diff --git a/data_feeds/.gitignore b/data_feeds/.gitignore index d765fa3..28960ab 100644 --- a/data_feeds/.gitignore +++ b/data_feeds/.gitignore @@ -3,3 +3,12 @@ __pycache__/ *.pyc .uv-cache/ uv.lock + +# Config local : peut contenir des creds (RTE OAuth, GCN Kafka). +# Le fichier de reference est config.toml.example (tracke). +# config.toml et config.data-only.toml restent committes tant qu'ils +# ne contiennent que des champs vides ; surveiller pre-commit hook. +*.local.toml + +# Modeles YOLO telecharges au runtime +*.pt diff --git a/data_feeds/bridge.py b/data_feeds/bridge.py index 9b0bc7e..f319e45 100644 --- a/data_feeds/bridge.py +++ b/data_feeds/bridge.py @@ -12,6 +12,7 @@ import argparse import asyncio import importlib import logging +import re import signal import sys import time @@ -19,6 +20,11 @@ from dataclasses import dataclass from pathlib import Path from typing import Any +# Whitelist des noms de feed : empeche l'injection de modules arbitraires +# via un config.toml malveillant (importlib resolve sur du . ou .. ferait +# remonter dans l'arborescence). +_FEED_NAME_RE = re.compile(r"^[a-z][a-z0-9_]{0,30}$") + try: import tomllib # py311+ except ModuleNotFoundError: @@ -52,6 +58,10 @@ def load_config(path: Path) -> dict[str, Any]: async def run_feed(name: str, cfg: dict[str, Any], ctx: Context) -> None: """Charge `data_feeds.feeds.` et appelle `run(ctx)`.""" + if not _FEED_NAME_RE.match(name): + LOG.error("invalid feed name %r — must match %s", + name, _FEED_NAME_RE.pattern) + return try: mod = importlib.import_module(f"data_feeds.feeds.{name}") except ModuleNotFoundError: @@ -62,7 +72,8 @@ async def run_feed(name: str, cfg: dict[str, Any], ctx: Context) -> None: while True: try: await mod.run(ctx) - # Si run() retourne sans exception, on redémarre poliment + # Retour propre : on reset le backoff et on rejoue en boucle. + backoff = 1.0 await asyncio.sleep(2.0) except asyncio.CancelledError: raise @@ -108,8 +119,19 @@ async def main_async(cfg: dict[str, Any]) -> None: loop = asyncio.get_running_loop() stop = loop.create_future() + sig_count = 0 + + def _on_signal() -> None: + nonlocal sig_count + sig_count += 1 + if sig_count > 1: + LOG.warning("second signal received — forcing exit") + sys.exit(1) + if not stop.done(): + stop.cancel() + for sig in (signal.SIGINT, signal.SIGTERM): - loop.add_signal_handler(sig, lambda: stop.cancel() if not stop.done() else None) + loop.add_signal_handler(sig, _on_signal) try: await stop except asyncio.CancelledError: diff --git a/data_feeds/config.data-only.toml b/data_feeds/config.data-only.toml new file mode 100644 index 0000000..6b83078 --- /dev/null +++ b/data_feeds/config.data-only.toml @@ -0,0 +1,90 @@ +# Profil "data-only" : que les flux open-data + pose YOLO, vers SC + oF. +# +# Pas de sclang/oscope/web → la chaine consomatrice est libre de +# brancher ce qu'elle veut sur les ports 57121 / 57123. +# +# Lancer avec : uv run python bridge.py -c config.data-only.toml -v + +[osc] +targets = [ + { host = "127.0.0.1", port = 57121 }, # SuperCollider (si lance) + { host = "127.0.0.1", port = 57123 }, # openFrameworks +] +prefix = "/data" + +# -- Pose / webcam -------------------------------------------------------- +[feeds.pose] +# Active uniquement quand le pont tourne dans le bundle launcher (TCC OK). +# En CLI le prompt webcam ne peut pas apparaitre -> camera silently refused. +enabled = false +model = "yolov8n-pose.pt" +device = "mps" +camera = 0 +width = 640 +height = 480 +target_fps = 20 +conf_thresh = 0.35 +max_persons = 4 +emit_keypoints = true + +# -- Sismique / géophysique ------------------------------------------------ +[feeds.usgs] +enabled = true +url = "https://earthquake.usgs.gov/earthquakes/feed/v1.0/summary/all_hour.geojson" +poll_seconds = 60 + +[feeds.swpc] +enabled = true +url_plasma = "https://services.swpc.noaa.gov/products/solar-wind/plasma-1-day.json" +url_mag = "https://services.swpc.noaa.gov/products/solar-wind/mag-1-day.json" +url_kp = "https://services.swpc.noaa.gov/products/noaa-planetary-k-index.json" +url_xray = "https://services.swpc.noaa.gov/json/goes/primary/xrays-1-day.json" +poll_seconds = 60 + +# -- Réseau électrique ----------------------------------------------------- +[feeds.netzfrequenz] +# WARN : mainsfrequenz.de a ferme son WS public (NXDOMAIN 2026-05). +# Garde a false jusqu'a trouver une source alternative (gridradar ? +# swissgrid open data ?). +enabled = false +ws_url = "wss://www.mainsfrequenz.de/frequenz.socket" + +[feeds.rte_eco2mix] +enabled = false +client_id = "" +client_secret = "" +poll_seconds = 900 + +# -- Foudre / atmosphère --------------------------------------------------- +[feeds.blitzortung] +enabled = true +ws_url = "wss://ws1.blitzortung.org:443/" + +# -- Aviation / mouvement -------------------------------------------------- +[feeds.opensky] +enabled = true +url = "https://opensky-network.org/api/states/all" +# API anonyme limitee a 1 req/15s. On laisse 20s de marge pour eviter +# les 429 sous drift d'horloge. +poll_seconds = 20 +bbox = [45.5, 4.6, 46.0, 5.2] + +# -- Pouls numérique ------------------------------------------------------- +[feeds.bluesky] +enabled = true +ws_url = "wss://jetstream2.us-east.bsky.network/subscribe?wantedCollections=app.bsky.feed.post" +sample_rate = 0.02 + +[feeds.mempool] +enabled = false +ws_url = "wss://mempool.space/api/v1/ws" + +[feeds.github] +enabled = false +url = "https://api.github.com/events" +poll_seconds = 30 + +[feeds.gcn] +enabled = false +client_id = "" +client_secret = "" diff --git a/data_feeds/config.toml b/data_feeds/config.toml index cad3da2..963afe3 100644 --- a/data_feeds/config.toml +++ b/data_feeds/config.toml @@ -65,7 +65,9 @@ poll_seconds = 60 # -- Réseau électrique ----------------------------------------------------- [feeds.netzfrequenz] -enabled = true +# WARN : mainsfrequenz.de a ferme son WS public (NXDOMAIN 2026-05). +# Voir alternatives : gridradar.net (auth requise), swissgrid open data. +enabled = false # Mainsfrequenz.de WebSocket (résolution ~200 ms, mesure Karlsruhe) ws_url = "wss://www.mainsfrequenz.de/frequenz.socket" # /data/grid/freq 50.000 ± 0.200 diff --git a/data_feeds/feeds/github.py b/data_feeds/feeds/github.py index db77961..d0a05cd 100644 --- a/data_feeds/feeds/github.py +++ b/data_feeds/feeds/github.py @@ -15,7 +15,9 @@ async def run(ctx) -> None: cfg = ctx.cfg url = cfg["url"] period = float(cfg.get("poll_seconds", 30)) - last_id = "" + # IDs GitHub sont des strings numeriques croissants : on compare en int + # (la compare string casse des que le nombre de digits change). + last_id: int = 0 async with httpx.AsyncClient(timeout=20.0, headers={"Accept": "application/vnd.github+json"}) as cli: while True: @@ -23,7 +25,11 @@ async def run(ctx) -> None: r = await cli.get(url) r.raise_for_status() for ev in reversed(r.json()): - eid = ev.get("id", "") + raw = ev.get("id", "") + try: + eid = int(raw) + except (TypeError, ValueError): + continue if eid <= last_id: continue ctx.send( diff --git a/data_feeds/feeds/pose.py b/data_feeds/feeds/pose.py index 00252d5..eb2f8d9 100644 --- a/data_feeds/feeds/pose.py +++ b/data_feeds/feeds/pose.py @@ -99,19 +99,28 @@ async def run(ctx) -> None: ok, frame = cap.read() return frame if ok else None + # ThreadPoolExecutor dedie : empeche l'inference longue (>20ms sur MPS) + # de monopoliser le pool partage et de bloquer les autres feeds. + import concurrent.futures + pool = concurrent.futures.ThreadPoolExecutor(max_workers=1, + thread_name_prefix="pose") + + def _infer(fr): + return model.predict(fr, device=device, conf=conf_thresh, + verbose=False, max_det=max_persons) + try: while True: t0 = time.monotonic() - frame = await loop.run_in_executor(None, _grab) + frame = await loop.run_in_executor(pool, _grab) if frame is None: await asyncio.sleep(period) continue h, w = frame.shape[:2] try: - results = model.predict( - frame, device=device, conf=conf_thresh, - verbose=False, max_det=max_persons, - ) + # Non-bloquant : l'event loop continue de servir les autres + # feeds pendant les ~20-80 ms d'inference. + results = await loop.run_in_executor(pool, _infer, frame) except Exception as e: # noqa: BLE001 LOG.warning("inference failed: %s", e) await asyncio.sleep(period) @@ -156,3 +165,4 @@ async def run(ctx) -> None: await asyncio.sleep(period - dt) finally: cap.release() + pool.shutdown(wait=False, cancel_futures=True) diff --git a/data_feeds/feeds/rte_eco2mix.py b/data_feeds/feeds/rte_eco2mix.py index f3f39a8..7cef722 100644 --- a/data_feeds/feeds/rte_eco2mix.py +++ b/data_feeds/feeds/rte_eco2mix.py @@ -12,6 +12,23 @@ LOG = logging.getLogger("feed.rte_eco2mix") TOKEN_URL = "https://digital.iservices.rte-france.com/token/oauth/" API_URL = "https://digital.iservices.rte-france.com/open_api/actual_generation/v1/actual_generations_per_production_type" +# Mapping des `production_type` RTE (verbose) vers nos categories courtes. +# L'API agrege HYDRO_* et WIND_* pour le contexte musical : on additionne. +_RTE_MAP = { + "NUCLEAR": "NUCLEAR", + "FOSSIL_GAS": "GAS", + "FOSSIL_HARD_COAL": "COAL", + "FOSSIL_OIL": "OIL", + "HYDRO_WATER_RESERVOIR": "HYDRO", + "HYDRO_RUN_OF_RIVER_AND_POUNDAGE":"HYDRO", + "HYDRO_PUMPED_STORAGE": "HYDRO", + "WIND_ONSHORE": "WIND", + "WIND_OFFSHORE": "WIND", + "SOLAR": "SOLAR", + "BIOMASS": "BIOENERGY", + "WASTE": "BIOENERGY", +} + async def _get_token(cli: httpx.AsyncClient, cid: str, csec: str) -> tuple[str, float]: r = await cli.post(TOKEN_URL, auth=(cid, csec), @@ -38,12 +55,18 @@ async def run(ctx) -> None: r = await cli.get(API_URL, headers={"Authorization": f"Bearer {token}"}) r.raise_for_status() j = r.json() - latest = {} + latest: dict[str, float] = {} for series in (j.get("actual_generations_per_production_type") or []): - typ = series.get("production_type", "?") + raw = series.get("production_type", "?") + typ = _RTE_MAP.get(raw) + if typ is None: + continue # type non-mappe, ignore vals = series.get("values") or [] if vals: - latest[typ] = float(vals[-1].get("value", 0.0)) + # Aggregation : on additionne les sous-categories + # (ex: WIND_ONSHORE + WIND_OFFSHORE -> WIND). + latest[typ] = latest.get(typ, 0.0) \ + + float(vals[-1].get("value", 0.0)) # mapping standard RTE → ordre des args ctx.send( "mix", diff --git a/data_feeds/feeds/swpc.py b/data_feeds/feeds/swpc.py index b646362..6cfb23b 100644 --- a/data_feeds/feeds/swpc.py +++ b/data_feeds/feeds/swpc.py @@ -68,14 +68,20 @@ async def run(ctx) -> None: pass if urls["kp"]: j = await _fetch_json(cli, urls["kp"]) - row = _last_row(j) - if row: - # ["time_tag","Kp","a_running","station_count"] + # NOAA renvoie maintenant une liste de dicts pour Kp + # ({"time_tag":..., "Kp":..., "a_running":...}) au lieu + # de la liste-de-listes historique. On supporte les deux. + if j: + last = j[-1] try: - kp = float(row[1]) - a = float(row[2]) + if isinstance(last, dict): + kp = float(last.get("Kp", 0.0)) + a = float(last.get("a_running", 0.0)) + else: + kp = float(last[1]) + a = float(last[2]) ctx.send("kp", kp, a) - except (TypeError, ValueError): + except (TypeError, ValueError, KeyError, IndexError): pass if urls["xray"]: j = await _fetch_json(cli, urls["xray"]) @@ -86,5 +92,5 @@ async def run(ctx) -> None: l = float(long_.get("flux", 0.0)) if long_ else 0.0 ctx.send("xray", s, l, _flare_class_norm(l)) except Exception as e: # noqa: BLE001 - LOG.warning("fetch failed: %s", e) + LOG.warning("fetch failed: %s: %s", type(e).__name__, e) await asyncio.sleep(period) diff --git a/data_feeds/feeds/usgs.py b/data_feeds/feeds/usgs.py index 7c2e1f4..298d2ff 100644 --- a/data_feeds/feeds/usgs.py +++ b/data_feeds/feeds/usgs.py @@ -2,6 +2,7 @@ from __future__ import annotations import asyncio +import collections import logging import time @@ -16,7 +17,11 @@ async def run(ctx) -> None: cfg = ctx.cfg url = cfg["url"] period = float(cfg.get("poll_seconds", 60)) - seen: set[str] = set() + # OrderedDict avec eviction LRU : conserve les 4096 derniers IDs vus + # dans l'ORDRE d'arrivee. set() perdait l'ordre au pruning, ce qui + # pouvait re-emettre un evenement deja vu. + seen: "collections.OrderedDict[str, None]" = collections.OrderedDict() + SEEN_MAX = 4096 rate = RateMeter(window=3600.0) async with httpx.AsyncClient(timeout=20.0) as cli: while True: @@ -34,7 +39,7 @@ async def run(ctx) -> None: fid = feat.get("id") if not fid or fid in seen: continue - seen.add(fid) + seen[fid] = None props = feat.get("properties") or {} coords = (feat.get("geometry") or {}).get("coordinates") or [0, 0, 0] mag = float(props.get("mag") or 0.0) @@ -45,7 +50,8 @@ async def run(ctx) -> None: rate.tick() ctx.send("rate", rate.rate * 3600.0) - # garde la mémoire bornée - if len(seen) > 4096: - seen = set(list(seen)[-2048:]) + # garde la mémoire bornée — evict les plus anciens en preservant + # l'ordre d'insertion (LRU front, head most-recent). + while len(seen) > SEEN_MAX: + seen.popitem(last=False) await asyncio.sleep(period)