752 lines
32 KiB
Python
752 lines
32 KiB
Python
"""B06 확정 종·횡단 산출물을 B07 CAD 도면으로 변환하는 라우터."""
|
|
|
|
import asyncio
|
|
import json
|
|
import logging
|
|
import re
|
|
from pathlib import Path
|
|
from typing import Any
|
|
from uuid import UUID
|
|
|
|
from fastapi import APIRouter
|
|
from fastapi.responses import JSONResponse
|
|
|
|
from B03_FileInput.B03_FileInput_Repository import get_project_storage_relative_path
|
|
from B05_Profile.B05_Profile_Engine_Sections import prune_stale_cross_files
|
|
from B06_Section.B06_Section_Engine_Design import compute_cross_design
|
|
from B06_Section.B06_Section_Repository import (
|
|
get_confirmed_route_context,
|
|
get_cross_section_design,
|
|
get_cross_section_designs,
|
|
get_longitudinal_section,
|
|
merge_cross_section_design_by_round,
|
|
)
|
|
from B07_DesignDetail.B07_DesignDetail_Engine_Cad import (
|
|
DRAWING_FORMAT,
|
|
build_cross_drawing,
|
|
infer_station_interval,
|
|
station_no_label,
|
|
)
|
|
from B07_DesignDetail.B07_DesignDetail_Engine_Cad_Long import (
|
|
build_longitudinal_drawing,
|
|
longitudinal_chunks,
|
|
)
|
|
from B07_DesignDetail.B07_DesignDetail_Engine_Cad_Sheet import (
|
|
CROSS_SHEET_ID,
|
|
build_cross_sheet,
|
|
plan_cross_sheets,
|
|
section_block_size,
|
|
)
|
|
from B07_DesignDetail.B07_DesignDetail_Engine_Cad_Table import (
|
|
QUANTITY_VALUE_KEYS,
|
|
extract_quantity_table,
|
|
)
|
|
from B07_DesignDetail.B07_DesignDetail_Schema import (
|
|
DesignDrawingConfirmRequest,
|
|
DesignDrawingConfirmResponse,
|
|
DesignDrawingInvalidateResponse,
|
|
DesignDrawingItem,
|
|
DesignDrawingListResponse,
|
|
DesignDrawingResponse,
|
|
)
|
|
from common_util.common_util_route_profile import design_elevation_from_longitudinal
|
|
from common_util.common_util_storage import resolve_stored_project_path
|
|
from common_util.common_util_workflow_state import complete_stage, start_stage
|
|
from config.config_db import get_db_pool
|
|
|
|
logger = logging.getLogger(__name__)
|
|
router = APIRouter(prefix="/api/projects", tags=["B07 Design Detail"])
|
|
|
|
_CROSS_ID = re.compile(r"^cross_(\d+)m$")
|
|
_LONG_ID = re.compile(r"^longitudinal(?:_(\d+))?$")
|
|
_STAGE_DIR = "B07_DesignDetail"
|
|
|
|
|
|
async def _confirmed_source(project_id: UUID) -> tuple[int, Path, Path]:
|
|
"""확정된 B06 종단 레코드와 프로젝트 저장 경로를 반환한다."""
|
|
pool = get_db_pool()
|
|
async with pool.acquire() as connection:
|
|
route_context = await get_confirmed_route_context(connection, project_id)
|
|
if not route_context:
|
|
raise FileNotFoundError("확정된 경로가 없습니다.")
|
|
route_id = int(route_context["route_id"])
|
|
longitudinal = await get_longitudinal_section(connection, project_id, route_id)
|
|
if not longitudinal or longitudinal.get("status") != "CONFIRMED":
|
|
raise PermissionError("B06 종·횡단 확정 후 상세 설계를 진행할 수 있습니다.")
|
|
stored_path = await get_project_storage_relative_path(connection, project_id)
|
|
|
|
root = Path(resolve_stored_project_path(stored_path)).resolve()
|
|
longitudinal_path = (root / str(longitudinal["longitudinal_file_path"])).resolve()
|
|
if root not in longitudinal_path.parents or not longitudinal_path.is_file():
|
|
raise FileNotFoundError("B06 종단면 파일을 찾을 수 없습니다.")
|
|
return route_id, root, longitudinal_path
|
|
|
|
|
|
def _read_json(path: Path) -> dict[str, Any]:
|
|
payload = json.loads(path.read_text(encoding="utf-8"))
|
|
if not isinstance(payload, dict):
|
|
raise ValueError("도면 원본 JSON 형식이 올바르지 않습니다.")
|
|
return payload
|
|
|
|
|
|
def _cross_files(longitudinal_path: Path, longitudinal: dict[str, Any]) -> list[Path]:
|
|
cross_dir = longitudinal_path.parent.parent / "cross_sections"
|
|
if not cross_dir.is_dir():
|
|
raise FileNotFoundError("B06 횡단면 파일을 찾을 수 없습니다.")
|
|
stations = longitudinal.get("stations")
|
|
valid_names = prune_stale_cross_files(cross_dir, stations if isinstance(stations, list) else [])
|
|
files = sorted(cross_dir.glob("cross_*.json"))
|
|
if valid_names:
|
|
files = [path for path in files if path.name in valid_names]
|
|
return files
|
|
|
|
|
|
def _station_map(longitudinal: dict[str, Any]) -> dict[int, dict[str, Any]]:
|
|
stations = longitudinal.get("stations", [])
|
|
if not isinstance(stations, list):
|
|
return {}
|
|
return {
|
|
round(float(station.get("chainage_m", 0))): station
|
|
for station in stations
|
|
if isinstance(station, dict)
|
|
}
|
|
|
|
|
|
def _design_root(project_root: Path) -> Path:
|
|
return project_root / _STAGE_DIR
|
|
|
|
|
|
def _read_manifest(project_root: Path) -> dict[str, Any]:
|
|
path = _design_root(project_root) / "manifest.json"
|
|
if not path.is_file():
|
|
return {"drawings": {}}
|
|
payload = _read_json(path)
|
|
return payload if isinstance(payload.get("drawings"), dict) else {"drawings": {}}
|
|
|
|
|
|
def _write_manifest(project_root: Path, manifest: dict[str, Any]) -> None:
|
|
stage_root = _design_root(project_root)
|
|
stage_root.mkdir(parents=True, exist_ok=True)
|
|
path = stage_root / "manifest.json"
|
|
temporary = path.with_suffix(".tmp")
|
|
temporary.write_text(json.dumps(manifest, ensure_ascii=False, indent=2), encoding="utf-8")
|
|
temporary.replace(path)
|
|
|
|
|
|
def _drawing_list(
|
|
project_root: Path,
|
|
longitudinal_path: Path,
|
|
designs: dict[int, dict[str, Any]] | None = None,
|
|
) -> list[DesignDrawingItem]:
|
|
longitudinal = _read_json(longitudinal_path)
|
|
station_by_chainage = _station_map(longitudinal)
|
|
manifest_drawings = _read_manifest(project_root)["drawings"]
|
|
# 종단도: 측점 30개 초과 시 30개 단위 분할 도면을 각각 목록에 노출한다(N-1-1).
|
|
drawings = [
|
|
DesignDrawingItem(
|
|
id=str(chunk["id"]),
|
|
kind="longitudinal",
|
|
label=str(chunk["label"]),
|
|
confirmed=bool(manifest_drawings.get(str(chunk["id"]), {}).get("confirmed")),
|
|
)
|
|
for chunk in longitudinal_chunks(longitudinal)
|
|
]
|
|
for sheet in _cross_sheet_plan(project_root, longitudinal_path, designs):
|
|
chainages = sheet["chainages"]
|
|
first = station_by_chainage.get(chainages[0], {})
|
|
last = station_by_chainage.get(chainages[-1], {})
|
|
span = str(first.get("label") or chainages[0])
|
|
if len(chainages) > 1:
|
|
span = f"{span}~{last.get('label') or chainages[-1]}"
|
|
drawings.append(
|
|
DesignDrawingItem(
|
|
id=sheet["id"],
|
|
kind="cross",
|
|
label=f"{sheet['number']}장 ({span})",
|
|
chainage_m=float(first.get("chainage_m", chainages[0])),
|
|
confirmed=bool(manifest_drawings.get(sheet["id"], {}).get("confirmed")),
|
|
)
|
|
)
|
|
return drawings
|
|
|
|
|
|
def _cross_sheet_plan(
|
|
project_root: Path,
|
|
longitudinal_path: Path,
|
|
designs: dict[int, dict[str, Any]] | None = None,
|
|
) -> list[dict[str, Any]]:
|
|
"""측점 도면을 A1 장으로 나눈 계획 — 목록과 도면 생성이 같은 결과를 쓴다.
|
|
|
|
한 장에 몇 개가 들어가는지는 단면 블록 크기가 정한다(작성 척도 1/100 +
|
|
설계선이 원지반과 갈라지는 구간). 설계 지정이 없으면 원지반 기준 폭이 된다.
|
|
"""
|
|
longitudinal = _read_json(longitudinal_path)
|
|
blocks: list[tuple[int, float, float]] = []
|
|
for path in sorted(_cross_files(longitudinal_path, longitudinal)):
|
|
match = _CROSS_ID.fullmatch(path.stem)
|
|
if not match:
|
|
continue
|
|
chainage = int(match.group(1))
|
|
source = _read_json(path)
|
|
design = (designs or {}).get(chainage)
|
|
design_line = _cross_design_line(longitudinal_path, source, design)
|
|
width, height = section_block_size(source, design_line)
|
|
blocks.append((chainage, width, height))
|
|
return plan_cross_sheets(blocks)
|
|
|
|
|
|
def _quantity_table(source: dict[str, Any]) -> dict[str, float | None]:
|
|
"""횡단면 원본에서 편집 가능한 수량 산출표 값을 구조화한다 (납품 양식 키).
|
|
|
|
center_z→지반고, design_elevation_m→계획고, 절토고/성토고는 파생 초기값.
|
|
나머지 항목은 source["quantities"]에 같은 키가 있으면 읽고 없으면 None으로
|
|
두어 CAD 테이블에서 사용자가 채운다.
|
|
"""
|
|
|
|
def num(value: Any) -> float | None:
|
|
return float(value) if isinstance(value, (int, float)) else None
|
|
|
|
ground = num(source.get("center_z"))
|
|
planned = num(source.get("planned_elevation_m", source.get("design_elevation_m")))
|
|
cut = max(ground - planned, 0.0) if ground is not None and planned is not None else None
|
|
fill = max(planned - ground, 0.0) if ground is not None and planned is not None else None
|
|
quantities = source.get("quantities") if isinstance(source.get("quantities"), dict) else {}
|
|
|
|
table: dict[str, float | None] = {
|
|
"ground": ground,
|
|
"planned": planned,
|
|
"cut": cut,
|
|
"fill": fill,
|
|
}
|
|
for key in QUANTITY_VALUE_KEYS:
|
|
table.setdefault(key, num(quantities.get(key)))
|
|
return table
|
|
|
|
|
|
def _route_cross_frame(
|
|
longitudinal_path: Path, longitudinal: dict[str, Any]
|
|
) -> dict[str, float] | None:
|
|
"""노선 전체 횡단면의 기준 레이아웃 {half, half_height}를 구한다.
|
|
|
|
측점별 단면 높이(최대-최소 표고)와 폭의 노선 최대값 — 모든 횡단도가 자기
|
|
콘텐츠 중심 기준으로 같은 크기 레이아웃에 배치되도록 하는 기준값(테이블
|
|
y 고정용). 실패 시 None(측점 자체 범위 폴백).
|
|
"""
|
|
half = half_height = 0.0
|
|
found = False
|
|
try:
|
|
for path in _cross_files(longitudinal_path, longitudinal):
|
|
source = _read_json(path)
|
|
min_e: float | None = None
|
|
max_e: float | None = None
|
|
for sample in source.get("samples", []):
|
|
if not isinstance(sample, dict) or not sample.get("valid", False):
|
|
continue
|
|
offset = sample.get("offset_m")
|
|
elevation = sample.get("elevation_m")
|
|
if isinstance(offset, (int, float)):
|
|
half = max(half, abs(float(offset)))
|
|
if isinstance(elevation, (int, float)):
|
|
value = float(elevation)
|
|
min_e = value if min_e is None else min(min_e, value)
|
|
max_e = value if max_e is None else max(max_e, value)
|
|
if min_e is not None and max_e is not None:
|
|
half_height = max(half_height, (max_e - min_e) / 2.0)
|
|
found = True
|
|
except (OSError, ValueError, json.JSONDecodeError, FileNotFoundError):
|
|
return None
|
|
if not found:
|
|
return None
|
|
# 여유: 측구 깊이·사면 연장 등 설계선이 지반 포락선을 소폭 벗어나는 분 반영.
|
|
return {"half": half or 12.0, "half_height": half_height + 1.5}
|
|
|
|
|
|
def _cross_design_line(
|
|
longitudinal_path: Path, source: dict[str, Any], stored_design: dict[str, Any] | None
|
|
) -> list[Any] | None:
|
|
"""횡단 CAD 계획선용 design_line을 정한다: 저장 설계 우선, 없으면 기본값 계산."""
|
|
if isinstance(stored_design, dict) and isinstance(stored_design.get("design_line"), list):
|
|
return stored_design["design_line"]
|
|
try:
|
|
longitudinal = _read_json(longitudinal_path)
|
|
design = compute_cross_design(
|
|
source.get("samples", []),
|
|
design_elevation_from_longitudinal(longitudinal, float(source.get("chainage_m", 0.0))),
|
|
ground_type="soil",
|
|
section_mode="left_cut",
|
|
)
|
|
return design["design_line"]
|
|
except (ValueError, KeyError, OSError, json.JSONDecodeError):
|
|
return None
|
|
|
|
|
|
def _cross_section_input(
|
|
longitudinal_path: Path,
|
|
longitudinal: dict[str, Any],
|
|
chainage: int,
|
|
design: dict[str, Any] | None,
|
|
) -> dict[str, Any] | None:
|
|
"""한 측점의 장 배치 입력(원본·계획선·수량표·제목)을 만든다."""
|
|
path = longitudinal_path.parent.parent / "cross_sections" / f"cross_{chainage:05d}m.json"
|
|
if not path.is_file():
|
|
return None
|
|
source = _read_json(path)
|
|
interval = infer_station_interval(longitudinal.get("stations") or [])
|
|
return {
|
|
"chainage": chainage,
|
|
"source": source,
|
|
"design": design,
|
|
"design_line": _cross_design_line(longitudinal_path, source, design),
|
|
"quantity_table": _quantity_table(source),
|
|
"title": station_no_label(float(source.get("chainage_m", chainage)), interval),
|
|
}
|
|
|
|
|
|
def _read_cross_sheet(
|
|
project_root: Path,
|
|
longitudinal_path: Path,
|
|
drawing_id: str,
|
|
stored_designs: dict[str, Any] | None,
|
|
) -> tuple[str, str, dict[str, Any], bool, dict[str, float | None] | None]:
|
|
"""횡단 장 하나를 만든다. stored_designs는 {측점: 설계 지정} 묶음이다."""
|
|
designs = stored_designs if isinstance(stored_designs, dict) else {}
|
|
plan = _cross_sheet_plan(project_root, longitudinal_path, designs)
|
|
sheet = next((item for item in plan if item["id"] == drawing_id), None)
|
|
if sheet is None:
|
|
raise FileNotFoundError("요청한 횡단 장을 찾을 수 없습니다.")
|
|
longitudinal = _read_json(longitudinal_path)
|
|
sections = [
|
|
section
|
|
for section in (
|
|
_cross_section_input(longitudinal_path, longitudinal, chainage, designs.get(chainage))
|
|
for chainage in sheet["chainages"]
|
|
)
|
|
if section is not None
|
|
]
|
|
label = f"횡단면도 {sheet['number']}장"
|
|
return "cross", label, build_cross_sheet(sheet, sections), False, None
|
|
|
|
|
|
def _read_drawing(
|
|
project_root: Path,
|
|
longitudinal_path: Path,
|
|
drawing_id: str,
|
|
stored_design: dict[str, Any] | None = None,
|
|
) -> tuple[str, str, dict[str, Any], bool, dict[str, float | None] | None]:
|
|
"""(kind, label, drawing, confirmed, quantity_table)를 반환한다.
|
|
|
|
quantity_table은 횡단도에서만 채워지며, 확정본은 manifest에 저장된 사용자
|
|
편집값을 우선하고 없으면 원본에서 파생한 초기값을 계산한다. 횡단도의 계획선은
|
|
stored_design(없으면 기본값)에서 만든다.
|
|
"""
|
|
manifest_entry = _read_manifest(project_root)["drawings"].get(drawing_id, {})
|
|
saved_path = _design_root(project_root) / "drawings" / f"{drawing_id}.json"
|
|
if manifest_entry.get("confirmed") and saved_path.is_file():
|
|
saved = _read_json(saved_path)
|
|
# 포맷 버전이 다르면(테이블·레이어 구성 변경 전 저장본) 캐시를 버리고
|
|
# 아래에서 원본 기준으로 재생성한다. 확정 상태도 무효로 응답해 재확정 유도.
|
|
if saved.get("format") == DRAWING_FORMAT:
|
|
kind = "longitudinal" if _LONG_ID.fullmatch(drawing_id) else "cross"
|
|
label = str(manifest_entry.get("label") or drawing_id)
|
|
stored_table = manifest_entry.get("quantity_table")
|
|
table = stored_table if kind == "cross" and isinstance(stored_table, dict) else None
|
|
return kind, label, saved, True, table
|
|
if _LONG_ID.fullmatch(drawing_id):
|
|
source = _read_json(longitudinal_path)
|
|
chunk = next(
|
|
(item for item in longitudinal_chunks(source) if item["id"] == drawing_id), None
|
|
)
|
|
if chunk is None:
|
|
raise FileNotFoundError("요청한 종단도 분할 도면을 찾을 수 없습니다.")
|
|
return (
|
|
"longitudinal",
|
|
str(chunk["label"]),
|
|
build_longitudinal_drawing(source, drawing_id, chunk),
|
|
False,
|
|
None,
|
|
)
|
|
|
|
if CROSS_SHEET_ID.fullmatch(drawing_id):
|
|
return _read_cross_sheet(project_root, longitudinal_path, drawing_id, stored_design)
|
|
|
|
if not _CROSS_ID.fullmatch(drawing_id):
|
|
raise ValueError("올바르지 않은 도면 ID입니다.")
|
|
path = longitudinal_path.parent.parent / "cross_sections" / f"{drawing_id}.json"
|
|
if not path.is_file():
|
|
raise FileNotFoundError("요청한 횡단도를 찾을 수 없습니다.")
|
|
source = _read_json(path)
|
|
label = str(source.get("label") or drawing_id)
|
|
design_line = _cross_design_line(longitudinal_path, source, stored_design)
|
|
quantity_table = _quantity_table(source)
|
|
# 수량표 제목행 No. 표기: 종단 측점 간격 기준 (납품 도면 양식)
|
|
longitudinal = _read_json(longitudinal_path)
|
|
interval = infer_station_interval(longitudinal.get("stations") or [])
|
|
title = station_no_label(float(source.get("chainage_m", 0.0)), interval)
|
|
# 계획고(로컬좌표 기준) + 노선 공통 범위 — 측점 이동 시 화면 배치 고정.
|
|
design_elevation = design_elevation_from_longitudinal(
|
|
longitudinal, float(source.get("chainage_m", 0.0))
|
|
)
|
|
frame = _route_cross_frame(longitudinal_path, longitudinal)
|
|
return (
|
|
"cross",
|
|
label,
|
|
build_cross_drawing(
|
|
source,
|
|
drawing_id,
|
|
design_line,
|
|
stored_design,
|
|
quantity_table,
|
|
title,
|
|
design_elevation,
|
|
frame,
|
|
),
|
|
False,
|
|
quantity_table,
|
|
)
|
|
|
|
|
|
def _store_confirmed_drawing(
|
|
project_root: Path,
|
|
item: DesignDrawingItem,
|
|
drawing: dict[str, Any],
|
|
expected_ids: set[str],
|
|
quantity_table: dict[str, Any] | None = None,
|
|
quantity_tables: dict[str, Any] | None = None,
|
|
) -> bool:
|
|
if not isinstance(drawing.get("entities"), list) or not isinstance(drawing.get("layers"), list):
|
|
raise ValueError("CAD 도면 스키마가 올바르지 않습니다.")
|
|
# CAD 앱 직렬화본에는 format이 없으므로 저장 시 현재 포맷 버전을 스탬프한다.
|
|
drawing = {"format": DRAWING_FORMAT, **drawing}
|
|
drawings_dir = _design_root(project_root) / "drawings"
|
|
drawings_dir.mkdir(parents=True, exist_ok=True)
|
|
path = drawings_dir / f"{item.id}.json"
|
|
temporary = path.with_suffix(".tmp")
|
|
temporary.write_text(json.dumps(drawing, ensure_ascii=False, indent=2), encoding="utf-8")
|
|
temporary.replace(path)
|
|
|
|
manifest = _read_manifest(project_root)
|
|
entry: dict[str, Any] = {
|
|
"kind": item.kind,
|
|
"label": item.label,
|
|
"confirmed": True,
|
|
"file": f"drawings/{item.id}.json",
|
|
}
|
|
if item.kind == "cross" and isinstance(quantity_table, dict):
|
|
entry["quantity_table"] = quantity_table
|
|
if isinstance(quantity_tables, dict) and quantity_tables:
|
|
# 장 도면: 측점별 수량표를 그대로 보관한다(뒷단계 수량산출이 측점 단위).
|
|
entry["quantity_tables"] = quantity_tables
|
|
manifest["drawings"][item.id] = entry
|
|
_write_manifest(project_root, manifest)
|
|
confirmed_ids = {
|
|
item_id for item_id, entry in manifest["drawings"].items() if entry.get("confirmed")
|
|
}
|
|
return expected_ids.issubset(confirmed_ids)
|
|
|
|
|
|
def _invalidate_drawing(project_root: Path, drawing_id: str) -> None:
|
|
manifest = _read_manifest(project_root)
|
|
entry = manifest["drawings"].get(drawing_id)
|
|
if entry:
|
|
entry["confirmed"] = False
|
|
_write_manifest(project_root, manifest)
|
|
|
|
|
|
async def _designs_by_chainage(route_id: int) -> dict[int, dict[str, Any]]:
|
|
"""노선 전체의 측점별 설계 지정 {측점(m): design}. 장 배치·목록이 함께 쓴다."""
|
|
pool = get_db_pool()
|
|
async with pool.acquire() as connection:
|
|
rows = await get_cross_section_designs(connection, route_id)
|
|
return {
|
|
int(round(float(row["chainage_m"]))): row["design"]
|
|
for row in rows
|
|
if isinstance(row, dict) and isinstance(row.get("design"), dict)
|
|
}
|
|
|
|
|
|
@router.get("/{project_id}/design-drawings", response_model=DesignDrawingListResponse)
|
|
async def get_design_drawing_list(
|
|
project_id: UUID,
|
|
) -> DesignDrawingListResponse | JSONResponse:
|
|
"""B07 좌측 패널용 도면 메타데이터만 캐시한다."""
|
|
try:
|
|
route_id, project_root, longitudinal_path = await _confirmed_source(project_id)
|
|
designs = await _designs_by_chainage(route_id)
|
|
drawings = await asyncio.to_thread(_drawing_list, project_root, longitudinal_path, designs)
|
|
return DesignDrawingListResponse(
|
|
project_id=str(project_id), route_id=route_id, drawings=drawings
|
|
)
|
|
except FileNotFoundError as exc:
|
|
return JSONResponse(status_code=404, content={"status": "error", "message": str(exc)})
|
|
except PermissionError as exc:
|
|
return JSONResponse(status_code=409, content={"status": "error", "message": str(exc)})
|
|
except Exception:
|
|
logger.exception("B07 도면 목록 조회 실패: project_id=%s", project_id)
|
|
return JSONResponse(
|
|
status_code=500,
|
|
content={"status": "error", "message": "상세 설계 도면 목록을 읽지 못했습니다."},
|
|
)
|
|
|
|
|
|
@router.get("/{project_id}/design-drawings/{drawing_id}", response_model=DesignDrawingResponse)
|
|
async def get_design_drawing(
|
|
project_id: UUID, drawing_id: str
|
|
) -> DesignDrawingResponse | JSONResponse:
|
|
"""선택한 도면 원본 한 건만 읽어 CAD 스키마로 변환한다."""
|
|
try:
|
|
route_id, project_root, longitudinal_path = await _confirmed_source(project_id)
|
|
# 횡단도는 B06 지정 설계를 먼저 읽어 CAD 계획선(design_line)과 응답에 함께 쓴다.
|
|
design: dict[str, Any] | None = None
|
|
source_design: Any = None
|
|
cross_match = _CROSS_ID.fullmatch(drawing_id)
|
|
if cross_match:
|
|
pool = get_db_pool()
|
|
async with pool.acquire() as connection:
|
|
design = await get_cross_section_design(
|
|
connection, route_id, int(cross_match.group(1))
|
|
)
|
|
source_design = design
|
|
elif CROSS_SHEET_ID.fullmatch(drawing_id):
|
|
# 장은 여러 측점을 담으므로 노선 전체 지정을 한 번에 읽어 넘긴다.
|
|
source_design = await _designs_by_chainage(route_id)
|
|
kind, label, drawing, confirmed, quantity_table = await asyncio.to_thread(
|
|
_read_drawing, project_root, longitudinal_path, drawing_id, source_design
|
|
)
|
|
return DesignDrawingResponse(
|
|
project_id=str(project_id),
|
|
route_id=route_id,
|
|
id=drawing_id,
|
|
kind=kind,
|
|
label=label,
|
|
drawing=drawing,
|
|
confirmed=confirmed,
|
|
quantity_table=quantity_table,
|
|
design=design,
|
|
)
|
|
except ValueError as exc:
|
|
return JSONResponse(status_code=400, content={"status": "error", "message": str(exc)})
|
|
except FileNotFoundError as exc:
|
|
return JSONResponse(status_code=404, content={"status": "error", "message": str(exc)})
|
|
except PermissionError as exc:
|
|
return JSONResponse(status_code=409, content={"status": "error", "message": str(exc)})
|
|
except Exception:
|
|
logger.exception(
|
|
"B07 단건 도면 조회 실패: project_id=%s drawing_id=%s", project_id, drawing_id
|
|
)
|
|
return JSONResponse(
|
|
status_code=500,
|
|
content={"status": "error", "message": "상세 설계 도면을 읽지 못했습니다."},
|
|
)
|
|
|
|
|
|
def _recompute_confirmed_design(
|
|
longitudinal_path: Path, cross_stem: str, designation: dict[str, Any]
|
|
) -> dict[str, Any]:
|
|
"""B06 지정값과 현재 계획고로 절·성토 단면적을 재계산해 확정치(status=confirmed)로 만든다.
|
|
|
|
B07 CAD에는 아직 편집 가능한 설계선이 없으므로, 저장된 지정값(지반유형·단면유형·
|
|
측구위치)과 계획고로 동일 엔진을 재실행해 확정 시점 값을 고정한다.
|
|
"""
|
|
longitudinal = _read_json(longitudinal_path)
|
|
cross_path = longitudinal_path.parent.parent / "cross_sections" / f"{cross_stem}.json"
|
|
source = _read_json(cross_path)
|
|
samples = source.get("samples")
|
|
if not isinstance(samples, list):
|
|
raise ValueError("횡단 상세 파일 형식이 올바르지 않습니다.")
|
|
design_elevation = design_elevation_from_longitudinal(
|
|
longitudinal, float(source.get("chainage_m", 0.0))
|
|
)
|
|
design = compute_cross_design(
|
|
samples,
|
|
design_elevation,
|
|
ground_type=designation["ground_type"],
|
|
section_mode=designation["section_mode"],
|
|
ditch_side=designation.get("ditch_side"),
|
|
)
|
|
design["status"] = "confirmed"
|
|
return design
|
|
|
|
|
|
@router.put(
|
|
"/{project_id}/design-drawings/{drawing_id}/confirm",
|
|
response_model=DesignDrawingConfirmResponse,
|
|
)
|
|
async def confirm_design_drawing(
|
|
project_id: UUID, drawing_id: str, request: DesignDrawingConfirmRequest
|
|
) -> DesignDrawingConfirmResponse | JSONResponse:
|
|
"""현재 편집 도면을 영구 저장하고 도면별 확정 상태를 기록한다.
|
|
|
|
횡단도 확정 시 B06 지정 잠정치를 동일 엔진으로 재계산해 확정치로 승격·저장한다.
|
|
"""
|
|
try:
|
|
route_id, project_root, longitudinal_path = await _confirmed_source(project_id)
|
|
designs = await _designs_by_chainage(route_id)
|
|
items = await asyncio.to_thread(_drawing_list, project_root, longitudinal_path, designs)
|
|
item = next((candidate for candidate in items if candidate.id == drawing_id), None)
|
|
if not item:
|
|
raise FileNotFoundError("확정할 도면을 찾을 수 없습니다.")
|
|
sheet = None
|
|
if CROSS_SHEET_ID.fullmatch(drawing_id):
|
|
plan = await asyncio.to_thread(
|
|
_cross_sheet_plan, project_root, longitudinal_path, designs
|
|
)
|
|
sheet = next((entry for entry in plan if entry["id"] == drawing_id), None)
|
|
# 수량표는 CAD 테이블(Text 엔티티)에서 역추출을 우선하고, 없으면 요청 본문 폴백.
|
|
# 장에는 측점이 여럿이라 측점별로 뽑는다 (엔티티 id 씨앗 = "{장id}:{측점}").
|
|
quantity_table = None
|
|
quantity_tables: dict[str, Any] = {}
|
|
if sheet is not None:
|
|
for chainage in sheet["chainages"]:
|
|
table = extract_quantity_table(f"{drawing_id}:{chainage}", request.drawing)
|
|
if table:
|
|
quantity_tables[str(chainage)] = table
|
|
elif item.kind == "cross":
|
|
quantity_table = (
|
|
extract_quantity_table(drawing_id, request.drawing) or request.quantity_table
|
|
)
|
|
# 단계 완료 기준은 횡단도(cross)만 본다. 종단도(longitudinal)는 확정 여부와 무관.
|
|
all_confirmed = await asyncio.to_thread(
|
|
_store_confirmed_drawing,
|
|
project_root,
|
|
item,
|
|
request.drawing,
|
|
{candidate.id for candidate in items if candidate.kind == "cross"},
|
|
quantity_table,
|
|
quantity_tables or None,
|
|
)
|
|
|
|
# 횡단도면이면 확정 단면적을 재계산한다 (재계산 실패는 도면 확정을 막지 않음).
|
|
# 장은 담긴 측점 전부를 함께 확정한다.
|
|
recomputed: list[tuple[int, dict[str, Any]]] = []
|
|
cross_match = _CROSS_ID.fullmatch(drawing_id)
|
|
pool = get_db_pool()
|
|
targets: list[int] = []
|
|
if sheet is not None:
|
|
targets = list(sheet["chainages"])
|
|
elif item.kind == "cross" and cross_match:
|
|
targets = [int(cross_match.group(1))]
|
|
for chainage_int in targets:
|
|
designation = designs.get(chainage_int)
|
|
if not designation:
|
|
continue
|
|
try:
|
|
recomputed.append(
|
|
(
|
|
chainage_int,
|
|
await asyncio.to_thread(
|
|
_recompute_confirmed_design,
|
|
longitudinal_path,
|
|
f"cross_{chainage_int:05d}m",
|
|
designation,
|
|
),
|
|
)
|
|
)
|
|
except (ValueError, KeyError, FileNotFoundError, OSError):
|
|
logger.warning(
|
|
"B07 확정 단면적 재계산 실패 (도면 확정은 유지): drawing_id=%s 측점=%s",
|
|
drawing_id,
|
|
chainage_int,
|
|
exc_info=True,
|
|
)
|
|
|
|
async with pool.acquire() as connection:
|
|
await connection.begin()
|
|
try:
|
|
for chainage_int, confirmed_design in recomputed:
|
|
await merge_cross_section_design_by_round(
|
|
connection,
|
|
route_id=route_id,
|
|
chainage_int=chainage_int,
|
|
patch=confirmed_design,
|
|
)
|
|
async with connection.cursor() as cursor:
|
|
if all_confirmed:
|
|
await complete_stage(cursor, str(project_id), 4)
|
|
else:
|
|
await start_stage(cursor, str(project_id), 4)
|
|
await connection.commit()
|
|
except Exception:
|
|
await connection.rollback()
|
|
raise
|
|
return DesignDrawingConfirmResponse(
|
|
project_id=str(project_id),
|
|
id=drawing_id,
|
|
confirmed=True,
|
|
all_confirmed=all_confirmed,
|
|
design=recomputed[0][1] if len(recomputed) == 1 else None,
|
|
)
|
|
except ValueError as exc:
|
|
return JSONResponse(status_code=400, content={"status": "error", "message": str(exc)})
|
|
except FileNotFoundError as exc:
|
|
return JSONResponse(status_code=404, content={"status": "error", "message": str(exc)})
|
|
except PermissionError as exc:
|
|
return JSONResponse(status_code=409, content={"status": "error", "message": str(exc)})
|
|
except Exception:
|
|
logger.exception("B07 도면 확정 실패: project_id=%s drawing_id=%s", project_id, drawing_id)
|
|
return JSONResponse(
|
|
status_code=500,
|
|
content={"status": "error", "message": "상세 설계 도면을 확정하지 못했습니다."},
|
|
)
|
|
|
|
|
|
@router.post(
|
|
"/{project_id}/design-drawings/{drawing_id}/invalidate",
|
|
response_model=DesignDrawingInvalidateResponse,
|
|
)
|
|
async def invalidate_design_drawing(
|
|
project_id: UUID, drawing_id: str
|
|
) -> DesignDrawingInvalidateResponse | JSONResponse:
|
|
"""확정 도면 편집 시 B07 및 이후 단계를 미확정 상태로 되돌린다."""
|
|
try:
|
|
route_id, project_root, longitudinal_path = await _confirmed_source(project_id)
|
|
items = await asyncio.to_thread(_drawing_list, project_root, longitudinal_path)
|
|
if drawing_id not in {item.id for item in items}:
|
|
raise FileNotFoundError("변경된 도면을 찾을 수 없습니다.")
|
|
await asyncio.to_thread(_invalidate_drawing, project_root, drawing_id)
|
|
|
|
cross_match = _CROSS_ID.fullmatch(drawing_id)
|
|
stale: list[int] = []
|
|
if cross_match:
|
|
stale = [int(cross_match.group(1))]
|
|
elif CROSS_SHEET_ID.fullmatch(drawing_id):
|
|
designs = await _designs_by_chainage(route_id)
|
|
plan = await asyncio.to_thread(
|
|
_cross_sheet_plan, project_root, longitudinal_path, designs
|
|
)
|
|
sheet = next((entry for entry in plan if entry["id"] == drawing_id), None)
|
|
stale = list(sheet["chainages"]) if sheet else []
|
|
pool = get_db_pool()
|
|
async with pool.acquire() as connection:
|
|
await connection.begin()
|
|
try:
|
|
# 확정 도면을 편집하면 담긴 측점 설계도 잠정 상태로 되돌린다.
|
|
for chainage_int in stale:
|
|
await merge_cross_section_design_by_round(
|
|
connection,
|
|
route_id=route_id,
|
|
chainage_int=chainage_int,
|
|
patch={"status": "provisional"},
|
|
)
|
|
async with connection.cursor() as cursor:
|
|
await start_stage(cursor, str(project_id), 4)
|
|
await connection.commit()
|
|
except Exception:
|
|
await connection.rollback()
|
|
raise
|
|
return DesignDrawingInvalidateResponse(
|
|
project_id=str(project_id),
|
|
id=drawing_id,
|
|
)
|
|
except FileNotFoundError as exc:
|
|
return JSONResponse(status_code=404, content={"status": "error", "message": str(exc)})
|
|
except PermissionError as exc:
|
|
return JSONResponse(status_code=409, content={"status": "error", "message": str(exc)})
|
|
except Exception:
|
|
logger.exception(
|
|
"B07 도면 확정 해제 실패: project_id=%s drawing_id=%s", project_id, drawing_id
|
|
)
|
|
return JSONResponse(
|
|
status_code=500,
|
|
content={"status": "error", "message": "상세 설계 도면 상태를 되돌리지 못했습니다."},
|
|
)
|