From 76e9a9c745dd9319ee40a85e24ed5b845dfcb288 Mon Sep 17 00:00:00 2001 From: yellowcooln <12516003+yellowcooln@users.noreply.github.com> Date: Thu, 20 Aug 2026 12:17:26 -0400 Subject: [PATCH] feat: migrate openHop modem sensor --- config.yaml.example | 2 +- repeater/config.py | 2 +- repeater/data_acquisition/gps_service.py | 2 +- repeater/modem_config.py | 34 +++ repeater/sensors/openhop_modem.py | 264 +++++++++++++++++++++++ repeater/sensors/pymc_modem.py | 210 +----------------- tests/test_modem_config_migration.py | 18 ++ tests/test_openhop_modem_sensor.py | 147 +++++++++++++ tests/test_sensors.py | 4 +- 9 files changed, 477 insertions(+), 206 deletions(-) create mode 100644 repeater/sensors/openhop_modem.py create mode 100644 tests/test_openhop_modem_sensor.py diff --git a/config.yaml.example b/config.yaml.example index a8471b2..95c1ce0 100644 --- a/config.yaml.example +++ b/config.yaml.example @@ -324,7 +324,7 @@ sensors: # HTTP /api/stats JSON API for modem diagnostics, including GPS fields # when the modem exposes them. If password is set to the same value as # modem_tcp.token, the repeater can use one shared modem secret. - # - type: pymc_modem + # - type: openhop_modem # name: modem # enabled: true # settings: diff --git a/repeater/config.py b/repeater/config.py index 2379f6e..9911c98 100644 --- a/repeater/config.py +++ b/repeater/config.py @@ -106,7 +106,7 @@ class NullRadio: class BaselineCrcCounterRadio: """Radio proxy that exposes CRC errors relative to repeater startup. - pyMC modem transports report the modem firmware's cumulative CRC counter. + openHop Modem transports report the modem firmware's cumulative CRC counter. The SX1262 wrapper's counter starts at process startup, which lets the engine persist deltas without knowing the radio backend. Mirror that wrapper flow here by normalizing the modem's raw counter at the transport boundary. diff --git a/repeater/data_acquisition/gps_service.py b/repeater/data_acquisition/gps_service.py index b6b8ce3..47d195a 100644 --- a/repeater/data_acquisition/gps_service.py +++ b/repeater/data_acquisition/gps_service.py @@ -283,7 +283,7 @@ class NMEAParser: self.ingest_sentence(line) def ingest_modem_payload(self, payload: Dict[str, Any]) -> bool: - """Ingest a generic pyMC modem HTTP GPS payload. + """Ingest a generic openHop Modem HTTP GPS payload. This is for modems that expose parsed GPS JSON instead of raw NMEA. The payload may be the whole /api/stats response or the nested gps dict. diff --git a/repeater/modem_config.py b/repeater/modem_config.py index 082c6ad..07f207d 100644 --- a/repeater/modem_config.py +++ b/repeater/modem_config.py @@ -13,6 +13,14 @@ LEGACY_MODEM_RADIO_TYPES = { "pymc_usb": "modem_usb", } +LEGACY_MODEM_SENSOR_TYPES = { + "pymc_modem": "openhop_modem", +} + +LEGACY_MODEM_GPS_SOURCES = { + "pymc_modem": "modem_http", +} + def _normalize_board_config(config: dict[str, Any], *, warn: bool) -> None: radio_type = config.get("radio_type") @@ -36,6 +44,31 @@ def _normalize_board_config(config: dict[str, Any], *, warn: bool) -> None: del config[legacy_key] +def _normalize_sensor_and_gps_aliases(config: dict[str, Any]) -> None: + sensors = config.get("sensors") + if isinstance(sensors, dict): + for definitions_key in ("definitions", "sensors"): + definitions = sensors.get(definitions_key) + if not isinstance(definitions, list): + continue + for definition in definitions: + if not isinstance(definition, dict): + continue + sensor_type = definition.get("type") + if isinstance(sensor_type, str): + normalized_type = sensor_type.strip().lower() + definition["type"] = LEGACY_MODEM_SENSOR_TYPES.get( + normalized_type, normalized_type + ) + + gps = config.get("gps") + if isinstance(gps, dict): + source = gps.get("source") + if isinstance(source, str): + normalized_source = source.strip().lower() + gps["source"] = LEGACY_MODEM_GPS_SOURCES.get(normalized_source, normalized_source) + + def normalize_modem_config_in_place(config: dict, *, warn: bool = True) -> dict: """Canonicalize modem transport keys in place without replacing unrelated objects.""" if not isinstance(config, dict): @@ -47,6 +80,7 @@ def normalize_modem_config_in_place(config: dict, *, warn: bool = True) -> dict: for entry in radios: if isinstance(entry, dict): _normalize_board_config(entry, warn=warn) + _normalize_sensor_and_gps_aliases(config) return config diff --git a/repeater/sensors/openhop_modem.py b/repeater/sensors/openhop_modem.py new file mode 100644 index 0000000..ea299f1 --- /dev/null +++ b/repeater/sensors/openhop_modem.py @@ -0,0 +1,264 @@ +from __future__ import annotations + +import base64 +import json +import urllib.error +import urllib.request +from typing import Any, Dict, Optional +from urllib.parse import urljoin, urlparse + +from .base import SensorBase +from .registry import SensorRegistry + + +def _single_cell_voltage_to_percent(voltage_v: float) -> int: + """Piecewise linear SoC estimate for a single Li-ion/LiPo cell.""" + if voltage_v >= 4.20: + return 100 + if voltage_v >= 4.00: + return int(85 + (voltage_v - 4.00) / 0.20 * 15) + if voltage_v >= 3.80: + return int(60 + (voltage_v - 3.80) / 0.20 * 25) + if voltage_v >= 3.70: + return int(40 + (voltage_v - 3.70) / 0.10 * 20) + if voltage_v >= 3.50: + return int(15 + (voltage_v - 3.50) / 0.20 * 25) + if voltage_v >= 3.00: + return int((voltage_v - 3.00) / 0.50 * 15) + return 0 + + +@SensorRegistry.register("openhop_modem") +class OpenHopModemSensor(SensorBase): + """Read diagnostics exposed by an openHop Modem HTTP API.""" + + sensor_type = "openhop_modem" + + def __init__(self, name: str, config: Optional[Dict[str, Any]] = None, log=None): + super().__init__(name=name, config=config, log=log) + self.poll_interval_seconds = float(self.settings.get("poll_interval_seconds", 60.0)) + self.timeout_seconds = float(self.settings.get("timeout_seconds", 2.0)) + self.endpoint = str(self.settings.get("endpoint", "/api/stats") or "/api/stats") + self.url = self._build_url() + self.username = str(self.settings.get("username", "admin") or "admin") + self.password = self.settings.get("password") + + def _build_url(self) -> str: + base_url = self.settings.get("base_url") + if base_url: + base = str(base_url).rstrip("/") + "/" + return self._validate_url(urljoin(base, self.endpoint.lstrip("/"))) + + host = str(self.settings.get("host", "") or "").strip() + if not host: + raise ValueError("openhop_modem requires settings.host or settings.base_url") + scheme = str(self.settings.get("scheme", "http") or "http").lower() + if scheme not in {"http", "https"}: + raise ValueError("openhop_modem scheme must be http or https") + port = self.settings.get("port") + netloc = host + if port not in (None, ""): + netloc = f"{host}:{int(port)}" + return self._validate_url( + f"{scheme}://{netloc}{self.endpoint if self.endpoint.startswith('/') else '/' + self.endpoint}" + ) + + @staticmethod + def _validate_url(url: str) -> str: + parsed = urlparse(url) + if parsed.scheme not in {"http", "https"}: + raise ValueError("openhop_modem URL scheme must be http or https") + if not parsed.netloc: + raise ValueError("openhop_modem URL must include a host") + return url + + def _read(self) -> Dict[str, Any]: + request = urllib.request.Request(self.url, headers={"Accept": "application/json"}) + if self.password not in (None, ""): + raw = f"{self.username}:{self.password}".encode("utf-8") + request.add_header("Authorization", "Basic " + base64.b64encode(raw).decode("ascii")) + + try: + with urllib.request.urlopen(request, timeout=self.timeout_seconds) as response: # nosec B310 + status = int(getattr(response, "status", 200) or 200) + body = response.read() + except urllib.error.HTTPError as exc: + raise RuntimeError(f"openHop Modem HTTP {exc.code} reading {self.url}") from exc + except urllib.error.URLError as exc: + raise RuntimeError(f"openHop Modem request failed: {exc.reason}") from exc + + if status < 200 or status >= 300: + raise RuntimeError(f"openHop Modem HTTP {status} reading {self.url}") + + try: + payload = json.loads(body.decode("utf-8")) + except Exception as exc: + raise RuntimeError("openHop Modem response was not valid JSON") from exc + if not isinstance(payload, dict): + raise RuntimeError("openHop Modem response was not a JSON object") + + return self._normalize_payload(payload) + + def _normalize_payload(self, payload: Dict[str, Any]) -> Dict[str, Any]: + raw_gps = payload.get("gps") + gps: Dict[str, Any] = raw_gps if isinstance(raw_gps, dict) else {} + system = self._first_dict(payload.get("system")) + counters = self._first_dict(payload.get("counters")) + radio = self._first_dict(payload.get("radio")) + network = self._first_dict(payload.get("network")) + position = self._first_dict( + gps.get("position"), + gps.get("gps_position"), + gps.get("location"), + payload.get("gps_position"), + payload.get("position"), + payload, + ) + fix = self._first_dict(gps.get("fix"), payload.get("fix")) + satellites = self._first_dict(gps.get("satellites"), payload.get("satellites")) + time_data = self._first_dict( + gps.get("time"), gps.get("time_data"), payload.get("time_data") + ) + motion = self._first_dict(gps.get("motion"), payload.get("motion")) + nmea = self._first_dict(gps.get("nmea"), payload.get("nmea")) + die_temperature_c = self._float( + system.get("die_temperature_c", payload.get("die_temperature_c")) + ) + + out: Dict[str, Any] = { + "source": "openhop_modem", + "url": self.url, + "system_board": system.get("board"), + "system_firmware": system.get("firmware"), + "system_uptime": system.get("uptime"), + "uptime_seconds": self._int(system.get("uptime_sec", payload.get("uptime_sec"))), + # The modem reports its MCU die temperature. Keep the precise field + # and provide the generic sensor convention used by RepeaterUI. + "die_temperature_c": die_temperature_c, + "temperature_c": die_temperature_c, + "rx_packets": self._int(counters.get("rx_packets", payload.get("rx_count"))), + "tx_packets": self._int(counters.get("tx_packets", payload.get("tx_count"))), + "crc_errors": self._int(counters.get("crc_errors", payload.get("crc_errors"))), + "last_rssi_dbm": self._float(counters.get("last_rssi_dbm", payload.get("last_rssi"))), + "last_snr_db": self._float(counters.get("last_snr_db", payload.get("last_snr"))), + "noise_floor_dbm": self._float( + counters.get("noise_floor_dbm", payload.get("noise_floor")) + ), + "radio_state": radio.get("state"), + "radio_standby": self._bool_or_none(radio.get("standby")), + "frequency_hz": self._int(radio.get("frequency_hz")), + "frequency_mhz": self._float(radio.get("frequency_mhz")), + "bandwidth_hz": self._int(radio.get("bandwidth_hz")), + "bandwidth_khz": self._float(radio.get("bandwidth_khz")), + "spreading_factor": self._int(radio.get("spreading_factor")), + "coding_rate": self._int(radio.get("coding_rate")), + "tx_power_dbm": self._int(radio.get("tx_power_dbm")), + "preamble_length": self._int(radio.get("preamble_len")), + "syncword": radio.get("syncword"), + "syncword_value": self._int(radio.get("syncword_value")), + "auto_cad_enabled": self._bool_or_none(radio.get("auto_cad_enabled")), + "network_interface": network.get("interface"), + "network_mode": network.get("mode"), + "network_live": self._bool_or_none(network.get("live")), + "network_current_ip": network.get("current_ip"), + "tcp_port": self._int(network.get("tcp_port")), + "token_configured": self._bool_or_none(network.get("pymc_token_set")), + "gps_available": self._bool_or_none(gps.get("available")), + "gps_enabled": self._bool_or_none(gps.get("enabled")), + "gps_seen": self._bool_or_none(gps.get("seen")), + "latitude": self._float(position.get("latitude")), + "longitude": self._float(position.get("longitude")), + "altitude_m": self._float(position.get("altitude_m")), + "fix_valid": self._bool_or_none(fix.get("valid")), + "fix_quality": self._int(fix.get("quality")), + "satellites_used": self._int( + satellites.get("used_count", satellites.get("satellites_used")) + ), + "satellites_in_view": self._int( + satellites.get("in_view_count", satellites.get("satellites_in_view")) + ), + "datetime_utc": time_data.get("datetime_utc") or payload.get("datetime_utc"), + "speed_kmh": self._float(motion.get("speed_kmh", payload.get("speed_kmh"))), + "course_degrees": self._float( + motion.get("course_degrees", payload.get("course_degrees")) + ), + "gps_nmea_age_ms": self._int(nmea.get("age_ms")), + "gps_nmea_valid_sentences": self._int(nmea.get("valid_sentence_count")), + "gps_nmea_invalid_checksums": self._int(nmea.get("invalid_checksum_count")), + "gps_nmea_raw_bytes": self._int(nmea.get("raw_byte_count")), + "gps_nmea_last_sentence": nmea.get("last_sentence_type"), + } + + for key in ( + "battery_voltage_mv", + "battery_voltage_v", + "battery_percent", + "battery_percentage", + "solar_charge_rate_percent_per_hour", + ): + if key in payload: + out[key] = payload[key] + + for key in ("battery_voltage_mv", "battery_voltage_v"): + if out.get(key) is None and system.get(key) is not None: + out[key] = system[key] + + if "battery_percent" not in out: + battery_voltage_v = self._float(out.get("battery_voltage_v")) + if battery_voltage_v is None: + battery_voltage_mv = self._float(out.get("battery_voltage_mv")) + if battery_voltage_mv is not None: + battery_voltage_v = battery_voltage_mv / 1000.0 + if battery_voltage_v is not None: + out["battery_percent"] = _single_cell_voltage_to_percent(battery_voltage_v) + + return {key: value for key, value in out.items() if value is not None} + + def _battery_voltage_v(self, payload: Dict[str, Any]) -> Optional[float]: + voltage_v = self._float(payload.get("battery_voltage_v")) + if voltage_v is not None: + return voltage_v + voltage_mv = self._float(payload.get("battery_voltage_mv")) + if voltage_mv is not None: + return voltage_mv / 1000.0 + return None + + @staticmethod + def _first_dict(*values: Any) -> Dict[str, Any]: + for value in values: + if isinstance(value, dict): + return value + return {} + + @staticmethod + def _float(value: Any) -> Optional[float]: + if value in (None, ""): + return None + try: + result = float(value) + except (TypeError, ValueError): + return None + return result + + @staticmethod + def _int(value: Any) -> Optional[int]: + if value in (None, ""): + return None + try: + return int(value) + except (TypeError, ValueError): + return None + + @staticmethod + def _bool_or_none(value: Any) -> Optional[bool]: + if value is None: + return None + if isinstance(value, bool): + return value + if isinstance(value, str): + lowered = value.strip().lower() + if lowered in {"1", "true", "yes", "on"}: + return True + if lowered in {"0", "false", "no", "off"}: + return False + return bool(value) diff --git a/repeater/sensors/pymc_modem.py b/repeater/sensors/pymc_modem.py index 39f3b4a..6330cab 100644 --- a/repeater/sensors/pymc_modem.py +++ b/repeater/sensors/pymc_modem.py @@ -1,207 +1,15 @@ -from __future__ import annotations +"""Compatibility alias for the renamed openHop Modem sensor.""" -import base64 -import json -import urllib.error -import urllib.request -from typing import Any, Dict, Optional -from urllib.parse import urljoin, urlparse - -from .base import SensorBase +from . import openhop_modem as _implementation +from .openhop_modem import OpenHopModemSensor from .registry import SensorRegistry +# Keep the old module-level urllib seam used by downstream tests/integrations. +urllib = _implementation.urllib -def _single_cell_voltage_to_percent(voltage_v: float) -> int: - """Piecewise linear SoC estimate for a single Li-ion/LiPo cell.""" - if voltage_v >= 4.20: - return 100 - if voltage_v >= 4.00: - return int(85 + (voltage_v - 4.00) / 0.20 * 15) - if voltage_v >= 3.80: - return int(60 + (voltage_v - 3.80) / 0.20 * 25) - if voltage_v >= 3.70: - return int(40 + (voltage_v - 3.70) / 0.10 * 20) - if voltage_v >= 3.50: - return int(15 + (voltage_v - 3.50) / 0.20 * 25) - if voltage_v >= 3.00: - return int((voltage_v - 3.00) / 0.50 * 15) - return 0 +SensorRegistry.register("pymc_modem", OpenHopModemSensor) +# Preserve the old import path/class name for existing integrations. +PymcModemSensor = OpenHopModemSensor -@SensorRegistry.register("pymc_modem") -class PymcModemSensor(SensorBase): - """Read diagnostics exposed by a pyMC modem HTTP API.""" - - sensor_type = "pymc_modem" - - def __init__(self, name: str, config: Optional[Dict[str, Any]] = None, log=None): - super().__init__(name=name, config=config, log=log) - self.poll_interval_seconds = float(self.settings.get("poll_interval_seconds", 60.0)) - self.timeout_seconds = float(self.settings.get("timeout_seconds", 2.0)) - self.endpoint = str(self.settings.get("endpoint", "/api/stats") or "/api/stats") - self.url = self._build_url() - self.username = str(self.settings.get("username", "admin") or "admin") - self.password = self.settings.get("password") - - def _build_url(self) -> str: - base_url = self.settings.get("base_url") - if base_url: - base = str(base_url).rstrip("/") + "/" - return self._validate_url(urljoin(base, self.endpoint.lstrip("/"))) - - host = str(self.settings.get("host", "") or "").strip() - if not host: - raise ValueError("pymc_modem requires settings.host or settings.base_url") - scheme = str(self.settings.get("scheme", "http") or "http").lower() - if scheme not in {"http", "https"}: - raise ValueError("pymc_modem scheme must be http or https") - port = self.settings.get("port") - netloc = host - if port not in (None, ""): - netloc = f"{host}:{int(port)}" - return self._validate_url( - f"{scheme}://{netloc}{self.endpoint if self.endpoint.startswith('/') else '/' + self.endpoint}" - ) - - @staticmethod - def _validate_url(url: str) -> str: - parsed = urlparse(url) - if parsed.scheme not in {"http", "https"}: - raise ValueError("pymc_modem URL scheme must be http or https") - if not parsed.netloc: - raise ValueError("pymc_modem URL must include a host") - return url - - def _read(self) -> Dict[str, Any]: - request = urllib.request.Request(self.url, headers={"Accept": "application/json"}) - if self.password not in (None, ""): - raw = f"{self.username}:{self.password}".encode("utf-8") - request.add_header("Authorization", "Basic " + base64.b64encode(raw).decode("ascii")) - - try: - with urllib.request.urlopen(request, timeout=self.timeout_seconds) as response: # nosec B310 - status = int(getattr(response, "status", 200) or 200) - body = response.read() - except urllib.error.HTTPError as exc: - raise RuntimeError(f"pyMC modem HTTP {exc.code} reading {self.url}") from exc - except urllib.error.URLError as exc: - raise RuntimeError(f"pyMC modem request failed: {exc.reason}") from exc - - if status < 200 or status >= 300: - raise RuntimeError(f"pyMC modem HTTP {status} reading {self.url}") - - try: - payload = json.loads(body.decode("utf-8")) - except Exception as exc: - raise RuntimeError("pyMC modem response was not valid JSON") from exc - if not isinstance(payload, dict): - raise RuntimeError("pyMC modem response was not a JSON object") - - return self._normalize_payload(payload) - - def _normalize_payload(self, payload: Dict[str, Any]) -> Dict[str, Any]: - raw_gps = payload.get("gps") - gps: Dict[str, Any] = raw_gps if isinstance(raw_gps, dict) else {} - position = self._first_dict( - gps.get("position"), - gps.get("gps_position"), - gps.get("location"), - payload.get("gps_position"), - payload.get("position"), - payload, - ) - fix = self._first_dict(gps.get("fix"), payload.get("fix")) - satellites = self._first_dict(gps.get("satellites"), payload.get("satellites")) - time_data = self._first_dict( - gps.get("time"), gps.get("time_data"), payload.get("time_data") - ) - motion = self._first_dict(gps.get("motion"), payload.get("motion")) - - out: Dict[str, Any] = { - "source": "pymc_modem", - "url": self.url, - "gps_enabled": self._bool_or_none(gps.get("enabled")), - "gps_seen": self._bool_or_none(gps.get("seen")), - "latitude": self._float(position.get("latitude")), - "longitude": self._float(position.get("longitude")), - "altitude_m": self._float(position.get("altitude_m")), - "fix_valid": self._bool_or_none(fix.get("valid")), - "fix_quality": self._int(fix.get("quality")), - "satellites_used": self._int( - satellites.get("used_count", satellites.get("satellites_used")) - ), - "satellites_in_view": self._int( - satellites.get("in_view_count", satellites.get("satellites_in_view")) - ), - "datetime_utc": time_data.get("datetime_utc") or payload.get("datetime_utc"), - "speed_kmh": self._float(motion.get("speed_kmh", payload.get("speed_kmh"))), - "course_degrees": self._float( - motion.get("course_degrees", payload.get("course_degrees")) - ), - } - - for key in ( - "battery_voltage_mv", - "battery_voltage_v", - "battery_percent", - "battery_percentage", - "solar_charge_rate_percent_per_hour", - ): - if key in payload: - out[key] = payload[key] - - if "battery_percent" not in out: - battery_voltage_v = self._battery_voltage_v(payload) - if battery_voltage_v is not None: - out["battery_percent"] = _single_cell_voltage_to_percent(battery_voltage_v) - - return {key: value for key, value in out.items() if value is not None} - - def _battery_voltage_v(self, payload: Dict[str, Any]) -> Optional[float]: - voltage_v = self._float(payload.get("battery_voltage_v")) - if voltage_v is not None: - return voltage_v - voltage_mv = self._float(payload.get("battery_voltage_mv")) - if voltage_mv is not None: - return voltage_mv / 1000.0 - return None - - @staticmethod - def _first_dict(*values: Any) -> Dict[str, Any]: - for value in values: - if isinstance(value, dict): - return value - return {} - - @staticmethod - def _float(value: Any) -> Optional[float]: - if value in (None, ""): - return None - try: - result = float(value) - except (TypeError, ValueError): - return None - return result - - @staticmethod - def _int(value: Any) -> Optional[int]: - if value in (None, ""): - return None - try: - return int(value) - except (TypeError, ValueError): - return None - - @staticmethod - def _bool_or_none(value: Any) -> Optional[bool]: - if value is None: - return None - if isinstance(value, bool): - return value - if isinstance(value, str): - lowered = value.strip().lower() - if lowered in {"1", "true", "yes", "on"}: - return True - if lowered in {"0", "false", "no", "off"}: - return False - return bool(value) +__all__ = ["PymcModemSensor"] diff --git a/tests/test_modem_config_migration.py b/tests/test_modem_config_migration.py index 69f2616..570c35f 100644 --- a/tests/test_modem_config_migration.py +++ b/tests/test_modem_config_migration.py @@ -152,3 +152,21 @@ def test_save_config_serializes_only_canonical_modem_keys(tmp_path): "radio_type": "modem_usb", "modem_usb": {"port": "/dev/ttyACM0"}, } + + +def test_sensor_and_gps_aliases_migrate_to_canonical_names(): + normalized = normalize_modem_config( + { + "sensors": { + "definitions": [ + {"name": "modem", "type": "pymc_modem", "settings": {"host": "x"}}, + {"name": "system", "type": "hardware_stats"}, + ] + }, + "gps": {"source": "pymc_modem"}, + } + ) + + assert normalized["sensors"]["definitions"][0]["type"] == "openhop_modem" + assert normalized["sensors"]["definitions"][1]["type"] == "hardware_stats" + assert normalized["gps"]["source"] == "modem_http" diff --git a/tests/test_openhop_modem_sensor.py b/tests/test_openhop_modem_sensor.py new file mode 100644 index 0000000..1132b3b --- /dev/null +++ b/tests/test_openhop_modem_sensor.py @@ -0,0 +1,147 @@ +import json + +from repeater.sensors.openhop_modem import OpenHopModemSensor +from repeater.sensors.pymc_modem import PymcModemSensor +from repeater.sensors.registry import SensorRegistry + + +class _Response: + status = 200 + + def __init__(self, payload): + self.payload = payload + + def __enter__(self): + return self + + def __exit__(self, *_args): + return False + + def read(self): + return json.dumps(self.payload).encode() + + +def test_openhop_modem_sensor_exposes_full_flat_diagnostics(monkeypatch): + payload = { + "battery_voltage_mv": 4112, + "battery_voltage_v": 4.112, + "solar_charge_rate_percent_per_hour": 4.5, + "system": { + "board": "Photon", + "firmware": "1.2.3", + "uptime": "1d 2h", + "uptime_sec": 93600, + "die_temperature_c": 54, + }, + "counters": { + "rx_packets": 101, + "tx_packets": 22, + "crc_errors": 3, + "last_rssi_dbm": -91, + "last_snr_db": 7.5, + "noise_floor_dbm": -116.2, + }, + "radio": { + "state": "rx", + "standby": False, + "frequency_hz": 910525000, + "frequency_mhz": 910.525, + "bandwidth_hz": 62500, + "bandwidth_khz": 62.5, + "spreading_factor": 7, + "coding_rate": 5, + "tx_power_dbm": 18, + "preamble_len": 32, + "syncword": "0x12", + "syncword_value": 18, + "auto_cad_enabled": True, + }, + "network": { + "interface": "ethernet", + "mode": "static", + "live": True, + "current_ip": "192.0.2.10", + "tcp_port": 5055, + "pymc_token_set": True, + }, + "gps": { + "available": True, + "enabled": True, + "seen": True, + "fix": {"valid": True, "quality": 1}, + "position": {"latitude": 42.3, "longitude": -71.0, "altitude_m": 10.5}, + "satellites": {"used_count": 8, "in_view_count": 12}, + "time": {"datetime_utc": "2026-08-20T14:00:00Z"}, + "motion": {"speed_kmh": 1.2, "course_degrees": 180.0}, + "nmea": { + "age_ms": 250, + "valid_sentence_count": 400, + "invalid_checksum_count": 2, + "raw_byte_count": 12000, + "last_sentence_type": "GGA", + }, + }, + } + monkeypatch.setattr( + "repeater.sensors.openhop_modem.urllib.request.urlopen", + lambda *_args, **_kwargs: _Response(payload), + ) + + reading = OpenHopModemSensor( + "modem", {"settings": {"base_url": "http://openhop-modem.local"}} + ).read() + + assert reading["ok"] is True + data = reading["data"] + assert data["source"] == "openhop_modem" + assert data["die_temperature_c"] == 54 + assert data["temperature_c"] == 54 + assert data["system_board"] == "Photon" + assert data["system_firmware"] == "1.2.3" + assert data["system_uptime"] == "1d 2h" + assert data["uptime_seconds"] == 93600 + assert data["rx_packets"] == 101 + assert data["tx_packets"] == 22 + assert data["crc_errors"] == 3 + assert data["last_rssi_dbm"] == -91 + assert data["last_snr_db"] == 7.5 + assert data["noise_floor_dbm"] == -116.2 + assert data["radio_state"] == "rx" + assert data["radio_standby"] is False + assert data["frequency_hz"] == 910525000 + assert data["bandwidth_hz"] == 62500 + assert data["spreading_factor"] == 7 + assert data["coding_rate"] == 5 + assert data["tx_power_dbm"] == 18 + assert data["preamble_length"] == 32 + assert data["auto_cad_enabled"] is True + assert data["network_interface"] == "ethernet" + assert data["network_live"] is True + assert data["tcp_port"] == 5055 + assert data["token_configured"] is True + assert data["gps_available"] is True + assert data["gps_nmea_age_ms"] == 250 + assert data["gps_nmea_valid_sentences"] == 400 + assert data["gps_nmea_invalid_checksums"] == 2 + assert data["gps_nmea_raw_bytes"] == 12000 + assert data["gps_nmea_last_sentence"] == "GGA" + assert data["battery_voltage_mv"] == 4112 + assert data["solar_charge_rate_percent_per_hour"] == 4.5 + + +def test_legacy_pymc_modem_module_and_registry_create_canonical_sensor(): + assert PymcModemSensor is OpenHopModemSensor + legacy = SensorRegistry.create( + "pymc_modem", + name="legacy", + config={"settings": {"base_url": "http://legacy-modem.local"}}, + ) + canonical = SensorRegistry.create( + "openhop_modem", + name="canonical", + config={"settings": {"base_url": "http://openhop-modem.local"}}, + ) + + assert isinstance(legacy, OpenHopModemSensor) + assert legacy.sensor_type == "openhop_modem" + assert canonical.sensor_type == "openhop_modem" diff --git a/tests/test_sensors.py b/tests/test_sensors.py index fa51682..5607180 100644 --- a/tests/test_sensors.py +++ b/tests/test_sensors.py @@ -234,7 +234,7 @@ def test_pymc_modem_sensor_reads_modem_stats(monkeypatch): assert captured["url"] == "http://192.168.0.205/api/stats" assert captured["auth"].startswith("Basic ") assert captured["timeout"] == 3.5 - assert reading["data"]["source"] == "pymc_modem" + assert reading["data"]["source"] == "openhop_modem" assert reading["data"]["latitude"] == 42.360082 assert reading["data"]["longitude"] == -71.05888 assert reading["data"]["altitude_m"] == 12.5 @@ -281,7 +281,7 @@ def test_pymc_modem_sensor_accepts_stats_without_gps_coordinates(monkeypatch): ).read() assert reading["ok"] is True - assert reading["data"]["source"] == "pymc_modem" + assert reading["data"]["source"] == "openhop_modem" assert reading["data"]["battery_voltage_mv"] == 3681 assert reading["data"]["battery_voltage_v"] == 3.681 assert reading["data"]["battery_percent"] == 37