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
477 lines
24 KiB
Python
477 lines
24 KiB
Python
"""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",
|
||
"AVE", "AVLO", "ALVIA", "EUROMED", # Spanien
|
||
"EIC", "EIP", "TLK", # Polen
|
||
"ICL", "WB", "SJ", "SNT", "X"} # Dänemark (IC Lyn), Westbahn, Schweden (SJ, Snälltåget, X2000)
|
||
# Zugnummer mit Gattung ("RJ 658"): bei ÖBB steht der Fernverkehr nur so, die Linie heisst "A10-1"
|
||
TRIP_CAT = re.compile(r"^([A-Z]{1,5}) ?(\d+)$")
|
||
NO_FJERNTOG = re.compile(r"^F\d+$") # Norwegen: F4, F5, F6 = Fernzüge (aber dänische S-Bahn "F" nicht)
|
||
# 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"
|
||
# Steige im deutschen Fahrplan: "S+U Rathaus Spandau Bstggl.2", "Zingster Str., Ausstieg Gl.3-5"
|
||
STOP_SUFFIX = re.compile(r",? (Ausstieg\b.*|Bstg\S*|Gl\.\s?\S+)$")
|
||
|
||
|
||
def route_mode(route_type: str) -> str | None:
|
||
"""rail, tram, subway – oder None (Bus, Schiff, Seilbahn …). GTFS-Grundtypen und "extended route types"."""
|
||
t = int(route_type)
|
||
if t == 2 or 100 <= t <= 199:
|
||
return "rail"
|
||
if t == 0 or 900 <= t <= 906:
|
||
return "tram"
|
||
if t == 1 or 400 <= t <= 404:
|
||
return "subway"
|
||
return None
|
||
|
||
|
||
SBAHN = re.compile(r"^(SN? ?\d+[A-Z]?|S)$") # S1, S 5, S41, SN7 (Nacht-S-Bahn), nur "S"
|
||
|
||
|
||
def sbahn(cat: str, line: str, route_type: str = "") -> bool:
|
||
"""S-Bahn: Schweiz über die Kategorie, Deutschland/Belgien über den Liniennamen, Dänemark (S-tog)
|
||
über den Routentyp 109. Niederländische Sprinter und französische TER zählen als Regionalzug."""
|
||
return cat in ("S", "SN") or route_type == "109" or bool(SBAHN.match(line))
|
||
|
||
|
||
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 or bool(NO_FJERNTOG.match(line or ""))
|
||
|
||
|
||
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, all_sbahn: bool = False):
|
||
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(STOP_SUFFIX.sub("", 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)])
|
||
if len(line) > 12: # Entur: "Ø Göteborg-Halmstad-…" -> "Ø"; Kartenbeschriftung bleibt kurz
|
||
line = line.split()[0][:12]
|
||
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 "",
|
||
"mode": route_mode(r[h["route_type"]]) or "rail", "type": r[h["route_type"]],
|
||
}
|
||
|
||
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"]
|
||
# Routentyp 101/102/105 (Hochgeschwindigkeit, Fern-, Nachtzug): Schweden kennt sonst nur Nummern
|
||
route["fv"] = route["mode"] == "rail" and (long_distance(route["cat"], route["line"])
|
||
or route["type"] in ("101", "102", "105"))
|
||
route["sb"] = route["mode"] == "rail" and not route["fv"] and (all_sbahn or sbahn(route["cat"], route["line"], route["type"]))
|
||
self.trip_cat = {} # Fahrt -> Gattung aus der Zugnummer, wo die Linie keine Fernverkehrs-Gattung hat
|
||
for ti, num in enumerate(self.trip_number):
|
||
m = TRIP_CAT.match(num) if num and not num.isdigit() else None
|
||
if m and m.group(1) in LONG_DISTANCE and not self.routes[self.trip_route[ti]]["fv"]:
|
||
self.trip_cat[ti], self.trip_number[ti] = m.group(1), m.group(2)
|
||
|
||
# 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, ahead: tuple = ()):
|
||
"""Fahrzeuge zur Zeit now. Mit ahead (Sekunden) zusätzlich "_ahead": Strecken-Meter zu now+s,
|
||
solange der Zug fährt – ohne den Fahrplan dafür jedes Mal neu durchzurechnen."""
|
||
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 = self.rt_for(rt, 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
|
||
g = self.geometry(ti)
|
||
sd = g[2]
|
||
o = self.off[ti]
|
||
|
||
def at(tt): # -> (Halt i, Strecken-Meter); Zug zwischen Halt i-1 und i oder steht an i
|
||
i = min(bisect_right(pd, tt), len(pd) - 1)
|
||
if tt >= pa[i]:
|
||
return i, sd[i]
|
||
return i, sd[i - 1] + travelled(sd[i] - sd[i - 1], pa[i] - pd[i - 1], tt - pd[i - 1])
|
||
|
||
i, dist = at(t)
|
||
if t >= pa[i]: # steht am Halt i
|
||
delay = dep[i] - self.st_dep[o + i] if t >= arr[i] else arr[i] - self.st_arr[o + i]
|
||
else:
|
||
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, tc = self.routes[self.trip_route[ti]], self.trip_cat.get(ti)
|
||
feats.append({
|
||
"type": "Feature",
|
||
"geometry": {"type": "Point", "coordinates": [lon, lat]},
|
||
"properties": {
|
||
"id": f"{dstr}/{ti}", "line": tc or route["line"], "cat": route["cat"], "mode": route["mode"],
|
||
"fv": bool(tc) or route["fv"], "sb": not tc and route["sb"],
|
||
"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
|
||
"_ahead": [at(t + s)[1] for s in ahead if t + s <= arr[live[-1]]],
|
||
})
|
||
return feats
|
||
|
||
def rt_for(self, rt: dict, ti: int, date: str):
|
||
# Renfe liefert kein Datum; dort ist die Trip-ID je Betriebstag eindeutig
|
||
return rt.get((self.trip_ids[ti], date)) or rt.get((self.trip_ids[ti], ""))
|
||
|
||
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 = self.rt_for(rt, 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": self.trip_cat.get(ti, 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]}
|