eomsangdonandClaude Opus 5
b97cb4e2e6
- 전개가 structures.json 만 읽던 구멍 — pipe_points.json 도 읽음(관 자체는 관 줄이 셈) - 관 부설(품셈 12-11 m당)에 기슭막이 몫 없음 → 기슭막이는 전개 한 곳에서만 · 사유에 적음 - 안 적힌 벽 칸은 등록부 기본값(횡단도 그림과 같은 값) + 「기본값으로 섰음」 사유 - 구조물 목록에 남은 관 지점 종류 옛 저장분은 안 셈(두 번 방지) - 기슭막이 공종 = 형태로 돌쌓기 찰·메 코드 · 독립 기슭막이 한쪽 설치에 두 칸이 다르면 사유만 - 산출식 없는 세월교가 「터파기 줄 없음」 거짓 사유를 안 내게 · 구조물도 제원 저장이 관 지점 시설엔 안 먹힘을 알림 Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_016VBGFXB9AbJBwXP19z75Qq
622 lines
29 KiB
Python
622 lines
29 KiB
Python
"""B08 구조물 원단위·자재총괄 조회 라우터 (일감 6·7 · PLAN 8-2·8-6·8-7).
|
|
|
|
값은 어디서 오나
|
|
치수 정본은 **`structures.json` 하나**다(B05 가 주인). B08 은 자기 치수표를 들지 않고
|
|
그 제원을 읽어 전개할 뿐이다 — 도면은 H=1.5 인데 수량은 옛 치수로 도는 사고를 막는다.
|
|
|
|
⚠ `design_owner` 가 붙은 타입은 건너뛴다
|
|
측구가 그렇다 — 횡단 설계가 이미 터파기 단면적까지 셈하므로 구조물로 또 세면 **같은 것을
|
|
두 번 계상**한다(레지스트리 주석, 2026-09-07 조사). 건너뛴 것은 숨기지 않고 응답에 적는다.
|
|
|
|
⚠ 할증은 자재총괄 한 곳뿐이다 (㉠)
|
|
원단위표는 할증 **전** 값(`surcharge_applied: False`)으로 오고, 자재총괄이 한 번 붙인다.
|
|
응답에 두 깃발이 다 실리므로 화면·B09 가 어느 쪽 값인지 헷갈릴 일이 없다.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import logging
|
|
from pathlib import Path
|
|
from typing import Any
|
|
from uuid import UUID
|
|
|
|
from fastapi import APIRouter
|
|
from fastapi.responses import JSONResponse
|
|
from pydantic import BaseModel, ConfigDict, Field
|
|
|
|
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
|
|
from B08_Quantity.B08_Quantity_Engine_MaterialSummary import build_table as build_material_table
|
|
from B08_Quantity.B08_Quantity_Provenance import quantity_provenance
|
|
from B08_Quantity.B08_Quantity_Engine_StructureLibrary import project_templates
|
|
from B08_Quantity.B08_Quantity_Engine_UnitQuantity import build_table as build_unit_table
|
|
from B08_Quantity.B08_Quantity_Engine_UnitQuantity import (
|
|
ground_types_from_designs,
|
|
section_modes_from_designs,
|
|
)
|
|
from common_util.common_util_project_settings import (
|
|
concrete_placing_method,
|
|
quantity_settings,
|
|
rock_classes,
|
|
rock_method,
|
|
)
|
|
from B08_Quantity.B08_Quantity_Engine_HaulInputs import haul_inputs
|
|
from B08_Quantity.B08_Quantity_Engine_Preparation import frame_material_rows
|
|
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
|
|
|
|
logger = logging.getLogger(__name__)
|
|
router = APIRouter(prefix="/api/projects", tags=["B08 Quantity"])
|
|
|
|
|
|
def _collect_structures(
|
|
project_root: str,
|
|
) -> tuple[list[dict[str, Any]], dict[str, str], list[str]]:
|
|
"""전개 대상 구조물·타입명·건너뛴 사유를 함께 낸다.
|
|
|
|
⭐ 정본 둘을 다 읽음(A1, 2026-09-14) — `structures.json` + `pipe_points.json`(계곡 통과 시설).
|
|
앞서 뒤엣것을 안 읽어 집수정·날개벽·관 유입/유출 기슭막이·독립 기슭막이·물넘이포장이
|
|
원단위·인계·내역에 한 줄도 안 섰음.
|
|
⚠ 관 지점 종류(`managed_by`)가 `structures.json` 에 옛 저장분으로 남아 있어도 안 셈 —
|
|
관 지점 정본이 주인이라 두 번 세지 않음(구조물 집계표와 같은 규칙).
|
|
"""
|
|
from B08_Quantity.B08_Quantity_Engine_Pipe import facility_structures
|
|
from common_util.common_util_drainage_pipes import pipe_points_path_in, read_pipe_points_file
|
|
|
|
_revision, items = load_structures(project_root)
|
|
types = structure_type_map()
|
|
targets: list[dict[str, Any]] = []
|
|
names: dict[str, str] = {}
|
|
skipped: list[str] = []
|
|
for item in items:
|
|
payload = item.model_dump()
|
|
type_id = str(payload.get("type_id") or "")
|
|
definition = types.get(type_id)
|
|
if definition is None:
|
|
skipped.append(f"{type_id}: 레지스트리에 없는 타입")
|
|
continue
|
|
names[type_id] = definition.name
|
|
if definition.managed_by:
|
|
skipped.append(
|
|
f"{definition.name}: 관 지점 정본이 주인 — 구조물 목록 옛 저장분은 안 셈"
|
|
)
|
|
continue
|
|
if definition.design_owner:
|
|
skipped.append(
|
|
f"{definition.name}: {definition.design_owner} 가 이미 셈 — 중복 계상 방지"
|
|
)
|
|
continue
|
|
if definition.reference_only:
|
|
skipped.append(f"{definition.name}: 전문 상세설계 대상 — 배치까지만")
|
|
continue
|
|
if definition.group == "B":
|
|
# ⚠ B군(종단배수)은 **연장표**로 간다 — `common_util_structure_lengths` 가
|
|
# 겹친 구간을 합쳐 주기 때문이다. 구조물별로 세면 겹친 구간을 두 번 센다.
|
|
# 여기서 빼지 않으면 **같은 시설이 두 줄로** 나간다.
|
|
continue
|
|
targets.append(payload)
|
|
points = read_pipe_points_file(pipe_points_path_in(Path(project_root)))
|
|
for row in facility_structures([point.as_dict() for point in points]):
|
|
for type_id in (row["type_id"], row.get("attachment_parent_type")):
|
|
if type_id in types:
|
|
names[type_id] = types[type_id].name
|
|
targets.append(row)
|
|
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]:
|
|
"""측점별 단면유형(`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 section_modes_from_designs(designs)
|
|
|
|
|
|
@router.get("/{project_id}/quantity/material-summary")
|
|
async def get_material_summary(project_id: UUID) -> JSONResponse:
|
|
"""구조물 원단위와 자재총괄을 **한 응답**으로 낸다.
|
|
|
|
자재총괄은 원단위의 `material` 성분만 모은 것이라 따로 부르면 같은 전개를 두 번 돈다.
|
|
"""
|
|
try:
|
|
stored_path = await run_with_connection(get_project_storage_relative_path, project_id)
|
|
project_root = resolve_stored_project_path(stored_path)
|
|
except Exception:
|
|
logger.exception("B08 자재총괄 조회 실패(경로): project_id=%s", project_id)
|
|
return JSONResponse(
|
|
status_code=404,
|
|
content={"status": "error", "message": "프로젝트 저장 폴더를 찾지 못했습니다."},
|
|
)
|
|
|
|
try:
|
|
structures, names, skipped = _collect_structures(project_root)
|
|
except Exception:
|
|
logger.exception("B08 구조물 정본 읽기 실패: project_id=%s", project_id)
|
|
return JSONResponse(
|
|
status_code=500,
|
|
content={"status": "error", "message": "구조물 정본을 읽지 못했습니다."},
|
|
)
|
|
|
|
settings = quantity_settings(project_root)
|
|
unit_table = build_unit_table(
|
|
structures,
|
|
names,
|
|
await _section_modes(project_id),
|
|
await _ground_types(project_id),
|
|
settings.get("rubble_base_thickness_m"),
|
|
# 구조물도에서 고친 식 — 안 넘기면 원단위·자재총괄이 구조물도와 갈림(PLAN 3장 ⑤).
|
|
structure_formulas=settings.get("structure_formula_overrides"),
|
|
# 프로젝트에 박힌 양식(PLAN 4장) — 같은 까닭.
|
|
structure_templates=project_templates(project_root),
|
|
)
|
|
material_table = build_material_table(
|
|
unit_table,
|
|
supply_map=settings.get("material_supply") or {},
|
|
# 콘크리트 할증은 **레미콘일 때만** 붙는다 — 방식이 이름을 가른다(확정 3차 ⑥).
|
|
concrete_placing_method=settings.get("concrete_placing_method"),
|
|
)
|
|
# 묶음으로 서는 구조물의 조각을 화면에도 보인다 — 코드만으로는 사람이 검증 못 한다.
|
|
handoff = build_handoff(unit_quantity_table=unit_table)
|
|
composite = [
|
|
{
|
|
"name": row["name"],
|
|
"parts": row["composite_parts"],
|
|
"not_ready": row.get("composite_not_ready"),
|
|
}
|
|
for row in handoff["work_items"]
|
|
# 조각이 없어도(원단위 자체가 없어 못 세운 경우) 사유는 보여야 한다.
|
|
if row.get("composite_parts") or row.get("composite_not_ready")
|
|
]
|
|
body: dict[str, Any] = {
|
|
"unit_quantity": unit_table,
|
|
"material": material_table,
|
|
"composite": composite,
|
|
"skipped_structures": skipped,
|
|
"structure_count": len(structures),
|
|
}
|
|
# 근거 사전(PLAN 8-36 ④) — ⚠ **개발환경에서만** 실린다. 운영에서는 `None` 이라
|
|
# 칸 자체가 안 생긴다 — 화면에서 숨기는 것이 아니라 안 보내는 것이 요점이다.
|
|
provenance = quantity_provenance()
|
|
if provenance is not None:
|
|
body["provenance"] = provenance
|
|
return JSONResponse(content=body)
|
|
|
|
|
|
@router.get("/{project_id}/quantity/structure-summary")
|
|
async def get_structure_summary(project_id: UUID) -> JSONResponse:
|
|
"""**구조물 집계표**(PLAN 2장) — 측점별 한 줄 · 종류별 표 · 칸마다 출처(자동·사용자·라이브러리).
|
|
|
|
⚠ 값을 셈하지 않음 — 정본 둘(`structures.json`·`pipe_points.json`)과 B06 관 연장을 읽기만.
|
|
"""
|
|
from B08_Quantity.B08_Quantity_Engine_StructureSummary import (
|
|
USER_CELLS_KEY,
|
|
build_summary,
|
|
pipe_lengths_from_designs,
|
|
)
|
|
from B08_Quantity.B08_Quantity_Engine_StructureTemplate import template_of
|
|
from common_util.common_util_drainage_pipes import pipe_points_path_in, read_pipe_points_file
|
|
|
|
try:
|
|
stored_path = await run_with_connection(get_project_storage_relative_path, project_id)
|
|
project_root = resolve_stored_project_path(stored_path)
|
|
revision, items = load_structures(project_root)
|
|
points = read_pipe_points_file(pipe_points_path_in(Path(project_root)))
|
|
except Exception:
|
|
logger.exception("B08 구조물 집계표 — 정본 읽기 실패: project_id=%s", project_id)
|
|
return JSONResponse(
|
|
status_code=500,
|
|
content={"status": "error", "message": "구조물 정본을 읽지 못했습니다."},
|
|
)
|
|
types = structure_type_map()
|
|
embedded = project_templates(project_root)
|
|
templates = {type_id: template_of(type_id, embedded) or {} for type_id in types}
|
|
body = build_summary(
|
|
[item.model_dump() for item in items],
|
|
[point.as_dict() for point in points],
|
|
types,
|
|
templates,
|
|
pipe_lengths_from_designs(await _designs(project_id)),
|
|
quantity_settings(project_root).get(USER_CELLS_KEY) or {},
|
|
)
|
|
return JSONResponse(
|
|
content={"status": "success", "revision": revision, "project_id": str(project_id), **body}
|
|
)
|
|
|
|
|
|
async def _designs(project_id: UUID) -> list[dict[str, Any]]:
|
|
"""현재 노선의 저장된 횡단 설계. 못 읽으면 빈 목록 — 관 연장이 빈칸으로 섬(0 아님)."""
|
|
try:
|
|
context = await run_with_connection(get_workflow_route_context, project_id)
|
|
route_id = int((context or {}).get("route_id") or 0)
|
|
return await run_with_connection(get_cross_section_designs, route_id) if route_id else []
|
|
except Exception:
|
|
logger.exception("B08 구조물 집계표 — 횡단 설계 조회 실패: project_id=%s", project_id)
|
|
return []
|
|
|
|
|
|
class StructureSummaryEdit(BaseModel):
|
|
"""집계표 칸 하나 — 빈 값(null·"")은 그 칸을 지움."""
|
|
|
|
model_config = ConfigDict(extra="forbid")
|
|
|
|
id: str = Field(min_length=1, max_length=200)
|
|
key: str = Field(min_length=1, max_length=100)
|
|
value: float | str | None = None
|
|
|
|
|
|
class StructureSummaryEditRequest(BaseModel):
|
|
"""[저장] 한 번에 보내는 손 고침 — 읽어 간 구조물 판번호와 함께."""
|
|
|
|
model_config = ConfigDict(extra="forbid")
|
|
|
|
base_revision: int = Field(ge=0)
|
|
edits: list[StructureSummaryEdit] = Field(max_length=500)
|
|
|
|
|
|
def _summary_error(status: int, message: str) -> JSONResponse:
|
|
return JSONResponse(status_code=status, content={"status": "error", "message": message})
|
|
|
|
|
|
@router.put("/{project_id}/quantity/structure-summary")
|
|
async def put_structure_summary(
|
|
project_id: UUID, payload: StructureSummaryEditRequest
|
|
) -> JSONResponse:
|
|
"""집계표 손 고침을 **정본에 바로 씀**(브레인 판정) + 산출 조건에 「손댄 칸」 표.
|
|
|
|
⚠ 덮개층을 두지 않음 — `structures.json`·`pipe_points.json` 이 곧 도면·수량의 한 값(지침 5장).
|
|
⚠ 다 맞춰 본 뒤에 씀 — 한 칸이라도 틀리면 아무것도 안 씀(구조물 판번호가 어긋나도 409).
|
|
⛔ 놓기 칸(자리·길이·높이·관경)은 거절 — 시·종점과 한 벌이라 B05 몫.
|
|
"""
|
|
import asyncio
|
|
|
|
from B05_Profile.B05_Profile_Structures_Repository import (
|
|
StructureRevisionConflict,
|
|
save_structures,
|
|
)
|
|
from B05_Profile.B05_Profile_Structures_Schema import StructureInstance
|
|
from B08_Quantity.B08_Quantity_Engine_StructureSummary import (
|
|
USER_CELLS_KEY,
|
|
apply_edits,
|
|
pipe_row_id,
|
|
prune_marks,
|
|
)
|
|
from common_util.common_util_drainage_pipes import pipe_points_path_in
|
|
from common_util.common_util_json import atomic_write_json
|
|
from common_util.common_util_project_settings import save_section
|
|
|
|
try:
|
|
stored_path = await run_with_connection(get_project_storage_relative_path, project_id)
|
|
project_root = resolve_stored_project_path(stored_path)
|
|
revision, items = load_structures(project_root)
|
|
path = pipe_points_path_in(Path(project_root))
|
|
document = json.loads(path.read_text(encoding="utf-8")) if path.is_file() else {}
|
|
except Exception:
|
|
logger.exception("B08 구조물 집계표 저장 — 정본 읽기 실패: project_id=%s", project_id)
|
|
return _summary_error(500, "구조물 정본을 읽지 못했습니다.")
|
|
if revision != payload.base_revision:
|
|
return _summary_error(409, "그 사이 구조물이 바뀌었습니다 — 표를 다시 불러와 고칠 것.")
|
|
|
|
structures = [item.model_dump() for item in items]
|
|
points = [point for point in document.get("points") or [] if isinstance(point, dict)]
|
|
settings = quantity_settings(project_root)
|
|
try:
|
|
changed, marks = apply_edits(
|
|
structures,
|
|
points,
|
|
structure_type_map(),
|
|
[edit.model_dump() for edit in payload.edits],
|
|
settings.get(USER_CELLS_KEY) or {},
|
|
)
|
|
instances = [StructureInstance.model_validate(item) for item in structures]
|
|
except (LookupError, ValueError) as exc:
|
|
return _summary_error(422, str(exc))
|
|
|
|
new_revision = revision
|
|
pipe_ids = {pipe_row_id(point) for point in points}
|
|
try:
|
|
if any(row_id not in pipe_ids for row_id in changed):
|
|
new_revision = await asyncio.to_thread(
|
|
save_structures, project_root, instances, base_revision=revision
|
|
)
|
|
if any(row_id in pipe_ids for row_id in changed):
|
|
# 관 지점 파일은 옵션만 바꿔 **그대로** 씀 — 노선 지문·좌표·다른 칸은 손대지 않음.
|
|
await asyncio.to_thread(atomic_write_json, path, {**document, "points": points})
|
|
except StructureRevisionConflict:
|
|
return _summary_error(409, "그 사이 구조물이 바뀌었습니다 — 표를 다시 불러와 고칠 것.")
|
|
except ValueError as exc:
|
|
return _summary_error(422, str(exc))
|
|
await asyncio.to_thread(
|
|
save_section,
|
|
project_root,
|
|
"quantity",
|
|
{USER_CELLS_KEY: prune_marks(marks, structures, points)},
|
|
replace_keys=[USER_CELLS_KEY],
|
|
)
|
|
return JSONResponse(
|
|
content={"status": "success", "revision": new_revision, "changed_rows": len(changed)}
|
|
)
|
|
|
|
|
|
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)
|
|
settings = quantity_settings(project_root)
|
|
unit_table = build_unit_table(
|
|
structures,
|
|
names,
|
|
await _section_modes(project_id),
|
|
await _ground_types(project_id),
|
|
settings.get("rubble_base_thickness_m"),
|
|
structure_formulas=settings.get("structure_formula_overrides"),
|
|
structure_templates=project_templates(project_root),
|
|
)
|
|
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).
|
|
|
|
⚠ **한 벌로 합치지 않는다.** 내역 줄은 작업 공종이고 자재는 자재다.
|
|
자재에 공종코드를 붙이면 자재가 내역 줄로 오해된다(8-2 이중계상 함정).
|
|
|
|
⚠ 토공·운반은 토적표 라우터가 이미 만드는 표를 그대로 받는다 — 여기서 다시 계산하지
|
|
않는다. 같은 값을 두 벌로 짜지 않는다는 규칙(CLAUDE.md 5장)이 여기에도 걸린다.
|
|
"""
|
|
try:
|
|
stored_path = await run_with_connection(get_project_storage_relative_path, project_id)
|
|
project_root = resolve_stored_project_path(stored_path)
|
|
except Exception:
|
|
logger.exception("B08 인계 조회 실패(경로): project_id=%s", project_id)
|
|
return JSONResponse(
|
|
status_code=404,
|
|
content={"status": "error", "message": "프로젝트 저장 폴더를 찾지 못했습니다."},
|
|
)
|
|
|
|
structures, names, skipped = _collect_structures(project_root)
|
|
settings = quantity_settings(project_root)
|
|
modes = await _section_modes(project_id)
|
|
unit_table = build_unit_table(
|
|
structures,
|
|
names,
|
|
modes,
|
|
await _ground_types(project_id),
|
|
settings.get("rubble_base_thickness_m"),
|
|
structure_formulas=settings.get("structure_formula_overrides"),
|
|
structure_templates=project_templates(project_root),
|
|
)
|
|
material_table = build_material_table(
|
|
unit_table,
|
|
supply_map=settings.get("material_supply") or {},
|
|
concrete_placing_method=settings.get("concrete_placing_method"),
|
|
)
|
|
|
|
# 토공·운반 표는 토적표 라우터의 것을 그대로 쓴다 — 여기서 다시 만들지 않는다.
|
|
earthwork = await _earthwork_tables(project_id)
|
|
|
|
# 규준틀 재료 — **개소가 선 뒤에야 설 수 있어** 준비공 표를 받은 다음 자재 축에 얹는다.
|
|
# ⚠ 값은 **제안값(실무 관측)**이고 산출 조건에서 고칠 수 있다 — 그 사실이 줄 사유에 적힌다.
|
|
frame_rows = [
|
|
row
|
|
for row in ((earthwork.get("preparation") or {}).get("rows") or [])
|
|
if str(row.get("item") or "").endswith("규준틀")
|
|
]
|
|
material_table = build_material_table(
|
|
unit_table,
|
|
supply_map=settings.get("material_supply") or {},
|
|
concrete_placing_method=settings.get("concrete_placing_method"),
|
|
extra_materials=frame_material_rows(frame_rows, settings.get("frame_material") or {}),
|
|
)
|
|
|
|
handoff = build_handoff(
|
|
summary_table=earthwork.get("summary"),
|
|
haul_table=earthwork.get("haul"),
|
|
unit_quantity_table=unit_table,
|
|
material_table=material_table,
|
|
# 준비공·사방공 — 값이 서는 줄도, 못 내는 줄도 함께 넘긴다(빼면 빠진 줄이 안 보임).
|
|
preparation_table=earthwork.get("preparation"),
|
|
# B군 종단배수 — 겹침을 합친 연장. 그 규칙이 이미 그 함수에 있어 두 벌로 안 짠다.
|
|
length_table=[row for row in structure_lengths(project_root) if row.get("group") == "B"],
|
|
# 배수관 — 관 정본은 `pipe_points.json`, 연장은 측점 `design.pipe_length_m` 다.
|
|
pipe_table=_pipe_table(
|
|
project_root,
|
|
earthwork.get("pipe_lengths") or [],
|
|
earthwork.get("section_chainages") or [],
|
|
),
|
|
ground_class_set=settings.get("rock_class_set"),
|
|
ground_classes=rock_classes(settings),
|
|
ground_methods={name: rock_method(settings, name) for name in rock_classes(settings)},
|
|
# 타설 방식 — 안 정했으면 기본값으로 서되 그 사실을 `placing_notes` 가 알린다.
|
|
concrete_placing_method=concrete_placing_method(settings)[0],
|
|
# 층따기 길이 — 면적 × 이 값으로 ㎥ 를 낸다(확정 2차 ①). 안 넣었으면 막히고 사유가 감.
|
|
bench_cut_depth_m=settings.get("bench_cut_depth_m"),
|
|
# 용수 유무 — 기본 「육상」은 **통상값**이다(확정 3차 ④). 사유·화면에 그 사실이 뜬다.
|
|
structure_trench_water=settings.get("structure_trench_water"),
|
|
# 임목축적 등급 — 지장목제거 뿌리뽑기(제근 9-21)의 품 갈래. 안 넣으면 B09 가 후보를 보임.
|
|
stand_volume_class=settings.get("stand_volume_class"),
|
|
# 구조물도 양식 일위대가로 셀 장 — 그 구조물은 호표 `AX-ST` 줄 하나로(PLAN 6장 ②).
|
|
priced_sheets=_priced_sheets(project_root, unit_table, modes, settings),
|
|
)
|
|
handoff["summary"] = summarize(handoff)
|
|
handoff["skipped_structures"] = skipped
|
|
handoff["earthwork_available"] = bool(earthwork)
|
|
return JSONResponse(content=handoff)
|
|
|
|
|
|
def _priced_sheets(
|
|
project_root: str,
|
|
unit_table: dict[str, Any],
|
|
section_modes: dict[float, str] | None,
|
|
settings: dict[str, Any],
|
|
) -> list[dict[str, Any]]:
|
|
"""양식 일위대가로 셀 구조물도 장 — 구조물도 탭과 **같은 접기**(`build_standard_sheets` →
|
|
`apply_templates`)로 세움. 인계와 내역 금액이 이 한 목록을 봄."""
|
|
from B08_Quantity.B08_Quantity_Engine_StructurePriceLink import priced_sheets
|
|
from B08_Quantity.B08_Quantity_Engine_StructureSheet import build_standard_sheets
|
|
from B08_Quantity.B08_Quantity_Engine_StructureTemplate import apply_templates, template_of
|
|
from B08_Quantity.B08_Quantity_Engine_StructureUnitPrice import ROWS_KEY
|
|
|
|
embedded = project_templates(project_root)
|
|
payload = build_standard_sheets(unit_table, section_modes)
|
|
apply_templates(payload, settings, embedded)
|
|
sheets = payload.get("sheets") or []
|
|
templates = {
|
|
type_id: template
|
|
for type_id in {str(sheet.get("type_id") or "") for sheet in sheets}
|
|
if (template := template_of(type_id, embedded))
|
|
}
|
|
return priced_sheets(sheets, templates, settings.get(ROWS_KEY))
|
|
|
|
|
|
async def structure_bill_prices(project_id: UUID, build: Any) -> dict[str, dict[str, Any]]:
|
|
"""B09 내역이 부름 — 구조물도 호표 금액. **B09 가 넘긴 단가표**로 B08 일위대가 엔진을 돌림.
|
|
|
|
⚠ 금액 정본은 B09 — 여기서 단가표를 따로 짓지 않음. 수동 단가(프로젝트 값)도 입력으로 씀.
|
|
⚠ 인계(`get_handoff`)와 **같은 전개**를 한 번 더 돎 — 두 요청이라 한 벌을 못 나눔.
|
|
"""
|
|
# ponytail: 인계와 전개를 두 번 돎 — 느려지면 인계 응답에 장 목록을 실어 한 번으로.
|
|
from B08_Quantity.B08_Quantity_Engine_StructureLibrary import available_templates
|
|
from B08_Quantity.B08_Quantity_Engine_StructurePriceLink import price_sheets
|
|
from B08_Quantity.B08_Quantity_Engine_StructureUnitPrice import MANUAL_KEY, ROWS_KEY, with_rows
|
|
from B09_Estimation.B09_Estimation_UnitPrice import find_variant_code
|
|
|
|
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)
|
|
settings = quantity_settings(project_root)
|
|
modes = await _section_modes(project_id)
|
|
unit_table = build_unit_table(
|
|
structures,
|
|
names,
|
|
modes,
|
|
await _ground_types(project_id),
|
|
settings.get("rubble_base_thickness_m"),
|
|
structure_formulas=settings.get("structure_formula_overrides"),
|
|
structure_templates=project_templates(project_root),
|
|
)
|
|
entries = _priced_sheets(project_root, unit_table, modes, settings)
|
|
if not entries:
|
|
return {}
|
|
row_edits = settings.get(ROWS_KEY) or {}
|
|
library = [
|
|
with_rows(template, row_edits.get(template.get("type_id")))
|
|
for template in available_templates(project_root)
|
|
]
|
|
return price_sheets(
|
|
entries,
|
|
build.book,
|
|
lambda code, value: find_variant_code(code, value, build),
|
|
library,
|
|
settings.get(MANUAL_KEY) or {},
|
|
)
|
|
|
|
|
|
def _pipe_table(
|
|
project_root: str,
|
|
pipe_lengths: list[dict[str, Any]],
|
|
section_chainages: list[Any] | None = None,
|
|
) -> dict[str, Any]:
|
|
"""배수관 표 — 정본 셋을 읽어 잇는다. 못 읽으면 **빈 표**(줄이 안 서는 것이 정직하다).
|
|
|
|
⚠ 관 정본은 `structures.json` 이 아니라 `pipe_points.json` 이다
|
|
(레지스트리 `pipe` 타입이 `managed_by: pipe_points`).
|
|
"""
|
|
from B08_Quantity.B08_Quantity_Engine_Pipe import build_rows as build_pipe_rows
|
|
from common_util.common_util_drainage_pipes import pipe_points_path_in
|
|
|
|
path = pipe_points_path_in(Path(project_root))
|
|
if not path.is_file():
|
|
return {"rows": [], "notes": [], "pipe_count": 0, "ready_count": 0, "length_total_m": 0.0}
|
|
try:
|
|
payload = json.loads(path.read_text(encoding="utf-8"))
|
|
except (OSError, ValueError):
|
|
logger.exception("B08 관 지점 읽기 실패: %s", path)
|
|
return {
|
|
"rows": [],
|
|
"notes": ["관 지점 파일을 읽지 못했습니다"],
|
|
"pipe_count": 0,
|
|
"ready_count": 0,
|
|
"length_total_m": 0.0,
|
|
}
|
|
points = payload.get("points") or payload.get("items") or []
|
|
# 토적표 라우터가 실어 준 모양을 엔진이 읽는 모양으로 옮긴다.
|
|
designs = [
|
|
{"chainage_m": row.get("chainage_m"), "design": {"pipe_length_m": row.get("pipe_length_m")}}
|
|
for row in pipe_lengths
|
|
]
|
|
return build_pipe_rows(
|
|
points,
|
|
designs,
|
|
(load_mapping().pipe or {}),
|
|
[float(x) for x in (section_chainages or []) if x is not None],
|
|
)
|
|
|
|
|
|
async def _earthwork_tables(project_id: UUID) -> dict[str, Any]:
|
|
"""토적표 라우터가 만든 집계·운반 표를 얻는다. 노선이 없으면 빈 값."""
|
|
from B08_Quantity.B08_Quantity_Router_Earthwork import (
|
|
get_earthwork_table_for_current_route,
|
|
)
|
|
|
|
try:
|
|
response = await get_earthwork_table_for_current_route(project_id)
|
|
except Exception:
|
|
logger.exception("B08 인계 — 토적표 조회 실패: project_id=%s", project_id)
|
|
return {}
|
|
if response.status_code != 200:
|
|
return {}
|
|
import json as _json
|
|
|
|
return _json.loads(bytes(response.body).decode("utf-8"))
|