From b48ab7047c1cfb749dfb333a3b497ed75c891ba3 Mon Sep 17 00:00:00 2001 From: umsangdon Date: Mon, 7 Sep 2026 20:11:53 +0900 Subject: [PATCH] =?UTF-8?q?feat(B08):=20=ED=86=A0=EC=A0=81=ED=91=9C=20?= =?UTF-8?q?=EC=97=94=EC=A7=84=C2=B7API=20=E2=80=94=20=ED=8F=89=EA=B7=A0?= =?UTF-8?q?=EB=8B=A8=EB=A9=B4=EC=A0=81=EB=B2=95=20=EC=B2=B4=EC=A0=81?= =?UTF-8?q?=ED=99=94?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit PLAN 8-4b 열 명세대로. B06 이 이미 낸 측점별 단면적을 다시 재지 않고 체적화만 함 — 새 수량을 낳지 않으므로 캐시·조작 경로 불필요(CLAUDE.md 5장). 열 구성 (실무 토적표 = 오솔길 1.BOM 36열과 1:1) 측점·거리·절토[토사·암 각 단면적/입적/보정량]·측구터파기[토사·암 각 3칸] ·보정량계·성토[단면적/입적]·유용토·차인토량·누가토량. 사면 계열(층따기·면고르기·법면보호공·지장목제거)은 사면길이가 아직 없어 일감 3. 계산 규칙 체적 = (앞 단면적 + 현 단면적)/2 x 거리. 첫 측점은 앞이 없어 체적 없음. 보정량 = 체적 x 다짐 환산계수 — 정의처는 EARTHWORK_CONVERSION_FACTORS 한 곳, 여기서 값을 다시 적지 않음(토사 0.90 / 리핑암 1.15 / 발파암 1.30). 값을 자르지 않음 — 품셈 1-2-2 는 표기 규칙이고 절사는 화면 몫(PLAN 8-16). 원가 쪽(줄마다 원 단위 절사)과 규칙이 반대라 섞지 말 것. TODO(미결 · PLAN 8-4b) — 설계가 측구를 토사·암으로 안 나눠 줌(ditch_area_m2 한 값). 잠정으로 그 측점 절토 토사:암 면적비로 안분함. 측구는 절토부에 파므로 같은 지반을 만난다는 것이 근거. 설계가 측구 지반을 따로 내면 _split_ditch 만 교체. API — GET /api/projects/{id}/quantity/{route_id}/earthwork-table, 경로 생략형은 워크플로 최신 노선으로. main.py 는 자기 두 줄만 추가. 검증 — tmp/tests/test_b08_earthwork_table.py 14건 통과. 거창 실무 BOM 실측값 재현(단면적 1.89 → 체적 9.45 → 보정 8.505, 다음 측점 18.10 → 16.29, 측구 0.18㎡ → 10m당 1.80 → 1.62). 공용 브라우저에서 실 API 호출로 route 150·측점 65곳 확인 — 측점 20 에서 (0.2723+1.9615)/2x20 = 22.338, x0.9 = 20.1042, 암 25.427 x1.15 = 29.24105, 측구 안분 0.0784+0.1016 = 0.18 로 전건 일치. Co-Authored-By: Claude Opus 5 (1M context) --- .../B08_Quantity_Engine_EarthworkTable.py | 209 ++++++++++++++++++ B08_Quantity/B08_Quantity_Router_Earthwork.py | 64 ++++++ main.py | 2 + 3 files changed, 275 insertions(+) create mode 100644 B08_Quantity/B08_Quantity_Engine_EarthworkTable.py create mode 100644 B08_Quantity/B08_Quantity_Router_Earthwork.py diff --git a/B08_Quantity/B08_Quantity_Engine_EarthworkTable.py b/B08_Quantity/B08_Quantity_Engine_EarthworkTable.py new file mode 100644 index 00000000..5cec5f9c --- /dev/null +++ b/B08_Quantity/B08_Quantity_Engine_EarthworkTable.py @@ -0,0 +1,209 @@ +"""토적표 — 측점별 단면적을 평균단면적법으로 체적화한다 (B08 일감 2 · PLAN 8-4b). + +무엇을 만드나 + 실무 토적표의 열 구성 그대로다. 거창 실무 워크북 `토적표` 시트와 오솔길 `1.BOM` 36열이 + 서로 1:1 로 맞물리는 것을 확인해 열 이름을 그대로 옮겼다(PLAN 8-4b). + + 측점 · 거리 · 절토[토사·암 각 (단면적·입적·보정량)] · 측구터파기[토사·암 각 3칸] + · 보정량계 · 성토[단면적·입적] · 유용토 · 차인토량 · 누가토량 + + 사면 계열(층따기·면고르기·법면보호공·지장목제거)은 사면길이가 아직 없어 일감 3에서 붙인다. + +평균단면적법 (신규 문서 5장 「다. 공사수량의 산출」) + 체적 = (앞 측점 단면적 + 현 측점 단면적) ÷ 2 × 두 측점 사이 거리. + 첫 측점은 앞이 없으므로 체적이 없다(거창 실무 토적표도 첫 행 체적이 비어 있다). + +보정량 = 체적 × 토량환산계수(다짐) + 절취한 흙이 다져지면 줄거나 부푼다. 성토에 쓸 수 있는 양으로 환산한 것이 보정량이다. + 계수의 유일한 정의처는 `config.config_system_design.EARTHWORK_CONVERSION_FACTORS` 이며 + 여기서 값을 다시 적지 않는다. + +⚠ 숫자는 자르지 않는다 (PLAN 8-16) + 품셈 1-2-2 의 소수 자리는 **표기 규칙**이다. 계산은 전정밀로 두고 화면·출력에서만 + 반올림한다. 원가 쪽(줄마다 원 단위 절사)과 규칙이 반대이므로 섞지 말 것. +""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from typing import Any, Iterable + +from config.config_system_design import EARTHWORK_CONVERSION_FACTORS + +# 절토 암을 어느 환산계수로 볼지 — 측점의 `cut_rock_kind` 를 그대로 쓴다. +# 값이 없으면 리핑암으로 본다(발파암보다 보수적으로 적은 쪽). +_DEFAULT_ROCK_KIND = "ripping_rock" + + +def _factor(kind: str) -> float: + """지반유형 → 다짐 환산계수. 모르는 유형이면 토사로 본다.""" + entry = EARTHWORK_CONVERSION_FACTORS.get(kind) or EARTHWORK_CONVERSION_FACTORS["soil"] + return float(entry["compacted"]) + + +@dataclass(slots=True) +class StationArea: + """토적표 한 줄이 필요로 하는 측점 값. B06 설계 결과에서 그대로 옮겨 담는다.""" + + chainage_m: float + cut_soil_area_m2: float = 0.0 + cut_rock_area_m2: float = 0.0 + fill_area_m2: float = 0.0 + ditch_area_m2: float = 0.0 + cut_rock_kind: str | None = None + + @classmethod + def from_design(cls, chainage_m: float, design: dict[str, Any]) -> "StationArea": + def num(key: str) -> float: + value = design.get(key) + return float(value) if isinstance(value, (int, float)) else 0.0 + + return cls( + chainage_m=float(chainage_m), + cut_soil_area_m2=num("cut_soil_area_m2"), + cut_rock_area_m2=num("cut_rock_area_m2"), + fill_area_m2=num("fill_area_m2"), + ditch_area_m2=num("ditch_area_m2"), + cut_rock_kind=design.get("cut_rock_kind") or None, + ) + + +@dataclass(slots=True) +class EarthworkRow: + """토적표 한 줄. 열 이름은 실무 토적표(PLAN 8-4b)를 따른다.""" + + chainage_m: float + distance_m: float = 0.0 + cut_soil_area_m2: float = 0.0 + cut_soil_volume_m3: float = 0.0 + cut_soil_adjusted_m3: float = 0.0 + cut_rock_area_m2: float = 0.0 + cut_rock_volume_m3: float = 0.0 + cut_rock_adjusted_m3: float = 0.0 + ditch_soil_area_m2: float = 0.0 + ditch_soil_volume_m3: float = 0.0 + ditch_soil_adjusted_m3: float = 0.0 + ditch_rock_area_m2: float = 0.0 + ditch_rock_volume_m3: float = 0.0 + ditch_rock_adjusted_m3: float = 0.0 + adjusted_total_m3: float = 0.0 + fill_area_m2: float = 0.0 + fill_volume_m3: float = 0.0 + diverted_m3: float = 0.0 + balance_m3: float = 0.0 + cumulative_m3: float = 0.0 + notes: list[str] = field(default_factory=list) + + +def _split_ditch(area: StationArea) -> tuple[float, float]: + """측구터파기 단면적을 토사·암으로 가른다. + + ⚠ TODO(미결 · PLAN 8-4b) — 설계가 측구를 토사·암으로 나눠 주지 않는다(`ditch_area_m2` + 한 값뿐). 실무 토적표는 둘로 갈라 적으므로, **그 측점의 절토 토사:암 면적비로 안분**한다. + 측구는 절토부에 파므로 같은 지반을 만난다는 것이 근거다. 설계가 측구 지반을 따로 내주게 + 되면 이 함수만 갈아끼운다. + """ + ditch = area.ditch_area_m2 + if ditch <= 0: + return 0.0, 0.0 + soil, rock = area.cut_soil_area_m2, area.cut_rock_area_m2 + total = soil + rock + if total <= 0: + return ditch, 0.0 # 절토가 없으면 토사로 본다. + return ditch * soil / total, ditch * rock / total + + +def build_rows(stations: Iterable[StationArea]) -> list[EarthworkRow]: + """측점 목록 → 토적표 줄 목록. 측점은 이정 순으로 정렬해 받는다.""" + ordered = sorted(stations, key=lambda s: s.chainage_m) + rows: list[EarthworkRow] = [] + previous: StationArea | None = None + previous_ditch: tuple[float, float] = (0.0, 0.0) + cumulative = 0.0 + + for station in ordered: + ditch_soil, ditch_rock = _split_ditch(station) + soil_factor = _factor("soil") + rock_factor = _factor(station.cut_rock_kind or _DEFAULT_ROCK_KIND) + row = EarthworkRow( + chainage_m=station.chainage_m, + cut_soil_area_m2=station.cut_soil_area_m2, + cut_rock_area_m2=station.cut_rock_area_m2, + ditch_soil_area_m2=ditch_soil, + ditch_rock_area_m2=ditch_rock, + fill_area_m2=station.fill_area_m2, + ) + if previous is not None: + distance = station.chainage_m - previous.chainage_m + row.distance_m = distance + + def mean_volume(before: float, now: float) -> float: + return (before + now) / 2.0 * distance + + row.cut_soil_volume_m3 = mean_volume( + previous.cut_soil_area_m2, station.cut_soil_area_m2 + ) + row.cut_rock_volume_m3 = mean_volume( + previous.cut_rock_area_m2, station.cut_rock_area_m2 + ) + row.ditch_soil_volume_m3 = mean_volume(previous_ditch[0], ditch_soil) + row.ditch_rock_volume_m3 = mean_volume(previous_ditch[1], ditch_rock) + row.fill_volume_m3 = mean_volume(previous.fill_area_m2, station.fill_area_m2) + + row.cut_soil_adjusted_m3 = row.cut_soil_volume_m3 * soil_factor + row.cut_rock_adjusted_m3 = row.cut_rock_volume_m3 * rock_factor + row.ditch_soil_adjusted_m3 = row.ditch_soil_volume_m3 * soil_factor + row.ditch_rock_adjusted_m3 = row.ditch_rock_volume_m3 * rock_factor + + row.adjusted_total_m3 = ( + row.cut_soil_adjusted_m3 + + row.cut_rock_adjusted_m3 + + row.ditch_soil_adjusted_m3 + + row.ditch_rock_adjusted_m3 + ) + # 유용토 = 그 측점에서 절취분과 성토분이 서로 만나는 몫. + row.diverted_m3 = min(row.adjusted_total_m3, row.fill_volume_m3) + row.balance_m3 = row.adjusted_total_m3 - row.fill_volume_m3 + cumulative += row.balance_m3 + row.cumulative_m3 = cumulative + + rows.append(row) + previous = station + previous_ditch = (ditch_soil, ditch_rock) + return rows + + +def totals(rows: list[EarthworkRow]) -> dict[str, float]: + """합계 행. 단면적은 합이 뜻이 없어 싣지 않는다(실무 토적표도 비워 둔다).""" + keys = ( + "distance_m", + "cut_soil_volume_m3", + "cut_soil_adjusted_m3", + "cut_rock_volume_m3", + "cut_rock_adjusted_m3", + "ditch_soil_volume_m3", + "ditch_soil_adjusted_m3", + "ditch_rock_volume_m3", + "ditch_rock_adjusted_m3", + "adjusted_total_m3", + "fill_volume_m3", + "diverted_m3", + "balance_m3", + ) + return {key: sum(getattr(row, key) for row in rows) for key in keys} + + +def build_table(stations: Iterable[StationArea]) -> dict[str, Any]: + """화면·API 가 그대로 쓰는 모양. 값은 자르지 않는다(PLAN 8-16).""" + rows = build_rows(stations) + return { + "method": "average_end_area", + "conversion_factors": EARTHWORK_CONVERSION_FACTORS, + "rows": [row.__dict__ if not hasattr(row, "__slots__") else _as_dict(row) for row in rows], + "totals": totals(rows), + "station_count": len(rows), + } + + +def _as_dict(row: EarthworkRow) -> dict[str, Any]: + return {name: getattr(row, name) for name in EarthworkRow.__slots__} diff --git a/B08_Quantity/B08_Quantity_Router_Earthwork.py b/B08_Quantity/B08_Quantity_Router_Earthwork.py new file mode 100644 index 00000000..ff91d278 --- /dev/null +++ b/B08_Quantity/B08_Quantity_Router_Earthwork.py @@ -0,0 +1,64 @@ +"""B08 토적표 조회 라우터 (일감 2 · PLAN 8-4b). + +값은 어디서 오나 + 측점별 단면적은 **B06 이 이미 낸 정본**이다(`cross_sections.data.design` 의 + `cut_soil_area_m2`·`cut_rock_area_m2`·`fill_area_m2`·`ditch_area_m2`). + B08 은 그것을 다시 재지 않고 **평균단면적법으로 체적화만** 한다. + +계산 자리 (CLAUDE.md 5장) + 초기값은 서버가 한 번 계산해 영구저장한다. 여기서는 저장된 단면적을 읽어 표를 만든다 — + 새 수량을 낳지 않으므로 캐시·조작 경로가 따로 필요 없다. +""" + +from __future__ import annotations + +import logging +from typing import Any +from uuid import UUID + +from fastapi import APIRouter +from fastapi.responses import JSONResponse + +from B06_Section.B06_Section_Repository import ( + get_cross_section_designs, + get_workflow_route_context, +) +from B08_Quantity.B08_Quantity_Engine_EarthworkTable import StationArea, build_table +from config.config_db import run_with_connection + +logger = logging.getLogger(__name__) +router = APIRouter(prefix="/api/projects", tags=["B08 Quantity"]) + + +def _stations(designs: list[dict[str, Any]]) -> list[StationArea]: + return [ + StationArea.from_design(item["chainage_m"], item.get("design") or {}) for item in designs + ] + + +@router.get("/{project_id}/quantity/{route_id}/earthwork-table") +async def get_earthwork_table(project_id: UUID, route_id: int) -> JSONResponse: + """토적표 — 측점별 단면적을 평균단면적법으로 체적화한 표.""" + try: + designs = await run_with_connection(get_cross_section_designs, route_id) + except Exception: + logger.exception("B08 토적표 조회 실패: project_id=%s route_id=%s", project_id, route_id) + return JSONResponse( + status_code=500, + content={"status": "error", "message": "토적표를 만들지 못했습니다."}, + ) + table = build_table(_stations(designs)) + table["route_id"] = route_id + return JSONResponse(content=table) + + +@router.get("/{project_id}/quantity/earthwork-table") +async def get_earthwork_table_for_current_route(project_id: UUID) -> JSONResponse: + """경로를 안 주면 워크플로가 보고 있는 경로로 낸다 — 화면이 route_id 를 모를 때 쓴다.""" + context = await run_with_connection(get_workflow_route_context, project_id) + if not context or not context.get("route_id"): + return JSONResponse( + status_code=404, + content={"status": "error", "message": "이 프로젝트에 확정된 노선이 없습니다."}, + ) + return await get_earthwork_table(project_id, int(context["route_id"])) diff --git a/main.py b/main.py index 6088b1f7..4add704f 100644 --- a/main.py +++ b/main.py @@ -59,6 +59,7 @@ from B06_Section.B06_Section_Router_HaulPlan import ( from B07_DesignDetail.B07_DesignDetail_Router import router as b07_design_router from B07_DesignDetail.B07_DesignDetail_Router_Frame import router as b07_frame_router from B08_Quantity.B08_Quantity_Router import router as b08_quantity_router +from B08_Quantity.B08_Quantity_Router_Earthwork import router as b08_earthwork_router from common_util.common_util_audit import note_api_call, record_call_burst from common_util.common_util_auth import ( require_company, @@ -535,6 +536,7 @@ app.include_router(b06_section_haul_plan_router, dependencies=protected_with_com app.include_router(b07_design_router, dependencies=protected_with_company) app.include_router(b07_frame_router, dependencies=protected_with_company) app.include_router(b08_quantity_router, dependencies=protected_with_company) +app.include_router(b08_earthwork_router, dependencies=protected_with_company) # ─────────────────────────────────────────────────────────────────────────