Claude
ef67df6e00
- mehrere Quellen je Land (country), Grenzen aus Natural Earth, OSM und Kachelausschnitt erweitert - Fahrpläne glätten: Unterordner im Zip, aufgefüllte Felder, fehlende parent_station, level_id - Gattung aus der Zugnummer (ÖBB 'RJ 658'), Fernverkehr auch über Routentyp 101/102/105 - Echtzeit: Verspätung vor Uhrzeit (Renfe), Fahrten ohne Datum, Key in der Adresse (Trafiklab) - httpx-Logs aus (würden Keys zeigen) Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01X48pE4bzjHej8w9HyZXCjb
50 lines
2.2 KiB
Python
50 lines
2.2 KiB
Python
"""GTFS-RT Trip Updates von opentransportdata.swiss holen und auf das Nötigste eindampfen."""
|
|
import httpx
|
|
from google.transit import gtfs_realtime_pb2 as pb
|
|
|
|
from . import config as C
|
|
|
|
CANCELED = pb.TripDescriptor.CANCELED
|
|
SKIPPED = pb.TripUpdate.StopTimeUpdate.SKIPPED
|
|
|
|
|
|
def parse(data: bytes, known_trips: dict):
|
|
"""-> (feed_version, {(trip_id, start_date): (ausgefallen, [(seq, stop_id, ank, ank_absolut, abf, abf_absolut, fällt_aus), ...])})
|
|
seq ist None, wenn die Quelle nur die Haltestellen-ID liefert (Frankreich, Niederlande).
|
|
ank/abf sind Verspätung in s, oder Unix-Zeit wenn *_absolut; None = keine Angabe."""
|
|
feed = pb.FeedMessage()
|
|
feed.ParseFromString(data)
|
|
out = {}
|
|
for e in feed.entity:
|
|
if not e.HasField("trip_update"):
|
|
continue
|
|
tu = e.trip_update
|
|
if tu.trip.trip_id not in known_trips:
|
|
continue # nicht Bahn
|
|
upd = []
|
|
for s in tu.stop_time_update:
|
|
ev = []
|
|
for name in ("arrival", "departure"):
|
|
if s.HasField(name):
|
|
x = getattr(s, name)
|
|
# Verspätung vor Uhrzeit: Renfe liefert beides, und die Uhrzeit ist dort oft falsch
|
|
ev += [x.delay, False] if x.HasField("delay") else [x.time, True] if x.HasField("time") else [None, False]
|
|
else:
|
|
ev += [None, False]
|
|
upd.append((s.stop_sequence if s.HasField("stop_sequence") else None, s.stop_id, *ev,
|
|
s.schedule_relationship == SKIPPED))
|
|
out[(tu.trip.trip_id, tu.trip.start_date)] = (tu.trip.schedule_relationship == CANCELED, upd)
|
|
version = feed.header.feed_version if feed.header.HasField("feed_version") else None
|
|
return version, out
|
|
|
|
|
|
def fetch(url: str, key: str = "") -> bytes:
|
|
headers = {"User-Agent": C.USER_AGENT, "ET-Client-Name": "trainpulse"} # Entur möchte eine Kennung
|
|
if "{key}" in url: # Schweden: Key in der Adresse
|
|
url = url.replace("{key}", key)
|
|
elif key:
|
|
headers["Authorization"] = f"Bearer {key}"
|
|
r = httpx.get(url, follow_redirects=True, timeout=30, headers=headers)
|
|
r.raise_for_status()
|
|
return r.content
|