11"""
22KingWatch Pro v17 - core/network.py
3- Upload/Download speeds, ping, signal strength, 2G/3G/4G/5G band detection.
3+ Upload/download speeds from /proc/net/dev.
4+ Band detection: tries TelephonyManager via jnius, falls back to sysfs.
5+ Ping from /proc/net/tcp RTT estimate.
46"""
57import time
68import os
79
8- _prev_rx = 0
9- _prev_tx = 0
10- _prev_time = 0.0
10+ _prev_rx = 0
11+ _prev_tx = 0
12+ _prev_time = 0.0
13+ _band_cache = ""
14+ _band_time = 0.0
1115
1216
1317def _read_net_bytes ():
@@ -18,7 +22,8 @@ def _read_net_bytes():
1822 if ":" not in line :
1923 continue
2024 iface , data = line .split (":" , 1 )
21- if iface .strip () in ("lo" ,):
25+ iface = iface .strip ()
26+ if iface == "lo" :
2227 continue
2328 parts = data .split ()
2429 if len (parts ) >= 9 :
@@ -29,7 +34,7 @@ def _read_net_bytes():
2934 return rx , tx
3035
3136
32- def _human (bps ) :
37+ def _human (bps : float ) -> str :
3338 if bps >= 1_000_000 :
3439 return f"{ bps / 1_000_000 :.1f} MB/s"
3540 if bps >= 1_000 :
@@ -38,47 +43,67 @@ def _human(bps):
3843
3944
4045def _ping () -> str :
41- try :
42- with open ("/proc/net/tcp" ) as f :
43- lines = f .readlines ()[1 :8 ]
44- rtts = []
45- for ln in lines :
46- p = ln .split ()
47- if len (p ) >= 13 :
48- rto = int (p [12 ], 16 ) * 4 # jiffies at ~250Hz → ms
49- if 4 < rto < 2000 :
50- rtts .append (rto )
51- if rtts :
52- return f"{ min (rtts )} ms"
53- except Exception :
54- pass
46+ # Use /proc/net/tcp6 or tcp to estimate round-trip
47+ for path in ("/proc/net/tcp6" , "/proc/net/tcp" ):
48+ try :
49+ with open (path ) as f :
50+ lines = f .readlines ()[1 :10 ]
51+ rtts = []
52+ for ln in lines :
53+ p = ln .split ()
54+ if len (p ) >= 13 :
55+ # column 12 = retransmit timeout in jiffies (250Hz kernel → 4ms each)
56+ try :
57+ rto = int (p [12 ], 16 ) * 4
58+ if 4 < rto < 3000 :
59+ rtts .append (rto )
60+ except Exception :
61+ pass
62+ if rtts :
63+ return f"{ min (rtts )} ms"
64+ except Exception :
65+ continue
5566 return "N/A"
5667
5768
5869def _band () -> str :
59- """Detect mobile network type via Android TelephonyManager."""
60- try :
61- from jnius import autoclass
62- PythonActivity = autoclass ("org.kivy.android.PythonActivity" )
63- Context = autoclass ("android.content.Context" )
64- ctx = PythonActivity .mActivity
65- tm = ctx .getSystemService (Context .TELEPHONY_SERVICE )
66- nt = tm .getNetworkType ()
70+ """Network band - cached for 10s to avoid pyjnius overhead."""
71+ global _band_cache , _band_time
72+ now = time .monotonic ()
73+ if _band_cache and now - _band_time < 10 :
74+ return _band_cache
75+
76+ result = _detect_band ()
77+ _band_cache = result
78+ _band_time = now
79+ return result
6780
81+
82+ def _detect_band () -> str :
83+ # 1. Android TelephonyManager
84+ try :
85+ from jnius import autoclass # type: ignore
86+ PythonActivity = autoclass ("org.kivy.android.PythonActivity" )
87+ Context = autoclass ("android.content.Context" )
88+ ctx = PythonActivity .mActivity
89+ tm = ctx .getSystemService (Context .TELEPHONY_SERVICE )
90+ nt = tm .getNetworkType ()
6891 _MAP = {
69- 1 : "2G GPRS " , 2 : "2G EDGE " , 3 : "3G UMTS " ,
70- 4 : "2G CDMA " , 5 : "3G EVDO-0 " , 6 : "3G EVDO-A " ,
71- 7 : "2G 1xRTT" , 8 : "3G HSDPA " , 9 : "3G HSUPA " ,
72- 10 : "3G HSPA " , 11 : "2G iDEN " , 12 : "3G EVDO-B " ,
73- 13 : "4G LTE" , 14 : "3G eHRPD" , 15 : "4G HSPA+ " ,
74- 16 : "2G GSM " , 17 : "4G TDLTE " , 18 : "WiFi " ,
75- 19 : "4G LTE-CA" , 20 : "5G NR" ,
92+ 0 : "Unknown " , 1 : "2G GPRS " , 2 : "2G EDGE " ,
93+ 3 : "3G UMTS " , 4 : "2G CDMA " , 5 : "3G EVDO-0 " ,
94+ 6 : "3G EVDO-A" , 7 : "2G 1xRTT " , 8 : "3G HSDPA " ,
95+ 9 : "3G HSUPA " , 10 : "3G HSPA " , 11 : "2G iDEN " ,
96+ 12 : "3G EVDO-B" , 13 : "4G LTE" , 14 : "3G eHRPD" ,
97+ 15 : "4G HSPA+ " , 16 : "2G GSM " , 17 : "4G TD-LTE " ,
98+ 18 : "WiFi Call" , 19 : "4G LTE-CA" ,20 : "5G NR" ,
7699 }
77- return _MAP .get (nt , f"Net#{ nt } " )
100+ band = _MAP .get (nt , f"Net#{ nt } " )
101+ if band != "Unknown" :
102+ return band
78103 except Exception :
79104 pass
80105
81- # Fallback: check /proc/net/wireless for WiFi
106+ # 2. WiFi signal from /proc/net/wireless
82107 try :
83108 with open ("/proc/net/wireless" ) as f :
84109 lines = f .readlines ()
@@ -93,12 +118,20 @@ def _band() -> str:
93118 except Exception :
94119 pass
95120
96- # Check interface names
121+ # 3. Check active interface names
97122 try :
98- for iface in os .listdir ("/sys/class/net" ):
99- if iface .startswith ("wlan" ):
123+ for iface in sorted (os .listdir ("/sys/class/net" )):
124+ op = f"/sys/class/net/{ iface } /operstate"
125+ try :
126+ with open (op ) as f :
127+ state = f .read ().strip ()
128+ except Exception :
129+ state = ""
130+ if state != "up" :
131+ continue
132+ if iface .startswith ("wlan" ) or iface .startswith ("wlp" ):
100133 return "WiFi"
101- if iface .startswith (("rmnet" , "ccmni" , "seth_lte " )):
134+ if iface .startswith (("rmnet" , "ccmni" , "seth" , "usb " )):
102135 return "Mobile"
103136 except Exception :
104137 pass
@@ -111,10 +144,10 @@ def get_network() -> dict:
111144
112145 now = time .monotonic ()
113146 rx , tx = _read_net_bytes ()
114- elapsed = now - _prev_time if _prev_time > 0 else 1.0
115147
116- dl = max (0 , (rx - _prev_rx ) / elapsed ) if _prev_rx else 0
117- ul = max (0 , (tx - _prev_tx ) / elapsed ) if _prev_tx else 0
148+ elapsed = now - _prev_time if _prev_time > 0 else 1.0
149+ dl = max (0.0 , (rx - _prev_rx ) / elapsed ) if _prev_rx > 0 else 0.0
150+ ul = max (0.0 , (tx - _prev_tx ) / elapsed ) if _prev_tx > 0 else 0.0
118151
119152 _prev_rx = rx
120153 _prev_tx = tx
0 commit comments