45 lines
1.3 KiB
Python
45 lines
1.3 KiB
Python
from typing import Any, List, Dict
|
|
from fastapi import APIRouter
|
|
from app.services.tjnetwork import (
|
|
get_pipe_risk_probability_now,
|
|
get_pipe_risk_probability,
|
|
get_pipes_risk_probability,
|
|
get_network_pipe_risk_probability_now,
|
|
get_pipe_risk_probability_geometries,
|
|
)
|
|
|
|
router = APIRouter()
|
|
|
|
@router.get("/getpiperiskprobabilitynow/")
|
|
async def fastapi_get_pipe_risk_probability_now(
|
|
network: str, pipe_id: str
|
|
) -> dict[str, Any]:
|
|
return get_pipe_risk_probability_now(network, pipe_id)
|
|
|
|
|
|
@router.get("/getpiperiskprobability/")
|
|
async def fastapi_get_pipe_risk_probability(
|
|
network: str, pipe_id: str
|
|
) -> dict[str, Any]:
|
|
return get_pipe_risk_probability(network, pipe_id)
|
|
|
|
|
|
@router.get("/getpipesriskprobability/")
|
|
async def fastapi_get_pipes_risk_probability(
|
|
network: str, pipe_ids: str
|
|
) -> list[dict[str, Any]]:
|
|
pipeids = pipe_ids.split(",")
|
|
return get_pipes_risk_probability(network, pipeids)
|
|
|
|
|
|
@router.get("/getnetworkpiperiskprobabilitynow/")
|
|
async def fastapi_get_network_pipe_risk_probability_now(
|
|
network: str,
|
|
) -> list[dict[str, Any]]:
|
|
return get_network_pipe_risk_probability_now(network)
|
|
|
|
|
|
@router.get("/getpiperiskprobabilitygeometries/")
|
|
async def fastapi_get_pipe_risk_probability_geometries(network: str) -> dict[str, Any]:
|
|
return get_pipe_risk_probability_geometries(network)
|