34 lines
1.1 KiB
Python
34 lines
1.1 KiB
Python
from typing import Any
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException
|
|
from pydantic import BaseModel
|
|
|
|
from app.auth.keycloak_dependencies import get_current_keycloak_username
|
|
from app.services.burst_location import run_burst_location_by_network
|
|
|
|
router = APIRouter()
|
|
|
|
|
|
class BurstLocationRequest(BaseModel):
|
|
network: str
|
|
pressure_scada_ids: list[str]
|
|
burst_pressure: dict[str, float] | list[dict[str, Any]]
|
|
normal_pressure: dict[str, float] | list[dict[str, Any]]
|
|
burst_leakage: float
|
|
flow_scada_ids: list[str] | None = None
|
|
burst_flow: dict[str, float] | list[dict[str, Any]] | None = None
|
|
normal_flow: dict[str, float] | list[dict[str, Any]] | None = None
|
|
min_dpressure: float = 2.0
|
|
basic_pressure: float = 10.0
|
|
|
|
|
|
@router.post("/locate/")
|
|
async def locate_burst(
|
|
data: BurstLocationRequest,
|
|
_username: str = Depends(get_current_keycloak_username),
|
|
) -> dict[str, Any]:
|
|
try:
|
|
return run_burst_location_by_network(**data.model_dump())
|
|
except (TypeError, ValueError) as exc:
|
|
raise HTTPException(status_code=400, detail=str(exc))
|