tcl-filtrage/main.py

import collections
import re
from typing import DefaultDict, 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=["*"],
)


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("/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 API",
            )
        if infos_res.status_code != 200:
            raise HTTPException(
                status_code=infos_res.status_code,
                detail="HTTP error during call to remote API",
            )

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

    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():
        passages_list.append(
            Passage(ligne=key[0], delais=delais, destination=stop_infos.get(key[1]))
        )

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