36 lines
1.5 KiB
Python
36 lines
1.5 KiB
Python
|
from __future__ import annotations
|
||
|
|
|
||
|
|
from fastapi import APIRouter, Depends, HTTPException
|
||
|
|
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
|
||
|
|
|
||
|
|
from backend.api.schemas import UnityCommandRequest, UnityCommandResponse, UnityInitConfigRequest, UnityInitConfigResponse
|
||
|
|
from backend.auth.deps import get_current_user
|
||
|
|
from backend.services.simulation_manager import SimulationManager
|
||
|
|
|
||
|
|
|
||
|
|
def get_router(simulation_manager: SimulationManager, session_factory: async_sessionmaker[AsyncSession]) -> APIRouter:
|
||
|
|
router = APIRouter(prefix="/api/unity", tags=["unity"])
|
||
|
|
|
||
|
|
def _require_admin(user: dict) -> None:
|
||
|
|
if user.get("role_id") != "admin":
|
||
|
|
raise HTTPException(status_code=403, detail="forbidden")
|
||
|
|
|
||
|
|
@router.post("/initconfig", response_model=UnityInitConfigResponse)
|
||
|
|
async def initconfig(
|
||
|
|
body: UnityInitConfigRequest, current_user: dict = Depends(get_current_user(session_factory))
|
||
|
|
) -> UnityInitConfigResponse:
|
||
|
|
_require_admin(current_user)
|
||
|
|
simulation_id = await simulation_manager.init_config(body.payload)
|
||
|
|
return UnityInitConfigResponse(simulation_id=simulation_id)
|
||
|
|
|
||
|
|
@router.post("/command", response_model=UnityCommandResponse)
|
||
|
|
async def command(
|
||
|
|
body: UnityCommandRequest, current_user: dict = Depends(get_current_user(session_factory))
|
||
|
|
) -> UnityCommandResponse:
|
||
|
|
_require_admin(current_user)
|
||
|
|
await simulation_manager.send_command(body.payload)
|
||
|
|
return UnityCommandResponse()
|
||
|
|
|
||
|
|
return router
|
||
|
|
|