132 lines
4.0 KiB
Python
132 lines
4.0 KiB
Python
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)
|