Compare commits
2
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
38a51ed65c | ||
|
|
a3b4e75317 |
@@ -1,10 +1,11 @@
|
||||
"""Topology-only valve isolation search."""
|
||||
|
||||
from collections import defaultdict, deque
|
||||
from typing import Any, Iterable
|
||||
from typing import Any, Iterable, Literal
|
||||
|
||||
|
||||
VALVE_LINK_TYPE = "valve"
|
||||
AccidentElementType = Literal["auto", "link", "node"]
|
||||
|
||||
|
||||
def _parse_link_entry(link_entry: str) -> tuple[str, str, str, str]:
|
||||
@@ -14,12 +15,37 @@ def _parse_link_entry(link_entry: str) -> tuple[str, str, str, str]:
|
||||
return parts[0], parts[1], parts[2], parts[3]
|
||||
|
||||
|
||||
def _reachable_nodes(
|
||||
adjacency: dict[str, list[tuple[str, str]]],
|
||||
start_nodes: Iterable[str],
|
||||
blocked_links: set[str],
|
||||
) -> set[str]:
|
||||
visited = set(start_nodes)
|
||||
queue = deque(visited)
|
||||
while queue:
|
||||
for neighbor, link_id in adjacency.get(queue.popleft(), ()):
|
||||
if link_id not in blocked_links and neighbor not in visited:
|
||||
visited.add(neighbor)
|
||||
queue.append(neighbor)
|
||||
return visited
|
||||
|
||||
|
||||
def valve_isolation_analysis(
|
||||
link_entries: Iterable[str],
|
||||
accident_elements: str | list[str],
|
||||
disabled_valves: list[str] | None = None,
|
||||
*,
|
||||
accident_element_type: AccidentElementType = "auto",
|
||||
water_source_nodes: Iterable[str] | None = None,
|
||||
closed_links: Iterable[str] = (),
|
||||
) -> dict[str, Any]:
|
||||
"""Determine boundary valves and affected nodes from a topology snapshot."""
|
||||
"""Find nearest valve boundaries, then check source connectivity.
|
||||
|
||||
Legacy untyped IDs prefer links; explicit node mode handles cross-type ID
|
||||
collisions. Supply impact is a before/after connectivity difference, not
|
||||
a hydraulic pressure or demand calculation. Without source metadata only
|
||||
the provisional boundary search is available.
|
||||
"""
|
||||
|
||||
disabled_valves_set = set(disabled_valves or [])
|
||||
target_elements = (
|
||||
@@ -27,8 +53,13 @@ def valve_isolation_analysis(
|
||||
if isinstance(accident_elements, str)
|
||||
else accident_elements
|
||||
)
|
||||
if not target_elements:
|
||||
raise ValueError("At least one accident element is required")
|
||||
if accident_element_type not in {"auto", "link", "node"}:
|
||||
raise ValueError(f"Invalid accident element type: {accident_element_type}")
|
||||
closed_links_set = set(closed_links)
|
||||
|
||||
pipe_adj: dict[str, set[str]] = defaultdict(set)
|
||||
adjacency: dict[str, list[tuple[str, str]]] = defaultdict(list)
|
||||
all_valves: dict[str, tuple[str, str]] = {}
|
||||
link_lookup: dict[str, tuple[str, str, str]] = {}
|
||||
node_set: set[str] = set()
|
||||
@@ -39,46 +70,31 @@ def valve_isolation_analysis(
|
||||
node_set.update((node1, node2))
|
||||
if link_type_name == VALVE_LINK_TYPE:
|
||||
all_valves[link_id] = (node1, node2)
|
||||
else:
|
||||
pipe_adj[node1].add(node2)
|
||||
pipe_adj[node2].add(node1)
|
||||
if link_id not in closed_links_set:
|
||||
adjacency[node1].append((node2, link_id))
|
||||
adjacency[node2].append((node1, link_id))
|
||||
|
||||
unknown_disabled = disabled_valves_set - all_valves.keys()
|
||||
if unknown_disabled:
|
||||
raise ValueError(f"Disabled valve was not found in topology: {sorted(unknown_disabled)}")
|
||||
|
||||
start_nodes: set[str] = set()
|
||||
for element in target_elements:
|
||||
if element in node_set:
|
||||
start_nodes.add(element)
|
||||
elif element in link_lookup:
|
||||
if accident_element_type != "node" and element in link_lookup:
|
||||
node1, node2, _ = link_lookup[element]
|
||||
start_nodes.update((node1, node2))
|
||||
elif accident_element_type != "link" and element in node_set:
|
||||
start_nodes.add(element)
|
||||
else:
|
||||
raise ValueError(f"Accident element {element} was not found in topology")
|
||||
|
||||
extra_adj: dict[str, list[str]] = defaultdict(list)
|
||||
boundary_valves: dict[str, tuple[str, str]] = {}
|
||||
for valve_id, (node1, node2) in all_valves.items():
|
||||
if valve_id in disabled_valves_set:
|
||||
extra_adj[node1].append(node2)
|
||||
extra_adj[node2].append(node1)
|
||||
else:
|
||||
boundary_valves[valve_id] = (node1, node2)
|
||||
|
||||
affected_nodes: set[str] = set()
|
||||
queue = deque(start_nodes)
|
||||
while queue:
|
||||
node = queue.popleft()
|
||||
if node in affected_nodes:
|
||||
continue
|
||||
affected_nodes.add(node)
|
||||
queue.extend(pipe_adj.get(node, set()) - affected_nodes)
|
||||
queue.extend(
|
||||
neighbor
|
||||
for neighbor in extra_adj.get(node, ())
|
||||
if neighbor not in affected_nodes
|
||||
)
|
||||
closable_valves = set(all_valves) - disabled_valves_set - closed_links_set
|
||||
affected_nodes = _reachable_nodes(adjacency, start_nodes, closable_valves)
|
||||
|
||||
must_close_valves: list[str] = []
|
||||
optional_valves: list[str] = []
|
||||
for valve_id, (node1, node2) in boundary_valves.items():
|
||||
for valve_id in closable_valves:
|
||||
node1, node2 = all_valves[valve_id]
|
||||
node1_affected = node1 in affected_nodes
|
||||
node2_affected = node2 in affected_nodes
|
||||
if node1_affected and node2_affected:
|
||||
@@ -89,14 +105,37 @@ def valve_isolation_analysis(
|
||||
must_close_valves.sort()
|
||||
optional_valves.sort()
|
||||
isolatable = bool(must_close_valves)
|
||||
failure_reason = None if isolatable else "no_boundary_valves"
|
||||
sources = set(water_source_nodes or ())
|
||||
source_check_performed = bool(sources)
|
||||
supply_affected_nodes = None
|
||||
if water_source_nodes is not None:
|
||||
if not sources:
|
||||
isolatable = False
|
||||
failure_reason = "water_sources_unavailable"
|
||||
else:
|
||||
after_closure = _reachable_nodes(adjacency, sources, set(must_close_valves))
|
||||
if affected_nodes & after_closure:
|
||||
isolatable = False
|
||||
failure_reason = "connected_to_water_source"
|
||||
if isolatable:
|
||||
before_closure = _reachable_nodes(adjacency, sources, set())
|
||||
supply_affected_nodes = sorted(before_closure - after_closure)
|
||||
result: dict[str, Any] = {
|
||||
"accident_elements": target_elements,
|
||||
"accident_element_type": accident_element_type,
|
||||
"disabled_valves": disabled_valves,
|
||||
"affected_nodes": sorted(affected_nodes) if isolatable else [],
|
||||
"affected_node_count": len(affected_nodes),
|
||||
"must_close_valves": must_close_valves,
|
||||
"optional_valves": optional_valves,
|
||||
"isolatable": isolatable,
|
||||
"source_check_performed": source_check_performed,
|
||||
"isolation_failure_reason": failure_reason,
|
||||
"supply_affected_nodes": supply_affected_nodes,
|
||||
"supply_affected_node_count": (
|
||||
len(supply_affected_nodes) if supply_affected_nodes is not None else None
|
||||
),
|
||||
}
|
||||
if len(target_elements) == 1:
|
||||
result["accident_element"] = target_elements[0]
|
||||
|
||||
@@ -7,6 +7,7 @@ from sqlalchemy.engine.url import make_url
|
||||
from sqlalchemy.exc import IntegrityError, SQLAlchemyError
|
||||
from sqlalchemy.ext.asyncio import create_async_engine
|
||||
|
||||
from app.api.pagination import PaginatedList
|
||||
from app.auth.metadata_dependencies import (
|
||||
get_current_metadata_admin,
|
||||
get_metadata_repository,
|
||||
@@ -236,7 +237,9 @@ async def list_metadata_users(
|
||||
metadata_repo: MetadataRepository = Depends(get_metadata_repository),
|
||||
) -> List[MetadataUserResponse]:
|
||||
users = await metadata_repo.list_users(skip=skip, limit=limit)
|
||||
return [MetadataUserResponse.model_validate(user) for user in users]
|
||||
total = await metadata_repo.count_users()
|
||||
items = [MetadataUserResponse.model_validate(user) for user in users]
|
||||
return PaginatedList(items, total=total)
|
||||
|
||||
|
||||
@router.get("/admin/projects", response_model=List[AdminProjectResponse])
|
||||
|
||||
@@ -180,6 +180,9 @@ def valve_isolation_endpoint(
|
||||
network: str = Query(..., description="管网名称(或数据库名称)"),
|
||||
accident_element: List[str] = Query(..., description="发生事故的管段/节点ID列表"),
|
||||
disabled_valves: List[str] = Query(None, description="已故障的阀门ID列表(可选)"),
|
||||
accident_element_type: Literal["auto", "link", "node"] = Query(
|
||||
"auto", description="事故要素类型;auto优先管段再节点;分析同名节点须指定node,纯管段分析可指定link",
|
||||
),
|
||||
):
|
||||
"""
|
||||
阀门隔离分析
|
||||
@@ -187,13 +190,19 @@ def valve_isolation_endpoint(
|
||||
- **network**: 管网名称(或数据库名称)
|
||||
- **accident_element**: 发生事故的管段/节点ID列表
|
||||
- **disabled_valves**: 已故障的阀门ID列表(可选)
|
||||
- **accident_element_type**: auto优先管段;link仅匹配管段;node仅匹配节点
|
||||
|
||||
返回隔离方案,包括:
|
||||
- must_close_valves: 必须关闭的阀门列表
|
||||
- optional_valves: 可选关闭的阀门列表
|
||||
- affected_nodes: 受影响的节点列表;不可隔离时为空列表
|
||||
- affected_node_count: 受影响的节点总数
|
||||
- isolatable: 是否可以有效隔离
|
||||
- affected_nodes / affected_node_count: 最近阀门围成的直接隔离区;不可隔离时列表为空,计数保留
|
||||
- supply_affected_nodes / supply_affected_node_count: 按建议关阀后新增失去水源连接的节点;未验证或不可隔离时为null
|
||||
- source_check_performed: 是否已执行水源连通检查
|
||||
- isolation_failure_reason: 不可隔离原因;可隔离时为null
|
||||
- isolatable: 存在关阀边界,且直接隔离区不再连通任何水源
|
||||
|
||||
使用模型初始开关状态进行无向拓扑分析,水池按潜在水源处理。
|
||||
不包含时变控制、水池蓄水量或压力流量计算,不是现场停水范围的水力证明。
|
||||
"""
|
||||
# result = {
|
||||
# "accident_element": "P461309",
|
||||
@@ -212,7 +221,13 @@ def valve_isolation_endpoint(
|
||||
# "must_close_valves": ["210521658", "V12974", "V12986", "V12993"],
|
||||
# "optional_valves": [],
|
||||
# }
|
||||
result = analyze_valve_isolation(network, accident_element, disabled_valves)
|
||||
try:
|
||||
result = analyze_valve_isolation(
|
||||
network, accident_element, disabled_valves,
|
||||
accident_element_type=accident_element_type,
|
||||
)
|
||||
except ValueError as exc:
|
||||
raise HTTPException(status_code=422, detail=str(exc)) from exc
|
||||
return result
|
||||
|
||||
|
||||
|
||||
@@ -4,7 +4,7 @@ from typing import Optional, List
|
||||
from uuid import UUID, uuid4
|
||||
|
||||
from cryptography.fernet import InvalidToken
|
||||
from sqlalchemy import delete, select
|
||||
from sqlalchemy import delete, func, select
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
|
||||
from app.core.encryption import (
|
||||
@@ -122,6 +122,12 @@ class MetadataRepository:
|
||||
)
|
||||
return list(result.scalars().all())
|
||||
|
||||
async def count_users(self) -> int:
|
||||
result = await self.session.execute(
|
||||
select(func.count()).select_from(models.User)
|
||||
)
|
||||
return int(result.scalar() or 0)
|
||||
|
||||
async def upsert_user_from_keycloak(
|
||||
self,
|
||||
*,
|
||||
|
||||
@@ -1,4 +1,4 @@
|
||||
"""Read-only projections backed by the GIS materialized-view query layer."""
|
||||
"""Read-only GIS projections and authoritative network topology snapshots."""
|
||||
|
||||
from typing import Any
|
||||
|
||||
@@ -69,6 +69,46 @@ def get_network_link_nodes(name: str) -> list[str]:
|
||||
]
|
||||
|
||||
|
||||
def get_valve_isolation_topology(name: str) -> tuple[list[str], list[str], list[str]]:
|
||||
"""Read links, closed statuses and sources from one consistent SQL snapshot.
|
||||
|
||||
Isolation must include model links without published geometries, and use
|
||||
initial statuses rather than treating every link as open. UNION ALL keeps
|
||||
source nodes in the same statement snapshot without repeating them for
|
||||
every link or relying on materialized view refresh timing.
|
||||
"""
|
||||
rows = read_all(
|
||||
name,
|
||||
"""
|
||||
SELECT l.id, l.link_type::text AS element_type,
|
||||
l.start_node_id, l.end_node_id,
|
||||
COALESCE(s.status::text, p.status::text) AS status
|
||||
FROM network.links AS l
|
||||
LEFT JOIN network.pipes AS p ON p.link_id = l.id
|
||||
LEFT JOIN network.link_initial_settings AS s ON s.link_id = l.id
|
||||
UNION ALL
|
||||
SELECT n.id, n.node_type::text AS element_type,
|
||||
NULL, NULL, NULL
|
||||
FROM network.nodes AS n
|
||||
WHERE n.node_type::text IN ('reservoir', 'tank')
|
||||
""",
|
||||
)
|
||||
link_entries: list[str] = []
|
||||
source_nodes: list[str] = []
|
||||
closed_links: list[str] = []
|
||||
for row in rows:
|
||||
element_id = str(row["id"])
|
||||
if row["element_type"] in {"reservoir", "tank"}:
|
||||
source_nodes.append(element_id)
|
||||
else:
|
||||
link_entries.append(
|
||||
f"{element_id}:{row['element_type']}:{row['start_node_id']}:{row['end_node_id']}"
|
||||
)
|
||||
if row["status"] == "CLOSED":
|
||||
closed_links.append(element_id)
|
||||
return link_entries, source_nodes, closed_links
|
||||
|
||||
|
||||
def get_major_pipe_nodes(name: str, diameter: int) -> list[str]:
|
||||
"""Return large pipes in the established API wire format."""
|
||||
rows = read_all(
|
||||
|
||||
@@ -1,13 +1,22 @@
|
||||
from typing import Any
|
||||
|
||||
from app.algorithms.valve_isolation.topology_search import valve_isolation_analysis
|
||||
from app.native.wndb.gis.network_views import get_network_link_nodes
|
||||
from app.algorithms.valve_isolation.topology_search import (
|
||||
AccidentElementType,
|
||||
valve_isolation_analysis,
|
||||
)
|
||||
from app.native.wndb.gis.network_views import get_valve_isolation_topology
|
||||
|
||||
|
||||
def analyze_valve_isolation(
|
||||
network: str,
|
||||
accident_element: str | list[str],
|
||||
disabled_valves: list[str] = None,
|
||||
disabled_valves: list[str] | None = None,
|
||||
accident_element_type: AccidentElementType = "auto",
|
||||
) -> dict[str, Any]:
|
||||
link_entries = get_network_link_nodes(network)
|
||||
return valve_isolation_analysis(link_entries, accident_element, disabled_valves)
|
||||
link_entries, source_nodes, closed_links = get_valve_isolation_topology(network)
|
||||
return valve_isolation_analysis(
|
||||
link_entries, accident_element, disabled_valves,
|
||||
accident_element_type=accident_element_type,
|
||||
water_source_nodes=source_nodes,
|
||||
closed_links=closed_links,
|
||||
)
|
||||
|
||||
@@ -3,7 +3,7 @@
|
||||
"contracts": {
|
||||
"server": {
|
||||
"file": "server-v1.openapi.json",
|
||||
"sha256": "b565d841061c9091f48ff3118bcc0cbb1b918b0cb8c2316d177570e8b2d8ba29"
|
||||
"sha256": "5b8ec0b469f1194e72214003a9cf427489f48de19ae31ee6492821cf64916b6d"
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -35969,6 +35969,23 @@
|
||||
"type": "array"
|
||||
}
|
||||
},
|
||||
{
|
||||
"description": "事故要素类型;auto优先管段再节点;分析同名节点须指定node,纯管段分析可指定link",
|
||||
"in": "query",
|
||||
"name": "accident_element_type",
|
||||
"required": false,
|
||||
"schema": {
|
||||
"default": "auto",
|
||||
"description": "事故要素类型;auto优先管段再节点;分析同名节点须指定node,纯管段分析可指定link",
|
||||
"enum": [
|
||||
"auto",
|
||||
"link",
|
||||
"node"
|
||||
],
|
||||
"title": "Accident Element Type",
|
||||
"type": "string"
|
||||
}
|
||||
},
|
||||
{
|
||||
"in": "header",
|
||||
"name": "X-Project-Id",
|
||||
|
||||
@@ -7,8 +7,9 @@ import pytest
|
||||
from fastapi import HTTPException
|
||||
from fastapi import Response
|
||||
|
||||
from app.auth.metadata_dependencies import get_current_metadata_admin
|
||||
from app.api.v1.endpoints import admin_metadata
|
||||
from app.api.pagination import PaginatedList
|
||||
from app.auth.metadata_dependencies import get_current_metadata_admin
|
||||
from app.domain.schemas.admin_metadata import (
|
||||
AdminProjectCreateRequest,
|
||||
MetadataUsersBatchSyncRequest,
|
||||
@@ -83,6 +84,28 @@ def test_to_async_sqlalchemy_url_preserves_password():
|
||||
assert "***" not in url
|
||||
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_list_metadata_users_returns_page_with_real_total():
|
||||
users = [_user(username="alice"), _user(username="bob")]
|
||||
repo = SimpleNamespace(
|
||||
list_users=AsyncMock(return_value=users),
|
||||
count_users=AsyncMock(return_value=7),
|
||||
)
|
||||
|
||||
response = await admin_metadata.list_metadata_users(
|
||||
skip=2,
|
||||
limit=2,
|
||||
current_user=_user(role="admin", is_superuser=True),
|
||||
metadata_repo=repo,
|
||||
)
|
||||
|
||||
assert isinstance(response, PaginatedList)
|
||||
assert response.total == 7
|
||||
assert [item.username for item in response] == ["alice", "bob"]
|
||||
repo.list_users.assert_awaited_once_with(skip=2, limit=2)
|
||||
repo.count_users.assert_awaited_once_with()
|
||||
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_sync_metadata_user_upserts_without_password(monkeypatch):
|
||||
keycloak_id = uuid4()
|
||||
|
||||
@@ -254,14 +254,16 @@ def test_committed_write_refresh_failure_has_explicit_http_contract() -> None:
|
||||
assert response.json()["code"] == "materialized_view_refresh_failed_after_commit"
|
||||
|
||||
|
||||
def test_valve_isolation_runtime_accepts_frontend_query(monkeypatch) -> None:
|
||||
@pytest.mark.parametrize("element_type", ["auto", "link", "node"])
|
||||
def test_valve_isolation_runtime_accepts_frontend_query(monkeypatch, element_type) -> None:
|
||||
captured: dict[str, object] = {}
|
||||
|
||||
def fake_analyze_valve_isolation(network, accident_element, disabled_valves):
|
||||
def fake_analyze_valve_isolation(network, accident_element, disabled_valves, accident_element_type):
|
||||
captured.update(
|
||||
network=network,
|
||||
accident_element=accident_element,
|
||||
disabled_valves=disabled_valves,
|
||||
accident_element_type=accident_element_type,
|
||||
business_dsn=get_project_pgconn_string(network),
|
||||
timescale_dsn=get_project_timescale_pgconn_string(network),
|
||||
)
|
||||
@@ -288,6 +290,7 @@ def test_valve_isolation_runtime_accepts_frontend_query(monkeypatch) -> None:
|
||||
("accident_element", "P-1"),
|
||||
("accident_element", "P-2"),
|
||||
("disabled_valves", "V-9"),
|
||||
*([] if element_type == "auto" else [("accident_element_type", element_type)]),
|
||||
],
|
||||
)
|
||||
|
||||
@@ -297,11 +300,39 @@ def test_valve_isolation_runtime_accepts_frontend_query(monkeypatch) -> None:
|
||||
"network": "fengyang",
|
||||
"accident_element": ["P-1", "P-2"],
|
||||
"disabled_valves": ["V-9"],
|
||||
"accident_element_type": element_type,
|
||||
"business_dsn": "postgresql://user:password@biz/fengyang",
|
||||
"timescale_dsn": "postgresql://user:password@timescale/fengyang",
|
||||
}
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"target,disabled,detail",
|
||||
[("P", "typo", "Disabled valve"), ("missing", "V", "Accident element")],
|
||||
)
|
||||
def test_valve_isolation_invalid_ids_return_validation_error(monkeypatch, target, disabled, detail):
|
||||
from app.services import valve_isolation as service
|
||||
|
||||
monkeypatch.setattr(
|
||||
service, "get_valve_isolation_topology",
|
||||
lambda _network: (["P:pipe:A:B", "V:valve:B:R"], ["R"], []),
|
||||
)
|
||||
monkeypatch.setattr(simulation_endpoint, "analyze_valve_isolation", service.analyze_valve_isolation)
|
||||
app = FastAPI()
|
||||
app.include_router(api_router, prefix="/api/v1")
|
||||
_override_project_routing(app, ProjectContext(
|
||||
project_id=uuid4(), project_code="project", user_id=uuid4(), project_role="member",
|
||||
))
|
||||
|
||||
response = TestClient(app, raise_server_exceptions=False).post(
|
||||
"/api/v1/valve-isolation-analyses",
|
||||
params={"accident_element": target, "disabled_valves": disabled},
|
||||
)
|
||||
|
||||
assert response.status_code == 422
|
||||
assert detail in response.json()["detail"]
|
||||
|
||||
|
||||
def test_scada_cleaning_runs_are_post() -> None:
|
||||
methods_by_path = {
|
||||
route.path: route.methods
|
||||
|
||||
@@ -1,3 +1,5 @@
|
||||
import pytest
|
||||
|
||||
from app.algorithms.valve_isolation import topology_search as valve
|
||||
|
||||
|
||||
@@ -23,6 +25,183 @@ def test_isolatable_keeps_affected_node_ids_and_count():
|
||||
assert result["must_close_valves"] == ["V-close"]
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def colliding_topology():
|
||||
# The reported adjacent pipes and their same-numbered nodes are in
|
||||
# different valve-bounded regions. Keep this fixture independent of a DB.
|
||||
return [
|
||||
"19025:pipe:17038:17037",
|
||||
"19026:pipe:17038:17039",
|
||||
"AUTO_TCV_4:valve:17037:D",
|
||||
"AUTO_TCV_5:valve:17039:E",
|
||||
"AUTO_TCV_546:valve:17038:R",
|
||||
"32252:pipe:19025:X",
|
||||
"15383:pipe:19026:Y",
|
||||
"V-wrong-25:valve:X:R",
|
||||
"V-wrong-26:valve:Y:R",
|
||||
]
|
||||
|
||||
|
||||
@pytest.mark.parametrize("target", ["19025", "19026", ["19025", "19026"]])
|
||||
def test_adjacent_pipes_take_priority_over_same_numbered_nodes(colliding_topology, target):
|
||||
result = valve.valve_isolation_analysis(colliding_topology, target)
|
||||
|
||||
assert result["affected_nodes"] == ["17037", "17038", "17039"]
|
||||
assert result["must_close_valves"] == ["AUTO_TCV_4", "AUTO_TCV_5", "AUTO_TCV_546"]
|
||||
assert result["optional_valves"] == []
|
||||
|
||||
|
||||
def test_explicit_node_type_resolves_the_other_side_of_an_id_collision(colliding_topology):
|
||||
result = valve.valve_isolation_analysis(
|
||||
colliding_topology, "19025", accident_element_type="node",
|
||||
)
|
||||
|
||||
assert result["affected_nodes"] == ["19025", "X"]
|
||||
assert result["must_close_valves"] == ["V-wrong-25"]
|
||||
|
||||
|
||||
def test_unambiguous_legacy_node_input_still_works():
|
||||
result = valve.valve_isolation_analysis(["P:pipe:A:B", "V:valve:B:R"], "A")
|
||||
assert result["affected_nodes"] == ["A", "B"]
|
||||
|
||||
|
||||
def test_explicit_link_type_does_not_fall_back_to_a_node():
|
||||
with pytest.raises(ValueError, match="not found"):
|
||||
valve.valve_isolation_analysis(
|
||||
["P:pipe:A:B"], "A", accident_element_type="link",
|
||||
)
|
||||
|
||||
|
||||
def test_a_boundary_valve_does_not_prove_source_isolation():
|
||||
result = valve.valve_isolation_analysis(
|
||||
["P:pipe:R:A", "V:valve:A:B"], "P", water_source_nodes=["R"],
|
||||
)
|
||||
|
||||
assert result["isolatable"] is False
|
||||
assert result["source_check_performed"] is True
|
||||
assert result["isolation_failure_reason"] == "connected_to_water_source"
|
||||
assert result["affected_nodes"] == []
|
||||
assert result["affected_node_count"] == 2
|
||||
assert result["supply_affected_nodes"] is None
|
||||
|
||||
|
||||
def test_supply_impact_includes_downstream_regions_but_not_preexisting_outages():
|
||||
links = [
|
||||
"P:pipe:A:B", "V-source:valve:A:R", "V-downstream:valve:B:C",
|
||||
"P-downstream:pipe:C:D", "P-already-off:pipe:X:Y",
|
||||
]
|
||||
result = valve.valve_isolation_analysis(links, "P", water_source_nodes=["R"])
|
||||
|
||||
assert result["isolatable"] is True
|
||||
assert result["source_check_performed"] is True
|
||||
assert result["affected_nodes"] == ["A", "B"]
|
||||
assert result["affected_node_count"] == 2
|
||||
assert result["supply_affected_nodes"] == ["A", "B", "C", "D"]
|
||||
assert result["supply_affected_node_count"] == 4
|
||||
|
||||
|
||||
def test_closed_links_are_not_traversed_or_recommended_for_closure():
|
||||
result = valve.valve_isolation_analysis(
|
||||
["P:pipe:A:B", "P-closed:pipe:B:R", "V:valve:A:R", "V-closed:valve:B:C"],
|
||||
"P", water_source_nodes=["R"], closed_links=["P-closed", "V-closed"],
|
||||
)
|
||||
|
||||
assert result["isolatable"] is True
|
||||
assert result["affected_nodes"] == ["A", "B"]
|
||||
assert result["must_close_valves"] == ["V"]
|
||||
assert result["supply_affected_nodes"] == ["A", "B"]
|
||||
|
||||
|
||||
def test_disabled_valve_cannot_cut_the_source_path():
|
||||
result = valve.valve_isolation_analysis(
|
||||
["P:pipe:A:B", "V-disabled:valve:A:R", "V-other:valve:B:C"],
|
||||
"P", disabled_valves=["V-disabled"], water_source_nodes=["R"],
|
||||
)
|
||||
|
||||
assert result["isolatable"] is False
|
||||
assert result["isolation_failure_reason"] == "connected_to_water_source"
|
||||
|
||||
|
||||
def test_joint_accidents_fail_if_any_region_remains_connected_to_a_source():
|
||||
result = valve.valve_isolation_analysis(
|
||||
["P1:pipe:A:B", "V1:valve:B:R", "P2:pipe:R:C", "V2:valve:C:D"],
|
||||
["P1", "P2"], water_source_nodes=["R"],
|
||||
)
|
||||
assert result["isolatable"] is False
|
||||
|
||||
|
||||
def test_missing_sources_do_not_claim_verified_isolation():
|
||||
result = valve.valve_isolation_analysis(
|
||||
["P:pipe:A:B", "V:valve:B:C"], "P", water_source_nodes=[],
|
||||
)
|
||||
assert result["isolatable"] is False
|
||||
assert result["source_check_performed"] is False
|
||||
assert result["isolation_failure_reason"] == "water_sources_unavailable"
|
||||
assert result["supply_affected_node_count"] is None
|
||||
|
||||
|
||||
@pytest.mark.parametrize("sources", [["R1"], ["R1", "R2"]])
|
||||
def test_all_water_sources_are_considered(sources):
|
||||
result = valve.valve_isolation_analysis(
|
||||
["P:pipe:A:B", "V:valve:A:R1", "P2:pipe:B:R2"],
|
||||
"P", water_source_nodes=sources,
|
||||
)
|
||||
assert result["isolatable"] is (len(sources) == 1)
|
||||
|
||||
|
||||
def test_empty_targets_are_rejected():
|
||||
with pytest.raises(ValueError, match="At least one"):
|
||||
valve.valve_isolation_analysis(["P:pipe:A:B"], [])
|
||||
|
||||
|
||||
def test_unknown_disabled_valve_is_rejected():
|
||||
with pytest.raises(ValueError, match="Disabled valve"):
|
||||
valve.valve_isolation_analysis(["P:pipe:A:B"], "P", disabled_valves=["typo"])
|
||||
|
||||
|
||||
def test_service_passes_authoritative_sources_statuses_and_element_type(monkeypatch):
|
||||
from app.services import valve_isolation as service
|
||||
|
||||
projects = []
|
||||
|
||||
def snapshot(network):
|
||||
projects.append(network)
|
||||
return (
|
||||
["P:pipe:A:B", "P-closed:pipe:B:R", "V:valve:A:R"],
|
||||
["R"], ["P-closed"],
|
||||
)
|
||||
|
||||
monkeypatch.setattr(service, "get_valve_isolation_topology", snapshot)
|
||||
result = service.analyze_valve_isolation(
|
||||
"trusted-project", ["A"], accident_element_type="node",
|
||||
)
|
||||
|
||||
assert projects == ["trusted-project"]
|
||||
assert result["isolatable"] is True
|
||||
assert result["source_check_performed"] is True
|
||||
assert result["must_close_valves"] == ["V"]
|
||||
assert result["supply_affected_nodes"] == ["A", "B"]
|
||||
|
||||
|
||||
def test_parallel_links_preserve_an_open_bypass():
|
||||
result = valve.valve_isolation_analysis(
|
||||
["P:pipe:A:B", "V:valve:B:R", "P-bypass:pipe:B:R"],
|
||||
"P", water_source_nodes=["R"],
|
||||
)
|
||||
assert result["isolatable"] is False
|
||||
assert result["isolation_failure_reason"] == "connected_to_water_source"
|
||||
|
||||
|
||||
def test_closed_disabled_valve_does_not_reopen_a_known_closed_link():
|
||||
result = valve.valve_isolation_analysis(
|
||||
["P:pipe:A:B", "V:valve:B:R", "V-disabled:valve:A:R"],
|
||||
"P", disabled_valves=["V-disabled"], closed_links=["V-disabled"],
|
||||
water_source_nodes=["R"],
|
||||
)
|
||||
assert result["isolatable"] is True
|
||||
assert result["must_close_valves"] == ["V"]
|
||||
|
||||
|
||||
def test_disabled_valve_expands_affected_area_before_counting():
|
||||
links = [
|
||||
"P-1:pipe:A:B",
|
||||
|
||||
@@ -52,6 +52,36 @@ def test_network_link_nodes_use_one_unified_view_query(monkeypatch) -> None:
|
||||
assert "FROM gis.network_links" in calls[0][1]
|
||||
|
||||
|
||||
def test_isolation_snapshot_reads_authoritative_links_statuses_and_sources_once(monkeypatch):
|
||||
calls = []
|
||||
|
||||
def fake_read_all(name, statement, params=None):
|
||||
calls.append((name, statement, params))
|
||||
return [
|
||||
{"id": "P", "element_type": "pipe", "start_node_id": "A", "end_node_id": "B", "status": None},
|
||||
{"id": "P-closed", "element_type": "pipe", "start_node_id": "A", "end_node_id": "R", "status": "CLOSED"},
|
||||
{"id": "V", "element_type": "valve", "start_node_id": "B", "end_node_id": "R", "status": "CLOSED"},
|
||||
{"id": "PU", "element_type": "pump", "start_node_id": "T", "end_node_id": "A", "status": "CLOSED"},
|
||||
{"id": "R", "element_type": "reservoir", "start_node_id": None, "end_node_id": None, "status": None},
|
||||
{"id": "T", "element_type": "tank", "start_node_id": None, "end_node_id": None, "status": None},
|
||||
]
|
||||
|
||||
monkeypatch.setattr(network_views, "read_all", fake_read_all)
|
||||
|
||||
assert network_views.get_valve_isolation_topology("project_a") == (
|
||||
["P:pipe:A:B", "P-closed:pipe:A:R", "V:valve:B:R", "PU:pump:T:A"],
|
||||
["R", "T"], ["P-closed", "V", "PU"],
|
||||
)
|
||||
assert len(calls) == 1
|
||||
name, statement, _params = calls[0]
|
||||
assert name == "project_a"
|
||||
assert "FROM network.links" in statement
|
||||
assert "network.link_initial_settings" in statement
|
||||
assert "COALESCE(s.status::text, p.status::text)" in statement
|
||||
assert "FROM network.nodes" in statement
|
||||
assert "UNION ALL" in statement
|
||||
|
||||
|
||||
def test_topology_rows_are_loaded_in_two_batch_queries(monkeypatch) -> None:
|
||||
calls = []
|
||||
responses = [
|
||||
|
||||
Reference in New Issue
Block a user