Skip to content

Commit 96c7015

Browse files
authored
Update network.py
1 parent 38b1f22 commit 96c7015

1 file changed

Lines changed: 110 additions & 27 deletions

File tree

core/network.py

Lines changed: 110 additions & 27 deletions
Original file line numberDiff line numberDiff line change
@@ -1,61 +1,144 @@
11
"""
2-
KingWatch Pro - core/network.py
3-
Network RX/TX speeds from /proc/net/dev. No psutil.
2+
KingWatch Pro v17 - core/network.py
3+
Network speeds, ping estimate, signal strength, band detection (2G/3G/4G/5G).
4+
No psutil - uses /proc/net/dev + Android TelephonyManager via pyjnius.
45
"""
56
import time
7+
import subprocess
68

79
_prev_rx = 0
810
_prev_tx = 0
911
_prev_time = 0.0
1012

1113

1214
def _read_net_bytes():
13-
"""Sum RX/TX bytes across all non-loopback interfaces."""
14-
rx_total = tx_total = 0
15+
rx = tx = 0
1516
try:
1617
with open("/proc/net/dev") as f:
1718
for line in f:
18-
line = line.strip()
1919
if ":" not in line:
2020
continue
2121
iface, data = line.split(":", 1)
2222
if iface.strip() in ("lo",):
2323
continue
2424
parts = data.split()
2525
if len(parts) >= 9:
26-
rx_total += int(parts[0]) # receive bytes
27-
tx_total += int(parts[8]) # transmit bytes
26+
rx += int(parts[0])
27+
tx += int(parts[8])
2828
except Exception:
2929
pass
30-
return rx_total, tx_total
30+
return rx, tx
3131

3232

33-
def _bytes_to_human(bps: float) -> str:
33+
def _human(bps):
3434
if bps >= 1_000_000:
35-
return f"{bps / 1_000_000:.1f} MB/s"
35+
return f"{bps/1_000_000:.1f} MB/s"
3636
if bps >= 1_000:
37-
return f"{bps / 1_000:.1f} KB/s"
38-
return f"{bps:.0f} B/s"
37+
return f"{bps/1_000:.1f} KB/s"
38+
return f"{int(bps)} B/s"
3939

4040

41-
def _get_signal() -> str:
42-
"""Try to read WiFi signal from /proc/net/wireless."""
41+
def _ping_ms() -> str:
42+
"""RTT from /proc/net/snmp RetransSegs ratio as rough proxy, else N/A."""
43+
try:
44+
# Use kernel TCP RTT estimate from /proc/net/tcp if available
45+
with open("/proc/net/tcp") as f:
46+
lines = f.readlines()[1:6] # first 5 connections
47+
rtts = []
48+
for ln in lines:
49+
parts = ln.split()
50+
if len(parts) >= 14:
51+
# timeout field in jiffies (rough proxy at 250Hz = 4ms/jiffy)
52+
rto = int(parts[12], 16)
53+
ms = rto * 4
54+
if 4 < ms < 2000:
55+
rtts.append(ms)
56+
if rtts:
57+
return f"{min(rtts)} ms"
58+
except Exception:
59+
pass
60+
return "N/A"
61+
62+
63+
def _wifi_signal() -> str:
64+
"""WiFi signal level from /proc/net/wireless."""
4365
try:
4466
with open("/proc/net/wireless") as f:
4567
lines = f.readlines()
46-
for line in lines[2:]:
47-
parts = line.split()
68+
for ln in lines[2:]:
69+
parts = ln.split()
4870
if len(parts) >= 4:
4971
lvl = parts[3].rstrip(".")
50-
return f"Signal: {lvl} dBm"
72+
try:
73+
dbm = int(float(lvl))
74+
return f"WiFi {dbm} dBm"
75+
except Exception:
76+
pass
5177
except Exception:
5278
pass
5379
return ""
5480

5581

56-
def _get_ping() -> str:
57-
"""Simple ICMP-less ping estimate via /proc/net/snmp."""
58-
return "N/A"
82+
def _mobile_band() -> str:
83+
"""
84+
Detect mobile network type via Android TelephonyManager (pyjnius).
85+
Falls back to reading /sys/class/net interface names.
86+
Returns: 5G / 4G LTE / 3G / 2G / WiFi / Unknown
87+
"""
88+
try:
89+
from jnius import autoclass # type: ignore
90+
PythonActivity = autoclass("org.kivy.android.PythonActivity")
91+
Context = autoclass("android.content.Context")
92+
TelephonyManager = autoclass("android.telephony.TelephonyManager")
93+
94+
ctx = PythonActivity.mActivity
95+
tm = ctx.getSystemService(Context.TELEPHONY_SERVICE)
96+
nt = tm.getNetworkType()
97+
98+
# Android network type constants
99+
NR = 20 # 5G NR
100+
LTE = 13
101+
LTE_CA = 19
102+
HSPA_PLUS = 15
103+
HSPA = 10
104+
HSDPA = 8
105+
HSUPA = 9
106+
UMTS = 3
107+
EDGE = 2
108+
GPRS = 1
109+
CDMA = 4
110+
EVDO_0 = 5
111+
EVDO_A = 6
112+
EVDO_B = 12
113+
EHRPD = 14
114+
IDEN = 11
115+
GSM = 16
116+
IWLAN = 18
117+
118+
if nt == NR: return "5G NR"
119+
if nt in (LTE,LTE_CA): return "4G LTE"
120+
if nt in (HSPA_PLUS,): return "4G HSPA+"
121+
if nt in (HSPA,HSDPA,HSUPA,UMTS): return "3G HSPA"
122+
if nt in (EDGE,GPRS,CDMA,IDEN,GSM): return "2G"
123+
if nt in (EVDO_0,EVDO_A,EVDO_B,EHRPD): return "3G EVDO"
124+
if nt == IWLAN: return "WiFi Call"
125+
return f"Net#{nt}"
126+
except Exception:
127+
pass
128+
129+
# Fallback: check interface names
130+
try:
131+
import os
132+
ifaces = os.listdir("/sys/class/net")
133+
for iface in ifaces:
134+
if iface.startswith("wlan"):
135+
return _wifi_signal() or "WiFi"
136+
if iface.startswith(("rmnet", "ccmni", "seth")):
137+
return "Mobile"
138+
except Exception:
139+
pass
140+
141+
return _wifi_signal() or "Unknown"
59142

60143

61144
def get_network() -> dict:
@@ -65,16 +148,16 @@ def get_network() -> dict:
65148
rx, tx = _read_net_bytes()
66149
elapsed = now - _prev_time if _prev_time > 0 else 1.0
67150

68-
dl_bps = (rx - _prev_rx) / elapsed if _prev_rx else 0
69-
ul_bps = (tx - _prev_tx) / elapsed if _prev_tx else 0
151+
dl = max(0, (rx - _prev_rx) / elapsed) if _prev_rx else 0
152+
ul = max(0, (tx - _prev_tx) / elapsed) if _prev_tx else 0
70153

71154
_prev_rx = rx
72155
_prev_tx = tx
73156
_prev_time = now
74157

75158
return {
76-
"dl": _bytes_to_human(max(0, dl_bps)),
77-
"ul": _bytes_to_human(max(0, ul_bps)),
78-
"ping": _get_ping(),
79-
"signal": _get_signal(),
80-
}
159+
"dl": _human(dl),
160+
"ul": _human(ul),
161+
"ping": _ping_ms(),
162+
"signal": _mobile_band(),
163+
}

0 commit comments

Comments
 (0)