hardware.py: remove NM dbus (#38005)

* hardware: read network info without NetworkManager DBus

* hardware: simplify wpa_cli SSID escape decoding

* hardware: restore cellular block in get_network_type to match master

* hardware: factor wpa_cli helper for key=value parsing

* hardware: comment SSID byte conversion for keyfile match

* hardware: comment NM metered enum values

* hardware: use check_output for ip route and wpa_cli helpers

* hardware: read default route iface from /proc/net/route

* hardware: simplify default route iface parsing

* hardware: only check for metered == 1

* hardware: also look for *.nmconnection in /data/etc/NetworkManager/system-connections

* hardware: use nmcli for runtime metered guess on wifi

* socket

* cleanup

* poor

* lil more

* mv that

---------

Co-authored-by: Adeeb Shihadeh <adeebshihadeh@gmail.com>
This commit is contained in:
Andi Radulescu
2026-05-30 21:09:17 +03:00
committed by GitHub
parent 8499de6afe
commit d937401511
+59 -52
View File
@@ -1,8 +1,9 @@
import configparser
import json
import os
import socket
import subprocess
import time
from enum import IntEnum
from functools import cached_property, lru_cache
from pathlib import Path
@@ -15,22 +16,7 @@ from openpilot.system.hardware.tici.lpa import TiciLPA
from openpilot.system.hardware.tici.pins import GPIO
from openpilot.system.hardware.tici.amplifier import Amplifier
NM = 'org.freedesktop.NetworkManager'
NM_CON_ACT = NM + '.Connection.Active'
NM_DEV = NM + '.Device'
NM_DEV_WL = NM + '.Device.Wireless'
NM_AP = NM + '.AccessPoint'
DBUS_PROPS = 'org.freedesktop.DBus.Properties'
class NMMetered(IntEnum):
NM_METERED_UNKNOWN = 0
NM_METERED_YES = 1
NM_METERED_NO = 2
NM_METERED_GUESS_YES = 3
NM_METERED_GUESS_NO = 4
MODEM_STATE_PATH = "/dev/shm/modem"
TIMEOUT = 0.1
NetworkType = log.DeviceState.NetworkType
NetworkStrength = log.DeviceState.NetworkStrength
@@ -52,16 +38,27 @@ def get_device_type():
model = f.read().strip('\x00')
return model.split('comma ')[-1]
def wpa_supplicant_cmd(cmd: str, timeout: float = 0.2) -> dict[str, str]:
with socket.socket(socket.AF_UNIX, socket.SOCK_DGRAM) as sock:
sock.settimeout(timeout)
sock.bind(f"\0openpilot-wpa-{os.getpid()}-{time.monotonic_ns()}")
sock.connect("/run/wpa_supplicant/wlan0")
sock.send(cmd.encode())
while True:
out = sock.recv(8192).decode("utf-8", "replace")
if out.startswith("<"):
continue
if out.startswith("FAIL"):
return {}
return dict(l.split("=", 1) for l in out.splitlines() if "=" in l)
def get_default_route_iface():
with open("/proc/net/route") as f:
routes = [(int(route[6]), route[0]) for line in f.readlines()[1:] if (route := line.split())[1] == "00000000" and int(route[3], 16) & 0x1]
return min(routes)[1] if routes else None
class Tici(HardwareBase):
@cached_property
def bus(self):
import dbus
return dbus.SystemBus()
@cached_property
def nm(self):
return self.bus.get_object(NM, '/org/freedesktop/NetworkManager')
@cached_property
def amplifier(self):
if self.get_device_type() == "mici":
@@ -115,13 +112,11 @@ class Tici(HardwareBase):
def get_network_type(self):
try:
primary_connection = self.nm.Get(NM, 'PrimaryConnection', dbus_interface=DBUS_PROPS, timeout=TIMEOUT)
primary_connection = self.bus.get_object(NM, primary_connection)
primary_type = primary_connection.Get(NM_CON_ACT, 'Type', dbus_interface=DBUS_PROPS, timeout=TIMEOUT)
if primary_type == '802-3-ethernet':
return NetworkType.ethernet
elif primary_type == '802-11-wireless':
return NetworkType.wifi
if (iface := get_default_route_iface()):
if iface.startswith('wlan'):
return NetworkType.wifi
if iface.startswith('eth'):
return NetworkType.ethernet
except Exception:
pass
@@ -138,10 +133,6 @@ class Tici(HardwareBase):
return NetworkType.cell2G
return NetworkType.none
def get_wlan(self):
wlan_path = self.nm.GetDeviceByIpIface('wlan0', dbus_interface=NM, timeout=TIMEOUT)
return self.bus.get_object(NM, wlan_path)
def get_sim_info(self):
ms = self.get_modem_state()
sim_id = ms.get('iccid', '')
@@ -191,13 +182,14 @@ class Tici(HardwareBase):
try:
if network_type == NetworkType.none:
pass
elif network_type == NetworkType.ethernet:
network_strength = NetworkStrength.great
elif network_type == NetworkType.wifi:
wlan = self.get_wlan()
active_ap_path = wlan.Get(NM_DEV_WL, 'ActiveAccessPoint', dbus_interface=DBUS_PROPS, timeout=TIMEOUT)
if active_ap_path != "/":
active_ap = self.bus.get_object(NM, active_ap_path)
strength = int(active_ap.Get(NM_AP, 'Strength', dbus_interface=DBUS_PROPS, timeout=TIMEOUT))
network_strength = self.parse_strength(strength)
rssi = wpa_supplicant_cmd("SIGNAL_POLL").get("RSSI")
if rssi is not None:
dbm = int(rssi)
if -100 < dbm <= 0:
network_strength = self.parse_strength(120 + max(-100, min(-20, dbm)))
else: # Cellular
network_strength = self.parse_strength(self.get_modem_state().get('signal_quality', 0))
except Exception:
@@ -210,17 +202,32 @@ class Tici(HardwareBase):
from openpilot.common.params import Params
return Params().get_bool("GsmMetered")
try:
primary_connection = self.nm.Get(NM, 'PrimaryConnection', dbus_interface=DBUS_PROPS, timeout=TIMEOUT)
primary_connection = self.bus.get_object(NM, primary_connection)
primary_devices = primary_connection.Get(NM_CON_ACT, 'Devices', dbus_interface=DBUS_PROPS, timeout=TIMEOUT)
if network_type == NetworkType.wifi:
ssid = wpa_supplicant_cmd("STATUS").get("ssid", "")
if ssid:
# wpa_supplicant escapes non-printable bytes as \xNN; NM keyfile stores ASCII SSIDs as a literal and others as a byte;byte; list
ssid_bytes = ssid.encode().decode('unicode_escape').encode('latin-1')
ssid_keyfile_list = ';'.join(str(b) for b in ssid_bytes) + ';'
for dev in primary_devices:
dev_obj = self.bus.get_object(NM, str(dev))
metered_prop = dev_obj.Get(NM_DEV, 'Metered', dbus_interface=DBUS_PROPS, timeout=TIMEOUT)
if network_type == NetworkType.wifi:
if metered_prop in [NMMetered.NM_METERED_YES, NMMetered.NM_METERED_GUESS_YES]:
return True
nm_dirs = ("/run/NetworkManager/system-connections", "/data/etc/NetworkManager/system-connections")
for fpath in (p for d in nm_dirs for p in Path(d).glob("*.nmconnection")):
raw = sudo_read(str(fpath))
if not raw:
continue
cp = configparser.ConfigParser(interpolation=None)
try:
cp.read_string(raw)
keyfile_ssid = cp.get("wifi", "ssid", fallback="")
if keyfile_ssid != ssid and keyfile_ssid != ssid_keyfile_list:
continue
metered = cp.getint("connection", "metered", fallback=0)
except (configparser.Error, ValueError):
continue
if metered == 1: # NM_METERED_YES
return True
if metered == 2: # NM_METERED_NO
return False
break
except Exception:
pass