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 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 State:
    feed: Feed | None = None
    rt: dict = {}
    rt_version: str | None = None
    rt_at: float | None = None
    rt_error: str | None = None
    rt_tried_at: float | None = None
    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):
    def run():
        while True:
            t = time.time()
            try:
                fn()
            except Exception:
                log.exception("Fehler in %s", fn.__name__)
            time.sleep(max(1.0, every - (time.time() - t)))
    threading.Thread(target=run, daemon=True, name=fn.__name__).start()


def watch_feeds():
    """Lädt den passenden Fahrplan: den, auf den sich die Echtzeitdaten beziehen (die schalten
    erst um 15 Uhr auf einen neuen Fahrplan um), sonst den neusten."""
    ready = sorted(m.parent.name for m in C.FEEDS_DIR.glob("*/meta.json"))
    if not ready:
        return
    want = S.rt_version if S.rt_version in ready else ready[-1]
    today = dt.datetime.now(TZ).date()
    if S.feed and S.feed.version == want and S.feed.covers(today + dt.timedelta(days=1)):
        return
    log.info("Lade Fahrplan %s …", want)
    t = time.time()
    S.feed = Feed(C.FEEDS_DIR / want)
    log.info("Fahrplan %s geladen in %.0f s (%d Fahrten)", want, time.time() - t, len(S.feed.trip_ids))
    try:  # Echtzeit sofort holen statt auf den nächsten Abruf zu warten
        poll_realtime()
    except Exception:
        log.exception("Erster Echtzeit-Abruf fehlgeschlagen")
    compute_positions()


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


def compute_positions():
    if not S.feed:
        return
    now = time.time()
    feats = S.feed.vehicles(now, S.rt)
    future = [{f["properties"]["id"]: f["geometry"]["coordinates"] + [f["properties"]["heading"]]
               for f in S.feed.vehicles(now + STEP * k, S.rt)}
              for k in range(1, STEPS + 1)]
    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)
    loop(C.RT_INTERVAL, poll_realtime)
    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/{date}/{ti}")
def trip(date: str, ti: int):
    f = S.feed
    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(), S.rt)


@app.get("/api/status")
def status():
    f = S.feed
    return {
        "feed_version": f.version if f else None,
        "shapes": f.meta.get("shapes") if f else None,
        "realtime": bool(C.RT_KEY),
        "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),
        "vehicles": S.count,
        # erst "bereit", wenn Positionen mit (dem ersten Versuch) Echtzeit berechnet sind
        "ready": bool(f and S.computed_at and (not C.RT_KEY or (S.rt_tried_at and S.computed_at >= S.rt_tried_at))),
    }


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