Spanien (Renfe + Cercanías), Polen, Norwegen, Dänemark, Österreich (ÖBB + Wiener Linien), Schweden (Key)
- 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
This commit is contained in:
+44
-4
@@ -208,6 +208,8 @@ def test_all():
|
||||
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", "")
|
||||
assert long_distance("", "AVE") and long_distance("", "EIP") and long_distance("", "F5") # ES, PL, Norwegen
|
||||
assert not long_distance("", "F") and sbahn("", "F", "109") # Kopenhagen S-tog F
|
||||
from trainpulse.feed import LINE_PREFIX
|
||||
assert LINE_PREFIX.sub("", "Stoptrein RS4") == "RS4" and LINE_PREFIX.sub("", "Intercity direct") == "Intercity direct"
|
||||
|
||||
@@ -215,12 +217,14 @@ def test_all():
|
||||
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
|
||||
(8.540, 47.378): "ch", (6.13, 49.61): None, (14.51, 46.05): None, # Paris … Luxemburg, Ljubljana
|
||||
(11.40, 47.26): "at", (16.37, 48.21): "at", (-3.69, 40.41): "es", (21.01, 52.23): "pl",
|
||||
(10.75, 59.91): "no", (12.565, 55.672): "dk", (18.06, 59.33): "se"} # Innsbruck, Wien, Madrid, …
|
||||
for (x, y), c in places.items():
|
||||
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, 14.51, 46.05) and shows("ch", both, 14.51, 46.05) # Ausland: beide
|
||||
assert shows("de", both, 2.355, 48.881) # Frankreich ohne eigenen Fahrplan -> DE zeigt
|
||||
|
||||
|
||||
@@ -236,7 +240,7 @@ def test_feed_choice():
|
||||
loaded = []
|
||||
|
||||
class Stub:
|
||||
def __init__(self, path):
|
||||
def __init__(self, path, **kw):
|
||||
loaded.append(path.name)
|
||||
self.version, self.meta, self.trip_ids = path.name, {}, []
|
||||
|
||||
@@ -253,7 +257,43 @@ def test_feed_choice():
|
||||
assert loaded == ["20261021"], loaded
|
||||
|
||||
|
||||
def test_new_countries():
|
||||
"""Renfe Cercanías: aufgefüllte Felder; ÖBB: Unterordner und Gattung nur in der Zugnummer;
|
||||
Renfe-Echtzeit ohne Datum."""
|
||||
with tempfile.TemporaryDirectory() as d:
|
||||
files = dict(FILES)
|
||||
files["routes.txt"] = ("route_id ,agency_id,route_short_name,route_long_name,route_desc,route_type\n"
|
||||
"R1 ,11,A10-1,,,2 \nB1,11,31,,B,700\n")
|
||||
files["trips.txt"] = files["trips.txt"].replace("R1,S,T1,Ziel,812,", "R1 ,S,T1,Ziel,RJ 658,")
|
||||
files["calendar.txt"] = files["calendar.txt"].replace("end_date", "end_date ") # auch nur im Kopf
|
||||
files["stops.txt"] = "\n".join(l + (",level_id" if i == 0 else ",Level 0") if l else l
|
||||
for i, l in enumerate(files["stops.txt"].split("\n"))) # ÖBB
|
||||
zp = Path(d) / "in.zip"
|
||||
with zipfile.ZipFile(zp, "w") as z:
|
||||
for n, c in files.items():
|
||||
z.writestr("GTFS_Fahrplan_2026/" + n, c)
|
||||
flat = prepare.normalize_zip(zp, Path(d))
|
||||
meta = prepare.filter_rail(flat, Path(d) / "rail", "20261020")
|
||||
assert meta["trips"] == 1, meta # Route trotz "R1 " gefunden
|
||||
assert "Level 0" not in (Path(d) / "rail" / "stops.txt").read_text() # levels.txt fehlt -> Verweis leer
|
||||
(Path(d) / "rail" / "meta.json").write_text(json.dumps(meta))
|
||||
f = Feed(Path(d) / "rail")
|
||||
p = f.vehicles(at(8, 5), {})[0]["properties"]
|
||||
assert (p["line"], p["nr"], p["fv"], p["sb"]) == ("RJ", "658", True, False), p
|
||||
assert f.trip_detail("20261021", 0, at(8, 5), {})["line"] == "RJ"
|
||||
fm = pb.FeedMessage()
|
||||
fm.header.gtfs_realtime_version = "2.0"
|
||||
e = fm.entity.add(id="1")
|
||||
e.trip_update.trip.trip_id = "T1" # ohne start_date (Renfe)
|
||||
st = e.trip_update.stop_time_update.add(stop_id="B")
|
||||
st.arrival.delay = st.departure.delay = 300
|
||||
_, rt = realtime.parse(fm.SerializeToString(), f.trip_index)
|
||||
assert f.vehicles(at(8, 14), rt)[0]["properties"]["delay"] == 300
|
||||
assert prepare.normalize_zip(flat, Path(d)) == flat # schon sauber: unverändert
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
test_all()
|
||||
test_feed_choice()
|
||||
test_new_countries()
|
||||
print("alle Tests ok")
|
||||
|
||||
Reference in New Issue
Block a user