"""배수관 물량 — **정본 셋을 잇기만 한다** (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`). ⚠ **유출·유입부 기슭막이는 관 줄에 넣지 않는다 — `facility_structures` 가 원단위 전개로 한 번 셈.** 관 옵션(`outlet_revet_*`)이 정본이다(2026-08-28 이관). 관 부설(품셈 12-11 m당: 관·기초콘크리트· 거푸집)에 기슭막이 몫이 없어 겹치지 않는다. 2026-09-14 까지는 **어느 쪽도 안 세고 있었다**(A1). ⚠ **터파기·되메우기를 관 줄에 붙이지 않는다.** 관 부설과 굴착이 각각 오면 **같은 굴착을 두 번** 센다(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 = ( "그 측점의 횡단 자체가 없습니다 — 관은 놓였는데 횡단이 안 만들어진 자리라 " "[저장]으로는 안 풀립니다. 횡단설계에서 그 측점이 서야 합니다" ) # ── 허용 오차 둘 — **묻는 것이 다르다.** 한 자리에 모아 둔다 ──────────────── # · `LENGTH_MATCH_TOLERANCE_M` — 「이 관의 **길이를 어디서 가져오나**」. # 넉넉해도 된다. 값을 못 찾는 것보다 옆 측점 길이를 쓰는 편이 낫다. # · `SECTION_MATCH_TOLERANCE_M` — 「이 측점에 **횡단 설계가 있나**」. # # ⚠⚠ **둘 다 0.5 다. 좁히지 말 것** — 앞서 「좁아야 한다」던 판단이 실측으로 뒤집혔다 # (2026-09-09 랩탑 메인). 측점을 만들 때 **정수 미터가 같은 격자 측점이 있으면 그리로 # 스냅**한다(횡단 파일명이 정수 미터라 두 측점이 한 파일을 덮어쓰는 것을 막는 가드). # 그래서 관 440.241 의 횡단은 **측점 440.0** 이고 구조물 이름표까지 달고 있다 — # **없는 것이 아니다.** 좁게 보면 「그 측점의 횡단 자체가 없습니다」라는 **거짓 사유**가 # 뜬다(반대 방향의 거짓). 스냅 폭이 「정수 미터 반올림」이라 최대 어긋남이 0.5m 이고, # 그래서 길이 찾기와 **같은 값**이 된다 — 우연이 아니라 같은 까닭이다. # ⚠ 값이 같아졌다고 **하나로 합치지 말 것.** 묻는 것이 둘이라 근거도 둘이고, 한쪽 근거가 # 바뀌면 한쪽만 움직여야 한다. # ⚠ 두 값은 **증상**이지 병이 아니다. 병은 **관 자리에 측점이 없는 것**이고 # (측점을 만드는 자리는 B05 노선 [확정] 한 곳뿐 — 관을 나중에 놓거나 옮기면 안 생긴다, # 계획서 3-14), 그것이 고쳐지면 이 값들이 **아무 관도 안 건드린다.** LENGTH_MATCH_TOLERANCE_M = 0.5 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 = LENGTH_MATCH_TOLERANCE_M, ) -> float | None: """관 측점과 단면 측점이 소수점에서 어긋날 수 있어 **가까운 것**을 본다. ⚠ 좁게 본다(기본 0.5m) — 넓히면 옆 측점의 길이를 물어 와 조용히 틀린다. ⚠ 「횡단이 있나」를 재는 `SECTION_MATCH_TOLERANCE_M`(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), } INLET_BASIN = "집수정" #: 기슭막이 벽 칸 — 관·독립 기슭막이 모두 `{쪽}_revet_{칸}` · 독립의 옛 저장분은 `{칸}`(B06 `_revet_side`). REVET_FIELDS = ( ("form", "형태"), ("height_m", "높이"), ("length_m", "길이"), ("before_m", ""), ("after_m", ""), ) #: (저장 채널, 관에 딸린 줄 이름 꼬리, 독립 기슭막이 줄 이름 꼬리) REVET_ROLES = ( ("inlet", "유입부 기슭막이", "유입 칸 벽"), ("outlet", "유출부 기슭막이", "유출 칸 벽"), ) #: 관 벽 높이를 안 적었으면 **횡단도가 그리는 관경 기준 최소 높이**(B06 `pipe_wall_min_height_m`, #: TS 짝 `pipeWallMinPureHeight`)로 섬 — 등록부·구조물도·횡단도가 한 값(2026-09-14 브레인 판정). #: ⚠ 「최소」라 실제 설계값이 더 클 수 있어 **미확정·금액 밖은 그대로**. NOTE_DEFAULT_WALL = ( "⚠ 시설 지점에 {filled} 을 안 적어 기본값으로 섰음 — 미확정이라 금액에 안 들어감" ) #: 꼬리 — 관 벽은 횡단도와 같은 높이 · 독립 기슭막이는 아직 갈림(관경이 없음). NOTE_WALL_HEIGHT_PIPE = " · 높이는 횡단도와 같은 관경 기준 최소 높이(실제 설계값은 더 클 수 있음)" NOTE_WALL_HEIGHT_OWN = " · ⚠ 횡단도는 저장 높이가 없으면 근입 0.5m 만 그림 — 표 높이와 다름" #: 금액에서 뺄 까닭(2026-09-14 브레인 판정) — 줄·표·그림은 서되 내역 합에는 안 듦. UNCONFIRMED_WALL = "벽 치수({filled})가 시설 지점에 없어 기본값으로 섰음 — 적으면 금액이 섬" NOTE_SIDE_UNKNOWN = ( "⚠ 설치 측 「{side}」인데 두 칸(유입·유출) 값이 달라 어느 칸이 그쪽인지 서버가 못 가림" "(횡단 지형이 정함) — 값을 안 세움 · 두 칸을 같게 적으면 섬" ) def _blank(value: Any) -> bool: return value is None or value == "" def _num_or_zero(value: Any) -> float: try: return float(value) except (TypeError, ValueError): return 0.0 def _revet_values( options: dict[str, Any], role: str, defaults: dict[str, Any], legacy: bool, min_height_m: float | None = None, ) -> tuple[dict[str, Any], list[str]]: """벽 한쪽 제원과 기본값으로 채운 칸 — 등록부 기본값 · 관 벽 높이는 관경 기준 최소 높이.""" values: dict[str, Any] = {} filled: list[str] = [] for name, label in REVET_FIELDS: value = options.get(f"{role}_revet_{name}") if _blank(value) and legacy: value = options.get(name) if _blank(value): value = defaults.get(name if legacy else f"{role}_revet_{name}") if value is None and name == "height_m" and min_height_m is not None: value = min_height_m label = f"{label}(관경 기준)" if label and value is not None: filled.append(f"{label} {value}") values[name] = value return values, filled def _pipe_diameter_m(options: dict[str, Any], defaults: dict[str, Any]) -> float: """관경(m) — 저장값, 없으면 등록부 기본(B06 세트 스펙과 같은 채움).""" raw = options.get("pipe_diameter_mm") if _blank(raw): raw = defaults.get("pipe_diameter_mm") return _num_or_zero(raw) / 1000.0 or 1.0 def facility_structures(points: list[dict[str, Any]]) -> list[dict[str, Any]]: """계곡 통과 시설(`pipe_points.json`) → 원단위 전개가 읽는 구조물 줄 (A1, 2026-09-14). ⚠ 관 자체는 안 냄 — `build_rows` 가 관 연장으로 셈. ⚠ 안 적힌 벽 칸은 **등록부 기본값** — 그 사실을 줄 사유로 붙이고 `unconfirmed` 로 금액 합에서 뺌 (2026-09-14 브레인 판정 「줄은 서되 금액은 실제 값이 있을 때만」). ⚠ 집수정·기슭막이 터파기·되메우기는 전개 성분(`destination: earthwork`)이라 토공집계로만 감. """ from B05_Profile.B05_Profile_Structures_Schema import structure_type_map from B06_Section.B06_Section_Engine_Culvert import pipe_wall_min_height_m from B08_Quantity.B08_Quantity_Engine_StructureSummary import pipe_row_id from B08_Quantity.B08_Quantity_Engine_UnitQuantity import attachments_of, child_row types = structure_type_map() def defaults(type_id: str) -> dict[str, Any]: found = types.get(type_id) return {option.key: option.default for option in found.options} if found else {} rows: list[dict[str, Any]] = [] for point in points or []: facility = str(point.get("facility") or FACILITY_PIPE) chainage = float(point.get("chainage_m") or 0.0) options = dict(point.get("options") or {}) base = { "structure_id": pipe_row_id(point), "type_id": facility, "chainage_m": chainage, "start_m": point.get("start_m") if point.get("start_m") is not None else chainage, "end_m": point.get("end_m") if point.get("end_m") is not None else chainage, "options": options, } if facility not in (FACILITY_PIPE, "revetment"): rows.append(base) # BOX암거·물넘이·세월교 — 전개식·관측값이 없으면 사유로 섬 continue legacy = facility == "revetment" # 독립 기슭막이 — 벽 칸 밖의 제원(뒷길이 등)도 벽이 씀 wall_defaults = defaults(facility) roles = list(REVET_ROLES) inlet_notes: list[str] = [] if not legacy: parent = {**base, "type_id": "pipe"} children = [r for r in attachments_of(parent) if r["type_id"] != "pipe_inlet_basin"] inlet = str(options.get("inlet_type") or wall_defaults.get("inlet_type") or "") if inlet == INLET_BASIN: # 형식을 안 골랐어도 줄은 세움 — 관측표가 「형식을 골라야 섬」 사유를 냄. children.append(child_row(parent, "pipe_inlet_basin", "유입부 집수정")) roles = roles[1:] elif options.get("inlet_basin_form"): inlet_notes.append( f"⚠ 집수정 형식({options['inlet_basin_form']})이 적혀 있으나 유입구 구조가 " f"「{inlet}」라 집수정은 안 셈 — 유입구를 「집수정」으로 바꾸면 섬" ) rows.extend(children) walls = [] min_height = ( None if legacy else pipe_wall_min_height_m(_pipe_diameter_m(options, wall_defaults)) ) for role, pipe_label, own_label in roles: values, filled = _revet_values(options, role, wall_defaults, legacy, min_height) notes = list(inlet_notes) if role == "inlet" else [] if filled: tail = NOTE_WALL_HEIGHT_OWN if legacy else NOTE_WALL_HEIGHT_PIPE notes.append(NOTE_DEFAULT_WALL.format(filled=" · ".join(filled)) + tail) walls.append([role, own_label if legacy else pipe_label, values, notes, False, filled]) side = str(options.get("side") or "") if legacy else "" if side in ("좌", "우"): # 좌·우가 유입/유출 어느 칸인지는 **횡단 지형**이 정함 — 두 칸이 같을 때만 한 벽으로 셈. same = walls[0][2] == walls[1][2] walls = [[walls[0][0], f"{side} 벽", walls[0][2], walls[0][3], not same, walls[0][5]]] if not same: walls[0][3].append(NOTE_SIDE_UNKNOWN.format(side=side)) foundation = options.get("foundation" if legacy else "revet_foundation") kept = {k: v for k, v in options.items() if not k.startswith(("inlet_", "outlet_"))} for role, label, values, notes, withheld, filled in walls: before, after = _num_or_zero(values["before_m"]), _num_or_zero(values["after_m"]) row = child_row(base, "revetment", label, f"{role}_revet") row.update( start_m=chainage - before, end_m=chainage + after, options={**(kept if legacy else {}), **values, "foundation": foundation}, notes=notes, withheld=withheld, unconfirmed=UNCONFIRMED_WALL.format(filled=" · ".join(filled)) if filled else "", ) rows.append(row) return rows