feat(B08): 구조물터파기 토질을 측점 설계값에서 끌어옴 — 남은 축은 용수 하나
- 새 저장 칸을 만들지 않음. 측점마다 이미 있는 `design.ground_type` (soil·ripping_rock·blasting_rock)이 품셈 9-13 토질 3구분과 그대로 맞물림 - 판정 규칙: 구조물이 **걸친 측점 전부**를 보고 갈래가 하나일 때만 값을 냄. 섞이면 다수결로 고르지 않고 갈래별 측점 수를 사유에 적음(암 단가가 몇 배라 임의 선택이 금액으로 굳음). 걸친 측점이 없으면 가장 가까운 측점 + 그 사실을 근거에 - 터파기 줄이 토질·심도로 갈려 서고, 막힌 사유가 「용수 유무」 하나로 좁혀짐 - tmp/tests 2건 추가(측점값 반영·섞임 판정) Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
@@ -25,6 +25,7 @@ from B08_Quantity.B08_Quantity_Engine_Handoff_Mapping import (
|
|||||||
ORIGIN_EARTHWORK,
|
ORIGIN_EARTHWORK,
|
||||||
WorkItemMapping,
|
WorkItemMapping,
|
||||||
)
|
)
|
||||||
|
from B08_Quantity.B08_Quantity_Engine_UnitQuantity import GROUND_TYPE_LABEL
|
||||||
from common_util.common_util_excavation import (
|
from common_util.common_util_excavation import (
|
||||||
WALL_BLINDING_DEPTH_M,
|
WALL_BLINDING_DEPTH_M,
|
||||||
WALL_FOUNDATION_DEPTH_M,
|
WALL_FOUNDATION_DEPTH_M,
|
||||||
@@ -38,10 +39,14 @@ TRENCH_GROUP = "구조물터파기"
|
|||||||
BACKFILL_GROUP = "되메우기"
|
BACKFILL_GROUP = "되메우기"
|
||||||
SPOIL_GROUP = "잔토처리"
|
SPOIL_GROUP = "잔토처리"
|
||||||
|
|
||||||
TRENCH_BLOCKED_REASON = (
|
#: ⚠ 남은 축은 **용수 하나**다(2026-09-08). 토질은 측점 설계값(`design.ground_type`)에서
|
||||||
"토질(토사·암절취·발파암)과 용수 유무가 저장 제원에 없어 품셈 9-13 의 18구분 중 어느 칸인지"
|
#: 끌어오고, 심도는 구조물 제원(직고 + 기초 깊이)에서 나온다. 용수는 저장에 칸이 없어
|
||||||
" 못 고름 — 심도만 구조물 제원(직고 + 기초 깊이)에서 갈라 둠"
|
#: **입력 칸이 서야 하는 자리**다 — 모른다고 「육상」으로 눅이지 않는다.
|
||||||
|
WATER_BLOCKED_REASON = (
|
||||||
|
"용수 유무가 저장에 없어 품셈 9-13 의 18구분 중 육상·용수 어느 쪽인지 못 고름"
|
||||||
|
" — 모른다고 육상으로 눅이지 않음"
|
||||||
)
|
)
|
||||||
|
GROUND_BLOCKED_REASON = "토질을 못 가름"
|
||||||
SPOIL_REASON = (
|
SPOIL_REASON = (
|
||||||
"사토로 실어 내는 몫이라 유토곡선(B06)이 세야 겹치지 않음 — ⚠ 지금 그 통로가 없어"
|
"사토로 실어 내는 몫이라 유토곡선(B06)이 세야 겹치지 않음 — ⚠ 지금 그 통로가 없어"
|
||||||
" 어디에도 안 실림. 값을 버리지 않고 사유와 함께 넘김"
|
" 어디에도 안 실림. 값을 버리지 않고 사유와 함께 넘김"
|
||||||
@@ -106,13 +111,15 @@ def structure_earthwork_rows(
|
|||||||
unit_quantity_table: dict[str, Any], mapping: WorkItemMapping
|
unit_quantity_table: dict[str, Any], mapping: WorkItemMapping
|
||||||
) -> tuple[list[dict[str, Any]], list[str]]:
|
) -> tuple[list[dict[str, Any]], list[str]]:
|
||||||
"""구조물이 낸 터파기·되메우기·잔토를 **공종 축**으로 올린다."""
|
"""구조물이 낸 터파기·되메우기·잔토를 **공종 축**으로 올린다."""
|
||||||
trench: dict[str, float] = {}
|
trench: dict[tuple[str, str | None, str], float] = {}
|
||||||
backfill = 0.0
|
backfill = 0.0
|
||||||
spoil = 0.0
|
spoil = 0.0
|
||||||
for structure in unit_quantity_table.get("structures") or []:
|
for structure in unit_quantity_table.get("structures") or []:
|
||||||
options = structure.get("options") or {}
|
options = structure.get("options") or {}
|
||||||
height = float(structure.get("height_m") or options.get("height_m") or 0.0)
|
height = float(structure.get("height_m") or options.get("height_m") or 0.0)
|
||||||
band = _depth_band(height + _foundation_depth(options))
|
band = _depth_band(height + _foundation_depth(options))
|
||||||
|
ground = structure.get("ground_type") or None
|
||||||
|
ground_basis = str(structure.get("ground_type_basis") or "")
|
||||||
for component in structure.get("components") or []:
|
for component in structure.get("components") or []:
|
||||||
if str(component.get("destination") or "") != "earthwork":
|
if str(component.get("destination") or "") != "earthwork":
|
||||||
continue
|
continue
|
||||||
@@ -121,7 +128,8 @@ def structure_earthwork_rows(
|
|||||||
if amount <= 0:
|
if amount <= 0:
|
||||||
continue
|
continue
|
||||||
if name == "터파기":
|
if name == "터파기":
|
||||||
trench[band] = trench.get(band, 0.0) + amount
|
key = (band, ground, ground_basis)
|
||||||
|
trench[key] = trench.get(key, 0.0) + amount
|
||||||
elif name == "되메우기":
|
elif name == "되메우기":
|
||||||
backfill += amount
|
backfill += amount
|
||||||
elif name == "잔토처리":
|
elif name == "잔토처리":
|
||||||
@@ -134,21 +142,28 @@ def structure_earthwork_rows(
|
|||||||
code = (entry or {}).get("work_item_code")
|
code = (entry or {}).get("work_item_code")
|
||||||
if code is None:
|
if code is None:
|
||||||
unmatched.append(TRENCH_GROUP)
|
unmatched.append(TRENCH_GROUP)
|
||||||
for _, band in (*DEPTH_BANDS, (0.0, DEPTH_OVER)):
|
order = [band for _, band in DEPTH_BANDS] + [DEPTH_OVER]
|
||||||
amount = trench.get(band)
|
for (band, ground, ground_basis), amount in sorted(
|
||||||
|
trench.items(), key=lambda item: (order.index(item[0][0]), str(item[0][1] or ""))
|
||||||
|
):
|
||||||
if not amount:
|
if not amount:
|
||||||
continue
|
continue
|
||||||
|
label = GROUND_TYPE_LABEL.get(str(ground), str(ground)) if ground else None
|
||||||
|
reason = WATER_BLOCKED_REASON if ground else f"{GROUND_BLOCKED_REASON} — {ground_basis}"
|
||||||
rows.append(
|
rows.append(
|
||||||
_row(
|
_row(
|
||||||
work_item_code=code,
|
work_item_code=code,
|
||||||
name=TRENCH_GROUP,
|
name=TRENCH_GROUP,
|
||||||
spec=f"심도 {band}",
|
spec=f"{label} · 심도 {band}" if label else f"심도 {band}",
|
||||||
spec_detail=f"구조물 전개 합 · 심도 {band}",
|
ground_class=label,
|
||||||
|
spec_detail=f"구조물 전개 합 · {ground_basis}"
|
||||||
|
if ground_basis
|
||||||
|
else "구조물 전개 합",
|
||||||
quantity=amount,
|
quantity=amount,
|
||||||
blocked_kind=BLOCKED_INPUT_MISSING,
|
blocked_kind=BLOCKED_INPUT_MISSING,
|
||||||
blocked_reason=TRENCH_BLOCKED_REASON,
|
blocked_reason=reason,
|
||||||
in_bill=False,
|
in_bill=False,
|
||||||
in_bill_reason=TRENCH_BLOCKED_REASON,
|
in_bill_reason=reason,
|
||||||
)
|
)
|
||||||
)
|
)
|
||||||
if backfill > 0:
|
if backfill > 0:
|
||||||
|
|||||||
@@ -342,6 +342,10 @@ class StructureQuantity:
|
|||||||
#: 비어 있으면 종전대로 「m · 연장」으로 선다.
|
#: 비어 있으면 종전대로 「m · 연장」으로 선다.
|
||||||
billing_unit: str = ""
|
billing_unit: str = ""
|
||||||
billing_quantity: float = 0.0
|
billing_quantity: float = 0.0
|
||||||
|
#: 구조물이 놓인 자리의 지반 갈래(`soil`·`ripping_rock`·`blasting_rock`)와 그 판정 근거.
|
||||||
|
#: 품셈 9-13 구조물터파기의 **토질 축**이 이 값으로 갈린다. 못 가르면 `None` 이다.
|
||||||
|
ground_type: str | None = None
|
||||||
|
ground_type_basis: str = ""
|
||||||
|
|
||||||
|
|
||||||
#: 표준경사표 — 품셈 13-4-4 [주]⑪. 파일이 없으면 종전 기본값(0.3)으로 돈다.
|
#: 표준경사표 — 품셈 13-4-4 [주]⑪. 파일이 없으면 종전 기본값(0.3)으로 돈다.
|
||||||
@@ -1097,6 +1101,68 @@ def section_modes_from_designs(designs: Iterable[dict[str, Any]]) -> dict[float,
|
|||||||
return modes
|
return modes
|
||||||
|
|
||||||
|
|
||||||
|
#: 저장된 지반 갈래 ↔ 품셈 9-13 토질 3구분. **새 칸을 만들지 않는다** — 측점마다 이미
|
||||||
|
#: `design.ground_type` 이 저장돼 있고(재생성 사고 때 이 값이 비어 B08 이 통째로 0 이 됐던
|
||||||
|
#: 그 키다), 값 셋이 품셈 구분과 그대로 맞물린다(2026-09-08 조율 창 확인).
|
||||||
|
GROUND_TYPE_LABEL = {
|
||||||
|
"soil": "토사",
|
||||||
|
"ripping_rock": "암절취",
|
||||||
|
"blasting_rock": "발파암",
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
|
def ground_types_from_designs(designs: Iterable[dict[str, Any]]) -> dict[float, str]:
|
||||||
|
"""저장된 횡단 설계 목록 → `{측점: 지반갈래}`. 값이 없는 측점은 담지 않는다."""
|
||||||
|
grounds: dict[float, str] = {}
|
||||||
|
for item in designs or ():
|
||||||
|
design = item.get("design") if isinstance(item, dict) else None
|
||||||
|
ground = str((design or {}).get("ground_type") or "").strip()
|
||||||
|
if not ground:
|
||||||
|
continue
|
||||||
|
grounds[float(_num(item.get("chainage_m")))] = ground
|
||||||
|
return grounds
|
||||||
|
|
||||||
|
|
||||||
|
def ground_type_at(
|
||||||
|
structure: dict[str, Any], ground_types: dict[float, str] | None
|
||||||
|
) -> tuple[str | None, str]:
|
||||||
|
"""(토질, 근거). 구조물이 **걸친 측점 전부**를 보고 갈래가 하나일 때만 값을 낸다.
|
||||||
|
|
||||||
|
⚠ **판정 규칙 — 섞이면 안 고른다.** 구조물은 구간(start~end)이고 지반은 측점 값이라
|
||||||
|
한 구조물이 토사 측점과 암 측점에 걸칠 수 있다. 그때 다수결로 한쪽을 고르면 **임의값이
|
||||||
|
금액으로 굳는다**(암 단가가 몇 배다). 섞였다는 사실과 갈래별 측점 수를 근거에 적고
|
||||||
|
값은 `None` 으로 둔다 — 성절토·용수에서 지킨 그대로다.
|
||||||
|
⚠ 걸친 측점이 하나도 없으면(구간이 측점 사이에 통째로 들어간 짧은 구조물) **가장 가까운
|
||||||
|
측점**을 쓴다 — 그 사실도 근거에 적는다.
|
||||||
|
"""
|
||||||
|
if not ground_types:
|
||||||
|
return None, "측점 지반 갈래가 저장에 없어 못 가름"
|
||||||
|
start, end = _num(structure.get("start_m")), _num(structure.get("end_m"))
|
||||||
|
if end < start:
|
||||||
|
start, end = end, start
|
||||||
|
inside = {
|
||||||
|
chainage: kind for chainage, kind in ground_types.items() if start <= float(chainage) <= end
|
||||||
|
}
|
||||||
|
if not inside:
|
||||||
|
center = (start + end) / 2.0
|
||||||
|
nearest = min(ground_types, key=lambda chainage: abs(float(chainage) - center))
|
||||||
|
kind = ground_types[nearest]
|
||||||
|
return (
|
||||||
|
kind,
|
||||||
|
f"걸친 측점이 없어 가장 가까운 측점({nearest:g}m)의 {GROUND_TYPE_LABEL.get(kind, kind)}",
|
||||||
|
)
|
||||||
|
counts: dict[str, int] = {}
|
||||||
|
for kind in inside.values():
|
||||||
|
counts[kind] = counts.get(kind, 0) + 1
|
||||||
|
if len(counts) == 1:
|
||||||
|
kind = next(iter(counts))
|
||||||
|
return kind, f"걸친 측점 {len(inside)}곳이 모두 {GROUND_TYPE_LABEL.get(kind, kind)}"
|
||||||
|
breakdown = " · ".join(
|
||||||
|
f"{GROUND_TYPE_LABEL.get(kind, kind)} {count}곳" for kind, count in sorted(counts.items())
|
||||||
|
)
|
||||||
|
return None, f"걸친 측점의 지반이 섞여 못 가름 — {breakdown}"
|
||||||
|
|
||||||
|
|
||||||
def _section_mode_at(
|
def _section_mode_at(
|
||||||
structure: dict[str, Any], section_modes: dict[float, str] | None
|
structure: dict[str, Any], section_modes: dict[float, str] | None
|
||||||
) -> str | None:
|
) -> str | None:
|
||||||
@@ -1120,6 +1186,7 @@ def build_table(
|
|||||||
structures: Iterable[dict[str, Any]],
|
structures: Iterable[dict[str, Any]],
|
||||||
names: dict[str, str] | None = None,
|
names: dict[str, str] | None = None,
|
||||||
section_modes: dict[float, str] | None = None,
|
section_modes: dict[float, str] | None = None,
|
||||||
|
ground_types: dict[float, str] | None = None,
|
||||||
) -> dict[str, Any]:
|
) -> dict[str, Any]:
|
||||||
"""화면·API 가 그대로 쓰는 모양. 성분별 총량과 구조물별 내역을 함께 낸다."""
|
"""화면·API 가 그대로 쓰는 모양. 성분별 총량과 구조물별 내역을 함께 낸다."""
|
||||||
observed = load_observed_table()
|
observed = load_observed_table()
|
||||||
@@ -1128,10 +1195,11 @@ def build_table(
|
|||||||
for item in structures:
|
for item in structures:
|
||||||
expanded_inputs.append(item)
|
expanded_inputs.append(item)
|
||||||
expanded_inputs.extend(attachments_of(item))
|
expanded_inputs.extend(attachments_of(item))
|
||||||
quantities = [
|
quantities = []
|
||||||
expand(item, names, observed, _section_mode_at(item, section_modes))
|
for item in expanded_inputs:
|
||||||
for item in expanded_inputs
|
quantity = expand(item, names, observed, _section_mode_at(item, section_modes))
|
||||||
]
|
quantity.ground_type, quantity.ground_type_basis = ground_type_at(item, ground_types)
|
||||||
|
quantities.append(quantity)
|
||||||
violations = verify_no_mix_components(quantities)
|
violations = verify_no_mix_components(quantities)
|
||||||
|
|
||||||
totals: dict[str, dict[str, Any]] = {}
|
totals: dict[str, dict[str, Any]] = {}
|
||||||
@@ -1163,6 +1231,10 @@ def build_table(
|
|||||||
"billing_quantity": item.billing_quantity,
|
"billing_quantity": item.billing_quantity,
|
||||||
# 저장된 제원 — 형식(반중력식…)처럼 **뒤 단계가 읽어야 하는** 값이 여기 있다.
|
# 저장된 제원 — 형식(반중력식…)처럼 **뒤 단계가 읽어야 하는** 값이 여기 있다.
|
||||||
"options": item.options,
|
"options": item.options,
|
||||||
|
# 구조물이 놓인 자리의 지반 갈래 — **품셈 9-13 토질 3구분**이 이 값으로 갈린다.
|
||||||
|
# ⚠ 여기서 새로 만드는 값이 아니라 측점 설계값(`design.ground_type`)을 옮긴 것이다.
|
||||||
|
"ground_type": item.ground_type,
|
||||||
|
"ground_type_basis": item.ground_type_basis,
|
||||||
"notes": item.notes,
|
"notes": item.notes,
|
||||||
"components": [
|
"components": [
|
||||||
{
|
{
|
||||||
|
|||||||
@@ -32,7 +32,10 @@ from B05_Profile.B05_Profile_Structures_Schema import structure_type_map
|
|||||||
from B08_Quantity.B08_Quantity_Engine_Handoff import build_handoff, load_mapping, summarize
|
from B08_Quantity.B08_Quantity_Engine_Handoff import build_handoff, load_mapping, summarize
|
||||||
from B08_Quantity.B08_Quantity_Engine_MaterialSummary import build_table as build_material_table
|
from B08_Quantity.B08_Quantity_Engine_MaterialSummary import build_table as build_material_table
|
||||||
from B08_Quantity.B08_Quantity_Engine_UnitQuantity import build_table as build_unit_table
|
from B08_Quantity.B08_Quantity_Engine_UnitQuantity import build_table as build_unit_table
|
||||||
from B08_Quantity.B08_Quantity_Engine_UnitQuantity import section_modes_from_designs
|
from B08_Quantity.B08_Quantity_Engine_UnitQuantity import (
|
||||||
|
ground_types_from_designs,
|
||||||
|
section_modes_from_designs,
|
||||||
|
)
|
||||||
from common_util.common_util_project_settings import (
|
from common_util.common_util_project_settings import (
|
||||||
concrete_placing_method,
|
concrete_placing_method,
|
||||||
quantity_settings,
|
quantity_settings,
|
||||||
@@ -81,6 +84,24 @@ def _collect_structures(
|
|||||||
return targets, names, sorted(set(skipped))
|
return targets, names, sorted(set(skipped))
|
||||||
|
|
||||||
|
|
||||||
|
async def _ground_types(project_id: UUID) -> dict[float, str]:
|
||||||
|
"""측점별 지반 갈래(`soil`·`ripping_rock`·`blasting_rock`).
|
||||||
|
|
||||||
|
구조물터파기(품셈 9-13)의 **토질 축**이 이 값으로 갈린다. 단면유형과 같은 자리에서
|
||||||
|
오므로 읽는 방식도 같다 — 못 읽으면 빈 표로 두고 판정이 「못 가름」이 되게 한다.
|
||||||
|
"""
|
||||||
|
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 ground_types_from_designs(designs)
|
||||||
|
|
||||||
|
|
||||||
async def _section_modes(project_id: UUID) -> dict[float, str]:
|
async def _section_modes(project_id: UUID) -> dict[float, str]:
|
||||||
"""측점별 단면유형(`left_cut` 등). 구조물이 **성토면인가 절토면인가**를 가릴 때 쓴다.
|
"""측점별 단면유형(`left_cut` 등). 구조물이 **성토면인가 절토면인가**를 가릴 때 쓴다.
|
||||||
|
|
||||||
@@ -124,7 +145,9 @@ async def get_material_summary(project_id: UUID) -> JSONResponse:
|
|||||||
content={"status": "error", "message": "구조물 정본을 읽지 못했습니다."},
|
content={"status": "error", "message": "구조물 정본을 읽지 못했습니다."},
|
||||||
)
|
)
|
||||||
|
|
||||||
unit_table = build_unit_table(structures, names, await _section_modes(project_id))
|
unit_table = build_unit_table(
|
||||||
|
structures, names, await _section_modes(project_id), await _ground_types(project_id)
|
||||||
|
)
|
||||||
settings = quantity_settings(project_root)
|
settings = quantity_settings(project_root)
|
||||||
material_table = build_material_table(
|
material_table = build_material_table(
|
||||||
unit_table,
|
unit_table,
|
||||||
@@ -174,7 +197,9 @@ async def get_handoff(project_id: UUID) -> JSONResponse:
|
|||||||
)
|
)
|
||||||
|
|
||||||
structures, names, skipped = _collect_structures(project_root)
|
structures, names, skipped = _collect_structures(project_root)
|
||||||
unit_table = build_unit_table(structures, names, await _section_modes(project_id))
|
unit_table = build_unit_table(
|
||||||
|
structures, names, await _section_modes(project_id), await _ground_types(project_id)
|
||||||
|
)
|
||||||
settings = quantity_settings(project_root)
|
settings = quantity_settings(project_root)
|
||||||
material_table = build_material_table(
|
material_table = build_material_table(
|
||||||
unit_table, supply_map=settings.get("material_supply") or {}
|
unit_table, supply_map=settings.get("material_supply") or {}
|
||||||
|
|||||||
Reference in New Issue
Block a user