258 lines
7.8 KiB
Python
258 lines
7.8 KiB
Python
from datetime import datetime
|
|
from io import BytesIO
|
|
from typing import Any
|
|
|
|
from openpyxl import Workbook
|
|
from openpyxl.styles import Alignment, Font, PatternFill
|
|
from openpyxl.worksheet.worksheet import Worksheet
|
|
from openpyxl.utils import get_column_letter
|
|
from pyproj import Transformer
|
|
|
|
from app.native import wndb
|
|
|
|
|
|
class SensorPlacementNotFoundError(LookupError):
|
|
pass
|
|
|
|
|
|
class SensorPlacementValidationError(ValueError):
|
|
pass
|
|
|
|
|
|
class SensorPlacementConflictError(RuntimeError):
|
|
pass
|
|
|
|
|
|
_to_wgs84 = Transformer.from_crs("EPSG:3857", "EPSG:4326", always_xy=True)
|
|
_STATUS_LABELS = {
|
|
"current": "当前方案",
|
|
"original": "原方案",
|
|
"added": "新增",
|
|
"replaced": "替换",
|
|
}
|
|
_COORDINATE_DESCRIPTION = (
|
|
"工程 X/Y: 项目地方坐标系；地图 X/Y: EPSG:3857；经纬度: WGS84"
|
|
)
|
|
_LIST_HEADERS = (
|
|
"序号",
|
|
"节点 ID",
|
|
"经度",
|
|
"纬度",
|
|
"工程 X",
|
|
"工程 Y",
|
|
"地图 X",
|
|
"地图 Y",
|
|
"高程",
|
|
"调整状态",
|
|
)
|
|
_LIST_COLUMN_WIDTHS = (8, 20, 16, 16, 18, 18, 18, 18, 14, 14)
|
|
|
|
|
|
def _normalize_locations(sensor_location: list[str]) -> list[str]:
|
|
normalized = [str(node_id).strip() for node_id in sensor_location]
|
|
if not normalized or any(not node_id for node_id in normalized):
|
|
raise SensorPlacementValidationError("监测点列表不能为空")
|
|
if len(set(normalized)) != len(normalized):
|
|
raise SensorPlacementValidationError("监测点列表不能包含重复节点")
|
|
return normalized
|
|
|
|
|
|
def _sensor_points(
|
|
network: str,
|
|
sensor_location: list[str],
|
|
) -> list[dict[str, Any]]:
|
|
nodes = wndb.get_sensor_placement_nodes(network, sensor_location)
|
|
by_id = {str(node["node_id"]): node for node in nodes}
|
|
missing = [node_id for node_id in sensor_location if node_id not in by_id]
|
|
if missing:
|
|
raise SensorPlacementValidationError(
|
|
f"以下节点不存在或不是 junction: {', '.join(missing)}"
|
|
)
|
|
|
|
points: list[dict[str, Any]] = []
|
|
for node_id in sensor_location:
|
|
node = by_id[node_id]
|
|
project_x = float(node["project_x"])
|
|
project_y = float(node["project_y"])
|
|
map_x = float(node["map_x"])
|
|
map_y = float(node["map_y"])
|
|
longitude, latitude = _to_wgs84.transform(map_x, map_y)
|
|
points.append(
|
|
{
|
|
"node_id": node_id,
|
|
"project_x": project_x,
|
|
"project_y": project_y,
|
|
"map_x": map_x,
|
|
"map_y": map_y,
|
|
"longitude": float(longitude),
|
|
"latitude": float(latitude),
|
|
"elevation": float(node["elevation"]),
|
|
}
|
|
)
|
|
return points
|
|
|
|
|
|
def validate_sensor_placement_nodes(
|
|
network: str,
|
|
sensor_location: list[str],
|
|
) -> None:
|
|
_sensor_points(network, _normalize_locations(sensor_location))
|
|
|
|
|
|
def get_sensor_placement_scheme(network: str, scheme_id: int) -> dict[str, Any]:
|
|
scheme = wndb.get_sensor_placement(network, scheme_id)
|
|
if scheme is None:
|
|
raise SensorPlacementNotFoundError("监测点方案不存在")
|
|
|
|
locations = [str(item) for item in (scheme.get("sensor_location") or [])]
|
|
return {
|
|
**scheme,
|
|
"sensor_number": len(locations),
|
|
"sensor_location": locations,
|
|
"sensor_points": _sensor_points(network, locations),
|
|
}
|
|
|
|
|
|
def update_sensor_placement_scheme(
|
|
network: str,
|
|
scheme_id: int,
|
|
*,
|
|
expected_sensor_location: list[str],
|
|
sensor_location: list[str],
|
|
) -> dict[str, Any]:
|
|
expected = _normalize_locations(expected_sensor_location)
|
|
next_locations = _normalize_locations(sensor_location)
|
|
_sensor_points(network, next_locations)
|
|
|
|
updated = wndb.update_sensor_placement(
|
|
network,
|
|
scheme_id,
|
|
expected_sensor_location=expected,
|
|
sensor_location=next_locations,
|
|
)
|
|
if updated is None:
|
|
if wndb.get_sensor_placement(network, scheme_id) is None:
|
|
raise SensorPlacementNotFoundError("监测点方案不存在")
|
|
raise SensorPlacementConflictError("方案已被其他用户修改，请重新加载")
|
|
return get_sensor_placement_scheme(network, scheme_id)
|
|
|
|
|
|
def can_edit_sensor_placement(user: Any, scheme: dict[str, Any]) -> bool:
|
|
return bool(
|
|
getattr(user, "is_superuser", False)
|
|
or getattr(user, "role", None) == "admin"
|
|
or getattr(user, "username", None) == scheme.get("username")
|
|
)
|
|
|
|
|
|
def _safe_excel_text(value: Any) -> str:
|
|
text = "" if value is None else str(value)
|
|
if text.startswith(("=", "+", "-", "@")):
|
|
return f"'{text}"
|
|
return text
|
|
|
|
|
|
def _populate_info_sheet(
|
|
sheet: Worksheet,
|
|
*,
|
|
network: str,
|
|
scheme: dict[str, Any],
|
|
location_count: int,
|
|
is_draft: bool,
|
|
) -> None:
|
|
created_at = scheme["create_time"]
|
|
if isinstance(created_at, datetime):
|
|
created_at = created_at.isoformat(timespec="minutes")
|
|
|
|
rows = [
|
|
("项目", network),
|
|
("方案名称", scheme["scheme_name"]),
|
|
("监测点数量", location_count),
|
|
("最小管径", scheme["min_diameter"]),
|
|
("创建人", scheme["username"]),
|
|
("创建时间", created_at),
|
|
("导出时间", datetime.now().astimezone().isoformat(timespec="minutes")),
|
|
("文档状态", "未保存草稿" if is_draft else "当前方案"),
|
|
("坐标说明", _COORDINATE_DESCRIPTION),
|
|
]
|
|
for row_index, (label, value) in enumerate(rows, start=1):
|
|
sheet.cell(row=row_index, column=1, value=label)
|
|
safe_value = _safe_excel_text(value) if isinstance(value, str) else value
|
|
sheet.cell(row=row_index, column=2, value=safe_value)
|
|
sheet.column_dimensions["A"].width = 18
|
|
sheet.column_dimensions["B"].width = 64
|
|
|
|
|
|
def _populate_list_sheet(
|
|
sheet: Worksheet,
|
|
*,
|
|
points: list[dict[str, Any]],
|
|
adjustment_status: dict[str, str],
|
|
) -> None:
|
|
sheet.append(_LIST_HEADERS)
|
|
for index, point in enumerate(points, start=1):
|
|
status = adjustment_status.get(point["node_id"], "current")
|
|
sheet.append(
|
|
[
|
|
index,
|
|
_safe_excel_text(point["node_id"]),
|
|
point["longitude"],
|
|
point["latitude"],
|
|
point["project_x"],
|
|
point["project_y"],
|
|
point["map_x"],
|
|
point["map_y"],
|
|
point["elevation"],
|
|
_STATUS_LABELS.get(status, "当前方案"),
|
|
]
|
|
)
|
|
|
|
header_fill = PatternFill("solid", fgColor="257DD4")
|
|
for cell in sheet[1]:
|
|
cell.fill = header_fill
|
|
cell.font = Font(color="FFFFFF", bold=True)
|
|
cell.alignment = Alignment(horizontal="center", vertical="center")
|
|
sheet.freeze_panes = "A2"
|
|
sheet.auto_filter.ref = sheet.dimensions
|
|
for index, width in enumerate(_LIST_COLUMN_WIDTHS, start=1):
|
|
sheet.column_dimensions[get_column_letter(index)].width = width
|
|
for row in sheet.iter_rows(min_row=2):
|
|
row[0].alignment = Alignment(horizontal="center")
|
|
for cell in row[2:9]:
|
|
cell.number_format = "0.000000"
|
|
|
|
|
|
def build_sensor_placement_workbook(
|
|
*,
|
|
network: str,
|
|
scheme: dict[str, Any],
|
|
sensor_location: list[str],
|
|
adjustment_status: dict[str, str],
|
|
) -> BytesIO:
|
|
locations = _normalize_locations(sensor_location)
|
|
points = _sensor_points(network, locations)
|
|
is_draft = locations != list(scheme["sensor_location"])
|
|
|
|
workbook = Workbook()
|
|
info_sheet = workbook.active
|
|
info_sheet.title = "方案信息"
|
|
_populate_info_sheet(
|
|
info_sheet,
|
|
network=network,
|
|
scheme=scheme,
|
|
location_count=len(locations),
|
|
is_draft=is_draft,
|
|
)
|
|
|
|
list_sheet = workbook.create_sheet("监测点清单")
|
|
_populate_list_sheet(
|
|
list_sheet,
|
|
points=points,
|
|
adjustment_status=adjustment_status,
|
|
)
|
|
|
|
output = BytesIO()
|
|
workbook.save(output)
|
|
output.seek(0)
|
|
return output
|