"""Selbsttest mit einem Mini-Fahrplan: python -m pytest tests (oder: python tests/test_core.py)""" import csv import datetime as dt import io import json import sys import tempfile import zipfile from pathlib import Path sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) from google.transit import gtfs_realtime_pb2 as pb # noqa: E402 from trainpulse import prepare, realtime # noqa: E402 from trainpulse.feed import TZ, Feed, day_start # noqa: E402 FILES = { "agency.txt": "agency_id,agency_name,agency_url,agency_timezone\n11,SBB,https://sbb.ch,Europe/Zurich\n", "routes.txt": "route_id,agency_id,route_short_name,route_long_name,route_desc,route_type\n" "R1,11,IC5,,IC,102\nB1,11,31,,B,700\n", "trips.txt": "route_id,service_id,trip_id,trip_headsign,trip_short_name,shape_id\n" "R1,S,T1,Ziel,812,SH1\nB1,S,BUS,Irgendwo,1,\n", "stop_times.txt": "trip_id,arrival_time,departure_time,stop_id,stop_sequence,shape_dist_traveled\n" "T1,08:00:00,08:00:00,A,1,0\nT1,08:10:00,08:12:00,B,2,8000\nT1,08:20:00,08:20:00,C,3,19000\n" "BUS,08:00:00,08:00:00,A,1,\nBUS,08:05:00,08:05:00,X,2,\n", "stops.txt": "stop_id,stop_name,stop_lat,stop_lon,parent_station,platform_code\n" "A,Alpha,47.0,8.0,,1\nB,Beta,47.0,8.1,,2\nC,Gamma,47.1,8.1,,3\nX,Bushalt,47.5,8.5,,\n", "calendar.txt": "service_id,monday,tuesday,wednesday,thursday,friday,saturday,sunday,start_date,end_date\n" "S,1,1,1,1,1,1,1,20260101,20261231\n", "calendar_dates.txt": "service_id,date,exception_type\nS,20261025,2\n", "feed_info.txt": "feed_publisher_name,feed_version\nX,20261017\n", } # Streckenverlauf mit Umweg nach Süden zwischen A und B # so wie pfaedle es schreibt: mit shape_dist_traveled in Metern (absichtlich nicht sortiert) SHAPE = "shape_id,shape_pt_lat,shape_pt_lon,shape_pt_sequence,shape_dist_traveled\n" \ "SH1,46.98,8.05,2,4000\nSH1,47.0,8.0,1,0\nSH1,47.0,8.1,3,8000\nSH1,47.1,8.1,4,19000\n" DAY = dt.date(2026, 10, 21) def german_style(files: dict) -> dict: """Wie gtfs.de: Spalten in anderer Reihenfolge, kein Fahrtziel/keine Zugnummer, Version 'latest-free'.""" out = {} for n, c in files.items(): rows = list(csv.reader(io.StringIO(c))) if n == "trips.txt": rows = [r[:3] for r in rows] # route_id, service_id, trip_id if n == "feed_info.txt": rows[1][1] = "latest-free" rows = [r[::-1] for r in rows] # Spalten rückwärts out[n] = "".join(",".join(r) + "\n" for r in rows) return out def make_feed(tmp: Path, shapes=True, files=FILES) -> Feed: tmp.mkdir(parents=True, exist_ok=True) zp = tmp / "in.zip" with zipfile.ZipFile(zp, "w") as z: for n, c in files.items(): z.writestr(n, "\ufeff" + c) # wie die echten Schweizer Dateien: mit BOM meta = prepare.filter_rail(zp, tmp / "rail", "20261020") assert meta["feed_version"] == ("20261017" if files is FILES else "20261020"), meta assert meta["trips"] == 1 and meta["stop_times"] == 3, meta if shapes: trips = (tmp / "rail" / "trips.txt").read_text() assert "SH1" not in trips # Verweise auf nicht kopierte shapes werden geleert (tmp / "rail" / "trips.txt").write_text(trips.replace("812,", "812,SH1")) # wie pfaedle (tmp / "rail" / "shapes.txt").write_text(SHAPE) prepare.pack_shapes(tmp / "rail") assert not (tmp / "rail" / "shapes.txt").exists() with open(tmp / "rail" / "trips.txt") as f: assert "SH1" in f.read() (tmp / "rail" / "meta.json").write_text(json.dumps(meta)) return Feed(tmp / "rail", today=DAY) def at(h, m): return dt.datetime(DAY.year, DAY.month, DAY.day, h, m, tzinfo=TZ).timestamp() def rt_feed(delay_b=None, cancel=False, version="20261017"): fm = pb.FeedMessage() fm.header.gtfs_realtime_version = "2.0" fm.header.feed_version = version e = fm.entity.add(id="1") e.trip_update.trip.trip_id = "T1" e.trip_update.trip.start_date = "20261021" if cancel: e.trip_update.trip.schedule_relationship = pb.TripDescriptor.CANCELED s = e.trip_update.stop_time_update.add(stop_sequence=1) s.departure.delay = 0 if delay_b is not None: s = e.trip_update.stop_time_update.add(stop_sequence=2) s.arrival.delay = delay_b s.departure.delay = delay_b e2 = fm.entity.add(id="2") e2.trip_update.trip.trip_id = "BUS" # nicht im Bahn-Fahrplan -> muss ignoriert werden return fm.SerializeToString() def test_all(): assert prepare.timetable_year(dt.date(2026, 12, 12)) == 2026 assert prepare.timetable_year(dt.date(2026, 12, 13)) == 2027 # 2. Sonntag im Dezember assert prepare.timetable_year(dt.date(2025, 12, 14)) == 2026 with tempfile.TemporaryDirectory() as d: f = make_feed(Path(d)) pts, cum, sd = f.geometry(0) assert len(pts) == 4 and abs(pts[1][1] - 46.98) < 1e-4 and sd == [0, 8000, 19000] and cum[1] == 4000 assert f.vehicles(at(7, 59), {}) == [] v = f.vehicles(at(8, 5), {}) assert len(v) == 1 and v[0]["properties"]["delay"] is None lon, lat = v[0]["geometry"]["coordinates"] assert lat < 47.0 and abs(lon - 8.05) < 0.01, (lon, lat) # halbe Fahrzeit = Umweg-Scheitelpunkt assert v[0]["properties"]["next"] == "Beta" and v[0]["properties"]["fv"] is True # IC = Fernverkehr assert abs(v[0]["properties"]["heading"] - 90) < 15, v[0]["properties"]["heading"] # Scheitel: Richtung Osten assert f.vehicles(at(8, 11), {})[0]["properties"]["heading"] is not None # auch im Stillstand # Gleisverlauf für den Browser: folgt der Kurve über den Umweg-Scheitel, Zeiten steigen an tr = f.track(0, [3000, 4500, 6000], 2) assert tr[0][2] == 0 and tr[-1][2] == 4 and [p[2] for p in tr] == sorted(p[2] for p in tr) assert any(abs(p[0] - 8.05) < 1e-4 and abs(p[1] - 46.98) < 1e-4 for p in tr), tr # Scheitelpunkt drin assert len(f.track(0, [5000, 5000, 5000], 2)) == 3 # steht: Punkte bleiben (Zeit läuft) assert tr[0][3] == 3000 and tr[-1][3] == 6000 # Strecken-km für "nie rückwärts" im Browser v = f.vehicles(at(8, 18), {}, (60, 120, 180)) # Vorausschau = einzelne Abfragen, endet mit der Fahrt assert v[0]["_ahead"] == [f.vehicles(at(8, m), {})[0]["_dist"] for m in (19, 20)] and not f.vehicles(at(8, 21), {}) v = f.vehicles(at(8, 2), {}, (60, 120)) assert v[0]["_ahead"] == [f.vehicles(at(8, m), {})[0]["_dist"] for m in (3, 4)] # Mindesthaltezeit: 0 s Halt wird für die Animation verlängert, Endhalte bleiben from trainpulse.feed import with_min_dwell pa, pd = with_min_dwell([0, 600, 1200], [0, 600, 1200]) assert pa[1] == 580 and pd[1] == 620 and pa[0] == 0 and pd[2] == 1200 # Anfahren/Bremsen: symmetrisch, am Anfang/Ende langsamer als linear, Gesamtstrecke stimmt from trainpulse.feed import travelled assert travelled(8000, 600, 0) == 0 and abs(travelled(8000, 600, 600) - 8000) < 1e-6 assert abs(travelled(8000, 600, 300) - 4000) < 1e-6 assert travelled(8000, 600, 10) < 8000 * 10 / 600 / 2 assert abs(travelled(8000, 600, 590) - (8000 - travelled(8000, 600, 10))) < 1e-6 assert all(travelled(200, 60, x) <= travelled(200, 60, x + 1) for x in range(60)) # kurz: nie rückwärts assert f.vehicles(at(8, 11), {})[0]["geometry"]["coordinates"] == [8.1, 47.0] # steht in Beta ver, rt = realtime.parse(rt_feed(delay_b=300), f.trip_index) assert ver == "20261017" and list(rt) == [("T1", "20261021")] v = f.vehicles(at(8, 14), rt) # planmässig weg, mit 5 min Verspätung noch unterwegs nach Beta assert v[0]["properties"]["delay"] == 300 and v[0]["properties"]["next"] == "Beta" v = f.vehicles(at(8, 24), rt) # Verspätung pflanzt sich bis Gamma fort assert v and v[0]["properties"]["delay"] == 300 # Frankreich/Niederlande: nur stop_id, kein stop_sequence fm = pb.FeedMessage() fm.header.gtfs_realtime_version = "2.0" e = fm.entity.add(id="1") e.trip_update.trip.trip_id, e.trip_update.trip.start_date = "T1", "20261021" st = e.trip_update.stop_time_update.add(stop_id="B_7") # Belgien: Gleis als Suffix st.arrival.delay = st.departure.delay = 300 _, rt = realtime.parse(fm.SerializeToString(), f.trip_index) assert rt[("T1", "20261021")][1][0][0] is None assert f.vehicles(at(8, 14), rt)[0]["properties"]["delay"] == 300 _, rt = realtime.parse(rt_feed(cancel=True), f.trip_index) assert f.vehicles(at(8, 5), rt) == [] def skipped(*seqs): # Ausfall einzelner Halte (so meldet Deutschland auch ganze Ausfälle) fm = pb.FeedMessage() fm.header.gtfs_realtime_version = "2.0" e = fm.entity.add(id="1") e.trip_update.trip.trip_id, e.trip_update.trip.start_date = "T1", "20261021" for q in seqs: e.trip_update.stop_time_update.add(stop_sequence=q).schedule_relationship = \ pb.TripUpdate.StopTimeUpdate.SKIPPED return realtime.parse(fm.SerializeToString(), f.trip_index)[1] assert f.vehicles(at(8, 5), skipped(1, 2, 3)) == [] # alle Halte fallen aus = Zug fällt aus rt = skipped(3) # endet vorzeitig in Beta assert f.vehicles(at(8, 5), rt) and f.vehicles(at(8, 15), rt) == [] assert [x["skipped"] for x in f.trip_detail("20261021", 0, at(8, 5), rt)["stops"]] == [False, False, True] assert f.trip_detail("20261021", 0, at(8, 5), skipped(1, 2, 3))["canceled"] assert f.vehicles(dt.datetime(2026, 10, 25, 8, 5, tzinfo=TZ).timestamp(), {}) == [] # calendar_dates _, rt = realtime.parse(rt_feed(delay_b=120), f.trip_index) det = f.trip_detail("20261021", 0, at(8, 13), rt) assert [s["name"] for s in det["stops"]] == ["Alpha", "Beta", "Gamma"] assert (det["stops"][1]["lon"], det["stops"][1]["lat"]) == (8.1, 47.0) # Halt auf der Strecke assert det["stops"][1]["arr"] == "08:10" and det["stops"][1]["arr_delay"] == 120 assert det["stops"][0]["passed"] and not det["stops"][1]["passed"] assert day_start(DAY) == at(0, 0) g = make_feed(Path(d) / "noshape", shapes=False) # ohne shapes: Luftlinie assert g.vehicles(at(8, 5), {})[0]["geometry"]["coordinates"] == [8.05, 47.0] de = make_feed(Path(d) / "de", shapes=False, files=german_style(FILES)) # Format von gtfs.de v = de.vehicles(at(8, 5), {}) assert len(v) == 1 and v[0]["properties"]["to"] == "Gamma" and v[0]["properties"]["nr"] == "" assert de.vehicles(dt.datetime(2026, 10, 25, 8, 5, tzinfo=TZ).timestamp(), {}) == [] # calendar_dates from trainpulse.feed import STOP_SUFFIX, long_distance, route_mode assert [STOP_SUFFIX.sub("", n) for n in ("S+U Rathaus Spandau Bstggl.2", "Zingster Str., Ausstieg Gl.3-5", "Virchow-Klinikum, Ausstieg", "Zürich HB", "Bahnhof, Oranienburg")] == \ ["S+U Rathaus Spandau", "Zingster Str.", "Virchow-Klinikum", "Zürich HB", "Bahnhof, Oranienburg"] assert long_distance("", "ICE 75") and long_distance("IR", "IR36") and long_distance("", "ECE 20") assert not long_distance("S", "S5") and not long_distance("", "RE6") and not long_distance("", "IRE 3") assert not long_distance("TER", "K1") and not long_distance("", "") assert [route_mode(t) for t in ("2", "106", "0", "900", "1", "401", "3", "700", "1400")] == \ ["rail", "rail", "tram", "tram", "subway", "subway", None, None, None] # Bus, Standseilbahn: nein assert long_distance("", "NJ") and long_distance("TGV", "") and long_distance("OUIGO", "") from trainpulse.feed import LINE_PREFIX assert LINE_PREFIX.sub("", "Stoptrein RS4") == "RS4" and LINE_PREFIX.sub("", "Intercity direct") == "Intercity direct" from trainpulse.border import country, in_switzerland, shows assert in_switzerland(8.540, 47.378) and not in_switzerland(9.177, 47.659) # Zürich HB / Konstanz assert not in_switzerland(8.690, 47.697) # Büsingen (deutsche Exklave) places = {(2.355, 48.881): "fr", (4.357, 50.836): "be", (4.900, 52.379): "nl", (13.369, 52.525): "de", (8.540, 47.378): "ch", (6.13, 49.61): None, (11.40, 47.26): None} # Paris, Brüssel, Amsterdam, for (x, y), c in places.items(): # Berlin, Zürich, Luxemburg, Innsbruck assert country(x, y) == c, (x, y, country(x, y)) both = {"ch", "de"} assert shows("ch", both, 8.54, 47.378) and not shows("de", both, 8.54, 47.378) assert shows("de", both, 11.40, 47.26) and shows("ch", both, 11.40, 47.26) # Ausland: beide assert shows("de", both, 2.355, 48.881) # Frankreich ohne eigenen Fahrplan -> DE zeigt print("alle Tests ok") if __name__ == "__main__": test_all()