feat(B08): 유토곡선에 넘길 구조물 몫 — 채집석 공제·구조물 잔토를 한 자리에서 냄

- `B08_Quantity_Engine_HaulInputs.haul_inputs(unit_table)` — 둘 다 양수 ㎥,
  빼고 더하는 것은 받는 쪽(B06) 몫
- ⚠ 측점별로도 냄(`structure_spoil_points`) — 총량만 주면 잔량 비례로 흩어져
  운반거리가 틀어짐. 구조물은 구간이라 가운데 측점을 자리로 봄
- ⚠ 「아직 안 옴(None)」과 「없음(0)」을 가름
- 라우터에 `project_haul_inputs(project_id)` + `GET .../quantity/haul-inputs`
  — B06 이 이 함수를 부르면 됨(같은 계산을 두 벌로 짜지 않게)
- tmp/tests/test_b08_haul_inputs.py 신설(5건)

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
2026-09-08 22:37:22 +09:00
co-authored by Claude Opus 5
parent 7225bab2b2
commit 428fe7aeed
2 changed files with 94 additions and 0 deletions
@@ -0,0 +1,67 @@
"""유토곡선이 받아야 할 **구조물 몫** — 채집석 공제와 구조물 잔토 (2026-09-08).
⚠⚠ **B08 은 내기만 하고 빼거나 더하지 않는다.** 두 값 다 **양수 ㎥** 로 주고,
공제(빼기)와 사토 가산(더하기)은 **유토곡선(B06)에서 한 번씩만** 일어난다.
부호를 넘기면 받는 쪽에서 두 번 뒤집힌다 — 실무 시트가 `−274.66` 으로 적혀 있어
실제로 겪은 자리다.
채집석 공제는 사토에서 한 번만 뺀다.
구조물 잔토는 사토에 한 번만 더한다.
⚠ **측점별로도 낸다.** 총량 하나만 주면 받는 쪽이 잔량 크기에 비례해 나눌 수밖에 없고,
그러면 **운반거리가 틀어진다**(한 곳에 몰리면 거리가 어긋남 — 공제 때 이미 짚은 자리).
구조물은 구간(start~end)이라 **그 가운데 측점**을 자리로 본다.
"""
from __future__ import annotations
from typing import Any
from B08_Quantity.B08_Quantity_Engine_UnitQuantity import COLLECTED_STONE_KEY
#: 유토곡선이 읽는 칸 이름 — 양쪽이 같은 낱말을 써야 인계에서 어긋나지 않는다.
STRUCTURE_SPOIL_KEY = "structure_spoil_m3"
STRUCTURE_SPOIL_POINTS_KEY = "structure_spoil_points"
SPOIL_COMPONENT = "잔토처리"
def _center(structure: dict[str, Any]) -> float | None:
start, end = structure.get("start_m"), structure.get("end_m")
if start is None and end is None:
return None
values = [float(v) for v in (start, end) if v is not None]
return sum(values) / len(values)
def haul_inputs(unit_quantity_table: dict[str, Any] | None) -> dict[str, Any]:
"""(채집석 공제, 구조물 잔토, 측점별 잔토). 값이 없으면 `None` — 0 으로 눅이지 않는다.
⚠ `None` 과 `0.0` 은 다르다. 「아직 안 옴」과 「없음」을 받는 쪽이 갈라 봐야 한다.
"""
if not unit_quantity_table:
return {
COLLECTED_STONE_KEY: None,
STRUCTURE_SPOIL_KEY: None,
STRUCTURE_SPOIL_POINTS_KEY: [],
}
total = 0.0
points: list[dict[str, float]] = []
for structure in unit_quantity_table.get("structures") or []:
amount = 0.0
for component in structure.get("components") or []:
if str(component.get("name") or "") != SPOIL_COMPONENT:
continue
amount += float(component.get("amount") or 0.0)
if amount <= 0:
continue
total += amount
chainage = _center(structure)
if chainage is not None:
points.append({"chainage_m": chainage, "spoil_m3": round(amount, 3)})
collected = unit_quantity_table.get(COLLECTED_STONE_KEY)
return {
COLLECTED_STONE_KEY: collected,
STRUCTURE_SPOIL_KEY: round(total, 3) if points or total else None,
STRUCTURE_SPOIL_POINTS_KEY: sorted(points, key=lambda row: row["chainage_m"]),
}
@@ -42,6 +42,7 @@ from common_util.common_util_project_settings import (
rock_classes,
rock_method,
)
from B08_Quantity.B08_Quantity_Engine_HaulInputs import haul_inputs
from common_util.common_util_storage import resolve_stored_project_path
from common_util.common_util_structure_lengths import structure_lengths
from config.config_db import run_with_connection
@@ -176,6 +177,32 @@ async def get_material_summary(project_id: UUID) -> JSONResponse:
)
async def project_haul_inputs(project_id: UUID) -> dict[str, Any]:
"""유토곡선(B06)이 받아야 할 **구조물 몫** — 채집석 공제 · 구조물 잔토.
⚠ **B06 이 이 함수를 부르면 된다.** 두 값 다 B08 전개에서 나오는 것이라 저쪽이 다시
세면 같은 계산이 두 벌이 된다(CLAUDE.md 5장). 값은 **양수 ㎥** 이고 빼고 더하는 것은
받는 쪽 몫이다. 못 읽으면 빈 값(`None`) — 0 으로 눅이지 않는다.
"""
try:
stored_path = await run_with_connection(get_project_storage_relative_path, project_id)
project_root = resolve_stored_project_path(stored_path)
structures, names, _skipped = _collect_structures(project_root)
unit_table = build_unit_table(
structures, names, await _section_modes(project_id), await _ground_types(project_id)
)
except Exception:
logger.exception("B08 유토곡선 입력 조회 실패: project_id=%s", project_id)
return haul_inputs(None)
return haul_inputs(unit_table)
@router.get("/{project_id}/quantity/haul-inputs")
async def get_haul_inputs(project_id: UUID) -> JSONResponse:
"""같은 값을 화면·다른 창이 볼 수 있게 낸 자리. 계산은 위 함수 한 벌이다."""
return JSONResponse(content=await project_haul_inputs(project_id))
@router.get("/{project_id}/quantity/handoff")
async def get_handoff(project_id: UUID) -> JSONResponse:
"""B09 로 넘길 두 벌 — 작업 공종 축과 자재 축 (일감 9).