Claude
53b7c24a81
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01X48pE4bzjHej8w9HyZXCjb
203 lines
11 KiB
Python
203 lines
11 KiB
Python
"""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)
|
|
|
|
# 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) == []
|
|
|
|
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 long_distance
|
|
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 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()
|