Files
Trainpulse/trainpulse/feed.py
T

424 lines
20 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""Hält einen aufbereiteten Bahn-Fahrplan im Speicher und berechnet daraus Zugpositionen
entlang der Strecke (Fahrplan + Echtzeit-Verspätung -> Position auf dem Streckenverlauf)."""
import csv
import datetime as dt
import json
import math
import mmap
import re
from array import array
from bisect import bisect_right
from pathlib import Path
from zoneinfo import ZoneInfo
from . import config as C
TZ = ZoneInfo(C.TZ)
CAL_DAYS = 14 # so viele Tage Ausnahmen aus calendar_dates laden (die Datei ist riesig)
def hms(s: str) -> int:
h, m, sec = s.split(":")
return int(h) * 3600 + int(m) * 60 + int(sec)
def day_start(d: dt.date) -> float:
# GTFS-Zeiten zählen ab "Mittag minus 12 h" – stimmt auch an Tagen mit Zeitumstellung
return dt.datetime(d.year, d.month, d.day, 12, tzinfo=TZ).timestamp() - 43200
def read(path: Path):
with open(path, encoding="utf-8-sig", newline="") as f:
r = csv.reader(f)
h = {k: i for i, k in enumerate(next(r))}
for row in r:
yield h, row
def xy(lon, lat, lat0): # lokale Projektion in Metern, reicht für Abstände entlang einer Strecke
return lon * 111320 * math.cos(math.radians(lat0)), lat * 110540
BRAKE_M = 150 # so viele Meter vor/nach einem Halt bremst bzw. beschleunigt der Zug
def travelled(length: float, duration: float, t: float) -> float:
"""Zurückgelegte Meter auf einem Abschnitt: sanft anfahren, konstant fahren, sanft bremsen
(trapezförmiges Geschwindigkeitsprofil), statt mit voller Geschwindigkeit in den Halt zu fahren."""
if length <= 0 or duration <= 0:
return 0.0
t = max(0.0, min(t, duration))
ta = min(duration / 2, 2 * BRAKE_M * duration / (length + 2 * BRAKE_M)) # Zeit fürs Anfahren/Bremsen
vmax = length / (duration - ta)
a = vmax / ta
if t < ta:
return 0.5 * a * t * t
if t > duration - ta:
return length - 0.5 * a * (duration - t) ** 2
return 0.5 * a * ta * ta + vmax * (t - ta)
# Gattungen des Fernverkehrs; alles andere (RE, RB, S, R, TER, …) gilt als Nahverkehr
LONG_DISTANCE = {"ICE", "IC", "ICN", "ICD", "EC", "ECE", "ECM", "EN", "NJ", "RJ", "RJX", "TGV", "OUIGO", "LYRIA",
"THA", "EST", "EUROSTAR", "EUROCITY", "NIGHTJET", "EUROPEAN", "FLX", "IR", "PE", "D"}
# Frankreich (SNCF): Gattung steht nur in der Trip-ID ("…_F:OUI:…"), die Linie heisst oft "INCONNU"
FR_CATEGORY = {"OUI": "TGV", "OGO": "OUIGO", "LYR": "Lyria", "ICE": "ICE", "IC": "IC", "ICN": "IC",
"TER": "TER", "CTE": "TER", "NAV": "Navette", "TRN": "TER", "CRE": "TER"}
FR_TRIP = re.compile(r"_F:([A-Z]+):")
# Niederlande (OVapi): lange Gattungsnamen kürzen, damit die Beschriftung auf der Karte passt
LINE_ALIAS = {"Intercity": "IC", "Intercity direct": "ICD", "Sprinter": "SPR", "Nightjet": "NJ", "Eurocity": "EC"}
LINE_PREFIX = re.compile(r"^(Stoptrein|Sneltrein) (?=\S)") # "Stoptrein RS4" -> "RS4"
def long_distance(cat: str, line: str) -> bool:
# Schweiz: Gattung in route_desc; gtfs.de: nur in der Linie ("ICE 75", "IC 3")
m = re.match(r"[A-Za-z]+", cat or line or "")
return bool(m) and m.group(0).upper() in LONG_DISTANCE
MIN_DWELL = 40 # s; im Fahrplan steht oft Ankunft = Abfahrt, dann würde der Zug nie sichtbar halten
def with_min_dwell(arr, dep):
"""Halte mit weniger als MIN_DWELL Haltezeit für die Animation verlängern: etwas früher ankommen,
etwas später abfahren – höchstens ein Viertel der Fahrzeit davor/danach."""
pa, pd = list(arr), list(dep)
for k in range(1, len(arr) - 1):
short = MIN_DWELL - (dep[k] - arr[k])
if short > 0:
h = min(short / 2, (arr[k] - dep[k - 1]) / 4, (arr[k + 1] - dep[k]) / 4)
if h > 0:
pa[k], pd[k] = arr[k] - h, dep[k] + h
return pa, pd
def simplify(p: list, tol_m: float, fixed=()) -> list:
"""Douglas-Peucker auf [lon, lat, t]-Punkten (Toleranz in Metern, grob über die Breite umgerechnet).
Punkte in `fixed` bleiben immer erhalten."""
if len(p) < 3:
return p
kx = 111320 * math.cos(math.radians(p[0][1]))
keep = {0, len(p) - 1, *fixed}
cuts = sorted(keep)
stack = list(zip(cuts, cuts[1:]))
while stack:
a, b = stack.pop()
(x1, y1), (x2, y2) = (p[a][0] * kx, p[a][1] * 110540), (p[b][0] * kx, p[b][1] * 110540)
dx, dy = x2 - x1, y2 - y1
n = math.hypot(dx, dy)
best, idx = 0.0, -1
for i in range(a + 1, b):
x, y = p[i][0] * kx, p[i][1] * 110540
d = abs(dy * x - dx * y + x2 * y1 - y2 * x1) / n if n else math.hypot(x - x1, y - y1)
if d > best:
best, idx = d, i
if best > tol_m:
keep.add(idx)
stack += [(a, idx), (idx, b)]
return [p[i] for i in sorted(keep)]
class Feed:
def __init__(self, path: Path, today: dt.date | None = None):
self.path = path
self.meta = json.loads((path / "meta.json").read_text())
self.version = self.meta["feed_version"]
self.stop_index, self.stop_name, self.stop_lat, self.stop_lon, self.stop_platform = {}, [], [], [], []
self.stop_group = array("i") # Bahnhof eines Halts (alle Gleise eines Bahnhofs -> gleiche Zahl)
groups = {}
for h, r in read(path / "stops.txt"):
parent = r[h["parent_station"]] if "parent_station" in h else ""
self.stop_group.append(groups.setdefault(parent or r[h["stop_id"]], len(groups)))
self.stop_index[r[h["stop_id"]]] = len(self.stop_name)
self.stop_name.append(r[h["stop_name"]])
self.stop_lat.append(float(r[h["stop_lat"]]))
self.stop_lon.append(float(r[h["stop_lon"]]))
self.stop_platform.append(r[h["platform_code"]] if "platform_code" in h else "")
agencies = {r[h["agency_id"]]: r[h["agency_name"]] for h, r in read(path / "agency.txt")}
self.routes = {}
for h, r in read(path / "routes.txt"):
line = LINE_PREFIX.sub("", r[h["route_short_name"]] or r[h.get("route_long_name", 0)])
self.routes[r[h["route_id"]]] = {
"line": LINE_ALIAS.get(line, "" if line == "INCONNU" else line),
"cat": r[h["route_desc"]] if "route_desc" in h else "",
"agency": agencies.get(r[h["agency_id"]], "") if "agency_id" in h else "",
}
self.trip_index, self.trip_ids, self.trip_route, self.trip_service = {}, [], [], []
self.trip_headsign, self.trip_number, self.trip_shape = [], [], []
for h, r in read(path / "trips.txt"):
self.trip_index[r[h["trip_id"]]] = len(self.trip_route)
self.trip_ids.append(r[h["trip_id"]])
self.trip_route.append(r[h["route_id"]])
self.trip_service.append(r[h["service_id"]])
self.trip_headsign.append(r[h["trip_headsign"]] if "trip_headsign" in h else "")
self.trip_number.append(r[h["trip_short_name"]] if "trip_short_name" in h else "")
self.trip_shape.append(r[h["shape_id"]] if "shape_id" in h else "")
route = self.routes[r[h["route_id"]]]
m = FR_TRIP.search(r[h["trip_id"]]) if not route["cat"] else None
if m:
route["cat"] = FR_CATEGORY.get(m.group(1), m.group(1))
if self.trip_headsign[-1].isdigit() and not self.trip_number[-1]: # SNCF: Ziel = Zugnummer
self.trip_number[-1], self.trip_headsign[-1] = self.trip_headsign[-1], ""
for route in self.routes.values():
route["line"] = route["line"] or route["cat"]
route["fv"] = long_distance(route["cat"], route["line"])
# Haltezeiten flach in Arrays; off[i]..off[i+1] sind die Halte von Fahrt i
n = len(self.trip_route)
self.off = array("i", [0]) * (n + 1)
self.st_stop, self.st_seq, self.st_arr, self.st_dep = array("i"), array("i"), array("i"), array("i")
self.st_dist = array("f") # Meter entlang des Streckenverlaufs (von pfaedle), -1 = unbekannt
rows = {}
for h, r in read(path / "stop_times.txt"):
dist = r[h["shape_dist_traveled"]] if "shape_dist_traveled" in h else ""
rows.setdefault(self.trip_index[r[h["trip_id"]]], []).append(
(int(r[h["stop_sequence"]]), self.stop_index[r[h["stop_id"]]], hms(r[h["arrival_time"]]),
hms(r[h["departure_time"]]), float(dist) if dist else -1.0))
for ti in range(n):
self.off[ti] = len(self.st_stop)
for seq, s, a, d, m in sorted(rows.pop(ti, ())):
self.st_seq.append(seq)
self.st_stop.append(s)
self.st_arr.append(a)
self.st_dep.append(d)
self.st_dist.append(m)
self.off[n] = len(self.st_stop)
for ti in range(n): # gtfs.de liefert kein Fahrtziel: dann der letzte Halt
if not self.trip_headsign[ti] and self.off[ti + 1] > self.off[ti]:
self.trip_headsign[ti] = self.stop_name[self.st_stop[self.off[ti + 1] - 1]]
self.shape_index, self.shape_data = {}, None
if (path / "shapes.bin").exists() and (path / "shapes.bin").stat().st_size:
self.shape_index = json.loads((path / "shapes.json").read_text())
with open(path / "shapes.bin", "rb") as f:
self.shape_data = memoryview(mmap.mmap(f.fileno(), 0, access=mmap.ACCESS_READ)).cast("f")
self.calendar = {}
for h, r in read(path / "calendar.txt") if (path / "calendar.txt").exists() else (): # SNCF: nur Ausnahmen
self.calendar[r[h["service_id"]]] = (
[r[h[d]] == "1" for d in ("monday", "tuesday", "wednesday", "thursday", "friday", "saturday", "sunday")],
r[h["start_date"]], r[h["end_date"]])
today = today or dt.datetime.now(TZ).date()
self.cal_from = today - dt.timedelta(days=1)
self.cal_until = today + dt.timedelta(days=CAL_DAYS)
window = {(self.cal_from + dt.timedelta(days=i)).strftime("%Y%m%d") for i in range(CAL_DAYS + 2)}
self.exceptions = {} # datum -> {service_id: 1 (fährt) | 2 (fährt nicht)}
cd = path / "calendar_dates.txt"
if cd.exists():
with open(cd, encoding="utf-8-sig") as f:
h = next(f).strip().replace('"', "").split(",") # Spaltenreihenfolge ist je Quelle anders
si, di, ti_ = h.index("service_id"), h.index("date"), h.index("exception_type")
for line in f:
v = line.rstrip("\r\n").replace('"', "").split(",") # kurze Datei ohne Kommas in Werten
if v[di] in window:
self.exceptions.setdefault(v[di], {})[v[si]] = int(v[ti_])
self._day_cache = {}
self._geom_cache = {}
# ---------- Kalender ----------
def trips_on(self, d: dt.date):
"""Fahrten eines Betriebstags, sortiert nach Abfahrt: (start, ende, trip_index)."""
if d not in self._day_cache:
ds = d.strftime("%Y%m%d")
exc = self.exceptions.get(ds, {})
active = {sid for sid, (days, a, b) in self.calendar.items() if a <= ds <= b and days[d.weekday()]}
active |= {sid for sid, t in exc.items() if t == 1}
active -= {sid for sid, t in exc.items() if t == 2}
lst = sorted((self.st_dep[self.off[ti]], self.st_arr[self.off[ti + 1] - 1], ti)
for ti, sid in enumerate(self.trip_service)
if sid in active and self.off[ti + 1] - self.off[ti] >= 2)
if len(self._day_cache) > 4:
self._day_cache.clear()
self._day_cache[d] = lst
return self._day_cache[d]
def covers(self, d: dt.date) -> bool:
return self.cal_from <= d <= self.cal_until
# ---------- Geometrie ----------
def geometry(self, ti: int):
"""Streckenverlauf einer Fahrt: Punkte, kumulierte Meter, Meter-Position jedes Halts.
Ohne (brauchbaren) Streckenverlauf: Luftlinie zwischen den Halten."""
o0, o1 = self.off[ti], self.off[ti + 1]
key = (self.trip_shape[ti], tuple(self.st_stop[o0:o1]))
g = self._geom_cache.get(key)
if g:
return g
shp = self.shape_index.get(self.trip_shape[ti])
stop_d = list(self.st_dist[o0:o1])
if shp and shp[1] >= 2 and min(stop_d) >= 0:
v = self.shape_data[shp[0] * 3:(shp[0] + shp[1]) * 3].tolist()
pts, cum = list(zip(v[0::3], v[1::3])), v[2::3]
else:
pts = [(self.stop_lon[s], self.stop_lat[s]) for s in self.st_stop[o0:o1]]
lat0 = pts[0][1]
P = [xy(lon, lat, lat0) for lon, lat in pts]
cum = [0.0]
for (x1, y1), (x2, y2) in zip(P, P[1:]):
cum.append(cum[-1] + math.hypot(x2 - x1, y2 - y1))
stop_d = cum
if len(self._geom_cache) > 20000:
self._geom_cache.clear()
g = (pts, cum, stop_d)
self._geom_cache[key] = g
return g
def track(self, ti: int, dists: list, step: float, tolerance: float = 3.0) -> list:
"""Gleispunkte von der aktuellen Position bis zur Position in len(dists)-1 Schritten, je mit der
Zeit (s ab jetzt), zu der der Zug dort ist. Gerade Stücke werden auf ~3 m vereinfacht."""
pts, cum, _ = self.geometry(ti)
out, fixed = [], []
for k in range(len(dists)):
d0 = dists[k]
fixed.append(len(out)) # Zeitpunkte alle `step` s behalten, sonst ginge das Bremsen verloren
out.append((d0, k * step))
if k + 1 < len(dists) and dists[k + 1] > d0:
d1 = dists[k + 1]
for j in range(bisect_right(cum, d0), bisect_right(cum, d1)):
if d0 < cum[j] < d1:
out.append((cum[j], (k + (cum[j] - d0) / (d1 - d0)) * step))
res = [[*self.point_at((pts, cum, None), d), round(t, 2), round(d)] for d, t in out] # d: m ab Fahrtbeginn
return simplify(res, tolerance, fixed)
@staticmethod
def point_at(g, d: float):
pts, cum, _ = g
k = max(0, min(bisect_right(cum, d) - 1, len(pts) - 2))
seg = cum[k + 1] - cum[k]
t = 0.0 if seg <= 0 else (d - cum[k]) / seg
(x1, y1), (x2, y2) = pts[k], pts[k + 1]
return round(x1 + t * (x2 - x1), 6), round(y1 + t * (y2 - y1), 6)
# ---------- Echtzeit ----------
def times(self, ti: int, ds: float, upd):
"""Ankunft/Abfahrt je Halt (Sekunden ab Betriebstag), mit Verspätungen aus GTFS-RT.
Eine Verspätung gilt für alle folgenden Halte, bis eine neue gemeldet wird.
Dazu die Nummern (0 …) der ausfallenden Halte."""
o0, o1 = self.off[ti], self.off[ti + 1]
arr, dep = list(self.st_arr[o0:o1]), list(self.st_dep[o0:o1])
if not upd:
return arr, dep, set()
seqs, stops = self.st_seq[o0:o1], self.st_stop[o0:o1]
by_i, j = {}, 0
for u in upd:
if u[0] is not None: # Schweiz, Deutschland: über die Halt-Nummer
i = bisect_right(seqs, u[0]) - 1
if i >= 0 and seqs[i] == u[0]:
by_i[i] = u
continue
# Frankreich, Niederlande: nur Haltestellen-ID; über den Bahnhof abgleichen, damit es
# auch bei Gleiswechsel passt, und der Reihe nach (eine Fahrt kann einen Bahnhof zweimal haben)
k = self.stop_index.get(u[1])
if k is None: # Belgien meldet das Gleis ("8863008_10"), der Fahrplan nur den Bahnhof ("8863008")
k = self.stop_index.get(u[1].rsplit("_", 1)[0])
if k is None:
continue
for i in range(j, len(stops)):
if self.stop_group[stops[i]] == self.stop_group[k]:
by_i[i], j = u, i + 1
break
cur, prev = None, -10 ** 9
for i in range(len(arr)):
u = by_i.get(i)
if u and u[2] is not None:
a = u[2] - ds if u[3] else arr[i] + u[2]
cur = a - arr[i]
else:
a = arr[i] + (cur or 0)
if u and u[4] is not None:
d = u[4] - ds if u[5] else dep[i] + u[4]
cur = d - dep[i]
else:
d = dep[i] + (cur or 0)
a = max(a, prev)
d = max(d, a)
arr[i], dep[i], prev = a, d, d
return arr, dep, {i for i, u in by_i.items() if u[6]}
# ---------- Positionen ----------
def vehicles(self, now: float, rt: dict):
feats = []
today = dt.datetime.fromtimestamp(now, TZ).date()
for d in (today - dt.timedelta(days=1), today):
ds, dstr = day_start(d), d.strftime("%Y%m%d")
t = now - ds
for start, end, ti in self.trips_on(d):
if start > t + 600:
break # Liste ist nach Abfahrt sortiert
if end + 7200 < t: # grosszügig, damit verspätete Züge nicht verschwinden
continue
r = rt.get((self.trip_ids[ti], dstr))
if r and r[0]:
continue # ausgefallen
arr, dep, skip = self.times(ti, ds, r[1] if r else None)
# Teilausfall: fährt nur zwischen erstem und letztem bedienten Halt; Deutschland meldet
# auch ganze Ausfälle so (alle Halte fallen aus)
live = [k for k in range(len(arr)) if k not in skip]
if not live or not (dep[live[0]] <= t <= arr[live[-1]]):
continue
pa, pd = with_min_dwell(arr, dep) # nur für die Position; Zeiten/Verspätung bleiben echt
i = min(bisect_right(pd, t), len(pd) - 1) # Zug ist zwischen Halt i-1 und i (oder steht an Halt i)
g = self.geometry(ti)
sd = g[2]
o = self.off[ti]
if t >= pa[i]: # steht am Halt i
dist = sd[i]
delay = dep[i] - self.st_dep[o + i] if t >= arr[i] else arr[i] - self.st_arr[o + i]
else:
dist = sd[i - 1] + travelled(sd[i] - sd[i - 1], pa[i] - pd[i - 1], t - pd[i - 1])
delay = arr[i] - self.st_arr[o + i]
lon, lat = self.point_at(g, dist)
# Richtung aus dem Gleisverlauf (±25 m), stimmt auch im Stillstand
(x1, y1), (x2, y2) = self.point_at(g, dist - 25), self.point_at(g, dist + 25)
dx, dy = (x2 - x1) * math.cos(math.radians(lat)), y2 - y1
heading = round(math.degrees(math.atan2(dx, dy)), 1) if dx or dy else None
route = self.routes[self.trip_route[ti]]
feats.append({
"type": "Feature",
"geometry": {"type": "Point", "coordinates": [lon, lat]},
"properties": {
"id": f"{dstr}/{ti}", "line": route["line"], "cat": route["cat"], "fv": route["fv"],
"nr": self.trip_number[ti], "from": self.stop_name[self.st_stop[o]], "to": self.trip_headsign[ti],
"delay": delay if r else None, "next": self.stop_name[self.st_stop[o + i]],
"heading": heading,
},
"_ti": ti, "_dist": dist, # intern, für track(); werden vor dem Versand entfernt
})
return feats
def trip_detail(self, date: str, ti: int, now: float, rt: dict):
d = dt.datetime.strptime(date, "%Y%m%d").date()
ds = day_start(d)
r = rt.get((self.trip_ids[ti], date))
arr, dep, skip = self.times(ti, ds, r[1] if r else None)
o0, o1 = self.off[ti], self.off[ti + 1]
def hhmm(sec):
return dt.datetime.fromtimestamp(ds + sec, TZ).strftime("%H:%M")
stops, g = [], self.geometry(ti)
for k, i in enumerate(range(o0, o1)):
s = self.st_stop[i]
stops.append({
"name": self.stop_name[s], "platform": self.stop_platform[s],
**dict(zip(("lon", "lat"), self.point_at(g, g[2][k]))), # Punkt auf der Strecke
"arr": hhmm(self.st_arr[i]) if k > 0 else None,
"dep": hhmm(self.st_dep[i]) if i < o1 - 1 else None,
"arr_delay": arr[k] - self.st_arr[i] if r else None,
"dep_delay": dep[k] - self.st_dep[i] if r else None,
"passed": now - ds > dep[k], "skipped": k in skip,
})
route = self.routes[self.trip_route[ti]]
return {"line": route["line"], "cat": route["cat"], "agency": route["agency"],
"nr": self.trip_number[ti], "to": self.trip_headsign[ti],
"realtime": bool(r), "canceled": bool(r and r[0]) or len(skip) == len(stops),
"stops": stops, "shape": self.geometry(ti)[0]}