Compare commits

..

2 Commits

2 changed files with 34 additions and 60 deletions

View File

@@ -1,3 +1,4 @@
version: "3.9"
name: "road_vision" name: "road_vision"
services: services:
mqtt: mqtt:

View File

@@ -2,6 +2,7 @@ import asyncio
import json import json
from typing import Set, Dict, List, Any from typing import Set, Dict, List, Any
from fastapi import FastAPI, HTTPException, WebSocket, WebSocketDisconnect, Body from fastapi import FastAPI, HTTPException, WebSocket, WebSocketDisconnect, Body
from fastapi.encoders import jsonable_encoder
from sqlalchemy import ( from sqlalchemy import (
create_engine, create_engine,
MetaData, MetaData,
@@ -125,10 +126,36 @@ async def send_data_to_subscribers(user_id: int, data):
@app.post("/processed_agent_data/") @app.post("/processed_agent_data/")
async def create_processed_agent_data(data: List[ProcessedAgentData]): async def create_processed_agent_data(data: List[ProcessedAgentData], user_id: int = Body(..., embed=True)):
# Insert data to database session = SessionLocal()
# Send data to subscribers try:
pass created_data = [
{
"road_state": item.road_state,
"user_id": user_id,
"x": item.agent_data.accelerometer.x,
"y": item.agent_data.accelerometer.y,
"z": item.agent_data.accelerometer.z,
"latitude": item.agent_data.gps.latitude,
"longitude": item.agent_data.gps.longitude,
"timestamp": item.agent_data.timestamp,
}
for item in data
]
stmt = processed_agent_data.insert().values(created_data).returning(processed_agent_data)
result = session.execute(stmt)
created_records = [dict(row._mapping) for row in result.fetchall()]
session.commit()
for record in created_records:
await send_data_to_subscribers(user_id, jsonable_encoder(record))
return created_records
except Exception as err:
session.rollback()
print(f"Database error: {err}")
raise HTTPException(status_code=500, detail="Internal Server Error")
finally:
session.close()
@app.get( @app.get(
@@ -152,41 +179,7 @@ def list_processed_agent_data():
) )
def update_processed_agent_data(processed_agent_data_id: int, data: ProcessedAgentData): def update_processed_agent_data(processed_agent_data_id: int, data: ProcessedAgentData):
# Update data # Update data
session = SessionLocal() pass
try:
query = select(processed_agent_data).where(
processed_agent_data.c.id == processed_agent_data_id
)
result = session.execute(query).fetchone()
if not result:
raise HTTPException(status_code=404, detail="Data not found")
update_query = (
processed_agent_data.update()
.where(processed_agent_data.c.id == processed_agent_data_id)
.values(
road_state=data.road_state,
user_id=data.agent_data.user_id,
x=data.agent_data.accelerometer.x,
y=data.agent_data.accelerometer.y,
z=data.agent_data.accelerometer.z,
latitude=data.agent_data.gps.latitude,
longitude=data.agent_data.gps.longitude,
timestamp=data.agent_data.timestamp,
)
)
session.execute(update_query)
session.commit()
updated_result = session.execute(query).fetchone()
return ProcessedAgentDataInDB(**updated_result._mapping)
finally:
session.close()
@app.delete( @app.delete(
@@ -195,28 +188,8 @@ def update_processed_agent_data(processed_agent_data_id: int, data: ProcessedAge
) )
def delete_processed_agent_data(processed_agent_data_id: int): def delete_processed_agent_data(processed_agent_data_id: int):
# Delete by id # Delete by id
session = SessionLocal() pass
try:
query = select(processed_agent_data).where(
processed_agent_data.c.id == processed_agent_data_id
)
result = session.execute(query).fetchone()
if not result:
raise HTTPException(status_code=404, detail="Data not found")
delete_query = processed_agent_data.delete().where(
processed_agent_data.c.id == processed_agent_data_id
)
session.execute(delete_query)
session.commit()
return ProcessedAgentDataInDB(**result._mapping)
finally:
session.close()
if __name__ == "__main__": if __name__ == "__main__":
import uvicorn import uvicorn