Aislo/common_util/common_util_stream_crossing.py

"""세류선 교차점 · 자동 세류선 관 라벨(507-2).

교차점 = 노선 평면 선형 × 세류선(`find_stream_crossings`) — 지물 id(`stream_id`)를 함께 단다.

자동 세류선 관 라벨(`PipePoint.stream_label`) = `{stream, order, x, y}` — 세류선 지물 id · 그
세류선의 노선 위 교차 차례 · 처음 교차점 좌표. 끌거나 옵션을 고쳐 `source` 가 바뀌어도 남음.

  - 노선 바뀐 [저장](`B05_Profile_Route_Save.followed_pipes`) — 라벨(관 · 지움 표지)마다 같은
    세류선의 새 교차점 중 라벨 좌표와 가장 가까운 것을 짝으로 잡음 · 짝 잡힌 교차점엔 관 안 더함.
  - 지움 표지(`pipe_points.json` `dismissed_streams`) = 사용자가 지운 자동 관 라벨 — 다시 안 생김.
    라벨 없는 자동 관을 지우면 그 자리(x · y)만 남겨 두고 다음 노선 [저장] 때 라벨로 바꿈.
  - 라벨 없는 옛 관(`source` stream) = 노선 [저장] 때 옛 노선 교차점 5 m 안이면 그 교차 라벨 ·
    교차 하나에 관 하나(가장 가까운 관) · 나머지 겹친 관은 지우지 않고 수만 셈.
  - 추천 [적용] · [초기화] = 초기값 파일 그대로(지움 표지도 초기값).
"""

from __future__ import annotations

import math
from typing import Any, Sequence

from shapely.geometry import LineString, Point, shape

from common_util.common_util_route_geometry import RouteVertex, StructureCandidate
from config.config_system import DRAINAGE_PIPE_MIN_SPACING_M


def stream_key(feature: dict[str, Any], index: int) -> str:
    """세류선 지물 id — 도엽 UFID · 없으면 피처 id · 그것도 없으면 파일 안 차례."""
    properties = feature.get("properties") or {}
    value = properties.get("UFID") or feature.get("id")
    return str(value) if value not in (None, "") else f"#{index}"


def find_stream_crossings(
    vertices: list[RouteVertex],
    stream_features: list[dict[str, Any]],
) -> list[StructureCandidate]:
    """노선 평면 선형과 세류선의 교차 지점을 누가거리 순으로 찾는다."""
    if len(vertices) < 2:
        return []
    route_line = LineString([(vertex.x, vertex.y) for vertex in vertices])
    candidates: list[StructureCandidate] = []
    for index, feature in enumerate(stream_features):
        geometry = feature.get("geometry")
        if not geometry:
            continue
        try:
            stream = shape(geometry)
        except Exception:  # noqa: BLE001 - 손상된 피처는 건너뛴다
            continue
        if stream.is_empty:
            continue
        intersection = route_line.intersection(stream)
        if intersection.is_empty:
            continue
        name = _stream_name(feature)
        for point in _collect_points(intersection):
            candidates.append(
                StructureCandidate(
                    chainage_m=route_line.project(point),
                    x=point.x,
                    y=point.y,
                    reason="stream",
                    stream_name=name,
                    stream_id=stream_key(feature, index),
                )
            )
    candidates.sort(key=lambda item: item.chainage_m)
    return candidates


def _stream_name(feature: dict[str, Any]) -> str | None:
    properties = feature.get("properties") or {}
    for key in ("명칭", "하천명", "NAME", "name"):
        value = properties.get(key)
        if value:
            return str(value)
    return None


def _collect_points(geometry: Any) -> list[Point]:
    """교차 결과(Point/MultiPoint/LineString 등)에서 대표 점들을 뽑는다."""
    if geometry.geom_type == "Point":
        return [geometry]
    if geometry.geom_type in {"MultiPoint", "GeometryCollection"}:
        points: list[Point] = []
        for part in geometry.geoms:
            points.extend(_collect_points(part))
        return points
    # 선분끼리 겹쳐 선으로 나온 경우는 중점을 대표로 쓴다.
    if geometry.geom_type in {"LineString", "MultiLineString"}:
        return [geometry.interpolate(0.5, normalized=True)]
    return []


# ── 자동 세류선 관 라벨(507-2) ────────────────────────────────────────────────


def crossing_labels(crossings: Sequence[Any]) -> list[dict[str, Any]]:
    """교차점(누가거리 순)마다 라벨 — 차례 = 같은 세류선 안에서 몇 번째 교차인가."""
    seen: dict[Any, int] = {}
    labels = []
    for crossing in crossings:
        stream = getattr(crossing, "stream_id", None)
        seen[stream] = seen.get(stream, 0) + 1
        labels.append(
            {
                "stream": stream,
                "order": seen[stream],
                "x": round(float(crossing.x), 3),
                "y": round(float(crossing.y), 3),
            }
        )
    return labels


def _gap(label: dict[str, Any], x: float, y: float) -> float:
    return math.hypot(float(label["x"]) - x, float(label["y"]) - y)


def claim_crossings(labels: Sequence[dict[str, Any]], crossings: Sequence[Any]) -> set[int]:
    """라벨마다 같은 세류선 교차점 중 라벨 좌표와 가장 가까운 것 하나 — 잡힌 교차점 번호.

    가까운 짝부터 하나씩(라벨 하나 · 교차점 하나) · 거리 한계 없음.
    """
    pairs = sorted(
        (_gap(label, float(crossing.x), float(crossing.y)), li, ci)
        for li, label in enumerate(labels)
        if "stream" in label
        for ci, crossing in enumerate(crossings)
        if getattr(crossing, "stream_id", None) == label["stream"]
    )
    used_labels: set[int] = set()
    used: set[int] = set()
    for _, li, ci in pairs:
        if li in used_labels or ci in used:
            continue
        used_labels.add(li)
        used.add(ci)
    return used


def adopt_labels(
    pipes: Sequence[Any],
    dismissed: list[dict[str, Any]],
    crossings: Sequence[Any],
    old_xy: Sequence[Sequence[float]],
) -> int:
    """옛 노선 교차점 중 라벨(관 · 지움 표지)이 안 잡은 것 — 옛 노선 누가거리 5 m 안 라벨 없는
    stream 관(가장 가까운 것) · 없으면 자리만 남은 지움 표지에 그 교차 라벨(제자리) · 돌려줌 =
    겹쳐 라벨 못 받은 stream 관 수.

    거리 = 옛 노선에 투영한 누가거리 차(옛 관 좌표가 다른 선 위에 남은 저장분 · 옆 15 m 실측).
    """
    from common_util.common_util_drainage_pipes import PIPE_SOURCE_STREAM

    line = LineString(old_xy)
    labels = crossing_labels(crossings)
    held = [pipe.stream_label for pipe in pipes if pipe.stream_label] + [
        mark for mark in dismissed if "stream" in mark
    ]
    claimed = claim_crossings(held, crossings)
    loose = [
        (line.project(Point(pipe.x, pipe.y)), pipe)
        for pipe in pipes
        if not pipe.stream_label and pipe.source == PIPE_SOURCE_STREAM and pipe.x is not None
    ]
    spots = [
        (line.project(Point(float(mark["x"]), float(mark["y"]))), mark)
        for mark in dismissed
        if "stream" not in mark
    ]

    def nearest(items: list, at: float) -> int | None:
        gaps = [(abs(chainage - at), order) for order, (chainage, _) in enumerate(items)]
        best = min(gaps, default=None)
        return best[1] if best and best[0] <= DRAINAGE_PIPE_MIN_SPACING_M else None

    for index, (label, crossing) in enumerate(zip(labels, crossings)):
        if index in claimed:
            continue
        at = float(crossing.chainage_m)
        pick = nearest(loose, at)
        if pick is not None:
            loose.pop(pick)[1].stream_label = dict(label)
            continue
        pick = nearest(spots, at)
        if pick is not None:
            spots.pop(pick)[1].update(label)
    near = [float(crossing.chainage_m) for crossing in crossings]
    return sum(
        1 for chainage, _ in loose if nearest([(c, None) for c in near], chainage) is not None
    )


def keep_labels(
    points: Sequence[Any], stored: Sequence[Any], dismissed: list[dict[str, Any]]
) -> None:
    """관 저장(PUT) — 화면이 모르는 라벨을 정본에서 id 로 되붙임 · 지운 자동 관 = 지움 표지(제자리).

    라벨 있는 관 = 라벨 · 라벨 없는 stream 관 = 그 자리(x · y · 다음 노선 [저장] 때 라벨로).
    """
    from common_util.common_util_drainage_pipes import PIPE_SOURCE_STREAM

    by_id = {pipe.id: pipe for pipe in stored if pipe.id}
    for point in points:
        old = by_id.get(point.id)
        if not point.stream_label and old is not None and old.stream_label:
            point.stream_label = dict(old.stream_label)
    kept = {point.id for point in points}
    for pipe in stored:
        if pipe.id in kept:
            continue
        if pipe.stream_label:
            mark = dict(pipe.stream_label)
        elif pipe.source == PIPE_SOURCE_STREAM and pipe.x is not None and pipe.y is not None:
            mark = {"x": round(float(pipe.x), 3), "y": round(float(pipe.y), 3)}
        else:
            continue
        if mark not in dismissed:
            dismissed.append(mark)