Replace inactive demo feeds with verified public telemetry

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],
     }
 
 

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.",
     ),
 )
 

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,
                     ),

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,18 +67,28 @@ 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:
         for feed in feeds(selection):
             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:

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,
+        ),
     )