Spaces:
Sleeping
Sleeping
File size: 1,772 Bytes
971a3e7 2c7f73f 9d72a91 a78f512 971a3e7 ee3c704 854e34f 3eeeacd c608381 9d72a91 2c7f73f 3eeeacd 722435e 3eeeacd 2c7f73f 971a3e7 ee3c704 9d72a91 3eeeacd 9d72a91 ac62257 30229c8 1841413 2c7f73f 4a84fec 1841413 3eeeacd 95608ed 1841413 9d72a91 2c7f73f 9d72a91 4a84fec 30229c8 4a84fec 9986845 4a84fec 9d72a91 6069409 9986845 2c7f73f |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 |
from fastapi import FastAPI, HTTPException,BackgroundTasks
import requests
import ast
import pymysql
import time
from database_manager import *
from mqtt_manager import *
from traffic_utils import *
from Utils import *
from threading import Thread
import state
from contextlib import asynccontextmanager
# -------------------------
# App Initialization
# -------------------------
@asynccontextmanager
async def lifespan(app: FastAPI):
# Starting code
start_mqtt()
yield
# At End
print("🔻 Shutting down...")
app = FastAPI(lifespan=lifespan)
#--------------------------
# Routes
# -------------------------
@app.get("/")
def main():
return {"message": "Hello from FastAPI on Hugging Face Spaces!"}
@app.post("/send-coordinates")
async def receive_coordinates(payload: CoordinatesPayload,background_tasks: BackgroundTasks):
coords= payload.coords
state_= payload.state
duration= payload.duration
delay= payload.delay
if isinstance(coords, str):
coords = ast.literal_eval(coords)
traffic_lights = get_traffic_in_container(coords)
tl_ids = check_signals(traffic_lights)
if tl_ids:
for tl_id in tl_ids:
#background_tasks.add_task(open_signal, tl_id, state_, duration,delay) #"EMR" # "ACC"
answer = await open_signal(tl_id, state_, duration,delay)
print(f"Signal {tl_id} opened with state {state_}, duration {duration}, delay {delay} ,answer: {answer}")
return {"answer": answer} # temp here
else:
print(f"Request failed with coords{coords} state {state_}, duration {duration}, delay {delay} ,answer: No traffic found found")
return {"answer": "No traffic signals found in database"+str(traffic_lights)}
|