Publish Storebaelt webcam freshness heartbeats

Author: Sam-Bolling

publishers/storebaelt_webcams/bootstrap_storebaelt_webcams.py

@@ -190,6 +190,13 @@ def _datastream_schema(camera: dict) -> dict:
                     {"type": "Text", "name": "sourceLastModifiedTime", "label": "Parsed Source Last-Modified Time", "definition": "http://sensorml.com/ont/swe/property/Timestamp"},
                     {"type": "Text", "name": "contentLength", "label": "Content Length", "definition": "http://sensorml.com/ont/swe/property/Size"},
                     {"type": "Text", "name": "imageSha256", "label": "Image SHA-256", "definition": "http://sensorml.com/ont/swe/property/Identifier"},
+                    {"type": "Boolean", "name": "imageChanged", "label": "Image Changed", "definition": "http://sensorml.com/ont/swe/property/Status"},
+                    {"type": "Text", "name": "firstSeenTime", "label": "Current Image First Seen Time", "definition": "http://sensorml.com/ont/swe/property/Timestamp"},
+                    {"type": "Text", "name": "lastSeenTime", "label": "Source Last Checked Time", "definition": "http://sensorml.com/ont/swe/property/Timestamp"},
+                    {"type": "Text", "name": "lastChangedTime", "label": "Image Last Changed Time", "definition": "http://sensorml.com/ont/swe/property/Timestamp"},
+                    {"type": "Count", "name": "unchangedPollCount", "label": "Unchanged Poll Count", "definition": "http://sensorml.com/ont/swe/property/Count"},
+                    {"type": "Text", "name": "stalenessStatus", "label": "Staleness Status", "definition": "http://sensorml.com/ont/swe/property/Status"},
+                    {"type": "Count", "name": "sourceAgeSeconds", "label": "Seconds Since Image Changed", "definition": "http://sensorml.com/ont/swe/property/ElapsedTime"},
                     {"type": "Text", "name": "sourceUrl", "label": "Source URL", "definition": "http://sensorml.com/ont/swe/property/ReferenceURL"},
                 ],
             },

publishers/storebaelt_webcams/storebaelt_webcams_publisher.py

@@ -21,6 +21,7 @@
 
 USER_AGENT = "OS4CSAPI Storebaelt Webcams Publisher/1.0"
 DS_OUTPUT_NAME = "storebaeltWebcamImage"
+DEFAULT_STALE_SECONDS = 15 * 60
 
 
 def _load_cameras() -> list[dict]:
@@ -51,6 +52,19 @@ def _parse_http_date(value: str | None) -> str | None:
     return dt.astimezone(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
 
 
+def _format_utc(dt: datetime) -> str:
+    return dt.astimezone(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
+
+
+def _parse_iso_utc(value: str | None) -> datetime | None:
+    if not value:
+        return None
+    try:
+        return datetime.fromisoformat(value.replace("Z", "+00:00")).astimezone(timezone.utc)
+    except ValueError:
+        return None
+
+
 def _probe_image_url(url: str) -> dict:
     headers = {
         "Accept": "image/jpeg,*/*;q=0.8",
@@ -136,8 +150,9 @@ def __init__(self, camera_filter: list[str] | None = None):
         self._base_url = os.environ.get("OSH_BASE_URL", f"https://{self.osh_address}/{self.osh_root}/api")
         self._auth = "Basic " + base64.b64encode(f"{self.osh_user}:{self.osh_pass}".encode()).decode()
         self._ds_ids: dict[str, str] = {}
-        self._seen: set[str] = set()
+        self._image_state: dict[str, dict] = {}
         self._request_delay = float(os.environ.get("STOREBAELT_WEBCAMS_REQUEST_DELAY", "0.5"))
+        self._stale_seconds = int(os.environ.get("STOREBAELT_WEBCAMS_STALE_SECONDS", str(DEFAULT_STALE_SECONDS)))
         self.stats = {"published": 0, "errors": 0, "reconnects": 0, "skipped": 0}
 
     def connect(self):
@@ -155,6 +170,7 @@ def connect(self):
                 print(f"  [WARN] Datastream {DS_OUTPUT_NAME} not found for camera {camera['id']}")
                 continue
             self._ds_ids[camera["id"]] = ds_id
+            self._seed_image_state(camera["id"], ds_id)
             connected += 1
             print(f"  Connected: {camera['id']} -> sys={sys_id} ds={ds_id}")
         print(f"  Ready: {connected}/{len(self.cameras)} cameras connected")
@@ -197,6 +213,78 @@ def _post_observation(self, ds_id: str, obs: dict):
             body_text = exc.read().decode("utf-8", errors="replace")[:500]
             raise RuntimeError(f"HTTP {exc.code} POST {url}: {body_text}") from exc
 
+    def _seed_image_state(self, camera_id: str, ds_id: str):
+        try:
+            data = api_get(self._base_url, f"datastreams/{ds_id}/observations?limit=1&resultTime=latest", self._auth)
+        except Exception as exc:
+            print(f"  [WARN] Could not seed freshness state for {camera_id}: {exc}")
+            return
+        items = (data or {}).get("items") or []
+        if not items:
+            return
+        obs = items[0]
+        result = obs.get("result") or {}
+        image_sha256 = result.get("imageSha256") or ""
+        if not image_sha256:
+            return
+        observed_time = obs.get("phenomenonTime") or obs.get("resultTime") or _format_utc(datetime.now(timezone.utc))
+        first_seen = result.get("firstSeenTime") or result.get("lastChangedTime") or observed_time
+        last_changed = result.get("lastChangedTime") or first_seen
+        last_seen = result.get("lastSeenTime") or observed_time
+        self._image_state[camera_id] = {
+            "imageSha256": image_sha256,
+            "firstSeenTime": first_seen,
+            "lastChangedTime": last_changed,
+            "lastSeenTime": last_seen,
+            "unchangedPollCount": int(result.get("unchangedPollCount") or 0),
+        }
+
+    def _apply_freshness_status(self, camera_id: str, latest: dict, poll_time: datetime) -> dict:
+        result = dict(latest["result"])
+        poll_time_iso = latest["phenomenonTime"]
+        effective_poll_time = _parse_iso_utc(poll_time_iso) or poll_time
+        image_sha256 = result.get("imageSha256") or latest["dedupeKey"].split("|", 1)[-1]
+        previous = self._image_state.get(camera_id)
+        image_changed = not previous or previous.get("imageSha256") != image_sha256
+
+        if image_changed:
+            first_seen = poll_time_iso
+            last_changed = poll_time_iso
+            unchanged_count = 0
+        else:
+            first_seen = previous.get("firstSeenTime") or poll_time_iso
+            last_changed = previous.get("lastChangedTime") or first_seen
+            unchanged_count = int(previous.get("unchangedPollCount") or 0) + 1
+
+        last_changed_dt = _parse_iso_utc(last_changed) or effective_poll_time
+        source_age_seconds = max(0, int((effective_poll_time - last_changed_dt).total_seconds()))
+        if image_changed:
+            staleness_status = "fresh"
+        elif source_age_seconds >= self._stale_seconds:
+            staleness_status = "stale"
+        else:
+            staleness_status = "unchanged"
+
+        result.update({
+            "imageChanged": image_changed,
+            "firstSeenTime": first_seen,
+            "lastSeenTime": poll_time_iso,
+            "lastChangedTime": last_changed,
+            "unchangedPollCount": unchanged_count,
+            "stalenessStatus": staleness_status,
+            "sourceAgeSeconds": source_age_seconds,
+        })
+        self._image_state[camera_id] = {
+            "imageSha256": image_sha256,
+            "firstSeenTime": first_seen,
+            "lastChangedTime": last_changed,
+            "lastSeenTime": poll_time_iso,
+            "unchangedPollCount": unchanged_count,
+        }
+        latest = dict(latest)
+        latest["result"] = result
+        return latest
+
     def publish_cycle(self, dry_run: bool = False) -> int:
         published = 0
         now = datetime.now(timezone.utc)
@@ -213,25 +301,23 @@ def publish_cycle(self, dry_run: bool = False) -> int:
                 self.stats["skipped"] += 1
                 print(f"  [{ts_label}] {camera_id}: no image metadata")
                 continue
-            if latest["dedupeKey"] in self._seen:
-                self.stats["skipped"] += 1
-                print(f"  [{ts_label}] {camera_id}: unchanged, skipping")
-                continue
+            latest = self._apply_freshness_status(camera_id, latest, now)
             obs = {
                 "phenomenonTime": latest["phenomenonTime"],
                 "resultTime": now.strftime("%Y-%m-%dT%H:%M:%SZ"),
                 "result": latest["result"],
             }
-            label = f"{latest['phenomenonTime']} {latest['result']['imageUrl']}"
+            status = latest["result"].get("stalenessStatus", "unknown")
+            changed = "changed" if latest["result"].get("imageChanged") else "unchanged"
+            age = latest["result"].get("sourceAgeSeconds", 0)
+            label = f"{latest['phenomenonTime']} {changed}/{status} age={age}s {latest['result']['imageUrl']}"
             if dry_run:
                 print(f"  [{ts_label}] {camera_id}: [DRY] {label}")
-                self._seen.add(latest["dedupeKey"])
             else:
                 try:
                     self._post_observation(ds_id, obs)
                     self.stats["published"] += 1
                     published += 1
-                    self._seen.add(latest["dedupeKey"])
                     print(f"  [{ts_label}] {camera_id}: OK {label}")
                 except Exception as exc:
                     self.stats["errors"] += 1

tests/test_storebaelt_webcams_publisher.py

@@ -53,12 +53,13 @@ def fake_probe(url):
     assert latest["result"]["imageSha256"] == "f" * 64
 
 
-def test_publish_cycle_dry_run_dedupes(monkeypatch):
+def test_publish_cycle_dry_run_publishes_unchanged_heartbeat(monkeypatch):
     publisher = StorebaeltWebcamsPublisher.__new__(StorebaeltWebcamsPublisher)
     publisher.cameras = [{"id": "sprogo", "title": "Sprogo Webcam"}]
     publisher._ds_ids = {}
-    publisher._seen = set()
+    publisher._image_state = {}
     publisher._request_delay = 0
+    publisher._stale_seconds = 900
     publisher.stats = {"published": 0, "errors": 0, "reconnects": 0, "skipped": 0}
 
     def fake_fetch(camera):
@@ -68,15 +69,77 @@ def fake_fetch(camera):
             "result": {
                 "cameraId": camera["id"],
                 "imageUrl": "https://stream.sob.m-dn.net/res/sb2-live.jpg",
+                "imageSha256": "abc",
             },
         }
 
     monkeypatch.setattr(storebaelt, "fetch_latest_image", fake_fetch)
 
     assert publisher.publish_cycle(dry_run=True) == 0
     assert publisher.publish_cycle(dry_run=True) == 0
-    assert "sprogo|abc" in publisher._seen
-    assert publisher.stats["skipped"] == 1
+    assert publisher._image_state["sprogo"]["imageSha256"] == "abc"
+    assert publisher._image_state["sprogo"]["unchangedPollCount"] == 1
+    assert publisher.stats["skipped"] == 0
+
+
+def test_freshness_status_tracks_changed_and_unchanged_images():
+    publisher = StorebaeltWebcamsPublisher.__new__(StorebaeltWebcamsPublisher)
+    publisher._image_state = {}
+    publisher._stale_seconds = 900
+    poll_time = datetime(2026, 6, 3, 12, 0, tzinfo=timezone.utc)
+    latest = {
+        "phenomenonTime": "2026-06-03T12:00:00Z",
+        "dedupeKey": "sprogo|abc",
+        "result": {"cameraId": "sprogo", "imageSha256": "abc"},
+    }
+
+    first = publisher._apply_freshness_status("sprogo", latest, poll_time)
+
+    assert first["result"]["imageChanged"] is True
+    assert first["result"]["stalenessStatus"] == "fresh"
+    assert first["result"]["unchangedPollCount"] == 0
+    assert first["result"]["firstSeenTime"] == "2026-06-03T12:00:00Z"
+    assert first["result"]["lastChangedTime"] == "2026-06-03T12:00:00Z"
+
+    second_latest = {
+        "phenomenonTime": "2026-06-03T12:05:00Z",
+        "dedupeKey": "sprogo|abc",
+        "result": {"cameraId": "sprogo", "imageSha256": "abc"},
+    }
+    second = publisher._apply_freshness_status("sprogo", second_latest, datetime(2026, 6, 3, 12, 5, tzinfo=timezone.utc))
+
+    assert second["result"]["imageChanged"] is False
+    assert second["result"]["stalenessStatus"] == "unchanged"
+    assert second["result"]["sourceAgeSeconds"] == 300
+    assert second["result"]["unchangedPollCount"] == 1
+    assert second["result"]["firstSeenTime"] == "2026-06-03T12:00:00Z"
+    assert second["result"]["lastChangedTime"] == "2026-06-03T12:00:00Z"
+
+
+def test_freshness_status_marks_stale_after_threshold():
+    publisher = StorebaeltWebcamsPublisher.__new__(StorebaeltWebcamsPublisher)
+    publisher._image_state = {
+        "sprogo": {
+            "imageSha256": "abc",
+            "firstSeenTime": "2026-06-03T12:00:00Z",
+            "lastChangedTime": "2026-06-03T12:00:00Z",
+            "lastSeenTime": "2026-06-03T12:10:00Z",
+            "unchangedPollCount": 2,
+        }
+    }
+    publisher._stale_seconds = 900
+    latest = {
+        "phenomenonTime": "2026-06-03T12:16:00Z",
+        "dedupeKey": "sprogo|abc",
+        "result": {"cameraId": "sprogo", "imageSha256": "abc"},
+    }
+
+    updated = publisher._apply_freshness_status("sprogo", latest, datetime(2026, 6, 3, 12, 16, tzinfo=timezone.utc))
+
+    assert updated["result"]["imageChanged"] is False
+    assert updated["result"]["stalenessStatus"] == "stale"
+    assert updated["result"]["sourceAgeSeconds"] == 960
+    assert updated["result"]["unchangedPollCount"] == 3
 
 
 def test_http_date_parser_normalizes_to_utc():