feat: add active network interface status

app/pigway_pi_control.py

@@ -1,7 +1,7 @@
 #!/usr/bin/env python3
 import os,time,socket,subprocess,configparser,signal
 from pathlib import Path
-APP_VERSION="3.1.0"
+APP_VERSION="3.2.0"
 from smbus2 import SMBus
 from PIL import Image
 from shutil import disk_usage
@@ -22,7 +22,8 @@ prev_cpu=None
 page=0; page_since=time.monotonic()
 cpu_hi=mem_hi=0
 net_ok=link_ok=True
-ip="NO IP"
+net_iface=""; net_kind="NET"; net_metric="DOWN"; ip_label="IP4"; ip="NO IP"
+wifi_weak_since=None; wifi_weak_active=False
 last_net=last_watch=0
 watch_state=[]
 
@@ -133,10 +134,37 @@ def mem_pct():
     return 100*(vals["MemTotal"]-vals["MemAvailable"])/vals["MemTotal"]
 def disk_pct():
     d=disk_usage("/"); return 100*d.used/d.total
+def _run(args):
+    try: return subprocess.check_output(args,text=True,stderr=subprocess.DEVNULL,timeout=2).strip()
+    except Exception: return ""
+
+def primary_network():
+    out=_run(["ip","-4","route","get","1.1.1.1"])
+    dev=re.search(r"\\bdev\\s+(\\S+)",out); src=re.search(r"\\bsrc\\s+(\\S+)",out); label="IP4"
+    if not dev:
+        out=_run(["ip","-6","route","get","2606:4700:4700::1111"])
+        dev=re.search(r"\\bdev\\s+(\\S+)",out); src=re.search(r"\\bsrc\\s+(\\S+)",out); label="IP6"
+    if not dev: return "","NET","DOWN","IP4","NO IP"
+    iface=dev.group(1); ipval=src.group(1) if src else ""
+    if Path(f"/sys/class/net/{iface}/wireless").exists():
+        kind="WIF"; metric="--"
+        try:
+            for line in Path("/proc/net/wireless").read_text().splitlines():
+                if line.lstrip().startswith(iface+":"):
+                    metric=str(int(float(line.split()[3].rstrip(".")))); break
+        except Exception: pass
+    else:
+        kind="ETH"; metric="UP"
+        try:
+            n=int(Path(f"/sys/class/net/{iface}/speed").read_text().strip())
+            metric=f"{n//1000}G" if n>=1000 and n%1000==0 else str(n)
+        except Exception: pass
+    if label=="IP6" and len(ipval)>18: ipval=ipval[:7]+".."+ipval[-7:]
+    return iface,kind,metric,label,(ipval or "NO IP")
+
 def get_ip():
-    try:
-        s=socket.socket(socket.AF_INET,socket.SOCK_DGRAM);s.connect(("1.1.1.1",53));v=s.getsockname()[0];s.close();return v
-    except: return "NO IP"
+    return primary_network()[4]
+
 def internet():
     try:
         s=socket.create_connection(("1.1.1.1",53),0.5);s.close();return True
@@ -176,6 +204,9 @@ TEMP_HIGH=cf("alerts","temp_high",68.0)
 TEMP_CRIT=cf("alerts","temp_critical",75.0)
 DISK_HIGH=cf("alerts","disk_high",90.0)
 DISK_CRIT=cf("alerts","disk_critical",95.0)
+WIFI_WEAK=cf("alerts","wifi_weak",-75.0)
+WIFI_RECOVER=cf("alerts","wifi_recover",-68.0)
+WIFI_WEAK_SECONDS=cf("alerts","wifi_weak_seconds",30.0)
 
 FAN_T=[cf("fan","start",50.0),cf("fan","level2",53.0),cf("fan","level3",56.0),
        cf("fan","level4",59.0),cf("fan","max",62.0)]
@@ -201,6 +232,7 @@ def alert(title,l2="",l3="",l4="",sev=1,color=(20,16,0),blink=0):
             "sev":sev,"color":color,"blink":blink}
 
 def make_alerts(cpu,temp,mem,disk):
+    global wifi_weak_since,wifi_weak_active
     global cpu_hi,mem_hi
     a=[]
     cpu_hi=cpu_hi+1 if cpu>=CPU_HIGH else 0 if cpu<CPU_RECOVER else cpu_hi
@@ -215,6 +247,19 @@ def make_alerts(cpu,temp,mem,disk):
     elif not net_ok:a.append(alert("INTERNET DOWN","LOCAL LINK OK",ip,"",2,(0,8,28),1))
     for w,ok in zip(watches,watch_state):
         if not ok:a.append(alert("SERVICE DOWN",w[1],"CHECK / RESTART","",2,(0,24,24),1))
+    if net_kind=="WIF" and str(net_metric).lstrip("-").isdigit():
+        rssi=float(net_metric)
+        if not wifi_weak_active:
+            if rssi <= WIFI_WEAK:
+                if wifi_weak_since is None: wifi_weak_since=time.monotonic()
+                if time.monotonic()-wifi_weak_since >= WIFI_WEAK_SECONDS: wifi_weak_active=True
+            else: wifi_weak_since=None
+        elif rssi >= WIFI_RECOVER:
+            wifi_weak_active=False; wifi_weak_since=None
+        if wifi_weak_active:
+            a.append(alert("WIFI SIGNAL WEAK",f"{rssi:.0f} dBm",f"IF {net_iface}","CHECK SIGNAL",1,(0,0,28),1))
+    else:
+        wifi_weak_since=None; wifi_weak_active=False
     a.sort(key=lambda x:x["sev"],reverse=True)
     return a
 
@@ -238,12 +283,12 @@ def render_home(cpu,temp,mem,disk):
         f"CPU {cpu:.1f}%",
         f"TMP {temp:.1f}C",
         f"DSK {disk:.1f}%",
-        ("IP6 " if ":" in ip else "IP4 ")+ip,
+        ip_label+" "+ip,
     ]
     right=[
         f"MEM {mem:.1f}%",
         f"FAN {FAN_NAME[fan_level]}",
-        f"NET {'OK' if net_ok else 'DOWN'}",
+        f"{net_kind} {net_metric}",
         "",
     ]
     for row in range(4):
@@ -302,7 +347,7 @@ while running:
             latest=(c,t,m,d); update_fan(t); last_sensor=now
 
         if now-last_net >= NETWORK_INTERVAL:
-            ip=get_ip(); link_ok=ip!="NO IP"; net_ok=internet() if link_ok else False
+            net_iface,net_kind,net_metric,ip_label,ip=primary_network(); link_ok=ip!="NO IP"; net_ok=internet() if link_ok else False
             last_net=now
 
         if now-last_watch >= SERVICE_INTERVAL: