Aislo/common_util/common_util_drainage_pipes.py

"""관 매설 지점 정본 저장소 (B04 관리자 화면 · B05 사용자 화면 공용).

해석 산출물(`01~03`)은 다시 돌리면 덮어써도 되지만, 사용자가 찍고 옮긴 관은 그러면 안 된다.
그래서 편집분만 `{배수유역 폴더}/edits/pipe_points.json`에 따로 남기고 두 화면이 같은 파일을
읽고 쓴다 — 관리자 화면에서 옮긴 관이 사용자 화면에서 다르게 보이면 안 되기 때문이다
(2026-08-01 사용자 지시).

위치는 좌표가 아니라 **누가거리(chainage_m)** 로 저장한다. 지면 필터나 지표면 모델을 바꾸면
종단 Z가 달라지지만 관이 놓인 자리는 그대로여야 하고, 그때는 세부유역만 다시 나누면 된다.
노선 자체가 바뀌면(`route_signature` 불일치) 기준이 사라지므로 전량 버리고 다시 만든다.
"""

from __future__ import annotations

import hashlib
import json
import logging
from dataclasses import dataclass
from pathlib import Path
from typing import Any

from B04_wf1_Surface.B04_wf1_Surface_Engine_Watershed_Export import drainage_dir
from common_util.common_util_json import atomic_write_json
from common_util.common_util_route_geometry import RouteVertex
from config.config_system import (
    DRAINAGE_DETAIL_FILENAME,
    DRAINAGE_EDITS_DIRNAME,
    DRAINAGE_PIPE_POINTS_FILENAME,
)

logger = logging.getLogger(__name__)

# 관이 그 자리에 있는 이유. 화면 마커 모양과 "자동/수동" 구분이 여기에 달려 있다.
PIPE_SOURCE_STREAM = "stream"  # 기본 관 — 도로 × 상류 세류선 교차점
PIPE_SOURCE_SPACING = "spacing"  # 자동 보충 — 관 최대 간격 규칙
PIPE_SOURCE_USER = "user"  # 수동 — 사용자가 우클릭으로 추가하거나 옮긴 관
_KNOWN_SOURCES = (PIPE_SOURCE_STREAM, PIPE_SOURCE_SPACING, PIPE_SOURCE_USER)


@dataclass
class PipePoint:
    """계획선 위 관 매설 지점 한 개."""

    chainage_m: float
    source: str = PIPE_SOURCE_USER

    def as_dict(self) -> dict[str, Any]:
        return {"chainage_m": round(float(self.chainage_m), 2), "source": self.source}


def edits_dir(stored_path: str) -> Path:
    return drainage_dir(stored_path) / DRAINAGE_EDITS_DIRNAME


def pipe_points_path(stored_path: str) -> Path:
    return edits_dir(stored_path) / DRAINAGE_PIPE_POINTS_FILENAME


def detail_basins_path(stored_path: str) -> Path:
    return drainage_dir(stored_path) / DRAINAGE_DETAIL_FILENAME


def route_signature(vertices: list[RouteVertex]) -> str:
    """노선이 바뀌었는지 판별할 지문. 정점 좌표를 0.01m로 끊어 해시한다.

    연장만 보면 노선이 통째로 옮겨져도 같은 값이 나온다. 좌표를 다 넣되 소수점을 끊어
    부동소수 잡음으로 지문이 흔들리지 않게 한다.
    """
    digest = hashlib.sha1(usedforsecurity=False)
    for vertex in vertices:
        digest.update(f"{vertex.x:.2f},{vertex.y:.2f};".encode())
    return f"{len(vertices)}-{digest.hexdigest()[:16]}"


def load_pipe_points(stored_path: str, signature: str) -> list[PipePoint] | None:
    """저장된 관 지점을 읽는다. 파일이 없거나 노선이 바뀌었으면 None(= 다시 만들어야 함)."""
    path = pipe_points_path(stored_path)
    if not path.exists():
        return None
    try:
        with path.open("r", encoding="utf-8") as file:
            document = json.load(file)
    except (OSError, json.JSONDecodeError):
        logger.warning("배수유역: 관 지점 파일을 읽지 못했습니다 (%s).", path)
        return None
    stored_signature = str(document.get("route_signature") or "")
    if stored_signature != signature:
        logger.info("배수유역: 노선이 바뀌어 저장된 관 지점을 버립니다 (%s).", path.name)
        return None
    return parse_pipe_points(document.get("points"))


def parse_pipe_points(values: Any) -> list[PipePoint]:
    """외부에서 들어온 관 목록(파일·요청 본문)을 정리한다. 누가거리 오름차순."""
    if not isinstance(values, list):
        return []
    points: list[PipePoint] = []
    for item in values:
        if isinstance(item, (int, float)):
            points.append(PipePoint(chainage_m=float(item)))
            continue
        if not isinstance(item, dict):
            continue
        chainage = item.get("chainage_m")
        if not isinstance(chainage, (int, float)):
            continue
        source = str(item.get("source") or PIPE_SOURCE_USER)
        points.append(
            PipePoint(
                chainage_m=float(chainage),
                source=source if source in _KNOWN_SOURCES else PIPE_SOURCE_USER,
            )
        )
    points.sort(key=lambda point: point.chainage_m)
    return points


def save_pipe_points(stored_path: str, signature: str, points: list[PipePoint]) -> int:
    """관 지점을 정본 파일에 쓴다. 저장된 개수를 돌려준다."""
    path = pipe_points_path(stored_path)
    path.parent.mkdir(parents=True, exist_ok=True)
    atomic_write_json(
        path,
        {
            "route_signature": signature,
            "points": [point.as_dict() for point in points],
        },
    )
    logger.info("배수유역: 관 지점 %d개를 저장했습니다 (%s).", len(points), path.name)
    return len(points)


def save_detail_basins(stored_path: str, features: list[dict[str, Any]]) -> Path:
    """세부유역을 GeoJSON으로 남긴다(파생물 — 관 지점만 있으면 언제든 다시 만든다)."""
    path = detail_basins_path(stored_path)
    path.parent.mkdir(parents=True, exist_ok=True)
    atomic_write_json(path, {"type": "FeatureCollection", "features": features})
    logger.info("배수유역: 세부유역 %d개를 저장했습니다 (%s).", len(features), path.name)
    return path