Aislo/B06_Section/B06_Section_Engine_Attach.py

"""형상 얹기 한 벌(323-4) — 측점(횡단 dict) + 현황(관 · 구조물) + 구조물 표 → 그 측점 그림 자료.

구조물 열쇠(`SHAPE_KEYS` · 빈칸 경고 `blank`)를 걷고 차례대로 다시 얹음 — 관 세트 → 벽 → 포장 → 맹암거 →
빈칸 경고. 고르기 = 표 형상(322-1). 서버 상세 읽기(`_read_section_detail`) · 재생성 응답이 이것 하나를 씀.
TS 짝 = `common_util/common_util_structure_attach.ts` `attachShapes`(화면 = 초안 ?? 저장분 현황) ·
거울 시험 = `resources/tester/test_323_4_attach_mirror.py`.
"""

from __future__ import annotations

import logging
import math
from pathlib import Path
from typing import Any

from B05_Profile.B05_Profile_Structures_Repository import load_structures
from B05_Profile.B05_Profile_Structures_Schema import structure_type_map
from B06_Section.B06_Section_Engine_Culvert import (
    attach_culvert_specs,
    culvert_sets_from,
    pipe_points_file,
)
from B06_Section.B06_Section_Engine_Structures_Wall import attach_wall_specs, wall_specs_from
from B06_Section.B06_Section_Engine_Tilt import attach_underdrain_specs, underdrains_from
from common_util.common_util_drainage_pipes import read_pipe_points_file
from common_util.common_util_structure_catalog import catalog_row, field_name, type_ids_with
from common_util.common_util_structure_pavement import (
    attach_pavement_specs,
    pavement_specs_from,
)
from common_util.common_util_structure_station import (
    PIPE_MATCH_M,
    SAME_STATION_M,
    blank_keys,
    plan_stations,
)

logger = logging.getLogger(__name__)

#: 측점 그림 자료 열쇠 — TS `common_util_structure_shape_keys.SHAPE_KEYS` 짝.
SHAPE_KEYS = ("culvert", "ford", "box", "ford_pavement", "revetment", "pavement", "underdrain")


def _blank_labels(type_id: str, options: dict | None, types: dict[str, Any]) -> list[str]:
    """빈 필수 칸 → 「종류 이름 칸 이름」(TS `blankLabels` 짝)."""
    name = (catalog_row(type_id) or {}).get("name") or type_id
    ordered = [types[type_id]] if type_id in types else []
    ordered += list(types.values())
    out = []
    for key in blank_keys(type_id, options):
        label = next(
            (o.label for t in ordered for o in t.options if o.key == key and o.label), None
        )
        out.append(f"{name} {label or field_name(key) or key}")
    return out


def _attach_blanks(sections: list[dict[str, Any]], pipes: list[Any], structures: list[Any]) -> None:
    """빈칸 경고 — 측점 계획 자리마다 덮는 관 · 구조물의 빈 필수 칸(TS `attachBlanks` 짝)."""
    types = structure_type_map()
    owners: dict[str, tuple[str, dict | None]] = {}
    for pipe in pipes:
        owner = getattr(pipe, "id", None) or f"pipe-{float(pipe.chainage_m):.2f}"
        owners[owner] = (pipe.facility or "pipe", pipe.options)
    for item in structures:
        if item.structure_id:
            owners[item.structure_id] = (item.type_id, item.options)
    for station in plan_stations(pipes, structures):
        if not station["blank"]:
            continue
        labels: list[str] = []
        for owner in station["owners"]:
            hit = owners.get(owner)
            for label in _blank_labels(hit[0], hit[1], types) if hit else []:
                if label not in labels:
                    labels.append(label)
        for section in sections:
            at = section.get("chainage_m")
            if isinstance(at, (int, float)) and abs(at - station["chainage_m"]) < SAME_STATION_M:
                section["blank"] = labels


def _finite(value: Any) -> bool:
    return isinstance(value, (int, float)) and math.isfinite(value)


def _outside_route(pipe: Any, sections: list[dict[str, Any]]) -> bool:
    """시종점 밖으로 밀린 관인가(458 · TS `outsideRoute` 짝) — 시종점 측점 자리인데 좌표가 노선 진행
    반대쪽으로 `PIPE_MATCH_M` 넘게 바깥(옆 거리는 안 봄 · 좌표 없으면 아님)."""
    placed = sorted(
        (
            s
            for s in sections
            if all(_finite(s.get(k)) for k in ("chainage_m", "center_x", "center_y"))
        ),
        key=lambda s: s["chainage_m"],
    )
    x, y = getattr(pipe, "x", None), getattr(pipe, "y", None)
    if len(placed) < 2 or not (_finite(x) and _finite(y)):
        return False
    for end, nxt in ((placed[0], placed[1]), (placed[-1], placed[-2])):
        if abs(end["chainage_m"] - float(pipe.chainage_m)) >= PIPE_MATCH_M:
            continue
        dx, dy = nxt["center_x"] - end["center_x"], nxt["center_y"] - end["center_y"]
        span = math.hypot(dx, dy)
        if (
            span > 0
            and ((x - end["center_x"]) * dx + (y - end["center_y"]) * dy) / span <= -PIPE_MATCH_M
        ):
            return True
    return False


def attach_shapes(
    sections: list[dict[str, Any]], pipes: list[Any], structures: list[Any]
) -> dict[str, list[float]]:
    """측점마다 그림 자료를 걷고 현황으로 다시 얹음(TS `attachShapes` 짝 · 제자리 고침) ·
    시종점 밖 관 · 주인 측점 먼 관은 안 얹고 돌려줌(458 · `{"outside", "far"}`)."""
    for section in sections:
        for key in (*SHAPE_KEYS, "blank"):
            section.pop(key, None)
    misses: dict[str, list[float]] = {"outside": [], "far": []}
    kept = []
    for pipe in pipes:
        if _outside_route(pipe, sections):
            misses["outside"].append(float(pipe.chainage_m))
        else:
            kept.append(pipe)
    pipes = kept
    attach_culvert_specs(sections, culvert_sets_from(pipes), misses["far"])
    # 관 세트가 선 측점은 벽을 안 얹음(`attach_wall_specs`) — 차례 = 화면과 같음
    attach_wall_specs(sections, wall_specs_from(structures))
    attach_pavement_specs(
        sections,
        pavement_specs_from([s.model_dump() for s in structures], type_ids_with(shape="pavement")),
    )
    attach_underdrain_specs(sections, underdrains_from(structures))
    _attach_blanks(sections, pipes, structures)
    if misses["outside"] or misses["far"]:
        logger.warning("B06 시설 그림 안 얹음 — 노선 밖 %s · 측점 멂 %s", *misses.values())
    return misses


def saved_pipes(project_root: Path, route_data_path: str | None = None) -> list[Any]:
    """관 지점 정본 — 노선 경로를 주면 그 노선에 얹어 읽음(472 덧) · 없으면 파일 그대로.

    측점 계획(`_load_pipe_points`) · 관 지점 GET · 화면과 같은 이월 — 지문이 어긋난 옛 저장분도
    측점이 선 자리에 관 세트가 얹힘."""
    if route_data_path:
        from B05_Profile.B05_Profile_Engine_Sections import (
            _load_pipe_points,
            _load_route_polyline,
        )

        try:
            return _load_pipe_points(
                project_root, _load_route_polyline(project_root, route_data_path)
            )
        except (FileNotFoundError, ValueError):
            logger.warning("B06 관 지점: 노선을 못 읽어 저장분 그대로 얹음 (%s)", route_data_path)
    return read_pipe_points_file(pipe_points_file(project_root))


def attach_saved_shapes(
    project_root: Path, sections: list[dict[str, Any]], route_data_path: str | None = None
) -> None:
    """저장분 현황(관 지점 · 구조물 정본)으로 얹음 — 관 = 그 노선에 얹은 것(`saved_pipes`) ·
    구조물 정본을 못 읽으면 관만(횡단 조회는 이어 감)."""
    pipes = saved_pipes(project_root, route_data_path)
    try:
        structures = load_structures(str(project_root))[1]
    except Exception:  # noqa: BLE001 — 정본을 못 읽어도 횡단 조회는 이어 간다
        logger.exception("B06 구조물 정본을 읽지 못했습니다 (관만 얹음)")
        structures = []
    attach_shapes(sections, pipes, structures)