"""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)] for d, t in out] 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.""" 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 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 # ---------- 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 = self.times(ti, ds, r[1] if r else None) if not (dep[0] <= t <= arr[-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], "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 = 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], }) 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]), "stops": stops, "shape": self.geometry(ti)[0]}