Trainpulse/trainpulse/server.py

"""Webserver: liefert die Karte aus und berechnet laufend die Zugpositionen.

    uvicorn trainpulse.server:app --host 0.0.0.0 --port 8080
"""
import datetime as dt
import gzip
import json
import logging
import threading
import time
from contextlib import asynccontextmanager
from pathlib import Path

from fastapi import FastAPI, HTTPException, Request, Response
from fastapi.staticfiles import StaticFiles

from . import config as C
from . import border, realtime
from .feed import TZ, Feed

log = logging.getLogger("trainpulse")
STEP, STEPS = 2, 5  # Positionen alle 2 s für die nächsten 10 s, damit der Browser Kurven nachfährt
WEB = Path(__file__).resolve().parent.parent / "web"


class Source:
    """Ein Fahrplan (Schweiz, Deutschland, …) mit seinem Echtzeit-Feed."""

    def __init__(self, name: str):
        self.name, self.cfg = name, C.SOURCES[name]
        self.feed: Feed | None = None
        self.rt: dict = {}
        self.rt_version: str | None = None
        self.rt_at: float | None = None
        self.rt_error: str | None = None
        self.rt_tried_at: float | None = None

    @property
    def has_rt(self) -> bool:
        return bool(self.cfg["rt_url"]) and (bool(self.cfg["rt_key"]) or not self.cfg["needs_key"])


class State:
    sources = {name: Source(name) for name in C.ENABLED}
    vehicles: bytes = b'{"type":"FeatureCollection","features":[]}'
    vehicles_gz: bytes = gzip.compress(vehicles)
    count = 0
    computed_at: float | None = None


S = State()


def loop(every: float, fn, *args):
    def run():
        while True:
            t = time.time()
            try:
                fn(*args)
            except Exception:
                log.exception("Fehler in %s %s", fn.__name__, args)
            time.sleep(max(1.0, every - (time.time() - t)))
    threading.Thread(target=run, daemon=True, name=fn.__name__).start()


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."""
    today = dt.datetime.now(TZ).date()
    for src in S.sources.values():
        ready = sorted(m.parent.name for m in (C.FEEDS_DIR / src.name).glob("*/meta.json"))
        if not ready:
            continue
        want = src.rt_version if src.rt_version in ready else ready[-1]
        if src.feed and src.feed.version == want and src.feed.covers(today + dt.timedelta(days=1)):
            continue
        log.info("[%s] Lade Fahrplan %s …", src.name, want)
        t = time.time()
        src.feed = Feed(C.FEEDS_DIR / src.name / want)
        log.info("[%s] Fahrplan %s geladen in %.0f s (%d Fahrten)", src.name, want, time.time() - t,
                 len(src.feed.trip_ids))
        try:  # Echtzeit sofort holen statt auf den nächsten Abruf zu warten
            poll_realtime(src)
        except Exception:
            log.exception("[%s] Erster Echtzeit-Abruf fehlgeschlagen", src.name)
        compute_positions()


def poll_realtime(src: Source):
    if not src.has_rt or not src.feed:
        return
    try:
        src.rt_version, src.rt = realtime.parse(realtime.fetch(src.cfg["rt_url"], src.cfg["rt_key"]),
                                                src.feed.trip_index)
        src.rt_at, src.rt_error = time.time(), None
    except Exception as e:  # Daten älter als 10 min sind wertlos -> verwerfen statt falsch anzeigen
        src.rt_error = str(e)
        if src.rt_at and time.time() - src.rt_at > 600:
            src.rt = {}
        raise
    finally:
        src.rt_tried_at = time.time()


def positions(src: Source, t: float) -> list:
    show = border.SHOW.get(src.name) if len(S.sources) > 1 else None
    feats = []
    for f in src.feed.vehicles(t, src.rt):
        if show and not show(*f["geometry"]["coordinates"]):
            continue  # gehört zum Fahrplan des Nachbarlandes
        f["properties"]["id"] = f"{src.name}/{f['properties']['id']}"
        feats.append(f)
    return feats


def compute_positions():
    now = time.time()
    feats, future = [], [{} for _ in range(STEPS)]
    for src in list(S.sources.values()):
        if not src.feed:
            continue
        feats += positions(src, now)
        for k in range(STEPS):
            future[k].update({f"{src.name}/{f['properties']['id']}": f["geometry"]["coordinates"] + [f["properties"]["heading"]]
                              for f in src.feed.vehicles(now + STEP * (k + 1), src.rt)})
    for f in feats:
        path = []
        for pos in future:
            p = pos.get(f["properties"]["id"])
            if p is None:
                break  # Fahrt endet
            path.append(p)
        f["path"] = path
    S.vehicles = json.dumps({"type": "FeatureCollection", "ts": now, "step": STEP, "features": feats},
                            separators=(",", ":")).encode()
    S.vehicles_gz = gzip.compress(S.vehicles, 5)
    S.count, S.computed_at = len(feats), time.time()


@asynccontextmanager
async def lifespan(app):
    logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
    C.FEEDS_DIR.mkdir(parents=True, exist_ok=True)
    loop(60, watch_feeds)
    for src in S.sources.values():
        loop(src.cfg["rt_interval"], poll_realtime, src)
    loop(3, compute_positions)
    yield


app = FastAPI(lifespan=lifespan, docs_url=None, redoc_url=None)


@app.get("/api/vehicles")
def vehicles(request: Request):
    headers = {"Cache-Control": "no-store", "X-Now": str(time.time())}  # für den Uhrenabgleich im Browser
    if "gzip" in request.headers.get("accept-encoding", ""):
        return Response(S.vehicles_gz, media_type="application/json", headers={**headers, "Content-Encoding": "gzip"})
    return Response(S.vehicles, media_type="application/json", headers=headers)


@app.get("/api/trip/{source}/{date}/{ti}")
def trip(source: str, date: str, ti: int):
    src = S.sources.get(source)
    f = src.feed if src else None
    if not f or not (0 <= ti < len(f.trip_ids)) or not (len(date) == 8 and date.isdigit()):
        raise HTTPException(404)
    return f.trip_detail(date, ti, time.time(), src.rt)


@app.get("/api/status")
def status():
    loaded = [s for s in S.sources.values() if s.feed]
    rt = [s for s in loaded if s.has_rt]
    ok = [s for s in rt if not s.rt_error and s.rt_at and time.time() - s.rt_at < 180]
    return {
        "loaded": bool(loaded),
        # erst "bereit", wenn Positionen mit (dem ersten Versuch) Echtzeit berechnet sind
        "ready": bool(loaded and S.computed_at and all(s.rt_tried_at and S.computed_at >= s.rt_tried_at for s in rt)),
        "vehicles": S.count,
        "rt_ok": bool(ok) and len(ok) == len(loaded),
        "rt_some": bool(ok),
        "sources": {s.name: {
            "feed_version": s.feed.version if s.feed else None,
            "shapes": s.feed.meta.get("shapes") if s.feed else None,
            "realtime": s.has_rt,
            "rt_version": s.rt_version,
            "rt_age": round(time.time() - s.rt_at) if s.rt_at else None,
            "rt_error": s.rt_error,
            "rt_trips": len(s.rt),
        } for s in S.sources.values()},
    }


app.mount("/tiles", StaticFiles(directory=C.DATA_DIR / "tiles", check_dir=False), name="tiles")  # PMTiles, Range-Requests
app.mount("/", StaticFiles(directory=WEB, html=True), name="web")