508 lines
23 KiB
Python
508 lines
23 KiB
Python
"""B06 종횡단 생성 FastAPI 라우터."""
|
|
|
|
import asyncio
|
|
import json
|
|
import logging
|
|
from pathlib import Path
|
|
from typing import Any
|
|
from uuid import UUID
|
|
|
|
import aiomysql
|
|
from fastapi import APIRouter
|
|
from fastapi.responses import JSONResponse
|
|
|
|
from B03_FileInput.B03_FileInput_Repository import get_project_storage_relative_path
|
|
from B05_wf2_Route.B05_wf2_Route_Engine_Sections import (
|
|
cross_filename,
|
|
prune_stale_cross_files,
|
|
run_section_generation,
|
|
)
|
|
from B05_wf2_Route.B05_wf2_Route_Engine_Sections_Core import SectionGenerationOptions
|
|
from B06_wf3_ProfileCross.B06_wf3_ProfileCross_Engine_Design import (
|
|
compute_cross_design,
|
|
design_elevation_from_longitudinal,
|
|
)
|
|
from B06_wf3_ProfileCross.B06_wf3_ProfileCross_Repository import (
|
|
confirm_sections_for_route,
|
|
count_cross_sections,
|
|
create_longitudinal_section,
|
|
delete_sections_for_route,
|
|
get_confirmed_route_context,
|
|
get_cross_section_designs,
|
|
get_cross_sections_missing_design_chainages,
|
|
get_latest_section_options,
|
|
get_longitudinal_section,
|
|
get_route_generation_source,
|
|
insert_cross_sections,
|
|
update_cross_section_design,
|
|
)
|
|
from B06_wf3_ProfileCross.B06_wf3_ProfileCross_Schema import (
|
|
CrossDesignRequest,
|
|
CrossDesignResponse,
|
|
SectionConfirmResponse,
|
|
SectionContextResponse,
|
|
SectionDetailResponse,
|
|
SectionOptionDefaults,
|
|
SectionRegenerateRequest,
|
|
SectionSummaryResponse,
|
|
)
|
|
from common_util.common_util_storage import resolve_stored_project_path
|
|
from common_util.common_util_surface_confirmation import get_surface_confirmation_params
|
|
from common_util.common_util_workflow_state import complete_stage, get_workflow_state
|
|
from config.config_db import get_db_pool
|
|
from config.config_system import FOREST_ROAD_MIN_WIDTH_M, SECTION_VERTICAL_EXAGGERATION
|
|
|
|
logger = logging.getLogger(__name__)
|
|
router = APIRouter(prefix="/api/projects", tags=["B06 Profile Cross"])
|
|
|
|
|
|
@router.get("/{project_id}/sections/context", response_model=SectionContextResponse)
|
|
async def get_section_context(project_id: UUID) -> SectionContextResponse | JSONResponse:
|
|
"""최신 확정 경로와 B04 확정값, config 기반 생성 기본값을 반환한다."""
|
|
pool = get_db_pool()
|
|
try:
|
|
async with pool.acquire() as connection:
|
|
route_context = await get_confirmed_route_context(connection, project_id)
|
|
surface_params = await get_surface_confirmation_params(connection, str(project_id))
|
|
|
|
defaults = SectionGenerationOptions()
|
|
return SectionContextResponse(
|
|
project_id=str(project_id),
|
|
route_id=route_context["route_id"] if route_context else None,
|
|
filter_key=surface_params["source_filter"] if route_context else None,
|
|
method=surface_params["method"] if route_context else None,
|
|
smooth=bool(surface_params["smooth"]) if route_context else None,
|
|
crs_epsg=route_context["crs_epsg"] if route_context else None,
|
|
defaults=SectionOptionDefaults(
|
|
station_interval_m=defaults.station_interval_m,
|
|
cross_half_width_m=defaults.cross_half_width_m,
|
|
cross_sample_interval_m=defaults.cross_sample_interval_m,
|
|
long_sample_interval_m=defaults.long_sample_interval_m,
|
|
vertical_exaggeration=SECTION_VERTICAL_EXAGGERATION,
|
|
),
|
|
)
|
|
except Exception:
|
|
logger.exception("B06 종횡단 컨텍스트 조회 실패: project_id=%s", project_id)
|
|
return JSONResponse(
|
|
status_code=500,
|
|
content={"status": "error", "message": "종횡단 컨텍스트 조회 중 오류가 발생했습니다."},
|
|
)
|
|
|
|
|
|
@router.get("/{project_id}/sections/road-widths")
|
|
async def get_forest_road_min_widths(project_id: UUID) -> dict[str, dict[str, float]]:
|
|
"""B05 측점 가로선에 적용할 임도 등급별 법정 최소너비를 반환한다."""
|
|
return {"forest_road_min_width_m": FOREST_ROAD_MIN_WIDTH_M}
|
|
|
|
|
|
@router.get("/{project_id}/sections/{route_id}", response_model=SectionSummaryResponse)
|
|
async def get_sections(project_id: UUID, route_id: int) -> SectionSummaryResponse | JSONResponse:
|
|
"""경로의 종단면 요약을 조회한다."""
|
|
pool = get_db_pool()
|
|
try:
|
|
async with pool.acquire() as connection:
|
|
longitudinal = await get_longitudinal_section(connection, project_id, route_id)
|
|
cross_section_count = await count_cross_sections(connection, route_id)
|
|
data = longitudinal.get("data") if longitudinal else None
|
|
length_m = data.get("length_m") if isinstance(data, dict) else None
|
|
return SectionSummaryResponse(
|
|
project_id=str(project_id),
|
|
route_id=route_id,
|
|
longitudinal=longitudinal,
|
|
length_m=length_m,
|
|
cross_section_count=cross_section_count,
|
|
)
|
|
except Exception:
|
|
logger.exception("B06 종횡단 조회 실패: project_id=%s", project_id)
|
|
return JSONResponse(
|
|
status_code=500,
|
|
content={"status": "error", "message": "종횡단 조회 중 오류가 발생했습니다."},
|
|
)
|
|
|
|
|
|
def _read_section_detail(project_root: Path, longitudinal_file_path: str) -> dict:
|
|
"""검증된 프로젝트 루트 안의 종단 및 횡단 JSON을 읽는다."""
|
|
root = project_root.resolve()
|
|
longitudinal_path = (root / longitudinal_file_path).resolve()
|
|
if root not in longitudinal_path.parents:
|
|
raise ValueError("종단면 파일 경로가 프로젝트 저장소를 벗어났습니다.")
|
|
if not longitudinal_path.is_file():
|
|
raise FileNotFoundError("종단면 상세 파일을 찾을 수 없습니다.")
|
|
|
|
stage_root = longitudinal_path.parent.parent
|
|
cross_dir = stage_root / "cross_sections"
|
|
if not cross_dir.is_dir():
|
|
raise FileNotFoundError("횡단면 상세 파일을 찾을 수 없습니다.")
|
|
|
|
longitudinal = json.loads(longitudinal_path.read_text(encoding="utf-8"))
|
|
stations = longitudinal.get("stations") if isinstance(longitudinal, dict) else None
|
|
valid_names = prune_stale_cross_files(cross_dir, stations if isinstance(stations, list) else [])
|
|
cross_sections = [
|
|
json.loads(path.read_text(encoding="utf-8"))
|
|
for path in sorted(cross_dir.glob("cross_*.json"))
|
|
if not valid_names or path.name in valid_names
|
|
]
|
|
if not isinstance(longitudinal, dict) or not all(
|
|
isinstance(section, dict) for section in cross_sections
|
|
):
|
|
raise ValueError("종횡단 상세 파일 형식이 올바르지 않습니다.")
|
|
return {"longitudinal": longitudinal, "cross_sections": cross_sections}
|
|
|
|
|
|
@router.get("/{project_id}/sections/{route_id}/detail", response_model=SectionDetailResponse)
|
|
async def get_section_detail(
|
|
project_id: UUID, route_id: int
|
|
) -> SectionDetailResponse | JSONResponse:
|
|
"""경로의 SVG 렌더링용 종단·횡단 원시 샘플을 반환한다."""
|
|
pool = get_db_pool()
|
|
try:
|
|
async with pool.acquire() as connection:
|
|
longitudinal = await get_longitudinal_section(connection, project_id, route_id)
|
|
if not longitudinal:
|
|
return JSONResponse(
|
|
status_code=404,
|
|
content={"status": "error", "message": "종횡단 상세 결과가 없습니다."},
|
|
)
|
|
stored_path = await get_project_storage_relative_path(connection, project_id)
|
|
designs = await get_cross_section_designs(connection, route_id)
|
|
project_root = Path(resolve_stored_project_path(stored_path))
|
|
detail = await asyncio.to_thread(
|
|
_read_section_detail,
|
|
project_root,
|
|
str(longitudinal["longitudinal_file_path"]),
|
|
)
|
|
# DB에만 있는 잠정 설계 지정을 chainage 근사로 각 횡단에 얹어 화면 복원을 돕는다.
|
|
for record in designs:
|
|
for section in detail["cross_sections"]:
|
|
if abs(float(section.get("chainage_m", 0.0)) - record["chainage_m"]) < 0.01:
|
|
section["design"] = record["design"]
|
|
break
|
|
# 지정값이 없는 측점은 기본값(토사/좌절토)으로 즉석 계산해 프리뷰로 채운다.
|
|
# (미저장 프리뷰: 실제 저장은 사용자가 카드를 조작하거나 확정할 때 이뤄진다.)
|
|
await asyncio.to_thread(
|
|
_attach_default_designs, detail["longitudinal"], detail["cross_sections"]
|
|
)
|
|
return SectionDetailResponse(**detail)
|
|
except FileNotFoundError as exc:
|
|
return JSONResponse(status_code=404, content={"status": "error", "message": str(exc)})
|
|
except (OSError, ValueError, json.JSONDecodeError) as exc:
|
|
logger.warning(
|
|
"B06 종횡단 상세 파일 조회 실패: project_id=%s route_id=%s error=%s",
|
|
project_id,
|
|
route_id,
|
|
exc,
|
|
)
|
|
return JSONResponse(
|
|
status_code=500,
|
|
content={"status": "error", "message": "종횡단 상세 파일을 읽지 못했습니다."},
|
|
)
|
|
except Exception:
|
|
logger.exception(
|
|
"B06 종횡단 상세 조회 실패: project_id=%s route_id=%s", project_id, route_id
|
|
)
|
|
return JSONResponse(
|
|
status_code=500,
|
|
content={"status": "error", "message": "종횡단 상세 조회 중 오류가 발생했습니다."},
|
|
)
|
|
|
|
|
|
def _regeneration_options(
|
|
stored_options: dict[str, Any] | None,
|
|
stage_params: dict[str, Any] | None,
|
|
cross_half_width_m: float,
|
|
) -> SectionGenerationOptions:
|
|
"""DB 저장 옵션(단일 소스) → stage 2 params → config 순으로 유지하고 반폭만 교체한다."""
|
|
defaults = SectionGenerationOptions()
|
|
stored = stored_options or {}
|
|
params = stage_params or {}
|
|
|
|
def pick(key: str, default: float) -> float:
|
|
return stored.get(key) or params.get(key) or default
|
|
|
|
return SectionGenerationOptions(
|
|
station_interval_m=pick("station_interval_m", defaults.station_interval_m),
|
|
cross_half_width_m=cross_half_width_m,
|
|
cross_sample_interval_m=pick("cross_sample_interval_m", defaults.cross_sample_interval_m),
|
|
long_sample_interval_m=pick("long_sample_interval_m", defaults.long_sample_interval_m),
|
|
include_endpoint=defaults.include_endpoint,
|
|
)
|
|
|
|
|
|
@router.post("/{project_id}/sections/{route_id}/regenerate", response_model=SectionDetailResponse)
|
|
async def regenerate_sections(
|
|
project_id: UUID, route_id: int, request: SectionRegenerateRequest
|
|
) -> SectionDetailResponse | JSONResponse:
|
|
"""표시 옵션의 횡단 반폭으로 종횡단을 재생성해 저장하고 상세를 반환한다."""
|
|
pool = get_db_pool()
|
|
try:
|
|
async with pool.acquire() as connection:
|
|
source = await get_route_generation_source(connection, project_id, route_id)
|
|
if not source:
|
|
return JSONResponse(
|
|
status_code=404,
|
|
content={"status": "error", "message": "재생성할 경로가 없습니다."},
|
|
)
|
|
surface_params = await get_surface_confirmation_params(connection, str(project_id))
|
|
stored_path = await get_project_storage_relative_path(connection, project_id)
|
|
stored_options = await get_latest_section_options(connection, project_id)
|
|
async with connection.cursor(aiomysql.DictCursor) as cursor:
|
|
workflow = await get_workflow_state(cursor, str(project_id))
|
|
route_stage = next(
|
|
(stage for stage in workflow["stages"] if stage["stage_no"] == 2),
|
|
None,
|
|
)
|
|
project_root = Path(resolve_stored_project_path(stored_path))
|
|
crs_epsg = source["crs_epsg"]
|
|
sections = await asyncio.to_thread(
|
|
run_section_generation,
|
|
project_root,
|
|
source["route_data_path"],
|
|
surface_params["source_filter"],
|
|
surface_params["method"],
|
|
bool(surface_params["smooth"]),
|
|
options=_regeneration_options(
|
|
stored_options,
|
|
route_stage.get("params") if route_stage else None,
|
|
request.cross_half_width_m,
|
|
),
|
|
crs=f"EPSG:{crs_epsg}" if crs_epsg is not None else None,
|
|
)
|
|
await connection.begin()
|
|
try:
|
|
await delete_sections_for_route(connection, route_id)
|
|
await create_longitudinal_section(
|
|
connection,
|
|
project_id=project_id,
|
|
route_id=route_id,
|
|
data=sections["longitudinal"]["data"],
|
|
longitudinal_file_path=sections["longitudinal"]["file_path"],
|
|
)
|
|
await insert_cross_sections(
|
|
connection,
|
|
project_id=project_id,
|
|
route_id=route_id,
|
|
sections=sections["cross_sections"],
|
|
)
|
|
await connection.commit()
|
|
except Exception:
|
|
await connection.rollback()
|
|
raise
|
|
result = sections["result"]
|
|
return SectionDetailResponse(
|
|
longitudinal=result["longitudinal"], cross_sections=result["cross_sections"]
|
|
)
|
|
except FileNotFoundError as exc:
|
|
return JSONResponse(status_code=404, content={"status": "error", "message": str(exc)})
|
|
except ValueError as exc:
|
|
return JSONResponse(status_code=400, content={"status": "error", "message": str(exc)})
|
|
except Exception:
|
|
logger.exception("B06 종횡단 재생성 실패: project_id=%s route_id=%s", project_id, route_id)
|
|
return JSONResponse(
|
|
status_code=500,
|
|
content={"status": "error", "message": "종횡단 재생성 처리 중 오류가 발생했습니다."},
|
|
)
|
|
|
|
|
|
def _read_cross_design_inputs(
|
|
project_root: Path, longitudinal_file_path: str, chainage_m: float
|
|
) -> tuple[list[dict], float | None]:
|
|
"""측점 하나의 지반 샘플과 계획고를 파일에서 읽는다 (경로 이탈 검증 포함)."""
|
|
root = project_root.resolve()
|
|
longitudinal_path = (root / longitudinal_file_path).resolve()
|
|
if root not in longitudinal_path.parents:
|
|
raise ValueError("종단면 파일 경로가 프로젝트 저장소를 벗어났습니다.")
|
|
if not longitudinal_path.is_file():
|
|
raise FileNotFoundError("종단면 상세 파일을 찾을 수 없습니다.")
|
|
longitudinal = json.loads(longitudinal_path.read_text(encoding="utf-8"))
|
|
design_elevation = design_elevation_from_longitudinal(longitudinal, chainage_m)
|
|
|
|
cross_dir = longitudinal_path.parent.parent / "cross_sections"
|
|
cross_path = (cross_dir / cross_filename(chainage_m)).resolve()
|
|
if cross_dir.resolve() not in cross_path.parents or not cross_path.is_file():
|
|
raise FileNotFoundError("해당 측점의 횡단 상세 파일을 찾을 수 없습니다.")
|
|
cross = json.loads(cross_path.read_text(encoding="utf-8"))
|
|
samples = cross.get("samples") if isinstance(cross, dict) else None
|
|
if not isinstance(samples, list):
|
|
raise ValueError("횡단 상세 파일 형식이 올바르지 않습니다.")
|
|
return samples, design_elevation
|
|
|
|
|
|
def _attach_default_designs(
|
|
longitudinal: dict[str, Any], cross_sections: list[dict[str, Any]]
|
|
) -> None:
|
|
"""지정 설계가 없는 횡단에 기본값(토사/좌절토) 프리뷰 설계를 즉석 계산해 얹는다.
|
|
|
|
detail 조회가 이미 읽어온 samples와 종단 계획선을 그대로 써서 추가 파일 I/O 없이
|
|
전 측점 프리뷰를 만든다(미저장). 계산 불가 측점은 건너뛴다.
|
|
"""
|
|
for section in cross_sections:
|
|
if section.get("design"):
|
|
continue
|
|
try:
|
|
design = compute_cross_design(
|
|
section.get("samples", []),
|
|
design_elevation_from_longitudinal(
|
|
longitudinal, float(section.get("chainage_m", 0.0))
|
|
),
|
|
ground_type="soil",
|
|
section_mode="left_cut",
|
|
)
|
|
design["status"] = "provisional"
|
|
section["design"] = design
|
|
except (ValueError, KeyError):
|
|
continue
|
|
|
|
|
|
def _compute_default_designs(
|
|
project_root: Path, longitudinal_file_path: str, chainages: list[float]
|
|
) -> list[tuple[float, dict[str, Any]]]:
|
|
"""미지정 측점들을 기본값(토사/좌절토)으로 계산한 (chainage, design) 목록을 만든다.
|
|
|
|
계획고 부재 등으로 계산 불가한 측점은 조용히 건너뛴다(확정을 막지 않기 위함).
|
|
"""
|
|
root = project_root.resolve()
|
|
longitudinal_path = (root / longitudinal_file_path).resolve()
|
|
if root not in longitudinal_path.parents or not longitudinal_path.is_file():
|
|
return []
|
|
longitudinal = json.loads(longitudinal_path.read_text(encoding="utf-8"))
|
|
cross_dir = longitudinal_path.parent.parent / "cross_sections"
|
|
results: list[tuple[float, dict[str, Any]]] = []
|
|
for chainage_m in chainages:
|
|
cross_path = cross_dir / cross_filename(chainage_m)
|
|
if not cross_path.is_file():
|
|
continue
|
|
try:
|
|
cross = json.loads(cross_path.read_text(encoding="utf-8"))
|
|
samples = cross.get("samples")
|
|
if not isinstance(samples, list):
|
|
continue
|
|
design = compute_cross_design(
|
|
samples,
|
|
design_elevation_from_longitudinal(longitudinal, chainage_m),
|
|
ground_type="soil",
|
|
section_mode="left_cut",
|
|
)
|
|
design["status"] = "provisional"
|
|
results.append((chainage_m, design))
|
|
except (ValueError, KeyError, OSError, json.JSONDecodeError):
|
|
continue
|
|
return results
|
|
|
|
|
|
@router.post("/{project_id}/sections/{route_id}/cross-design", response_model=CrossDesignResponse)
|
|
async def compute_cross_section_design(
|
|
project_id: UUID, route_id: int, request: CrossDesignRequest
|
|
) -> CrossDesignResponse | JSONResponse:
|
|
"""측점 표준횡단 설계(지반유형·단면유형)를 즉시 계산해 잠정치로 저장한다."""
|
|
pool = get_db_pool()
|
|
try:
|
|
async with pool.acquire() as connection:
|
|
longitudinal = await get_longitudinal_section(connection, project_id, route_id)
|
|
if not longitudinal:
|
|
return JSONResponse(
|
|
status_code=404,
|
|
content={"status": "error", "message": "종횡단 상세 결과가 없습니다."},
|
|
)
|
|
stored_path = await get_project_storage_relative_path(connection, project_id)
|
|
project_root = Path(resolve_stored_project_path(stored_path))
|
|
samples, design_elevation = await asyncio.to_thread(
|
|
_read_cross_design_inputs,
|
|
project_root,
|
|
str(longitudinal["longitudinal_file_path"]),
|
|
request.chainage_m,
|
|
)
|
|
design = compute_cross_design(
|
|
samples,
|
|
design_elevation,
|
|
ground_type=request.ground_type,
|
|
section_mode=request.section_mode,
|
|
ditch_side=request.ditch_side,
|
|
)
|
|
# B06 지정 시점은 잠정치. B07 도면 확정 시 동일 엔진으로 재계산해 confirmed로 승격한다.
|
|
design["status"] = "provisional"
|
|
async with pool.acquire() as connection:
|
|
await connection.begin()
|
|
try:
|
|
updated = await update_cross_section_design(
|
|
connection,
|
|
route_id=route_id,
|
|
chainage_m=request.chainage_m,
|
|
design=design,
|
|
)
|
|
await connection.commit()
|
|
except Exception:
|
|
await connection.rollback()
|
|
raise
|
|
if not updated:
|
|
return JSONResponse(
|
|
status_code=404,
|
|
content={"status": "error", "message": "해당 측점 횡단 레코드가 없습니다."},
|
|
)
|
|
return CrossDesignResponse(chainage_m=request.chainage_m, design=design)
|
|
except FileNotFoundError as exc:
|
|
return JSONResponse(status_code=404, content={"status": "error", "message": str(exc)})
|
|
except ValueError as exc:
|
|
return JSONResponse(status_code=400, content={"status": "error", "message": str(exc)})
|
|
except Exception:
|
|
logger.exception(
|
|
"B06 측점 횡단 설계 계산 실패: project_id=%s route_id=%s", project_id, route_id
|
|
)
|
|
return JSONResponse(
|
|
status_code=500,
|
|
content={"status": "error", "message": "횡단 설계 계산 중 오류가 발생했습니다."},
|
|
)
|
|
|
|
|
|
@router.post("/{project_id}/sections/{route_id}/confirm", response_model=SectionConfirmResponse)
|
|
async def confirm_sections(
|
|
project_id: UUID, route_id: int
|
|
) -> SectionConfirmResponse | JSONResponse:
|
|
"""경로의 종횡단면을 확정(CONFIRMED)한다.
|
|
|
|
지반유형을 지정하지 않은 측점은 기본값(토사/좌절토)으로 자동 채운 뒤 확정한다.
|
|
"""
|
|
pool = get_db_pool()
|
|
try:
|
|
async with pool.acquire() as connection:
|
|
existing = await get_longitudinal_section(connection, project_id, route_id)
|
|
if not existing:
|
|
return JSONResponse(
|
|
status_code=404,
|
|
content={"status": "error", "message": "확정할 종횡단이 없습니다."},
|
|
)
|
|
stored_path = await get_project_storage_relative_path(connection, project_id)
|
|
missing = await get_cross_sections_missing_design_chainages(connection, route_id)
|
|
|
|
# 미지정 측점을 기본값으로 계산해 채운다 (계산 불가 측점은 조용히 건너뜀).
|
|
default_designs: list[tuple[float, dict[str, Any]]] = []
|
|
if missing:
|
|
project_root = Path(resolve_stored_project_path(stored_path))
|
|
default_designs = await asyncio.to_thread(
|
|
_compute_default_designs,
|
|
project_root,
|
|
str(existing["longitudinal_file_path"]),
|
|
missing,
|
|
)
|
|
|
|
async with pool.acquire() as connection:
|
|
await connection.begin()
|
|
try:
|
|
for chainage_m, design in default_designs:
|
|
await update_cross_section_design(
|
|
connection, route_id=route_id, chainage_m=chainage_m, design=design
|
|
)
|
|
await confirm_sections_for_route(connection, route_id)
|
|
async with connection.cursor() as cursor:
|
|
await complete_stage(cursor, str(project_id), 3)
|
|
await connection.commit()
|
|
except Exception:
|
|
await connection.rollback()
|
|
raise
|
|
return SectionConfirmResponse(project_id=str(project_id), route_id=route_id)
|
|
except Exception:
|
|
logger.exception("B06 종횡단 확정 실패: project_id=%s", project_id)
|
|
return JSONResponse(
|
|
status_code=500,
|
|
content={"status": "error", "message": "종횡단 확정 처리 중 오류가 발생했습니다."},
|
|
)
|