Aislo/B08_Quantity/B08_Quantity_Engine_Pipe.py

"""배수관 물량 — **정본 셋을 잇기만 한다** (2026-09-08 두 창 합의).

값이 어디서 오나
    관 자체(있나·어디·관경·관종)  → `pipe_points.json`   (레지스트리 `pipe` 타입이
                                    `managed_by: pipe_points`)
    관 연장(m)                    → 측점 `design.pipe_length_m`
                                    (B06 횡단이 **서버 Node 로** 계산해 m 단위 올림까지
                                     끝낸 값을 정본에 남긴다 — 계산이 두 벌이 아니다)
    관종 → 공종코드                → `work_item_mapping` 의 `pipe.kind_codes`

⚠ **여기서 길이를 짓지 않는다.** 앞서 「도로폭 = 관 길이」처럼 잡을 뻔했는데 그것이 곧
  임의 수치다. 연장이 없는 관은 **줄을 세우되 막힌 사유와 함께** 보낸다.

⚠ **`facility` 가 `pipe` 인 점만 배관이다.**
  `pipe_points.json` 은 **계곡 통과 시설 전부의 정본**이라 BOX암거·물넘이·세월교·독립
  기슭막이가 같은 파일에 있다. `pipe_diameter_mm` 유무로 가르면 **관경 미지정 관을 놓친다**
  (실측: `5601e828` 11점 중 2점이 `facility: ford_bridge`).

⚠ **유출·유입부 기슭막이는 여기서 세지 않는다.**
  관 옵션(`outlet_revet_*`)이 정본이고 구조물 목록에서는 빠졌다(2026-08-28 이관).
  구조물 쪽으로 또 세면 이중계상이다.

⚠ **터파기·되메우기를 관 줄에 붙이지 않는다.**
  관 부설과 굴착이 각각 오면 **같은 굴착을 두 번** 센다(B09 ㉡ 가드와 같은 자리).
"""

from __future__ import annotations

from typing import Any

#: 배관으로 보는 `facility` 값. 그 밖(BOX암거·물넘이·세월교·독립 기슭막이)은 관이 아니다.
FACILITY_PIPE = "pipe"

#: 연장을 못 찾은 줄의 막힌 갈래 — 「입력하면 풀림」이 아니라 **앞 단계가 내야 하는 값**이다.
BLOCKED_LENGTH_MISSING = "input_missing"

#: ⚠ **둘을 갈라 말한다** — 「[저장]을 누르면 풀리는 것」과 「눌러도 안 풀리는 것」은
#: 사용자가 할 일이 다르다. 뒤엣것에 앞 문구를 쓰면 눌러 보고 안 되어 헤맨다
#: (2026-09-08 실측: 관 9개 중 3개가 **횡단 행 자체가 없는** 자리였다).
NOTE_LENGTH_MISSING = (
    "관 연장이 아직 정본에 없습니다 — 횡단설계에서 [저장]을 한 번 누르면 "
    "그 측점의 관 길이가 남고 값이 섭니다"
)
NOTE_SECTION_MISSING = (
    "그 측점의 횡단 자체가 없습니다 — 관은 놓였는데 횡단이 안 만들어진 자리라 "
    "[저장]으로는 안 풀립니다. 횡단설계에서 그 측점이 서야 합니다"
)
#: 관 자리에 횡단이 있는지 볼 때의 허용 오차(m).
#: ⚠⚠ **0.05 → 0.5 로 넓혔다**(2026-09-09 실측으로 앞의 판단이 뒤집힌 자리).
#:   측점을 만들 때 **정수 미터가 같은 격자 측점이 있으면 그리로 스냅**한다(횡단 파일명이
#:   정수 미터라 두 측점이 한 파일을 덮어쓰는 것을 막는 가드). 그래서 관 440.241 의 횡단은
#:   **측점 440.0** 이고, 그 측점은 구조물 이름표까지 달고 있다 — **없는 것이 아니다.**
#:   좁게 보면 「그 측점의 횡단 자체가 없습니다」라는 **거짓 사유**가 뜬다(반대 방향의 거짓).
#:   스냅 폭이 「정수 미터 반올림」이라 최대 어긋남이 0.5m 이고, 길이 찾기와 같은 값이 된다.
SECTION_MATCH_TOLERANCE_M = 0.5

NOTE_KIND_DEFAULT = "관종을 안 정해 기본값({kind})으로 섰습니다 — 정하면 공종이 갈립니다"
NOTE_KIND_UNKNOWN = "「{kind}」은(는) 아는 관종이 아니라 공종을 못 골랐습니다"


def _length_by_chainage(designs: list[dict[str, Any]], key: str) -> dict[float, float]:
    """측점별 관 길이. **없는 측점은 담지 않는다** — 0 으로 채우면 「없음」과 구별이 안 된다."""
    found: dict[float, float] = {}
    for row in designs or []:
        design = row.get("design") if isinstance(row, dict) else None
        if not isinstance(design, dict):
            continue
        value = design.get(key)
        if value is None:
            continue
        try:
            length = float(value)
        except (TypeError, ValueError):
            continue
        if length > 0:
            found[round(float(row.get("chainage_m") or 0.0), 3)] = length
    return found


def _nearest(lengths: dict[float, float], chainage: float, tolerance: float = 0.5) -> float | None:
    """관 측점과 단면 측점이 소수점에서 어긋날 수 있어 **가까운 것**을 본다.

    ⚠ 좁게 본다(기본 0.5m) — 넓히면 옆 측점의 길이를 물어 와 조용히 틀린다.
    """
    if not lengths:
        return None
    key = round(chainage, 3)
    if key in lengths:
        return lengths[key]
    best = min(lengths, key=lambda x: abs(x - chainage))
    return lengths[best] if abs(best - chainage) <= tolerance else None


def build_rows(
    pipe_points: list[dict[str, Any]],
    designs: list[dict[str, Any]],
    mapping: dict[str, Any] | None = None,
    section_chainages: list[float] | None = None,
) -> dict[str, Any]:
    """관 줄 목록. **값이 서는 줄도, 못 서는 줄도** 함께 낸다.

    `pipe_points` 는 `PipePoint.model_dump()` 또는 같은 모양의 딕셔너리 목록이다.
    """
    table = mapping or {}
    kind_codes: dict[str, str] = table.get("kind_codes") or {}
    kind_key = str(table.get("kind_option_key") or "pipe_kind")
    default_kind = str(table.get("default_kind") or "")
    length_key = str(table.get("length_key") or "pipe_length_m")
    diameter_key = str(table.get("diameter_option_key") or "pipe_diameter_mm")

    lengths = _length_by_chainage(designs, length_key)
    # 횡단이 **있는데 길이가 없는 것**과 **관 자리에 횡단이 없는 것**을 가르기 위한 목록.
    #   안 주면 종전대로 「[저장]하면 풀림」 하나로만 말한다.
    sections = [float(x) for x in (section_chainages or [])]
    rows: list[dict[str, Any]] = []
    notes: list[str] = []

    for point in pipe_points or []:
        if str(point.get("facility") or FACILITY_PIPE) != FACILITY_PIPE:
            continue  # 배관이 아닌 시설 — 그쪽 줄은 그쪽이 센다
        options = point.get("options") or {}
        chainage = float(point.get("chainage_m") or 0.0)

        stored_kind = str(options.get(kind_key) or "").strip()
        kind = stored_kind or default_kind
        code = kind_codes.get(kind)
        kind_note = ""
        if not stored_kind and default_kind:
            kind_note = NOTE_KIND_DEFAULT.format(kind=default_kind)
        elif stored_kind and code is None:
            kind_note = NOTE_KIND_UNKNOWN.format(kind=stored_kind)

        length = _nearest(lengths, chainage)
        # ⚠ 관이 선 자리의 횡단이 있는지를 본다. **스냅을 셈에 넣는다**(2026-09-09) —
        #   관 440.241 의 횡단은 측점 440.0 이고 그것이 정상이다. 옛 주석은 0.05m 로 좁게
        #   보라고 했으나, 그때는 **스냅 때문에 관이 측점에 안 붙던 것**을 「측점이 없다」로
        #   읽던 시절이라 판단이 뒤집혔다. 길이는 여전히 **주인 측점 하나에만** 실린다.
        has_section = not sections or any(
            abs(x - chainage) <= SECTION_MATCH_TOLERANCE_M for x in sections
        )
        blocked_kind = None if length else BLOCKED_LENGTH_MISSING
        blocked_reason = ""
        if not length:
            blocked_reason = NOTE_LENGTH_MISSING if has_section else NOTE_SECTION_MISSING
        if code is None:
            blocked_kind = blocked_kind or BLOCKED_LENGTH_MISSING
            blocked_reason = blocked_reason or kind_note

        rows.append(
            {
                "chainage_m": chainage,
                "work_item_code": code,
                "kind": kind,
                "kind_from_default": not stored_kind,
                # 갈래는 **저장 원본값**만 보낸다 — 「…㎜ 이하」 구간 나누기는 원문을 읽는 쪽 몫.
                "variant_axis": diameter_key,
                "variant_value": options.get(diameter_key),
                "unit": str(table.get("unit") or "m"),
                "quantity": float(length or 0.0),
                "blocked_kind": blocked_kind,
                "blocked_reason": blocked_reason or kind_note,
                "in_bill": bool(length and code),
            }
        )
        if kind_note and kind_note not in notes:
            notes.append(kind_note)

    missing = sum(1 for row in rows if not row["in_bill"])
    if missing:
        notes.append(f"관 {len(rows)}개 중 {missing}개가 아직 값이 안 섭니다")
    return {
        "rows": rows,
        "notes": notes,
        "pipe_count": len(rows),
        "ready_count": sum(1 for row in rows if row["in_bill"]),
        "length_total_m": round(sum(row["quantity"] for row in rows if row["in_bill"]), 3),
    }