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_PreProcess.B04_PreProcess_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)

# 계곡 통과 시설 종류 (2026-08-17 컨테이너 병합). 같은 계곡 교차 지점에서 유량·지형에
# 따라 택일하는 관계라 별도 정본을 만들지 않고 관 지점에 종류만 얹는다 — 유역 계산은
# 기준점(chainage_m)만 읽으므로 어느 종류든 계산이 같다. 교량은 임도용이 아니라 없다
# (2026-08-17 사용자 확정).
PIPE_FACILITY_PIPE = "pipe"  # 배관(횡단배수관) — 기본
PIPE_FACILITY_BOX = "box_culvert"  # BOX암거
PIPE_FACILITY_FORD_PAVEMENT = "ford_pavement"  # 물넘이포장
PIPE_FACILITY_FORD_BRIDGE = "ford_bridge"  # 세월교
# 독립 기슭막이(2026-08-28 사용자) — 배관 없이 성토 사면에 세우는 벽. 배관 세트 경로를
# 그대로 태우되 관을 숨긴다(hidden_pipe). 수량은 관 정보를 빼고 벽만 센다.
PIPE_FACILITY_REVET = "revetment"  # 독립 기슭막이(관 숨김)
_KNOWN_FACILITIES = (
    PIPE_FACILITY_PIPE,
    PIPE_FACILITY_BOX,
    PIPE_FACILITY_FORD_PAVEMENT,
    PIPE_FACILITY_FORD_BRIDGE,
    PIPE_FACILITY_REVET,
)


@dataclass
class PipePoint:
    """계획선 위 계곡 통과 시설 한 개 (배관·BOX암거·물넘이·세월교).

    `chainage_m`가 기준점(계곡 교차, 종단 마킹 위치)이고, `start_m`/`end_m`는 유입·유출
    부속이 차지하는 앞뒤 구간이다. 구간 미지정(None)은 폭 0 — 자동 배치분의 기본값이며
    사용자가 필요할 때 벌린다. 상세 치수는 B06/B07 몫이라 여기에는 유무·종류 수준의
    `options`(예: 세월교 관 종류/크기/수량)만 둔다 (2026-08-17 사용자 확정).
    """

    chainage_m: float
    source: str = PIPE_SOURCE_USER
    facility: str = PIPE_FACILITY_PIPE
    start_m: float | None = None
    end_m: float | None = None
    options: dict[str, Any] | None = None

    def as_dict(self) -> dict[str, Any]:
        # 구 형식 저장분이 확장 필드 없이 그대로 다시 저장되도록 기본값은 생략한다.
        payload: dict[str, Any] = {
            "chainage_m": round(float(self.chainage_m), 2),
            "source": self.source,
        }
        if self.facility != PIPE_FACILITY_PIPE:
            payload["facility"] = self.facility
        if self.start_m is not None and self.end_m is not None:
            payload["start_m"] = round(float(self.start_m), 2)
            payload["end_m"] = round(float(self.end_m), 2)
        if self.options:
            payload["options"] = self.options
        return payload


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_span(item: dict[str, Any], chainage: float) -> tuple[float | None, float | None]:
    """시작·종료 구간을 정규화한다 — 기준점을 항상 품고, 뒤집힘은 바로잡는다."""
    raw_start, raw_end = item.get("start_m"), item.get("end_m")
    start = float(raw_start) if isinstance(raw_start, (int, float)) else None
    end = float(raw_end) if isinstance(raw_end, (int, float)) else None
    if start is None and end is None:
        return None, None
    values = [value for value in (start, end) if value is not None] + [chainage]
    return min(values), max(values)


# 보호공 부위별 기본값 (2026-08-17 사용자 확정) — 구 "없음" 저장분을 끌어올릴 때 쓴다.
_PROTECTION_FALLBACK = {"inlet": "돌붙임(찰)", "outlet": "돌붙임(메)"}


def _migrate_protection(options: dict[str, Any]) -> dict[str, Any]:
    """구 저장분의 보호공 키를 새 한 축으로 옮긴다 (2026-08-17 보호공 개편).

    개편 전에는 `*_pitching`(있음/없음)과 `*_pitching_finish`(찰/메) 두 축이었다.
    읽는 순간 `*_protection`(돌붙임(찰)/돌붙임(메)/도수로) 한 축으로 바꿔 두면 화면도
    수량도 옛 키를 알 필요가 없다. "없음"은 선택지가 사라졌으므로 부위 기본값으로
    올린다 — 물이 흐르는 자리라 보호공은 반드시 있다(사용자 확정).
    """
    for side, fallback in _PROTECTION_FALLBACK.items():
        legacy = options.pop(f"{side}_pitching", None)
        finish = options.pop(f"{side}_pitching_finish", None)
        area = options.pop(f"{side}_pitching_area_m2", None)
        if legacy is None and finish is None and area is None:
            continue
        if f"{side}_protection" not in options:
            options[f"{side}_protection"] = (
                f"돌붙임({finish})" if legacy == "있음" and finish in ("찰", "메") else fallback
            )
        if area is not None and f"{side}_protection_area_m2" not in options:
            options[f"{side}_protection_area_m2"] = area
    return options


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)
        facility = str(item.get("facility") or PIPE_FACILITY_PIPE)
        start, end = _parse_span(item, float(chainage))
        options = item.get("options")
        points.append(
            PipePoint(
                chainage_m=float(chainage),
                source=source if source in _KNOWN_SOURCES else PIPE_SOURCE_USER,
                facility=facility if facility in _KNOWN_FACILITIES else PIPE_FACILITY_PIPE,
                start_m=start,
                end_m=end,
                options=(
                    _migrate_protection(dict(options))
                    if isinstance(options, dict) and options
                    else None
                ),
            )
        )
    points.sort(key=lambda point: point.chainage_m)
    return points


def carry_facility_attributes(base: list[PipePoint], reference: list[PipePoint]) -> list[PipePoint]:
    """계산기를 거쳐 재구성된 목록에 시설 종류·구간·옵션을 되붙인다.

    세부유역 계산기는 chainage만 다루므로 계산에서 돌아온 목록은 전부 기본 배관이 된다.
    그대로 저장하면 사용자가 고른 세월교·BOX암거가 배관으로 되돌아간다. 좌표가 미세
    조정(스냅)될 수 있어 정확 일치가 없으면 가장 가까운 원본에서 승계한다 — 생성 사유를
    되붙이는 `_retag`(B04 라우터)과 같은 기준이다.
    """
    if not reference:
        return base
    by_key = {round(ref.chainage_m, 2): ref for ref in reference}
    for point in base:
        ref = by_key.get(round(point.chainage_m, 2))
        if ref is None:
            ref = min(reference, key=lambda item: abs(item.chainage_m - point.chainage_m))
        point.facility = ref.facility
        point.start_m = ref.start_m
        point.end_m = ref.end_m
        point.options = ref.options
    return base


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 clear_pipe_points(stored_path: str) -> bool:
    """저장된 관 지점과 그 파생물을 지운다. 하나라도 지웠으면 True.

    "초기화"는 화면만 되돌리는 것이 아니라 **저장분까지** 되돌린다 — 화면만 되돌리면 다시
    들어왔을 때 옛 관이 살아나 사용자가 초기화한 적 없는 상태를 보게 된다
    (2026-08-02 사용자 보고).
    """
    removed = False
    for path in (pipe_points_path(stored_path), detail_basins_path(stored_path)):
        try:
            path.unlink()
            removed = True
        except FileNotFoundError:
            continue
        except OSError:
            logger.warning("배수유역: 저장분을 지우지 못했습니다 (%s).", path)
    if removed:
        logger.info("배수유역: 관 지점 저장분을 초기화했습니다 (%s).", stored_path)
    return removed


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


# ── 횡단배수 최소 계획고 (2026-08-23 사용자 확정) ─────────────────────────────
# 계획 종단선의 변화점(PVI)은 배수 시설 자리다. 그 자리에서 계획고를 지반고와 같게
# 두면 시설이 들어갈 자리가 없다 — 시설 제원만큼 계획고를 들어 올려야 한다.
#   배수관 Ø1000  → 지반고 + 1.0(관경) + 0.5(토피) = +1.5
#   BOX암거 2×2   → 지반고 + 2.0(구체 높이) + 0.5(토피) = +2.5
#   세월교 Ø1000  → 지반고 + 1.0(관경) + 0.5(토피) + 0.5(물넘이 몫) = +2.0
#   물넘이포장    → 도로에 그대로 앉히는 시설이라 요구 여유 없음 (2026-08-23 사용자)
# 세월교는 배관을 여러 개 묶어 다리 형태로 만든 것이라 배수관과 같은 산식을 쓰되,
# 그 위에 물넘이가 얹히므로 0.5m를 더 얹는다. 토피 0.5m는 B06 배수관
# 엔진(`MIN_PIPE_COVER_M`)과 같은 값이다.
MIN_PIPE_COVER_M = 0.5
FORD_BRIDGE_EXTRA_M = 0.5
DEFAULT_PIPE_DIAMETER_MM = 1000.0
DEFAULT_BOX_HEIGHT_M = 2.0


def _positive(value: Any, fallback: float) -> float:
    try:
        parsed = float(value)
    except (TypeError, ValueError):
        return fallback
    return parsed if parsed > 0 else fallback


def facility_clearance_m(facility: str, options: dict[str, Any] | None) -> float:
    """시설이 요구하는 지반고 대비 최소 여유(m). 계획선·경고가 같이 쓰는 정본 산식."""
    values = options or {}
    if facility == PIPE_FACILITY_BOX:
        return _positive(values.get("body_height_m"), DEFAULT_BOX_HEIGHT_M) + MIN_PIPE_COVER_M
    if facility == PIPE_FACILITY_FORD_PAVEMENT:
        # 물넘이포장은 도로 위에 그대로 만든다 — 들어 올릴 이유가 없다.
        return 0.0
    if facility == PIPE_FACILITY_REVET:
        # 독립 기슭막이는 성토 사면에 세우는 벽 — 관이 없어 들어 올릴 여유가 필요 없다.
        return 0.0
    diameter_m = _positive(values.get("pipe_diameter_mm"), DEFAULT_PIPE_DIAMETER_MM) / 1000.0
    extra = FORD_BRIDGE_EXTRA_M if facility == PIPE_FACILITY_FORD_BRIDGE else 0.0
    return diameter_m + MIN_PIPE_COVER_M + extra


def pipe_anchor_clearances(points: list[PipePoint]) -> list[tuple[float, float]]:
    """계획선 변화점으로 쓸 (누가거리, 최소 여유) 목록. 누가거리 오름차순."""
    return sorted(
        (float(point.chainage_m), facility_clearance_m(point.facility, point.options))
        for point in points
    )