feat(data-feeds): 10 open-data OSC publisher

data_feeds/config.avlivedata.toml

@@ -0,0 +1,37 @@
+[osc]
+host = "127.0.0.1"
+port = 57127
+
+[feeds.eco2mix]
+enabled = true
+interval_sec = 60
+
+[feeds.velib]
+enabled = true
+interval_sec = 120
+station_codes = []
+
+[feeds.hubeau]
+enabled = true
+interval_sec = 300
+codes = ["F050001001"]
+
+[feeds.gbfs]
+enabled = false
+interval_sec = 120
+url = "https://velib-metropole-opendata.smoove.pro/opendata/Velib_Metropole/station_status.json"
+
+[feeds.ais]
+enabled = false
+
+[feeds.carburants]
+enabled = false
+
+[feeds.prim]
+enabled = false
+
+[feeds.sytadin]
+enabled = false
+
+[feeds.teleray]
+enabled = false

data_feeds/feeds/__init__.py

@@ -0,0 +1,27 @@
+"""Registry of available feed classes (auto-discovery on import)."""
+from __future__ import annotations
+
+from .base import Feed
+from .eco2mix import Eco2MixFeed
+from .gbfs import GBFSFeed
+from .hubeau import HubeauFeed
+from .velib import VelibFeed
+from .ais import AISFeed
+from .carburants import CarburantsFeed
+from .prim import PRIMFeed
+from .sytadin import SytadinFeed
+from .teleray import TelerayFeed
+
+REGISTRY: dict[str, type[Feed]] = {
+    "eco2mix": Eco2MixFeed,
+    "gbfs": GBFSFeed,
+    "hubeau": HubeauFeed,
+    "velib": VelibFeed,
+    "ais": AISFeed,
+    "carburants": CarburantsFeed,
+    "prim": PRIMFeed,
+    "sytadin": SytadinFeed,
+    "teleray": TelerayFeed,
+}
+
+__all__ = ["Feed", "REGISTRY"]

data_feeds/feeds/ais.py

@@ -0,0 +1,22 @@
+"""AIS vessel positions feed — STUB.
+
+TODO: needs aisstream.io API key + websocket subscription.
+"""
+from __future__ import annotations
+
+import logging
+
+from .base import Feed
+
+LOG = logging.getLogger("data_feeds.ais")
+
+
+class AISFeed(Feed):
+    name = "ais"
+    interval_sec = 60.0
+
+    def fetch(self):
+        return None
+
+    def publish(self, payload) -> None:
+        LOG.info("stub")

data_feeds/feeds/base.py

@@ -0,0 +1,55 @@
+"""Abstract base class for data feeds."""
+from __future__ import annotations
+
+import abc
+import logging
+import time
+import threading
+from typing import Any
+
+LOG = logging.getLogger("data_feeds.base")
+
+
+class Feed(abc.ABC):
+    name: str = "feed"
+    interval_sec: float = 60.0
+
+    def __init__(self, osc_send, **cfg) -> None:
+        self.osc_send = osc_send
+        self.cfg = cfg
+        self._stop = threading.Event()
+        self._thread: threading.Thread | None = None
+        self.last_t: float = 0.0
+
+    def configure(self, **kwargs) -> None:
+        self.cfg.update(kwargs)
+        if "interval_sec" in kwargs:
+            self.interval_sec = float(kwargs["interval_sec"])
+
+    @abc.abstractmethod
+    def fetch(self) -> Any: ...
+
+    @abc.abstractmethod
+    def publish(self, payload: Any) -> None: ...
+
+    def tick(self) -> None:
+        try:
+            payload = self.fetch()
+            self.publish(payload)
+            self.last_t = time.time()
+            self.osc_send(f"/data/{self.name}/heartbeat", [self.last_t])
+        except Exception as e:  # noqa: BLE001
+            LOG.warning("%s fetch failed: %s", self.name, e)
+
+    def start(self) -> None:
+        self._thread = threading.Thread(
+            target=self._run, name=f"feed-{self.name}", daemon=True)
+        self._thread.start()
+
+    def stop(self) -> None:
+        self._stop.set()
+
+    def _run(self) -> None:
+        while not self._stop.is_set():
+            self.tick()
+            self._stop.wait(self.interval_sec)

data_feeds/feeds/carburants.py

@@ -0,0 +1,22 @@
+"""Prix carburants feed — STUB.
+
+TODO: needs prix-carburants.gouv.fr GeoJSON cache + station selection.
+"""
+from __future__ import annotations
+
+import logging
+
+from .base import Feed
+
+LOG = logging.getLogger("data_feeds.carburants")
+
+
+class CarburantsFeed(Feed):
+    name = "carburants"
+    interval_sec = 3600.0
+
+    def fetch(self):
+        return None
+
+    def publish(self, payload) -> None:
+        LOG.info("stub")

data_feeds/feeds/eco2mix.py

@@ -0,0 +1,58 @@
+"""RTE eco2mix feed — France electricity production mix in MW.
+
+Uses the public OpenDataSoft mirror of RTE eco2mix-national-tr (temps reel,
+15-min resolution). Stdlib HTTP only.
+"""
+from __future__ import annotations
+
+import json
+import logging
+import urllib.parse
+import urllib.request
+from typing import Any
+
+from .base import Feed
+
+LOG = logging.getLogger("data_feeds.eco2mix")
+
+# OpenDataSoft public mirror — no key required.
+URL = (
+    "https://odre.opendatasoft.com/api/records/1.0/search/"
+    "?dataset=eco2mix-national-tr&rows=1&sort=-date_heure"
+)
+
+
+class Eco2MixFeed(Feed):
+    name = "eco2mix"
+    interval_sec = 60.0
+
+    def fetch(self) -> Any:
+        req = urllib.request.Request(URL, headers={"User-Agent": "av-live/0.1"})
+        with urllib.request.urlopen(req, timeout=10) as r:
+            data = json.loads(r.read().decode("utf-8"))
+        records = data.get("records") or []
+        if not records:
+            return None
+        return records[0].get("fields") or {}
+
+    def publish(self, payload: Any) -> None:
+        if not isinstance(payload, dict):
+            return
+        # Pick a representative subset (MW). Keys per eco2mix-national-tr.
+        keys = [
+            "consommation", "nucleaire", "gaz", "charbon", "fioul",
+            "eolien", "solaire", "hydraulique", "bioenergies",
+            "ech_physiques",
+        ]
+        count = 0
+        for k in keys:
+            v = payload.get(k)
+            if v is None:
+                continue
+            try:
+                fv = float(v)
+            except (TypeError, ValueError):
+                continue
+            self.osc_send(f"/data/{self.name}/sample", [k, fv])
+            count += 1
+        self.osc_send(f"/data/{self.name}/count", [count])

data_feeds/feeds/gbfs.py

@@ -0,0 +1,53 @@
+"""Generic GBFS (General Bikeshare Feed Specification) reader.
+
+Reads a `station_status.json` URL and publishes aggregate counters.
+"""
+from __future__ import annotations
+
+import json
+import logging
+import urllib.request
+from typing import Any
+
+from .base import Feed
+
+LOG = logging.getLogger("data_feeds.gbfs")
+
+
+class GBFSFeed(Feed):
+    name = "gbfs"
+    interval_sec = 120.0
+
+    def fetch(self) -> Any:
+        url = self.cfg.get("url")
+        if not url:
+            LOG.info("gbfs disabled: no url configured")
+            return None
+        req = urllib.request.Request(url, headers={"User-Agent": "av-live/0.1"})
+        with urllib.request.urlopen(req, timeout=10) as r:
+            return json.loads(r.read().decode("utf-8"))
+
+    def publish(self, payload: Any) -> None:
+        if not isinstance(payload, dict):
+            return
+        stations = (payload.get("data") or {}).get("stations") or []
+        if not stations:
+            return
+        codes = set(self.cfg.get("station_codes") or [])
+        bikes = 0
+        docks = 0
+        operative = 0
+        sampled = 0
+        for s in stations:
+            sid = str(s.get("station_id", ""))
+            if codes and sid not in codes:
+                continue
+            bikes += int(s.get("num_bikes_available") or 0)
+            docks += int(s.get("num_docks_available") or 0)
+            if s.get("is_renting") or s.get("is_installed"):
+                operative += 1
+            sampled += 1
+        self.osc_send(f"/data/{self.name}/sample", ["bikes_available", float(bikes)])
+        self.osc_send(f"/data/{self.name}/sample", ["docks_available", float(docks)])
+        self.osc_send(f"/data/{self.name}/sample", ["stations_active", float(operative)])
+        self.osc_send(f"/data/{self.name}/count", [sampled])

data_feeds/feeds/hubeau.py

@@ -0,0 +1,66 @@
+"""Hub'Eau hydrometrie feed — water level and flow rate for French rivers.
+
+API: https://hubeau.eaufrance.fr/api/v1/hydrometrie/observations_tr
+Open, no API key required.
+"""
+from __future__ import annotations
+
+import json
+import logging
+import urllib.parse
+import urllib.request
+from typing import Any
+
+from .base import Feed
+
+LOG = logging.getLogger("data_feeds.hubeau")
+
+BASE = "https://hubeau.eaufrance.fr/api/v1/hydrometrie/observations_tr"
+
+
+class HubeauFeed(Feed):
+    name = "hubeau"
+    interval_sec = 300.0
+
+    def fetch(self) -> Any:
+        codes = self.cfg.get("codes") or ["F050001001"]
+        out: dict[str, dict[str, float]] = {}
+        for code in codes:
+            params = {
+                "code_entite": code,
+                "size": 1,
+                "sort": "desc",
+                "fields": "code_station,grandeur_hydro,resultat_obs,date_obs",
+            }
+            url = BASE + "?" + urllib.parse.urlencode(params)
+            try:
+                req = urllib.request.Request(
+                    url, headers={"User-Agent": "av-live/0.1"})
+                with urllib.request.urlopen(req, timeout=10) as r:
+                    data = json.loads(r.read().decode("utf-8"))
+            except Exception as e:  # noqa: BLE001
+                LOG.warning("hubeau %s failed: %s", code, e)
+                continue
+            for obs in data.get("data") or []:
+                station = obs.get("code_station") or code
+                gr = obs.get("grandeur_hydro") or "X"
+                v = obs.get("resultat_obs")
+                if v is None:
+                    continue
+                try:
+                    fv = float(v)
+                except (TypeError, ValueError):
+                    continue
+                out.setdefault(station, {})[gr] = fv
+        return out
+
+    def publish(self, payload: Any) -> None:
+        if not isinstance(payload, dict) or not payload:
+            return
+        count = 0
+        for station, vals in payload.items():
+            for gr, v in vals.items():
+                key = f"{station}_{gr}"
+                self.osc_send(f"/data/{self.name}/sample", [key, float(v)])
+                count += 1
+        self.osc_send(f"/data/{self.name}/count", [count])

data_feeds/feeds/prim.py

@@ -0,0 +1,22 @@
+"""PRIM Ile-de-France Mobilites feed — STUB.
+
+TODO: needs API key (https://prim.iledefrance-mobilites.fr/).
+"""
+from __future__ import annotations
+
+import logging
+
+from .base import Feed
+
+LOG = logging.getLogger("data_feeds.prim")
+
+
+class PRIMFeed(Feed):
+    name = "prim"
+    interval_sec = 60.0
+
+    def fetch(self):
+        return None
+
+    def publish(self, payload) -> None:
+        LOG.info("stub")

data_feeds/feeds/sytadin.py

@@ -0,0 +1,22 @@
+"""Sytadin Paris traffic feed — STUB.
+
+TODO: needs sytadin.fr scraping / cumulative km of congestion.
+"""
+from __future__ import annotations
+
+import logging
+
+from .base import Feed
+
+LOG = logging.getLogger("data_feeds.sytadin")
+
+
+class SytadinFeed(Feed):
+    name = "sytadin"
+    interval_sec = 300.0
+
+    def fetch(self):
+        return None
+
+    def publish(self, payload) -> None:
+        LOG.info("stub")

data_feeds/feeds/teleray.py

@@ -0,0 +1,22 @@
+"""IRSN Teleray radiation feed — STUB.
+
+TODO: needs https://teleray.irsn.fr/data endpoint research.
+"""
+from __future__ import annotations
+
+import logging
+
+from .base import Feed
+
+LOG = logging.getLogger("data_feeds.teleray")
+
+
+class TelerayFeed(Feed):
+    name = "teleray"
+    interval_sec = 600.0
+
+    def fetch(self):
+        return None
+
+    def publish(self, payload) -> None:
+        LOG.info("stub")

data_feeds/feeds/velib.py

@@ -0,0 +1,25 @@
+"""Velib Metropole feed — specialization of GBFS against the Paris URL."""
+from __future__ import annotations
+
+import logging
+
+from .gbfs import GBFSFeed
+
+LOG = logging.getLogger("data_feeds.velib")
+
+VELIB_URL = (
+    "https://velib-metropole-opendata.smoove.pro/opendata/"
+    "Velib_Metropole/station_status.json"
+)
+
+
+class VelibFeed(GBFSFeed):
+    name = "velib"
+    interval_sec = 120.0
+
+    def configure(self, **kwargs) -> None:
+        # Force the URL if caller did not provide one.
+        kwargs.setdefault("url", VELIB_URL)
+        super().configure(**kwargs)
+        if not self.cfg.get("url"):
+            self.cfg["url"] = VELIB_URL

data_feeds/main.py

@@ -0,0 +1,57 @@
+"""Run all enabled feeds, publish OSC to AVLiveBody."""
+from __future__ import annotations
+
+import argparse
+import logging
+import sys
+import time
+import tomllib
+from pathlib import Path
+
+from .feeds import REGISTRY
+from .osc_sender import OscSender
+
+
+def main(argv: list[str] | None = None) -> int:
+    p = argparse.ArgumentParser()
+    p.add_argument("--config", default="data_feeds/config.avlivedata.toml")
+    p.add_argument("--osc-host")
+    p.add_argument("--osc-port", type=int)
+    p.add_argument("-v", "--verbose", action="store_true")
+    args = p.parse_args(argv)
+    logging.basicConfig(
+        level=logging.INFO if args.verbose else logging.WARNING,
+        format="%(asctime)s %(levelname)s %(name)s %(message)s")
+    cfg = tomllib.loads(Path(args.config).read_text())
+    osc_cfg = cfg.get("osc", {})
+    host = args.osc_host or osc_cfg.get("host", "127.0.0.1")
+    port = args.osc_port or osc_cfg.get("port", 57127)
+    sender = OscSender(host, port)
+    feeds = []
+    for name, kwargs in (cfg.get("feeds") or {}).items():
+        if not kwargs.get("enabled", False):
+            continue
+        cls = REGISTRY.get(name)
+        if cls is None:
+            logging.warning("Unknown feed: %s", name)
+            continue
+        f = cls(sender.send)
+        f.configure(**kwargs)
+        f.start()
+        feeds.append(f)
+        logging.info("started feed %s (interval %.0fs)", name, f.interval_sec)
+    if not feeds:
+        logging.warning("No feeds enabled. Exiting.")
+        return 1
+    try:
+        while True:
+            time.sleep(60)
+    except KeyboardInterrupt:
+        return 0
+    finally:
+        for f in feeds:
+            f.stop()
+
+
+if __name__ == "__main__":
+    sys.exit(main())

data_feeds/osc_sender.py

@@ -0,0 +1,22 @@
+"""Wrapper around python-osc SimpleUDPClient with per-route helpers."""
+from __future__ import annotations
+
+import logging
+from typing import Any
+
+from pythonosc.udp_client import SimpleUDPClient
+
+LOG = logging.getLogger("data_feeds.osc")
+
+
+class OscSender:
+    def __init__(self, host: str, port: int) -> None:
+        self.host = host
+        self.port = port
+        self._client = SimpleUDPClient(host, port)
+
+    def send(self, addr: str, args: list[Any]) -> None:
+        try:
+            self._client.send_message(addr, args)
+        except OSError as e:
+            LOG.warning("send %s failed: %s", addr, e)