From c9d3ddb299d9d4f9ac1b9f325d4e8c3dfe1cf144 Mon Sep 17 00:00:00 2001 From: Geoffrey Kruse <6468053+doggkruse@users.noreply.github.com> Date: Thu, 11 Dec 2025 23:23:04 -0400 Subject: [PATCH 1/8] First pass at adding interfaces and modems --- gli4py/interfaces.py | 348 +++++++++++++++++++++++++++++++++++++++ gli4py/modem.py | 381 +++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 729 insertions(+) create mode 100644 gli4py/interfaces.py create mode 100644 gli4py/modem.py diff --git a/gli4py/interfaces.py b/gli4py/interfaces.py new file mode 100644 index 0000000..f12b6d8 --- /dev/null +++ b/gli4py/interfaces.py @@ -0,0 +1,348 @@ +"""Helpers for inspecting GL.iNet interface / Multi‑WAN state.""" + +from __future__ import annotations + +from dataclasses import dataclass +from enum import IntEnum +from typing import Any, Dict, Iterable, List, Optional + +from .modem import ModemInfo, ModemManager, ModemStatusEntry + +JSONDict = Dict[str, Any] + + +class InterfaceStatus(IntEnum): + """Interface connectivity status reported by kmwan.""" + + ONLINE = 0 + OFFLINE = 1 + ERROR = 2 + + +class MultiWANMode(IntEnum): + """Multi-WAN operating mode reported by kmwan.""" + + FAILOVER = 0 + LOAD_BALANCING = 1 + + +@dataclass(slots=True) +class InterfaceInfo: + """Joined view of kmwan configuration and status for a single interface. + + Attributes + ---------- + name: + Logical interface name as used by kmwan (e.g. ``"wan"``, ``"wwan"``, + ``"tethering"``, ``"modem_0001"``, ``"secondwan"``). + metric: + Fail‑over priority (lower value == higher priority). ``None`` if not + provided by the firmware / config. + weight: + Load‑balancing weight in balancing mode. ``None`` if not present. + status_v4: + IPv4 status as reported by ``kmwan.get_status``. + According to the device documentation this is typically: + + * ``0`` – online + * ``1`` – offline + + Some firmwares also use ``2`` for an explicit error state. + status_v6: + IPv6 status, with the same coding as :attr:`status_v4`. + """ + + name: str + metric: Optional[int] = None + weight: Optional[int] = None + status_v4: InterfaceStatus | None = None + status_v6: InterfaceStatus | None = None + modem: ModemDetails | None = None + + def ipv4_online(self) -> Optional[bool]: + """Return ``True`` if IPv4 is explicitly online, ``False`` if + explicitly offline, or ``None`` if there is no information. + """ + + if self.status_v4 is None: + return None + return self.status_v4 == InterfaceStatus.ONLINE + + def ipv6_online(self) -> Optional[bool]: + """Return ``True`` if IPv6 is explicitly online, ``False`` if + explicitly offline, or ``None`` if there is no information. + """ + + if self.status_v6 is None: + return None + return self.status_v6 == InterfaceStatus.ONLINE + + def is_online(self, prefer_ipv6: bool = False) -> bool: + """Coarse "is this usable" predicate. + + The rules are conservative and intentionally simple: + + * If *prefer_ipv6* is true and there is an IPv6 state, use that. + * Otherwise, prefer IPv4 if present. + * If we only have one of them, use whichever we have. + * Any non‑zero status is treated as *not online*. + """ + + v4 = self.ipv4_online() + v6 = self.ipv6_online() + + if prefer_ipv6 and v6 is not None: + return bool(v6) + if v4 is not None: + return bool(v4) + if v6 is not None: + return bool(v6) + return False + + +@dataclass(slots=True) +class ModemDetails: + """Combined modem runtime status and hardware info.""" + + status: ModemStatusEntry | None = None + info: ModemInfo | None = None + + +@dataclass(slots=True) +class MultiWANState: + """Snapshot of the router's Multi‑WAN configuration and status.""" + + mode: MultiWANMode + """Multi‑WAN mode.""" + + interfaces: Dict[str, InterfaceInfo] + """Mapping from interface name to :class:`InterfaceInfo`.""" + + primary: str | None = None + """Interface name that is currently considered primary, if known.""" + + +class InterfaceManager: + """High‑level helper for kmwan interface status. + + Parameters + ---------- + client: + A GL.iNet API client instance that exposes ``gen_sid_payload`` and + ``_request`` helpers like :class:`gli4py.glinet.GLinet`. + """ + + def __init__(self, client: Any) -> None: + self._client = client + self._modem_manager = ModemManager(client) + + # ------------------------------------------------------------------ + # Raw fetchers + # ------------------------------------------------------------------ + async def fetch_kmwan_status(self) -> JSONDict: + """Return the raw payload from ``kmwan.get_status``. + + Typical shape:: + + { + "interfaces": [ + { + "interface": "wan", + "status_v4": 0, + "status_v6": 1, + }, + ... + ] + } + """ + + return await self._client._request( + self._client.gen_sid_payload( + "call", + ["kmwan", "get_status"], + self._client.sid, + ) + ) + + async def fetch_kmwan_config(self) -> JSONDict: + """Return the raw payload from ``kmwan.get_config``. + + Firmware‑specific details vary, but you can generally expect:: + + { + "mode": 0 | 1, + "interfaces": [ + { + "interface": "wan", + "metric": 1, + "weight": 1, + ... + }, + ... + ] + } + """ + + return await self._client._request( + self._client.gen_sid_payload( + "call", + ["kmwan", "get_config"], + self._client.sid, + ) + ) + + async def fetch_network_interface_status(self) -> JSONDict: + """Return status for a single interface via ``network.interface status``.""" + + return await self._client._request( + self._client.gen_sid_payload( + "call", + ["network.interface", "status", {"interface": "wan"}], + self._client.sid, + ) + ) + + # ------------------------------------------------------------------ + # Joined / parsed state + # ------------------------------------------------------------------ + async def get_state(self, prefer_ipv6: bool = False) -> MultiWANState: + """Fetch and join kmwan configuration + status into one object.""" + + config = await self.fetch_kmwan_config() + mwan_status = await self.fetch_kmwan_status() + # iface_status = await self.fetch_network_interface_status() + + interfaces: Dict[str, InterfaceInfo] = {} + + # First layer in config (metric / weight). + for entry in config.get("interfaces", []) or []: + name = entry.get("interface") + if not name: + continue + info = interfaces.get(name) + if info is None: + info = InterfaceInfo(name=name) + interfaces[name] = info + + info.metric = entry.get("metric") + info.weight = entry.get("weight") + + # Second layer: live status (IPv4 / IPv6 online / offline). + for entry in mwan_status.get("interfaces", []) or []: + name = entry.get("interface") + if not name: + continue + info = interfaces.get(name) + if info is None: + info = InterfaceInfo(name=name) + interfaces[name] = info + + # kmwan uses 0 = online, 1 = offline; some firmwares extend this + # to 2 = error. We keep the raw integer and interpret it in + # InterfaceInfo helper methods. + info.status_v4 = self._parse_status(entry.get("status_v4")) + info.status_v6 = self._parse_status(entry.get("status_v6")) + + mode_value = int(config.get("mode", 0)) + try: + mode = MultiWANMode(mode_value) + except ValueError: + mode = MultiWANMode.FAILOVER + primary = self._select_primary(interfaces, mode, prefer_ipv6) + + # Attach modem status entries to modem interfaces if available. + modem_interfaces = sorted( + [name for name in interfaces if name.startswith("modem")] + ) + if modem_interfaces: + modem_statuses = await self._modem_manager.get_status() + modem_infos = await self._modem_manager.get_info() + for idx, iface_name in enumerate(modem_interfaces): + modem_details = ModemDetails() + if idx < len(modem_statuses): + modem_details.status = modem_statuses[idx] + if idx < len(modem_infos): + modem_details.info = modem_infos[idx] + interfaces[iface_name].modem = modem_details + + return MultiWANState( + mode=mode, + interfaces=interfaces, + primary=primary.name if primary else None, + ) + + # ------------------------------------------------------------------ + # High‑level helpers + # ------------------------------------------------------------------ + def _select_primary( + self, + interfaces: Dict[str, InterfaceInfo], + mode: MultiWANMode, + prefer_ipv6: bool, + ) -> Optional[InterfaceInfo]: + """Return the primary interface based on kmwan mode and status.""" + + if not interfaces: + return None + + def by_metric(iters: Iterable[InterfaceInfo]) -> List[InterfaceInfo]: + def metric_key(info: InterfaceInfo) -> int: + # Higher metrics are lower priority; push "None" to the end. + return info.metric if info.metric is not None else 10_000 + + return sorted(iters, key=metric_key) + + online = [ + i for i in interfaces.values() if i.is_online(prefer_ipv6=prefer_ipv6) + ] + if not online: + return None + + if mode == MultiWANMode.FAILOVER: + return by_metric(online)[0] + + def weight_metric_key(info: InterfaceInfo) -> tuple[int, int]: + w = info.weight if info.weight is not None else 0 + m = info.metric if info.metric is not None else 10_000 + return (-w, m) + + return sorted(online, key=weight_metric_key)[0] + + @staticmethod + def _parse_status(value: Any) -> InterfaceStatus | None: + """Convert kmwan status integers to InterfaceStatus.""" + + if value is None: + return None + if isinstance(value, InterfaceStatus): + return value + if isinstance(value, bool): + return InterfaceStatus.ONLINE if value else InterfaceStatus.OFFLINE + if isinstance(value, dict): + for key in ("up", "online", "connected", "link", "state", "status"): + if key in value: + v = value[key] + if isinstance(v, bool): + return InterfaceStatus.ONLINE if v else InterfaceStatus.OFFLINE + if isinstance(v, str): + lowered = v.lower() + if lowered in {"up", "online", "connected"}: + return InterfaceStatus.ONLINE + if lowered in {"down", "offline", "disconnected"}: + return InterfaceStatus.OFFLINE + if isinstance(v, int): + try: + return InterfaceStatus(int(v)) + except ValueError: + pass + # As a fallback, treat any truthy dict with no known key as unknown. + if isinstance(value, str): + lowered = value.lower() + if lowered in {"online", "up", "connected"}: + return InterfaceStatus.ONLINE + if lowered in {"offline", "down", "disconnected"}: + return InterfaceStatus.OFFLINE + try: + return InterfaceStatus(int(value)) + except ValueError: + return None diff --git a/gli4py/modem.py b/gli4py/modem.py new file mode 100644 index 0000000..a8cd9c2 --- /dev/null +++ b/gli4py/modem.py @@ -0,0 +1,381 @@ +"""Helpers for fetching and parsing modem status.""" + +from __future__ import annotations + +from dataclasses import dataclass +from enum import IntEnum +from typing import Any, Dict, List, Optional + +JSONDict = dict[str, Any] + + +class ModemConnectionState(IntEnum): + """Connection state derived from modem network status.""" + + UNKNOWN = -1 + DISCONNECTED = 0 + CONNECTED = 1 + + +class ModemRegistrationStatus(IntEnum): + """SIM registration status.""" + + REGISTERED = 0 + UNREGISTERED = 1 + NEEDS_PIN = 2 + + +class ModemNetworkStatus(IntEnum): + """Network status for modem network block.""" + + CONNECTED = 0 + CONNECTING = 1 + + +class ModemNetworkMode(IntEnum): + """Network mode reported in signal section.""" + + GSM = 2 + UMTS = 3 + LTE = 4 + FIVE_G = 5 + LTE_ADVANCED = 41 + + @property + def label(self) -> str: + """Return a user-friendly label for the network mode.""" + + return { + ModemNetworkMode.GSM: "2G", + ModemNetworkMode.UMTS: "3G", + ModemNetworkMode.LTE: "LTE", + ModemNetworkMode.FIVE_G: "5G", + ModemNetworkMode.LTE_ADVANCED: "4G+", + }.get(self, self.name) + + +class ModemSignalStrength(IntEnum): + """Overall signal strength buckets.""" + + POOR = 1 + FAIR = 2 + GOOD = 3 + EXCELLENT = 4 + + +class ModemAutoSwitchState(IntEnum): + """SIM auto-switch state for dual-SIM devices.""" + + ENABLED = 0 + DISABLED = 1 + + +class ModemType(IntEnum): + """Modem type as reported by modem.get_info.""" + + BUILT_IN = 0 + EXTERNAL = 1 + UNSUPPORTED = 2 + + +@dataclass(slots=True) +class SimCardInfo: + """SIM card details.""" + + iccid: str | None = None + phone_number: str | None = None + mcc: str | None = None + mnc: str | None = None + + +@dataclass(slots=True) +class ModemInfo: + """Hardware info returned by modem.get_info.""" + + bus: str | None + type: ModemType | None + at_port: str | None + data_port: str | None + sms_support: bool | None + lock_tower_support: bool | None + qcfg_unsupport: bool | None + imei: str | None + name: str | None + version: str | None + vendor: str | None + protocols: list[str] | None + devices: list[str] | None + simcard: SimCardInfo | None + + +@dataclass(slots=True) +class ModemSignal: + """Signal details for a SIM.""" + + mode: ModemNetworkMode | None + strength: ModemSignalStrength | None + rssi: int | None + rsrp: int | None + rsrq: int | None + sinr: int | None + ecio: int | None + + +@dataclass(slots=True) +class ModemNetworkIP: + """IP details for a stack.""" + + ip: str | None + netmask: str | None + gateway: str | None + dns: list[str] | None + + +@dataclass(slots=True) +class ModemNetwork: + """Network status for the modem.""" + + status: ModemNetworkStatus | None + traffic_total: int | None + ipv4: ModemNetworkIP | None + ipv6: ModemNetworkIP | None + + +@dataclass(slots=True) +class ModemStatusEntry: + """Status entry returned by modem.get_status.""" + + bus: str | None + current_sim: int | None + switch_status: ModemAutoSwitchState | None + sim_status: ModemRegistrationStatus | None + sim_operator: str | None + sim_iccid: str | None + sim_phone_number: str | None + sim_mcc: str | None + sim_mnc: str | None + signal: ModemSignal | None + network: ModemNetwork | None + connection_state: ModemConnectionState + new_sms_count: int | None + passthrough: JSONDict | None + err_code: int | None + err_msg: str | None + + +class ModemManager: + """High-level helper for modem status retrieval.""" + + def __init__(self, client: Any) -> None: + self._client = client + + async def _rpc_call( + self, namespace: str, method: str, params: Optional[JSONDict] = None + ) -> JSONDict: + """Thin wrapper around the parent client's JSON-RPC helper.""" + + return await self._client._request( # type: ignore[attr-defined,no-any-return] + self._client.gen_sid_payload( # type: ignore[attr-defined] + "call", [namespace, method, params or {}], self._client.sid # type: ignore[attr-defined] + ) + ) + + async def fetch_modem_status(self) -> JSONDict: + """Return raw modem runtime status for all modems.""" + + return await self._rpc_call("modem", "get_status", {}) + + async def fetch_modem_info(self) -> JSONDict: + """Return raw modem hardware info.""" + + return await self._rpc_call("modem", "get_info", {}) + + async def get_status(self) -> list[ModemStatusEntry]: + """Fetch and parse modem status.""" + + payload = await self.fetch_modem_status() + modems = payload.get("modems", []) if isinstance(payload, dict) else [] + return [self._parse_status_entry(entry) for entry in modems] + + async def get_info(self) -> list[ModemInfo]: + """Fetch and parse modem hardware info.""" + + payload = await self.fetch_modem_info() + modems = payload.get("modems", []) if isinstance(payload, dict) else [] + return [self._parse_info(entry) for entry in modems] + + @staticmethod + @staticmethod + def _parse_info(entry: JSONDict) -> ModemInfo: + """Parse a modem entry from modem.get_info.""" + + type_value = entry.get("type") + modem_type: ModemType | None = None + try: + modem_type = ModemType(int(type_value)) + except (TypeError, ValueError): + modem_type = None + + sim_entry = entry.get("simcard") or {} + simcard = SimCardInfo( + iccid=sim_entry.get("iccid"), + phone_number=sim_entry.get("phone_number"), + mcc=sim_entry.get("mcc"), + mnc=sim_entry.get("mnc"), + ) if sim_entry else None + + devices = entry.get("devices") + if isinstance(devices, list): + devices_list: List[str] | None = [str(dev) for dev in devices] + else: + devices_list = None + + protocols = entry.get("protocols") + if isinstance(protocols, list): + protocols_list: List[str] | None = [str(proto) for proto in protocols] + else: + protocols_list = None + + return ModemInfo( + bus=entry.get("bus"), + type=modem_type, + at_port=entry.get("at_port"), + data_port=entry.get("data_port"), + sms_support=entry.get("sms_support"), + lock_tower_support=entry.get("lock_tower_support"), + qcfg_unsupport=entry.get("qcfg_unsupport"), + imei=entry.get("imei"), + name=entry.get("name"), + version=entry.get("version"), + vendor=entry.get("vendor"), + protocols=protocols_list, + devices=devices_list, + simcard=simcard, + ) + + @staticmethod + def _parse_status_entry(entry: JSONDict) -> ModemStatusEntry: + """Parse a modem entry from modem.get_status.""" + + sim_entry = entry.get("simcard") or {} + signal_entry = sim_entry.get("signal") or {} + network_entry = entry.get("network") or {} + + def _parse_signal() -> ModemSignal | None: + if not signal_entry: + return None + mode_value = signal_entry.get("mode") + try: + mode = ModemNetworkMode(int(mode_value)) if mode_value is not None else None + except ValueError: + mode = None + strength_value = signal_entry.get("strength") + try: + strength = ( + ModemSignalStrength(int(strength_value)) + if strength_value is not None + else None + ) + except ValueError: + strength = None + return ModemSignal( + mode=mode, + strength=strength, + rssi=signal_entry.get("rssi"), + rsrp=signal_entry.get("rsrp"), + rsrq=signal_entry.get("rsrq"), + sinr=signal_entry.get("sinr"), + ecio=signal_entry.get("ecio"), + ) + + def _parse_ip(ip_entry: Any) -> ModemNetworkIP | None: + if not isinstance(ip_entry, dict): + return None + dns = ip_entry.get("dns") + if isinstance(dns, list): + dns_list: list[str] | None = [str(v) for v in dns] + else: + dns_list = None + return ModemNetworkIP( + ip=ip_entry.get("ip"), + netmask=ip_entry.get("netmask"), + gateway=ip_entry.get("gateway"), + dns=dns_list, + ) + + def _parse_network() -> tuple[ModemNetwork | None, ModemConnectionState]: + if not network_entry: + return None, ModemConnectionState.UNKNOWN + status_value = network_entry.get("status") + status: ModemNetworkStatus | None = None + if isinstance(status_value, str): + lowered = status_value.lower() + if lowered == "connected": + status = ModemNetworkStatus.CONNECTED + elif lowered == "connecting": + status = ModemNetworkStatus.CONNECTING + else: + try: + status = ModemNetworkStatus(int(status_value)) + except ValueError: + status = None + elif status_value is not None: + try: + status = ModemNetworkStatus(int(status_value)) + except ValueError: + status = None + conn_state = ModemConnectionState.UNKNOWN + if status == ModemNetworkStatus.CONNECTED: + conn_state = ModemConnectionState.CONNECTED + elif status == ModemNetworkStatus.CONNECTING: + conn_state = ModemConnectionState.DISCONNECTED + return ( + ModemNetwork( + status=status, + traffic_total=network_entry.get("traffic_total"), + ipv4=_parse_ip(network_entry.get("ipv4")), + ipv6=_parse_ip(network_entry.get("ipv6")), + ), + conn_state, + ) + + sim_status_value = sim_entry.get("status") + try: + sim_status = ( + ModemRegistrationStatus(int(sim_status_value)) + if sim_status_value is not None + else None + ) + except ValueError: + sim_status = None + + network_parsed, conn_state = _parse_network() + + switch_status_value = entry.get("switch_status") + try: + switch_status = ( + ModemAutoSwitchState(int(switch_status_value)) + if switch_status_value is not None + else None + ) + except ValueError: + switch_status = None + + return ModemStatusEntry( + bus=entry.get("bus"), + current_sim=entry.get("current_sim"), + switch_status=switch_status, + sim_status=sim_status, + sim_operator=sim_entry.get("carrier"), + sim_iccid=sim_entry.get("iccid"), + sim_phone_number=sim_entry.get("phone_number"), + sim_mcc=sim_entry.get("mcc"), + sim_mnc=sim_entry.get("mnc"), + signal=_parse_signal(), + network=network_parsed, + connection_state=conn_state, + new_sms_count=entry.get("new_sms_count"), + passthrough=entry.get("passthrough"), + err_code=entry.get("err_code"), + err_msg=entry.get("err_msg"), + ) From f296256d83574606e0c0fc778a8586d45dbc2668 Mon Sep 17 00:00:00 2001 From: Geoffrey Kruse <6468053+doggkruse@users.noreply.github.com> Date: Sun, 14 Dec 2025 11:40:24 -0400 Subject: [PATCH 2/8] Add cell tower info via modem.get_cells_info --- gli4py/modem.py | 99 +++++++++++++++++++++++++++++++++++++++++++++++-- 1 file changed, 96 insertions(+), 3 deletions(-) diff --git a/gli4py/modem.py b/gli4py/modem.py index a8cd9c2..9036f89 100644 --- a/gli4py/modem.py +++ b/gli4py/modem.py @@ -141,6 +141,27 @@ class ModemNetwork: ipv6: ModemNetworkIP | None +@dataclass(slots=True) +class CellInfo: + """Cell details from modem.get_cells_info.""" + + ul_bandwidth: str | None + dl_bandwidth: str | None + rsrp: int | None + id: str | None + rssi: int | None + tx_channel: str | None + sinr_level: int | None + rsrq_level: int | None + sinr: int | None + rsrq: int | None + rssi_level: int | None + rsrp_level: int | None + mode: str | None + band: int | None + type: str | None + + @dataclass(slots=True) class ModemStatusEntry: """Status entry returned by modem.get_status.""" @@ -156,6 +177,7 @@ class ModemStatusEntry: sim_mnc: str | None signal: ModemSignal | None network: ModemNetwork | None + cells_info: list[CellInfo] | None connection_state: ModemConnectionState new_sms_count: int | None passthrough: JSONDict | None @@ -190,12 +212,31 @@ async def fetch_modem_info(self) -> JSONDict: return await self._rpc_call("modem", "get_info", {}) + async def fetch_cells_info(self, bus: str) -> JSONDict: + """Return raw cell information for a modem.""" + + return await self._rpc_call("modem", "get_cells_info", {"bus": bus}) + async def get_status(self) -> list[ModemStatusEntry]: """Fetch and parse modem status.""" payload = await self.fetch_modem_status() modems = payload.get("modems", []) if isinstance(payload, dict) else [] - return [self._parse_status_entry(entry) for entry in modems] + + parsed_entries: list[ModemStatusEntry] = [] + for entry in modems: + if not isinstance(entry, dict): + continue + bus = entry.get("bus") + cells_info: list[CellInfo] | None = None + if bus is not None: + try: + cells_payload = await self.fetch_cells_info(str(bus)) + cells_info = self._parse_cells_info(cells_payload) + except Exception: # noqa: BLE001 + cells_info = None + parsed_entries.append(self._parse_status_entry(entry, cells_info)) + return parsed_entries async def get_info(self) -> list[ModemInfo]: """Fetch and parse modem hardware info.""" @@ -204,7 +245,12 @@ async def get_info(self) -> list[ModemInfo]: modems = payload.get("modems", []) if isinstance(payload, dict) else [] return [self._parse_info(entry) for entry in modems] - @staticmethod + async def get_cells_info(self, bus: str) -> list[CellInfo] | None: + """Fetch and return parsed cell information for a modem.""" + + payload = await self.fetch_cells_info(bus) + return self._parse_cells_info(payload) + @staticmethod def _parse_info(entry: JSONDict) -> ModemInfo: """Parse a modem entry from modem.get_info.""" @@ -254,7 +300,9 @@ def _parse_info(entry: JSONDict) -> ModemInfo: ) @staticmethod - def _parse_status_entry(entry: JSONDict) -> ModemStatusEntry: + def _parse_status_entry( + entry: JSONDict, cells_info: list[CellInfo] | None = None + ) -> ModemStatusEntry: """Parse a modem entry from modem.get_status.""" sim_entry = entry.get("simcard") or {} @@ -373,9 +421,54 @@ def _parse_network() -> tuple[ModemNetwork | None, ModemConnectionState]: sim_mnc=sim_entry.get("mnc"), signal=_parse_signal(), network=network_parsed, + cells_info=cells_info, connection_state=conn_state, new_sms_count=entry.get("new_sms_count"), passthrough=entry.get("passthrough"), err_code=entry.get("err_code"), err_msg=entry.get("err_msg"), ) + + @staticmethod + def _parse_cells_info(payload: JSONDict | None) -> list[CellInfo] | None: + """Parse cell info payload.""" + + if not isinstance(payload, dict): + return None + body: JSONDict = payload.get("result", payload) if isinstance(payload, dict) else {} + cells = body.get("cells") if isinstance(body, dict) else None + if not isinstance(cells, list): + return None + + parsed: list[CellInfo] = [] + + def _as_int(value: Any) -> int | None: + try: + return int(value) if value is not None else None + except (TypeError, ValueError): + return None + + for cell in cells: + if not isinstance(cell, dict): + continue + parsed.append( + CellInfo( + ul_bandwidth=cell.get("ul_bandwidth"), + dl_bandwidth=cell.get("dl_bandwidth"), + rsrp=_as_int(cell.get("rsrp")), + id=str(cell.get("id")) if cell.get("id") is not None else None, + rssi=_as_int(cell.get("rssi")), + tx_channel=str(cell.get("tx_channel")) if cell.get("tx_channel") is not None else None, + sinr_level=_as_int(cell.get("sinr_level")), + rsrq_level=_as_int(cell.get("rsrq_level")), + sinr=_as_int(cell.get("sinr")), + rsrq=_as_int(cell.get("rsrq")), + rssi_level=_as_int(cell.get("rssi_level")), + rsrp_level=_as_int(cell.get("rsrp_level")), + mode=cell.get("mode"), + band=_as_int(cell.get("band")), + type=cell.get("type"), + ) + ) + + return parsed or None From 19abf943f110f0a0f09d4c4513bb7b2e707ceb1a Mon Sep 17 00:00:00 2001 From: Geoffrey Kruse <6468053+doggkruse@users.noreply.github.com> Date: Sun, 14 Dec 2025 23:05:59 -0400 Subject: [PATCH 3/8] Add 51 for 5G-NSA --- gli4py/modem.py | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/gli4py/modem.py b/gli4py/modem.py index 9036f89..f78f5e7 100644 --- a/gli4py/modem.py +++ b/gli4py/modem.py @@ -38,8 +38,9 @@ class ModemNetworkMode(IntEnum): GSM = 2 UMTS = 3 LTE = 4 - FIVE_G = 5 LTE_ADVANCED = 41 + FIVE_G = 5 + FIVE_G_NSA = 51 @property def label(self) -> str: @@ -49,8 +50,9 @@ def label(self) -> str: ModemNetworkMode.GSM: "2G", ModemNetworkMode.UMTS: "3G", ModemNetworkMode.LTE: "LTE", - ModemNetworkMode.FIVE_G: "5G", ModemNetworkMode.LTE_ADVANCED: "4G+", + ModemNetworkMode.FIVE_G: "5G", + ModemNetworkMode.FIVE_G_NSA: "5G", }.get(self, self.name) From 905802dbc97ea9d942002772b2668d197179fd92 Mon Sep 17 00:00:00 2001 From: Geoffrey Kruse <6468053+doggkruse@users.noreply.github.com> Date: Fri, 16 Jan 2026 09:45:14 -0800 Subject: [PATCH 4/8] Add reboot modem command --- gli4py/modem.py | 12 ++++++++++++ 1 file changed, 12 insertions(+) diff --git a/gli4py/modem.py b/gli4py/modem.py index f78f5e7..0fe22b6 100644 --- a/gli4py/modem.py +++ b/gli4py/modem.py @@ -214,6 +214,18 @@ async def fetch_modem_info(self) -> JSONDict: return await self._rpc_call("modem", "get_info", {}) + async def reboot_modem(self, bus: str, *, hw_reboot: bool = True) -> JSONDict: + """Reboot a modem, optionally forcing hardware reboot.""" + + return await self._rpc_call( + "modem", + "reboot_modem", + { + "bus": bus, + "hw_reboot": hw_reboot, + }, + ) + async def fetch_cells_info(self, bus: str) -> JSONDict: """Return raw cell information for a modem.""" From b1fc6aed75e2508d8c0d2bbcb8a657d018e2263b Mon Sep 17 00:00:00 2001 From: Geoffrey Kruse <6468053+doggkruse@users.noreply.github.com> Date: Thu, 11 Dec 2025 23:23:04 -0400 Subject: [PATCH 5/8] First pass at adding interfaces and modems --- gli4py/interfaces.py | 348 +++++++++++++++++++++++++++++++++++++++ gli4py/modem.py | 381 +++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 729 insertions(+) create mode 100644 gli4py/interfaces.py create mode 100644 gli4py/modem.py diff --git a/gli4py/interfaces.py b/gli4py/interfaces.py new file mode 100644 index 0000000..f12b6d8 --- /dev/null +++ b/gli4py/interfaces.py @@ -0,0 +1,348 @@ +"""Helpers for inspecting GL.iNet interface / Multi‑WAN state.""" + +from __future__ import annotations + +from dataclasses import dataclass +from enum import IntEnum +from typing import Any, Dict, Iterable, List, Optional + +from .modem import ModemInfo, ModemManager, ModemStatusEntry + +JSONDict = Dict[str, Any] + + +class InterfaceStatus(IntEnum): + """Interface connectivity status reported by kmwan.""" + + ONLINE = 0 + OFFLINE = 1 + ERROR = 2 + + +class MultiWANMode(IntEnum): + """Multi-WAN operating mode reported by kmwan.""" + + FAILOVER = 0 + LOAD_BALANCING = 1 + + +@dataclass(slots=True) +class InterfaceInfo: + """Joined view of kmwan configuration and status for a single interface. + + Attributes + ---------- + name: + Logical interface name as used by kmwan (e.g. ``"wan"``, ``"wwan"``, + ``"tethering"``, ``"modem_0001"``, ``"secondwan"``). + metric: + Fail‑over priority (lower value == higher priority). ``None`` if not + provided by the firmware / config. + weight: + Load‑balancing weight in balancing mode. ``None`` if not present. + status_v4: + IPv4 status as reported by ``kmwan.get_status``. + According to the device documentation this is typically: + + * ``0`` – online + * ``1`` – offline + + Some firmwares also use ``2`` for an explicit error state. + status_v6: + IPv6 status, with the same coding as :attr:`status_v4`. + """ + + name: str + metric: Optional[int] = None + weight: Optional[int] = None + status_v4: InterfaceStatus | None = None + status_v6: InterfaceStatus | None = None + modem: ModemDetails | None = None + + def ipv4_online(self) -> Optional[bool]: + """Return ``True`` if IPv4 is explicitly online, ``False`` if + explicitly offline, or ``None`` if there is no information. + """ + + if self.status_v4 is None: + return None + return self.status_v4 == InterfaceStatus.ONLINE + + def ipv6_online(self) -> Optional[bool]: + """Return ``True`` if IPv6 is explicitly online, ``False`` if + explicitly offline, or ``None`` if there is no information. + """ + + if self.status_v6 is None: + return None + return self.status_v6 == InterfaceStatus.ONLINE + + def is_online(self, prefer_ipv6: bool = False) -> bool: + """Coarse "is this usable" predicate. + + The rules are conservative and intentionally simple: + + * If *prefer_ipv6* is true and there is an IPv6 state, use that. + * Otherwise, prefer IPv4 if present. + * If we only have one of them, use whichever we have. + * Any non‑zero status is treated as *not online*. + """ + + v4 = self.ipv4_online() + v6 = self.ipv6_online() + + if prefer_ipv6 and v6 is not None: + return bool(v6) + if v4 is not None: + return bool(v4) + if v6 is not None: + return bool(v6) + return False + + +@dataclass(slots=True) +class ModemDetails: + """Combined modem runtime status and hardware info.""" + + status: ModemStatusEntry | None = None + info: ModemInfo | None = None + + +@dataclass(slots=True) +class MultiWANState: + """Snapshot of the router's Multi‑WAN configuration and status.""" + + mode: MultiWANMode + """Multi‑WAN mode.""" + + interfaces: Dict[str, InterfaceInfo] + """Mapping from interface name to :class:`InterfaceInfo`.""" + + primary: str | None = None + """Interface name that is currently considered primary, if known.""" + + +class InterfaceManager: + """High‑level helper for kmwan interface status. + + Parameters + ---------- + client: + A GL.iNet API client instance that exposes ``gen_sid_payload`` and + ``_request`` helpers like :class:`gli4py.glinet.GLinet`. + """ + + def __init__(self, client: Any) -> None: + self._client = client + self._modem_manager = ModemManager(client) + + # ------------------------------------------------------------------ + # Raw fetchers + # ------------------------------------------------------------------ + async def fetch_kmwan_status(self) -> JSONDict: + """Return the raw payload from ``kmwan.get_status``. + + Typical shape:: + + { + "interfaces": [ + { + "interface": "wan", + "status_v4": 0, + "status_v6": 1, + }, + ... + ] + } + """ + + return await self._client._request( + self._client.gen_sid_payload( + "call", + ["kmwan", "get_status"], + self._client.sid, + ) + ) + + async def fetch_kmwan_config(self) -> JSONDict: + """Return the raw payload from ``kmwan.get_config``. + + Firmware‑specific details vary, but you can generally expect:: + + { + "mode": 0 | 1, + "interfaces": [ + { + "interface": "wan", + "metric": 1, + "weight": 1, + ... + }, + ... + ] + } + """ + + return await self._client._request( + self._client.gen_sid_payload( + "call", + ["kmwan", "get_config"], + self._client.sid, + ) + ) + + async def fetch_network_interface_status(self) -> JSONDict: + """Return status for a single interface via ``network.interface status``.""" + + return await self._client._request( + self._client.gen_sid_payload( + "call", + ["network.interface", "status", {"interface": "wan"}], + self._client.sid, + ) + ) + + # ------------------------------------------------------------------ + # Joined / parsed state + # ------------------------------------------------------------------ + async def get_state(self, prefer_ipv6: bool = False) -> MultiWANState: + """Fetch and join kmwan configuration + status into one object.""" + + config = await self.fetch_kmwan_config() + mwan_status = await self.fetch_kmwan_status() + # iface_status = await self.fetch_network_interface_status() + + interfaces: Dict[str, InterfaceInfo] = {} + + # First layer in config (metric / weight). + for entry in config.get("interfaces", []) or []: + name = entry.get("interface") + if not name: + continue + info = interfaces.get(name) + if info is None: + info = InterfaceInfo(name=name) + interfaces[name] = info + + info.metric = entry.get("metric") + info.weight = entry.get("weight") + + # Second layer: live status (IPv4 / IPv6 online / offline). + for entry in mwan_status.get("interfaces", []) or []: + name = entry.get("interface") + if not name: + continue + info = interfaces.get(name) + if info is None: + info = InterfaceInfo(name=name) + interfaces[name] = info + + # kmwan uses 0 = online, 1 = offline; some firmwares extend this + # to 2 = error. We keep the raw integer and interpret it in + # InterfaceInfo helper methods. + info.status_v4 = self._parse_status(entry.get("status_v4")) + info.status_v6 = self._parse_status(entry.get("status_v6")) + + mode_value = int(config.get("mode", 0)) + try: + mode = MultiWANMode(mode_value) + except ValueError: + mode = MultiWANMode.FAILOVER + primary = self._select_primary(interfaces, mode, prefer_ipv6) + + # Attach modem status entries to modem interfaces if available. + modem_interfaces = sorted( + [name for name in interfaces if name.startswith("modem")] + ) + if modem_interfaces: + modem_statuses = await self._modem_manager.get_status() + modem_infos = await self._modem_manager.get_info() + for idx, iface_name in enumerate(modem_interfaces): + modem_details = ModemDetails() + if idx < len(modem_statuses): + modem_details.status = modem_statuses[idx] + if idx < len(modem_infos): + modem_details.info = modem_infos[idx] + interfaces[iface_name].modem = modem_details + + return MultiWANState( + mode=mode, + interfaces=interfaces, + primary=primary.name if primary else None, + ) + + # ------------------------------------------------------------------ + # High‑level helpers + # ------------------------------------------------------------------ + def _select_primary( + self, + interfaces: Dict[str, InterfaceInfo], + mode: MultiWANMode, + prefer_ipv6: bool, + ) -> Optional[InterfaceInfo]: + """Return the primary interface based on kmwan mode and status.""" + + if not interfaces: + return None + + def by_metric(iters: Iterable[InterfaceInfo]) -> List[InterfaceInfo]: + def metric_key(info: InterfaceInfo) -> int: + # Higher metrics are lower priority; push "None" to the end. + return info.metric if info.metric is not None else 10_000 + + return sorted(iters, key=metric_key) + + online = [ + i for i in interfaces.values() if i.is_online(prefer_ipv6=prefer_ipv6) + ] + if not online: + return None + + if mode == MultiWANMode.FAILOVER: + return by_metric(online)[0] + + def weight_metric_key(info: InterfaceInfo) -> tuple[int, int]: + w = info.weight if info.weight is not None else 0 + m = info.metric if info.metric is not None else 10_000 + return (-w, m) + + return sorted(online, key=weight_metric_key)[0] + + @staticmethod + def _parse_status(value: Any) -> InterfaceStatus | None: + """Convert kmwan status integers to InterfaceStatus.""" + + if value is None: + return None + if isinstance(value, InterfaceStatus): + return value + if isinstance(value, bool): + return InterfaceStatus.ONLINE if value else InterfaceStatus.OFFLINE + if isinstance(value, dict): + for key in ("up", "online", "connected", "link", "state", "status"): + if key in value: + v = value[key] + if isinstance(v, bool): + return InterfaceStatus.ONLINE if v else InterfaceStatus.OFFLINE + if isinstance(v, str): + lowered = v.lower() + if lowered in {"up", "online", "connected"}: + return InterfaceStatus.ONLINE + if lowered in {"down", "offline", "disconnected"}: + return InterfaceStatus.OFFLINE + if isinstance(v, int): + try: + return InterfaceStatus(int(v)) + except ValueError: + pass + # As a fallback, treat any truthy dict with no known key as unknown. + if isinstance(value, str): + lowered = value.lower() + if lowered in {"online", "up", "connected"}: + return InterfaceStatus.ONLINE + if lowered in {"offline", "down", "disconnected"}: + return InterfaceStatus.OFFLINE + try: + return InterfaceStatus(int(value)) + except ValueError: + return None diff --git a/gli4py/modem.py b/gli4py/modem.py new file mode 100644 index 0000000..a8cd9c2 --- /dev/null +++ b/gli4py/modem.py @@ -0,0 +1,381 @@ +"""Helpers for fetching and parsing modem status.""" + +from __future__ import annotations + +from dataclasses import dataclass +from enum import IntEnum +from typing import Any, Dict, List, Optional + +JSONDict = dict[str, Any] + + +class ModemConnectionState(IntEnum): + """Connection state derived from modem network status.""" + + UNKNOWN = -1 + DISCONNECTED = 0 + CONNECTED = 1 + + +class ModemRegistrationStatus(IntEnum): + """SIM registration status.""" + + REGISTERED = 0 + UNREGISTERED = 1 + NEEDS_PIN = 2 + + +class ModemNetworkStatus(IntEnum): + """Network status for modem network block.""" + + CONNECTED = 0 + CONNECTING = 1 + + +class ModemNetworkMode(IntEnum): + """Network mode reported in signal section.""" + + GSM = 2 + UMTS = 3 + LTE = 4 + FIVE_G = 5 + LTE_ADVANCED = 41 + + @property + def label(self) -> str: + """Return a user-friendly label for the network mode.""" + + return { + ModemNetworkMode.GSM: "2G", + ModemNetworkMode.UMTS: "3G", + ModemNetworkMode.LTE: "LTE", + ModemNetworkMode.FIVE_G: "5G", + ModemNetworkMode.LTE_ADVANCED: "4G+", + }.get(self, self.name) + + +class ModemSignalStrength(IntEnum): + """Overall signal strength buckets.""" + + POOR = 1 + FAIR = 2 + GOOD = 3 + EXCELLENT = 4 + + +class ModemAutoSwitchState(IntEnum): + """SIM auto-switch state for dual-SIM devices.""" + + ENABLED = 0 + DISABLED = 1 + + +class ModemType(IntEnum): + """Modem type as reported by modem.get_info.""" + + BUILT_IN = 0 + EXTERNAL = 1 + UNSUPPORTED = 2 + + +@dataclass(slots=True) +class SimCardInfo: + """SIM card details.""" + + iccid: str | None = None + phone_number: str | None = None + mcc: str | None = None + mnc: str | None = None + + +@dataclass(slots=True) +class ModemInfo: + """Hardware info returned by modem.get_info.""" + + bus: str | None + type: ModemType | None + at_port: str | None + data_port: str | None + sms_support: bool | None + lock_tower_support: bool | None + qcfg_unsupport: bool | None + imei: str | None + name: str | None + version: str | None + vendor: str | None + protocols: list[str] | None + devices: list[str] | None + simcard: SimCardInfo | None + + +@dataclass(slots=True) +class ModemSignal: + """Signal details for a SIM.""" + + mode: ModemNetworkMode | None + strength: ModemSignalStrength | None + rssi: int | None + rsrp: int | None + rsrq: int | None + sinr: int | None + ecio: int | None + + +@dataclass(slots=True) +class ModemNetworkIP: + """IP details for a stack.""" + + ip: str | None + netmask: str | None + gateway: str | None + dns: list[str] | None + + +@dataclass(slots=True) +class ModemNetwork: + """Network status for the modem.""" + + status: ModemNetworkStatus | None + traffic_total: int | None + ipv4: ModemNetworkIP | None + ipv6: ModemNetworkIP | None + + +@dataclass(slots=True) +class ModemStatusEntry: + """Status entry returned by modem.get_status.""" + + bus: str | None + current_sim: int | None + switch_status: ModemAutoSwitchState | None + sim_status: ModemRegistrationStatus | None + sim_operator: str | None + sim_iccid: str | None + sim_phone_number: str | None + sim_mcc: str | None + sim_mnc: str | None + signal: ModemSignal | None + network: ModemNetwork | None + connection_state: ModemConnectionState + new_sms_count: int | None + passthrough: JSONDict | None + err_code: int | None + err_msg: str | None + + +class ModemManager: + """High-level helper for modem status retrieval.""" + + def __init__(self, client: Any) -> None: + self._client = client + + async def _rpc_call( + self, namespace: str, method: str, params: Optional[JSONDict] = None + ) -> JSONDict: + """Thin wrapper around the parent client's JSON-RPC helper.""" + + return await self._client._request( # type: ignore[attr-defined,no-any-return] + self._client.gen_sid_payload( # type: ignore[attr-defined] + "call", [namespace, method, params or {}], self._client.sid # type: ignore[attr-defined] + ) + ) + + async def fetch_modem_status(self) -> JSONDict: + """Return raw modem runtime status for all modems.""" + + return await self._rpc_call("modem", "get_status", {}) + + async def fetch_modem_info(self) -> JSONDict: + """Return raw modem hardware info.""" + + return await self._rpc_call("modem", "get_info", {}) + + async def get_status(self) -> list[ModemStatusEntry]: + """Fetch and parse modem status.""" + + payload = await self.fetch_modem_status() + modems = payload.get("modems", []) if isinstance(payload, dict) else [] + return [self._parse_status_entry(entry) for entry in modems] + + async def get_info(self) -> list[ModemInfo]: + """Fetch and parse modem hardware info.""" + + payload = await self.fetch_modem_info() + modems = payload.get("modems", []) if isinstance(payload, dict) else [] + return [self._parse_info(entry) for entry in modems] + + @staticmethod + @staticmethod + def _parse_info(entry: JSONDict) -> ModemInfo: + """Parse a modem entry from modem.get_info.""" + + type_value = entry.get("type") + modem_type: ModemType | None = None + try: + modem_type = ModemType(int(type_value)) + except (TypeError, ValueError): + modem_type = None + + sim_entry = entry.get("simcard") or {} + simcard = SimCardInfo( + iccid=sim_entry.get("iccid"), + phone_number=sim_entry.get("phone_number"), + mcc=sim_entry.get("mcc"), + mnc=sim_entry.get("mnc"), + ) if sim_entry else None + + devices = entry.get("devices") + if isinstance(devices, list): + devices_list: List[str] | None = [str(dev) for dev in devices] + else: + devices_list = None + + protocols = entry.get("protocols") + if isinstance(protocols, list): + protocols_list: List[str] | None = [str(proto) for proto in protocols] + else: + protocols_list = None + + return ModemInfo( + bus=entry.get("bus"), + type=modem_type, + at_port=entry.get("at_port"), + data_port=entry.get("data_port"), + sms_support=entry.get("sms_support"), + lock_tower_support=entry.get("lock_tower_support"), + qcfg_unsupport=entry.get("qcfg_unsupport"), + imei=entry.get("imei"), + name=entry.get("name"), + version=entry.get("version"), + vendor=entry.get("vendor"), + protocols=protocols_list, + devices=devices_list, + simcard=simcard, + ) + + @staticmethod + def _parse_status_entry(entry: JSONDict) -> ModemStatusEntry: + """Parse a modem entry from modem.get_status.""" + + sim_entry = entry.get("simcard") or {} + signal_entry = sim_entry.get("signal") or {} + network_entry = entry.get("network") or {} + + def _parse_signal() -> ModemSignal | None: + if not signal_entry: + return None + mode_value = signal_entry.get("mode") + try: + mode = ModemNetworkMode(int(mode_value)) if mode_value is not None else None + except ValueError: + mode = None + strength_value = signal_entry.get("strength") + try: + strength = ( + ModemSignalStrength(int(strength_value)) + if strength_value is not None + else None + ) + except ValueError: + strength = None + return ModemSignal( + mode=mode, + strength=strength, + rssi=signal_entry.get("rssi"), + rsrp=signal_entry.get("rsrp"), + rsrq=signal_entry.get("rsrq"), + sinr=signal_entry.get("sinr"), + ecio=signal_entry.get("ecio"), + ) + + def _parse_ip(ip_entry: Any) -> ModemNetworkIP | None: + if not isinstance(ip_entry, dict): + return None + dns = ip_entry.get("dns") + if isinstance(dns, list): + dns_list: list[str] | None = [str(v) for v in dns] + else: + dns_list = None + return ModemNetworkIP( + ip=ip_entry.get("ip"), + netmask=ip_entry.get("netmask"), + gateway=ip_entry.get("gateway"), + dns=dns_list, + ) + + def _parse_network() -> tuple[ModemNetwork | None, ModemConnectionState]: + if not network_entry: + return None, ModemConnectionState.UNKNOWN + status_value = network_entry.get("status") + status: ModemNetworkStatus | None = None + if isinstance(status_value, str): + lowered = status_value.lower() + if lowered == "connected": + status = ModemNetworkStatus.CONNECTED + elif lowered == "connecting": + status = ModemNetworkStatus.CONNECTING + else: + try: + status = ModemNetworkStatus(int(status_value)) + except ValueError: + status = None + elif status_value is not None: + try: + status = ModemNetworkStatus(int(status_value)) + except ValueError: + status = None + conn_state = ModemConnectionState.UNKNOWN + if status == ModemNetworkStatus.CONNECTED: + conn_state = ModemConnectionState.CONNECTED + elif status == ModemNetworkStatus.CONNECTING: + conn_state = ModemConnectionState.DISCONNECTED + return ( + ModemNetwork( + status=status, + traffic_total=network_entry.get("traffic_total"), + ipv4=_parse_ip(network_entry.get("ipv4")), + ipv6=_parse_ip(network_entry.get("ipv6")), + ), + conn_state, + ) + + sim_status_value = sim_entry.get("status") + try: + sim_status = ( + ModemRegistrationStatus(int(sim_status_value)) + if sim_status_value is not None + else None + ) + except ValueError: + sim_status = None + + network_parsed, conn_state = _parse_network() + + switch_status_value = entry.get("switch_status") + try: + switch_status = ( + ModemAutoSwitchState(int(switch_status_value)) + if switch_status_value is not None + else None + ) + except ValueError: + switch_status = None + + return ModemStatusEntry( + bus=entry.get("bus"), + current_sim=entry.get("current_sim"), + switch_status=switch_status, + sim_status=sim_status, + sim_operator=sim_entry.get("carrier"), + sim_iccid=sim_entry.get("iccid"), + sim_phone_number=sim_entry.get("phone_number"), + sim_mcc=sim_entry.get("mcc"), + sim_mnc=sim_entry.get("mnc"), + signal=_parse_signal(), + network=network_parsed, + connection_state=conn_state, + new_sms_count=entry.get("new_sms_count"), + passthrough=entry.get("passthrough"), + err_code=entry.get("err_code"), + err_msg=entry.get("err_msg"), + ) From 9f1adba7fe45d3b63e6c3159a1a0ced253873bbf Mon Sep 17 00:00:00 2001 From: Geoffrey Kruse <6468053+doggkruse@users.noreply.github.com> Date: Sun, 14 Dec 2025 11:40:24 -0400 Subject: [PATCH 6/8] Add cell tower info via modem.get_cells_info --- gli4py/modem.py | 99 +++++++++++++++++++++++++++++++++++++++++++++++-- 1 file changed, 96 insertions(+), 3 deletions(-) diff --git a/gli4py/modem.py b/gli4py/modem.py index a8cd9c2..9036f89 100644 --- a/gli4py/modem.py +++ b/gli4py/modem.py @@ -141,6 +141,27 @@ class ModemNetwork: ipv6: ModemNetworkIP | None +@dataclass(slots=True) +class CellInfo: + """Cell details from modem.get_cells_info.""" + + ul_bandwidth: str | None + dl_bandwidth: str | None + rsrp: int | None + id: str | None + rssi: int | None + tx_channel: str | None + sinr_level: int | None + rsrq_level: int | None + sinr: int | None + rsrq: int | None + rssi_level: int | None + rsrp_level: int | None + mode: str | None + band: int | None + type: str | None + + @dataclass(slots=True) class ModemStatusEntry: """Status entry returned by modem.get_status.""" @@ -156,6 +177,7 @@ class ModemStatusEntry: sim_mnc: str | None signal: ModemSignal | None network: ModemNetwork | None + cells_info: list[CellInfo] | None connection_state: ModemConnectionState new_sms_count: int | None passthrough: JSONDict | None @@ -190,12 +212,31 @@ async def fetch_modem_info(self) -> JSONDict: return await self._rpc_call("modem", "get_info", {}) + async def fetch_cells_info(self, bus: str) -> JSONDict: + """Return raw cell information for a modem.""" + + return await self._rpc_call("modem", "get_cells_info", {"bus": bus}) + async def get_status(self) -> list[ModemStatusEntry]: """Fetch and parse modem status.""" payload = await self.fetch_modem_status() modems = payload.get("modems", []) if isinstance(payload, dict) else [] - return [self._parse_status_entry(entry) for entry in modems] + + parsed_entries: list[ModemStatusEntry] = [] + for entry in modems: + if not isinstance(entry, dict): + continue + bus = entry.get("bus") + cells_info: list[CellInfo] | None = None + if bus is not None: + try: + cells_payload = await self.fetch_cells_info(str(bus)) + cells_info = self._parse_cells_info(cells_payload) + except Exception: # noqa: BLE001 + cells_info = None + parsed_entries.append(self._parse_status_entry(entry, cells_info)) + return parsed_entries async def get_info(self) -> list[ModemInfo]: """Fetch and parse modem hardware info.""" @@ -204,7 +245,12 @@ async def get_info(self) -> list[ModemInfo]: modems = payload.get("modems", []) if isinstance(payload, dict) else [] return [self._parse_info(entry) for entry in modems] - @staticmethod + async def get_cells_info(self, bus: str) -> list[CellInfo] | None: + """Fetch and return parsed cell information for a modem.""" + + payload = await self.fetch_cells_info(bus) + return self._parse_cells_info(payload) + @staticmethod def _parse_info(entry: JSONDict) -> ModemInfo: """Parse a modem entry from modem.get_info.""" @@ -254,7 +300,9 @@ def _parse_info(entry: JSONDict) -> ModemInfo: ) @staticmethod - def _parse_status_entry(entry: JSONDict) -> ModemStatusEntry: + def _parse_status_entry( + entry: JSONDict, cells_info: list[CellInfo] | None = None + ) -> ModemStatusEntry: """Parse a modem entry from modem.get_status.""" sim_entry = entry.get("simcard") or {} @@ -373,9 +421,54 @@ def _parse_network() -> tuple[ModemNetwork | None, ModemConnectionState]: sim_mnc=sim_entry.get("mnc"), signal=_parse_signal(), network=network_parsed, + cells_info=cells_info, connection_state=conn_state, new_sms_count=entry.get("new_sms_count"), passthrough=entry.get("passthrough"), err_code=entry.get("err_code"), err_msg=entry.get("err_msg"), ) + + @staticmethod + def _parse_cells_info(payload: JSONDict | None) -> list[CellInfo] | None: + """Parse cell info payload.""" + + if not isinstance(payload, dict): + return None + body: JSONDict = payload.get("result", payload) if isinstance(payload, dict) else {} + cells = body.get("cells") if isinstance(body, dict) else None + if not isinstance(cells, list): + return None + + parsed: list[CellInfo] = [] + + def _as_int(value: Any) -> int | None: + try: + return int(value) if value is not None else None + except (TypeError, ValueError): + return None + + for cell in cells: + if not isinstance(cell, dict): + continue + parsed.append( + CellInfo( + ul_bandwidth=cell.get("ul_bandwidth"), + dl_bandwidth=cell.get("dl_bandwidth"), + rsrp=_as_int(cell.get("rsrp")), + id=str(cell.get("id")) if cell.get("id") is not None else None, + rssi=_as_int(cell.get("rssi")), + tx_channel=str(cell.get("tx_channel")) if cell.get("tx_channel") is not None else None, + sinr_level=_as_int(cell.get("sinr_level")), + rsrq_level=_as_int(cell.get("rsrq_level")), + sinr=_as_int(cell.get("sinr")), + rsrq=_as_int(cell.get("rsrq")), + rssi_level=_as_int(cell.get("rssi_level")), + rsrp_level=_as_int(cell.get("rsrp_level")), + mode=cell.get("mode"), + band=_as_int(cell.get("band")), + type=cell.get("type"), + ) + ) + + return parsed or None From 6fe38d4644e1828d963e6fbc7a8313ff8fd5d24e Mon Sep 17 00:00:00 2001 From: Geoffrey Kruse <6468053+doggkruse@users.noreply.github.com> Date: Sun, 14 Dec 2025 23:05:59 -0400 Subject: [PATCH 7/8] Add 51 for 5G-NSA --- gli4py/modem.py | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/gli4py/modem.py b/gli4py/modem.py index 9036f89..f78f5e7 100644 --- a/gli4py/modem.py +++ b/gli4py/modem.py @@ -38,8 +38,9 @@ class ModemNetworkMode(IntEnum): GSM = 2 UMTS = 3 LTE = 4 - FIVE_G = 5 LTE_ADVANCED = 41 + FIVE_G = 5 + FIVE_G_NSA = 51 @property def label(self) -> str: @@ -49,8 +50,9 @@ def label(self) -> str: ModemNetworkMode.GSM: "2G", ModemNetworkMode.UMTS: "3G", ModemNetworkMode.LTE: "LTE", - ModemNetworkMode.FIVE_G: "5G", ModemNetworkMode.LTE_ADVANCED: "4G+", + ModemNetworkMode.FIVE_G: "5G", + ModemNetworkMode.FIVE_G_NSA: "5G", }.get(self, self.name) From 26bd69250cf872ac1d758b8e45bdaa8edb25f4c6 Mon Sep 17 00:00:00 2001 From: Geoffrey Kruse <6468053+doggkruse@users.noreply.github.com> Date: Fri, 16 Jan 2026 09:45:14 -0800 Subject: [PATCH 8/8] Add reboot modem command --- gli4py/modem.py | 12 ++++++++++++ 1 file changed, 12 insertions(+) diff --git a/gli4py/modem.py b/gli4py/modem.py index f78f5e7..0fe22b6 100644 --- a/gli4py/modem.py +++ b/gli4py/modem.py @@ -214,6 +214,18 @@ async def fetch_modem_info(self) -> JSONDict: return await self._rpc_call("modem", "get_info", {}) + async def reboot_modem(self, bus: str, *, hw_reboot: bool = True) -> JSONDict: + """Reboot a modem, optionally forcing hardware reboot.""" + + return await self._rpc_call( + "modem", + "reboot_modem", + { + "bus": bus, + "hw_reboot": hw_reboot, + }, + ) + async def fetch_cells_info(self, bus: str) -> JSONDict: """Return raw cell information for a modem."""