Claude
a064fccb23
- Server berechnet den Kurs aus der Strecke (±25 m) an der Zugposition; stimmt dadurch auch im Stillstand (vorher blieb der Pfeil verdreht stehen) - Trapezprofil: ~150 m vor/nach jedem Halt bremst bzw. beschleunigt der Zug, Ankunfts- und Abfahrtszeiten bleiben gleich Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01X48pE4bzjHej8w9HyZXCjb
135 lines
6.6 KiB
Python
135 lines
6.6 KiB
Python
"""Selbsttest mit einem Mini-Fahrplan: python -m pytest tests (oder: python tests/test_core.py)"""
|
||
import datetime as dt
|
||
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 make_feed(tmp: Path, shapes=True) -> 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, "" + c) # wie die echten Schweizer Dateien: mit BOM
|
||
meta = prepare.filter_rail(zp, tmp / "rail")
|
||
assert meta["feed_version"] == "20261017" and meta["trips"] == 1 and meta["stop_times"] == 3, meta
|
||
if shapes:
|
||
(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"
|
||
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
|
||
|
||
# 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
|
||
|
||
_, 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]["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]
|
||
print("alle Tests ok")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
test_all()
|