diff --git a/B08_Quantity/B08_Quantity_Engine_UnitQuantity.py b/B08_Quantity/B08_Quantity_Engine_UnitQuantity.py index cc2d9ace..d52dad12 100644 --- a/B08_Quantity/B08_Quantity_Engine_UnitQuantity.py +++ b/B08_Quantity/B08_Quantity_Engine_UnitQuantity.py @@ -37,6 +37,7 @@ from functools import lru_cache from dataclasses import dataclass, field from typing import Any, Iterable +from common_util.common_util_structure_face_role import structure_face_role from B08_Quantity.B08_Quantity_Engine_ObservedUnit import ( BASIS_DERIVED, BASIS_OBSERVED, @@ -281,7 +282,13 @@ def load_slope_table(path: str | None = None) -> dict[str, Any]: def face_slope_ratio( - options: dict[str, Any], *, wet: bool, height_m: float, table: dict[str, Any] | None = None + options: dict[str, Any], + *, + wet: bool, + height_m: float, + face: str | None = None, + face_reason: str = "", + table: dict[str, Any] | None = None, ) -> tuple[float, str]: """(기울기 1:n 의 n, 근거 문구). @@ -299,10 +306,15 @@ def face_slope_ratio( found = table if table is not None else load_slope_table() steps = found.get("steps_m") or [] bond = "찰쌓기" if wet else "메쌓기" - # ⚠ 성토/절토는 **저장이 아직 안 갖고 있다.** 레지스트리에 없는 키를 읽지 않는다 - # (읽으면 「없는 칸을 읽는」 자리가 되어 옵션 키 검사가 잡는다). 값이 생기면 - # 여기 한 줄만 바꾸면 된다 — 그때까지 성토 열로 잠정 적용하고 근거에 적는다. - face = DEFAULT_SLOPE_FACE + # ⚠ 성토/절토는 **판정 한 벌**(`common_util_structure_face_role`)이 준다. 못 가르면 + # `None` 이 오는데, **성토로 눅이지 않는다** — 임의값이 금액으로 굳으면 안 된다. + # 그때는 종전값으로 서되 **왜 못 갈랐는지**를 근거에 적는다. + if not face: + why = face_reason or "성절토를 가를 근거 없음" + return ( + LEGACY_FACE_SLOPE_RATIO, + f"⚠ {why} — 표준경사표를 못 골라 종전값 1:{LEGACY_FACE_SLOPE_RATIO} 로 섰음", + ) row = ((found.get("table") or {}).get(bond) or {}).get(face) if not row: return LEGACY_FACE_SLOPE_RATIO, f"표준경사표를 못 읽어 종전값 1:{LEGACY_FACE_SLOPE_RATIO}" @@ -315,7 +327,8 @@ def face_slope_ratio( ratio = float(row[min(index, len(row) - 1)]) label = f"직고 {height_m:g}m" + (f" ≤{steps[index]:g}m" if index < len(steps) else " 7m 초과") note = f"품셈 13-4-4 [주]⑪ 표준경사 · {bond} {face} · {label} → 1:{ratio:g}" - note += " · ⚠ 성토/절토가 저장에 없어 **성토 열로 잠정**" + if face_reason: + note += f" · {face_reason}" return ratio, note @@ -513,7 +526,12 @@ def boulder_masonry( def stone_masonry( - height_m: float, length_m: float, options: dict[str, Any], wet: bool + height_m: float, + length_m: float, + options: dict[str, Any], + wet: bool, + face: str | None = None, + face_reason: str = "", ) -> tuple[list[Component], list[str]]: """돌쌓기(찰/메) 1구간 전개 — 실무 `기슭막이(찰쌓기, H=1.5, 기초무)` 시트의 식. @@ -561,7 +579,9 @@ def stone_masonry( # 돌 메쌓기 2.0m 이하 **1:0.3** / 큰돌쌓기 **1:0.3 이상**(전도 방지)」 # (`지식DB 02_상세설계/구조물/돌쌓기.md §1`, 값은 `data_masonry` 의 `face_slope`). # 전면 기울기 — 품셈 표준경사표로 자동 판정하고, 저장 제원에 값이 있으면 그것이 이긴다. - slope_ratio, slope_basis = face_slope_ratio(options, wet=wet, height_m=height_m) + slope_ratio, slope_basis = face_slope_ratio( + options, wet=wet, height_m=height_m, face=face, face_reason=face_reason + ) constants = STONE_MASONRY face_area = height_m * length_m # 정면적 @@ -735,9 +755,11 @@ OBSERVED_SPEC_KEYS: dict[str, tuple[str, ...]] = { } EXPANDERS = { - "masonry_wet": lambda h, l, o: stone_masonry(h, l, o, wet=True), - "masonry_dry": lambda h, l, o: stone_masonry(h, l, o, wet=False), - "boulder_masonry": lambda h, l, o: boulder_masonry(h, l, o), + # 넷째·다섯째 인자는 **성토/절토와 그 까닭** — 표준경사 표가 그것으로 갈린다. + "masonry_wet": lambda h, l, o, f=None, r="": stone_masonry(h, l, o, True, f, r), + "masonry_dry": lambda h, l, o, f=None, r="": stone_masonry(h, l, o, False, f, r), + # 큰돌쌓기는 표준경사 표 대상이 아니라 성절토를 안 쓴다(교본 「1:0.3 이상」). + "boulder_masonry": lambda h, l, o, f=None, r="": boulder_masonry(h, l, o), # ⚠⚠ **큰돌쌓기(`boulder_masonry`)를 여기에 두지 않는다** (2026-09-07 발견). # 큰돌쌓기는 품셈 **13-6** 이고 돌쌓기는 **13-4** 다 — **규격 축이 다르다.** # 돌쌓기는 **뒷길이**(35·45·55·60㎝), 큰돌쌓기는 **직경**(40~60·60~80·80~100㎝). @@ -817,6 +839,7 @@ def expand( structure: dict[str, Any], names: dict[str, str] | None = None, observed: ObservedUnitTable | None = None, + section_mode: str | None = None, ) -> StructureQuantity: """구조물 하나를 전개한다. 치수는 저장된 제원에서만 읽는다(치수 두 벌 금지).""" type_id = str(structure.get("type_id") or "") @@ -862,7 +885,9 @@ def expand( f"{type_label(type_id, names)}의 수량 산출식이 아직 없습니다 — 물량이 서지 않습니다" ) return result - result.components, notes = expander(height, length, options) + # 성토/절토 — **판정 한 벌**을 부른다(우리가 따로 짜지 않는다). + face, face_reason = structure_face_role(section_mode, options.get("side")) + result.components, notes = expander(height, length, options, face, face_reason) result.notes.extend(notes) return result @@ -881,8 +906,29 @@ def verify_no_mix_components(quantities: Iterable[StructureQuantity]) -> list[st return found +def _section_mode_at( + structure: dict[str, Any], section_modes: dict[float, str] | None +) -> str | None: + """구조물이 선 자리의 단면유형. **가장 가까운 측점**의 값을 쓴다. + + ⚠ 구조물은 구간(start~end)이고 단면유형은 측점 값이라 딱 맞는 측점이 없을 수 있다. + 가장 가까운 측점을 쓰되, 목록이 없으면 `None`(가를 근거 없음)으로 둔다 — + **성토로 눅이지 않는다.** + """ + if not section_modes: + return None + center = _num(structure.get("chainage_m")) + if not center: + start, end = _num(structure.get("start_m")), _num(structure.get("end_m")) + center = (start + end) / 2.0 if (start or end) else 0.0 + nearest = min(section_modes, key=lambda chainage: abs(float(chainage) - center)) + return section_modes.get(nearest) + + def build_table( - structures: Iterable[dict[str, Any]], names: dict[str, str] | None = None + structures: Iterable[dict[str, Any]], + names: dict[str, str] | None = None, + section_modes: dict[float, str] | None = None, ) -> dict[str, Any]: """화면·API 가 그대로 쓰는 모양. 성분별 총량과 구조물별 내역을 함께 낸다.""" observed = load_observed_table() @@ -891,7 +937,10 @@ def build_table( for item in structures: expanded_inputs.append(item) expanded_inputs.extend(attachments_of(item)) - quantities = [expand(item, names, observed) for item in expanded_inputs] + quantities = [ + expand(item, names, observed, _section_mode_at(item, section_modes)) + for item in expanded_inputs + ] violations = verify_no_mix_components(quantities) totals: dict[str, dict[str, Any]] = {} diff --git a/B08_Quantity/B08_Quantity_Router_Material.py b/B08_Quantity/B08_Quantity_Router_Material.py index b95b5d64..659cb2fd 100644 --- a/B08_Quantity/B08_Quantity_Router_Material.py +++ b/B08_Quantity/B08_Quantity_Router_Material.py @@ -25,6 +25,8 @@ from fastapi import APIRouter from fastapi.responses import JSONResponse from B03_FileInput.B03_FileInput_Repository import get_project_storage_relative_path +from B06_Section.B06_Section_Repository import get_cross_section_designs +from B06_Section.B06_Section_Repository import get_workflow_route_context from B05_Profile.B05_Profile_Structures_Repository import load_structures from B05_Profile.B05_Profile_Structures_Schema import structure_type_map from B08_Quantity.B08_Quantity_Engine_Handoff import build_handoff, load_mapping, summarize @@ -78,6 +80,28 @@ def _collect_structures( return targets, names, sorted(set(skipped)) +async def _section_modes(project_id: UUID) -> dict[float, str]: + """측점별 단면유형(`left_cut` 등). 구조물이 **성토면인가 절토면인가**를 가릴 때 쓴다. + + ⚠ 새 저장 키를 만들지 않는다 — 이미 저장되는 `design.section_mode` 를 읽기만 한다. + 못 읽으면 빈 표로 두고, 판정이 「가를 근거 없음」이 되게 한다(성토로 눅이지 않음). + """ + try: + context = await run_with_connection(get_workflow_route_context, project_id) + route_id = int((context or {}).get("route_id") or 0) + if not route_id: + return {} + designs = await run_with_connection(get_cross_section_designs, route_id) + except Exception: + logger.exception("B08 단면유형 조회 실패: project_id=%s", project_id) + return {} + return { + float(item["chainage_m"]): str((item.get("design") or {}).get("section_mode") or "") + for item in designs + if (item.get("design") or {}).get("section_mode") + } + + @router.get("/{project_id}/quantity/material-summary") async def get_material_summary(project_id: UUID) -> JSONResponse: """구조물 원단위와 자재총괄을 **한 응답**으로 낸다. @@ -103,7 +127,7 @@ async def get_material_summary(project_id: UUID) -> JSONResponse: content={"status": "error", "message": "구조물 정본을 읽지 못했습니다."}, ) - unit_table = build_unit_table(structures, names) + unit_table = build_unit_table(structures, names, await _section_modes(project_id)) settings = quantity_settings(project_root) material_table = build_material_table( unit_table, @@ -153,7 +177,7 @@ async def get_handoff(project_id: UUID) -> JSONResponse: ) structures, names, skipped = _collect_structures(project_root) - unit_table = build_unit_table(structures, names) + unit_table = build_unit_table(structures, names, await _section_modes(project_id)) settings = quantity_settings(project_root) material_table = build_material_table( unit_table, supply_map=settings.get("material_supply") or {}