feat(burst-detection): update scada analysis flow

app/algorithms/__init__.py

@@ -1,27 +1,21 @@
-from app.algorithms.cleaning import flow_data_clean, pressure_data_clean
+"""Algorithm package with side-effect-free, lazy compatibility exports."""
-from app.algorithms.sensor import (
-    pressure_sensor_placement_sensitivity,
-    pressure_sensor_placement_kmeans,
-)
-from app.algorithms.isolation.valve import valve_isolation_analysis
-from app.algorithms.leakage import LeakageIdentifier
-from app.algorithms.health import PipelineHealthAnalyzer
-from app.algorithms.burst_location import run_burst_location
-from app.algorithms.simulation.scenarios import (
-    convert_to_local_unit,
-    burst_analysis,
-    valve_close_analysis,
-    flushing_analysis,
-    contaminant_simulation,
-    age_analysis,
-    pressure_regulation,
-)
 
-__all__ = [
+from importlib import import_module
-    "flow_data_clean",
+from typing import Any
-    "pressure_data_clean",
+
-    "pressure_sensor_placement_sensitivity",
+
-    "pressure_sensor_placement_kmeans",
+_EXPORT_MODULES = {
+    "flow_data_clean": "app.algorithms.cleaning",
+    "pressure_data_clean": "app.algorithms.cleaning",
+    "pressure_sensor_placement_sensitivity": "app.algorithms.sensor",
+    "pressure_sensor_placement_kmeans": "app.algorithms.sensor",
+    "valve_isolation_analysis": "app.algorithms.isolation.valve",
+    "LeakageIdentifier": "app.algorithms.leakage",
+    "PipelineHealthAnalyzer": "app.algorithms.health",
+    "run_burst_location": "app.algorithms.burst_location",
+    **{
+        name: "app.algorithms.simulation.scenarios"
+        for name in (
             "convert_to_local_unit",
             "burst_analysis",
             "valve_close_analysis",
@@ -29,8 +23,23 @@ __all__ = [
             "contaminant_simulation",
             "age_analysis",
             "pressure_regulation",
-    "valve_isolation_analysis",
+        )
-    "LeakageIdentifier",
+    },
-    "PipelineHealthAnalyzer",
+}
-    "run_burst_location",
+
-]
+__all__ = list(_EXPORT_MODULES)
+
+
+def __getattr__(name: str) -> Any:
+    try:
+        module_name = _EXPORT_MODULES[name]
+    except KeyError as exc:
+        raise AttributeError(f"module {__name__!r} has no attribute {name!r}") from exc
+
+    value = getattr(import_module(module_name), name)
+    globals()[name] = value
+    return value
+
+
+def __dir__() -> list[str]:
+    return sorted({*globals(), *__all__})

app/algorithms/leakage/identifier.py

@@ -124,7 +124,9 @@ def _worker_evaluate(raw_ratios: np.ndarray) -> float:
 class LeakageIdentifier:
     FLOW_UNIT_TO_M3S = {
         "m3/s": 1.0,
+        "m³/s": 1.0,
         "m3/h": 1.0 / 3600.0,
+        "m³/h": 1.0 / 3600.0,
         "L/s": 1.0 / 1000.0,
         "L/min": 1.0 / 60000.0,
     }

app/algorithms/simulation/runner.py

@@ -29,6 +29,7 @@ import pytz
 import requests
 import time
 import app.services.project_info as project_info
+from app.services.time_api import parse_clock_duration_seconds
 
 url_path = 'http://10.101.15.16:9000/loong'  # 内网
 # url_path = 'http://183.64.62.100:9057/loong'  # 外网
@@ -551,21 +552,11 @@ def from_clock_to_seconds (clock: str)->int:
     return hr*3600+mnt*60+seconds
 
 def from_clock_to_seconds_2 (clock: str)->int:
-    str_format="%H:%M:%S"
+    return parse_clock_duration_seconds(clock)
-    dt=datetime.strptime(clock,str_format)
-    hr=dt.hour
-    mnt=dt.minute
-    seconds=dt.second
-    return hr*3600+mnt*60+seconds
 
 
 def from_clock_to_seconds_3 (clock: str)->int:
-    str_format = "%H:%M" # 更新时间格式以适应 "小时:分钟" 格式
+    return parse_clock_duration_seconds(clock)
-    dt = datetime.strptime(clock,str_format)
-    hr = dt.hour
-    mnt = dt.minute
-    seconds = dt.second
-    return hr * 3600 + mnt * 60
 
 
 ###convert datetimestring 

app/api/v1/endpoints/burst_detection.py

@@ -30,6 +30,16 @@ class BurstDetectionRequest(BaseModel):
     points_per_day: int = Field(1440, description="每天的数据点数")
     mu: int = Field(100, description="异常值检测的参数")
     iforest_params: dict[str, Any] | None = Field(None, description="隔离森林算法参数")
+    target_time: datetime | None = Field(
+        None,
+        description="目标侦测时刻；为空时自动使用最近一个完整的监测时刻",
+    )
+    sampling_interval_minutes: int | None = Field(
+        None,
+        ge=1,
+        le=1440,
+        description="采样间隔（分钟）；为空时根据压力 SCADA 传输频率自动推断",
+    )
     scada_start: datetime | None = Field(None, description="SCADA数据起始时间")
     scada_end: datetime | None = Field(None, description="SCADA数据结束时间")
     sensor_nodes: list[str] | None = Field(None, description="传感器节点列表")

app/api/v1/endpoints/project_data.py

@@ -1,10 +1,9 @@
 from fastapi import APIRouter, Depends, HTTPException, Path, Query
 from psycopg import AsyncConnection
 
-import app.native.wndb as wndb
+from app.infra.db.postgresql.scada import ScadaInfoRepository
 from app.infra.db.postgresql.scheme import SchemeRepository
 from app.auth.project_dependencies import get_project_pg_connection
-from app.services import project_info
 
 router = APIRouter()
 
@@ -26,9 +25,7 @@ async def get_scada_info_with_connection(
     返回项目中所有的SCADA设备信息
     """
     try:
-        _ = conn
+        scada_data = await ScadaInfoRepository.get_scadas(conn)
-        network_name = project_info.name
-        scada_data = wndb.get_all_scada_info(network_name) if network_name else []
         return {"success": True, "data": scada_data, "count": len(scada_data)}
     except Exception as e:
         raise HTTPException(

app/api/v1/endpoints/simulation.py

@@ -35,7 +35,11 @@ from app.services.simulation_ops import (
     daily_scheduling_simulation,
 )
 from app.services.valve_isolation import analyze_valve_isolation
-from app.services.time_api import parse_aware_time, parse_utc_time
+from app.services.time_api import (
+    parse_aware_time,
+    parse_clock_duration_seconds,
+    parse_utc_time,
+)
 from pydantic import BaseModel, Field, field_validator
 
 router = APIRouter()
@@ -118,6 +122,14 @@ def run_simulation_manually_by_date(
     network_name: str, start_time: datetime, duration: int
 ) -> None:
     end_datetime = start_time + timedelta(minutes=duration)
+    time_properties = simulation.get_time(network_name)
+    hydraulic_step_seconds = parse_clock_duration_seconds(
+        time_properties["HYDRAULIC TIMESTEP"],
+        field_name="HYDRAULIC TIMESTEP",
+    )
+    if hydraulic_step_seconds <= 0:
+        raise ValueError("HYDRAULIC TIMESTEP must be greater than 0.")
+    hydraulic_step = timedelta(seconds=hydraulic_step_seconds)
     current_time = start_time
     while current_time < end_datetime:
         simulation.run_simulation(
@@ -125,7 +137,7 @@ def run_simulation_manually_by_date(
             simulation_type="realtime",
             modify_pattern_start_time=current_time.isoformat(timespec="seconds"),
         )
-        current_time += timedelta(minutes=15)
+        current_time += hydraulic_step
 
 
 # 必须用这个PlainTextResponse，不然每个key都有引号

app/infra/db/postgresql/scada.py

@@ -0,0 +1,51 @@
+from typing import Any
+
+from psycopg import AsyncConnection
+
+
+def _optional_text(value: Any) -> str | None:
+    return str(value).strip() if value is not None else None
+
+
+def _optional_float(value: Any) -> float | None:
+    return float(value) if value is not None else None
+
+
+class ScadaInfoRepository:
+    """Read SCADA metadata from the current project's business database."""
+
+    @staticmethod
+    async def get_scadas(conn: AsyncConnection) -> list[dict[str, Any]]:
+        async with conn.cursor() as cur:
+            await cur.execute(
+                """
+                SELECT id,
+                       type,
+                       associated_element_id,
+                       api_query_id,
+                       transmission_mode,
+                       transmission_frequency,
+                       reliability,
+                       x_coor,
+                       y_coor
+                FROM public.scada_info
+                """
+            )
+            records = await cur.fetchall()
+
+        return [
+            {
+                "id": str(record["id"]).strip(),
+                "type": str(record["type"]).strip().lower(),
+                "associated_element_id": _optional_text(
+                    record["associated_element_id"]
+                ),
+                "api_query_id": record["api_query_id"],
+                "transmission_mode": record["transmission_mode"],
+                "transmission_frequency": record["transmission_frequency"],
+                "reliability": _optional_float(record["reliability"]),
+                "x": _optional_float(record["x_coor"]),
+                "y": _optional_float(record["y_coor"]),
+            }
+            for record in records
+        ]

app/infra/db/timescaledb/composite_queries.py

@@ -1,18 +1,19 @@
 import time
-from typing import List, Optional, Any, Dict, Tuple
 from datetime import datetime, timedelta
-from psycopg import AsyncConnection
+from typing import Any, Dict, List, Optional, Tuple
-import pandas as pd
+
 import numpy as np
+import pandas as pd
+from psycopg import AsyncConnection
+
+import app.native.wndb as wndb
 from app.algorithms.cleaning.flow import clean_flow_data_df_kf
 from app.algorithms.cleaning.pressure import clean_pressure_data_df_km
 from app.algorithms.health.analyzer import PipelineHealthAnalyzer
-import app.native.wndb as wndb
+from app.infra.db.postgresql.scada import ScadaInfoRepository
-
 from app.infra.db.timescaledb.repositories.realtime import RealtimeRepository
 from app.infra.db.timescaledb.repositories.scheme import SchemeRepository
 from app.infra.db.timescaledb.repositories.scada import ScadaRepository
-from app.services import project_info
 
 
 class CompositeQueries:
@@ -20,6 +21,13 @@ class CompositeQueries:
     复合查询类，提供跨表查询功能
     """
 
+    @staticmethod
+    async def _get_project_scada_index(
+        postgres_conn: AsyncConnection,
+    ) -> Dict[str, Dict[str, Any]]:
+        scadas = await ScadaInfoRepository.get_scadas(postgres_conn)
+        return {scada["id"]: scada for scada in scadas}
+
     @staticmethod
     async def get_scada_associated_realtime_simulation_data(
         timescale_conn: AsyncConnection,
@@ -48,31 +56,22 @@ class CompositeQueries:
             ValueError: 当 SCADA 设备未找到或字段无效时
         """
         result = {}
-        # 1. 查询所有 SCADA 信息
+        scada_by_id = await CompositeQueries._get_project_scada_index(postgres_conn)
-        network_name = project_info.name
-        scada_infos = wndb.get_all_scada_info(network_name) if network_name else []
 
         for device_id in device_ids:
-            # 2. 根据 device_id 找到对应的 SCADA 信息
+            target_scada = scada_by_id.get(device_id)
-            target_scada = None
-            for scada in scada_infos:
-                if scada["id"] == device_id:
-                    target_scada = scada
-                    break
-
             if not target_scada:
                 raise ValueError(f"SCADA device {device_id} not found")
 
-            # 3. 根据 type 和 associated_element_id 查询对应的模拟数据
             element_id = target_scada["associated_element_id"]
             scada_type = target_scada["type"]
 
-            if scada_type.lower() == "pipe_flow":
+            if scada_type == "pipe_flow":
                 # 查询 link 模拟数据
                 res = await RealtimeRepository.get_link_field_by_time_range(
                     timescale_conn, start_time, end_time, element_id, "flow"
                 )
-            elif scada_type.lower() == "pressure":
+            elif scada_type == "pressure":
                 # 查询 node 模拟数据
                 res = await RealtimeRepository.get_node_field_by_time_range(
                     timescale_conn, start_time, end_time, element_id, "pressure"
@@ -115,26 +114,17 @@ class CompositeQueries:
             ValueError: 当 SCADA 设备未找到或字段无效时
         """
         result = {}
-        # 1. 查询所有 SCADA 信息
+        scada_by_id = await CompositeQueries._get_project_scada_index(postgres_conn)
-        network_name = project_info.name
-        scada_infos = wndb.get_all_scada_info(network_name) if network_name else []
 
         for device_id in device_ids:
-            # 2. 根据 device_id 找到对应的 SCADA 信息
+            target_scada = scada_by_id.get(device_id)
-            target_scada = None
-            for scada in scada_infos:
-                if scada["id"] == device_id:
-                    target_scada = scada
-                    break
-
             if not target_scada:
                 raise ValueError(f"SCADA device {device_id} not found")
 
-            # 3. 根据 type 和 associated_element_id 查询对应的模拟数据
             element_id = target_scada["associated_element_id"]
             scada_type = target_scada["type"]
 
-            if scada_type.lower() == "pipe_flow":
+            if scada_type == "pipe_flow":
                 # 查询 link 模拟数据
                 res = await SchemeRepository.get_link_field_by_scheme_and_time_range(
                     timescale_conn,
@@ -145,7 +135,7 @@ class CompositeQueries:
                     element_id,
                     "flow",
                 )
-            elif scada_type.lower() == "pressure":
+            elif scada_type == "pressure":
                 # 查询 node 模拟数据
                 res = await SchemeRepository.get_node_field_by_scheme_and_time_range(
                     timescale_conn,
@@ -167,19 +157,19 @@ class CompositeQueries:
     @staticmethod
     async def get_realtime_simulation_data(
         timescale_conn: AsyncConnection,
-        featureInfos: List[Tuple[str, str]],
+        feature_infos: List[Tuple[str, str]],
         start_time: datetime,
         end_time: datetime,
     ) -> Dict[str, List[Dict[str, Any]]]:
         """
         获取 link/node 模拟值
 
-        根据传入的 featureInfos，找到关联的 link/node，
+        根据传入的 feature_infos，找到关联的 link/node，
         并根据对应的 type，查询对应的模拟数据
 
         Args:
             timescale_conn: TimescaleDB 异步连接
-            featureInfos: 传入的 feature 信息列表，包含 (element_id, type)
+            feature_infos: 传入的 feature 信息列表，包含 (element_id, type)
             start_time: 开始时间
             end_time: 结束时间
 
@@ -190,20 +180,20 @@ class CompositeQueries:
             ValueError: 当 SCADA 设备未找到或字段无效时
         """
         result = {}
-        for feature_id, type in featureInfos:
+        for feature_id, feature_type in feature_infos:
 
-            if type.lower() == "pipe":
+            if feature_type.lower() == "pipe":
                 # 查询 link 模拟数据
                 res = await RealtimeRepository.get_link_field_by_time_range(
                     timescale_conn, start_time, end_time, feature_id, "flow"
                 )
-            elif type.lower() == "junction":
+            elif feature_type.lower() == "junction":
                 # 查询 node 模拟数据
                 res = await RealtimeRepository.get_node_field_by_time_range(
                     timescale_conn, start_time, end_time, feature_id, "pressure"
                 )
             else:
-                raise ValueError(f"Unknown type: {type}")
+                raise ValueError(f"Unknown type: {feature_type}")
             # 添加 scada_id 到每个数据项
             for item in res:
                 item["feature_id"] = feature_id
@@ -213,7 +203,7 @@ class CompositeQueries:
     @staticmethod
     async def get_scheme_simulation_data(
         timescale_conn: AsyncConnection,
-        featureInfos: List[Tuple[str, str]],
+        feature_infos: List[Tuple[str, str]],
         start_time: datetime,
         end_time: datetime,
         scheme_type: str,
@@ -222,12 +212,12 @@ class CompositeQueries:
         """
         获取 link/node scheme 模拟值
 
-        根据传入的 featureInfos，找到关联的 link/node，
+        根据传入的 feature_infos，找到关联的 link/node，
         并根据对应的 type，查询对应的模拟数据
 
         Args:
             timescale_conn: TimescaleDB 异步连接
-            featureInfos: 传入的 feature 信息列表，包含 (element_id, type)
+            feature_infos: 传入的 feature 信息列表，包含 (element_id, type)
             start_time: 开始时间
             end_time: 结束时间
             scheme_type: 工况类型
@@ -240,8 +230,8 @@ class CompositeQueries:
             ValueError: 当类型无效时
         """
         result = {}
-        for feature_id, type in featureInfos:
+        for feature_id, feature_type in feature_infos:
-            if type.lower() == "pipe":
+            if feature_type.lower() == "pipe":
                 # 查询 link 模拟数据
                 res = await SchemeRepository.get_link_field_by_scheme_and_time_range(
                     timescale_conn,
@@ -252,7 +242,7 @@ class CompositeQueries:
                     feature_id,
                     "flow",
                 )
-            elif type.lower() == "junction":
+            elif feature_type.lower() == "junction":
                 # 查询 node 模拟数据
                 res = await SchemeRepository.get_node_field_by_scheme_and_time_range(
                     timescale_conn,
@@ -264,7 +254,7 @@ class CompositeQueries:
                     "pressure",
                 )
             else:
-                raise ValueError(f"Unknown type: {type}")
+                raise ValueError(f"Unknown type: {feature_type}")
             # 添加 feature_id 到每个数据项
             for item in res:
                 item["feature_id"] = feature_id
@@ -301,33 +291,27 @@ class CompositeQueries:
             ValueError: 当元素类型无效时
         """
 
-        # 1. 查询所有 SCADA 信息
+        scada_by_id = await CompositeQueries._get_project_scada_index(postgres_conn)
-        network_name = project_info.name
+        associated_scada = next(
-        scada_infos = wndb.get_all_scada_info(network_name) if network_name else []
+            (
-
+                scada
-        # 2. 根据 element_type 和 element_id 找到关联的 SCADA 设备
+                for scada in scada_by_id.values()
-        associated_scada = None
+                if scada["associated_element_id"] == element_id
-        for scada in scada_infos:
+            ),
-            if scada["associated_element_id"] == element_id:
+            None,
-                associated_scada = scada
+        )
-                break
 
         if not associated_scada:
-            # 没有找到关联的 SCADA 设备
             return None
 
-        # 3. 通过 SCADA device_id 获取监测数据
         device_id = associated_scada["id"]
 
-        # 根据 use_cleaned 参数选择字段
         data_field = "cleaned_value" if use_cleaned else "monitored_value"
 
-        # 保证 device_id 以列表形式传递
         res = await ScadaRepository.get_scada_field_by_id_time_range(
             timescale_conn, [device_id], start_time, end_time, data_field
         )
 
-        # 将 device_id 替换为 element_id 返回
         return {element_id: res.get(device_id, [])}
 
     @staticmethod
@@ -351,97 +335,111 @@ class CompositeQueries:
             end_time: 结束时间
 
         Returns:
-            "success" 或错误信息
+            "success"
+
+        Raises:
+            ValueError: 当前项目没有可清洗设备或指定时间范围内没有监测数据
         """
-        try:
+        scada_by_id = await CompositeQueries._get_project_scada_index(postgres_conn)
-            # 获取所有 SCADA 信息
+        supported_types = {"pressure", "pipe_flow", "flow"}
-            network_name = project_info.name
+
-            scada_infos = wndb.get_all_scada_info(network_name) if network_name else []
+        if device_ids:
-            # 将列表转换为字典，以 device_id 为键
+            device_ids = [str(device_id).strip() for device_id in device_ids]
-            scada_device_info_dict = {info["id"]: info for info in scada_infos}
+            missing_metadata_ids = [
+                device_id
+                for device_id in device_ids
+                if device_id not in scada_by_id
+            ]
+            if missing_metadata_ids:
+                raise ValueError(
+                    f"当前项目中有 {len(missing_metadata_ids)} 个 SCADA 设备缺少元数据"
+                )
+
+            unsupported_ids = [
+                device_id
+                for device_id in device_ids
+                if scada_by_id[device_id]["type"] not in supported_types
+            ]
+            if unsupported_ids:
+                raise ValueError(
+                    f"当前项目中有 {len(unsupported_ids)} 个 SCADA 设备类型不支持清洗"
+                )
+        else:
+            device_ids = [
+                device_id
+                for device_id, info in scada_by_id.items()
+                if info["type"] in supported_types
+            ]
 
-            # 如果 device_ids 为空，则处理所有 SCADA 设备
         if not device_ids:
-                device_ids = list(scada_device_info_dict.keys())
+            raise ValueError("当前项目没有可清洗的 SCADA 设备")
 
-            # 批量查询所有设备的数据
         data = await ScadaRepository.get_scada_field_by_id_time_range(
             timescale_conn, device_ids, start_time, end_time, "monitored_value"
         )
-
         if not data:
-                return "error: fetch none scada data"  # 没有数据，直接返回
+            raise ValueError("指定时间范围内没有 SCADA 监测数据")
 
-            # 将嵌套字典转换为 DataFrame，使用 time 作为索引
+        normalized_data = {
-            # data 格式: {device_id: [{"time": "...", "value": ...}, ...]}
+            str(device_id): records for device_id, records in data.items()
-            all_records = []
+        }
-            for device_id, records in data.items():
+        missing_data_ids = [
-                for record in records:
+            device_id for device_id in device_ids if not normalized_data.get(device_id)
-                    all_records.append(
+        ]
+        if missing_data_ids:
+            raise ValueError(
+                f"指定时间范围内有 {len(missing_data_ids)} 个 SCADA 设备没有监测数据"
+            )
+
+        all_records = [
             {
                 "time": record["time"],
                 "device_id": device_id,
                 "value": record["value"],
             }
-                    )
+            for device_id, records in normalized_data.items()
-
+            for record in records
+        ]
         if not all_records:
-                return "error: fetch none scada data"  # 没有数据，直接返回
+            raise ValueError("指定时间范围内没有 SCADA 监测数据")
 
-            # 创建 DataFrame 并透视，使 device_id 成为列
         df_long = pd.DataFrame(all_records)
         df = df_long.pivot(index="time", columns="device_id", values="value")
 
-            # 根据type分类设备
         pressure_ids = [
-                id
+            device_id
-                for id in df.columns
+            for device_id in df.columns
-                if scada_device_info_dict.get(id, {}).get("type") == "pressure"
+            if scada_by_id[device_id]["type"] == "pressure"
         ]
         flow_ids = [
-                id
+            device_id
-                for id in df.columns
+            for device_id in df.columns
-                if scada_device_info_dict.get(id, {}).get("type") == "pipe_flow"
+            if scada_by_id[device_id]["type"] in {"pipe_flow", "flow"}
         ]
 
-            # 处理pressure数据
+        updated_rows = 0
-            if pressure_ids:
+        for grouped_ids, cleaning_function in (
-                pressure_df = df[pressure_ids]
+            (pressure_ids, clean_pressure_data_df_km),
-                # 重置索引，将 time 变为普通列
+            (flow_ids, clean_flow_data_df_kf),
-                pressure_df = pressure_df.reset_index()
+        ):
-                # 调用清洗方法
+            if not grouped_ids:
-                cleaned_df = clean_pressure_data_df_km(pressure_df)
+                continue
-                # 将清洗后的数据写回数据库
-                for device_id in pressure_ids:
-                    if device_id in cleaned_df.columns:
-                        cleaned_values = cleaned_df[device_id].tolist()
-                        time_values = cleaned_df["time"].tolist()
-                        for i, time_str in enumerate(time_values):
-                            time_dt = datetime.fromisoformat(time_str)
-                            value = cleaned_values[i]
-                            await ScadaRepository.update_scada_field(
-                                timescale_conn,
-                                time_dt,
-                                device_id,
-                                "cleaned_value",
-                                value,
-                            )
 
-            # 处理flow数据
+            source_df = df[grouped_ids].reset_index()
-            if flow_ids:
+            cleaned_df = cleaning_function(source_df)
-                flow_df = df[flow_ids]
-                # 重置索引，将 time 变为普通列
-                flow_df = flow_df.reset_index()
-                # 调用清洗方法
-                cleaned_df = clean_flow_data_df_kf(flow_df)
-                # 将清洗后的数据写回数据库
-                for device_id in flow_ids:
-                    if device_id in cleaned_df.columns:
-                        cleaned_values = cleaned_df[device_id].tolist()
             time_values = cleaned_df["time"].tolist()
-                        for i, time_str in enumerate(time_values):
+
-                            time_dt = datetime.fromisoformat(time_str)
+            for device_id in grouped_ids:
-                            value = cleaned_values[i]
+                if device_id not in cleaned_df.columns:
+                    raise ValueError(f"设备 {device_id} 的清洗结果缺少数据列")
+
+                cleaned_values = cleaned_df[device_id].tolist()
+                for time_value, value in zip(time_values, cleaned_values):
+                    time_dt = (
+                        time_value
+                        if isinstance(time_value, datetime)
+                        else datetime.fromisoformat(str(time_value))
+                    )
                     await ScadaRepository.update_scada_field(
                         timescale_conn,
                         time_dt,
@@ -449,10 +447,12 @@ class CompositeQueries:
                         "cleaned_value",
                         value,
                     )
+                    updated_rows += 1
+
+        if updated_rows == 0:
+            raise ValueError("SCADA 数据清洗未产生任何数据库更新")
 
         return "success"
-        except Exception as e:
-            return f"error: {str(e)}"
 
     @staticmethod
     async def predict_pipeline_health(

app/infra/db/timescaledb/internal_queries.py

@@ -50,6 +50,7 @@ class InternalStorage:
         link_result_list: List[dict],
         result_start_time: str,
         num_periods: int = 1,
+        result_timestep_seconds: int | None = None,
         db_name: str = None,
         max_retries: int = 3,
     ):
@@ -70,6 +71,7 @@ class InternalStorage:
                         link_result_list,
                         result_start_time,
                         num_periods,
+                        result_timestep_seconds,
                     )
                 break  # 成功
             except Exception as e:
@@ -167,6 +169,38 @@ class InternalQueries:
                 else:
                     raise
 
+    @staticmethod
+    def query_latest_scada_time(
+        device_ids: List[str],
+        before_time: str | datetime | None = None,
+        db_name: str = None,
+        max_retries: int = 3,
+    ) -> datetime | None:
+        """Return the latest SCADA timestamp for the selected devices."""
+        before_dt = (
+            parse_utc_time(before_time, field_name="before_time")
+            if before_time is not None
+            else None
+        )
+        for attempt in range(max_retries):
+            try:
+                conn_string = (
+                    get_timescaledb_pgconn_string(db_name=db_name)
+                    if db_name
+                    else get_timescaledb_pgconn_string()
+                )
+                with psycopg.Connection.connect(conn_string) as conn:
+                    return ScadaRepository.get_latest_scada_time_sync(
+                        conn,
+                        device_ids,
+                        before_dt,
+                    )
+            except Exception:
+                if attempt < max_retries - 1:
+                    time.sleep(1)
+                else:
+                    raise
+
     @staticmethod
     def query_realtime_simulation_by_ids_timerange(
         element_ids: List[str],

app/infra/db/timescaledb/repositories/scada.py

@@ -54,6 +54,27 @@ class ScadaRepository:
             )
             return cur.fetchall()
 
+    @staticmethod
+    def get_latest_scada_time_sync(
+        conn: Connection,
+        device_ids: List[str],
+        before_time: datetime | None = None,
+    ) -> datetime | None:
+        with conn.cursor(row_factory=dict_row) as cur:
+            if before_time is None:
+                cur.execute(
+                    "SELECT max(time) AS time FROM scada.scada_data WHERE device_id = ANY(%s)",
+                    (device_ids,),
+                )
+            else:
+                cur.execute(
+                    "SELECT max(time) AS time FROM scada.scada_data "
+                    "WHERE device_id = ANY(%s) AND time <= %s",
+                    (device_ids, before_time),
+                )
+            row = cur.fetchone()
+            return row["time"] if row else None
+
     @staticmethod
     async def get_scada_field_by_id_time_range(
         conn: AsyncConnection,

app/infra/db/timescaledb/repositories/scheme.py

@@ -3,10 +3,24 @@ from datetime import datetime, timedelta
 from collections import defaultdict
 from psycopg import AsyncConnection, Connection, sql
 import app.services.globals as globals
-from app.services.time_api import parse_utc_time
+from app.services.time_api import parse_clock_duration_seconds, parse_utc_time
 
 
 class SchemeRepository:
+    @staticmethod
+    def _get_result_timestep(result_timestep_seconds: int | None) -> timedelta:
+        if result_timestep_seconds is not None:
+            if result_timestep_seconds <= 0:
+                raise ValueError("result_timestep_seconds must be greater than 0.")
+            return timedelta(seconds=result_timestep_seconds)
+
+        timestep_seconds = parse_clock_duration_seconds(
+            globals.hydraulic_timestep,
+            field_name="HYDRAULIC TIMESTEP",
+        )
+        if timestep_seconds <= 0:
+            raise ValueError("HYDRAULIC TIMESTEP must be greater than 0.")
+        return timedelta(seconds=timestep_seconds)
 
     # --- Link Simulation ---
 
@@ -452,6 +466,7 @@ class SchemeRepository:
         link_result_list: List[Dict[str, any]],
         result_start_time: str,
         num_periods: int = 1,
+        result_timestep_seconds: int | None = None,
     ):
         """
         Store scheme simulation results to TimescaleDB.
@@ -468,20 +483,16 @@ class SchemeRepository:
             result_start_time, field_name="result_start_time"
         )
 
-        timestep_parts = globals.hydraulic_timestep.split(":")
+        timestep = SchemeRepository._get_result_timestep(result_timestep_seconds)
-        timestep = timedelta(
-            hours=int(timestep_parts[0]),
-            minutes=int(timestep_parts[1]),
-            seconds=int(timestep_parts[2]),
-        )
 
         # Prepare node data for batch insert
         node_data = []
         for node_result in node_result_list:
             node_id = node_result.get("node")
-            for period_index in range(num_periods):
+            result_rows = node_result.get("result", [])
+            for period_index in range(min(num_periods, len(result_rows))):
                 current_time = simulation_time + (timestep * period_index)
-                data = node_result.get("result", [])[period_index]
+                data = result_rows[period_index]
                 node_data.append(
                     {
                         "time": current_time,
@@ -499,9 +510,10 @@ class SchemeRepository:
         link_data = []
         for link_result in link_result_list:
             link_id = link_result.get("link")
-            for period_index in range(num_periods):
+            result_rows = link_result.get("result", [])
+            for period_index in range(min(num_periods, len(result_rows))):
                 current_time = simulation_time + (timestep * period_index)
-                data = link_result.get("result", [])[period_index]
+                data = result_rows[period_index]
                 link_data.append(
                     {
                         "time": current_time,
@@ -535,6 +547,7 @@ class SchemeRepository:
         link_result_list: List[Dict[str, any]],
         result_start_time: str,
         num_periods: int = 1,
+        result_timestep_seconds: int | None = None,
     ):
         """
         Store scheme simulation results to TimescaleDB (sync version).
@@ -551,20 +564,16 @@ class SchemeRepository:
             result_start_time, field_name="result_start_time"
         )
 
-        timestep_parts = globals.hydraulic_timestep.split(":")
+        timestep = SchemeRepository._get_result_timestep(result_timestep_seconds)
-        timestep = timedelta(
-            hours=int(timestep_parts[0]),
-            minutes=int(timestep_parts[1]),
-            seconds=int(timestep_parts[2]),
-        )
 
         # Prepare node data for batch insert
         node_data = []
         for node_result in node_result_list:
             node_id = node_result.get("node")
-            for period_index in range(num_periods):
+            result_rows = node_result.get("result", [])
+            for period_index in range(min(num_periods, len(result_rows))):
                 current_time = simulation_time + (timestep * period_index)
-                data = node_result.get("result", [])[period_index]
+                data = result_rows[period_index]
                 node_data.append(
                     {
                         "time": current_time,
@@ -582,9 +591,10 @@ class SchemeRepository:
         link_data = []
         for link_result in link_result_list:
             link_id = link_result.get("link")
-            for period_index in range(num_periods):
+            result_rows = link_result.get("result", [])
+            for period_index in range(min(num_periods, len(result_rows))):
                 current_time = simulation_time + (timestep * period_index)
-                data = link_result.get("result", [])[period_index]
+                data = result_rows[period_index]
                 link_data.append(
                     {
                         "time": current_time,

app/native/wndb/connection.py

@@ -1,3 +1,66 @@
+from collections.abc import Iterator
+from contextlib import contextmanager
+from threading import RLock
+
 import psycopg as pg
 
-g_conn_dict : dict[str, pg.Connection] = {}
+from app.core.config import get_pgconn_string
+
+g_conn_dict: dict[str, pg.Connection] = {}
+_registry_lock = RLock()
+_project_locks: dict[str, RLock] = {}
+
+
+def _is_closed(connection: pg.Connection) -> bool:
+    return bool(getattr(connection, "closed", False))
+
+
+def _close_connection(connection: pg.Connection) -> None:
+    if not _is_closed(connection):
+        connection.close()
+
+
+def _get_project_lock(name: str) -> RLock:
+    with _registry_lock:
+        lock = _project_locks.get(name)
+        if lock is None:
+            lock = RLock()
+            _project_locks[name] = lock
+        return lock
+
+
+def open_connection(name: str) -> pg.Connection:
+    with _get_project_lock(name):
+        connection = g_conn_dict.get(name)
+        if connection is None or _is_closed(connection):
+            if connection is not None:
+                _close_connection(connection)
+            connection = pg.connect(
+                conninfo=get_pgconn_string(db_name=name), autocommit=True
+            )
+            g_conn_dict[name] = connection
+        return connection
+
+
+def is_connection_open(name: str) -> bool:
+    with _get_project_lock(name):
+        connection = g_conn_dict.get(name)
+        if connection is None:
+            return False
+        if _is_closed(connection):
+            del g_conn_dict[name]
+            return False
+        return True
+
+
+def close_connection(name: str) -> None:
+    with _get_project_lock(name):
+        connection = g_conn_dict.pop(name, None)
+        if connection is not None:
+            _close_connection(connection)
+
+
+@contextmanager
+def project_connection(name: str) -> Iterator[pg.Connection]:
+    with _get_project_lock(name):
+        yield open_connection(name)

app/native/wndb/database.py

@@ -1,6 +1,6 @@
 from typing import Any
 from psycopg.rows import dict_row, Row
-from .connection import g_conn_dict as conn
+from .connection import project_connection
 
 API_ADD = 'add'
 API_UPDATE = 'update'
@@ -83,7 +83,8 @@ class DbChangeSet:
 
 
 def read(name: str, sql: str) -> Row:
-    with conn[name].cursor(row_factory=dict_row) as cur:
+    with project_connection(name) as conn:
+        with conn.cursor(row_factory=dict_row) as cur:
             cur.execute(sql)
             row = cur.fetchone()
             if row == None:
@@ -92,19 +93,22 @@ def read(name: str, sql: str) -> Row:
 
 
 def read_all(name: str, sql: str) -> list[Row]:
-    with conn[name].cursor(row_factory=dict_row) as cur:
+    with project_connection(name) as conn:
+        with conn.cursor(row_factory=dict_row) as cur:
             cur.execute(sql)
             return cur.fetchall()
 
 
 def try_read(name: str, sql: str) -> Row | None:
-    with conn[name].cursor(row_factory=dict_row) as cur:
+    with project_connection(name) as conn:
+        with conn.cursor(row_factory=dict_row) as cur:
             cur.execute(sql)
             return cur.fetchone()
 
 
 def write(name: str, sql: str) -> None:
-    with conn[name].cursor() as cur:
+    with project_connection(name) as conn:
+        with conn.cursor() as cur:
             cur.execute(sql)
 
 

app/native/wndb/project.py

@@ -2,7 +2,11 @@ import os
 import psycopg as pg
 from psycopg import sql
 from psycopg.rows import dict_row
-from .connection import g_conn_dict as conn
+from .connection import (
+    close_connection,
+    is_connection_open,
+    open_connection,
+)
 from app.core.config import get_pgconn_string, get_pg_config, get_pg_password
 
 # no undo/redo
@@ -31,9 +35,7 @@ def have_project(name: str) -> bool:
 
 
 def copy_project(source: str, new: str) -> None:
-    if source in conn:
+    close_connection(source)
-        conn[source].close()
-        del conn[source]
 
     with pg.connect(
         conninfo=get_pgconn_string(db_name="postgres"), autocommit=True
@@ -176,17 +178,12 @@ def clean_project(excluded: list[str] = []) -> None:
 
 
 def open_project(name: str) -> None:
-    if name not in conn:
+    open_connection(name)
-        conn[name] = pg.connect(
-            conninfo=get_pgconn_string(db_name=name), autocommit=True
-        )
 
 
 def is_project_open(name: str) -> bool:
-    return name in conn
+    return is_connection_open(name)
 
 
 def close_project(name: str) -> None:
-    if name in conn:
+    close_connection(name)
-        conn[name].close()
-        del conn[name]

app/native/wndb/s0_base.py

@@ -1,5 +1,5 @@
 from psycopg.rows import dict_row, Row
-from .connection import g_conn_dict as conn
+from .connection import project_connection
 from .database import read
 from typing import Any
 
@@ -47,7 +47,8 @@ ELEMENT_TYPES : dict[str, int] = {
 }
 
 def _get_from(name: str, id: str, base_type: str) -> Row | None:
-    with conn[name].cursor(row_factory=dict_row) as cur:
+    with project_connection(name) as conn:
+        with conn.cursor(row_factory=dict_row) as cur:
             cur.execute(f"select * from {base_type} where id = '{id}'")
             return cur.fetchone()
 
@@ -125,7 +126,8 @@ def is_region(name: str, id: str) -> bool:
 
 def _get_all(name: str, base_type: str) -> list[str]:
     ids : list[str] = []
-    with conn[name].cursor(row_factory=dict_row) as cur:
+    with project_connection(name) as conn:
+        with conn.cursor(row_factory=dict_row) as cur:
             cur.execute(f"select id from {base_type} order by id")
             for record in cur:
                 ids.append(record['id'])
@@ -138,7 +140,8 @@ def get_nodes(name: str) -> list[str]:
 # DingZQ
 def _get_nodes_by_type(name: str, type: str) -> list[str]:
     ids : list[str] = []
-    with conn[name].cursor(row_factory=dict_row) as cur:
+    with project_connection(name) as conn:
+        with conn.cursor(row_factory=dict_row) as cur:
             cur.execute(f"select id from {_NODE} where type = '{type}' order by id")
             for record in cur:
                 ids.append(record['id'])
@@ -147,7 +150,8 @@ def _get_nodes_by_type(name: str, type: str) -> list[str]:
 # DingZQ
 def get_nodes_id_and_type(name: str) -> dict[str, str]:
     nodes_id_and_type: dict[str, str] = {}
-    with conn[name].cursor(row_factory=dict_row) as cur:
+    with project_connection(name) as conn:
+        with conn.cursor(row_factory=dict_row) as cur:
             cur.execute(f"select id, type from {_NODE} order by id")
             for record in cur:
                 nodes_id_and_type[record['id']] = record['type']
@@ -156,7 +160,8 @@ def get_nodes_id_and_type(name: str) -> dict[str, str]:
 # DingZQ 2024-12-31
 def get_major_nodes(name: str, diameter: int) -> list[str]:
     major_nodes_set = set()
-    with conn[name].cursor(row_factory=dict_row) as cur:
+    with project_connection(name) as conn:
+        with conn.cursor(row_factory=dict_row) as cur:
             cur.execute(f"select node1, node2 from pipes where diameter > {diameter}")
             for record in cur:
                 major_nodes_set.add(record['node1'])
@@ -183,7 +188,8 @@ def get_links(name: str) -> list[str]:
 # DingZQ
 def _get_links_by_type(name: str, type: str) -> list[str]:
     ids : list[str] = []
-    with conn[name].cursor(row_factory=dict_row) as cur:
+    with project_connection(name) as conn:
+        with conn.cursor(row_factory=dict_row) as cur:
             cur.execute(f"select id from {_LINK} where type = '{type}' order by id")
             for record in cur:
                 ids.append(record['id'])
@@ -192,7 +198,8 @@ def _get_links_by_type(name: str, type: str) -> list[str]:
 # DingZQ
 def get_links_id_and_type(name: str) -> dict[str, str]:
     links_id_and_type: dict[str, str] = {}
-    with conn[name].cursor(row_factory=dict_row) as cur:
+    with project_connection(name) as conn:
+        with conn.cursor(row_factory=dict_row) as cur:
             cur.execute(f"select id, type from {_LINK} order by id")
             for record in cur:
                 links_id_and_type[record['id']] = record['type']
@@ -202,7 +209,8 @@ def get_links_id_and_type(name: str) -> dict[str, str]:
 # 获取直径大于800的管道
 def get_major_pipes(name: str, diameter: int) -> list[str]:
     major_pipe_ids: list[str] = []
-    with conn[name].cursor(row_factory=dict_row) as cur:
+    with project_connection(name) as conn:
+        with conn.cursor(row_factory=dict_row) as cur:
             cur.execute(f"select id from pipes where diameter > {diameter} order by id")
             for record in cur:
                 major_pipe_ids.append(record['id'])
@@ -232,7 +240,8 @@ def get_regions(name: str) -> list[str]:
     return _get_all(name, _REGION)
 
 def get_node_links(name: str, id: str) -> list[str]:
-    with conn[name].cursor(row_factory=dict_row) as cur:
+    with project_connection(name) as conn:
+        with conn.cursor(row_factory=dict_row) as cur:
             links: list[str] = []
             for p in cur.execute(f"select id from pipes where node1 = '{id}' or node2 = '{id}'").fetchall():
                 links.append(p['id'])
@@ -259,4 +268,3 @@ def get_region_type(name: str, id: str)->str:
         return type
 
 
-

app/native/wndb/s24_coordinates.py

@@ -1,5 +1,7 @@
 from .database import *
+from .connection import project_connection
 from .s0_base import get_link_nodes
+from psycopg.rows import dict_row
 
 def sql_update_coord(node: str, x: float, y: float) -> str:
     coord = f"st_geomfromtext('point({x} {y})')"
@@ -49,7 +51,8 @@ def get_links_in_extent(name: str, x1: float, y1: float, x2: float, y2: float) -
     node_ids = set([s.split(':')[0] for s in get_nodes_in_extent(name, x1, y1, x2, y2)])
 
     all_link_ids = []
-    with conn[name].cursor(row_factory=dict_row) as cur:
+    with project_connection(name) as conn:
+        with conn.cursor(row_factory=dict_row) as cur:
             cur.execute(f"select id from pipes")
             for record in cur:
                 all_link_ids.append(record['id'])

app/native/wndb/s41_pipe_risk_probability.py

@@ -1,5 +1,7 @@
 from .database import *
+from .connection import project_connection
 from .s0_base import *
+from psycopg.rows import dict_row
 import json
 
 def get_pipe_risk_probability_now(name: str, pipe_id: str) -> dict[str, Any]:
@@ -28,7 +30,8 @@ def get_pipe_risk_probability(name: str, pipe_id: str) -> dict[str, Any]:
 
 def get_network_pipe_risk_probability_now(name: str) -> list[dict[str, Any]]:
     pipe_risk_probability_list = []
-    with conn[name].cursor(row_factory=dict_row) as cur:
+    with project_connection(name) as conn:
+        with conn.cursor(row_factory=dict_row) as cur:
             cur.execute(f"select * from pipe_risk_probability")
             for record in cur:
                 #pipe_risk_probability_list.append(record)
@@ -42,7 +45,8 @@ def get_network_pipe_risk_probability_now(name: str) -> list[dict[str, Any]]:
 
 def get_pipes_risk_probability(name: str, pipe_ids: list[str]) -> list[dict[str, Any]]:
     pipe_risk_probability_list = []
-    with conn[name].cursor(row_factory=dict_row) as cur:
+    with project_connection(name) as conn:
+        with conn.cursor(row_factory=dict_row) as cur:
             cur.execute(f"select * from pipe_risk_probability")
             for record in cur:
                 if record['pipeid'] in pipe_ids:
@@ -67,7 +71,8 @@ def get_pipe_risk_probability_geometries(name: str) -> dict[str, Any]:
     # key_endnode = '下游节点'
     key_geometry = 'geometry'
 
-    with conn[name].cursor(row_factory=dict_row) as cur:
+    with project_connection(name) as conn:
+        with conn.cursor(row_factory=dict_row) as cur:
             cur.execute(f"select *, ST_AsGeoJSON(geometry) AS {key_geometry} from gis_pipe")
 
             for record in cur:

app/services/burst_detection.py

@@ -1,8 +1,10 @@
 from __future__ import annotations
 
-from datetime import datetime
+from collections import Counter
+from datetime import datetime, timedelta
 from typing import Any
 
+import numpy as np
 import pandas as pd
 
 from app.algorithms.burst_detection.burst_detector import BurstDetector
@@ -17,6 +19,15 @@ from app.services.tjnetwork import get_all_scada_info
 from app.services.time_api import extract_date, parse_utc_time, utc_now
 
 
+TARGET_DAY_COUNT = 15
+DEFAULT_SAMPLE_INTERVAL_MINUTES = 15
+TARGET_MU = 1
+TARGET_N_ESTIMATORS = 50
+TARGET_RANDOM_STATE = 42
+TARGET_SCORE_THRESHOLD = -0.04
+MIN_COMPLETE_SENSORS = 5
+
+
 def run_burst_detection(
     *,
     network: str,
@@ -31,6 +42,8 @@ def run_burst_detection(
     points_per_day: int = 1440,
     mu: int = 100,
     iforest_params: dict[str, Any] | None = None,
+    target_time: datetime | str | None = None,
+    sampling_interval_minutes: int | None = None,
     scada_start: datetime | str | None = None,
     scada_end: datetime | str | None = None,
     sensor_nodes: list[str] | None = None,
@@ -42,7 +55,8 @@ def run_burst_detection(
     """
     运行爆管侦测服务入口。
 
-    调用方式二选一：
+    调用方式三选一：
+    - 不传数据时间窗，自动侦测最近完整时刻；可用 `target_time` 回放历史时刻
     - 直接传 `observed_pressure_data`
     - 或传 `scada_start/scada_end` 让后端自动查询 SCADA 压力数据
 
@@ -74,8 +88,65 @@ def run_burst_detection(
         else None
     )
     use_scada_source = scada_start is not None or scada_end is not None
+    use_target_mode = (
+        observed_pressure_data is None
+        and not use_scada_source
+        and data_source != "simulation"
+    ) or target_time is not None
 
-    if use_scada_source:
+    resolved_target_time: datetime | None = None
+    requested_target_time: datetime | None = None
+    excluded_sensors: list[dict[str, str]] = []
+    daily_times: list[datetime] | None = None
+    resolved_sampling_interval_minutes: int | None = None
+
+    if use_target_mode:
+        if observed_pressure_data is not None or use_scada_source:
+            raise ValueError(
+                "target_time 不能与 observed_pressure_data 或 scada_start/scada_end 同时使用。"
+            )
+        scada_sensor_nodes = (
+            selected_sensor_nodes
+            if selected_sensor_nodes is not None
+            else _get_pressure_sensor_nodes(network)
+        )
+        requested_target_time = (
+            _to_datetime(target_time) if target_time is not None else None
+        )
+        resolved_sampling_interval_minutes = _resolve_sampling_interval_minutes(
+            network=network,
+            sensor_nodes=scada_sensor_nodes,
+            requested_interval=sampling_interval_minutes,
+        )
+        target_points_per_day = 1440 // resolved_sampling_interval_minutes
+        (
+            observed_input,
+            resolved_target_time,
+            excluded_sensors,
+        ) = _build_target_pressure_from_scada(
+            network=network,
+            sensor_nodes=scada_sensor_nodes,
+            requested_target_time=requested_target_time,
+            sampling_interval_minutes=resolved_sampling_interval_minutes,
+            points_per_day=target_points_per_day,
+        )
+        selected_sensor_nodes = list(observed_input.columns)
+        observed_source = (
+            "latest_monitoring" if target_time is None else "historical_monitoring"
+        )
+        points_per_day = target_points_per_day
+        mu = TARGET_MU
+        iforest_params = {
+            "n_estimators": TARGET_N_ESTIMATORS,
+            "random_state": TARGET_RANDOM_STATE,
+            "contamination": "auto",
+        }
+        daily_times = [
+            resolved_target_time - timedelta(days=offset)
+            for offset in range(TARGET_DAY_COUNT - 1, -1, -1)
+        ]
+
+    elif use_scada_source:
         scada_sensor_nodes = (
             selected_sensor_nodes
             if selected_sensor_nodes is not None
@@ -121,7 +192,16 @@ def run_burst_detection(
         sensor_nodes=selected_sensor_nodes,
     )
     resolved_sensor_nodes = list(result_df.attrs.get("sensor_nodes", []))
-    rows = _serialize_result_rows(result_df)
+    rows = _serialize_result_rows(
+        result_df,
+        daily_times=daily_times,
+        target_only=use_target_mode,
+    )
+    summary = _build_detection_summary(
+        result_df,
+        daily_times=daily_times,
+        target_only=use_target_mode,
+    )
     payload: dict[str, Any] = {
         "network": network,
         "sensor_nodes": resolved_sensor_nodes,
@@ -130,7 +210,17 @@ def run_burst_detection(
         "points_per_day": int(result_df.attrs.get("points_per_day", points_per_day)),
         "day_count": int(result_df.attrs.get("day_count", len(result_df))),
         "rows": rows,
-        "summary": _build_detection_summary(result_df),
+        "summary": summary,
+        "algorithm_params": {
+            "mu": mu,
+            "points_per_day": points_per_day,
+            "iforest_params": detector.iforest_params,
+            **(
+                {"score_threshold": TARGET_SCORE_THRESHOLD}
+                if use_target_mode
+                else {}
+            ),
+        },
     }
     if data_source == "simulation":
         payload["data_source"] = "simulation"
@@ -141,7 +231,50 @@ def run_burst_detection(
     else:
         payload["data_source"] = "monitoring"
 
-    if use_scada_source:
+    if (
+        use_target_mode
+        and resolved_target_time is not None
+        and resolved_sampling_interval_minutes is not None
+    ):
+        sample_start = resolved_target_time - timedelta(
+            days=TARGET_DAY_COUNT,
+            minutes=-resolved_sampling_interval_minutes,
+        )
+        payload.update(
+            {
+                "requested_target_time": (
+                    requested_target_time.isoformat()
+                    if requested_target_time is not None
+                    else None
+                ),
+                "target_time": resolved_target_time.isoformat(),
+                "reference_window": {
+                    "start": (resolved_target_time - timedelta(days=14)).isoformat(),
+                    "end": (resolved_target_time - timedelta(days=1)).isoformat(),
+                    "day_count": 14,
+                },
+                "sampling_interval_minutes": resolved_sampling_interval_minutes,
+                "daily_scores": [
+                    {
+                        "timestamp": row["Timestamp"],
+                        "role": row["Role"],
+                        "score": row["Score"],
+                        "raw_prediction": row["Prediction"],
+                    }
+                    for row in rows
+                ],
+                "data_quality": {
+                    "included_sensors": resolved_sensor_nodes,
+                    "excluded_sensors": excluded_sensors,
+                    "minimum_required_sensors": MIN_COMPLETE_SENSORS,
+                },
+                "scada_window": {
+                    "start": sample_start.isoformat(),
+                    "end": resolved_target_time.isoformat(),
+                },
+            }
+        )
+    elif use_scada_source:
         payload["scada_window"] = {
             "start": _to_datetime(scada_start).isoformat(),
             "end": _to_datetime(scada_end).isoformat(),
@@ -294,22 +427,49 @@ def _store_burst_detection_scheme(
     )
 
 
-def _serialize_result_rows(result_df: pd.DataFrame) -> list[dict[str, Any]]:
+def _serialize_result_rows(
+    result_df: pd.DataFrame,
+    *,
+    daily_times: list[datetime] | None = None,
+    target_only: bool = False,
+) -> list[dict[str, Any]]:
     rows: list[dict[str, Any]] = []
-    for row in result_df.to_dict(orient="records"):
+    raw_rows = result_df.to_dict(orient="records")
+    for index, row in enumerate(raw_rows):
+        is_target = index == len(raw_rows) - 1
+        is_burst = bool(row["IsBurst"])
+        if target_only:
+            is_burst = is_target and float(row["Score"]) <= TARGET_SCORE_THRESHOLD
         rows.append(
             {
                 "Day": int(row["Day"]),
                 "Score": float(row["Score"]),
                 "Prediction": int(row["Prediction"]),
-                "IsBurst": bool(row["IsBurst"]),
+                "IsBurst": is_burst,
+                **(
+                    {
+                        "Timestamp": daily_times[index].isoformat(),
+                        "Role": "target" if is_target else "reference",
+                    }
+                    if daily_times is not None
+                    else {}
+                ),
             }
         )
     return rows
 
 
-def _build_detection_summary(result_df: pd.DataFrame) -> dict[str, Any]:
+def _build_detection_summary(
-    rows = _serialize_result_rows(result_df)
+    result_df: pd.DataFrame,
+    *,
+    daily_times: list[datetime] | None = None,
+    target_only: bool = False,
+) -> dict[str, Any]:
+    rows = _serialize_result_rows(
+        result_df,
+        daily_times=daily_times,
+        target_only=target_only,
+    )
     if not rows:
         raise ValueError("爆管侦测结果为空。")
 
@@ -318,7 +478,7 @@ def _build_detection_summary(result_df: pd.DataFrame) -> dict[str, Any]:
     latest_row = rows[-1]
     anomaly_days = [row["Day"] for row in rows if row["IsBurst"]]
 
-    return {
+    summary = {
         "burst_detected": bool(latest_row["IsBurst"]),
         "latest_day": latest_row,
         "most_anomalous_day": int(result_df.iloc[most_anomalous_index]["Day"]),
@@ -326,6 +486,18 @@ def _build_detection_summary(result_df: pd.DataFrame) -> dict[str, Any]:
         "anomaly_day_count": len(anomaly_days),
         "latest_sensor_rankings": _build_latest_sensor_rankings(result_df),
     }
+    if target_only:
+        target_score = float(latest_row["Score"])
+        summary.update(
+            {
+                "target_score": target_score,
+                "score_threshold": TARGET_SCORE_THRESHOLD,
+                "target_rank": int(result_df["Score"].rank(method="min").iloc[-1]),
+                "target_time": latest_row.get("Timestamp"),
+                "reference_day_count": TARGET_DAY_COUNT - 1,
+            }
+        )
+    return summary
 
 
 def _build_latest_sensor_rankings(result_df: pd.DataFrame) -> list[dict[str, Any]]:
@@ -334,18 +506,192 @@ def _build_latest_sensor_rankings(result_df: pd.DataFrame) -> list[dict[str, Any
     if feature_matrix is None or len(sensor_nodes) == 0:
         return []
 
-    latest_values = feature_matrix[-1]
+    latest_values = np.asarray(feature_matrix[-1], dtype=float)
+    history = np.asarray(feature_matrix[:-1], dtype=float)
+    history_means = history.mean(axis=0)
+    history_stds = history.std(axis=0)
+    safe_stds = np.where(history_stds > 1e-9, history_stds, 1e-9)
+    deviations = (latest_values - history_means) / safe_stds
     ranking = sorted(
-        zip(sensor_nodes, latest_values, strict=False),
+        zip(
-        key=lambda item: item[1],
+            sensor_nodes,
+            latest_values,
+            history_means,
+            history_stds,
+            deviations,
+            strict=False,
+        ),
+        key=lambda item: item[4],
     )
     return [
         {
             "sensor_node": sensor_id,
             "latest_high_frequency_value": float(value),
+            "historical_mean": float(history_mean),
+            "historical_std": float(history_std),
+            "standardized_deviation": float(deviation),
         }
-        for sensor_id, value in ranking[: min(10, len(ranking))]
+        for sensor_id, value, history_mean, history_std, deviation in ranking[
+            : min(10, len(ranking))
         ]
+    ]
+
+
+def _build_target_pressure_from_scada(
+    *,
+    network: str,
+    sensor_nodes: list[str],
+    requested_target_time: datetime | None,
+    sampling_interval_minutes: int,
+    points_per_day: int,
+) -> tuple[pd.DataFrame, datetime, list[dict[str, str]]]:
+    node_query_id = _get_pressure_sensor_mapping(network)
+    mapped_nodes = [node for node in sensor_nodes if node in node_query_id]
+    excluded_without_mapping = [
+        {"sensor_node": node, "reason": "missing_api_query_id"}
+        for node in sensor_nodes
+        if node not in node_query_id
+    ]
+    if len(mapped_nodes) < MIN_COMPLETE_SENSORS:
+        raise ValueError(
+            f"可查询的压力测点少于 {MIN_COMPLETE_SENSORS} 个，无法执行爆管侦测。"
+        )
+
+    query_ids = [node_query_id[node] for node in mapped_nodes]
+    candidate_before = requested_target_time
+    last_excluded: list[dict[str, str]] = excluded_without_mapping
+
+    for _ in range(4):
+        resolved_target = InternalQueries.query_latest_scada_time(
+            db_name=network,
+            device_ids=query_ids,
+            before_time=candidate_before,
+        )
+        if resolved_target is None:
+            break
+
+        sample_start = resolved_target - timedelta(
+            days=TARGET_DAY_COUNT,
+            minutes=-sampling_interval_minutes,
+        )
+        expected_index = pd.date_range(
+            start=sample_start,
+            end=resolved_target,
+            freq=f"{sampling_interval_minutes}min",
+        )
+        scada_data = InternalQueries.query_scada_by_ids_timerange(
+            db_name=network,
+            device_ids=query_ids,
+            start_time=sample_start,
+            end_time=resolved_target,
+        )
+
+        complete_columns: dict[str, pd.Series] = {}
+        excluded = list(excluded_without_mapping)
+        for node_id in mapped_nodes:
+            query_id = node_query_id[node_id]
+            records = scada_data.get(query_id, [])
+            if not records:
+                excluded.append({"sensor_node": node_id, "reason": "no_data"})
+                continue
+
+            record_frame = pd.DataFrame.from_records(records)
+            record_frame["time"] = pd.to_datetime(record_frame["time"], utc=True)
+            record_frame["value"] = pd.to_numeric(
+                record_frame["value"], errors="coerce"
+            )
+            series = (
+                record_frame.drop_duplicates(subset="time", keep="last")
+                .set_index("time")["value"]
+                .reindex(expected_index)
+            )
+            if len(series) != TARGET_DAY_COUNT * points_per_day:
+                excluded.append(
+                    {"sensor_node": node_id, "reason": "unexpected_sample_count"}
+                )
+                continue
+            if series.isna().any():
+                excluded.append(
+                    {"sensor_node": node_id, "reason": "missing_or_invalid_samples"}
+                )
+                continue
+            complete_columns[node_id] = series
+
+        if len(complete_columns) >= MIN_COMPLETE_SENSORS:
+            observation_df = pd.DataFrame(complete_columns, index=expected_index)
+            return observation_df, resolved_target, excluded
+
+        last_excluded = excluded
+        candidate_before = resolved_target - timedelta(microseconds=1)
+
+    excluded_preview = ", ".join(
+        item["sensor_node"] for item in last_excluded[:10]
+    )
+    raise ValueError(
+        f"最近数据中完整压力测点少于 {MIN_COMPLETE_SENSORS} 个；"
+        f"请检查 15 天数据完整性。排除测点: {excluded_preview or '无'}"
+    )
+
+
+def _resolve_sampling_interval_minutes(
+    *,
+    network: str,
+    sensor_nodes: list[str],
+    requested_interval: int | None,
+) -> int:
+    if requested_interval is not None:
+        interval = int(requested_interval)
+    else:
+        selected_nodes = set(sensor_nodes)
+        inferred_intervals = [
+            parsed
+            for item in get_all_scada_info(network)
+            if str(item.get("type", "")).lower() == "pressure"
+            and str(item.get("associated_element_id", "")) in selected_nodes
+            and (
+                parsed := _parse_sampling_interval_minutes(
+                    item.get("transmission_frequency")
+                )
+            )
+            is not None
+        ]
+        interval = (
+            Counter(inferred_intervals).most_common(1)[0][0]
+            if inferred_intervals
+            else DEFAULT_SAMPLE_INTERVAL_MINUTES
+        )
+
+    if interval <= 0 or 1440 % interval != 0:
+        raise ValueError("采样间隔必须是能整除 1440 分钟的正整数。")
+    return interval
+
+
+def _parse_sampling_interval_minutes(value: Any) -> int | None:
+    if value is None:
+        return None
+    if isinstance(value, (int, float)):
+        minutes = float(value)
+    else:
+        try:
+            minutes = pd.to_timedelta(str(value)).total_seconds() / 60
+        except (TypeError, ValueError):
+            return None
+    rounded = round(minutes)
+    if minutes <= 0 or abs(minutes - rounded) > 1e-6:
+        return None
+    return int(rounded)
+
+
+def _get_pressure_sensor_mapping(network: str) -> dict[str, str]:
+    node_query_id: dict[str, str] = {}
+    for item in get_all_scada_info(network):
+        if str(item.get("type", "")).lower() != "pressure":
+            continue
+        node_id = item.get("associated_element_id")
+        query_id = item.get("api_query_id")
+        if node_id and query_id is not None:
+            node_query_id[str(node_id)] = str(query_id)
+    return node_query_id
 
 
 def _get_pressure_sensor_nodes(network: str) -> list[str]:
@@ -377,19 +723,7 @@ def _build_observed_pressure_from_scada(
     if start_dt >= end_dt:
         raise ValueError("SCADA 时间窗非法：scada_start 必须早于 scada_end。")
 
-    node_query_id: dict[str, str] = {}
+    node_query_id = _get_pressure_sensor_mapping(network)
-    for item in get_all_scada_info(network):
-        if str(item.get("type", "")).lower() != "pressure":
-            continue
-        node_id = item.get("associated_element_id")
-        query_id = item.get("api_query_id")
-        if (
-            isinstance(node_id, str)
-            and node_id
-            and isinstance(query_id, str)
-            and query_id
-        ):
-            node_query_id[node_id] = query_id
 
     missing_nodes = [node_id for node_id in sensor_nodes if node_id not in node_query_id]
     if missing_nodes:

app/services/burst_location.py

@@ -11,6 +11,7 @@ from app.infra.db.timescaledb.internal_queries import InternalQueries
 from app.services.scheme_management import (
     query_burst_location_scheme_detail,
     query_burst_location_schemes,
+    query_scheme_list,
     scheme_name_exists,
     store_scheme_info,
 )
@@ -353,9 +354,15 @@ def run_burst_location_by_network(
                 }
             )
     if normalized_data_source == "simulation":
+        simulation_burst_ids = _get_simulation_scheme_burst_ids(
+            network=network,
+            scheme_name=simulation_scheme_name,
+            scheme_type=resolved_simulation_scheme_type,
+        )
         payload["simulation_scheme"] = {
             "name": simulation_scheme_name,
             "type": resolved_simulation_scheme_type,
+            "burst_ids": simulation_burst_ids,
         }
     if scheme_name:
         _store_burst_scheme(
@@ -464,6 +471,30 @@ def _validate_time_window(
     return start_dt, end_dt
 
 
+def _get_simulation_scheme_burst_ids(
+    *, network: str, scheme_name: str | None, scheme_type: str
+) -> list[str]:
+    if not scheme_name:
+        return []
+    rows = query_scheme_list(network) or []
+    for row in rows:
+        if len(row) < 7:
+            continue
+        if row[1] != scheme_name or row[2] != scheme_type:
+            continue
+        detail = row[6] if isinstance(row[6], dict) else {}
+        return _normalize_burst_ids(detail.get("burst_ID"))
+    return []
+
+
+def _normalize_burst_ids(value: Any) -> list[str]:
+    if value is None:
+        return []
+    if isinstance(value, (list, tuple, set)):
+        return _dedupe_ids([str(item) for item in value])
+    return _dedupe_ids([str(value)])
+
+
 def _align_observed_series_pair(
     *,
     ids: list[str],

app/services/simulation.py

@@ -34,7 +34,7 @@ import psycopg
 import logging
 import app.services.globals as globals
 import app.services.project_info as project_info
-from app.services.time_api import parse_beijing_time
+from app.services.time_api import parse_beijing_time, parse_clock_duration_seconds
 from app.core.config import get_pgconn_string
 from app.infra.db.timescaledb.internal_queries import (
     InternalQueries as TimescaleInternalQueries,
@@ -757,11 +757,13 @@ def run_simulation(
 
     # 获取水力模拟步长，如’0:15:00‘
     globals.hydraulic_timestep = dic_time["HYDRAULIC TIMESTEP"]
-    # 将时间字符串转换为 timedelta 对象
+    # 转换为分钟浮点数，兼容 EPANET 的 H:MM 和 H:MM:SS 写法
-    time_obj = datetime.strptime(globals.hydraulic_timestep, "%H:%M:%S")
+    globals.PATTERN_TIME_STEP = (
-    # 转换为分钟浮点数
+        parse_clock_duration_seconds(
-    globals.PATTERN_TIME_STEP = float(
+            globals.hydraulic_timestep,
-        time_obj.hour * 60 + time_obj.minute + time_obj.second / 60
+            field_name="HYDRAULIC TIMESTEP",
+        )
+        / 60
     )
     # 对输入的时间参数进行处理
     pattern_start_time = convert_time_format(modify_pattern_start_time)
@@ -1258,6 +1260,9 @@ def run_simulation(
             node_result, link_result, modify_pattern_start_time, db_name=db_name
         )
     elif simulation_type.upper() == "EXTENDED":
+        result_timestep_seconds = times_info.get("report_step")
+        if result_timestep_seconds is None:
+            raise RuntimeError("run_project output missing times.report_step")
         TimescaleInternalStorage.store_scheme_simulation(
             scheme_type,
             scheme_name,
@@ -1265,6 +1270,7 @@ def run_simulation(
             link_result,
             modify_pattern_start_time,
             num_periods_result,
+            result_timestep_seconds,
             db_name=db_name,
         )
     endtime = time.time()

app/services/time_api.py

@@ -89,6 +89,41 @@ def to_time_range(dt: datetime, delta: float) -> tuple[datetime, datetime]:
 
     return (start_time, end_time)
 
+
+def parse_clock_duration_seconds(clock: str, field_name: str = "duration") -> int:
+    """
+    Parse EPANET-style clock durations into seconds.
+
+    Accepted formats include H:MM, HH:MM, H:MM:SS, and HH:MM:SS.
+    """
+    if not isinstance(clock, str):
+        raise ValueError(f"{field_name} must be a string clock duration.")
+
+    parts = clock.strip().split(":")
+    if len(parts) not in (2, 3):
+        raise ValueError(
+            f"{field_name} must use H:MM or H:MM:SS format, got {clock!r}."
+        )
+
+    try:
+        values = [int(part) for part in parts]
+    except ValueError as exc:
+        raise ValueError(
+            f"{field_name} must contain numeric clock parts, got {clock!r}."
+        ) from exc
+
+    if any(value < 0 for value in values):
+        raise ValueError(f"{field_name} must not contain negative values.")
+
+    hours, minutes = values[0], values[1]
+    seconds = values[2] if len(values) == 3 else 0
+    if minutes >= 60 or seconds >= 60:
+        raise ValueError(
+            f"{field_name} minutes and seconds must be less than 60, got {clock!r}."
+        )
+
+    return hours * 3600 + minutes * 60 + seconds
+
 def parse_beijing_date_range(query_date: str) -> tuple[datetime, datetime]:
     '''
     将一个日期字符串，转换成 start/end 时间段，传进来的日期被认为是北京时间

tests/api/test_simulation_endpoints.py

@@ -19,11 +19,18 @@ def _load_simulation_module(monkeypatch):
             timezone.utc
         )
 
+    def parse_clock_duration_seconds(value, field_name="duration"):
+        parts = [int(part) for part in value.split(":")]
+        hours, minutes = parts[0], parts[1]
+        seconds = parts[2] if len(parts) == 3 else 0
+        return hours * 3600 + minutes * 60 + seconds
+
     install_stub(
         monkeypatch,
         "app.services.time_api",
         {
             "parse_aware_time": parse_aware_time,
+            "parse_clock_duration_seconds": parse_clock_duration_seconds,
             "parse_utc_time": parse_utc_time,
         },
     )
@@ -31,6 +38,7 @@ def _load_simulation_module(monkeypatch):
         monkeypatch,
         "app.services.simulation",
         {
+            "get_time": lambda name: {"HYDRAULIC TIMESTEP": "0:15:00"},
             "run_simulation": lambda **kwargs: None,
             "query_corresponding_element_id_and_query_id": lambda name: None,
             "query_corresponding_pattern_id_and_query_id": lambda name: None,
@@ -227,6 +235,33 @@ def test_run_simulation_manually_by_date_uses_utc_aware_timestamps(monkeypatch):
     ]
 
 
+def test_run_simulation_manually_by_date_uses_hydraulic_timestep(monkeypatch):
+    module = _load_simulation_module(monkeypatch)
+    captured_calls = []
+
+    monkeypatch.setattr(
+        module.simulation,
+        "get_time",
+        lambda name: {"HYDRAULIC TIMESTEP": "1:00"},
+    )
+    monkeypatch.setattr(
+        module.simulation,
+        "run_simulation",
+        lambda **kwargs: captured_calls.append(kwargs),
+    )
+
+    module.run_simulation_manually_by_date(
+        "demo",
+        datetime(2025, 1, 1, 16, 0, 0, tzinfo=timezone.utc),
+        120,
+    )
+
+    assert [call["modify_pattern_start_time"] for call in captured_calls] == [
+        "2025-01-01T16:00:00+00:00",
+        "2025-01-01T17:00:00+00:00",
+    ]
+
+
 def test_runsimulationmanuallybydate_endpoint_accepts_timezone_aware_start_time(monkeypatch):
     module = _load_simulation_module(monkeypatch)
     captured = {}

tests/conftest.py

@@ -56,7 +56,7 @@ def install_stub(monkeypatch, name: str, attrs: dict | None = None, package: boo
                 parent = types.ModuleType(parent_name)
                 parent.__path__ = []
                 monkeypatch.setitem(sys.modules, parent_name, parent)
-        setattr(parent, child_name, module)
+        monkeypatch.setattr(parent, child_name, module, raising=False)
 
     return module
 

tests/unit/test_burst_detection_service.py

@@ -0,0 +1,160 @@
+from __future__ import annotations
+
+from datetime import datetime, timedelta, timezone
+
+import numpy as np
+import pandas as pd
+
+from app.services import burst_detection
+
+
+TARGET = datetime(2026, 6, 20, 5, 30, tzinfo=timezone.utc)
+
+
+def _complete_records(*, target: datetime, offset: float = 0.0) -> list[dict]:
+    start = target - timedelta(days=15) + timedelta(minutes=15)
+    return [
+        {
+            "time": (start + timedelta(minutes=15 * index)).isoformat(),
+            "value": float(index % 96) + offset,
+        }
+        for index in range(15 * 96)
+    ]
+
+
+def test_build_target_pressure_aligns_timestamps_and_excludes_incomplete_sensor(
+    monkeypatch,
+):
+    nodes = [f"J{index}" for index in range(6)]
+    mapping = {node: f"D{index}" for index, node in enumerate(nodes)}
+    scada_data = {
+        query_id: _complete_records(target=TARGET, offset=float(index))
+        for index, query_id in enumerate(mapping.values())
+    }
+    scada_data["D5"] = scada_data["D5"][:-1]
+
+    monkeypatch.setattr(
+        burst_detection,
+        "_get_pressure_sensor_mapping",
+        lambda _network: mapping,
+    )
+    monkeypatch.setattr(
+        burst_detection.InternalQueries,
+        "query_latest_scada_time",
+        lambda **_kwargs: TARGET,
+    )
+    monkeypatch.setattr(
+        burst_detection.InternalQueries,
+        "query_scada_by_ids_timerange",
+        lambda **_kwargs: scada_data,
+    )
+
+    frame, resolved_target, excluded = (
+        burst_detection._build_target_pressure_from_scada(
+            network="test",
+            sensor_nodes=nodes,
+            requested_target_time=TARGET,
+            sampling_interval_minutes=15,
+            points_per_day=96,
+        )
+    )
+
+    assert resolved_target == TARGET
+    assert frame.shape == (1440, 5)
+    assert frame.index[-1].to_pydatetime() == TARGET
+    assert excluded == [
+        {"sensor_node": "J5", "reason": "missing_or_invalid_samples"}
+    ]
+
+
+def test_target_mode_uses_fixed_parameters_and_only_classifies_target(monkeypatch):
+    index = pd.date_range(
+        start=TARGET - timedelta(days=15) + timedelta(minutes=15),
+        end=TARGET,
+        freq="15min",
+    )
+    values = np.tile(np.arange(96, dtype=float), 15)
+    frame = pd.DataFrame(
+        {f"J{sensor}": values + sensor for sensor in range(5)},
+        index=index,
+    )
+    monkeypatch.setattr(
+        burst_detection,
+        "_get_pressure_sensor_nodes",
+        lambda _network: list(frame.columns),
+    )
+    monkeypatch.setattr(
+        burst_detection,
+        "_build_target_pressure_from_scada",
+        lambda **_kwargs: (frame, TARGET, []),
+    )
+
+    payload = burst_detection.run_burst_detection(
+        network="test",
+        username="tester",
+        sampling_interval_minutes=15,
+    )
+
+    assert payload["target_time"] == TARGET.isoformat()
+    assert payload["sample_count"] == 1440
+    assert payload["points_per_day"] == 96
+    assert payload["algorithm_params"]["mu"] == 1
+    assert payload["summary"]["score_threshold"] == -0.04
+    assert [row["Role"] for row in payload["rows"]].count("target") == 1
+    assert all(not row["IsBurst"] for row in payload["rows"][:-1])
+    assert payload["reference_window"] == {
+        "start": (TARGET - timedelta(days=14)).isoformat(),
+        "end": (TARGET - timedelta(days=1)).isoformat(),
+        "day_count": 14,
+    }
+
+
+def test_sampling_interval_uses_scada_frequency_and_can_be_overridden(monkeypatch):
+    monkeypatch.setattr(
+        burst_detection,
+        "get_all_scada_info",
+        lambda _network: [
+            {
+                "type": "pressure",
+                "associated_element_id": "J1",
+                "transmission_frequency": "0:15:00",
+            },
+            {
+                "type": "pressure",
+                "associated_element_id": "J2",
+                "transmission_frequency": "0:15:00",
+            },
+        ],
+    )
+
+    assert (
+        burst_detection._resolve_sampling_interval_minutes(
+            network="test",
+            sensor_nodes=["J1", "J2"],
+            requested_interval=None,
+        )
+        == 15
+    )
+    assert (
+        burst_detection._resolve_sampling_interval_minutes(
+            network="test",
+            sensor_nodes=["J1", "J2"],
+            requested_interval=30,
+        )
+        == 30
+    )
+
+
+def test_target_threshold_is_applied_only_to_latest_row():
+    result = pd.DataFrame(
+        {
+            "Day": [1, 2, 3],
+            "Score": [-0.3, -0.2, -0.04],
+            "Prediction": [-1, -1, 1],
+            "IsBurst": [True, True, False],
+        }
+    )
+
+    rows = burst_detection._serialize_result_rows(result, target_only=True)
+
+    assert [row["IsBurst"] for row in rows] == [False, False, True]

tests/unit/test_burst_location_service.py

@@ -12,12 +12,19 @@ def _load_burst_location_module():
         Path(__file__).resolve().parents[2] / "app" / "services" / "burst_location.py"
     )
 
+    missing = object()
+    previous_modules = {}
+
+    def install_module(name: str, module: types.ModuleType) -> None:
+        previous_modules.setdefault(name, sys.modules.get(name, missing))
+        sys.modules[name] = module
+
     def ensure_package(name: str) -> types.ModuleType:
         module = sys.modules.get(name)
         if module is None:
             module = types.ModuleType(name)
             module.__path__ = []
-            sys.modules[name] = module
+            install_module(name, module)
         return module
 
     for package_name in [
@@ -46,11 +53,11 @@ def _load_burst_location_module():
         )
     )
     time_api_module.utc_now = lambda: datetime.now(timezone.utc)
-    sys.modules["app.services.time_api"] = time_api_module
+    install_module("app.services.time_api", time_api_module)
 
     algorithms_module = types.ModuleType("app.algorithms.burst_location")
     algorithms_module.run_burst_location = lambda **kwargs: {}
-    sys.modules["app.algorithms.burst_location"] = algorithms_module
+    install_module("app.algorithms.burst_location", algorithms_module)
 
     internal_queries_module = types.ModuleType(
         "app.infra.db.timescaledb.internal_queries"
@@ -70,25 +77,35 @@ def _load_burst_location_module():
             return {}
 
     internal_queries_module.InternalQueries = DummyInternalQueries
-    sys.modules["app.infra.db.timescaledb.internal_queries"] = internal_queries_module
+    install_module(
+        "app.infra.db.timescaledb.internal_queries", internal_queries_module
+    )
 
     scheme_management_module = types.ModuleType("app.services.scheme_management")
     scheme_management_module.query_burst_location_scheme_detail = lambda *args, **kwargs: {}
     scheme_management_module.query_burst_location_schemes = lambda *args, **kwargs: []
+    scheme_management_module.query_scheme_list = lambda *args, **kwargs: []
     scheme_management_module.scheme_name_exists = lambda *args, **kwargs: False
     scheme_management_module.store_scheme_info = lambda *args, **kwargs: None
-    sys.modules["app.services.scheme_management"] = scheme_management_module
+    install_module("app.services.scheme_management", scheme_management_module)
 
     tjnetwork_module = types.ModuleType("app.services.tjnetwork")
     tjnetwork_module.dump_inp = lambda *args, **kwargs: None
     tjnetwork_module.get_all_scada_info = lambda *args, **kwargs: []
-    sys.modules["app.services.tjnetwork"] = tjnetwork_module
+    install_module("app.services.tjnetwork", tjnetwork_module)
 
     module_name = "tests_burst_location_under_test"
     spec = importlib.util.spec_from_file_location(module_name, module_path)
     module = importlib.util.module_from_spec(spec)
     assert spec and spec.loader
+    try:
         spec.loader.exec_module(module)
+    finally:
+        for name, previous in reversed(previous_modules.items()):
+            if previous is missing:
+                sys.modules.pop(name, None)
+            else:
+                sys.modules[name] = previous
     return module
 
 
@@ -177,6 +194,21 @@ def test_run_burst_location_uses_single_timerange_with_burst_source_split(monkey
         "query_realtime_simulation_by_ids_timerange",
         staticmethod(fake_realtime_query),
     )
+    monkeypatch.setattr(
+        module,
+        "query_scheme_list",
+        lambda name: [
+            (
+                1,
+                "BurstSchemeA",
+                "burst_analysis",
+                "testuser",
+                None,
+                None,
+                {"burst_ID": ["Pipe-009", "Pipe-010"]},
+            )
+        ],
+    )
 
     result = module.run_burst_location_by_network(
         network="tjwater",
@@ -194,6 +226,7 @@ def test_run_burst_location_uses_single_timerange_with_burst_source_split(monkey
     assert result["simulation_scheme"] == {
         "name": "BurstSchemeA",
         "type": "burst_analysis",
+        "burst_ids": ["Pipe-009", "Pipe-010"],
     }
     assert result["pressure_samples"] == {"burst": 4, "normal": 4}
     assert result["flow_samples"] == {"burst": 4, "normal": 4}

tests/unit/test_leakage_flow_units.py

@@ -0,0 +1,20 @@
+import pytest
+
+from app.algorithms.leakage.identifier import LeakageIdentifier
+
+
+@pytest.mark.parametrize(
+    ("unit", "expected"),
+    [
+        ("m3/s", 1.0),
+        ("m³/s", 1.0),
+        ("m3/h", 3600.0),
+        ("m³/h", 3600.0),
+    ],
+)
+def test_leakage_identifier_accepts_display_flow_units(unit, expected):
+    assert LeakageIdentifier._flow_from_m3s(1.0, unit) == expected
+
+
+def test_leakage_identifier_accepts_display_flow_units_for_input():
+    assert LeakageIdentifier._flow_to_m3s(3600.0, "m³/h") == 1.0

tests/unit/test_postgres_scada_repository.py

@@ -0,0 +1,62 @@
+import asyncio
+
+from app.infra.db.postgresql.scada import ScadaInfoRepository
+
+
+class _FakeCursor:
+    def __init__(self):
+        self.query = None
+
+    async def __aenter__(self):
+        return self
+
+    async def __aexit__(self, exc_type, exc, tb):
+        return False
+
+    async def execute(self, query):
+        self.query = query
+
+    async def fetchall(self):
+        return [
+            {
+                "id": " 25470001 ",
+                "type": " PRESSURE ",
+                "associated_element_id": " J1 ",
+                "api_query_id": "query-1",
+                "transmission_mode": "realtime",
+                "transmission_frequency": None,
+                "reliability": "0.95",
+                "x_coor": "117.1",
+                "y_coor": "32.9",
+            }
+        ]
+
+
+class _FakeConnection:
+    def __init__(self):
+        self.cursor_instance = _FakeCursor()
+
+    def cursor(self):
+        return self.cursor_instance
+
+
+def test_get_scadas_normalizes_id_and_type():
+    conn = _FakeConnection()
+
+    result = asyncio.run(ScadaInfoRepository.get_scadas(conn))
+
+    assert result == [
+        {
+            "id": "25470001",
+            "type": "pressure",
+            "associated_element_id": "J1",
+            "api_query_id": "query-1",
+            "transmission_mode": "realtime",
+            "transmission_frequency": None,
+            "reliability": 0.95,
+            "x": 117.1,
+            "y": 32.9,
+        }
+    ]
+    assert "associated_element_id" in conn.cursor_instance.query
+    assert "FROM public.scada_info" in conn.cursor_instance.query

tests/unit/test_pressure_cleaning.py

@@ -1,47 +1,25 @@
-import importlib.util
-import sys
-import types
 from pathlib import Path
 
 import numpy as np
 import pandas as pd
+import pytest
+
+from app.algorithms.cleaning import pressure as pressure_cleaning
 
 
-def _load_pressure_cleaning_module():
+DATA_DIR = Path(__file__).resolve().parents[3] / "data"
-    project_root = Path(__file__).resolve().parents[2]
+RAW_DATA_PATH = DATA_DIR / "node_simulation.csv"
-    utils_path = project_root / "app" / "algorithms" / "_utils.py"
+NOISY_DATA_PATH = DATA_DIR / "node_simulation_noisy.csv"
-    pressure_path = project_root / "app" / "algorithms" / "cleaning" / "pressure.py"
+REQUIRES_PRESSURE_SAMPLES = pytest.mark.skipif(
-
+    not RAW_DATA_PATH.exists() or not NOISY_DATA_PATH.exists(),
-    app_module = sys.modules.setdefault("app", types.ModuleType("app"))
+    reason="pressure cleaning sample CSV files are not available",
-    algorithms_module = sys.modules.setdefault(
+)
-        "app.algorithms",
-        types.ModuleType("app.algorithms"),
-    )
-    setattr(app_module, "algorithms", algorithms_module)
-
-    utils_spec = importlib.util.spec_from_file_location("app.algorithms._utils", utils_path)
-    assert utils_spec and utils_spec.loader
-    utils_module = importlib.util.module_from_spec(utils_spec)
-    sys.modules["app.algorithms._utils"] = utils_module
-    utils_spec.loader.exec_module(utils_module)
-
-    pressure_spec = importlib.util.spec_from_file_location(
-        "tests_pressure_under_test",
-        pressure_path,
-    )
-    assert pressure_spec and pressure_spec.loader
-    pressure_module = importlib.util.module_from_spec(pressure_spec)
-    pressure_spec.loader.exec_module(pressure_module)
-    return pressure_module
-
 
+@REQUIRES_PRESSURE_SAMPLES
 def test_clean_pressure_data_df_km_repairs_long_form_pressure_series():
-    module = _load_pressure_cleaning_module()
+    raw_df = pd.read_csv(RAW_DATA_PATH)
-    repo_root = Path(__file__).resolve().parents[3]
+    noisy_df = pd.read_csv(NOISY_DATA_PATH)
-
+    cleaned_df = pressure_cleaning.clean_pressure_data_df_km(noisy_df)
-    raw_df = pd.read_csv(repo_root / "data" / "node_simulation.csv")
-    noisy_df = pd.read_csv(repo_root / "data" / "node_simulation_noisy.csv")
-    cleaned_df = module.clean_pressure_data_df_km(noisy_df)
 
     for df in (raw_df, noisy_df, cleaned_df):
         df["time"] = pd.to_datetime(df["time"])
@@ -50,7 +28,12 @@ def test_clean_pressure_data_df_km_repairs_long_form_pressure_series():
     assert set(cleaned_df.columns) == {"time", "id", "pressure"}
     assert cleaned_df["pressure"].isna().sum() == 0
 
-    noisy_joined = raw_df.merge(noisy_df, on=["time", "id"], how="inner", suffixes=("_raw", "_noisy"))
+    noisy_joined = raw_df.merge(
+        noisy_df,
+        on=["time", "id"],
+        how="inner",
+        suffixes=("_raw", "_noisy"),
+    )
     cleaned_joined = raw_df.merge(
         cleaned_df,
         on=["time", "id"],
@@ -59,10 +42,20 @@ def test_clean_pressure_data_df_km_repairs_long_form_pressure_series():
     )
 
     noisy_rmse = float(
-        np.sqrt(np.mean((noisy_joined["pressure_raw"] - noisy_joined["pressure_noisy"]) ** 2))
+        np.sqrt(
+            np.mean(
+                (noisy_joined["pressure_raw"] - noisy_joined["pressure_noisy"])
+                ** 2
+            )
+        )
     )
     cleaned_rmse = float(
-        np.sqrt(np.mean((cleaned_joined["pressure_raw"] - cleaned_joined["pressure_clean"]) ** 2))
+        np.sqrt(
+            np.mean(
+                (cleaned_joined["pressure_raw"] - cleaned_joined["pressure_clean"])
+                ** 2
+            )
+        )
     )
     noisy_mae = float(
         np.mean(np.abs(noisy_joined["pressure_raw"] - noisy_joined["pressure_noisy"]))
@@ -88,11 +81,9 @@ def test_clean_pressure_data_df_km_repairs_long_form_pressure_series():
     assert abs(spike_row - 28.018701553344727) < 2.0
 
 
+@REQUIRES_PRESSURE_SAMPLES
 def test_clean_pressure_data_df_km_accepts_single_sensor_wide_frame_with_utc_strings():
-    module = _load_pressure_cleaning_module()
+    noisy_df = pd.read_csv(NOISY_DATA_PATH)
-    repo_root = Path(__file__).resolve().parents[3]
-
-    noisy_df = pd.read_csv(repo_root / "data" / "node_simulation_noisy.csv")
     single_sensor = (
         noisy_df[noisy_df["id"] == 170490][["time", "pressure"]]
         .rename(columns={"pressure": "170490"})
@@ -102,7 +93,7 @@ def test_clean_pressure_data_df_km_accepts_single_sensor_wide_frame_with_utc_str
         pd.to_datetime(single_sensor["time"], utc=True).dt.strftime("%Y-%m-%dT%H:%M:%SZ")
     )
 
-    cleaned_df = module.clean_pressure_data_df_km(single_sensor)
+    cleaned_df = pressure_cleaning.clean_pressure_data_df_km(single_sensor)
 
     assert len(cleaned_df) == 192
     assert cleaned_df["170490"].isna().sum() == 0

tests/unit/test_project_scada_metadata.py

@@ -0,0 +1,121 @@
+import asyncio
+from datetime import datetime, timezone
+from unittest.mock import AsyncMock
+
+from app.api.v1.endpoints import project_data
+from app.infra.db.timescaledb import composite_queries
+
+
+PROJECT_SCADA = {
+    "id": "fengyang-pressure-1",
+    "type": "pressure",
+    "associated_element_id": "J1",
+    "api_query_id": "query-1",
+    "transmission_mode": "realtime",
+    "transmission_frequency": None,
+    "reliability": 1.0,
+    "x": 117.1,
+    "y": 32.9,
+}
+START_TIME = datetime(2026, 6, 1, tzinfo=timezone.utc)
+END_TIME = datetime(2026, 6, 2, tzinfo=timezone.utc)
+
+
+def _patch_project_scadas(monkeypatch):
+    monkeypatch.setattr(
+        composite_queries.ScadaInfoRepository,
+        "get_scadas",
+        AsyncMock(return_value=[PROJECT_SCADA.copy()]),
+    )
+
+
+def test_realtime_scada_simulation_uses_current_project_metadata(monkeypatch):
+    _patch_project_scadas(monkeypatch)
+    query_mock = AsyncMock(return_value=[{"time": START_TIME, "value": 26.5}])
+    monkeypatch.setattr(
+        composite_queries.RealtimeRepository,
+        "get_node_field_by_time_range",
+        query_mock,
+    )
+
+    result = asyncio.run(
+        composite_queries.CompositeQueries.get_scada_associated_realtime_simulation_data(
+            object(),
+            object(),
+            [PROJECT_SCADA["id"]],
+            START_TIME,
+            END_TIME,
+        )
+    )
+
+    assert result[PROJECT_SCADA["id"]][0]["scada_id"] == PROJECT_SCADA["id"]
+    assert query_mock.await_count == 1
+    assert query_mock.await_args.args[1:] == (
+        START_TIME,
+        END_TIME,
+        "J1",
+        "pressure",
+    )
+
+
+def test_scheme_scada_simulation_uses_current_project_metadata(monkeypatch):
+    _patch_project_scadas(monkeypatch)
+    query_mock = AsyncMock(return_value=[{"time": START_TIME, "value": 26.5}])
+    monkeypatch.setattr(
+        composite_queries.SchemeRepository,
+        "get_node_field_by_scheme_and_time_range",
+        query_mock,
+    )
+
+    result = asyncio.run(
+        composite_queries.CompositeQueries.get_scada_associated_scheme_simulation_data(
+            object(),
+            object(),
+            [PROJECT_SCADA["id"]],
+            START_TIME,
+            END_TIME,
+            "baseline",
+            "scheme-1",
+        )
+    )
+
+    assert result[PROJECT_SCADA["id"]][0]["scada_id"] == PROJECT_SCADA["id"]
+    assert query_mock.await_args.args[5:] == ("J1", "pressure")
+
+
+def test_element_scada_query_uses_current_project_metadata(monkeypatch):
+    _patch_project_scadas(monkeypatch)
+    query_mock = AsyncMock(
+        return_value={
+            PROJECT_SCADA["id"]: [{"time": START_TIME, "value": 26.5}]
+        }
+    )
+    monkeypatch.setattr(
+        composite_queries.ScadaRepository,
+        "get_scada_field_by_id_time_range",
+        query_mock,
+    )
+
+    result = asyncio.run(
+        composite_queries.CompositeQueries.get_element_associated_scada_data(
+            object(),
+            object(),
+            "J1",
+            START_TIME,
+            END_TIME,
+        )
+    )
+
+    assert result == {"J1": [{"time": START_TIME, "value": 26.5}]}
+
+
+def test_scada_info_endpoint_uses_current_project_connection(monkeypatch):
+    monkeypatch.setattr(
+        project_data.ScadaInfoRepository,
+        "get_scadas",
+        AsyncMock(return_value=[PROJECT_SCADA.copy()]),
+    )
+
+    result = asyncio.run(project_data.get_scada_info_with_connection(object()))
+
+    assert result == {"success": True, "data": [PROJECT_SCADA], "count": 1}

tests/unit/test_realtime_repository.py

@@ -1,48 +1,7 @@
 import asyncio
 from datetime import datetime, timezone
-import importlib.util
-from pathlib import Path
-import sys
-from types import ModuleType
 
-
+from app.infra.db.timescaledb.repositories.realtime import RealtimeRepository
-def _load_time_api_module():
-    module_path = (
-        Path(__file__).resolve().parents[2] / "app" / "services" / "time_api.py"
-    )
-    spec = importlib.util.spec_from_file_location("tests_time_api_under_test", module_path)
-    module = importlib.util.module_from_spec(spec)
-    assert spec and spec.loader
-    spec.loader.exec_module(module)
-    return module
-
-
-def _load_realtime_repository():
-    time_api_module = _load_time_api_module()
-    app_module = ModuleType("app")
-    services_module = ModuleType("app.services")
-    services_module.time_api = time_api_module
-    app_module.services = services_module
-    sys.modules["app"] = app_module
-    sys.modules["app.services"] = services_module
-    sys.modules["app.services.time_api"] = time_api_module
-
-    module_path = (
-        Path(__file__).resolve().parents[2]
-        / "app"
-        / "infra"
-        / "db"
-        / "timescaledb"
-        / "repositories"
-        / "realtime.py"
-    )
-    spec = importlib.util.spec_from_file_location(
-        "tests_realtime_repo_under_test", module_path
-    )
-    module = importlib.util.module_from_spec(spec)
-    assert spec and spec.loader
-    spec.loader.exec_module(module)
-    return module.RealtimeRepository
 
 
 class _FakeCursor:
@@ -71,7 +30,6 @@ class _FakeConnection:
 
 
 def test_get_links_by_time_range_normalizes_inputs_to_utc():
-    RealtimeRepository = _load_realtime_repository()
     conn = _FakeConnection()
 
     asyncio.run(
@@ -91,7 +49,6 @@ def test_get_links_by_time_range_normalizes_inputs_to_utc():
 
 
 def test_get_nodes_by_time_range_normalizes_inputs_to_utc():
-    RealtimeRepository = _load_realtime_repository()
     conn = _FakeConnection()
 
     asyncio.run(

tests/unit/test_scada_cleaning.py

@@ -0,0 +1,200 @@
+import asyncio
+from datetime import datetime, timezone
+from unittest.mock import AsyncMock
+
+import pandas as pd
+import pytest
+from fastapi import HTTPException
+
+from app.api.v1.endpoints.timeseries import composite as composite_endpoint
+from app.infra.db.timescaledb import composite_queries
+
+
+def test_clean_scada_uses_current_project_metadata(monkeypatch):
+    """Fengyang data must not be classified with the global tjwater metadata."""
+
+    monkeypatch.setattr(
+        composite_queries.ScadaInfoRepository,
+        "get_scadas",
+        AsyncMock(
+            return_value=[{"id": "fengyang-pressure-1", "type": "pressure"}]
+        ),
+    )
+    monkeypatch.setattr(
+        composite_queries.ScadaRepository,
+        "get_scada_field_by_id_time_range",
+        AsyncMock(
+            return_value={
+                "fengyang-pressure-1": [
+                    {"time": "2026-06-01T00:00:00+08:00", "value": 26.5}
+                ]
+            }
+        ),
+    )
+    update_mock = AsyncMock()
+    monkeypatch.setattr(
+        composite_queries.ScadaRepository,
+        "update_scada_field",
+        update_mock,
+    )
+    monkeypatch.setattr(
+        composite_queries,
+        "clean_pressure_data_df_km",
+        lambda frame: pd.DataFrame(
+            {
+                "time": frame["time"],
+                "fengyang-pressure-1": frame["fengyang-pressure-1"],
+            }
+        ),
+    )
+
+    result = asyncio.run(
+        composite_queries.CompositeQueries.clean_scada_data(
+            object(),
+            object(),
+            ["fengyang-pressure-1"],
+            datetime(2026, 6, 1, tzinfo=timezone.utc),
+            datetime(2026, 6, 2, tzinfo=timezone.utc),
+        )
+    )
+
+    assert result == "success"
+    update_mock.assert_awaited_once()
+
+
+def test_clean_scada_rejects_devices_missing_from_project_metadata(monkeypatch):
+    monkeypatch.setattr(
+        composite_queries.ScadaInfoRepository,
+        "get_scadas",
+        AsyncMock(return_value=[{"id": "other-device", "type": "pressure"}]),
+    )
+    query_mock = AsyncMock()
+    monkeypatch.setattr(
+        composite_queries.ScadaRepository,
+        "get_scada_field_by_id_time_range",
+        query_mock,
+    )
+
+    with pytest.raises(ValueError, match="缺少元数据"):
+        asyncio.run(
+            composite_queries.CompositeQueries.clean_scada_data(
+                object(),
+                object(),
+                ["fengyang-pressure-1"],
+                datetime(2026, 6, 1, tzinfo=timezone.utc),
+                datetime(2026, 6, 2, tzinfo=timezone.utc),
+            )
+        )
+
+    query_mock.assert_not_awaited()
+
+
+def test_clean_scada_rejects_zero_database_updates(monkeypatch):
+    monkeypatch.setattr(
+        composite_queries.ScadaInfoRepository,
+        "get_scadas",
+        AsyncMock(
+            return_value=[{"id": "fengyang-pressure-1", "type": "pressure"}]
+        ),
+    )
+    monkeypatch.setattr(
+        composite_queries.ScadaRepository,
+        "get_scada_field_by_id_time_range",
+        AsyncMock(
+            return_value={
+                "fengyang-pressure-1": [
+                    {"time": "2026-06-01T00:00:00+08:00", "value": 26.5}
+                ]
+            }
+        ),
+    )
+    update_mock = AsyncMock()
+    monkeypatch.setattr(
+        composite_queries.ScadaRepository,
+        "update_scada_field",
+        update_mock,
+    )
+    monkeypatch.setattr(
+        composite_queries,
+        "clean_pressure_data_df_km",
+        lambda _frame: pd.DataFrame(
+            {"time": [], "fengyang-pressure-1": []}
+        ),
+    )
+
+    with pytest.raises(ValueError, match="未产生任何数据库更新"):
+        asyncio.run(
+            composite_queries.CompositeQueries.clean_scada_data(
+                object(),
+                object(),
+                ["fengyang-pressure-1"],
+                datetime(2026, 6, 1, tzinfo=timezone.utc),
+                datetime(2026, 6, 2, tzinfo=timezone.utc),
+            )
+        )
+
+    update_mock.assert_not_awaited()
+
+
+def test_clean_scada_propagates_write_failures(monkeypatch):
+    monkeypatch.setattr(
+        composite_queries.ScadaInfoRepository,
+        "get_scadas",
+        AsyncMock(
+            return_value=[{"id": "fengyang-pressure-1", "type": "pressure"}]
+        ),
+    )
+    monkeypatch.setattr(
+        composite_queries.ScadaRepository,
+        "get_scada_field_by_id_time_range",
+        AsyncMock(
+            return_value={
+                "fengyang-pressure-1": [
+                    {"time": "2026-06-01T00:00:00+08:00", "value": 26.5}
+                ]
+            }
+        ),
+    )
+    monkeypatch.setattr(
+        composite_queries.ScadaRepository,
+        "update_scada_field",
+        AsyncMock(side_effect=RuntimeError("database write failed")),
+    )
+    monkeypatch.setattr(
+        composite_queries,
+        "clean_pressure_data_df_km",
+        lambda frame: frame,
+    )
+
+    with pytest.raises(RuntimeError, match="database write failed"):
+        asyncio.run(
+            composite_queries.CompositeQueries.clean_scada_data(
+                object(),
+                object(),
+                ["fengyang-pressure-1"],
+                datetime(2026, 6, 1, tzinfo=timezone.utc),
+                datetime(2026, 6, 2, tzinfo=timezone.utc),
+            )
+        )
+
+
+def test_clean_scada_endpoint_returns_http_400_for_validation_error(monkeypatch):
+    monkeypatch.setattr(
+        composite_endpoint.CompositeQueries,
+        "clean_scada_data",
+        AsyncMock(side_effect=ValueError("当前项目没有可清洗的 SCADA 设备")),
+    )
+
+    with pytest.raises(HTTPException) as exc_info:
+        asyncio.run(
+            composite_endpoint.clean_scada_data(
+                device_ids="all",
+                start_time=datetime(2026, 6, 1, tzinfo=timezone.utc),
+                end_time=datetime(2026, 6, 2, tzinfo=timezone.utc),
+                timescale_conn=object(),
+                postgres_conn=object(),
+            )
+        )
+
+    assert exc_info.value.status_code == 400
+    assert exc_info.value.detail == "当前项目没有可清洗的 SCADA 设备"

tests/unit/test_scheme_simulation_timestep.py

@@ -0,0 +1,154 @@
+import json
+from datetime import timedelta
+
+import pytest
+
+from app.infra.db.timescaledb.repositories.scheme import SchemeRepository
+from app.services.time_api import parse_utc_time
+
+
+def _node_result(periods: int) -> list[dict]:
+    return [
+        {
+            "node": "J1",
+            "result": [
+                {"demand": index, "head": index, "pressure": index, "quality": index}
+                for index in range(periods)
+            ],
+        }
+    ]
+
+
+def _link_result(periods: int) -> list[dict]:
+    return [
+        {
+            "link": "P1",
+            "result": [
+                {
+                    "flow": index,
+                    "friction": index,
+                    "headloss": index,
+                    "quality": index,
+                    "reaction": index,
+                    "setting": index,
+                    "status": index,
+                    "velocity": index,
+                }
+                for index in range(periods)
+            ],
+        }
+    ]
+
+
+def test_store_scheme_simulation_uses_15_minute_report_step(monkeypatch):
+    inserted: dict[str, list[dict]] = {}
+
+    monkeypatch.setattr(
+        SchemeRepository,
+        "insert_nodes_batch_sync",
+        staticmethod(lambda conn, data: inserted.setdefault("nodes", data)),
+    )
+    monkeypatch.setattr(
+        SchemeRepository,
+        "insert_links_batch_sync",
+        staticmethod(lambda conn, data: inserted.setdefault("links", data)),
+    )
+
+    SchemeRepository.store_scheme_simulation_result_sync(
+        conn=object(),
+        scheme_type="burst_analysis",
+        scheme_name="five_hour_case",
+        node_result_list=_node_result(21),
+        link_result_list=_link_result(21),
+        result_start_time="2026-07-16T00:00:00Z",
+        num_periods=21,
+        result_timestep_seconds=900,
+    )
+
+    start_time = parse_utc_time("2026-07-16T00:00:00Z")
+    assert len(inserted["nodes"]) == 21
+    assert inserted["nodes"][0]["time"] == start_time
+    assert inserted["nodes"][-1]["time"] == start_time + timedelta(hours=5)
+    assert inserted["links"][-1]["time"] == start_time + timedelta(hours=5)
+
+
+def test_store_scheme_simulation_uses_hourly_report_step(monkeypatch):
+    inserted: dict[str, list[dict]] = {}
+
+    monkeypatch.setattr(
+        SchemeRepository,
+        "insert_nodes_batch_sync",
+        staticmethod(lambda conn, data: inserted.setdefault("nodes", data)),
+    )
+    monkeypatch.setattr(
+        SchemeRepository,
+        "insert_links_batch_sync",
+        staticmethod(lambda conn, data: inserted.setdefault("links", data)),
+    )
+
+    SchemeRepository.store_scheme_simulation_result_sync(
+        conn=object(),
+        scheme_type="burst_analysis",
+        scheme_name="hourly_case",
+        node_result_list=_node_result(6),
+        link_result_list=_link_result(6),
+        result_start_time="2026-07-16T00:00:00Z",
+        num_periods=6,
+        result_timestep_seconds=3600,
+    )
+
+    start_time = parse_utc_time("2026-07-16T00:00:00Z")
+    assert [item["time"] for item in inserted["nodes"]] == [
+        start_time + timedelta(hours=index) for index in range(6)
+    ]
+
+
+def test_run_simulation_passes_report_step_for_extended_scheme(monkeypatch):
+    import app.services.simulation as simulation
+
+    time_updates: list[dict] = []
+    storage_calls: list[tuple] = []
+
+    monkeypatch.setattr(simulation, "open_project", lambda name: None)
+    monkeypatch.setattr(
+        simulation,
+        "get_time",
+        lambda name: {
+            "HYDRAULIC TIMESTEP": "00:15:00",
+            "REPORT TIMESTEP": "1:00",
+            "DURATION": "0:00",
+            "PATTERN START": "0:00",
+        },
+    )
+    monkeypatch.setattr(
+        simulation,
+        "set_time",
+        lambda name, changeset: time_updates.append(changeset.operations[0]),
+    )
+    monkeypatch.setattr(simulation, "run_project", lambda name: json.dumps({
+        "simulation_result": "successful",
+        "output": {
+            "times": {"num_periods": 21, "report_step": 900},
+            "node_results": _node_result(21),
+            "link_results": _link_result(21),
+        },
+    }))
+    monkeypatch.setattr(
+        simulation.TimescaleInternalStorage,
+        "store_scheme_simulation",
+        staticmethod(lambda *args, **kwargs: storage_calls.append((args, kwargs))),
+    )
+
+    simulation.run_simulation(
+        name="fengyang",
+        simulation_type="extended",
+        modify_pattern_start_time="2026-07-16T00:00:00+08:00",
+        modify_total_duration=18000,
+        scheme_type="burst_analysis",
+        scheme_name="five_hour_case",
+    )
+
+    assert time_updates[0]["DURATION"] == "05:00:00"
+    assert time_updates[0]["REPORT TIMESTEP"] == "1:00"
+    assert storage_calls[0][0][5] == 21
+    assert storage_calls[0][0][6] == 900

tests/unit/test_time_api.py

@@ -43,3 +43,29 @@ def test_utc_now_returns_timezone_aware_utc_datetime():
 
     assert result.tzinfo == timezone.utc
     assert result.utcoffset() == timedelta(0)
+
+
+@pytest.mark.parametrize(
+    ("clock", "expected_seconds"),
+    [
+        ("1:00", 3600),
+        ("01:00", 3600),
+        ("0:05", 300),
+        ("0:05:00", 300),
+        ("24:00", 86400),
+    ],
+)
+def test_parse_clock_duration_seconds_accepts_epanet_clock_formats(
+    clock, expected_seconds
+):
+    module = _load_time_api_module()
+
+    assert module.parse_clock_duration_seconds(clock) == expected_seconds
+
+
+@pytest.mark.parametrize("clock", ["bad", "1:60", "1:00:60", "-1:00"])
+def test_parse_clock_duration_seconds_rejects_invalid_clock_formats(clock):
+    module = _load_time_api_module()
+
+    with pytest.raises(ValueError):
+        module.parse_clock_duration_seconds(clock)

tests/unit/test_wndb_connection.py

@@ -0,0 +1,78 @@
+import pytest
+
+from app.native.wndb import connection
+from app.native.wndb import database
+from app.native.wndb import project
+
+
+class _FakeCursor:
+    def __init__(self, rows):
+        self.rows = rows
+        self.executed = []
+
+    def __enter__(self):
+        return self
+
+    def __exit__(self, exc_type, exc, tb):
+        return False
+
+    def execute(self, sql):
+        self.executed.append(sql)
+
+    def fetchall(self):
+        return self.rows
+
+
+class _FakeConnection:
+    def __init__(self, rows=None, *, closed=False):
+        self.rows = list(rows or [])
+        self.closed = closed
+        self.close_calls = 0
+
+    def cursor(self, row_factory=None):
+        if self.closed:
+            raise RuntimeError("the connection is closed")
+        return _FakeCursor(self.rows)
+
+    def close(self):
+        self.close_calls += 1
+        self.closed = True
+
+
+@pytest.fixture(autouse=True)
+def clear_native_connections():
+    connection.g_conn_dict.clear()
+    connection._project_locks.clear()
+    yield
+    connection.g_conn_dict.clear()
+    connection._project_locks.clear()
+
+
+def test_is_project_open_drops_closed_cached_connection():
+    connection.g_conn_dict["fengyang"] = _FakeConnection(closed=True)
+
+    assert project.is_project_open("fengyang") is False
+    assert "fengyang" not in connection.g_conn_dict
+
+
+def test_read_all_reopens_closed_cached_connection(monkeypatch):
+    stale = _FakeConnection(closed=True)
+    fresh = _FakeConnection(rows=[{"key": "DURATION", "value": "01:00:00"}])
+    connection.g_conn_dict["fengyang"] = stale
+
+    opened = []
+
+    def fake_connect(*, conninfo, autocommit):
+        opened.append((conninfo, autocommit))
+        return fresh
+
+    monkeypatch.setattr(connection.pg, "connect", fake_connect)
+    monkeypatch.setattr(
+        connection, "get_pgconn_string", lambda db_name: f"dbname={db_name}"
+    )
+
+    rows = database.read_all("fengyang", "select * from times")
+
+    assert rows == [{"key": "DURATION", "value": "01:00:00"}]
+    assert opened == [("dbname=fengyang", True)]
+    assert connection.g_conn_dict["fengyang"] is fresh