jiang
fa188af0b1
Reorganize WNDB by responsibility and remove legacy scheme endpoints.\n\nRoute analysis and time-series access through project pools, preserve transactional realtime replacement, and refresh GIS materialized views after writes.\n\nAdd database architecture documentation, live pooling coverage, API contract updates, and executable container verification.\n\nBREAKING CHANGE: legacy scheme APIs and flat app.native.wndb module imports are removed.
95 lines
2.7 KiB
Python
95 lines
2.7 KiB
Python
from datetime import datetime
|
|
from typing import Literal
|
|
from uuid import UUID
|
|
|
|
from pydantic import BaseModel, Field, field_validator
|
|
|
|
|
|
AdjustmentStatus = Literal["current", "original", "added", "replaced"]
|
|
|
|
|
|
def _normalize_location_ids(value: list[str]) -> list[str]:
|
|
normalized = [str(item).strip() for item in value]
|
|
if any(not item for item in normalized):
|
|
raise ValueError("sensor locations cannot contain blank node IDs")
|
|
if len(set(normalized)) != len(normalized):
|
|
raise ValueError("sensor locations cannot contain duplicate node IDs")
|
|
return normalized
|
|
|
|
|
|
class SensorPlacementOptimizeRequest(BaseModel):
|
|
network: str = Field(
|
|
...,
|
|
min_length=1,
|
|
max_length=63,
|
|
pattern=r"^[^/\\\x00]+$",
|
|
)
|
|
run_name: str = Field(..., min_length=1, max_length=64)
|
|
sensor_type: Literal["pressure"]
|
|
method: Literal["sensitivity", "kmeans"]
|
|
sensor_count: int = Field(..., gt=0, le=200)
|
|
min_diameter: int = Field(default=0, ge=0)
|
|
|
|
@field_validator("network")
|
|
@classmethod
|
|
def validate_network(cls, value: str) -> str:
|
|
normalized = value.strip()
|
|
if normalized in {".", ".."}:
|
|
raise ValueError("network must be a project identifier")
|
|
return normalized
|
|
|
|
|
|
class SensorPlacementUpdateRequest(BaseModel):
|
|
expected_sensor_locations: list[str] = Field(
|
|
...,
|
|
min_length=1,
|
|
max_length=200,
|
|
)
|
|
sensor_locations: list[str] = Field(..., min_length=1, max_length=200)
|
|
|
|
@field_validator("expected_sensor_locations", "sensor_locations")
|
|
@classmethod
|
|
def validate_locations(cls, value: list[str]) -> list[str]:
|
|
return _normalize_location_ids(value)
|
|
|
|
|
|
class SensorPlacementExportRequest(BaseModel):
|
|
sensor_locations: list[str] = Field(..., min_length=1, max_length=200)
|
|
adjustment_status: dict[str, AdjustmentStatus] = Field(
|
|
default_factory=dict,
|
|
max_length=200,
|
|
)
|
|
|
|
@field_validator("sensor_locations")
|
|
@classmethod
|
|
def validate_locations(cls, value: list[str]) -> list[str]:
|
|
return _normalize_location_ids(value)
|
|
|
|
|
|
class SensorPointResponse(BaseModel):
|
|
node_id: str
|
|
max_pipe_diameter: float | None = Field(
|
|
...,
|
|
description="节点关联管道的最大管径，单位：毫米",
|
|
)
|
|
project_x: float
|
|
project_y: float
|
|
map_x: float
|
|
map_y: float
|
|
longitude: float
|
|
latitude: float
|
|
elevation: float
|
|
|
|
|
|
class SensorPlacementSchemeResponse(BaseModel):
|
|
run_id: UUID
|
|
name: str
|
|
sensor_count: int
|
|
min_diameter: int
|
|
created_by: str
|
|
created_at: datetime
|
|
status: str
|
|
sensor_locations: list[str]
|
|
sensor_points: list[SensorPointResponse]
|
|
can_edit: bool = False
|