diff --git a/data_feeds/__init__.py b/data_feeds/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/data_feeds/config.avlivedata.toml b/data_feeds/config.avlivedata.toml new file mode 100644 index 0000000..ebecaea --- /dev/null +++ b/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 diff --git a/data_feeds/feeds/__init__.py b/data_feeds/feeds/__init__.py index e69de29..aeba3e5 100644 --- a/data_feeds/feeds/__init__.py +++ b/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"] diff --git a/data_feeds/feeds/ais.py b/data_feeds/feeds/ais.py new file mode 100644 index 0000000..bf51df8 --- /dev/null +++ b/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") diff --git a/data_feeds/feeds/base.py b/data_feeds/feeds/base.py new file mode 100644 index 0000000..9446c1e --- /dev/null +++ b/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) diff --git a/data_feeds/feeds/carburants.py b/data_feeds/feeds/carburants.py new file mode 100644 index 0000000..fc06443 --- /dev/null +++ b/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") diff --git a/data_feeds/feeds/eco2mix.py b/data_feeds/feeds/eco2mix.py new file mode 100644 index 0000000..23a6627 --- /dev/null +++ b/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]) diff --git a/data_feeds/feeds/gbfs.py b/data_feeds/feeds/gbfs.py new file mode 100644 index 0000000..ab7db7b --- /dev/null +++ b/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]) diff --git a/data_feeds/feeds/hubeau.py b/data_feeds/feeds/hubeau.py new file mode 100644 index 0000000..c6d65a3 --- /dev/null +++ b/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]) diff --git a/data_feeds/feeds/prim.py b/data_feeds/feeds/prim.py new file mode 100644 index 0000000..54c8389 --- /dev/null +++ b/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") diff --git a/data_feeds/feeds/sytadin.py b/data_feeds/feeds/sytadin.py new file mode 100644 index 0000000..6470044 --- /dev/null +++ b/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") diff --git a/data_feeds/feeds/teleray.py b/data_feeds/feeds/teleray.py new file mode 100644 index 0000000..8a52164 --- /dev/null +++ b/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") diff --git a/data_feeds/feeds/velib.py b/data_feeds/feeds/velib.py new file mode 100644 index 0000000..79088ad --- /dev/null +++ b/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 diff --git a/data_feeds/main.py b/data_feeds/main.py new file mode 100644 index 0000000..2dacae9 --- /dev/null +++ b/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()) diff --git a/data_feeds/osc_sender.py b/data_feeds/osc_sender.py new file mode 100644 index 0000000..5c20101 --- /dev/null +++ b/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)