From 60112b07908cfafe4ef2d78787a74a025ea0ac94 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?L=27=C3=A9lectron=20rare?= <108685187+electron-rare@users.noreply.github.com> Date: Wed, 13 May 2026 19:17:36 +0200 Subject: [PATCH] feat(data-feeds): GDELT, Wikipedia, Tides, ATC 4 nouvelles sources open-data publiques, sans cle : - gdelt : GDELT 2.0 events CSV.zip toutes les 15 min ; emet /event lat lon tone country eid + /batch n countries avg_tone. Dedup par GLOBALEVENTID, ring buffer 8192. - wikimedia : EventStreams SSE firehose recentchange (toutes langues). Sample ~5% des edits + rate per second toutes les 5 s. Reconnect auto sur disconnect. - tides : NOAA CO-OPS water level observe + predit pour une station (Boston par defaut, station code configurable) + phase lunaire calculee (Conway approx, sans dep). Poll 6 min. - atc : LiveATC hub listeners scraping (KJFK/KLAX/KSFO/KORD/ EGLL/LFPG par defaut). Polite 500ms entre hubs, total poll 5 min. --- data_feeds/config.data-only.toml | 23 ++++++++ data_feeds/feeds/atc.py | 60 +++++++++++++++++++++ data_feeds/feeds/gdelt.py | 93 ++++++++++++++++++++++++++++++++ data_feeds/feeds/tides.py | 64 ++++++++++++++++++++++ data_feeds/feeds/wikimedia.py | 62 +++++++++++++++++++++ 5 files changed, 302 insertions(+) create mode 100644 data_feeds/feeds/atc.py create mode 100644 data_feeds/feeds/gdelt.py create mode 100644 data_feeds/feeds/tides.py create mode 100644 data_feeds/feeds/wikimedia.py diff --git a/data_feeds/config.data-only.toml b/data_feeds/config.data-only.toml index 1b660a3..32d4555 100644 --- a/data_feeds/config.data-only.toml +++ b/data_feeds/config.data-only.toml @@ -119,6 +119,29 @@ poll_seconds = 3600 enabled = true poll_seconds = 60 +# -- GDELT (evenements monde 15-min) -------------------------------------- +[feeds.gdelt] +enabled = true +poll_seconds = 900 + +# -- Wikipedia recent changes --------------------------------------------- +[feeds.wikimedia] +enabled = true +sample_rate = 0.05 +rate_window_s = 5 + +# -- NOAA tides + lune ---------------------------------------------------- +[feeds.tides] +enabled = true +station = "8443970" +poll_seconds = 360 + +# -- LiveATC hubs --------------------------------------------------------- +[feeds.atc] +enabled = true +hubs = ["KJFK", "KLAX", "KSFO", "KORD", "EGLL", "LFPG"] +poll_seconds = 300 + [feeds.gcn] enabled = false client_id = "" diff --git a/data_feeds/feeds/atc.py b/data_feeds/feeds/atc.py new file mode 100644 index 0000000..26bb6d7 --- /dev/null +++ b/data_feeds/feeds/atc.py @@ -0,0 +1,60 @@ +"""LiveATC feeds metadata — pas de pull audio, juste compteur de +streams actifs et indicateur d'activite par hub aeroport. + +LiveATC.net expose un endpoint stats JSON simplifie pour quelques +aeroports majeurs (KJFK, KLAX, KSFO, EGLL, LFPG). On polle pour +chacun le nombre d'auditeurs courant (proxy pour 'activite ATC'). + +OSC out : + /data/atc/hub icao listeners + /data/atc/total total_listeners n_hubs +""" +from __future__ import annotations + +import asyncio +import logging +import re + +import httpx + +LOG = logging.getLogger("feed.atc") + +# LiveATC publie un page HTML par feed avec "Listeners: N" — on parse +# ca via regex au lieu d'un API officielle (pas disponible). +HUB_URL = "https://www.liveatc.net/search/?icao={icao}" +LISTENERS_RE = re.compile(r"Listeners:\s*(\d+)", re.I) + + +async def _fetch_hub(cli: httpx.AsyncClient, icao: str + ) -> int: + r = await cli.get(HUB_URL.format(icao=icao), + headers={"User-Agent": "av-live-data-feeds/1.0"}) + r.raise_for_status() + matches = LISTENERS_RE.findall(r.text) + if not matches: + return 0 + return sum(int(m) for m in matches) + + +async def run(ctx) -> None: + cfg = ctx.cfg + hubs = cfg.get("hubs", + ["KJFK", "KLAX", "KSFO", "KORD", "EGLL", "LFPG"]) + period = float(cfg.get("poll_seconds", 300.0)) # 5 min + async with httpx.AsyncClient(timeout=20.0) as cli: + while True: + total = 0 + active = 0 + for icao in hubs: + try: + listeners = await _fetch_hub(cli, icao) + except Exception as e: # noqa: BLE001 + LOG.warning("atc %s failed: %s", icao, e) + continue + ctx.send("hub", icao, float(listeners)) + total += listeners + if listeners > 0: + active += 1 + await asyncio.sleep(0.5) # polite scrape + ctx.send("total", float(total), float(active)) + await asyncio.sleep(period) diff --git a/data_feeds/feeds/gdelt.py b/data_feeds/feeds/gdelt.py new file mode 100644 index 0000000..2c24876 --- /dev/null +++ b/data_feeds/feeds/gdelt.py @@ -0,0 +1,93 @@ +"""GDELT Project — feed des evenements 'world' 15-min. + +Source : GDELT 2.0 Events CSV ; updates toutes les 15 min. +On compte les evenements + extrait les top countries / themes du +dernier batch. + +OSC out : + /data/gdelt/batch n_events n_countries avg_tone + /data/gdelt/event lat lon tone country_code root_event_id +""" +from __future__ import annotations + +import asyncio +import collections +import logging +import time +import zipfile +from io import BytesIO + +import httpx + +LOG = logging.getLogger("feed.gdelt") + +# GDELT v2 master file list ; on prend juste le dernier .export.CSV.zip +MASTER = "http://data.gdeltproject.org/gdeltv2/lastupdate.txt" + + +async def _fetch_latest_csv(cli: httpx.AsyncClient) -> list[list[str]]: + r = await cli.get(MASTER) + r.raise_for_status() + # 3 lignes : events, mentions, gkg ; on prend events (premiere ligne) + first = (r.text.strip().splitlines() or [""])[0].split(" ") + if len(first) < 3: + return [] + url = first[2] + rz = await cli.get(url, follow_redirects=True) + rz.raise_for_status() + with zipfile.ZipFile(BytesIO(rz.content)) as zf: + name = zf.namelist()[0] + text = zf.read(name).decode("utf-8", errors="ignore") + return [line.split("\t") for line in text.splitlines() if line] + + +async def run(ctx) -> None: + cfg = ctx.cfg + period = float(cfg.get("poll_seconds", 900.0)) # 15 min + seen: collections.OrderedDict[str, None] = collections.OrderedDict() + SEEN_MAX = 8192 + async with httpx.AsyncClient(timeout=60.0) as cli: + while True: + try: + rows = await _fetch_latest_csv(cli) + if not rows: + LOG.warning("gdelt: empty batch") + await asyncio.sleep(period) + continue + count = 0 + tones = [] + countries: collections.Counter[str] = collections.Counter() + # GDELT 2.0 events CSV : 61 colonnes. + # idx 0 = GLOBALEVENTID, 7 = Actor1CountryCode, + # 34 = AvgTone, 39 = ActionGeo_Lat, 40 = ActionGeo_Long + for row in rows: + if len(row) < 41: + continue + eid = row[0] + if not eid or eid in seen: + continue + seen[eid] = None + if len(seen) > SEEN_MAX: + seen.popitem(last=False) + count += 1 + try: + tone = float(row[34] or "0") + except ValueError: + tone = 0.0 + tones.append(tone) + cc = row[7].strip()[:3] + if cc: + countries[cc] += 1 + try: + lat = float(row[39] or "0") + lon = float(row[40] or "0") + except ValueError: + lat = lon = 0.0 + if lat or lon: + ctx.send("event", lat, lon, tone, cc, eid) + avg_tone = (sum(tones) / len(tones)) if tones else 0.0 + ctx.send("batch", float(count), + float(len(countries)), float(avg_tone)) + except Exception as e: # noqa: BLE001 + LOG.warning("gdelt fetch failed: %s", e) + await asyncio.sleep(period) diff --git a/data_feeds/feeds/tides.py b/data_feeds/feeds/tides.py new file mode 100644 index 0000000..fcfc52d --- /dev/null +++ b/data_feeds/feeds/tides.py @@ -0,0 +1,64 @@ +"""NOAA tides (station configurable) + phase lunaire calculee. + +NOAA CO-OPS API : observed water level + predicted, station codee +(defaut Boston). Phase lunaire algorithme Conway approx (sans dep). + +OSC out : + /data/tides/level water_level_m predicted_m residual_m + /data/tides/moon phase_0_1 illum_0_1 +""" +from __future__ import annotations + +import asyncio +import logging +import math +import time + +import httpx + +LOG = logging.getLogger("feed.tides") + +NOAA_URL = ("https://api.tidesandcurrents.noaa.gov/api/prod/datagetter" + "?product={product}&application=AV-Live&format=json" + "&time_zone=gmt&datum=MLLW&units=metric" + "&date=latest&station={station}") + + +def _moon_phase(t: float) -> tuple[float, float]: + """t epoch -> (phase 0..1 ou 0 = new, 0.5 = full ; illum 0..1). + Approx Conway, +-1 jour de precision suffisant.""" + # Reference : 2000-01-06 18:14 UTC ~ new moon + new = 946755300.0 + cycle = 29.530588853 * 86400.0 + phase = ((t - new) % cycle) / cycle + illum = (1.0 - math.cos(2 * math.pi * phase)) / 2.0 + return phase, illum + + +async def _fetch_level(cli: httpx.AsyncClient, station: str + ) -> tuple[float, float]: + obs_url = NOAA_URL.format(product="water_level", station=station) + pred_url = NOAA_URL.format(product="predictions", station=station) + obs = await cli.get(obs_url) + pred = await cli.get(pred_url) + obs.raise_for_status() + pred.raise_for_status() + o = obs.json().get("data", [{}])[0] + p = pred.json().get("predictions", [{}])[0] + return float(o.get("v") or 0.0), float(p.get("v") or 0.0) + + +async def run(ctx) -> None: + cfg = ctx.cfg + station = str(cfg.get("station", "8443970")) # Boston by default + period = float(cfg.get("poll_seconds", 360.0)) # 6 min + async with httpx.AsyncClient(timeout=20.0) as cli: + while True: + try: + obs, pred = await _fetch_level(cli, station) + ctx.send("level", obs, pred, obs - pred) + except Exception as e: # noqa: BLE001 + LOG.warning("tides fetch failed: %s", e) + phase, illum = _moon_phase(time.time()) + ctx.send("moon", float(phase), float(illum)) + await asyncio.sleep(period) diff --git a/data_feeds/feeds/wikimedia.py b/data_feeds/feeds/wikimedia.py new file mode 100644 index 0000000..29faa30 --- /dev/null +++ b/data_feeds/feeds/wikimedia.py @@ -0,0 +1,62 @@ +"""Wikipedia recent changes — Wikimedia EventStreams SSE. + +Firehose des modifications Wikipedia toutes langues confondues. +Tres bavard (~10-30 events/s) — on echantillonne et on emet une +fraction + des compteurs. + +OSC out : + /data/wikimedia/edit lang title bot (sample) + /data/wikimedia/rate per_second +""" +from __future__ import annotations + +import asyncio +import json +import logging +import random +import time + +import httpx + +LOG = logging.getLogger("feed.wikimedia") + +URL = "https://stream.wikimedia.org/v2/stream/recentchange" + + +async def run(ctx) -> None: + cfg = ctx.cfg + sample = float(cfg.get("sample_rate", 0.05)) # emit 5% des edits + window = float(cfg.get("rate_window_s", 5.0)) + while True: + try: + async with httpx.AsyncClient(timeout=None) as cli: + async with cli.stream("GET", URL, + headers={"Accept": "text/event-stream"} + ) as r: + r.raise_for_status() + bucket = 0 + bucket_start = time.monotonic() + async for line in r.aiter_lines(): + if not line.startswith("data:"): + continue + try: + ev = json.loads(line[5:].strip()) + except json.JSONDecodeError: + continue + bucket += 1 + # Sample emit + if random.random() < sample: + lang = (ev.get("wiki") or "").replace("wiki", "") + title = str(ev.get("title", ""))[:64] + bot = 1.0 if ev.get("bot") else 0.0 + ctx.send("edit", lang, title, bot) + # Periodic rate + now = time.monotonic() + if now - bucket_start >= window: + per_s = bucket / (now - bucket_start) + ctx.send("rate", float(per_s)) + bucket = 0 + bucket_start = now + except Exception as e: # noqa: BLE001 + LOG.warning("wikimedia stream error: %s — reconnect 5s", e) + await asyncio.sleep(5.0)