From ddbb50173c06ba9ae5b383cfe43f7cc9f74c5e66 Mon Sep 17 00:00:00 2001 From: Jiang Date: Thu, 30 Jul 2026 16:16:51 +0800 Subject: [PATCH] feat(sensor-placement): add editable scheme APIs --- app/algorithms/sensor/__init__.py | 151 +++++---- app/api/v1/endpoints/network/geometry.py | 39 +-- app/api/v1/endpoints/sensor_placement.py | 208 ++++++++++++ app/api/v1/router.py | 2 + app/domain/schemas/sensor_placement.py | 81 +++++ app/infra/audit/middleware.py | 20 +- app/native/wndb/__init__.py | 13 +- app/native/wndb/s42_sensor_placement.py | 114 ++++++- app/services/sensor_placement.py | 257 +++++++++++++++ app/services/tjnetwork.py | 1 - tests/api/test_audit_middleware.py | 36 ++ tests/api/test_sensor_placement_endpoints.py | 325 +++++++++++++++++++ tests/unit/test_sensor_placement_service.py | 188 +++++++++++ 13 files changed, 1313 insertions(+), 122 deletions(-) create mode 100644 app/api/v1/endpoints/sensor_placement.py create mode 100644 app/domain/schemas/sensor_placement.py create mode 100644 app/services/sensor_placement.py create mode 100644 tests/api/test_audit_middleware.py create mode 100644 tests/api/test_sensor_placement_endpoints.py create mode 100644 tests/unit/test_sensor_placement_service.py diff --git a/app/algorithms/sensor/__init__.py b/app/algorithms/sensor/__init__.py index d6c1a48..998016b 100644 --- a/app/algorithms/sensor/__init__.py +++ b/app/algorithms/sensor/__init__.py @@ -1,14 +1,68 @@ -import psycopg +from contextlib import contextmanager +import fcntl +from pathlib import Path +from typing import Any from app.algorithms.sensor import kmeans as kmeans_sensor from app.algorithms.sensor import sensitivity -from app.core.config import get_pgconn_string +from app.native.wndb.s42_sensor_placement import create_sensor_placement +from app.services.sensor_placement import ( + SensorPlacementValidationError, + validate_sensor_placement_nodes, +) from app.services.tjnetwork import dump_inp +def _sensor_inp_path(name: str) -> Path: + if ( + not name + or name in {".", ".."} + or "/" in name + or "\\" in name + or "\x00" in name + ): + raise SensorPlacementValidationError("管网名称不是有效的项目标识") + return Path("db_inp") / f"{name}.db.inp" + + +@contextmanager +def _sensor_inp_lock(name: str): + inp_path = _sensor_inp_path(name) + inp_path.parent.mkdir(parents=True, exist_ok=True) + lock_path = inp_path.with_suffix(".sensor.lock") + with lock_path.open("w", encoding="utf-8") as lock_file: + fcntl.flock(lock_file.fileno(), fcntl.LOCK_EX) + try: + yield inp_path + finally: + fcntl.flock(lock_file.fileno(), fcntl.LOCK_UN) + + +def _create_validated_placement( + name: str, + *, + scheme_name: str, + min_diameter: int, + username: str, + sensor_location: list[str], +) -> dict[str, Any]: + validate_sensor_placement_nodes(name, sensor_location) + return create_sensor_placement( + name, + scheme_name=scheme_name, + min_diameter=min_diameter, + username=username, + sensor_location=sensor_location, + ) + + def pressure_sensor_placement_sensitivity( - name: str, scheme_name: str, sensor_number: int, min_diameter: int, username: str -) -> None: + name: str, + scheme_name: str, + sensor_number: int, + min_diameter: int, + username: str, +) -> dict[str, Any]: """ 基于改进灵敏度法进行压力监测点优化布置 :param name: 数据库名称 @@ -16,41 +70,32 @@ def pressure_sensor_placement_sensitivity( :param sensor_number: 传感器数目 :param min_diameter: 最小管径 :param username: 用户名 - :return: + :return: 新建的监测点方案 """ - sensor_location = sensitivity.get_ID( - name=name, sensor_num=sensor_number, min_diameter=min_diameter + with _sensor_inp_lock(name): + sensor_location = sensitivity.get_ID( + name=name, + sensor_num=sensor_number, + min_diameter=min_diameter, + ) + return _create_validated_placement( + name, + scheme_name=scheme_name, + min_diameter=min_diameter, + username=username, + sensor_location=sensor_location, ) - try: - conn_string = get_pgconn_string(db_name=name) - with psycopg.connect(conn_string) as conn: - with conn.cursor() as cur: - sql = """ - INSERT INTO sensor_placement (scheme_name, sensor_number, min_diameter, username, sensor_location) - VALUES (%s, %s, %s, %s, %s) - """ - - cur.execute( - sql, - ( - scheme_name, - sensor_number, - min_diameter, - username, - sensor_location, - ), - ) - conn.commit() - print("方案信息存储成功！") - except Exception as e: - print(f"存储方案信息时出错：{e}") # 2025/08/21 # 基于kmeans聚类法进行压力监测点优化布置 def pressure_sensor_placement_kmeans( - name: str, scheme_name: str, sensor_number: int, min_diameter: int, username: str -) -> None: + name: str, + scheme_name: str, + sensor_number: int, + min_diameter: int, + username: str, +) -> dict[str, Any]: """ 基于聚类法进行压力监测点优化布置 :param name: 数据库名称（注意，此处数据库名称也是inp文件名称，inp文件与pg库名要一样） @@ -58,34 +103,20 @@ def pressure_sensor_placement_kmeans( :param sensor_number: 传感器数目 :param min_diameter: 最小管径 :param username: 用户名 - :return: + :return: 新建的监测点方案 """ # dump_inp - inp_name = f"./db_inp/{name}.db.inp" - dump_inp(name, inp_name, "2") - sensor_location = kmeans_sensor.kmeans_sensor_placement( - name=name, sensor_num=sensor_number, min_diameter=min_diameter + with _sensor_inp_lock(name) as inp_path: + dump_inp(name, str(inp_path), "2") + sensor_location = kmeans_sensor.kmeans_sensor_placement( + name=name, + sensor_num=sensor_number, + min_diameter=min_diameter, + ) + return _create_validated_placement( + name, + scheme_name=scheme_name, + min_diameter=min_diameter, + username=username, + sensor_location=sensor_location, ) - try: - conn_string = get_pgconn_string(db_name=name) - with psycopg.connect(conn_string) as conn: - with conn.cursor() as cur: - sql = """ - INSERT INTO sensor_placement (scheme_name, sensor_number, min_diameter, username, sensor_location) - VALUES (%s, %s, %s, %s, %s) - """ - - cur.execute( - sql, - ( - scheme_name, - sensor_number, - min_diameter, - username, - sensor_location, - ), - ) - conn.commit() - print("方案信息存储成功！") - except Exception as e: - print(f"存储方案信息时出错：{e}") diff --git a/app/api/v1/endpoints/network/geometry.py b/app/api/v1/endpoints/network/geometry.py index 5adb575..8acf2f4 100644 --- a/app/api/v1/endpoints/network/geometry.py +++ b/app/api/v1/endpoints/network/geometry.py @@ -1,18 +1,14 @@ -from fastapi import APIRouter, Request, Depends, Query, Path, Body -from typing import Any, List, Dict, Union +from typing import Any + +from fastapi import APIRouter, Query + from app.services.tjnetwork import ( - Any, - get_all_scada_info, get_major_node_coords, get_major_pipe_nodes, get_network_in_extent, get_network_link_nodes, - get_network_node_coords, get_node_coord, ) -from app.auth.metadata_dependencies import get_current_metadata_user -from app.infra.cache.redis_client import redis_client, encode_datetime, decode_datetime -import msgpack router = APIRouter() @@ -62,33 +58,6 @@ async def fastapi_get_network_in_extent( """获取地理范围内的网络几何信息。""" return get_network_in_extent(network, x1, y1, x2, y2) -@router.get( - "/getnetworkgeometries/", - dependencies=[Depends(get_current_metadata_user)], - summary="获取完整网络几何信息", - description="获取整个水网的所有节点、管线和SCADA点的几何信息(需要身份验证)" -) -async def fastapi_get_network_geometries( - network: str = Query(..., description="管网名称（或数据库名称）") -) -> dict[str, Any] | None: - """获取完整的网络几何信息，包括所有节点、管线和SCADA点。结果从缓存返回。""" - cache_key = f"getnetworkgeometries_{network}" - data = redis_client.get(cache_key) - if data: - loaded_dict = msgpack.unpackb(data, object_hook=decode_datetime) - return loaded_dict - - coords = get_network_node_coords(network) - nodes = [] - for node_id, coord in coords.items(): - nodes.append(f"{node_id}:{coord['type']}:{coord['x']}:{coord['y']}") - links = get_network_link_nodes(network) - scadas = get_all_scada_info(network) - - results = {"nodes": nodes, "links": links, "scadas": scadas} - redis_client.set(cache_key, msgpack.packb(results, default=encode_datetime)) - return results - @router.get( "/getmajornodecoords/", summary="获取主要节点坐标", diff --git a/app/api/v1/endpoints/sensor_placement.py b/app/api/v1/endpoints/sensor_placement.py new file mode 100644 index 0000000..454da49 --- /dev/null +++ b/app/api/v1/endpoints/sensor_placement.py @@ -0,0 +1,208 @@ +import logging +from typing import Any +from urllib.parse import quote + +from fastapi import APIRouter, Depends, HTTPException, Query, status +from fastapi.responses import StreamingResponse +from starlette.concurrency import run_in_threadpool + +from app.algorithms.sensor import ( + pressure_sensor_placement_kmeans, + pressure_sensor_placement_sensitivity, +) +from app.auth.metadata_dependencies import get_current_metadata_user +from app.auth.project_dependencies import ProjectContext, get_project_context +from app.domain.schemas.sensor_placement import ( + SensorPlacementExportRequest, + SensorPlacementOptimizeRequest, + SensorPlacementSchemeResponse, + SensorPlacementUpdateRequest, +) +from app.services.sensor_placement import ( + SensorPlacementConflictError, + SensorPlacementNotFoundError, + SensorPlacementValidationError, + build_sensor_placement_workbook, + can_edit_sensor_placement, + get_sensor_placement_scheme, + update_sensor_placement_scheme, +) + +router = APIRouter() +logger = logging.getLogger(__name__) + + +def _project_network(network: str, project_context: ProjectContext) -> str: + if network != project_context.project_code: + raise HTTPException( + status_code=status.HTTP_403_FORBIDDEN, + detail="请求的管网不属于当前项目", + ) + return project_context.project_code + + +def _service_http_error(exc: Exception) -> HTTPException: + if isinstance(exc, SensorPlacementNotFoundError): + return HTTPException( + status_code=status.HTTP_404_NOT_FOUND, + detail=str(exc), + ) + if isinstance(exc, SensorPlacementConflictError): + return HTTPException( + status_code=status.HTTP_409_CONFLICT, + detail=str(exc), + ) + return HTTPException( + status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, + detail=str(exc), + ) + + +def _get_scheme_response( + network: str, + scheme_id: int, + current_user: Any, +) -> dict[str, Any]: + try: + scheme = get_sensor_placement_scheme(network, scheme_id) + return { + **scheme, + "can_edit": can_edit_sensor_placement(current_user, scheme), + } + except ( + SensorPlacementNotFoundError, + SensorPlacementValidationError, + ) as exc: + raise _service_http_error(exc) from exc + + +@router.post( + "/sensor-placement-schemes/optimize", + response_model=SensorPlacementSchemeResponse, + summary="创建并返回监测点优化方案", +) +async def optimize_sensor_placement_scheme( + payload: SensorPlacementOptimizeRequest, + project_context: ProjectContext = Depends(get_project_context), + current_user=Depends(get_current_metadata_user), +) -> dict[str, Any]: + network = _project_network(payload.network, project_context) + optimizer = ( + pressure_sensor_placement_sensitivity + if payload.method == "sensitivity" + else pressure_sensor_placement_kmeans + ) + try: + created = await run_in_threadpool( + optimizer, + name=network, + scheme_name=payload.scheme_name, + sensor_number=payload.sensor_count, + min_diameter=payload.min_diameter, + username=current_user.username, + ) + scheme = get_sensor_placement_scheme(network, int(created["id"])) + return {**scheme, "can_edit": True} + except (SensorPlacementValidationError, ValueError) as exc: + raise _service_http_error(exc) from exc + except Exception as exc: + logger.exception("Sensor placement optimization failed") + raise HTTPException( + status_code=500, + detail="监测点优化失败，请稍后重试", + ) from exc + + +@router.get( + "/sensor-placement-schemes/{scheme_id}", + response_model=SensorPlacementSchemeResponse, + summary="获取监测点方案详情", +) +async def get_sensor_placement_scheme_detail( + scheme_id: int, + network: str = Query(..., min_length=1), + project_context: ProjectContext = Depends(get_project_context), + current_user=Depends(get_current_metadata_user), +) -> dict[str, Any]: + return _get_scheme_response( + _project_network(network, project_context), + scheme_id, + current_user, + ) + + +@router.put( + "/sensor-placement-schemes/{scheme_id}", + response_model=SensorPlacementSchemeResponse, + summary="覆盖保存监测点方案", +) +async def overwrite_sensor_placement_scheme( + scheme_id: int, + payload: SensorPlacementUpdateRequest, + network: str = Query(..., min_length=1), + project_context: ProjectContext = Depends(get_project_context), + current_user=Depends(get_current_metadata_user), +) -> dict[str, Any]: + network = _project_network(network, project_context) + scheme = _get_scheme_response(network, scheme_id, current_user) + if not scheme["can_edit"]: + raise HTTPException(status_code=403, detail="无权修改该监测点方案") + + try: + updated = update_sensor_placement_scheme( + network, + scheme_id, + expected_sensor_location=payload.expected_sensor_location, + sensor_location=payload.sensor_location, + ) + return {**updated, "can_edit": True} + except ( + SensorPlacementConflictError, + SensorPlacementNotFoundError, + SensorPlacementValidationError, + ) as exc: + raise _service_http_error(exc) from exc + + +@router.post( + "/sensor-placement-schemes/{scheme_id}/exports/excel", + summary="导出监测点工程清单", +) +async def export_sensor_placement_excel( + scheme_id: int, + payload: SensorPlacementExportRequest, + network: str = Query(..., min_length=1), + project_context: ProjectContext = Depends(get_project_context), + current_user=Depends(get_current_metadata_user), +) -> StreamingResponse: + network = _project_network(network, project_context) + scheme = _get_scheme_response(network, scheme_id, current_user) + if ( + payload.sensor_location != scheme["sensor_location"] + and not scheme["can_edit"] + ): + raise HTTPException(status_code=403, detail="无权导出该方案的未保存草稿") + + try: + workbook = build_sensor_placement_workbook( + network=network, + scheme=scheme, + sensor_location=payload.sensor_location, + adjustment_status=payload.adjustment_status, + ) + except SensorPlacementValidationError as exc: + raise _service_http_error(exc) from exc + + filename = f"{scheme['scheme_name']}_监测点清单.xlsx" + encoded_filename = quote(filename) + return StreamingResponse( + workbook, + media_type=( + "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet" + ), + headers={ + "Content-Disposition": ( + f"attachment; filename*=UTF-8''{encoded_filename}" + ) + }, + ) diff --git a/app/api/v1/router.py b/app/api/v1/router.py index 99da712..df73930 100644 --- a/app/api/v1/router.py +++ b/app/api/v1/router.py @@ -5,6 +5,7 @@ from app.api.v1.endpoints import ( project, simulation, scada, + sensor_placement, extension, snapshots, # data_query, @@ -89,6 +90,7 @@ api_router.include_router(visuals.router, tags=["Visuals"]) api_router.include_router(simulation.router, tags=["Simulation Control"]) # api_router.include_router(data_query.router, tags=["Data Query & InfluxDB"]) api_router.include_router(scada.router) +api_router.include_router(sensor_placement.router, tags=["Sensor Placement"]) api_router.include_router(snapshots.router, tags=["Snapshots"]) api_router.include_router(users.router, tags=["Users"]) api_router.include_router(schemes.router, tags=["Schemes"]) diff --git a/app/domain/schemas/sensor_placement.py b/app/domain/schemas/sensor_placement.py new file mode 100644 index 0000000..8dcfe69 --- /dev/null +++ b/app/domain/schemas/sensor_placement.py @@ -0,0 +1,81 @@ +from datetime import datetime +from typing import Literal + +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]+$", + ) + scheme_name: str = Field(..., min_length=1, max_length=32) + 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_location: list[str] = Field(..., min_length=1) + sensor_location: list[str] = Field(..., min_length=1) + + @field_validator("expected_sensor_location", "sensor_location") + @classmethod + def validate_locations(cls, value: list[str]) -> list[str]: + return _normalize_location_ids(value) + + +class SensorPlacementExportRequest(BaseModel): + sensor_location: list[str] = Field(..., min_length=1) + adjustment_status: dict[str, AdjustmentStatus] = Field(default_factory=dict) + + @field_validator("sensor_location") + @classmethod + def validate_locations(cls, value: list[str]) -> list[str]: + return _normalize_location_ids(value) + + +class SensorPointResponse(BaseModel): + node_id: str + project_x: float + project_y: float + map_x: float + map_y: float + longitude: float + latitude: float + elevation: float + + +class SensorPlacementSchemeResponse(BaseModel): + id: int + scheme_name: str + sensor_number: int + min_diameter: int + username: str + create_time: datetime + sensor_location: list[str] + sensor_points: list[SensorPointResponse] + can_edit: bool = False diff --git a/app/infra/audit/middleware.py b/app/infra/audit/middleware.py index d0774d4..fbb3f02 100644 --- a/app/infra/audit/middleware.py +++ b/app/infra/audit/middleware.py @@ -58,6 +58,8 @@ class AuditMiddleware(BaseHTTPMiddleware): "/meta/projects", "/api/v1/openproject/", "/openproject/", + "/api/v1/audit/session-events", + "/audit/session-events", } EXCLUDED_PATH_PREFIXES = ( ) @@ -80,27 +82,9 @@ class AuditMiddleware(BaseHTTPMiddleware): request_data = None if should_capture_body: try: - # 注意：读取 body 后需要重新设置，避免影响后续处理 - original_receive = request._receive body = await request.body() if body: request_data = json.loads(body.decode()) - - # 重新构造请求以供后续使用：仅回放一次，后续回落原始 receive - body_sent = False - - async def receive(): - nonlocal body_sent - if not body_sent: - body_sent = True - return { - "type": "http.request", - "body": body, - "more_body": False, - } - return await original_receive() - - request._receive = receive except Exception as e: logger.warning(f"Failed to read request body for audit: {e}") diff --git a/app/native/wndb/__init__.py b/app/native/wndb/__init__.py index 57d230b..4765495 100644 --- a/app/native/wndb/__init__.py +++ b/app/native/wndb/__init__.py @@ -320,7 +320,11 @@ from .s23_options_util import ( from .s23_options_util import get_option_v3_schema, get_option_v3 from .batch_api import set_option_v3_ex -from .s24_coordinates import get_node_coord, get_nodes_in_extent, get_links_in_extent +from .s24_coordinates import ( + get_links_in_extent, + get_node_coord, + get_nodes_in_extent, +) from .s25_vertices import ( get_vertex_schema, @@ -468,6 +472,11 @@ from .s41_pipe_risk_probability import ( get_pipe_risk_probability_geometries, ) -from .s42_sensor_placement import get_all_sensor_placements +from .s42_sensor_placement import ( + get_all_sensor_placements, + get_sensor_placement, + get_sensor_placement_nodes, + update_sensor_placement, +) from .s43_burst_locate_result import get_all_burst_locate_results diff --git a/app/native/wndb/s42_sensor_placement.py b/app/native/wndb/s42_sensor_placement.py index 5ad8b2a..c65dc66 100644 --- a/app/native/wndb/s42_sensor_placement.py +++ b/app/native/wndb/s42_sensor_placement.py @@ -1,7 +1,109 @@ -from .database import * -from .s0_base import * -from .s42_sensor_placement import * -import json +from typing import Any -def get_all_sensor_placements(name: str) -> list[dict[Any, Any]]: - return read_all(name, "select * from sensor_placement") \ No newline at end of file +from psycopg.rows import dict_row + +from .connection import project_connection +from .database import read_all + + +def get_all_sensor_placements(name: str) -> list[dict[str, Any]]: + return read_all(name, "select * from sensor_placement") + + +def create_sensor_placement( + name: str, + *, + scheme_name: str, + min_diameter: int, + username: str, + sensor_location: list[str], +) -> dict[str, Any]: + with project_connection(name) as conn: + with conn.cursor(row_factory=dict_row) as cur: + cur.execute( + """ + INSERT INTO sensor_placement ( + scheme_name, + sensor_number, + min_diameter, + username, + sensor_location + ) + VALUES (%s, %s, %s, %s, %s) + RETURNING * + """, + ( + scheme_name, + len(sensor_location), + min_diameter, + username, + sensor_location, + ), + ) + created = cur.fetchone() + if created is None: + raise RuntimeError("监测点方案写入失败") + return dict(created) + + +def get_sensor_placement(name: str, scheme_id: int) -> dict[str, Any] | None: + with project_connection(name) as conn: + with conn.cursor(row_factory=dict_row) as cur: + cur.execute( + "SELECT * FROM sensor_placement WHERE id = %s", + (scheme_id,), + ) + return cur.fetchone() + + +def get_sensor_placement_nodes( + name: str, + node_ids: list[str], +) -> list[dict[str, Any]]: + if not node_ids: + return [] + with project_connection(name) as conn: + with conn.cursor(row_factory=dict_row) as cur: + cur.execute( + """ + SELECT DISTINCT ON (gj.id) + gj.id AS node_id, + gj.elevation, + ST_X(c.coord) AS project_x, + ST_Y(c.coord) AS project_y, + ST_X(gj.geom) AS map_x, + ST_Y(gj.geom) AS map_y + FROM geo_junctions_mat AS gj + JOIN coordinates AS c ON c.node = gj.id + WHERE gj.id = ANY(%s) + ORDER BY gj.id + """, + (node_ids,), + ) + return list(cur.fetchall()) + + +def update_sensor_placement( + name: str, + scheme_id: int, + *, + expected_sensor_location: list[str], + sensor_location: list[str], +) -> dict[str, Any] | None: + with project_connection(name) as conn: + with conn.cursor(row_factory=dict_row) as cur: + cur.execute( + """ + UPDATE sensor_placement + SET sensor_location = %s, sensor_number = %s + WHERE id = %s AND sensor_location = %s + RETURNING * + """, + ( + sensor_location, + len(sensor_location), + scheme_id, + expected_sensor_location, + ), + ) + return cur.fetchone() diff --git a/app/services/sensor_placement.py b/app/services/sensor_placement.py new file mode 100644 index 0000000..3d44686 --- /dev/null +++ b/app/services/sensor_placement.py @@ -0,0 +1,257 @@ +from datetime import datetime +from io import BytesIO +from typing import Any + +from openpyxl import Workbook +from openpyxl.styles import Alignment, Font, PatternFill +from openpyxl.worksheet.worksheet import Worksheet +from openpyxl.utils import get_column_letter +from pyproj import Transformer + +from app.native import wndb + + +class SensorPlacementNotFoundError(LookupError): + pass + + +class SensorPlacementValidationError(ValueError): + pass + + +class SensorPlacementConflictError(RuntimeError): + pass + + +_to_wgs84 = Transformer.from_crs("EPSG:3857", "EPSG:4326", always_xy=True) +_STATUS_LABELS = { + "current": "当前方案", + "original": "原方案", + "added": "新增", + "replaced": "替换", +} +_COORDINATE_DESCRIPTION = ( + "工程 X/Y: 项目地方坐标系；地图 X/Y: EPSG:3857；经纬度: WGS84" +) +_LIST_HEADERS = ( + "序号", + "节点 ID", + "经度", + "纬度", + "工程 X", + "工程 Y", + "地图 X", + "地图 Y", + "高程", + "调整状态", +) +_LIST_COLUMN_WIDTHS = (8, 20, 16, 16, 18, 18, 18, 18, 14, 14) + + +def _normalize_locations(sensor_location: list[str]) -> list[str]: + normalized = [str(node_id).strip() for node_id in sensor_location] + if not normalized or any(not node_id for node_id in normalized): + raise SensorPlacementValidationError("监测点列表不能为空") + if len(set(normalized)) != len(normalized): + raise SensorPlacementValidationError("监测点列表不能包含重复节点") + return normalized + + +def _sensor_points( + network: str, + sensor_location: list[str], +) -> list[dict[str, Any]]: + nodes = wndb.get_sensor_placement_nodes(network, sensor_location) + by_id = {str(node["node_id"]): node for node in nodes} + missing = [node_id for node_id in sensor_location if node_id not in by_id] + if missing: + raise SensorPlacementValidationError( + f"以下节点不存在或不是 junction: {', '.join(missing)}" + ) + + points: list[dict[str, Any]] = [] + for node_id in sensor_location: + node = by_id[node_id] + project_x = float(node["project_x"]) + project_y = float(node["project_y"]) + map_x = float(node["map_x"]) + map_y = float(node["map_y"]) + longitude, latitude = _to_wgs84.transform(map_x, map_y) + points.append( + { + "node_id": node_id, + "project_x": project_x, + "project_y": project_y, + "map_x": map_x, + "map_y": map_y, + "longitude": float(longitude), + "latitude": float(latitude), + "elevation": float(node["elevation"]), + } + ) + return points + + +def validate_sensor_placement_nodes( + network: str, + sensor_location: list[str], +) -> None: + _sensor_points(network, _normalize_locations(sensor_location)) + + +def get_sensor_placement_scheme(network: str, scheme_id: int) -> dict[str, Any]: + scheme = wndb.get_sensor_placement(network, scheme_id) + if scheme is None: + raise SensorPlacementNotFoundError("监测点方案不存在") + + locations = [str(item) for item in (scheme.get("sensor_location") or [])] + return { + **scheme, + "sensor_number": len(locations), + "sensor_location": locations, + "sensor_points": _sensor_points(network, locations), + } + + +def update_sensor_placement_scheme( + network: str, + scheme_id: int, + *, + expected_sensor_location: list[str], + sensor_location: list[str], +) -> dict[str, Any]: + expected = _normalize_locations(expected_sensor_location) + next_locations = _normalize_locations(sensor_location) + _sensor_points(network, next_locations) + + updated = wndb.update_sensor_placement( + network, + scheme_id, + expected_sensor_location=expected, + sensor_location=next_locations, + ) + if updated is None: + if wndb.get_sensor_placement(network, scheme_id) is None: + raise SensorPlacementNotFoundError("监测点方案不存在") + raise SensorPlacementConflictError("方案已被其他用户修改，请重新加载") + return get_sensor_placement_scheme(network, scheme_id) + + +def can_edit_sensor_placement(user: Any, scheme: dict[str, Any]) -> bool: + return bool( + getattr(user, "is_superuser", False) + or getattr(user, "role", None) == "admin" + or getattr(user, "username", None) == scheme.get("username") + ) + + +def _safe_excel_text(value: Any) -> str: + text = "" if value is None else str(value) + if text.startswith(("=", "+", "-", "@")): + return f"'{text}" + return text + + +def _populate_info_sheet( + sheet: Worksheet, + *, + network: str, + scheme: dict[str, Any], + location_count: int, + is_draft: bool, +) -> None: + created_at = scheme["create_time"] + if isinstance(created_at, datetime): + created_at = created_at.isoformat(timespec="minutes") + + rows = [ + ("项目", network), + ("方案名称", scheme["scheme_name"]), + ("监测点数量", location_count), + ("最小管径", scheme["min_diameter"]), + ("创建人", scheme["username"]), + ("创建时间", created_at), + ("导出时间", datetime.now().astimezone().isoformat(timespec="minutes")), + ("文档状态", "未保存草稿" if is_draft else "当前方案"), + ("坐标说明", _COORDINATE_DESCRIPTION), + ] + for row_index, (label, value) in enumerate(rows, start=1): + sheet.cell(row=row_index, column=1, value=label) + safe_value = _safe_excel_text(value) if isinstance(value, str) else value + sheet.cell(row=row_index, column=2, value=safe_value) + sheet.column_dimensions["A"].width = 18 + sheet.column_dimensions["B"].width = 64 + + +def _populate_list_sheet( + sheet: Worksheet, + *, + points: list[dict[str, Any]], + adjustment_status: dict[str, str], +) -> None: + sheet.append(_LIST_HEADERS) + for index, point in enumerate(points, start=1): + status = adjustment_status.get(point["node_id"], "current") + sheet.append( + [ + index, + _safe_excel_text(point["node_id"]), + point["longitude"], + point["latitude"], + point["project_x"], + point["project_y"], + point["map_x"], + point["map_y"], + point["elevation"], + _STATUS_LABELS.get(status, "当前方案"), + ] + ) + + header_fill = PatternFill("solid", fgColor="257DD4") + for cell in sheet[1]: + cell.fill = header_fill + cell.font = Font(color="FFFFFF", bold=True) + cell.alignment = Alignment(horizontal="center", vertical="center") + sheet.freeze_panes = "A2" + sheet.auto_filter.ref = sheet.dimensions + for index, width in enumerate(_LIST_COLUMN_WIDTHS, start=1): + sheet.column_dimensions[get_column_letter(index)].width = width + for row in sheet.iter_rows(min_row=2): + row[0].alignment = Alignment(horizontal="center") + for cell in row[2:9]: + cell.number_format = "0.000000" + + +def build_sensor_placement_workbook( + *, + network: str, + scheme: dict[str, Any], + sensor_location: list[str], + adjustment_status: dict[str, str], +) -> BytesIO: + locations = _normalize_locations(sensor_location) + points = _sensor_points(network, locations) + is_draft = locations != list(scheme["sensor_location"]) + + workbook = Workbook() + info_sheet = workbook.active + info_sheet.title = "方案信息" + _populate_info_sheet( + info_sheet, + network=network, + scheme=scheme, + location_count=len(locations), + is_draft=is_draft, + ) + + list_sheet = workbook.create_sheet("监测点清单") + _populate_list_sheet( + list_sheet, + points=points, + adjustment_status=adjustment_status, + ) + + output = BytesIO() + workbook.save(output) + output.seek(0) + return output diff --git a/app/services/tjnetwork.py b/app/services/tjnetwork.py index 25f4a0f..fa36d76 100644 --- a/app/services/tjnetwork.py +++ b/app/services/tjnetwork.py @@ -1370,4 +1370,3 @@ def get_all_sensor_placements(name: str) -> list[dict[Any, Any]]: ############################################################ def get_all_burst_locate_results(name: str) -> list[dict[Any, Any]]: return api.get_all_burst_locate_results(name) - diff --git a/tests/api/test_audit_middleware.py b/tests/api/test_audit_middleware.py new file mode 100644 index 0000000..96f53ac --- /dev/null +++ b/tests/api/test_audit_middleware.py @@ -0,0 +1,36 @@ +from io import BytesIO +from unittest.mock import AsyncMock + +from fastapi import FastAPI +from fastapi.responses import StreamingResponse +from fastapi.testclient import TestClient + +from app.infra.audit import middleware as audit_middleware +from app.infra.audit.middleware import AuditMiddleware + + +def test_post_streaming_response_survives_audit_body_capture(monkeypatch): + log_audit_event = AsyncMock() + monkeypatch.setattr(audit_middleware, "log_audit_event", log_audit_event) + + app = FastAPI() + app.add_middleware(AuditMiddleware) + + @app.post("/exports") + async def export(payload: dict[str, str]) -> StreamingResponse: + assert payload == {"format": "xlsx"} + return StreamingResponse( + BytesIO(b"xlsx"), + media_type=( + "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet" + ), + ) + + response = TestClient(app, raise_server_exceptions=False).post( + "/exports", + json={"format": "xlsx"}, + ) + + assert response.status_code == 200 + assert response.content == b"xlsx" + log_audit_event.assert_awaited_once() diff --git a/tests/api/test_sensor_placement_endpoints.py b/tests/api/test_sensor_placement_endpoints.py new file mode 100644 index 0000000..1b67ef0 --- /dev/null +++ b/tests/api/test_sensor_placement_endpoints.py @@ -0,0 +1,325 @@ +from datetime import datetime, timezone +from io import BytesIO +from types import SimpleNamespace + +from fastapi.testclient import TestClient + +from tests.conftest import build_test_app, install_stub, load_module_from_path + + +class NotFoundError(LookupError): + pass + + +class ValidationError(ValueError): + pass + + +class ConflictError(RuntimeError): + pass + + +def _scheme(**overrides): + value = { + "id": 7, + "scheme_name": "北区测压点", + "sensor_number": 2, + "min_diameter": 300, + "username": "alice", + "create_time": datetime(2026, 7, 30, 8, 0, tzinfo=timezone.utc), + "sensor_location": ["J1", "J2"], + "sensor_points": [ + { + "node_id": "J1", + "project_x": 13500000.0, + "project_y": 3600000.0, + "map_x": 13500000.0, + "map_y": 3600000.0, + "longitude": 121.0, + "latitude": 31.0, + "elevation": 4.5, + }, + { + "node_id": "J2", + "project_x": 13500100.0, + "project_y": 3600100.0, + "map_x": 13500100.0, + "map_y": 3600100.0, + "longitude": 121.001, + "latitude": 31.001, + "elevation": 5.0, + }, + ], + } + value.update(overrides) + return value + + +def _load_module(monkeypatch): + install_stub(monkeypatch, "app.algorithms", package=True) + install_stub( + monkeypatch, + "app.algorithms.sensor", + { + "pressure_sensor_placement_kmeans": lambda **kwargs: {"id": 7}, + "pressure_sensor_placement_sensitivity": lambda **kwargs: {"id": 7}, + }, + ) + install_stub(monkeypatch, "app.auth", package=True) + + async def current_user(): + return SimpleNamespace( + username="alice", + role="user", + is_superuser=False, + ) + + install_stub( + monkeypatch, + "app.auth.metadata_dependencies", + {"get_current_metadata_user": current_user}, + ) + + class ProjectContext: + def __init__(self, project_code: str): + self.project_code = project_code + + async def project_context(): + return ProjectContext("tjwater") + + install_stub( + monkeypatch, + "app.auth.project_dependencies", + { + "ProjectContext": ProjectContext, + "get_project_context": project_context, + }, + ) + install_stub(monkeypatch, "app.services", package=True) + install_stub( + monkeypatch, + "app.services.sensor_placement", + { + "SensorPlacementConflictError": ConflictError, + "SensorPlacementNotFoundError": NotFoundError, + "SensorPlacementValidationError": ValidationError, + "build_sensor_placement_workbook": lambda **kwargs: BytesIO(b"xlsx"), + "can_edit_sensor_placement": ( + lambda user, scheme: user.username == scheme["username"] + or user.role == "admin" + or user.is_superuser + ), + "get_sensor_placement_scheme": lambda network, scheme_id: _scheme( + id=scheme_id + ), + "update_sensor_placement_scheme": ( + lambda network, scheme_id, **kwargs: _scheme( + id=scheme_id, + sensor_location=kwargs["sensor_location"], + sensor_number=len(kwargs["sensor_location"]), + ) + ), + }, + ) + return load_module_from_path( + "tests_sensor_placement_endpoints_module", + "app/api/v1/endpoints/sensor_placement.py", + ) + + +def _client(module, user=None): + app = build_test_app(module.router, "/api/v1") + if user is None: + user = SimpleNamespace( + username="alice", + role="user", + is_superuser=False, + ) + app.dependency_overrides[module.get_current_metadata_user] = lambda: user + app.dependency_overrides[module.get_project_context] = lambda: ( + module.ProjectContext("tjwater") + ) + return TestClient(app) + + +def test_optimize_returns_created_scheme(monkeypatch): + module = _load_module(monkeypatch) + captured = {} + + def optimize(**kwargs): + captured.update(kwargs) + return {"id": 7} + + monkeypatch.setattr(module, "pressure_sensor_placement_kmeans", optimize) + response = _client(module).post( + "/api/v1/sensor-placement-schemes/optimize", + json={ + "network": "tjwater", + "scheme_name": "北区测压点", + "sensor_type": "pressure", + "method": "kmeans", + "sensor_count": 2, + "min_diameter": 300, + }, + ) + + assert response.status_code == 200 + assert response.json()["sensor_location"] == ["J1", "J2"] + assert captured["username"] == "alice" + + +def test_optimize_rejects_unsupported_sensor_type(monkeypatch): + module = _load_module(monkeypatch) + response = _client(module).post( + "/api/v1/sensor-placement-schemes/optimize", + json={ + "network": "tjwater", + "scheme_name": "北区测流点", + "sensor_type": "flow", + "method": "kmeans", + "sensor_count": 2, + "min_diameter": 300, + }, + ) + + assert response.status_code == 422 + + +def test_optimize_rejects_network_outside_project_context(monkeypatch): + module = _load_module(monkeypatch) + response = _client(module).post( + "/api/v1/sensor-placement-schemes/optimize", + json={ + "network": "other_project", + "scheme_name": "越权方案", + "sensor_type": "pressure", + "method": "kmeans", + "sensor_count": 2, + "min_diameter": 300, + }, + ) + + assert response.status_code == 403 + + +def test_optimize_rejects_network_path_traversal(monkeypatch): + module = _load_module(monkeypatch) + response = _client(module).post( + "/api/v1/sensor-placement-schemes/optimize", + json={ + "network": "../other_project", + "scheme_name": "非法路径", + "sensor_type": "pressure", + "method": "kmeans", + "sensor_count": 2, + "min_diameter": 300, + }, + ) + + assert response.status_code == 422 + + +def test_optimize_rejects_unbounded_sensor_count(monkeypatch): + module = _load_module(monkeypatch) + response = _client(module).post( + "/api/v1/sensor-placement-schemes/optimize", + json={ + "network": "tjwater", + "scheme_name": "超大方案", + "sensor_type": "pressure", + "method": "kmeans", + "sensor_count": 201, + "min_diameter": 300, + }, + ) + + assert response.status_code == 422 + + +def test_update_rejects_non_owner(monkeypatch): + module = _load_module(monkeypatch) + response = _client( + module, + SimpleNamespace(username="bob", role="user", is_superuser=False), + ).put( + "/api/v1/sensor-placement-schemes/7", + params={"network": "tjwater"}, + json={ + "expected_sensor_location": ["J1", "J2"], + "sensor_location": ["J1", "J3"], + }, + ) + + assert response.status_code == 403 + + +def test_admin_can_overwrite_scheme(monkeypatch): + module = _load_module(monkeypatch) + response = _client( + module, + SimpleNamespace(username="ops", role="admin", is_superuser=False), + ).put( + "/api/v1/sensor-placement-schemes/7", + params={"network": "tjwater"}, + json={ + "expected_sensor_location": ["J1", "J2"], + "sensor_location": ["J1", "J3"], + }, + ) + + assert response.status_code == 200 + assert response.json()["sensor_number"] == 2 + assert response.json()["sensor_location"] == ["J1", "J3"] + + +def test_update_maps_concurrent_change_to_409(monkeypatch): + module = _load_module(monkeypatch) + + def conflict(*args, **kwargs): + raise ConflictError("方案已被其他用户修改，请重新加载") + + monkeypatch.setattr(module, "update_sensor_placement_scheme", conflict) + response = _client(module).put( + "/api/v1/sensor-placement-schemes/7", + params={"network": "tjwater"}, + json={ + "expected_sensor_location": ["J1", "J2"], + "sensor_location": ["J1", "J3"], + }, + ) + + assert response.status_code == 409 + assert "重新加载" in response.json()["detail"] + + +def test_update_rejects_duplicate_nodes_before_service(monkeypatch): + module = _load_module(monkeypatch) + response = _client(module).put( + "/api/v1/sensor-placement-schemes/7", + params={"network": "tjwater"}, + json={ + "expected_sensor_location": ["J1", "J2"], + "sensor_location": ["J1", "J1"], + }, + ) + + assert response.status_code == 422 + + +def test_export_returns_xlsx_download(monkeypatch): + module = _load_module(monkeypatch) + response = _client(module).post( + "/api/v1/sensor-placement-schemes/7/exports/excel", + params={"network": "tjwater"}, + json={ + "sensor_location": ["J1", "J2"], + "adjustment_status": {"J1": "original", "J2": "replaced"}, + }, + ) + + assert response.status_code == 200 + assert response.content == b"xlsx" + assert response.headers["content-type"].startswith( + "application/vnd.openxmlformats-officedocument" + ) + assert "filename*=UTF-8" in response.headers["content-disposition"] diff --git a/tests/unit/test_sensor_placement_service.py b/tests/unit/test_sensor_placement_service.py new file mode 100644 index 0000000..b4131aa --- /dev/null +++ b/tests/unit/test_sensor_placement_service.py @@ -0,0 +1,188 @@ +from datetime import datetime, timezone +from unittest.mock import MagicMock + +import pytest +from openpyxl import load_workbook + +from app.native.wndb import s42_sensor_placement +from app.services import sensor_placement + + +def _mock_project_cursor(monkeypatch): + cursor = MagicMock() + connection = MagicMock() + connection.cursor.return_value.__enter__.return_value = cursor + connection_context = MagicMock() + connection_context.__enter__.return_value = connection + monkeypatch.setattr( + s42_sensor_placement, + "project_connection", + lambda _network: connection_context, + ) + return cursor + + +def test_build_workbook_contains_engineering_columns(monkeypatch): + monkeypatch.setattr( + sensor_placement, + "_sensor_points", + lambda network, locations: [ + { + "node_id": "J1", + "project_x": 13500000.0, + "project_y": 3600000.0, + "map_x": 13500000.0, + "map_y": 3600000.0, + "longitude": 121.0, + "latitude": 31.0, + "elevation": 4.5, + } + ], + ) + scheme = { + "id": 7, + "scheme_name": "北区测压点", + "sensor_number": 2, + "min_diameter": 300, + "username": "alice", + "create_time": datetime(2026, 7, 30, tzinfo=timezone.utc), + "sensor_location": ["J1", "J2"], + } + + output = sensor_placement.build_sensor_placement_workbook( + network="tjwater", + scheme=scheme, + sensor_location=["J1"], + adjustment_status={"J1": "replaced"}, + ) + workbook = load_workbook(output) + + assert workbook.sheetnames == ["方案信息", "监测点清单"] + headers = [cell.value for cell in workbook["监测点清单"][1]] + assert headers == [ + "序号", + "节点 ID", + "经度", + "纬度", + "工程 X", + "工程 Y", + "地图 X", + "地图 Y", + "高程", + "调整状态", + ] + assert workbook["监测点清单"]["J2"].value == "替换" + assert workbook["方案信息"]["B8"].value == "未保存草稿" + + +def test_sensor_points_keep_engineering_coordinates_and_transform_map_coordinates( + monkeypatch, +): + monkeypatch.setattr( + sensor_placement.wndb, + "get_sensor_placement_nodes", + lambda network, node_ids: [ + { + "node_id": "J1", + "project_x": 3038.94, + "project_y": -34446.59, + "map_x": 13525191.530279, + "map_y": 3622984.760237, + "elevation": 4.5, + } + ], + ) + + point = sensor_placement._sensor_points("tjwater", ["J1"])[0] + + assert point["project_x"] == 3038.94 + assert point["project_y"] == -34446.59 + assert point["longitude"] == pytest.approx(121.498863, abs=1e-6) + assert point["latitude"] == pytest.approx(30.924784, abs=1e-6) + + +def test_update_validates_nodes_before_write(monkeypatch): + monkeypatch.setattr( + sensor_placement.wndb, + "get_sensor_placement_nodes", + lambda network, node_ids: [], + ) + + try: + sensor_placement.update_sensor_placement_scheme( + "tjwater", + 7, + expected_sensor_location=["J1"], + sensor_location=["missing"], + ) + except sensor_placement.SensorPlacementValidationError as exc: + assert "missing" in str(exc) + else: + raise AssertionError("expected invalid node to be rejected") + + +def test_sensor_nodes_use_materialized_web_mercator_geometry(monkeypatch): + cursor = _mock_project_cursor(monkeypatch) + cursor.fetchall.return_value = [] + + s42_sensor_placement.get_sensor_placement_nodes("tjwater", ["J1"]) + + query = cursor.execute.call_args.args[0] + assert "geo_junctions_mat" in query + assert "ST_X(c.coord)" in query + assert "ST_Y(c.coord)" in query + assert "ST_X(gj.geom)" in query + assert "ST_Y(gj.geom)" in query + + +def test_workbook_escapes_formula_in_scheme_metadata(monkeypatch): + monkeypatch.setattr( + sensor_placement, + "_sensor_points", + lambda network, locations: [ + { + "node_id": "J1", + "project_x": 3038.94, + "project_y": -34446.59, + "map_x": 13525191.53, + "map_y": 3622984.76, + "longitude": 121.49, + "latitude": 30.92, + "elevation": 4.5, + } + ], + ) + output = sensor_placement.build_sensor_placement_workbook( + network="tjwater", + scheme={ + "scheme_name": "=1+1", + "sensor_location": ["J1"], + "min_diameter": 300, + "username": "alice", + "create_time": datetime(2026, 7, 30, tzinfo=timezone.utc), + }, + sensor_location=["J1"], + adjustment_status={}, + ) + + workbook = load_workbook(output, data_only=False) + assert workbook["方案信息"]["B2"].value == "'=1+1" + assert workbook["方案信息"]["B2"].data_type == "s" + + +def test_create_sensor_placement_returns_inserted_record(monkeypatch): + cursor = _mock_project_cursor(monkeypatch) + cursor.fetchone.return_value = {"id": 7, "sensor_location": ["J1", "J2"]} + + created = s42_sensor_placement.create_sensor_placement( + "tjwater", + scheme_name="北区测压点", + min_diameter=300, + username="alice", + sensor_location=["J1", "J2"], + ) + + assert created["id"] == 7 + query, parameters = cursor.execute.call_args.args + assert "INSERT INTO sensor_placement" in query + assert parameters == ("北区测压点", 2, 300, "alice", ["J1", "J2"])