123 lines
3.5 KiB
Python
123 lines
3.5 KiB
Python
from contextlib import contextmanager
|
|
import fcntl
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
from app.algorithms.sensor import kmeans as kmeans_sensor
|
|
from app.algorithms.sensor import sensitivity
|
|
from app.native.wndb.s42_sensor_placement import create_sensor_placement
|
|
from app.services.sensor_placement import (
|
|
SensorPlacementValidationError,
|
|
validate_sensor_placement_nodes,
|
|
)
|
|
from app.services.tjnetwork import dump_inp
|
|
|
|
|
|
def _sensor_inp_path(name: str) -> Path:
|
|
if (
|
|
not name
|
|
or name in {".", ".."}
|
|
or "/" in name
|
|
or "\\" in name
|
|
or "\x00" in name
|
|
):
|
|
raise SensorPlacementValidationError("管网名称不是有效的项目标识")
|
|
return Path("db_inp") / f"{name}.db.inp"
|
|
|
|
|
|
@contextmanager
|
|
def _sensor_inp_lock(name: str):
|
|
inp_path = _sensor_inp_path(name)
|
|
inp_path.parent.mkdir(parents=True, exist_ok=True)
|
|
lock_path = inp_path.with_suffix(".sensor.lock")
|
|
with lock_path.open("w", encoding="utf-8") as lock_file:
|
|
fcntl.flock(lock_file.fileno(), fcntl.LOCK_EX)
|
|
try:
|
|
yield inp_path
|
|
finally:
|
|
fcntl.flock(lock_file.fileno(), fcntl.LOCK_UN)
|
|
|
|
|
|
def _create_validated_placement(
|
|
name: str,
|
|
*,
|
|
scheme_name: str,
|
|
min_diameter: int,
|
|
username: str,
|
|
sensor_location: list[str],
|
|
) -> dict[str, Any]:
|
|
validate_sensor_placement_nodes(name, sensor_location)
|
|
return create_sensor_placement(
|
|
name,
|
|
scheme_name=scheme_name,
|
|
min_diameter=min_diameter,
|
|
username=username,
|
|
sensor_location=sensor_location,
|
|
)
|
|
|
|
|
|
def pressure_sensor_placement_sensitivity(
|
|
name: str,
|
|
scheme_name: str,
|
|
sensor_number: int,
|
|
min_diameter: int,
|
|
username: str,
|
|
) -> dict[str, Any]:
|
|
"""
|
|
基于改进灵敏度法进行压力监测点优化布置
|
|
:param name: 数据库名称
|
|
:param scheme_name: 监测优化布置方案名称
|
|
:param sensor_number: 传感器数目
|
|
:param min_diameter: 最小管径
|
|
:param username: 用户名
|
|
:return: 新建的监测点方案
|
|
"""
|
|
with _sensor_inp_lock(name):
|
|
sensor_location = sensitivity.get_ID(
|
|
name=name,
|
|
sensor_num=sensor_number,
|
|
min_diameter=min_diameter,
|
|
)
|
|
return _create_validated_placement(
|
|
name,
|
|
scheme_name=scheme_name,
|
|
min_diameter=min_diameter,
|
|
username=username,
|
|
sensor_location=sensor_location,
|
|
)
|
|
|
|
|
|
# 2025/08/21
|
|
# 基于kmeans聚类法进行压力监测点优化布置
|
|
def pressure_sensor_placement_kmeans(
|
|
name: str,
|
|
scheme_name: str,
|
|
sensor_number: int,
|
|
min_diameter: int,
|
|
username: str,
|
|
) -> dict[str, Any]:
|
|
"""
|
|
基于聚类法进行压力监测点优化布置
|
|
:param name: 数据库名称（注意，此处数据库名称也是inp文件名称，inp文件与pg库名要一样）
|
|
:param scheme_name: 监测优化布置方案名称
|
|
:param sensor_number: 传感器数目
|
|
:param min_diameter: 最小管径
|
|
:param username: 用户名
|
|
:return: 新建的监测点方案
|
|
"""
|
|
# dump_inp
|
|
with _sensor_inp_lock(name) as inp_path:
|
|
dump_inp(name, str(inp_path), "2")
|
|
sensor_location = kmeans_sensor.kmeans_sensor_placement(
|
|
name=name,
|
|
sensor_num=sensor_number,
|
|
min_diameter=min_diameter,
|
|
)
|
|
return _create_validated_placement(
|
|
name,
|
|
scheme_name=scheme_name,
|
|
min_diameter=min_diameter,
|
|
username=username,
|
|
sensor_location=sensor_location,
|
|
)
|