Schweiz: alten Fahrplan ohne Trams nicht mehr laden, nur weil die Echtzeit noch auf ihn verweist

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01X48pE4bzjHej8w9HyZXCjb

tests/test_core.py

@@ -222,8 +222,38 @@ def test_all():
     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")
+
+
+def test_feed_choice():
+    """Echtzeit verweist auf alten Fahrplan ohne Trams -> trotzdem den neuen (mit Trams) laden."""
+    from trainpulse import server
+    with tempfile.TemporaryDirectory() as d:
+        for v, rtype in (("20261017", "102"), ("20261021", "900")):
+            p = Path(d) / "ch" / v
+            p.mkdir(parents=True)
+            (p / "meta.json").write_text("{}")
+            (p / "routes.txt").write_text(f"route_id,route_type\nR,{rtype}\n")
+        loaded = []
+
+        class Stub:
+            def __init__(self, path):
+                loaded.append(path.name)
+                self.version, self.meta, self.trip_ids = path.name, {}, []
+
+            def covers(self, day):
+                return True
+        old = server.C.FEEDS_DIR, server.Feed, server.compute_positions, dict(server.S.sources)
+        try:
+            server.C.FEEDS_DIR, server.Feed, server.compute_positions = Path(d), Stub, lambda: None
+            server.S.sources = {"ch": server.Source("ch")}
+            server.S.sources["ch"].rt_version = "20261017"
+            server.watch_feeds()
+        finally:
+            server.C.FEEDS_DIR, server.Feed, server.compute_positions, server.S.sources = old
+        assert loaded == ["20261021"], loaded
 
 
 if __name__ == "__main__":
     test_all()
+    test_feed_choice()
+    print("alle Tests ok")

trainpulse/prepare.py

@@ -332,6 +332,7 @@ def prepare(source: str, state: dict, zip_path: Path | None = None, force: bool
     if shaped:
         pack_shapes(shaped)
     meta["source_file"] = name
+    meta["format"] = FORMAT
     meta["prepared_at"] = int(time.time())
     publish(source, shaped or work / "rail", meta)
     shutil.rmtree(work, ignore_errors=True)

trainpulse/server.py

@@ -16,7 +16,7 @@ from fastapi.staticfiles import StaticFiles
 
 from . import config as C
 from . import border, realtime
-from .feed import TZ, Feed
+from .feed import TZ, Feed, read, route_mode
 
 log = logging.getLogger("trainpulse")
 STEP, STEPS = 2, 6  # Gleisverlauf für die nächsten 12 s (Abruf alle 5 s, Berechnung alle 3 s -> Reserve)
@@ -68,6 +68,13 @@ def loop(every: float, fn, *args):
     threading.Thread(target=run, daemon=True, name=fn.__name__).start()
 
 
+def feed_format(path: Path, meta: dict) -> int:
+    """Aufbereitungs-Format; ältere Ordner ohne Angabe: 2, wenn Trams/U-Bahnen drin sind."""
+    if "format" in meta:
+        return meta["format"]
+    return 2 if any(route_mode(r[h["route_type"]]) != "rail" for h, r in read(path / "routes.txt")) else 1
+
+
 def watch_feeds():
     """Lädt je Quelle den passenden Fahrplan: den, auf den sich die Echtzeitdaten beziehen (Schweiz:
     die schalten erst um 15 Uhr auf einen neuen Fahrplan um), sonst den neusten."""
@@ -77,11 +84,15 @@ def watch_feeds():
         found[src.name] = sorted(m.parent.name for m in (C.FEEDS_DIR / src.name).glob("*/meta.json"))
         src.on_disk = bool(found[src.name])
     for src in S.sources.values():
-        ready = found[src.name]
+        if not found[src.name]:
-        if not ready:
             continue
+        metas = {v: json.loads((C.FEEDS_DIR / src.name / v / "meta.json").read_text()) for v in found[src.name]}
+        # nur Versionen im neusten Aufbereitungs-Format (sonst z. B. alter Fahrplan ohne Trams, nur weil
+        # die Echtzeit sich noch auf ihn bezieht)
+        fmts = {v: feed_format(C.FEEDS_DIR / src.name / v, m) for v, m in metas.items()}
+        ready = [v for v in found[src.name] if fmts[v] == max(fmts.values())]
         want = src.rt_version if src.rt_version in ready else ready[-1]
-        meta = json.loads((C.FEEDS_DIR / src.name / want / "meta.json").read_text())
+        meta = metas[want]
         if (src.feed and src.feed.version == want and src.feed.covers(today + dt.timedelta(days=1))
                 and src.feed.meta.get("prepared_at") == meta.get("prepared_at")):  # gleiche Version neu aufbereitet?
             continue