diff --git a/B07_DesignDetail/B07_DesignDetail_Quantity_Proto.py b/B07_DesignDetail/B07_DesignDetail_Quantity_Proto.py new file mode 100644 index 00000000..a8e315c3 --- /dev/null +++ b/B07_DesignDetail/B07_DesignDetail_Quantity_Proto.py @@ -0,0 +1,592 @@ +"""B07 수량 산출 프로토타입 (처분 가능 — disposable). + +사용자 평가용 시안이다. 이 파일 + `B07_DesignDetail_Router.py`의 include 2줄이 전부라 +마음에 안 들면 그대로 지우면 된다. 정본을 읽기만 하고 어디에도 쓰지 않는다. + +페이지: GET /api/projects/quantity-proto/page — 프로젝트 목록 + GET /api/projects/quantity-proto/page/{id} — 수량 산출 보고서 + +구조물 출처(정본 3곳, CLAUDE.md 5장): + ① pipe_points.json — 계곡 통과 시설(배수관·BOX암거·물넘이·세월교·독립 기슭막이) + ② structures.json — 일반 구조물(C·D·E군) + ③ 횡단 설계 patch(DB) — 기준벽 조작·다단 벽(extra_wall_counts·extra_spans) + +원단위 출처: 실무 관측 **참조 전용** — 채용 미확정(개발 단계 사용자 협의 대상). + · 울진1 수량집계 13종 (resources/knowledge/original/실무문서/_종합비교/04_임도구조물_원단위.md) + · 울진소광 구조도 숨김탭 (resources/knowledge/original/실무문서/_원단위라이브러리_울진소광.md) +""" + +import json +import logging +from collections import defaultdict +from pathlib import Path +from typing import Any +from uuid import UUID + +from fastapi import APIRouter +from fastapi.responses import HTMLResponse + +from B01_Dashboard.B01_Dashboard_Repository import list_all_projects +from B03_FileInput.B03_FileInput_Repository import get_project_storage_relative_path +from B05_Profile.B05_Profile_Structures_Repository import load_structures +from B06_Section.B06_Section_Repository import ( + get_confirmed_route_context, + get_cross_section_designs, + get_longitudinal_section, +) +from common_util.common_util_drainage_pipes import parse_pipe_points +from common_util.common_util_storage import resolve_stored_project_path +from config.config_db import get_db_pool + +logger = logging.getLogger(__name__) + +proto_router = APIRouter() + +# ---------------------------------------------------------------- 타입 이름표 + +_FACILITY_NAMES = { + "pipe": "배수관", + "box_culvert": "BOX암거", + "ford_pavement": "물넘이포장", + "ford_bridge": "세월교", + "revetment": "기슭막이(독립)", +} + +_REGISTRY_PATH = Path(__file__).parent.parent / "B05_Profile" / "B05_Profile_Structure_Types.json" +_type_names: dict[str, str] | None = None + + +def _type_name(type_id: str) -> str: + global _type_names + if _type_names is None: + try: + payload = json.loads(_REGISTRY_PATH.read_text(encoding="utf-8")) + _type_names = {t["type_id"]: t["name"] for t in payload.get("types", [])} + except (OSError, ValueError, KeyError): + _type_names = {} + return _type_names.get(type_id, type_id) + + +# ---------------------------------------------------------------- 원단위 라이브러리 +# ⚠ 참조 전용(과거 공사지 관측값). 프로그램 기본값 아님 — 채용은 사용자 협의로 확정. +# items: {공종명(단위): 단위수량} + +_SRC_ULJIN1 = "울진1 수량집계" +_SRC_SOGWANG = "울진소광 구조도 숨김탭" + +UNIT_RATES: dict[tuple[str, str], dict[str, Any]] = { + ("basin", "Ø800"): { + "label": "집수정 Ø800", + "per": "개소", + "source": _SRC_SOGWANG, + "items": { + "터파기(㎥)": 10.64, + "되메우기(㎥)": 6.44, + "잔토처리(㎥)": 4.20, + "콘크리트(㎥)": 2.84, + "이형철근 D13(kg)": 4.78, + "면목 A25(m)": 12.67, + "거푸집(㎡)": 21.28, + }, + }, + ("erosion_check", "찰쌓기·평균"): { + "label": "돌골막이(찰쌓기) — 임도 실측 6개소 평균", + "per": "개소", + "source": _SRC_SOGWANG, + "note": "관측 높이 2.5~3.0m는 사방기술교본 '2m 이내' 초과 — 판정 없이 기록(지식DB 주석 그대로)", + "items": { + "돌쌓기(㎡)": 19.62, + "돌붙임(㎡)": 2.63, + "야면석(ton)": 19.58, + "고임돌(㎥)": 3.34, + "막자갈(㎥)": 10.54, + "채움콘크리트(㎥)": 4.45, + "물구멍(m)": 4.91, + "바닥파기(㎥)": 8.61, + "되메우기(㎥)": 5.87, + "잔토처리(㎥)": 2.74, + }, + }, + ("retaining_wall", "반중력식 H=2.0"): { + "label": "반중력식옹벽 H=2.0", + "per": "m", + "source": _SRC_SOGWANG, + "items": { + "콘크리트(㎥)": 1.35, + "버림콘크리트(㎥)": 0.15, + "거푸집-유로폼(㎡)": 3.20, + "거푸집-기초(㎡)": 0.60, + "물빼기 Ø50(m)": 0.32, + "이형철근 D13(kg)": 13.45, + "이형철근 D16(kg)": 30.42, + }, + }, + ("revet_wet", "H=1.5"): { + "label": "돌기슭막이 찰쌓기 H=1.5", + "per": "m", + "source": _SRC_ULJIN1, + "items": { + "콘크리트(㎥)": 0.31, + "모르터(㎥)": 0.014, + "돌쌓기(㎡)": 1.57, + "석재(ton)": 0.91, + "터파기(㎥)": 1.55, + "되메우기(㎥)": 0.30, + "잔토처리(㎥)": 1.25, + }, + }, + ("revet_wet", "H=2.0"): { + "label": "돌기슭막이 찰쌓기 H=2.0", + "per": "m", + "source": _SRC_ULJIN1, + "items": { + "콘크리트(㎥)": 0.42, + "모르터(㎥)": 0.019, + "돌쌓기(㎡)": 2.09, + "석재(ton)": 0.91, + "터파기(㎥)": 2.06, + "되메우기(㎥)": 0.40, + "잔토처리(㎥)": 1.66, + }, + }, + ("revet_wet", "H=2.5"): { + "label": "돌기슭막이 찰쌓기 H=2.5", + "per": "m", + "source": _SRC_ULJIN1, + "items": { + "콘크리트(㎥)": 0.52, + "모르터(㎥)": 0.023, + "돌쌓기(㎡)": 2.61, + "석재(ton)": 0.91, + "터파기(㎥)": 2.58, + "되메우기(㎥)": 0.50, + "잔토처리(㎥)": 2.08, + }, + }, + ("completion_sign", "화강암 300×200×600"): { + "label": "임도준공판 (화강암 300×200×600)", + "per": "개소", + "source": _SRC_SOGWANG, + "items": { + "화강암(㎥)": 0.036, + "기초콘크리트(㎥)": 0.014, + "거푸집-유로폼(㎡)": 0.35, + "터파기(㎥)": 0.105, + "되메우기(㎥)": 0.08, + "잔토처리(㎥)": 0.025, + }, + }, + # 참고 표기용 — 관 부설 연장이 미산정이라 rollup에는 적용하지 않는다. + ("pipe_ref", "Ø800"): { + "label": "관공 Ø800 (참고 — 연장 미산정으로 미적용)", + "per": "m", + "source": _SRC_ULJIN1, + "items": {"터파기(㎥)": 4.5, "되메우기(㎥)": 3.99, "잔토처리(㎥)": 2.93}, + }, + ("pipe_ref", "Ø1000"): { + "label": "관공 Ø1000 (참고 — 연장 미산정으로 미적용)", + "per": "m", + "source": _SRC_ULJIN1, + "items": {"터파기(㎥)": 5.6, "되메우기(㎥)": 4.81, "잔토처리(㎥)": 3.83}, + }, +} + + +# ---------------------------------------------------------------- 데이터 수집 + + +async def _collect(project_id: UUID) -> dict[str, Any]: + """정본 3곳을 읽어 화면 조립에 필요한 원자료만 모은다 (쓰기 없음).""" + pool = get_db_pool() + async with pool.acquire() as connection: + stored_path = await get_project_storage_relative_path(connection, project_id) + route_context = await get_confirmed_route_context(connection, project_id) + route_id = int(route_context["route_id"]) if route_context else None + confirmed = False + designs: list[dict[str, Any]] = [] + if route_id is not None: + longitudinal = await get_longitudinal_section(connection, project_id, route_id) + confirmed = bool(longitudinal and longitudinal.get("status") == "CONFIRMED") + rows = await get_cross_section_designs(connection, route_id) + designs = [row for row in rows if isinstance(row.get("design"), dict)] + + root = Path(resolve_stored_project_path(stored_path)).resolve() + + pipe_file = root / "B04_PreProcess" / "drainage" / "edits" / "pipe_points.json" + pipes: list[Any] = [] + pipe_file_exists = pipe_file.is_file() + if pipe_file_exists: + try: + document = json.loads(pipe_file.read_text(encoding="utf-8")) + pipes = parse_pipe_points(document.get("points")) + except (OSError, ValueError): + logger.warning("수량 프로토: pipe_points.json 읽기 실패 — %s", pipe_file) + + revision, structures = load_structures(str(root)) + + return { + "route_id": route_id, + "confirmed": confirmed, + "designs": designs, + "pipes": pipes, + "pipe_file_exists": pipe_file_exists, + "structures": structures, + "structures_revision": revision, + } + + +# ---------------------------------------------------------------- 집계·매칭 + + +def _pipe_spec(point: Any) -> str: + options = point.options or {} + if point.facility == "pipe": + kind = options.get("pipe_kind", "파형강관") + dia = options.get("pipe_diameter_mm", 1000) + return f"{kind} Ø{dia}" + if point.facility == "box_culvert": + return str(options.get("box_size", "규격 미지정")) + return "—" + + +def _span_text(point: Any) -> str: + if point.start_m is not None and point.end_m is not None: + return f"{point.start_m:.2f}~{point.end_m:.2f} (L={point.end_m - point.start_m:.2f}m)" + return "미지정(폭 0)" + + +def _structure_position(instance: Any) -> str: + if instance.placement == "interval" and instance.start_m is not None: + end = instance.end_m if instance.end_m is not None else instance.start_m + return f"{instance.start_m:.2f}~{end:.2f} (L={end - instance.start_m:.2f}m)" + if instance.chainage_m is not None: + return f"{instance.chainage_m:.2f}m" + return "—" + + +def _wall_rows(designs: list[dict[str, Any]]) -> list[dict[str, Any]]: + """횡단 patch에서 벽 조작·다단 벽 흔적이 있는 측점만 표로 만든다.""" + rows: list[dict[str, Any]] = [] + for row in designs: + design = row["design"] + adjust = design.get("revet_adjust") or {} + counts = design.get("extra_wall_counts") or {} + spans = design.get("extra_spans") or {} + if not adjust and not any(counts.values() if isinstance(counts, dict) else []): + continue + walls = [] + for key, patch in adjust.items() if isinstance(adjust, dict) else []: + if not isinstance(patch, dict): + continue + height = patch.get("h") + span = spans.get(key) if isinstance(spans, dict) else None + length = span.get("length_m") if isinstance(span, dict) else None + walls.append( + { + "key": key, + "height": f"{height:.2f}m" if isinstance(height, (int, float)) else "자동", + "material": patch.get("m") or "자동", + "length": f"{length:.2f}m" if isinstance(length, (int, float)) else "—", + } + ) + rows.append( + { + "chainage": float(row.get("chainage_m", 0)), + "counts": counts if isinstance(counts, dict) else {}, + "walls": walls, + } + ) + rows.sort(key=lambda r: r["chainage"]) + return rows + + +def _rollup(data: dict[str, Any]) -> dict[str, Any]: + """규격별 집계 → 원단위 매칭 → 공종별 합계. 매칭 불가는 사유와 함께 보류 목록으로.""" + matched: list[dict[str, Any]] = [] # 산출근거 행 + deferred: list[dict[str, Any]] = [] # 보류 행 + totals: dict[str, float] = defaultdict(float) + + def apply(rate_key: tuple[str, str], quantity: float, positions: str) -> None: + rate = UNIT_RATES[rate_key] + lines = [] + for item, unit_qty in rate["items"].items(): + value = unit_qty * quantity + totals[item] += value + lines.append({"item": item, "unit_qty": unit_qty, "value": value}) + matched.append( + { + "label": rate["label"], + "per": rate["per"], + "quantity": quantity, + "source": rate["source"], + "note": rate.get("note", ""), + "positions": positions, + "lines": lines, + } + ) + + # ① 배수 시설 — 관 본체는 연장 미산정으로 보류, 유입 집수정만 개소당 적용 + basin_positions: list[str] = [] + pipe_counts: dict[str, int] = defaultdict(int) + for point in data["pipes"]: + facility = _FACILITY_NAMES.get(point.facility, point.facility) + options = point.options or {} + if point.facility == "pipe": + pipe_counts[_pipe_spec(point)] += 1 + if options.get("inlet_type") == "집수정": + basin_positions.append(f"{point.chainage_m:.2f}m") + else: + deferred.append( + { + "name": f"{facility} @{point.chainage_m:.2f}m", + "reason": "원단위 미확보 (상세 치수·산식은 B06/구조도 확정 후)", + } + ) + if basin_positions: + apply(("basin", "Ø800"), len(basin_positions), " · ".join(basin_positions)) + for spec, count in sorted(pipe_counts.items()): + deferred.append( + { + "name": f"배수관 {spec} × {count}개소", + "reason": "관 부설 연장 미산정 (B06 횡단 세트 연장 필요) — 관공 m당 원단위는 라이브러리 참고행", + } + ) + + # ② 일반 구조물 (structures.json) + for instance in data["structures"]: + name = _type_name(instance.type_id) + position = _structure_position(instance) + if instance.type_id == "erosion_check": + apply(("erosion_check", "찰쌓기·평균"), 1, position) + continue + if instance.type_id == "completion_sign": + apply(("completion_sign", "화강암 300×200×600"), 1, position) + continue + if instance.type_id == "retaining_wall" and instance.placement == "interval": + length = (instance.end_m or 0) - (instance.start_m or 0) + if length > 0: + apply(("retaining_wall", "반중력식 H=2.0"), length, position) + continue + deferred.append( + {"name": f"{name} @{position}", "reason": "원단위 미확보 또는 규격(치수) 미확정"} + ) + + # ③ 횡단 벽 패치 — 높이·연장이 자동 계산 몫이라 전량 보류 (일람으로만 제시) + wall_station_count = len(_wall_rows(data["designs"])) + if wall_station_count: + deferred.append( + { + "name": f"기준벽·다단 기슭막이 — 조작 측점 {wall_station_count}곳", + "reason": "벽 높이(자동)·연장이 B06 엔진 산출 몫 — 규격 확정 후 돌기슭막이 H별 m당 원단위 적용 예정", + } + ) + + return {"matched": matched, "deferred": deferred, "totals": dict(totals)} + + +# ---------------------------------------------------------------- HTML + + +_CSS = """ +body{font-family:'Malgun Gothic',sans-serif;margin:24px;background:#f5f6f8;color:#222} +h1{font-size:20px} h2{font-size:16px;margin-top:28px;border-left:4px solid #4f8ef7;padding-left:8px} +table{border-collapse:collapse;background:#fff;margin:8px 0;font-size:13px} +th,td{border:1px solid #ccd;padding:4px 10px;text-align:left} +th{background:#eef2fa} td.num{text-align:right;font-variant-numeric:tabular-nums} +.banner{background:#fff3cd;border:1px solid #e0c060;padding:10px 14px;border-radius:6px;font-size:13px} +.muted{color:#777;font-size:12px} .warn{color:#a55} .src{color:#476;font-size:12px} +a{color:#3366cc} +""" + + +def _fmt(value: float) -> str: + return f"{value:,.2f}" + + +def _page(title: str, body: str) -> HTMLResponse: + return HTMLResponse( + "" + f"{title}{body}" + ) + + +def _render_report(project: dict[str, Any] | None, project_id: UUID, data: dict[str, Any]) -> str: + rollup = _rollup(data) + name = (project or {}).get("name") or str(project_id) + parts: list[str] = [f"

수량 산출 프로토타입 — {name}

"] + parts.append( + "" + ) + + # 데이터 원천 상태 + parts.append("

데이터 원천

") + parts.append( + f"" + ) + parts.append( + f"" + ) + patch_state = ( + f"확정 노선 route_id={data['route_id']} — 설계 저장 측점 {len(data['designs'])}곳" + if data["route_id"] is not None + else "확정 노선 없음 — 횡단 벽 패치 제외" + ) + if data["route_id"] is not None and not data["confirmed"]: + patch_state += " (B06 종단 미확정)" + parts.append(f"
정본상태
pipe_points.json (배수·독립 기슭막이){'있음 — ' + str(len(data['pipes'])) + '건' if data['pipe_file_exists'] else '없음'}
structures.json (일반 구조물, rev {data['structures_revision']}){len(data['structures'])}건
횡단 설계 patch (DB){patch_state}
") + + # ① 배수 시설 일람 + parts.append("

① 배수 시설 일람 (pipe_points.json)

") + if data["pipes"]: + parts.append( + "" + ) + for point in sorted(data["pipes"], key=lambda p: p.chainage_m): + options = point.options or {} + inlet = options.get("inlet_type", "기슭막이(기본)") if point.facility == "pipe" else "—" + parts.append( + f"" + f"" + ) + parts.append("
측점시설규격부속 구간(전~후)유입구출처
{point.chainage_m:.2f}m{_FACILITY_NAMES.get(point.facility, point.facility)}{_pipe_spec(point)}{_span_text(point)}{inlet}{point.source}
") + else: + parts.append("

배치된 배수 시설이 없다.

") + + # ② 일반 구조물 일람 + parts.append("

② 일반 구조물 일람 (structures.json)

") + if data["structures"]: + parts.append( + "" + ) + for instance in data["structures"]: + option_text = ", ".join(f"{k}={v}" for k, v in (instance.options or {}).items()) or "—" + parts.append( + f"" + f"" + ) + parts.append("
타입위치옵션배치상태
{_type_name(instance.type_id)}{_structure_position(instance)}{option_text}{instance.placement_source}{instance.status}
") + else: + parts.append("

배치된 일반 구조물이 없다.

") + + # ③ 횡단 벽 패치 일람 + parts.append("

③ 기준벽·다단 기슭막이 (횡단 설계 patch)

") + wall_rows = _wall_rows(data["designs"]) + if wall_rows: + parts.append( + "" + ) + for row in wall_rows: + counts = row["counts"] + count_text = f"{counts.get('outlet', 0)} / {counts.get('basin', 0)}" + if not row["walls"]: + parts.append( + f"" + ) + for index, wall in enumerate(row["walls"]): + lead = ( + ( + f"" + f"" + ) + if index == 0 + else "" + ) + parts.append( + f"{lead}" + f"" + ) + parts.append("
측점다단(유출/계류)벽 키높이재질단별 구간
{row['chainage']:.2f}m{count_text}벽 조작값 없음(단 수만 지정)
{row['chainage']:.2f}m{count_text}
{wall['key']}{wall['height']}{wall['material']}{wall['length']}
") + parts.append( + "

벽 좌우·상하·높이·형태는 횡단 설계 patch, 다단 단별 구간값은 extra_spans 정본 (CLAUDE.md 5장).

" + ) + else: + parts.append("

벽 조작·다단 지정 측점이 없다.

") + + # ④ 산출근거 (원단위 × 수량) + parts.append("

④ 산출근거 — 원단위 × 수량

") + if rollup["matched"]: + for entry in rollup["matched"]: + parts.append( + f"

{entry['label']} × {_fmt(entry['quantity'])}{entry['per']}" + f" [출처: {entry['source']} — 참조 전용]
" + f"위치: {entry['positions']}" + + (f"
{entry['note']}" if entry["note"] else "") + + "

" + ) + parts.append( + "" + ) + for line in entry["lines"]: + parts.append( + f"" + f"" + ) + parts.append("
공종단위수량수량산출값
{line['item']}{line['unit_qty']}{_fmt(entry['quantity'])}{_fmt(line['value'])}
") + else: + parts.append("

원단위 매칭 가능한 구조물이 없다.

") + + # ⑤ 공종별 합계 + parts.append("

⑤ 공종별 수량 합계

") + if rollup["totals"]: + parts.append("") + for item, value in sorted(rollup["totals"].items()): + parts.append(f"") + parts.append("
공종합계
{item}{_fmt(value)}
") + else: + parts.append("

합산할 항목이 없다.

") + + # ⑥ 산출 보류 + parts.append("

⑥ 산출 보류 (사유)

") + if rollup["deferred"]: + parts.append("") + for row in rollup["deferred"]: + parts.append(f"") + parts.append("
대상사유
{row['name']}{row['reason']}
") + else: + parts.append("

보류 항목 없음.

") + + parts.append( + "

근거 문서: resources/knowledge/original/실무문서/_원단위라이브러리_울진소광.md · " + "_종합비교/04_임도구조물_원단위.md · technical_info/01_임도/04_수량분석정보/ (수량산출_일반·구조물_수량·배수공_수량)

" + ) + return "".join(parts) + + +# ---------------------------------------------------------------- 라우트 + + +@proto_router.get("/quantity-proto/page", response_class=HTMLResponse) +async def quantity_proto_index() -> HTMLResponse: + """프로젝트 선택 페이지.""" + projects = await list_all_projects() + rows = "".join( + f"{p.get('name') or p['id']}" + f"{p.get('region') or '—'}{p.get('status') or '—'}{p.get('updated_at') or ''}" + for p in projects + ) + body = ( + "

수량 산출 프로토타입 — 프로젝트 선택

" + "" + f"{rows}
프로젝트지역상태수정
" + if projects + else "

수량 산출 프로토타입

프로젝트가 없다.

" + ) + return _page("수량 산출 프로토타입", body) + + +@proto_router.get("/quantity-proto/page/{project_id}", response_class=HTMLResponse) +async def quantity_proto_report(project_id: UUID) -> HTMLResponse: + """프로젝트 1건의 수량 산출 보고서.""" + try: + data = await _collect(project_id) + except FileNotFoundError as error: + return _page( + "수량 산출 프로토타입", f"

수량 산출 프로토타입

{error}

" + ) + projects = await list_all_projects() + project = next((p for p in projects if str(p.get("id")) == str(project_id)), None) + body = _render_report(project, project_id, data) + body += "

← 프로젝트 선택으로

" + return _page("수량 산출 프로토타입", body) diff --git a/B07_DesignDetail/B07_DesignDetail_Router.py b/B07_DesignDetail/B07_DesignDetail_Router.py index 3a01ed65..34f239fe 100644 --- a/B07_DesignDetail/B07_DesignDetail_Router.py +++ b/B07_DesignDetail/B07_DesignDetail_Router.py @@ -50,6 +50,11 @@ from config.config_db import get_db_pool logger = logging.getLogger(__name__) router = APIRouter(prefix="/api/projects", tags=["B07 Design Detail"]) +# 수량 산출 프로토타입 (처분 가능) — 파일 삭제 시 아래 2줄도 함께 지운다. +from B07_DesignDetail.B07_DesignDetail_Quantity_Proto import proto_router # noqa: E402 + +router.include_router(proto_router) + _CROSS_ID = re.compile(r"^cross_(\d+)m$") _LONG_ID = re.compile(r"^longitudinal(?:_(\d+))?$")