eomsangdonandClaude Opus 5.5
1a8a16a99b
- 평면 눈금 굵은 13px · 종단 12px · 횡단 판 · B06 카드 · 상세 창 제목 = --color-warning(밝음 · 어두움) - 측점 갈아끼움 이름 = 서버 꼴 STA… (MC) - B08 토적표 줄 mc = 서버가 종단 파일 kind curve 로 실음 - test_507_3_mc_label Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
108 lines
4.6 KiB
Python
108 lines
4.6 KiB
Python
"""B08 토공 탭 길 (PLAN 61-3 · 69-2) — 읽기만, 아무 파일 · DB 에도 쓰지 않음.
|
||
|
||
GET /api/projects/{id}/quantity/earthwork
|
||
토적표 + 토공집계표 + 평균운반거리 · 종방향 무대운반거리
|
||
|
||
운반 Q · L = 종단 레코드에 저장된 유토곡선 배분(`mass_haul.haul_plan`) — B07 토적도와 같은 정본.
|
||
|
||
단면적 = B06 횡단 설계 저장값(`cross_sections.data.design`) · 노선 = 확정 노선(없으면 최신 노선) ·
|
||
토량환산계수 = 유토곡선과 같은 `mixed_conversion_factors(quantity_settings)` · 암 = 연암 · 경암 비율 가중 C ·
|
||
`rock_c` = 가중 전 리핑암 · 발파암 C(조작 중 브라우저가 다시 가중 · 125-1).
|
||
암 → 연암 · 경암 = 암 × 연암 %(`rock_ratios_pct` · 기본 90 · PLAN 122) — 토적표 줄 · 합계 · 토공집계표.
|
||
잔토처리 = 잔토 − 채집석 공제(자재 총괄 저장본 · `picked_deduction` · 160-5).
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import asyncio
|
||
import json
|
||
from pathlib import Path
|
||
from typing import Any
|
||
from uuid import UUID
|
||
|
||
from fastapi import APIRouter, HTTPException
|
||
|
||
from B06_Section.B06_Section_Repository import (
|
||
get_confirmed_route_context,
|
||
get_cross_section_designs,
|
||
get_longitudinal_section,
|
||
get_workflow_route_context,
|
||
)
|
||
from B08_Quantity.B08_Quantity_Engine_Earthwork import StationArea, build_table
|
||
from B08_Quantity.B08_Quantity_Engine_Earthwork_Summary import (
|
||
haul_lists,
|
||
split_rock_columns,
|
||
summary_rows,
|
||
)
|
||
from B08_Quantity.B08_Quantity_Router_Hydro import _project_root
|
||
from B08_Quantity.B08_Quantity_Router_Material import picked_deduction
|
||
from common_util.common_util_project_settings import (
|
||
earthwork_conversion_factors,
|
||
mixed_conversion_factors,
|
||
quantity_settings,
|
||
soft_rock_pct,
|
||
)
|
||
from config.config_db import get_db_pool
|
||
|
||
router = APIRouter(prefix="/api/projects", tags=["B08 Earthwork"])
|
||
|
||
|
||
def _mc_chainages(root: Path, section: dict[str, Any] | None) -> set[float]:
|
||
"""MC(곡선 가운데 · 종단 파일 `kind` curve) 누가거리 — 토적표 측점 뒤 「 (MC)」(507-3)."""
|
||
relative = (section or {}).get("longitudinal_file_path")
|
||
path = root / str(relative) if relative else None
|
||
if path is None or not path.is_file():
|
||
return set()
|
||
try:
|
||
stations = json.loads(path.read_text(encoding="utf-8")).get("stations") or []
|
||
except (OSError, ValueError, AttributeError):
|
||
return set()
|
||
return {
|
||
round(float(station["chainage_m"]), 3)
|
||
for station in stations
|
||
if isinstance(station, dict)
|
||
and station.get("kind") == "curve"
|
||
and isinstance(station.get("chainage_m"), (int, float))
|
||
}
|
||
|
||
|
||
@router.get("/{project_id}/quantity/earthwork")
|
||
async def earthwork_table(project_id: UUID) -> dict[str, Any]:
|
||
root = await _project_root(project_id)
|
||
async with get_db_pool().acquire() as connection:
|
||
context = await get_confirmed_route_context(connection, project_id)
|
||
confirmed = context is not None
|
||
context = context or await get_workflow_route_context(connection, project_id)
|
||
if not context:
|
||
raise HTTPException(status_code=404, detail="노선이 없습니다 — 노선 · 횡단 설계부터.")
|
||
route_id = int(context["route_id"])
|
||
rows = await get_cross_section_designs(connection, route_id)
|
||
section = await get_longitudinal_section(connection, project_id, route_id)
|
||
settings = await asyncio.to_thread(quantity_settings, root)
|
||
soft = soft_rock_pct(settings)
|
||
stations = [StationArea.from_design(row["chainage_m"], row["design"]) for row in rows]
|
||
table = build_table(stations, mixed_conversion_factors(settings))
|
||
curves = await asyncio.to_thread(_mc_chainages, root, section)
|
||
for row in table["rows"]:
|
||
if round(float(row["chainage_m"]), 3) in curves:
|
||
row["mc"] = True
|
||
for record in [*table["rows"], table["totals"]]:
|
||
split_rock_columns(record, soft)
|
||
mass_haul = ((section or {}).get("data") or {}).get("mass_haul") or {}
|
||
haul = haul_lists(mass_haul.get("haul_plan"))
|
||
deduct = await picked_deduction(project_id, root)
|
||
return {
|
||
**table,
|
||
"summary": summary_rows(table["totals"], haul, soft, deduct),
|
||
"rock_ratios_pct": {"연암": soft, "경암": 100.0 - soft},
|
||
"rock_c": {
|
||
kind: entry["compacted"]
|
||
for kind, entry in earthwork_conversion_factors(settings).items()
|
||
if kind != "soil"
|
||
},
|
||
"haul": haul,
|
||
"haul_saved": bool(mass_haul.get("haul_plan")),
|
||
"route_id": route_id,
|
||
"confirmed": confirmed,
|
||
}
|