33 lines
1.2 KiB
Python
33 lines
1.2 KiB
Python
from typing import Any
|
|
|
|
from psycopg import AsyncConnection
|
|
|
|
|
|
class NetworkSettingsRepository:
|
|
@staticmethod
|
|
async def get_result_units(conn: AsyncConnection) -> dict[str, str]:
|
|
async with conn.cursor() as cur:
|
|
await cur.execute(
|
|
"""
|
|
SELECT engine_version, key, value
|
|
FROM network.simulation_settings
|
|
WHERE (engine_version = 'v3' AND key IN ('FLOW_UNITS', 'PRESSURE_UNITS'))
|
|
OR (engine_version = 'legacy' AND key IN ('UNITS', 'PRESSURE'))
|
|
ORDER BY CASE engine_version WHEN 'v3' THEN 0 ELSE 1 END
|
|
"""
|
|
)
|
|
rows: list[dict[str, Any]] = await cur.fetchall()
|
|
|
|
units: dict[str, str] = {}
|
|
for row in rows:
|
|
key = str(row["key"])
|
|
metric = "flow" if key in {"FLOW_UNITS", "UNITS"} else "pressure"
|
|
units.setdefault(metric, str(row["value"]).strip())
|
|
missing = {"flow", "pressure"} - units.keys()
|
|
if missing:
|
|
raise ValueError(
|
|
"network simulation settings missing result units: "
|
|
+ ", ".join(sorted(missing))
|
|
)
|
|
return units
|