70 lines
2.1 KiB
Python
70 lines
2.1 KiB
Python
import re
|
|
from typing import DefaultDict, List, Optional
|
|
|
|
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 Passage(BaseModel):
|
|
ligne: str
|
|
delais: List[str]
|
|
|
|
|
|
class Passages(BaseModel):
|
|
passages: List[Passage]
|
|
stop_id: int
|
|
|
|
|
|
@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)",
|
|
),
|
|
):
|
|
if authorization is None:
|
|
raise HTTPException(status_code=401, detail="Not authenticated")
|
|
|
|
headers = {"Authorization": authorization}
|
|
async with httpx.AsyncClient(headers=headers) as client:
|
|
res = await client.get(
|
|
"https://download.data.grandlyon.com/ws/rdata/tcl_sytral.tclpassagearret/all.json?maxfeatures=-1"
|
|
)
|
|
if res.status_code != 200:
|
|
raise HTTPException(
|
|
status_code=res.status_code,
|
|
detail="HTTP error during call to remote API",
|
|
)
|
|
|
|
passages = DefaultDict(list)
|
|
for passage in res.json().get("values"):
|
|
if passage.get("id") == stop_id:
|
|
ligne = passage.get("ligne")
|
|
ligne = re.sub(
|
|
"[A-Z]$", "", ligne
|
|
) # Remove letter suffix to group by commercial line name
|
|
passages[ligne].append(passage.get("delaipassage"))
|
|
passages_list = []
|
|
for ligne, delais in passages.items():
|
|
passages_list.append(Passage(ligne=ligne, delais=delais))
|
|
return Passages(passages=passages_list, stop_id=stop_id)
|