-
Notifications
You must be signed in to change notification settings - Fork 0
/
app_async.py
75 lines (59 loc) · 2.3 KB
/
app_async.py
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
64
65
66
67
68
69
70
71
72
73
74
75
import json
from fastapi.staticfiles import StaticFiles
from fastapi import FastAPI, Request
from fastapi.templating import Jinja2Templates
from fastapi.responses import HTMLResponse, StreamingResponse, JSONResponse
from fastapi.middleware.cors import CORSMiddleware
import asyncio
templates = Jinja2Templates(directory="templates")
app = FastAPI()
app.mount("/static", StaticFiles(directory="static"), name="static")
app.mount("/data", StaticFiles(directory="data"), name="data")
def load_json(file_path):
with open(file_path, 'r') as file:
data = json.load(file)
return data
def get_cords_from_geojson(file_path:str):
geo_data = load_json(file_path)
cords = []
for inner_dicts in geo_data["features"]:
cords.append(inner_dicts['geometry']['coordinates'])
return cords
cords2stream_truck1 = get_cords_from_geojson('data/gps_04_02.geojson')
cords2stream_truck2 = get_cords_from_geojson('data/gps_03_23.geojson')
cords2stream_truck3 = get_cords_from_geojson('data/gps_03_25.geojson')
min_len = min(len(cords2stream_truck1), len(cords2stream_truck2), len(cords2stream_truck3))
current_index = 0
async def background_stream(min_len=min_len):
global current_index
while True:
current_index = (current_index + 1) % min_len
await asyncio.sleep(1)
@app.on_event("startup")
async def startup_event():
asyncio.create_task(background_stream())
async def stream_cords():
global current_index
while True:
yield f"data: {json.dumps({'truck_1': cords2stream_truck1[current_index], 'truck_2': cords2stream_truck2[current_index], 'truck_3': cords2stream_truck3[current_index]})}\n\n"
await asyncio.sleep(2)
# @app.get("/", response_class=HTMLResponse)
# def home(request: Request):
# return templates.TemplateResponse("gps_temp.html", {"request": request})
@app.get("/stream-cords")
async def stream():
return StreamingResponse(stream_cords(), media_type="text/event-stream")
@app.get("/data")
async def gps_data():
data2 = get_cords_from_geojson()
return JSONResponse(content=data2)
app.add_middleware(
CORSMiddleware,
allow_origins=["http://127.0.0.1:8888"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
if __name__ == '__main__':
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=5000)