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.
This commit is contained in:
L'électron rare
@@ -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 = ""
|
||||
|
||||
@@ -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*<b>(\d+)</b>", 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)
|
||||
@@ -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)
|
||||
@@ -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)
|
||||
@@ -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)
|
||||
Reference in New Issue
Block a user