From b4d7ab6b0b32759f9834df9a02c149fb828743de Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Oliver=20Hennh=C3=B6fer?= Date: Fri, 13 Nov 2026 20:51:15 +0100 Subject: [PATCH] Replace inactive demo feeds with verified public telemetry --- .../core/src/off_key_core/demo/catalogue.py | 27 +++-- .../core/src/off_key_core/demo/sources.py | 22 ++-- .../src/off_key_mqtt_proxy/demo/normalize.py | 40 +++---- .../src/off_key_mqtt_proxy/demo/probe.py | 22 ++-- .../src/off_key_mqtt_proxy/demo/relay.py | 10 +- backend/tests/demo/test_sources.py | 112 ++++++++++++++---- docs/operations/public-demo.md | 40 ++++--- .../inventories/demo/group_vars/all/vars.yml | 3 +- 8 files changed, 182 insertions(+), 94 deletions(-) diff --git a/backend/libs/core/src/off_key_core/demo/catalogue.py b/backend/libs/core/src/off_key_core/demo/catalogue.py index d69b4cba..be8d65ea 100644 --- a/backend/libs/core/src/off_key_core/demo/catalogue.py +++ b/backend/libs/core/src/off_key_core/demo/catalogue.py @@ -18,18 +18,15 @@ ) from .sources import SOURCES -Identifier = Annotated[str, Field(pattern=r"^[A-Za-z0-9_-]{1,60}$")] - class FeedSelection(BaseModel): model_config = ConfigDict(extra="forbid", frozen=True) - weather_station: int = Field(default=4057, gt=0, lt=1_000_000) + weather_station: int = Field(default=1003, gt=0, lt=1_000_000) traffic_station: int = Field(default=23438, gt=0, lt=1_000_000) hamburg_detectors: list[Annotated[int, Field(gt=0)]] = Field( default_factory=lambda: [13, 30], min_length=1, max_length=3 ) - solar_inverter: Identifier = "INV_001_001" @field_validator("hamburg_detectors") @classmethod @@ -39,8 +36,7 @@ def distinct_detectors(cls, values: list[int]) -> list[int]: return values -# Digitraffic road sensor ids by normalized field. Both networks report every -# station around the clock, about once a minute. +# Digitraffic road sensor ids by normalized field. Cadence varies by station. WEATHER_SENSORS = { "air_temperature": 1, "road_temperature": 3, @@ -53,6 +49,13 @@ def distinct_detectors(cls, values: list[int]) -> list[int]: "speed_1": 5122, "speed_2": 5125, } +MOSQUITTO_ENTITY = "test-mosquitto-org" +MOSQUITTO_SENSORS = { + "clients": "$SYS/broker/clients/connected", + "messages_received": "$SYS/broker/load/messages/received/1min", + "messages_sent": "$SYS/broker/load/messages/sent/1min", + "bytes_received": "$SYS/broker/load/bytes/received/1min", +} # (field in normalized message, visible label, unit) FIELDS = { @@ -69,11 +72,11 @@ def distinct_detectors(cls, values: list[int]) -> list[int]: ("speed_2", "Average speed, direction 2", "km/h"), ], "hamburg": [("occupancy", "Occupancy", "%")], - "solar": [ - ("power", "AC power", "kW"), - ("voltage", "DC voltage", "V"), - ("temperature", "Module temperature", "°C"), - ("efficiency", "Efficiency", "%"), + "mosquitto": [ + ("clients", "Connected clients", "clients"), + ("messages_received", "Messages received, 1-minute average", "messages/min"), + ("messages_sent", "Messages sent, 1-minute average", "messages/min"), + ("bytes_received", "Bytes received, 1-minute average", "B/min"), ], } @@ -83,7 +86,7 @@ def entities(selection: FeedSelection) -> dict[str, list[str]]: "weather": [str(selection.weather_station)], "traffic": [str(selection.traffic_station)], "hamburg": [str(value) for value in selection.hamburg_detectors], - "solar": [selection.solar_inverter], + "mosquitto": [MOSQUITTO_ENTITY], } diff --git a/backend/libs/core/src/off_key_core/demo/sources.py b/backend/libs/core/src/off_key_core/demo/sources.py index c753f910..d93e03b2 100644 --- a/backend/libs/core/src/off_key_core/demo/sources.py +++ b/backend/libs/core/src/off_key_core/demo/sources.py @@ -24,7 +24,7 @@ class Source: "Fintraffic / Digitraffic", "CC BY 4.0", CC_BY, - "One roadside weather station, reporting about once a minute.", + "One roadside weather station. Updates can be several minutes apart.", ), Source( "traffic", @@ -43,18 +43,18 @@ class Source: "Freie und Hansestadt Hamburg", "DL-DE BY 2.0", "https://www.govdata.de/dl-de/by-2-0", - "Event-driven observations; quiet periods are expected.", + "Vehicle detector occupancy in percent. Brief spikes and zeros are " + "normal; observations arrive when the detector changes.", ), Source( - "solar", - "TDengine solar simulation", - "https://tdengine.com/free-public-mqtt-data-feed-real-time-solar-fleet-simulation-now-available/", - "TDengine", - "Provider's public feed terms", - "https://tdengine.com/free-public-mqtt-data-feed-real-time-solar-fleet-simulation-now-available/", - "Simulated industrial telemetry. " - "Availability and cadence depend on the provider.", - True, + "mosquitto", + "Mosquitto public broker metrics", + "https://test.mosquitto.org/", + "Eclipse Mosquitto public test broker", + "Public test service terms", + "https://test.mosquitto.org/", + "Real client counts and message traffic from a public test broker. " + "Fresh updates are timestamped on arrival; service interruptions are possible.", ), ) diff --git a/backend/services/mqtt/proxy/src/off_key_mqtt_proxy/demo/normalize.py b/backend/services/mqtt/proxy/src/off_key_mqtt_proxy/demo/normalize.py index ede0af08..133ffd71 100644 --- a/backend/services/mqtt/proxy/src/off_key_mqtt_proxy/demo/normalize.py +++ b/backend/services/mqtt/proxy/src/off_key_mqtt_proxy/demo/normalize.py @@ -5,6 +5,8 @@ from datetime import datetime from off_key_core.demo.catalogue import ( + MOSQUITTO_ENTITY, + MOSQUITTO_SENSORS, TRAFFIC_SENSORS, WEATHER_SENSORS, FeedSelection, @@ -56,34 +58,23 @@ def _hamburg(mqtt_topic: str, data: dict, selection: FeedSelection): return entity, data["phenomenonTime"], {"occupancy": data.get("result")} -def _solar(mqtt_topic: str, data: dict, selection: FeedSelection): - entity = data["Inverter_ID"] - if entity != selection.solar_inverter: +def _mosquitto(mqtt_topic: str, data, _selection: FeedSelection): + field = {value: key for key, value in MOSQUITTO_SENSORS.items()}.get(mqtt_topic) + if field is None or type(data) not in (int, float) or data < 0: return None - return ( - entity, - data["ts"], - { - name: data.get(key) - for name, key in { - "power": "Output_AC_Power_kW", - "voltage": "Input_DC_Voltage", - "temperature": "Temperature_Module", - "efficiency": "Efficiency_%", - }.items() - }, - ) + # $SYS values have no observation timestamp. Never invent one for the source. + return MOSQUITTO_ENTITY, None, {field: data} _DECODERS = { "weather": _weather, "traffic": _traffic, "hamburg": _hamburg, - "solar": _solar, + "mosquitto": _mosquitto, } -def decode(source: str, mqtt_topic: str, data: dict, selection: FeedSelection): +def decode(source: str, mqtt_topic: str, data, selection: FeedSelection): return _DECODERS[source](mqtt_topic, data, selection) @@ -110,15 +101,16 @@ def messages( return [] try: data = json.loads(payload) - if not isinstance(data, dict): + if source != "mosquitto" and not isinstance(data, dict): return [] decoded = decode(source, mqtt_topic, data, self.selection) if decoded is None: return [] entity, source_time, fields = decoded - observed_at = timestamp(source_time) + observed_at = None if source == "mosquitto" else timestamp(source_time) + event_time = received_at if observed_at is None else observed_at max_age = MAX_AGE.get(source, DEFAULT_MAX_AGE) - if not received_at - max_age <= observed_at <= received_at + 10: + if not received_at - max_age <= event_time <= received_at + 10: return [] except ( ValueError, @@ -132,7 +124,7 @@ def messages( result = [] for field, raw in fields.items(): destination = topic(source, entity, field) - if destination not in self.allowed or observed_at <= self.latest.get( + if destination not in self.allowed or event_time <= self.latest.get( destination, 0 ): continue @@ -142,7 +134,7 @@ def messages( continue if not math.isfinite(value): continue - self.latest[destination] = observed_at + self.latest[destination] = event_time result.append( ( destination, @@ -153,7 +145,7 @@ def messages( "entity": entity, "observed_at": observed_at, "received_at": received_at, - "simulated": source == "solar", + "simulated": False, }, allow_nan=False, ), diff --git a/backend/services/mqtt/proxy/src/off_key_mqtt_proxy/demo/probe.py b/backend/services/mqtt/proxy/src/off_key_mqtt_proxy/demo/probe.py index e0826f0c..6cabffd3 100644 --- a/backend/services/mqtt/proxy/src/off_key_mqtt_proxy/demo/probe.py +++ b/backend/services/mqtt/proxy/src/off_key_mqtt_proxy/demo/probe.py @@ -22,8 +22,6 @@ def candidate(source: str, mqtt_topic: str, data: dict) -> dict: if source in ("weather", "traffic"): return {f"{source}_station": int(mqtt_topic.split("/")[1])} - if source == "solar": - return {"solar_inverter": data["Inverter_ID"]} return {} @@ -69,10 +67,11 @@ def probe(selection: FeedSelection, seconds: int, discover: bool) -> FeedSelecti counts: dict[str, int] = {} clients = [] discovery = { - # One sensor per network, so the first station heard is one that reports. - "weather": ("weather-v2/+/1",), - "traffic": ("tms-v2/+/5122",), - "solar": ("inverters",), + # Use Digitraffic's documented network subscription. The '+' filters + # do not deliver measurements on this service. Narrow after the first + # valid observation, retaining only a bounded selected catalogue. + "weather": ("weather-v2/#",), + "traffic": ("tms-v2/#",), } try: @@ -80,7 +79,16 @@ def probe(selection: FeedSelection, seconds: int, discover: bool) -> FeedSelecti if discover and feed.id in discovery: feed = replace(feed, topics=discovery[feed.id]) clients.append( - subscriber(feed, _receiver(feed, selection, discover, chosen, counts)) + subscriber( + feed, + _receiver( + feed, + selection, + discover and feed.id in discovery, + chosen, + counts, + ), + ) ) threading.Event().wait(seconds) finally: diff --git a/backend/services/mqtt/proxy/src/off_key_mqtt_proxy/demo/relay.py b/backend/services/mqtt/proxy/src/off_key_mqtt_proxy/demo/relay.py index a7c02335..60a30d22 100644 --- a/backend/services/mqtt/proxy/src/off_key_mqtt_proxy/demo/relay.py +++ b/backend/services/mqtt/proxy/src/off_key_mqtt_proxy/demo/relay.py @@ -16,6 +16,7 @@ from uuid import uuid4 from off_key_core.demo.catalogue import ( + MOSQUITTO_SENSORS, TRAFFIC_SENSORS, WEATHER_SENSORS, FeedSelection, @@ -72,7 +73,14 @@ def feeds(selected: FeedSelection) -> tuple[Feed, ...]: for value in selected.hamburg_detectors ), ), - Feed("solar", "mqtt.tdengine.com", ("inverters",), 1883, None, False), + Feed( + "mosquitto", + "test.mosquitto.org", + tuple(MOSQUITTO_SENSORS.values()), + 8886, + None, + True, + ), ) diff --git a/backend/tests/demo/test_sources.py b/backend/tests/demo/test_sources.py index 50a38e62..3df5cb5a 100644 --- a/backend/tests/demo/test_sources.py +++ b/backend/tests/demo/test_sources.py @@ -2,11 +2,14 @@ import json from datetime import UTC, datetime +from types import SimpleNamespace +from unittest.mock import Mock import pytest -from off_key_core.demo.catalogue import FeedSelection, catalogue +from off_key_core.demo.catalogue import MOSQUITTO_SENSORS, FeedSelection, catalogue from off_key_core.demo.sources import public_sources from off_key_core.schemas.system import SystemDocument +from off_key_mqtt_proxy.demo import probe as live_probe from off_key_mqtt_proxy.demo.normalize import DEFAULT_MAX_AGE, MAX_AGE, Normalizer from off_key_mqtt_proxy.demo.relay import feeds @@ -16,7 +19,7 @@ MESSAGES = [ ( "weather", - "weather-v2/4057/3", + "weather-v2/1003/3", {"value": -1.4, "time": int(NOW)}, 1, ), @@ -32,19 +35,6 @@ {"phenomenonTime": ISO, "result": 35}, 1, ), - ( - "solar", - "inverters", - { - "ts": ISO, - "Inverter_ID": "INV_001_001", - "Output_AC_Power_kW": 2000, - "Input_DC_Voltage": 850, - "Temperature_Module": 45, - "Efficiency_%": 97.5, - }, - 4, - ), ] @@ -64,7 +54,7 @@ def test_sources_preserve_event_and_arrival_time_and_reject_duplicates( message = json.loads(encoded) assert message["observed_at"] == NOW assert message["received_at"] == NOW + 1 - assert message["simulated"] == (source == "solar") + assert message["simulated"] is False assert normalizer.messages(source, topic, wire, received_at=NOW + 2) == [] @@ -87,7 +77,7 @@ def test_road_feeds_take_only_the_selected_station_and_known_sensors(): reading = json.dumps({"value": 3.0, "time": int(NOW)}).encode() for topic in ( "weather-v2/9999/3", - "weather-v2/4057/999", + "weather-v2/1003/999", "weather-v2/status", "tms-v2/23438", ): @@ -105,15 +95,95 @@ def test_road_feeds_take_only_the_selected_station_and_known_sensors(): assert destination == "demo/traffic/23438/volume_1" -def test_multiplexed_assets_are_not_scored(): +def test_unselected_hamburg_detector_is_not_recorded(): + normalizer = Normalizer(SELECTION) + assert ( + normalizer.messages( + "hamburg", + "v1.1/Datastreams(99)/Observations", + json.dumps(MESSAGES[2][2]).encode(), + received_at=NOW, + ) + == [] + ) + + +@pytest.mark.parametrize("field,mqtt_topic", MOSQUITTO_SENSORS.items()) +def test_broker_metrics_record_only_fresh_numeric_updates(field, mqtt_topic): + normalizer = Normalizer(SELECTION) + assert ( + normalizer.messages( + "mosquitto", mqtt_topic, b"12606", received_at=NOW, retained=True + ) + == [] + ) + [(destination, encoded)] = normalizer.messages( + "mosquitto", mqtt_topic, b"12606", received_at=NOW + ) + assert destination == f"demo/mosquitto/test-mosquitto-org/{field}" + message = json.loads(encoded) + assert message["value"] == 12606 + assert message["observed_at"] is None + assert message["received_at"] == NOW + assert message["simulated"] is False + assert normalizer.messages("mosquitto", mqtt_topic, b"12606", received_at=NOW) == [] + assert normalizer.messages("mosquitto", mqtt_topic, b"12607", received_at=NOW + 10) + + +@pytest.mark.parametrize( + "wire", [b"true", b"null", b"{}", b"[]", b'"12"', b"-1", b"NaN", b"Infinity"] +) +def test_broker_metrics_reject_invalid_values_and_unknown_topics(wire): normalizer = Normalizer(SELECTION) - payload = MESSAGES[3][2] | {"Inverter_ID": "another-inverter"} assert ( normalizer.messages( - "solar", "inverters", json.dumps(payload).encode(), received_at=NOW + "mosquitto", "$SYS/broker/clients/connected", wire, received_at=NOW ) == [] ) + assert ( + normalizer.messages("mosquitto", "someone/clients", b"12", received_at=NOW) + == [] + ) + assert normalizer.latest == {} + + +def test_discovery_uses_supported_network_topics_then_pins_the_fresh_station( + monkeypatch, +): + clients = {} + monkeypatch.setattr(live_probe.time, "time", lambda: NOW) + + def subscribe(feed, receive): + client = clients[feed.id] = Mock() + # Digitraffic's network-wide subscription delivers measurements; the + # unsupported '+' variant used to leave discovery at an invalid default. + observations = { + "weather": ("weather-v2/#", "weather-v2/1001/1"), + "traffic": ("tms-v2/#", "tms-v2/23439/5122"), + } + if feed.id in observations: + subscription, mqtt_topic = observations[feed.id] + if subscription in feed.topics: + receive( + client, + None, + SimpleNamespace( + retain=False, + topic=mqtt_topic, + payload=json.dumps({"time": NOW, "value": 1.3}).encode(), + ), + ) + return client + + monkeypatch.setattr(live_probe, "subscriber", subscribe) + selected = live_probe.probe(SELECTION, seconds=0, discover=True) + assert selected.weather_station == 1001 + assert selected.traffic_station == 23439 + clients["weather"].unsubscribe.assert_called_once_with(["weather-v2/#"]) + subscriptions = clients["weather"].subscribe.call_args.args[0] + assert ("weather-v2/1001/1", 0) in subscriptions + assert ("weather-v2/status", 0) in subscriptions @pytest.mark.parametrize( @@ -139,7 +209,7 @@ def test_catalogue_roundtrips_and_changed_entities_get_new_reading_ids(): "weather", "traffic", "hamburg", - "solar", + "mosquitto", } before = { reading.id diff --git a/docs/operations/public-demo.md b/docs/operations/public-demo.md index 4f44f16f..f28719f2 100644 --- a/docs/operations/public-demo.md +++ b/docs/operations/public-demo.md @@ -20,10 +20,10 @@ monitoring evidence, pipelines and Dojo datasets become public. ## Public feeds The initial catalogue contains four providers, five devices and fourteen numeric -readings, recorded every five seconds. Every selected source reports around the -clock, so monitors see a continuous signal with daily patterns rather than the -gaps of a single vehicle going in and out of service. A small relay separates individual assets -and gives moving or multiplexed upstream topics stable application topics. +readings, sampled at five-second intervals when new messages arrive. The sources +report at different cadences; a quiet or unavailable upstream never produces +invented readings. A small relay separates individual assets and gives upstream +topics stable application topics. The normal collection and monitoring services process those topics. | Provider | Selected readings | Reference | @@ -31,22 +31,27 @@ The normal collection and monitoring services process those topics. | Fintraffic road weather | Air and road surface temperature, mean wind, humidity at one station | [Road traffic API](https://www.digitraffic.fi/en/road-traffic/) | | Fintraffic traffic counts | Vehicles per hour and average speed in both directions at one station | [Road traffic API](https://www.digitraffic.fi/en/road-traffic/) | | Hamburg traffic | Occupancy for detectors 13 and 30 | [Hamburg SensorThings](https://tld.iot.hamburg.de/v1.1/) | -| TDengine solar | Inverter power, voltage, temperature, efficiency | [Simulated solar feed](https://tdengine.com/free-public-mqtt-data-feed-real-time-solar-fleet-simulation-now-available/) | +| Mosquitto public broker | Connected clients, received/sent messages per minute, received bytes per minute | [Public test broker](https://test.mosquitto.org/) | -TDengine is explicitly labeled **Simulated**. PONTOS is excluded for now. Provider attribution and license links are available from **Demo → Data sources**. -The Fintraffic road stations publish about once a minute; traffic counts are -five-minute sliding values. Hamburg detectors are event-driven and quieter at -night. TDengine simulates a few-second cadence. Single trams, trains and vessels +Fintraffic weather updates can be several minutes apart; traffic counts are +five-minute sliding values. Hamburg detectors report short vehicle-presence +pulses: zero and sharp occupancy peaks are normal, with quiet periods between +vehicles. Five-second sampling can omit intermediate changes. Mosquitto reports +real operational metrics, typically about every ten seconds. It is a public test +service, so interruptions are possible. Single trams, trains and vessels were dropped from the initial selection because they only report while that one vehicle is in service. The relay rejects retained messages, duplicates, out-of-order readings, non-finite values and observations older than two minutes (ten for the Fintraffic road stations, which publish a measurement about a minute after taking it) or more -than ten seconds in the future. It preserves observation and arrival timestamps in the normalized payload; -the current application's charts use arrival time. Missing upstream data stays +than ten seconds in the future. It preserves observation and arrival timestamps +where the provider supplies them. Mosquitto's numeric `$SYS` messages have no source +timestamp: their observation time remains null and fresh delivery is recorded +at arrival. Retained snapshots are discarded. The current application's charts +use arrival time. Missing upstream data stays missing. There is no fallback generator. The relay retries failed connections with bounded backoff and keeps no history queue to replay as fresh data. @@ -59,14 +64,17 @@ backend/.venv/bin/python -m off_key_mqtt_proxy.demo.probe --discover --seconds 2 backend/.venv/bin/python -m off_key_mqtt_proxy.demo.probe --selection /tmp/demo-selection.json --seconds 20 ``` -Discovery subscribes briefly to the relevant fleet, selects a fresh asset and -narrows the subscription. For the road networks it picks the first station heard, +Discovery subscribes briefly to the documented `weather-v2/#` and `tms-v2/#` +network topics, selects a fresh asset and narrows the subscription. For the road +networks it picks the first station heard, so prefer a known station with every sensor and check it with `--selection`. It never publishes to a provider. Review the output before copying it into the inventory. A zero count means no fresh observation was received during that bounded check, not proof that a source is permanently down. -The implementation check on 2026-11-12 received Hamburg data and TDengine did not -connect. The road stations 4057 and 23438 are unverified defaults; probe them on -the deployment network first. +On 2026-11-13, weather station 1003 supplied all four selected measurements, +traffic station 23438 and Hamburg supplied fresh readings, and all four Mosquitto +metrics arrived over verified TLS from the demo worker. The invalid weather +station 4057 and unreachable TDengine endpoint were replaced. These checks do not +guarantee future provider availability; verify freshness before each launch. When changing an asset after launch, update the relay selection and import an updated catalogue through normal administration. Generate it with: diff --git a/infra/ansible/inventories/demo/group_vars/all/vars.yml b/infra/ansible/inventories/demo/group_vars/all/vars.yml index 7b4da0c4..5957634f 100644 --- a/infra/ansible/inventories/demo/group_vars/all/vars.yml +++ b/infra/ansible/inventories/demo/group_vars/all/vars.yml @@ -28,10 +28,9 @@ telemetry_storage_cap_gb: 2 # Pin chosen assets. Review the probe's --discover output after a bounded live check; # never silently switch entities underneath an existing monitor. demo_feed_selection: - weather_station: 4057 + weather_station: 1003 traffic_station: 23438 hamburg_detectors: [13, 30] - solar_inverter: INV_001_001 # Run the whole demo on one worker, away from the manager and production's # nodes. Label it once from the manager: # docker node update --label-add offkey-demo=true