OrgTJWater/TJWaterServerBinary
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(simulation): use hydraulic timestep

Browse Source
This commit is contained in:
jiang
2026-07-16 14:16:06 +08:00
parent d33619d0d5
commit 9f7f5536d7
6 changed files with 527 additions and 20 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/algorithms/simulation/runner.py
+3 -12
View File
@@ -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
app/api/v1/endpoints/simulation.py
+14 -2
View File
@@ -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/services/simulation.py
+9 -6
View File
@@ -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)
app/services/time_api.py
+35
View File
@@ -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
+395
View File
@@ -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
tests/unit/test_time_api.py
+71
View File
@@ -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)