66 lines
2.2 KiB
Python
66 lines
2.2 KiB
Python
from typing import Any
|
|
from datetime import datetime
|
|
|
|
from typing import Literal
|
|
|
|
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 (
|
|
get_burst_location_scheme_detail,
|
|
list_burst_location_schemes,
|
|
run_burst_location_by_network,
|
|
)
|
|
|
|
router = APIRouter()
|
|
|
|
|
|
class BurstLocationRequest(BaseModel):
|
|
network: str
|
|
data_source: Literal["monitoring", "simulation"] = "monitoring"
|
|
pressure_scada_ids: list[str] | None = None
|
|
burst_pressure: dict[str, float] | list[dict[str, Any]] | None = None
|
|
normal_pressure: dict[str, float] | list[dict[str, Any]] | None = None
|
|
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
|
|
scada_burst_start: datetime | None = None
|
|
scada_burst_end: datetime | None = None
|
|
use_scada_flow: bool = False
|
|
scheme_name: str | None = None
|
|
simulation_scheme_name: str | None = None
|
|
simulation_scheme_type: str | None = None
|
|
|
|
|
|
@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(), username=username)
|
|
except (TypeError, ValueError) as exc:
|
|
raise HTTPException(status_code=400, detail=str(exc))
|
|
|
|
|
|
@router.get("/schemes/")
|
|
async def query_burst_schemes(
|
|
network: str, query_date: datetime | None = None
|
|
) -> list[dict[str, Any]]:
|
|
try:
|
|
return list_burst_location_schemes(network=network, query_date=query_date)
|
|
except Exception as exc:
|
|
raise HTTPException(status_code=400, detail=str(exc))
|
|
|
|
|
|
@router.get("/schemes/{scheme_name}")
|
|
async def query_burst_scheme_detail(network: str, scheme_name: str) -> dict[str, Any]:
|
|
try:
|
|
return get_burst_location_scheme_detail(network=network, scheme_name=scheme_name)
|
|
except Exception as exc:
|
|
raise HTTPException(status_code=400, detail=str(exc))
|