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, size=-1): body = json.dumps(self.payload).encode() return body if size < 0 else body[:size] 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, "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_modem_sensor_defaults_to_http_port_not_radio_tcp_port(): defaults = {field["key"]: field["default"] for field in OpenHopModemSensor._settings_schema} assert defaults["port"] == 80 defaults["host"] = "modem.local" sensor = OpenHopModemSensor("modem", {"settings": defaults}) assert sensor.url == "http://modem.local:80/api/stats" explicit = OpenHopModemSensor("other", {"settings": {"host": "modem.local", "port": 8080}}) assert explicit.url == "http://modem.local:8080/api/stats" def test_openhop_modem_sensor_accepts_legacy_token_flag(): sensor = OpenHopModemSensor("modem", {"settings": {"base_url": "http://openhop-modem.local"}}) data = sensor._normalize_payload({"network": {"pymc_token_set": True}}) assert data["token_configured"] is True def test_openhop_modem_sensor_prefers_top_level_power_telemetry(): sensor = OpenHopModemSensor("modem", {"settings": {"base_url": "http://modem.local"}}) data = sensor._normalize_payload( { "bus_voltage_v": 12.46, "current_ma": 59.7, "power_mw": 744.0, "system": { "station_g3_input_voltage_v": 11.0, "station_g3_current_ma": 50.0, "station_g3_power_w": 0.55, }, } ) assert data["bus_voltage_v"] == 12.46 assert data["current_ma"] == 59.7 assert data["power_mw"] == 744.0 def test_openhop_modem_sensor_ignores_legacy_station_g3_power_fields(): sensor = OpenHopModemSensor("modem", {"settings": {"base_url": "http://modem.local"}}) data = sensor._normalize_payload( { "system": { "station_g3_input_voltage_v": "12.46", "station_g3_current_ma": "-59.7", "station_g3_power_w": "-0.744", }, } ) assert "bus_voltage_v" not in data assert "current_ma" not in data assert "power_mw" not in data def test_openhop_modem_sensor_omits_invalid_power_telemetry(): sensor = OpenHopModemSensor("modem", {"settings": {"base_url": "http://modem.local"}}) data = sensor._normalize_payload( { "bus_voltage_v": "invalid", "current_ma": "invalid", "power_mw": "invalid", "system": { "station_g3_input_voltage_v": None, "station_g3_current_ma": None, "station_g3_power_w": None, }, } ) assert "bus_voltage_v" not in data assert "current_ma" not in data assert "power_mw" not in data 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"