diff --git a/B08_Quantity/B08_Quantity_Engine_EarthworkSummary.py b/B08_Quantity/B08_Quantity_Engine_EarthworkSummary.py index 37b5c20a..36c3afcd 100644 --- a/B08_Quantity/B08_Quantity_Engine_EarthworkSummary.py +++ b/B08_Quantity/B08_Quantity_Engine_EarthworkSummary.py @@ -68,17 +68,24 @@ def _ratio(source: SummaryInput, key: str) -> float: return float(value) if isinstance(value, (int, float)) else 1.0 -def _split_by_rock(total: float, source: SummaryInput) -> list[tuple[str, float]]: - """암 총량을 설계자가 준 비율(%)로 갈래별로 나눈다. +def _split_by_rock(total: float, source: SummaryInput) -> list[tuple[str, float, str]]: + """암 총량을 설계자가 준 비율(%)로 갈래별로 나눈다. `(이름, 물량, 비고)`. 비율이 아직 없으면 **나누지 않고 「암」 한 줄로** 낸다 — 지어낸 비율로 쪼개지 않는다. + + ⚠ 합이 100 이 아니어도 **총량은 보존**한다 — 준 비율끼리 안분한다. 물량이 조용히 + 사라지면 안 되기 때문이다. 다만 값이 말없이 바뀌는 것이므로 **비고에 드러낸다** + (60/30 을 넣으면 실제로는 66.7/33.3 으로 돈다). """ classes = [name for name in source.rock_classes if name != "토사"] ratios = {name: float(source.rock_ratios_pct.get(name, 0) or 0) for name in classes} given = sum(ratios.values()) if given <= 0: - return [("암", total)] - return [(name, total * ratios[name] / given) for name in classes if ratios[name] > 0] + return [("암", total, "")] + note = "" if abs(given - 100.0) < 1e-9 else f"입력 합 {given:g} % → 100 % 로 안분" + return [ + (name, total * ratios[name] / given, note) for name in classes if ratios[name] > 0 + ] def build_rows(source: SummaryInput) -> list[SummaryRow]: @@ -97,8 +104,12 @@ def build_rows(source: SummaryInput) -> list[SummaryRow]: group=group, item="토사", spec="기계(굴삭기)", amount=earth.get(soil_key, 0.0) ) ) - for name, amount in _split_by_rock(earth.get(rock_key, 0.0), source): - rows.append(SummaryRow(group=group, item=name, spec="굴삭기+브레카", amount=amount)) + for name, amount, note in _split_by_rock(earth.get(rock_key, 0.0), source): + rows.append( + SummaryRow( + group=group, item=name, spec="굴삭기+브레카", amount=amount, note=note + ) + ) rows.append(SummaryRow(group="보정량계", amount=earth.get("adjusted_total_m3", 0.0))) rows.append(SummaryRow(group="성토", amount=earth.get("fill_volume_m3", 0.0))) diff --git a/B08_Quantity/B08_Quantity_Router_Earthwork.py b/B08_Quantity/B08_Quantity_Router_Earthwork.py index 509ef116..e9b9b314 100644 --- a/B08_Quantity/B08_Quantity_Router_Earthwork.py +++ b/B08_Quantity/B08_Quantity_Router_Earthwork.py @@ -12,20 +12,35 @@ from __future__ import annotations +import asyncio import logging from typing import Any from uuid import UUID from fastapi import APIRouter from fastapi.responses import JSONResponse +from pydantic import BaseModel +from B03_FileInput.B03_FileInput_Repository import get_project_storage_relative_path from B06_Section.B06_Section_Repository import ( get_cross_section_designs, + get_longitudinal_section, get_workflow_route_context, ) +from B08_Quantity.B08_Quantity_Engine_EarthworkSummary import SummaryInput +from B08_Quantity.B08_Quantity_Engine_EarthworkSummary import build_table as build_summary_table from B08_Quantity.B08_Quantity_Engine_EarthworkTable import StationArea, build_table +from B08_Quantity.B08_Quantity_Engine_HaulSummary import build_table as build_haul_table +from B08_Quantity.B08_Quantity_Engine_HaulSummary import summary_input_rows from B08_Quantity.B08_Quantity_Engine_SlopeArea import build_table as build_slope_table from B08_Quantity.B08_Quantity_Engine_SlopeLength import station_slopes +from common_util.common_util_project_settings import ( + application_ratio, + quantity_settings, + rock_classes, + save_section, +) +from common_util.common_util_storage import resolve_stored_project_path from config.config_db import run_with_connection logger = logging.getLogger(__name__) @@ -54,12 +69,98 @@ async def get_earthwork_table(project_id: UUID, route_id: int) -> JSONResponse: content={"status": "error", "message": "토적표를 만들지 못했습니다."}, ) table = build_table(_stations(designs)) - # 사면 계열은 저장된 설계선에서 유도한다 — 반영률은 기본 100 %(설계자 입력은 후속). - table["slope"] = build_slope_table(station_slopes(designs)) + # 사면 계열은 저장된 설계선에서 유도한다. + slope = build_slope_table(station_slopes(designs)) + table["slope"] = slope + + settings, project_root = await _project_settings(project_id) + plan = await _stored_haul_plan(project_id, route_id) + haul = build_haul_table(plan) + table["haul"] = haul + # 운반계획은 [저장]·[확정]에서 정본에 남는 값이다 — 아직 없으면 빈 표가 정직하다. + table["haul_available"] = bool(plan) + + table["summary"] = build_summary_table( + SummaryInput( + earthwork_totals=table.get("totals") or {}, + slope_totals=slope.get("totals") or {}, + haul_rows=summary_input_rows(haul), + rock_classes=rock_classes(settings), + rock_ratios_pct=settings.get("rock_ratios_pct") or {}, + application_ratios={ + key: application_ratio(settings, key) + for key in (settings.get("application_ratios_pct") or {}) + }, + ) + ) + table["settings"] = settings + table["project_root_known"] = project_root is not None table["route_id"] = route_id return JSONResponse(content=table) +async def _project_settings(project_id: UUID) -> tuple[dict[str, Any], str | None]: + """프로젝트 설정을 읽는다. 경로를 못 찾아도 기본값으로 화면은 선다.""" + try: + stored_path = await run_with_connection(get_project_storage_relative_path, project_id) + root = resolve_stored_project_path(stored_path) + except Exception: + logger.warning("B08 프로젝트 경로를 못 찾음: project_id=%s", project_id) + from common_util.common_util_project_settings import default_settings + + return default_settings()["quantity"], None + return quantity_settings(root), root + + +async def _stored_haul_plan(project_id: UUID, route_id: int) -> dict[str, Any] | None: + """정본에 남은 운반계획. [확정]을 아직 안 돌렸으면 없다.""" + try: + row = await run_with_connection(get_longitudinal_section, project_id, route_id) + except Exception: + logger.exception("B08 운반계획 조회 실패: route_id=%s", route_id) + return None + data = (row or {}).get("data") or {} + plan = data.get("mass_haul") if isinstance(data, dict) else None + return plan if isinstance(plan, dict) and plan else None + + +class QuantitySettingsBody(BaseModel): + """[저장]이 보내는 산출 조건. 보내지 않은 칸은 저장분을 그대로 둔다.""" + + rock_class_set: str | None = None + rock_classes: list[str] | None = None + rock_ratios_pct: dict[str, float] | None = None + application_ratios_pct: dict[str, float] | None = None + + +@router.put("/{project_id}/quantity/settings") +async def save_quantity_settings(project_id: UUID, body: QuantitySettingsBody) -> JSONResponse: + """산출 조건을 정본에 남긴다 — [저장]이 부르는 자리. + + ⚠ 자동저장이 아니다(CLAUDE.md 5장). 조작은 캐시에 쌓이고 여기서만 작업본으로 넘어간다. + ⚠ `quantity` 구획만 쓴다 — `estimation` 은 B09 것이라 손대지 않는다(모듈이 막고 있다). + """ + try: + stored_path = await run_with_connection(get_project_storage_relative_path, project_id) + 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": "프로젝트 저장 폴더를 찾지 못했습니다."}, + ) + values = {key: value for key, value in body.model_dump().items() if value is not None} + try: + saved = await asyncio.to_thread(save_section, root, "quantity", values) + except Exception: + logger.exception("B08 설정 저장 실패(쓰기): project_id=%s", project_id) + return JSONResponse( + status_code=500, + content={"status": "error", "message": "산출 조건을 저장하지 못했습니다."}, + ) + return JSONResponse(content={"status": "success", "quantity": saved.get("quantity") or {}}) + + @router.get("/{project_id}/quantity/earthwork-table") async def get_earthwork_table_for_current_route(project_id: UUID) -> JSONResponse: """경로를 안 주면 워크플로가 보고 있는 경로로 낸다 — 화면이 route_id 를 모를 때 쓴다."""