diff --git a/B08_Quantity/B08_Quantity_Engine_Handoff_Rows.py b/B08_Quantity/B08_Quantity_Engine_Handoff_Rows.py index b0fe9ba5..68adef0a 100644 --- a/B08_Quantity/B08_Quantity_Engine_Handoff_Rows.py +++ b/B08_Quantity/B08_Quantity_Engine_Handoff_Rows.py @@ -378,6 +378,11 @@ def _structure_rows( "큰돌쌓기 쌓기 방식이 아직 입력되지 않았습니다 — 구조물 상세 입력에서 " "메쌓기·찰쌓기 중 하나를 고르면 공종이 정해집니다" ) + if entry.get("class_from") == "form": + # 기슭막이 — **형태**가 공종을 가름(돌쌓기 찰·메). 그 밖 형태는 전개 사유가 막음. + form = str((structure.get("options") or {}).get("form") or "").strip() + code = (entry.get("form_codes") or {}).get(form) + class_basis = f"형태 「{form}」 → {code}" if code else "" if code and entry.get("class_from") == "back_length": class_key, class_basis = masonry_class(structure.get("options") or {}) diff --git a/B08_Quantity/B08_Quantity_Engine_Handoff_Trench.py b/B08_Quantity/B08_Quantity_Engine_Handoff_Trench.py index 7acc15f2..55ccbc65 100644 --- a/B08_Quantity/B08_Quantity_Engine_Handoff_Trench.py +++ b/B08_Quantity/B08_Quantity_Engine_Handoff_Trench.py @@ -243,7 +243,8 @@ def structure_earthwork_rows( backfill += amount elif name == "잔토처리": spoil += amount - if not any( + # 성분이 아예 없는 구조물(세월교 등 산출식 없음)은 제 줄이 이미 사유로 막힘 — 여기 안 적음. + if structure.get("components") and not any( str(component.get("name") or "") == "터파기" and float(component.get("amount") or 0.0) > 0 for component in structure.get("components") or [] diff --git a/B08_Quantity/B08_Quantity_Engine_Pipe.py b/B08_Quantity/B08_Quantity_Engine_Pipe.py index 812b3649..e0b5ca9e 100644 --- a/B08_Quantity/B08_Quantity_Engine_Pipe.py +++ b/B08_Quantity/B08_Quantity_Engine_Pipe.py @@ -16,9 +16,9 @@ 기슭막이가 같은 파일에 있다. `pipe_diameter_mm` 유무로 가르면 **관경 미지정 관을 놓친다** (실측: `5601e828` 11점 중 2점이 `facility: ford_bridge`). -⚠ **유출·유입부 기슭막이는 여기서 세지 않는다.** - 관 옵션(`outlet_revet_*`)이 정본이고 구조물 목록에서는 빠졌다(2026-08-28 이관). - 구조물 쪽으로 또 세면 이중계상이다. +⚠ **유출·유입부 기슭막이는 관 줄에 넣지 않는다 — `facility_structures` 가 원단위 전개로 한 번 셈.** + 관 옵션(`outlet_revet_*`)이 정본이다(2026-08-28 이관). 관 부설(품셈 12-11 m당: 관·기초콘크리트· + 거푸집)에 기슭막이 몫이 없어 겹치지 않는다. 2026-09-14 까지는 **어느 쪽도 안 세고 있었다**(A1). ⚠ **터파기·되메우기를 관 줄에 붙이지 않는다.** 관 부설과 굴착이 각각 오면 **같은 굴착을 두 번** 센다(B09 ㉡ 가드와 같은 자리). @@ -192,3 +192,136 @@ def build_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", "유출부 기슭막이", "유출 칸 벽"), +) +NOTE_DEFAULT_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 +) -> tuple[dict[str, Any], list[str]]: + """벽 한쪽 제원과 등록부 기본값으로 채운 칸 — B06 횡단 그림(`_side_spec`)과 같은 채움.""" + 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 label and value is not None: + filled.append(f"{label} {value}") + values[name] = value + return values, filled + + +def facility_structures(points: list[dict[str, Any]]) -> list[dict[str, Any]]: + """계곡 통과 시설(`pipe_points.json`) → 원단위 전개가 읽는 구조물 줄 (A1, 2026-09-14). + + ⚠ 관 자체는 안 냄 — `build_rows` 가 관 연장으로 셈. + ⚠ 안 적힌 벽 칸은 **등록부 기본값** — 횡단 그림이 쓰는 값과 같고, 그 사실을 줄 사유로 붙임. + ⚠ 집수정·기슭막이 터파기·되메우기는 전개 성분(`destination: earthwork`)이라 토공집계로만 감. + """ + from B05_Profile.B05_Profile_Structures_Schema import structure_type_map + 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 = [] + for role, pipe_label, own_label in roles: + values, filled = _revet_values(options, role, wall_defaults, legacy) + notes = list(inlet_notes) if role == "inlet" else [] + if filled: + notes.append(NOTE_DEFAULT_WALL.format(filled=" · ".join(filled))) + walls.append([role, own_label if legacy else pipe_label, values, notes, False]) + 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]] + 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 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, + ) + rows.append(row) + return rows diff --git a/B08_Quantity/B08_Quantity_Engine_UnitQuantity.py b/B08_Quantity/B08_Quantity_Engine_UnitQuantity.py index b5769d65..dcd043c2 100644 --- a/B08_Quantity/B08_Quantity_Engine_UnitQuantity.py +++ b/B08_Quantity/B08_Quantity_Engine_UnitQuantity.py @@ -198,24 +198,26 @@ def wing_wall_double_count(options: dict[str, Any]) -> str | None: return WING_WALL_DOUBLE_COUNT if options.get("inlet_basin_form") else None +def child_row(structure: dict[str, Any], type_id: str, label: str, suffix: str = "") -> dict: + """딸린 줄 한 벌 — 이름에 「어디에 딸렸나」가 남음(`expand` 가 부모 이름 · 꼬리로 붙임).""" + return { + **structure, + "structure_id": f"{structure.get('structure_id')}-{suffix or type_id}", + "type_id": type_id, + "attachment_of": structure.get("structure_id"), + "attachment_parent_type": structure.get("type_id"), + "attachment_label": label, + } + + def attachments_of(structure: dict[str, Any]) -> list[dict[str, Any]]: """구조물에 딸린 **별도 줄**을 만든다. 제원은 원본을 그대로 물려준다(치수 두 벌 금지).""" - rows: list[dict[str, Any]] = [] options = structure.get("options") or {} - for type_id, gate_key, label in ATTACHMENTS.get(str(structure.get("type_id") or ""), ()): - if not options.get(gate_key): - continue # 그 부속이 없는 배치다 — 빈 줄을 만들지 않는다 - rows.append( - { - **structure, - "structure_id": f"{structure.get('structure_id')}-{type_id}", - "type_id": type_id, - "attachment_of": structure.get("structure_id"), - "attachment_parent_type": structure.get("type_id"), - "attachment_label": label, - } - ) - return rows + return [ + child_row(structure, type_id, label) + for type_id, gate_key, label in ATTACHMENTS.get(str(structure.get("type_id") or ""), ()) + if options.get(gate_key) # 그 부속이 없는 배치다 — 빈 줄을 만들지 않는다 + ] def _observed_components( @@ -288,6 +290,10 @@ def expand( end_m=end if structure.get("end_m") is not None else None, options=dict(options), ) + # 줄을 만든 쪽이 붙인 사유(관 지점 시설의 기본값 · 못 가른 자리) — `withheld` 면 값을 안 세움. + result.notes.extend(str(note) for note in structure.get("notes") or []) + if structure.get("withheld"): + return result withheld = EXPANDER_WITHHELD.get(type_id) if withheld: result.notes.append(f"전개식 미확보 — {withheld}") diff --git a/B08_Quantity/B08_Quantity_Router_Material.py b/B08_Quantity/B08_Quantity_Router_Material.py index acfb3708..2b767788 100644 --- a/B08_Quantity/B08_Quantity_Router_Material.py +++ b/B08_Quantity/B08_Quantity_Router_Material.py @@ -58,7 +58,17 @@ router = APIRouter(prefix="/api/projects", tags=["B08 Quantity"]) def _collect_structures( project_root: str, ) -> tuple[list[dict[str, Any]], dict[str, str], list[str]]: - """전개 대상 구조물·타입명·건너뛴 사유를 함께 낸다.""" + """전개 대상 구조물·타입명·건너뛴 사유를 함께 낸다. + + ⭐ 정본 둘을 다 읽음(A1, 2026-09-14) — `structures.json` + `pipe_points.json`(계곡 통과 시설). + 앞서 뒤엣것을 안 읽어 집수정·날개벽·관 유입/유출 기슭막이·독립 기슭막이·물넘이포장이 + 원단위·인계·내역에 한 줄도 안 섰음. + ⚠ 관 지점 종류(`managed_by`)가 `structures.json` 에 옛 저장분으로 남아 있어도 안 셈 — + 관 지점 정본이 주인이라 두 번 세지 않음(구조물 집계표와 같은 규칙). + """ + from B08_Quantity.B08_Quantity_Engine_Pipe import facility_structures + from common_util.common_util_drainage_pipes import pipe_points_path_in, read_pipe_points_file + _revision, items = load_structures(project_root) types = structure_type_map() targets: list[dict[str, Any]] = [] @@ -72,6 +82,11 @@ def _collect_structures( skipped.append(f"{type_id}: 레지스트리에 없는 타입") continue names[type_id] = definition.name + if definition.managed_by: + skipped.append( + f"{definition.name}: 관 지점 정본이 주인 — 구조물 목록 옛 저장분은 안 셈" + ) + continue if definition.design_owner: skipped.append( f"{definition.name}: {definition.design_owner} 가 이미 셈 — 중복 계상 방지" @@ -86,6 +101,12 @@ def _collect_structures( # 여기서 빼지 않으면 **같은 시설이 두 줄로** 나간다. continue targets.append(payload) + points = read_pipe_points_file(pipe_points_path_in(Path(project_root))) + for row in facility_structures([point.as_dict() for point in points]): + for type_id in (row["type_id"], row.get("attachment_parent_type")): + if type_id in types: + names[type_id] = types[type_id].name + targets.append(row) return targets, names, sorted(set(skipped)) diff --git a/B08_Quantity/B08_Quantity_Router_StructureSheet.py b/B08_Quantity/B08_Quantity_Router_StructureSheet.py index e268d3c8..d47daa53 100644 --- a/B08_Quantity/B08_Quantity_Router_StructureSheet.py +++ b/B08_Quantity/B08_Quantity_Router_StructureSheet.py @@ -259,6 +259,13 @@ async def put_structure_sheet_spec( try: revision, stored = await asyncio.to_thread(load_structures, project_root) updated, changed = apply_spec(stored, member_ids, spec) + outside = member_ids - {str(item.structure_id) for item in stored} + if outside: + # 관 지점 시설(A1)은 `pipe_points.json` 이 정본 — 조용히 안 먹히지 않게 알림. + notes.append( + f"관 지점 시설 {len(outside)}개소(배수관 기슭막이 등)는 이 화면에서 제원을 못 적음" + " — B05 배수 시설·구조물 집계표에서 고칠 것" + ) new_revision = await asyncio.to_thread( save_structures, project_root, updated, base_revision=payload.base_revision ) diff --git a/resources/data_work_item_mapping/work_item_mapping_2026-01-01.json b/resources/data_work_item_mapping/work_item_mapping_2026-01-01.json index 86ceb393..7c163680 100644 --- a/resources/data_work_item_mapping/work_item_mapping_2026-01-01.json +++ b/resources/data_work_item_mapping/work_item_mapping_2026-01-01.json @@ -220,6 +220,21 @@ "work_item_code": "FP-12-15", "master_name": "집수정" }, + { + "type_id": "revetment", + "work_item_code": null, + "class_from": "form", + "form_codes": { + "돌쌓기(찰)": "FP-13-04-05", + "돌쌓기(메)": "FP-13-04-02" + }, + "class_note": "기슭막이는 **형태가 돌쌓기면 돌쌓기 식**(`_UnitQuantity_Revetment`) — 공종도 돌쌓기 찰·메와 같음(실무 정본 탭 「돌기슭막이(H=2.0m, 찰쌓기, 기초유)」). 그 밖 형태(콘크리트·돌망태·통나무·바자)는 전개식이 없어 사유로 막힘. 관 유입·유출부 기슭막이와 독립 기슭막이가 이 줄로 섬(A1, 2026-09-14).", + "secondary_axes": [ + "stone_kind" + ], + "billing_component": "돌쌓기", + "variant_axis": "back_len_cm" + }, { "type_id": "ford_pavement", "work_item_code": "FP-12-06", @@ -414,7 +429,7 @@ "diameter_option_key": "pipe_diameter_mm", "variant_axis": "pipe_diameter_mm", "facility_rule": "⚠ `pipe_points.json` 은 **계곡 통과 시설 전부의 정본**이다(배관·BOX암거·물넘이·세월교·독립 기슭막이). `facility` 가 `pipe` 인 점만 배관이다 — `pipe_diameter_mm` 유무로 가르면 **관경 미지정 관을 놓친다**(2026-09-08 랩탑 창).", - "revetment_note": "⚠ 유출·유입부 기슭막이는 **관 옵션(`inlet_revet_*`·`outlet_revet_*`)이 정본**이다. 레지스트리 `revetment` 타입이 `managed_by: pipe_points` 라 구조물 목록에서 빠졌으므로 **구조물 쪽으로 또 세지 않는다**(2026-08-28 이관).", + "revetment_note": "⚠ 유출·유입부 기슭막이는 **관 옵션(`inlet_revet_*`·`outlet_revet_*`)이 정본**이다. 레지스트리 `revetment` 타입이 `managed_by: pipe_points` 라 구조물 목록에서 빠졌다(2026-08-28 이관). ⭐ **관 줄에는 안 넣고 원단위 전개가 한 번 셈** — B08 이 관 지점을 읽어 「배수관 · 유입부/유출부 기슭막이」 줄로 세움(`Engine_Pipe.facility_structures`, A1 2026-09-14). 관 부설(품셈 12-11 m당: 관·기초콘크리트·거푸집)에 기슭막이 몫이 없어 겹치지 않음. 그전에는 **어느 쪽도 안 세고 있었음**.", "not_ready": { "흄관 밑수 두 벌": "`FP-12-11-02` 는 밑수가 「1 m」와 「1 개소」 두 벌이다(표가 둘). B09 가 `#갈래` 로 두 표를 각각 세우므로 B08 은 `variant_value` 로 어느 쪽인지 보내면 된다.", "터파기·되메우기": "⚠ 관 부설과 터파기·되메우기가 각각 오면 **같은 굴착을 두 번 셀 수 있다**(B09 ㉡ 가드). 관 줄에는 지금 터파기를 붙이지 않는다." diff --git a/resources/tester/test_b08_facility_structures.py b/resources/tester/test_b08_facility_structures.py new file mode 100644 index 00000000..8a684a5a --- /dev/null +++ b/resources/tester/test_b08_facility_structures.py @@ -0,0 +1,86 @@ +"""계곡 통과 시설(`pipe_points.json`)이 원단위 전개·인계에 선다 (A1, 2026-09-14). + +⚠ 앞서 전개가 `structures.json` 만 읽어 관 유입·유출부 기슭막이·집수정·독립 기슭막이가 + 원단위·내역에 한 줄도 안 섰음. 관 줄(품셈 12-11 m당)에는 기슭막이 몫이 없어 **한 번만** 셈. +""" + +from __future__ import annotations + +import sys +from pathlib import Path + +ROOT = Path(__file__).resolve().parents[2] +if str(ROOT) not in sys.path: + sys.path.insert(0, str(ROOT)) + +from B05_Profile.B05_Profile_Structures_Schema import StructureInstance # noqa: E402 +from B08_Quantity import B08_Quantity_Router_Material as material # noqa: E402 +from B08_Quantity.B08_Quantity_Engine_Handoff import build_handoff # noqa: E402 +from B08_Quantity.B08_Quantity_Engine_Pipe import build_rows, facility_structures # noqa: E402 +from B08_Quantity.B08_Quantity_Engine_UnitQuantity import build_table # noqa: E402 + +NAMES = {"pipe": "배수관", "revetment": "기슭막이", "ford_bridge": "세월교"} + + +def _table(points: list[dict]) -> dict: + return build_table(facility_structures(points), NAMES) + + +def test_관_하나에_기슭막이_둘이_기본값_사유와_함께_서고_관_줄에는_없다() -> None: + point = {"chainage_m": 100.0, "options": {"pipe_diameter_mm": 1000}} + table = _table([point]) + names = [s["name"] for s in table["structures"]] + assert names == ["배수관 · 유입부 기슭막이", "배수관 · 유출부 기슭막이"] + inlet, outlet = table["structures"] + assert inlet["options"]["form"] == "돌쌓기(찰)" and outlet["options"]["form"] == "돌쌓기(메)" + assert all(s["components"] and s["length_m"] == 10 for s in table["structures"]) + assert "등록부 기본값으로 섰음" in inlet["notes"][0] + # 관 줄은 관 연장만 — 기슭막이가 거기 섞이지 않음(두 번 안 셈). + pipe = build_rows([point], [{"chainage_m": 100.0, "design": {"pipe_length_m": 8}}]) + assert [row["quantity"] for row in pipe["rows"]] == [8.0] + + work = {row["name"]: row for row in build_handoff(unit_quantity_table=table)["work_items"]} + assert work["배수관 · 유입부 기슭막이"]["work_item_code"] == "FP-13-04-05" + assert work["배수관 · 유출부 기슭막이"]["work_item_code"] == "FP-13-04-02" + assert ( + work["배수관 · 유입부 기슭막이"]["in_bill"] + and work["배수관 · 유입부 기슭막이"]["unit"] == "㎡" + ) + + +def test_유입구가_집수정이면_집수정_줄이_사유로_서고_유입_기슭막이는_없다() -> None: + table = _table([{"chainage_m": 5.0, "options": {"inlet_type": "집수정"}}]) + basin, outlet = table["structures"] + assert basin["type_id"] == "pipe_inlet_basin" and not basin["components"] and basin["notes"] + assert outlet["name"] == "배수관 · 유출부 기슭막이" + + +def test_유입구가_기슭막이인데_집수정_형식이_남아_있으면_집수정은_안_세고_알린다() -> None: + options = {"inlet_type": "기슭막이", "inlet_basin_form": "□형(기본형)"} + inlet = _table([{"chainage_m": 5.0, "options": options}])["structures"][0] + assert inlet["type_id"] == "revetment" and "집수정은 안 셈" in inlet["notes"][0] + + +def test_독립_기슭막이_한쪽이면_두_칸이_같을_때만_세고_다르면_사유() -> None: + same = {"facility": "revetment", "chainage_m": 50.0, "options": {"side": "좌"}} + rows = _table([same])["structures"] + assert [r["name"] for r in rows] == ["기슭막이 · 좌 벽"] and rows[0]["components"] + differ = {**same, "options": {"side": "우", "inlet_revet_height_m": 3.0}} + row = _table([differ])["structures"][0] + assert not row["components"] and "서버가 못 가림" in row["notes"][-1] + + +def test_산출식_없는_세월교는_사유로_서고_터파기_빠짐_줄을_안_만든다() -> None: + table = _table([{"facility": "ford_bridge", "chainage_m": 30.0, "options": {}}]) + assert "산출식이 아직 없습니다" in table["structures"][0]["notes"][0] + rows = build_handoff(unit_quantity_table=table)["work_items"] + assert not any("터파기가 안 선 구조물" in str(row.get("spec_detail")) for row in rows) + + +def test_구조물_목록의_관_지점_종류_옛_저장분은_안_셈(monkeypatch, tmp_path) -> None: + old = StructureInstance( + structure_id="old", type_id="revetment", placement="point", chainage_m=10.0 + ) + monkeypatch.setattr(material, "load_structures", lambda root: (1, [old])) + targets, _names, skipped = material._collect_structures(str(tmp_path)) + assert targets == [] and any("관 지점 정본이 주인" in note for note in skipped)