tcl-filtrage/main.py

import collections
import re
from datetime import datetime
from typing import Dict, List, Optional, Tuple

import httpx
from fastapi import FastAPI, HTTPException
from fastapi.middleware.cors import CORSMiddleware
from fastapi.params import Header, Path
from pydantic import BaseModel

app = FastAPI()

origins = ["http://localhost:3000", "https://display.augendre.info"]

app.add_middleware(
    CORSMiddleware,
    allow_origins=origins,
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)


Delai = tuple[str, int]


class Stop(BaseModel):
    id: int
    name: str


class Passage(BaseModel):
    ligne: str
    delais: List[str]
    destination: Stop


class Passages(BaseModel):
    passages: List[Passage]
    stop: Stop


@app.get("/")
async def root():
    return {"status": "ok"}


@app.get("/stop/{stop_id}", response_model=Passages)
async def stop(
    stop_id: int = Path(
        None,
        description="Stop id to monitor. Can be obtained using https://data.grandlyon.com/jeux-de-donnees/points-arret-reseau-transports-commun-lyonnais/donnees",
    ),
    authorization: Optional[str] = Header(
        None,
        alias="Authorization",
        description="Basic auth for remote API (data grand lyon)",
    ),
):
    monitored_stop_id = stop_id
    if authorization is None:
        raise HTTPException(status_code=401, detail="Not authenticated")

    headers = {"Authorization": authorization}
    async with httpx.AsyncClient(headers=headers) as client:
        passages_res = client.get(
            "https://download.data.grandlyon.com/ws/rdata/tcl_sytral.tclpassagearret/all.json?maxfeatures=-1"
        )
        infos_res = client.get(
            "https://download.data.grandlyon.com/ws/rdata/tcl_sytral.tclarret/all.json?maxfeatures=-1"
        )
        passages_res = await passages_res
        infos_res = await infos_res
        if passages_res.status_code != 200:
            raise HTTPException(
                status_code=passages_res.status_code,
                detail="HTTP error during call to remote passages API",
            )
        if infos_res.status_code != 200:
            raise HTTPException(
                status_code=infos_res.status_code,
                detail="HTTP error during call to remote info API",
            )

    stop_ids = {monitored_stop_id}
    passages: Dict[Tuple[str, int], list[Delai]] = collections.defaultdict(list)
    for passage in passages_res.json().get("values"):
        if passage.get("id") == monitored_stop_id and passage.get("type") == "E":
            ligne = passage.get("ligne")
            ligne = re.sub(
                "[A-Z]$", "", ligne
            )  # Remove letter suffix to group by commercial line name
            destination = passage.get("idtarretdestination")
            heure_passage = passage.get("heurepassage")
            delai = get_delai(heure_passage)
            passages[(ligne, destination)].append(delai)
            stop_ids.add(destination)

    if not passages:
        raise HTTPException(status_code=404, detail="Stop not found")

    stop_infos: Dict[int, Stop] = {}
    for info in infos_res.json().get("values"):
        stop_id = info.get("id")
        if stop_id in stop_ids:
            stop_infos[stop_id] = Stop(id=stop_id, name=info.get("nom"))
        if len(stop_infos) == len(stop_ids):
            break

    passages_list = []
    for key, delais in passages.items():
        delais = list(map(lambda x: x[0], sorted(delais, key=lambda x: x[1])))
        passages_list.append(
            Passage(ligne=key[0], delais=delais, destination=stop_infos.get(key[1]))
        )
    passages_list.sort(key=lambda x: x.ligne)

    return Passages(passages=passages_list, stop=stop_infos.get(monitored_stop_id))


def get_delai(heure_passage: str) -> Delai:
    dt = datetime.strptime(heure_passage, "%Y-%m-%d %H:%M:%S")
    now = datetime.now()
    if now > dt:
        return ("Passé", -2)
    delai = dt - now
    minutes = delai.seconds // 60
    if minutes <= 0:
        return ("Proche", -1)
    return (f"{minutes} min", minutes)