feat(b09): 운반공 줄 = 거리이정표 행 1:1 (253-1)

- 자동 상자 = 옛 키 · 사용자 상자 = haul:r-{id} · 둘째 행 haul:r-{id}.{n}
- 이름 = 항목 + 운반 · 수량 빈칸 · 연결 = 경로 ref(설계내역서 덮개가 이김)
- B09 가 거리이정표 표(compute_milestone)를 읽어 행 · 거리(L1 포장 · L2 비포장)를 셈 덧값으로

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01JywHxQuWPk6UYXTebyrBoU
This commit is contained in:
2026-10-11 13:59:10 +09:00
co-authored by Claude Opus 5.5
parent c172c4a462
commit a1c033c3cb
4 changed files with 142 additions and 15 deletions
+34
View File
@@ -127,3 +127,37 @@ def with_haul_routes(data: dict[str, Any], groups: list[dict[str, Any]]) -> dict
}
)
return {**data, "routes": routes}
def route_bill_key(route: dict[str, Any], index: int = 0) -> str:
"""거리이정표 행 → B09 운반공 줄 키(253-1) — 자동 상자 = 옛 키 · 사용자 상자 = `haul:r-{id}` ·
한 상자 둘째 행부터 `haul:r-{id}.{n}`(경로 id 에 「.」 없음 — 안 겹침)."""
if route.get("haul"):
return bill_key(str(route["haul"]))
return f"haul:r-{route.get('id')}" + (f".{index}" if index else "")
def haul_rows(data: dict[str, Any], table: list[dict[str, Any]]) -> list[dict[str, Any]]:
"""거리이정표 표 행 하나 = 운반공 줄 하나(253-1) — [{key, name, ref, user, L1, L2}] ·
L1 = 포장 계(2차 + 1차 + 포장) · L2 = 비포장(`compute_milestone` 결과) ·
user = 사용자 상자(자동 상자는 총괄자재 묶음 줄이 이미 있음 · 둘째 행은 안 씀)."""
routes = {r.get("id"): r for r in data.get("routes") or []}
out, seen = [], set()
for row in table:
route = routes.get(row.get("route")) or {}
key = route_bill_key(route, int(row.get("index") or 0))
if key in seen:
continue
seen.add(key)
item = str(row.get("item") or "").strip()
out.append(
{
"key": key,
"name": f"{item}운반" if item else "운반",
"ref": route.get("ref"),
"user": not route.get("haul"),
"L1": row.get("paved"),
"L2": row.get("unpaved"),
}
)
return out
+24 -3
View File
@@ -106,9 +106,12 @@ def structure_items(summary: dict[str, Any] | None) -> list[dict[str, Any]]:
def haul_items(
materials: dict[str, Any] | None, refs: dict[str, Any] | None = None
materials: dict[str, Any] | None,
refs: dict[str, Any] | None = None,
routes: list[dict[str, Any]] | None = None,
) -> list[dict[str, Any]]:
"""운반공 줄 — B08 운반 묶음(251-1 · 장비 둘 + 총괄자재 묶음 · 옛 키 haul:1~6 그대로) ·
"""운반공 줄 — B08 운반 묶음(251-1 · 장비 둘 + 총괄자재 묶음 · 옛 키 haul:1~6 그대로) +
거리이정표 사용자 상자 행(253-1 · `B08_Quantity_Haul.haul_rows` · 수량 빈칸) ·
연결 = 거리이정표 단가연결(251-5 · `refs{줄 키: ref}` · 설계내역서 연결 덮개가 이김)."""
out = []
for g in haul_groups(materials):
@@ -128,6 +131,24 @@ def haul_items(
"why": WHY_LOGIC if logic else WHY_HAUL,
}
)
for r in routes or []:
if not r.get("user"):
continue
ref = (refs or {}).get(r["key"])
logic = (ref or {}).get("tab") == "logic"
out.append(
{
"key": r["key"],
"section": "3",
"name": r["name"],
"spec": "",
"unit": "",
"qty": None,
"note": "B08 거리이정표",
"ref": ref,
"why": WHY_LOGIC if logic else WHY_HAUL,
}
)
return out
@@ -136,7 +157,7 @@ def bill_items(b08: dict[str, Any]) -> dict[str, Any]:
"items": [
*earth_items(b08.get("earthwork")),
*structure_items(b08.get("summary")),
*haul_items(b08.get("materials"), b08.get("haul_refs")),
*haul_items(b08.get("materials"), b08.get("haul_refs"), b08.get("haul_routes")),
]
}
+24 -12
View File
@@ -27,11 +27,13 @@ from uuid import UUID
from fastapi import APIRouter, Body, Depends, HTTPException
from pydantic import BaseModel
from B08_Quantity.B08_Quantity_Haul import bill_key
from B08_Quantity.B08_Quantity_Haul import haul_rows
from B08_Quantity.B08_Quantity_Milestone_Calc import compute_milestone
from B08_Quantity.B08_Quantity_Router_Earthwork import earthwork_table
from B08_Quantity.B08_Quantity_Router_Hydro import _project_root
from B08_Quantity.B08_Quantity_Router_Material import material_sheets
from B08_Quantity.B08_Quantity_Router_Milestone import read_data as read_milestone
from B08_Quantity.B08_Quantity_Router_Milestone import site_info
from B09_Estimation import B09_Estimation_Bill_Library as library
from B09_Estimation import B09_Estimation_Prices as prices
from B09_Estimation import B09_Estimation_Sangeun as sangeun
@@ -110,18 +112,27 @@ def compute_twice(root: Path, inputs: dict[str, Any], user: dict[str, Any]) -> t
return sheets, compute_summary(sheets, hopyo_lines(sheets))
def haul_refs(root: Path) -> dict[str, Any]:
"""거리이정표 운반 상자 단가연결(251-5) → `{설계내역서 줄 키: ref}` — 못 읽으면 빈 묶음."""
def haul_routes(root: Path) -> list[dict[str, Any]]:
"""거리이정표 행 → 운반공 줄(253-1 · `haul_rows` · 거리 L1 · L2 포함) — 못 읽으면 빈 목록."""
try:
routes = read_milestone(root).get("routes") or []
except Exception: # noqa: BLE001 — 거리이정표 없음 = 연결 없음
return {}
return {bill_key(r["haul"]): r["ref"] for r in routes if r.get("haul") and r.get("ref")}
data = read_milestone(root)
mid = site_info(root, data["site"]["mid_station_m"])["mid_default_m"]
return haul_rows(data, compute_milestone(data, mid))
except Exception: # noqa: BLE001 — 거리이정표 없음 = 운반 연결 없음
return []
def haul_user(rows: list[dict[str, Any]]) -> dict[str, Any]:
"""셈 덧값 — `haul_refs{줄 키: ref}`(251-5) · `haul_dist{줄 키: {L1, L2}}`(253-2 거리 채움)."""
return {
"haul_refs": {r["key"]: r["ref"] for r in rows if r.get("ref")},
"haul_dist": {r["key"]: {"L1": r["L1"], "L2": r["L2"]} for r in rows},
}
def calc_user(root: Path) -> dict[str, Any]:
"""셈 · 수요 · 사본 정리용 사용자 값 — 저장 값 + 거리이정표 운반 연결(화면엔 안 보냄)."""
return {**store.read_user(root), "haul_refs": haul_refs(root)}
"""셈 · 수요 · 사본 정리용 사용자 값 — 저장 값 + 거리이정표 운반 연결 · 거리(화면엔 안 보냄)."""
return {**store.read_user(root), **haul_user(haul_routes(root))}
async def _demand(root: Path) -> list[str]:
@@ -143,11 +154,12 @@ async def payload(project_id: UUID, root: Path) -> dict[str, Any]:
)
book = await asyncio.to_thread(store.read_book, root)
inputs = await asyncio.to_thread(prices.build_inputs, book, keys, b08["materials"])
inputs["bill"] = bill_items({**b08, "haul_refs": await asyncio.to_thread(haul_refs, root)})
hauls = await asyncio.to_thread(haul_routes, root)
inputs["bill"] = bill_items({**b08, **haul_user(hauls), "haul_routes": hauls})
inputs["hopyo"] = hopyo_prices(root)
inputs["daega"] = await asyncio.to_thread(daega_inputs, root)
user = await asyncio.to_thread(store.read_user, root)
calc = {**user, "haul_refs": await asyncio.to_thread(haul_refs, root)}
calc = {**user, **haul_user(hauls)}
sheets, summary = await asyncio.to_thread(compute_twice, root, inputs, calc)
if schedule_draft:
schedule_draft.attach(inputs, sheets, user)
@@ -176,7 +188,7 @@ async def estimation(project_id: UUID) -> dict[str, Any]:
async def save_estimation(project_id: UUID, body: dict[str, Any] = Body(...)) -> dict[str, Any]:
root = await _project_root(project_id)
user = await asyncio.to_thread(store.write_user, root, body.get("user"))
calc = {**user, "haul_refs": await asyncio.to_thread(haul_refs, root)}
calc = {**user, **haul_user(await asyncio.to_thread(haul_routes, root))}
await asyncio.to_thread(library.prune, root, calc) # 204-2 줄이 안 잇는 서고 사본
keys = await _demand(root)
book = await asyncio.to_thread(store.read_book, root)
+60
View File
@@ -0,0 +1,60 @@
"""253-1 거리이정표 행 1:1 → B09 운반공 줄 — 자동 상자 = 옛 키 · 사용자 상자 = `haul:r-{id}`."""
import sys
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
from B08_Quantity.B08_Quantity_Haul import haul_rows, route_bill_key # noqa: E402
from B08_Quantity.B08_Quantity_Milestone_Calc import compute_milestone # noqa: E402
from B09_Estimation.B09_Estimation_Bill_Items import haul_items # noqa: E402
from B09_Estimation.B09_Estimation_Router import haul_user # noqa: E402
LOGIC = {"tab": "logic", "code": "GF000134", "name": "골재운반", "given": {}}
DATA = {
"site": {"mid_station_m": 2000, "mid_road": "비포장"},
"base": {"enabled": False, "legs": []},
"routes": [
{"id": "h-sand", "name": "골재상", "haul": "sand", "via_base": False,
"legs": [{"road": "포장", "km": 25.1}], "rows": [{"item": "모래", "dest": "start"}]},
{"id": "u1", "name": "석산", "ref": LOGIC, "via_base": False,
"legs": [{"road": "포장", "km": 10}, {"road": "비포장", "km": 0.8}],
"rows": [{"item": "혼합골재", "dest": "start"}, {"item": "깬돌", "dest": "mid"}]},
],
} # fmt: skip
def test_keys():
assert route_bill_key({"id": "h-sand", "haul": "sand"}) == "haul:3"
assert route_bill_key({"id": "h-x", "haul": "remicon"}) == "haul:remicon"
assert route_bill_key({"id": "u1"}) == "haul:r-u1"
assert route_bill_key({"id": "u1"}, 1) == "haul:r-u1.1"
def test_rows_one_per_table_row():
rows = haul_rows(DATA, compute_milestone(DATA, None))
got = {r["key"]: r for r in rows}
assert list(got) == ["haul:3", "haul:r-u1", "haul:r-u1.1"]
assert (got["haul:3"]["L1"], got["haul:3"]["L2"], got["haul:3"]["user"]) == (25.1, 0, False)
assert (got["haul:r-u1"]["name"], got["haul:r-u1"]["L1"], got["haul:r-u1"]["L2"]) == (
"혼합골재운반", 10, 0.8) # fmt: skip
assert got["haul:r-u1.1"]["L2"] == 2.8 # 중점 = 시점 → 중점 2 km 비포장 더함
assert got["haul:r-u1"]["ref"] == LOGIC
def test_bill_items_user_rows():
rows = haul_rows(DATA, compute_milestone(DATA, None))
extra = haul_user(rows)
items = {i["key"]: i for i in haul_items(None, extra["haul_refs"], rows)}
assert (
"haul:3" in items and items["haul:3"]["name"] == "모래운반"
) # 자동 = 총괄자재 묶음 그대로
user = items["haul:r-u1"]
assert (user["section"], user["name"], user["qty"], user["ref"]) == (
"3",
"혼합골재운반",
None,
LOGIC,
)
assert items["haul:r-u1.1"]["name"] == "깬돌운반" and items["haul:r-u1.1"]["ref"] == LOGIC
assert extra["haul_dist"]["haul:r-u1"] == {"L1": 10, "L2": 0.8}