Trainpulse/tests/test_core.py

"""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\n"
                      "T1,08:00:00,08:00:00,A,1\nT1,08:10:00,08:12:00,B,2\nT1,08:20:00,08:20:00,C,3\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
SHAPE = "shape_id,shape_pt_lat,shape_pt_lon,shape_pt_sequence\n" \
        "SH1,47.0,8.0,1\nSH1,46.98,8.05,2\nSH1,47.0,8.1,3\nSH1,47.1,8.1,4\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)
        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))
        # Halt B liegt auf dem Streckenverlauf bei Punkt 3, nicht auf der Luftlinie
        pts, cum, sd = f.geometry(0)
        assert abs(sd[1] - cum[2]) < 1 and abs(sd[2] - cum[3]) < 1

        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 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()