From 9f7f5536d7fad8a44449b66c77e5287df3f6e265 Mon Sep 17 00:00:00 2001 From: Huarch Date: Thu, 16 Jul 2026 14:16:06 +0800 Subject: [PATCH] fix(simulation): use hydraulic timestep --- app/algorithms/simulation/runner.py | 15 +- app/api/v1/endpoints/simulation.py | 16 +- app/services/simulation.py | 15 +- app/services/time_api.py | 35 +++ tests/api/test_simulation_endpoints.py | 395 +++++++++++++++++++++++++ tests/unit/test_time_api.py | 71 +++++ 6 files changed, 527 insertions(+), 20 deletions(-) create mode 100644 tests/api/test_simulation_endpoints.py create mode 100644 tests/unit/test_time_api.py diff --git a/app/algorithms/simulation/runner.py b/app/algorithms/simulation/runner.py index 004caba..b59fa55 100644 --- a/app/algorithms/simulation/runner.py +++ b/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" - dt=datetime.strptime(clock,str_format) - hr=dt.hour - mnt=dt.minute - seconds=dt.second - return hr*3600+mnt*60+seconds + return parse_clock_duration_seconds(clock) def from_clock_to_seconds_3 (clock: str)->int: - str_format = "%H:%M" # 更新时间格式以适应 "小时:分钟" 格式 - dt = datetime.strptime(clock,str_format) - hr = dt.hour - mnt = dt.minute - seconds = dt.second - return hr * 3600 + mnt * 60 + return parse_clock_duration_seconds(clock) ###convert datetimestring diff --git a/app/api/v1/endpoints/simulation.py b/app/api/v1/endpoints/simulation.py index c8ad1f4..3a758cd 100644 --- a/app/api/v1/endpoints/simulation.py +++ b/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都有引号 diff --git a/app/services/simulation.py b/app/services/simulation.py index 825976d..5e910e9 100644 --- a/app/services/simulation.py +++ b/app/services/simulation.py @@ -28,12 +28,13 @@ import pytz import requests import time from typing import Optional, Tuple +import app.infra.db.influxdb.api as influxdb_api import typing 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, @@ -756,11 +757,13 @@ def run_simulation( # 获取水力模拟步长，如’0:15:00‘ globals.hydraulic_timestep = dic_time["HYDRAULIC TIMESTEP"] - # 将时间字符串转换为 timedelta 对象 - time_obj = datetime.strptime(globals.hydraulic_timestep, "%H:%M:%S") - # 转换为分钟浮点数 - globals.PATTERN_TIME_STEP = float( - time_obj.hour * 60 + time_obj.minute + time_obj.second / 60 + # 转换为分钟浮点数，兼容 EPANET 的 H:MM 和 H:MM:SS 写法 + globals.PATTERN_TIME_STEP = ( + parse_clock_duration_seconds( + globals.hydraulic_timestep, + field_name="HYDRAULIC TIMESTEP", + ) + / 60 ) # 对输入的时间参数进行处理 pattern_start_time = convert_time_format(modify_pattern_start_time) diff --git a/app/services/time_api.py b/app/services/time_api.py index 00904e2..569626a 100644 --- a/app/services/time_api.py +++ b/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 时间段，传进来的日期被认为是北京时间 diff --git a/tests/api/test_simulation_endpoints.py b/tests/api/test_simulation_endpoints.py new file mode 100644 index 0000000..e022f92 --- /dev/null +++ b/tests/api/test_simulation_endpoints.py @@ -0,0 +1,395 @@ +from pathlib import Path +from datetime import datetime, timezone + +from fastapi.testclient import TestClient + +from tests.conftest import build_test_app, install_stub, load_module_from_path + + +def _load_simulation_module(monkeypatch): + install_stub(monkeypatch, "app.services", package=True) + def parse_aware_time(value, field_name="datetime"): + dt = datetime.fromisoformat(str(value).replace("Z", "+00:00")) + if dt.tzinfo is None: + raise ValueError(f"{field_name} is missing timezone information.") + return dt + + def parse_utc_time(value, field_name="datetime"): + return parse_aware_time(value, field_name=field_name).astimezone( + 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, + }, + ) + install_stub( + 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, + "query_non_realtime_region": lambda name: [], + "get_source_outflow_region_id": lambda name, region_result: {}, + "query_realtime_region_pipe_flow_and_demand_id": lambda name, region_result: {}, + "query_pipe_flow_region_patterns": lambda name: {}, + "query_non_realtime_region_patterns": lambda name, region_result: {}, + "get_realtime_region_patterns": lambda name, source_outflow_region_id, realtime_region_pipe_flow_and_demand_id: ({}, {}), + }, + ) + install_stub(monkeypatch, "app.services.globals", {}) + install_stub( + monkeypatch, + "app.services.tjnetwork", + { + "run_project": lambda network: "report", + "run_project_return_dict": lambda network: {"output": {}, "report": "ok"}, + "run_inp": lambda network: "inp-report", + "dump_output": lambda output: f"dump::{output}", + }, + ) + install_stub(monkeypatch, "app.algorithms", package=True) + install_stub(monkeypatch, "app.algorithms.simulation", package=True) + install_stub( + monkeypatch, + "app.algorithms.simulation.scenarios", + { + "burst_analysis": lambda *args, **kwargs: "burst", + "valve_close_analysis": lambda *args, **kwargs: "valve", + "flushing_analysis": lambda *args, **kwargs: "flush", + "contaminant_simulation": lambda *args, **kwargs: "contaminant", + "age_analysis": lambda *args, **kwargs: "age", + "pressure_regulation": lambda *args, **kwargs: "pressure", + }, + ) + install_stub( + monkeypatch, + "app.algorithms.sensor", + { + "pressure_sensor_placement_sensitivity": lambda *args, **kwargs: [], + "pressure_sensor_placement_kmeans": lambda *args, **kwargs: [], + }, + ) + install_stub( + monkeypatch, + "app.services.network_import", + {"network_update": lambda *args, **kwargs: "updated"}, + ) + install_stub( + monkeypatch, + "app.services.simulation_ops", + { + "project_management": lambda *args, **kwargs: "managed", + "scheduling_simulation": lambda *args, **kwargs: "scheduled", + "daily_scheduling_simulation": lambda *args, **kwargs: "daily", + }, + ) + install_stub( + monkeypatch, + "app.services.valve_isolation", + {"analyze_valve_isolation": lambda *args, **kwargs: {}}, + ) + return load_module_from_path( + "tests_simulation_endpoints_module", + "app/api/v1/endpoints/simulation.py", + ) + + +def test_run_project_endpoint_returns_plain_text(monkeypatch): + module = _load_simulation_module(monkeypatch) + monkeypatch.setattr(module, "run_project", lambda network: f"report::{network}") + client = TestClient(build_test_app(module.router, "/api/v1")) + + response = client.get("/api/v1/runproject/", params={"network": "demo"}) + + assert response.status_code == 200 + assert response.text == "report::demo" + + +def test_scheduling_analysis_maps_request_body(monkeypatch): + module = _load_simulation_module(monkeypatch) + captured = {} + + def fake_schedule(network, start_time, pump_control, tank_id, water_plant_output_id, time_delta): + captured["args"] = ( + network, + start_time, + pump_control, + tank_id, + water_plant_output_id, + time_delta, + ) + return "scheduled" + + monkeypatch.setattr(module, "scheduling_simulation", fake_schedule) + client = TestClient(build_test_app(module.router, "/api/v1")) + + response = client.post( + "/api/v1/scheduling_analysis/", + json={ + "network": "demo", + "start_time": "2025-01-01T08:00:00+08:00", + "pump_control": {"P1": [1, 0, 1]}, + "tank_id": "T1", + "water_plant_output_id": "R1", + }, + ) + + assert response.status_code == 200 + assert response.json() == "scheduled" + assert captured["args"] == ( + "demo", + "2025-01-01T08:00:00+08:00", + {"P1": [1, 0, 1]}, + "T1", + "R1", + 300, + ) + + +def test_project_management_maps_named_arguments(monkeypatch): + module = _load_simulation_module(monkeypatch) + captured = {} + + def fake_project_management(**kwargs): + captured.update(kwargs) + return "managed" + + monkeypatch.setattr(module, "project_management", fake_project_management) + client = TestClient(build_test_app(module.router, "/api/v1")) + + response = client.post( + "/api/v1/project_management/", + json={ + "network": "demo", + "start_time": "2025-01-01T08:00:00+08:00", + "pump_control": {"P1": [1]}, + "tank_init_level": {"T1": 10.0}, + "region_demand": {"R1": 20.0}, + }, + ) + + assert response.status_code == 200 + assert response.json() == "managed" + assert captured == { + "prj_name": "demo", + "start_datetime": "2025-01-01T08:00:00+08:00", + "pump_control": {"P1": [1]}, + "tank_initial_level_control": {"T1": 10.0}, + "region_demand_control": {"R1": 20.0}, + } + + +def test_network_update_surfaces_service_error(monkeypatch, tmp_path): + module = _load_simulation_module(monkeypatch) + monkeypatch.chdir(tmp_path) + + def boom(_path): + raise RuntimeError("write failed") + + monkeypatch.setattr(module, "network_update", boom) + client = TestClient(build_test_app(module.router, "/api/v1")) + + response = client.post( + "/api/v1/network_update/", + files={"file": ("update.txt", b"payload")}, + ) + + assert response.status_code == 500 + assert "数据库操作失败: write failed" in response.json()["detail"] + assert list(Path(tmp_path).glob("network_update_*")) + + +def test_run_simulation_manually_by_date_uses_utc_aware_timestamps(monkeypatch): + module = _load_simulation_module(monkeypatch) + captured_calls = [] + + monkeypatch.setattr( + module.simulation, + "run_simulation", + lambda **kwargs: captured_calls.append(kwargs), + ) + + module.run_simulation_manually_by_date( + "demo", + datetime(2025, 1, 1, 19, 4, 5, tzinfo=timezone.utc), + 30, + ) + + assert [call["modify_pattern_start_time"] for call in captured_calls] == [ + "2025-01-01T19:04:05+00:00", + "2025-01-01T19:19:05+00:00", + ] + + +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 = {} + + def fake_run(network_name, start_time, duration): + captured["network_name"] = network_name + captured["start_time"] = start_time + captured["duration"] = duration + + monkeypatch.setattr(module, "run_simulation_manually_by_date", fake_run) + client = TestClient(build_test_app(module.router, "/api/v1")) + + response = client.post( + "/api/v1/runsimulationmanuallybydate/", + json={ + "name": "demo", + "start_time": "2025-01-02T03:04:05+08:00", + "duration": 30, + }, + ) + + assert response.status_code == 200 + assert response.json() == {"status": "success"} + assert captured["network_name"] == "demo" + assert captured["duration"] == 30 + assert captured["start_time"].isoformat() == "2025-01-01T19:04:05+00:00" + + +def test_runsimulationmanuallybydate_endpoint_rejects_naive_start_time(monkeypatch): + module = _load_simulation_module(monkeypatch) + client = TestClient(build_test_app(module.router, "/api/v1")) + + response = client.post( + "/api/v1/runsimulationmanuallybydate/", + json={ + "name": "demo", + "start_time": "2025-01-02T03:04:05", + "duration": 30, + }, + ) + + assert response.status_code == 422 + + +def test_valve_close_endpoint_passes_scheme_name(monkeypatch): + module = _load_simulation_module(monkeypatch) + captured = {} + + def fake_valve_close_analysis(**kwargs): + captured.update(kwargs) + return "ok" + + monkeypatch.setattr(module, "valve_close_analysis", fake_valve_close_analysis) + client = TestClient(build_test_app(module.router, "/api/v1")) + + response = client.get( + "/api/v1/valve_close_analysis/", + params={ + "network": "demo", + "start_time": "2025-01-02T03:04:05+08:00", + "valves": ["V1", "V2"], + "duration": 900, + "scheme_name": "valve_case_01", + }, + ) + + assert response.status_code == 200 + assert response.text == "ok" + assert captured == { + "name": "demo", + "modify_pattern_start_time": "2025-01-02T03:04:05+08:00", + "modify_total_duration": 900, + "modify_valve_opening": {"V1": 0.0, "V2": 0.0}, + "scheme_name": "valve_case_01", + } + + +def test_flushing_endpoint_passes_required_scheme_name(monkeypatch): + module = _load_simulation_module(monkeypatch) + captured = {} + + def fake_flushing_analysis(**kwargs): + captured.update(kwargs) + return "ok" + + monkeypatch.setattr(module, "flushing_analysis", fake_flushing_analysis) + client = TestClient(build_test_app(module.router, "/api/v1")) + + response = client.get( + "/api/v1/flushing_analysis/", + params={ + "network": "demo", + "start_time": "2025-01-02T03:04:05+08:00", + "valves": ["V1"], + "valves_k": [0.5], + "drainage_node_ID": "N1", + "flush_flow": 100.0, + "duration": 900, + "scheme_name": "flush_case_01", + }, + ) + + assert response.status_code == 200 + assert response.text == "ok" + assert captured == { + "name": "demo", + "modify_pattern_start_time": "2025-01-02T03:04:05+08:00", + "modify_total_duration": 900, + "modify_valve_opening": {"V1": 0.5}, + "drainage_node_ID": "N1", + "flushing_flow": 100.0, + "scheme_name": "flush_case_01", + } + + +def test_contaminant_endpoint_requires_scheme_name(monkeypatch): + module = _load_simulation_module(monkeypatch) + client = TestClient(build_test_app(module.router, "/api/v1")) + + response = client.get( + "/api/v1/contaminant_simulation/", + params={ + "network": "demo", + "start_time": "2025-01-02T03:04:05+08:00", + "source": "N1", + "concentration": 10.0, + "duration": 900, + }, + ) + + assert response.status_code == 422 diff --git a/tests/unit/test_time_api.py b/tests/unit/test_time_api.py new file mode 100644 index 0000000..319c316 --- /dev/null +++ b/tests/unit/test_time_api.py @@ -0,0 +1,71 @@ +import importlib.util +from datetime import date, datetime, timedelta, timezone +from pathlib import Path + +import pytest + + +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 test_parse_utc_time_rejects_naive_datetimes(): + module = _load_time_api_module() + + with pytest.raises(ValueError, match="timezone information"): + module.parse_utc_time("2025-01-01T08:00:00") + + +def test_parse_utc_time_normalizes_offset_datetime_to_utc(): + module = _load_time_api_module() + result = module.parse_utc_time("2025-01-01T08:00:00+08:00") + + assert result == datetime(2025, 1, 1, 0, 0, tzinfo=timezone.utc) + + +def test_extract_date_keeps_original_offset_calendar_day(): + module = _load_time_api_module() + result = module.extract_date("2025-01-01T00:30:00+08:00") + + assert result == date(2025, 1, 1) + + +def test_utc_now_returns_timezone_aware_utc_datetime(): + module = _load_time_api_module() + result = module.utc_now() + + 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)