diff --git a/B08_Quantity/B08_Quantity_Engine_EarthworkSummary.py b/B08_Quantity/B08_Quantity_Engine_EarthworkSummary.py index b7dea09c..fcc33a95 100644 --- a/B08_Quantity/B08_Quantity_Engine_EarthworkSummary.py +++ b/B08_Quantity/B08_Quantity_Engine_EarthworkSummary.py @@ -376,7 +376,7 @@ def _haul_rows(source: SummaryInput) -> list[SummaryRow]: for item in source.haul_rows: key = str(item.get("equipment") or "") label = HAUL_LABELS.get(key, key or "운반") - ground = str(item.get("ground") or "") + ground = str(item.get("rock_class") or item.get("ground") or "") distance = item.get("average_distance_m") note = f"평균운반거리 {float(distance):.2f} m" if isinstance(distance, (int, float)) else "" # ⚠ **자연상태로 싣는다** — 「운반거리 산정은 다짐상태, 내역서 수량은 자연상태」 diff --git a/B08_Quantity/B08_Quantity_Engine_Handoff_Rows.py b/B08_Quantity/B08_Quantity_Engine_Handoff_Rows.py index 2b9572c9..29d792d5 100644 --- a/B08_Quantity/B08_Quantity_Engine_Handoff_Rows.py +++ b/B08_Quantity/B08_Quantity_Engine_Handoff_Rows.py @@ -166,11 +166,8 @@ def _earthwork_rows( variant_value = template.format(**inputs) # 부모(갈래 고르기형)를 가리키는 매핑 — 설정값으로 잎 코드를 고름(초류종자살포 5-24 · 09-14 ㉮). leaf_from = str((entry or {}).get("leaf_from") or "") - leaf = ( - ((entry or {}).get("leaf_codes") or {}).get(inputs.get(leaf_from)) - if leaf_from - else None - ) + leaf_codes = (entry or {}).get("leaf_codes") or {} + leaf = leaf_codes.get(inputs.get(leaf_from)) if leaf_from else None code = leaf or code missing = not leaf if leaf_from else (not variant_value or not template_ready) # 매핑이 「이 칸이 비면 못 고름」이라 적은 갈래 — 금액 없이 입력 사유(면고르기 · 09-14 Ⓒ). @@ -247,10 +244,13 @@ def _haul_rows( # ⚠ 코드가 없으면 **줄에 막힘 표시를 단다**(2026-09-09) — 목록에만 실으면 줄 단위로 # 보는 쪽이 「멀쩡한 줄」로 읽어 금액이 조용히 빠진다(도자운반·덤프운반이 그랬다). # ⚠ `in_bill` 이 False 인 무대 줄은 **막힌 것이 아니다** — 품에 포함이라 안 세우는 것. - haul_blocked = BLOCKED_UNIT_DATA_MISSING if (code is None and in_bill) else None - haul_blocked_reason = ( - f"운반({equipment})의 품셈 공종을 아직 못 이었습니다" if haul_blocked else "" + no_code = code is None and in_bill # 암 줄 막힘은 운반표가 구성비로 가르며 단 것(㉱) + haul_blocked = ( + BLOCKED_UNIT_DATA_MISSING if no_code else in_bill and row.get("blocked_kind") or None ) + haul_blocked_reason = str(in_bill and row.get("blocked_reason") or "") + if no_code: + haul_blocked_reason = f"운반({equipment})의 품셈 공종을 아직 못 이었습니다" # ⚠⚠ **내역서 수량은 자연상태다** — 유토곡선은 다짐상태로 쌓고(운반거리를 그 기준으로 # 재야 맞는다) 내역에 오르는 수량은 되돌린 값이다(`config_system_design` 5-4-3 # 「운반거리 산정 시 모든 수량은 다짐상태로 환산해 계산하고, **내역서에 적용하는 @@ -275,7 +275,9 @@ def _haul_rows( { "work_item_code": code, "name": f"{equipment} 운반", - "spec": str(row.get("ground") or ""), + "spec": " · ".join( + dict.fromkeys(filter(None, (row.get("rock_class"), row.get("ground")))) + ), "unit": "㎥", "quantity": quantity, # 운반에는 반영률 개념이 없다 — 그래서 `None` 이다(0 이 아니다). diff --git a/B08_Quantity/B08_Quantity_Engine_Handoff_Spoil.py b/B08_Quantity/B08_Quantity_Engine_Handoff_Spoil.py index 98d3c228..5a7f442f 100644 --- a/B08_Quantity/B08_Quantity_Engine_Handoff_Spoil.py +++ b/B08_Quantity/B08_Quantity_Engine_Handoff_Spoil.py @@ -24,6 +24,7 @@ from B08_Quantity.B08_Quantity_Engine_Handoff_Mapping import ( WorkItemMapping, ) from B08_Quantity.B08_Quantity_Engine_Handoff_Rows import LOADING_EQUIPMENT +from B08_Quantity.B08_Quantity_Engine_RockSplit import split_amounts SPOIL_NAME = "사토 운반" #: ⚠ **사토장 사면 물량은 안 센다**(2026-09-09 세 창 확인). 교본 6장 3절은 「완료 구간 비탈면을 @@ -98,20 +99,28 @@ def spoil_haul_rows( unknown = float(spoil.get("ground_unknown_m3") or 0.0) if by_ground or unknown > 0: rows: list[dict[str, Any]] = [] - for label, amount in sorted(by_ground.items()): - leg = by_distance.get(label) + # 암은 운반표와 같은 구성비 몫으로(자리표시 리핑암을 값으로 안 씀 · ㉱ `RockSplit`). + shares = (haul_table or {}).get("rock_shares") + for label, amount, share_reason, rock_class, source in split_amounts(by_ground, shares): + leg = by_distance.get(source) leg_blocked = blocked if leg is None else False rows.append( _spoil_row( code, amount, distance if leg is None else leg, - leg_blocked, - reason if leg_blocked else "", + leg_blocked or bool(share_reason), + share_reason or (reason if leg_blocked else ""), note, ground=label, extra=" · ".join( - (DISTANCE_FROM_SETTING if leg is None else DISTANCE_FROM_SITE, state_note) + part + for part in ( + f"암 갈래 {rock_class}" if rock_class else "", + DISTANCE_FROM_SETTING if leg is None else DISTANCE_FROM_SITE, + state_note, + ) + if part ), ) ) diff --git a/B08_Quantity/B08_Quantity_Engine_HaulSummary.py b/B08_Quantity/B08_Quantity_Engine_HaulSummary.py index 3f335150..bd8e2cbe 100644 --- a/B08_Quantity/B08_Quantity_Engine_HaulSummary.py +++ b/B08_Quantity/B08_Quantity_Engine_HaulSummary.py @@ -254,6 +254,8 @@ def summary_input_rows(table: dict[str, Any]) -> list[dict[str, Any]]: { "equipment": row["equipment"], "ground": row["ground"], + # 암 갈래(구성비로 가른 줄 · `RockSplit`) — 집계표 공종 칸이 흙깎기와 같은 이름을 씀. + "rock_class": row.get("rock_class"), "volume_m3": row["volume_m3"], "volume_basis": row.get("volume_basis") or "compacted", "natural_m3": row.get("natural_m3"), diff --git a/B08_Quantity/B08_Quantity_Engine_RockSplit.py b/B08_Quantity/B08_Quantity_Engine_RockSplit.py new file mode 100644 index 00000000..75490295 --- /dev/null +++ b/B08_Quantity/B08_Quantity_Engine_RockSplit.py @@ -0,0 +1,110 @@ +"""암 운반량을 **구성비로 가르기** — ㉱ (가) (2026-09-14 브레인 판정 · 8-1 사용자 확정). + +B06 의 암은 **한 종류 자리표시**다 — 토사 토글을 끄면 저장값이 `ripping_rock` 이 되지만 그 뜻은 +「암」이고 「갈라 넣는 것은 설계내역 몫」(`B06_Section_UI_Cross_Design.ts` 머리 주석). 그런데 운반표·사토가 +그 이름(리핑암)을 **값처럼** 넘겨, 깎기 암은 구성비가 비어 막히는데 운반 암은 금액이 섰다. + +⇒ 운반표를 만든 **한 곳**(`Router_Earthwork`)에서 암 줄을 흙깎기와 **같은 구성비**(`_split_by_rock`)와 + 갈래별 시공법으로 가른다. 운반거리 탭·토공집계 운반 줄·인계·사토가 모두 이 표를 읽어 한 값이 된다. +⚠ 유토곡선 거리·다짐 부피는 그대로(자리표시 C 로 쌓은 값) — 부피를 몫대로 나눌 뿐이라 검산이 안 흔들림. + C 까지 구성비로 맞추는 것은 (나) 차례. +⚠ 구성비가 비면 「암」 한 줄로 **막는다** — 깎기와 같은 사유(지어낸 갈래로 금액을 세우지 않음). +""" + +from __future__ import annotations + +from typing import Any, Iterable + +from B08_Quantity.B08_Quantity_Engine_EarthworkSummary import SummaryInput, _split_by_rock +from B08_Quantity.B08_Quantity_Engine_Handoff_Mapping import ( + BLOCKED_INPUT_MISSING, + METHOD_TO_GROUND, + NOTE_METHOD_MISSING, + NOTE_ROCK_RATIO_MISSING, +) + +#: 유토곡선이 넘기는 암 이름 — 둘 다 자리표시로 본다(옛 자료 `blasting_rock` 도 「암」). +ROCK_GROUNDS = ("리핑암", "발파암") +#: 몫대로 나누는 수량 칸 — 운반표 줄(`HaulSummary.build_table`). +SHARED_KEYS = ("volume_m3", "natural_m3", "work_m3m") + + +def rock_shares( + classes: Iterable[str], ratios_pct: dict[str, Any], methods: dict[str, str | None] +) -> list[dict[str, Any]]: + """갈래별 몫 — `{rock_class, fraction, ground, blocked_reason, note}`. 흙깎기와 같은 안분.""" + source = SummaryInput(rock_classes=list(classes), rock_ratios_pct=dict(ratios_pct or {})) + shares = [] + for name, fraction, note in _split_by_rock(1.0, source): + ground = "암" if name == "암" else METHOD_TO_GROUND.get(methods.get(name) or "") + if name == "암": + reason = NOTE_ROCK_RATIO_MISSING + else: + reason = "" if ground else NOTE_METHOD_MISSING + shares.append( + { + "rock_class": name, + "fraction": fraction, + "ground": ground or name, + "blocked_reason": reason, + "note": note, + } + ) + return shares + + +def split_rows( + rows: Iterable[dict[str, Any]], shares: list[dict[str, Any]] +) -> list[dict[str, Any]]: + """암 줄을 몫마다 한 줄로 — 토사 줄은 그대로.""" + out: list[dict[str, Any]] = [] + for row in rows: + if row.get("ground") not in ROCK_GROUNDS: + out.append(row) + continue + for share in shares: + numbers = { + key: row[key] * share["fraction"] + for key in SHARED_KEYS + if isinstance(row.get(key), (int, float)) + } + out.append( + { + **row, + **numbers, + "ground": share["ground"], + "rock_class": share["rock_class"], + "rock_split_note": share["note"], + "blocked_kind": BLOCKED_INPUT_MISSING if share["blocked_reason"] else None, + "blocked_reason": share["blocked_reason"], + } + ) + return out + + +def split_amounts( + by_ground: dict[str, float], shares: list[dict[str, Any]] | None +) -> list[tuple[str, float, str, str | None, str]]: + """갈래별 물량(사토) → `(갈래, 물량, 막힘 사유, 암 갈래, 원래 이름)`. 몫이 없으면 그대로.""" + out = [] + for label, amount in sorted(by_ground.items()): + if shares and label in ROCK_GROUNDS: + out.extend( + (s["ground"], amount * s["fraction"], s["blocked_reason"], s["rock_class"], label) + for s in shares + ) + else: + out.append((label, amount, "", None, label)) + return out + + +def apply_rock_split( + haul: dict[str, Any], + classes: Iterable[str], + ratios_pct: dict[str, Any], + methods: dict[str, str | None], +) -> None: + """운반표(`rows`)를 자리에서 가르고 사토가 같은 몫을 쓰게 `rock_shares` 를 싣는다.""" + shares = rock_shares(classes, ratios_pct, methods) + haul["rows"] = split_rows(haul.get("rows") or [], shares) + haul["rock_shares"] = shares diff --git a/B08_Quantity/B08_Quantity_Router_Earthwork.py b/B08_Quantity/B08_Quantity_Router_Earthwork.py index ee9a71ee..26640669 100644 --- a/B08_Quantity/B08_Quantity_Router_Earthwork.py +++ b/B08_Quantity/B08_Quantity_Router_Earthwork.py @@ -44,6 +44,7 @@ from B08_Quantity.B08_Quantity_Engine_EarthworkTable import StationArea, build_t from B08_Quantity.B08_Quantity_Engine_Handoff import load_mapping from B08_Quantity.B08_Quantity_Engine_HaulSummary import build_table as build_haul_table from B08_Quantity.B08_Quantity_Engine_HaulSummary import check_against_plan, summary_input_rows +from B08_Quantity.B08_Quantity_Engine_RockSplit import apply_rock_split from B08_Quantity.B08_Quantity_Engine_Preparation import build_table as build_preparation_table from B08_Quantity.B08_Quantity_Engine_SlopeArea import build_table as build_slope_table from B08_Quantity.B08_Quantity_Engine_SlopeLength import road_surface_area, station_slopes @@ -58,6 +59,7 @@ from common_util.common_util_project_settings import ( haul_limit_choice, quantity_settings, rock_classes, + rock_method, save_section, topsoil_target, ) @@ -115,6 +117,10 @@ async def get_earthwork_table(project_id: UUID, route_id: int) -> JSONResponse: plan = await _stored_haul_plan(project_id, route_id) haul = build_haul_table(plan, factors) + # ㉱ (가) 암은 흙깎기와 같은 구성비·시공법으로 가름 — B06 리핑암은 자리표시(8-1 · 2026-09-14 브레인). + classes = rock_classes(settings) + methods = {name: rock_method(settings, name) for name in classes} + apply_rock_split(haul, classes, settings.get("rock_ratios_pct") or {}, methods) # 사토 — **운반 줄이 되는 값**인데 유토곡선의 띠·이동에는 안 들어 있다(잔량으로 남는다). # 여기서 그 값을 운반표에 실어 인계가 「사토 운반」 한 줄을 세우게 한다. # ⚠ 거리는 품셈이 정하지 않는다 — 설계 입력(`spoil_site_distance_m`)이고 없으면 막힌다. diff --git a/resources/tester/test_b08_haul_summary.py b/resources/tester/test_b08_haul_summary.py index a6f4f954..5f660d0b 100644 --- a/resources/tester/test_b08_haul_summary.py +++ b/resources/tester/test_b08_haul_summary.py @@ -153,6 +153,8 @@ def test_집계표_입력으로_줄임() -> None: "volume_basis", # 쓴 계수도 함께 온다 — 받는 쪽이 되짚을 수 있어야 한다. "conversion_c", + # 2026-09-14 ㉱ — 구성비로 가른 암 갈래(안 가른 줄은 None) · 집계표 공종 칸이 씀. + "rock_class", } for row in rows ) diff --git a/resources/tester/test_b08_rock_haul_split.py b/resources/tester/test_b08_rock_haul_split.py new file mode 100644 index 00000000..ea8ca7c9 --- /dev/null +++ b/resources/tester/test_b08_rock_haul_split.py @@ -0,0 +1,106 @@ +"""㉱ (가) 암 운반량을 구성비로 가르기 — 2026-09-14 브레인 판정(구성비가 정본 · 8-1 사용자 확정). + +B06 토사 토글 끔 = 저장값 `ripping_rock` 은 **자리표시**(「갈라 넣는 것은 설계내역 몫」)인데 +운반·사토가 그 이름을 값처럼 써서 「깎기 암은 구성비가 비어 막히는데 운반 암은 리핑암으로 금액이 섬」. +⇒ 운반표 한 곳에서 암 줄을 구성비·시공법으로 가름 · 비면 「암」 한 줄로 막음(깎기와 같은 사유). +유토곡선 거리·다짐 부피는 그대로 — 검산이 안 흔들림((나)는 뒤 차례). +""" + +from __future__ import annotations + +import sys +from pathlib import Path + +import pytest + +ROOT = Path(__file__).resolve().parents[2] +if str(ROOT) not in sys.path: + sys.path.insert(0, str(ROOT)) + +from B08_Quantity.B08_Quantity_Engine_EarthworkSummary import SummaryInput # noqa: E402 +from B08_Quantity.B08_Quantity_Engine_EarthworkSummary import build_table as summary # noqa: E402 +from B08_Quantity.B08_Quantity_Engine_Handoff import build_handoff # noqa: E402 +from B08_Quantity.B08_Quantity_Engine_Handoff_Mapping import ( # noqa: E402 + NOTE_METHOD_MISSING, + NOTE_ROCK_RATIO_MISSING, +) +from B08_Quantity.B08_Quantity_Engine_HaulSummary import build_table, check_against_plan # noqa: E402 +from B08_Quantity.B08_Quantity_Engine_RockSplit import apply_rock_split # noqa: E402 + +CLASSES = ["토사", "풍화암", "연암", "보통암", "경암"] +PLAN = { + "blocks": [ + { + "bands": [ + {"equipment": "dozer", "haul_distance_m": 40.0, "haul_from_m": 0.0, + "haul_to_m": 40.0, "ea_m3": 90.0, "rr_m3": 115.0, "br_m3": 0.0}, + ] + } + ], + "transfers": [], + "hauled_m3": 205.0, + "transferred_m3": 0.0, +} # fmt: skip +SPOIL = {"volume_m3": 23.0, "distance_m": 300.0, "by_ground_m3": {"rr_m3": 23.0}, + "natural_m3_by_ground": {"rr_m3": 20.0}} # fmt: skip + + +def _haul(ratios: dict, methods: dict) -> dict: + haul = build_table(PLAN) + haul["spoil"] = dict(SPOIL) + apply_rock_split(haul, CLASSES, ratios, methods) + return haul + + +def _rock(haul: dict) -> list[dict]: + return [r for r in haul["rows"] if r["ground"] != "토사"] + + +def test_구성비가_비면_운반_암도_한_줄_암으로_막힘() -> None: + haul = _haul({}, {}) + (rock,) = _rock(haul) + assert rock["ground"] == "암" and rock["natural_m3"] == pytest.approx(100.0) + assert rock["blocked_reason"] == NOTE_ROCK_RATIO_MISSING + items = build_handoff(haul_table=haul)["work_items"] + dozer = next(r for r in items if r["name"].endswith("운반") and r["ground_class"] == "암") + assert ( + dozer["blocked_kind"] == "input_missing" + and dozer["blocked_reason"] == NOTE_ROCK_RATIO_MISSING + ) + spoil = next(r for r in items if r["name"] == "사토 운반") + assert ( + spoil["blocked_kind"] == "input_missing" + and NOTE_ROCK_RATIO_MISSING in spoil["blocked_reason"] + ) + + +def test_구성비와_시공법대로_갈리고_검산은_그대로() -> None: + ratios = {"연암": 60.0, "보통암": 40.0} + methods = {"연암": "ripping", "보통암": "blasting"} + haul = _haul(ratios, methods) + got = [(r["rock_class"], r["ground"], round(r["natural_m3"], 6)) for r in _rock(haul)] + assert got == [("연암", "리핑암", 60.0), ("보통암", "발파암", 40.0)] + assert check_against_plan(haul, PLAN).difference_m3 == pytest.approx(0.0) + items = build_handoff(haul_table=haul)["work_items"] + dozer = [r for r in items if r["haul_equipment"] == "dozer" and r["ground_class"] != "토사"] + assert [(r["spec"], r["variant_value"], r["blocked_kind"]) for r in dozer] == [ + ("연암 · 리핑암", "리핑암", None), + ("보통암 · 발파암", "발파암", None), + ] + spoil = [r for r in items if r["name"] == "사토 운반"] + assert [(r["variant_value"], round(r["quantity"], 3)) for r in spoil] == [ + ("리핑암", 12.0), + ("발파암", 8.0), + ] + # 깎기와 같은 몫 — 흙깎기 암 1000 이 같은 구성비로 600 · 400 + rows = summary(SummaryInput(earthwork_totals={"cut_rock_volume_m3": 1000.0}, + rock_classes=CLASSES, rock_ratios_pct=ratios))["rows"] # fmt: skip + cut = {r["item"]: r["amount"] for r in rows if r["group"] == "흙깎기" and r["item"] != "토사"} + assert cut == {"연암": 600.0, "보통암": 400.0} + + +def test_시공법을_안_고른_갈래만_막힘() -> None: + haul = _haul({"연암": 50.0, "경암": 50.0}, {"연암": "ripping"}) + rows = {r["rock_class"]: r for r in _rock(haul)} + assert not rows["연암"]["blocked_reason"] + assert rows["경암"]["blocked_reason"] == NOTE_METHOD_MISSING