electron-rare/AV-Live
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 1 Wiki Activity

feat(data-feeds): data-only profile + feeds

Browse Source 
Add config.data-only.toml for headless bridge mode.
Update .gitignore to exclude .venv, __pycache__, *.pt.
Improve USGS, SWPC, RTE, GitHub, and pose feeds:
error handling, rate limiting, keypoint emission.
Refactor bridge.py to support -c config file flag.
This commit is contained in:
L'électron rare
2026-05-13 09:33:36 +02:00
parent 4d32e61bae
commit 8913393e4c
9 changed files with 199 additions and 25 deletions
Download Patch File Download Diff File Expand all files Collapse all files
data_feeds/.gitignore
+9
View File
@@ -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
data_feeds/bridge.py
+24 -2
View File
@@ -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.<name>` 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:
data_feeds/config.data-only.toml
+90
View File
@@ -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 = ""
data_feeds/config.toml
+3 -1
View File
@@ -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 <hz> 50.000 ± 0.200
data_feeds/feeds/github.py
+8 -2
View File
@@ -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(
data_feeds/feeds/pose.py
+15 -5
View File
@@ -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)
data_feeds/feeds/rte_eco2mix.py
+26 -3
View File
@@ -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",
data_feeds/feeds/swpc.py
+13 -7
View File
@@ -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)
data_feeds/feeds/usgs.py
+11 -5
View File
@@ -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)