eomsangdonandClaude Fable 5
0d5fba4400
B05에서 배치한 배수관 측점의 B06 횡단도에 배관·기슭막이(또는 집수정)·보호공을 분리 도형으로 그린다. 기존 도형(지반선·면적 밴드·포장·암 경계)은 손대지 않고, 계획선 표시 규칙만 세트와 정합을 맞춰 함께 정비했다. - 백엔드: pipe_points.json 정본(facility=="pipe")에 레지스트리 기본값을 채워 chainage ±0.02m로 맞는 횡단에 culvert 키를 얹는다(Engine_Culvert 신규, Router의 상세조회·재생성 두 경로에서 호출). 횡단 dict는 자유형이라 Pydantic 변경 없음. - 배관: 유입 invert = min(노견 아래 원지반, 노면고 − 관경 − 토피 0.5), 유출 앵커는 성토사면이 지반과 만나는 사면 끝(성토사면 길이 5m 한계 — 별표2 Ⅰ.2.차.(3)). 역경사는 수평 클램프, 길이는 m 단위 올림. 단면은 축 법선 방향 직사각형이고 외경선(실선)+내경선(점선) 2겹으로 관벽 두께를 보인다(파형강관 2.7T — KS D 3590). 끝단면은 구조물의 계류측 변을 복사해 자른다. - 기슭막이: 배면 수직 사다리꼴 + 전면 평행사변형 띠(1:0.3, t=0.45 — 실무 L3=45)의 합성 단면. 높이는 이음선 상단점이 성토 사면선과 만나는 접점까지, 하한 관경+0.45, 상한은 형태별 한계(찰 3.0/메 2.0 — 교본 7-3). 근입 0.5m(실무 기초콘크리트 H=0.5). 돌 해칭은 벽 폴리곤 clipPath 안에만 그려 어떤 높이에서도 벽 밖으로 새지 않는다. - 집수정: 절토측 유입이거나 원지반이 관 단면의 30% 이상을 막으면 자동 전환. I형(벽)·ㄴ형(+바닥)·ㄷ형(+반대측 막음), 기본 ㄴ형·내공 1.0m. 구조물이 원지반 안쪽에 박히면 절토선을 공사 계획선(보라 실선)으로 긋는다. 라벨은 I형 벽 하단. - 보호공: 시작 윗점 = 관 하단 꼭짓점, 시작 변은 기슭막이 전면과 같은 경사. 성토경사를 따라 원지반까지 내려간 뒤, 최소 길이(낙차고×2)를 채울 때까지 원지반을 따라 더 간다. - 계획선 표시: 원지반과 차이 없는 구간은 그리지 않되(보라 점선 폐지), 노면·노견· 측구부터 사면 1차 접점까지는 항상 그린다. 측구는 노견 바깥에 붙으므로 보호 범위를 측구 폭만큼 넓히고 접점 탐색도 그 바깥에서 시작한다. 배수관 측점은 벽 접점에서 트림. - B05: 배수관 측점 계획고를 지반고 + 관경 + 토피 0.5m 아래로 내리지 못하게 막고 경고 토스트를 띄운다. - 700줄 제한: 오버레이 파일이 919줄이 되어 기하(_Geom 678줄)와 그리기(247줄)로 분리했다. import 경로는 재수출로 유지해 B05·Cross_View는 건드리지 않았다. 근거 미확정(코드 주석·툴팁에 명시): 토피 0.5m(별표2 교량·암거 복토 50㎝ 교차 참조), 기슭막이 근입 0.5m(실무값), 흄관·VR관 관벽 두께(관경÷12 — KS F 4403 미확보). 검증: tsc, ruff, pytest 148 passed(신규 test_b06_culvert_set.py 6건), 헤드 브라우저 — 4+4.3·13+15.7 세트 표시, 보호공 시작점과 관 하단 꼭짓점 간격 0.6px 이내·시작 변 기울기 실측 0.223(= 0.3·H/(H+0.5) 이론값과 일치)·최소 길이 5.0m 확보, 계획선 세그먼트 연결 틈 0개(8·9·10·12·4+4.3). ⚠ 13+15.7 성토사면 5.03m로 법정 5m를 0.03m 초과 — 툴팁에 표기하고 접점 보간 여부는 사용자 판단 대기. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
825 lines
38 KiB
Python
825 lines
38 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, Depends
|
|
from fastapi.responses import JSONResponse
|
|
|
|
from B03_FileInput.B03_FileInput_Repository import get_project_storage_relative_path
|
|
from B05_Profile.B05_Profile_Engine_Grade import GradeDesignOptions, resolve_grade_options
|
|
from B05_Profile.B05_Profile_Engine_Grade_Profile import rebuild_alignment_profile
|
|
from B05_Profile.B05_Profile_Engine_Sections import (
|
|
cross_filename,
|
|
prune_stale_cross_files,
|
|
run_section_generation,
|
|
)
|
|
from B05_Profile.B05_Profile_Engine_Sections_Core import SectionGenerationOptions
|
|
from B06_Section.B06_Section_Engine_Culvert import attach_culvert_sets
|
|
from B06_Section.B06_Section_Engine_Design import compute_cross_design
|
|
from B06_Section.B06_Section_Repository import (
|
|
count_cross_sections,
|
|
create_longitudinal_section,
|
|
delete_sections_for_route,
|
|
get_confirmed_route_context,
|
|
get_cross_section_designs,
|
|
get_latest_section_options,
|
|
get_longitudinal_section,
|
|
get_project_standard_cross_section,
|
|
get_route_generation_source,
|
|
insert_cross_sections,
|
|
list_recent_company_projects,
|
|
update_cross_section_design,
|
|
)
|
|
from B06_Section.B06_Section_Schema import (
|
|
CompanyStandardListResponse,
|
|
CompanyStandardProject,
|
|
CompanyStandardResponse,
|
|
CrossDesignPreviewRequest,
|
|
CrossDesignPreviewResponse,
|
|
CrossDesignRequest,
|
|
CrossDesignResponse,
|
|
HaulEquipmentLimit,
|
|
SectionContextResponse,
|
|
SectionDetailResponse,
|
|
SectionOptionDefaults,
|
|
SectionRegenerateRequest,
|
|
SectionSummaryResponse,
|
|
)
|
|
from common_util.common_util_auth import verify_session
|
|
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_surface_confirmation import get_surface_confirmation_params
|
|
from common_util.common_util_workflow_state import get_workflow_state
|
|
from config.config_db import get_db_pool
|
|
from config.config_system import (
|
|
EARTHWORK_CONVERSION_FACTORS,
|
|
EARTHWORK_HAUL_EQUIPMENT_LIMITS_M,
|
|
FOREST_ROAD_MIN_WIDTH_M,
|
|
NATURAL_SPOIL_MIN_GROUND_SLOPE,
|
|
SECTION_VERTICAL_EXAGGERATION,
|
|
STANDARD_CROSS_SECTION,
|
|
STANDARD_ROCK_BOUNDARY_DEFAULT_OFFSET_M,
|
|
STANDARD_ROCK_BOUNDARY_STEP_M,
|
|
)
|
|
|
|
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))
|
|
# 임도 종류(projects.road_type) — B05가 계획선 법정 기준을 정하는 데 쓴다.
|
|
async with connection.cursor() as cursor:
|
|
await cursor.execute(
|
|
"SELECT road_type FROM projects WHERE id = %s", (str(project_id),)
|
|
)
|
|
row = await cursor.fetchone()
|
|
road_type = row[0] if row else None
|
|
|
|
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,
|
|
road_type=road_type,
|
|
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,
|
|
),
|
|
standard_cross_section=STANDARD_CROSS_SECTION,
|
|
rock_boundary_default_offset_m=STANDARD_ROCK_BOUNDARY_DEFAULT_OFFSET_M,
|
|
rock_boundary_step_m=STANDARD_ROCK_BOUNDARY_STEP_M,
|
|
earthwork_conversion=EARTHWORK_CONVERSION_FACTORS,
|
|
haul_equipment_limits=[
|
|
HaulEquipmentLimit(key=key, max_distance_m=limit)
|
|
for key, limit in EARTHWORK_HAUL_EQUIPMENT_LIMITS_M
|
|
],
|
|
natural_spoil_min_ground_slope=NATURAL_SPOIL_MIN_GROUND_SLOPE,
|
|
)
|
|
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}
|
|
|
|
|
|
# 아래 두 엔드포인트는 `/sections/{route_id}`(int)보다 먼저 선언해 라우팅 충돌을 막는다.
|
|
@router.get("/{project_id}/sections/company-standards", response_model=CompanyStandardListResponse)
|
|
async def list_company_standards(
|
|
project_id: UUID, session: dict[str, Any] = Depends(verify_session)
|
|
) -> CompanyStandardListResponse:
|
|
"""같은 회사의 최근 프로젝트 5개를 반환한다(설계값 보유 여부 무관)."""
|
|
company_id = session.get("company_id")
|
|
if company_id is None:
|
|
return CompanyStandardListResponse(projects=[])
|
|
pool = get_db_pool()
|
|
async with pool.acquire() as connection:
|
|
rows = await list_recent_company_projects(connection, company_id, project_id)
|
|
return CompanyStandardListResponse(
|
|
projects=[
|
|
CompanyStandardProject(project_id=str(row["project_id"]), name=row["name"])
|
|
for row in rows
|
|
]
|
|
)
|
|
|
|
|
|
@router.get(
|
|
"/{project_id}/sections/company-standards/{source_project_id}",
|
|
response_model=CompanyStandardResponse,
|
|
)
|
|
async def get_company_standard(
|
|
project_id: UUID,
|
|
source_project_id: UUID,
|
|
session: dict[str, Any] = Depends(verify_session),
|
|
) -> CompanyStandardResponse | JSONResponse:
|
|
"""특정 프로젝트의 표준횡단 설정값을 미리보기용으로 반환한다(회사 스코프 강제)."""
|
|
company_id = session.get("company_id")
|
|
if company_id is None:
|
|
return JSONResponse(
|
|
status_code=404, content={"status": "error", "message": "설계값을 찾을 수 없습니다."}
|
|
)
|
|
pool = get_db_pool()
|
|
async with pool.acquire() as connection:
|
|
standard = await get_project_standard_cross_section(
|
|
connection, company_id, source_project_id
|
|
)
|
|
if standard is None:
|
|
return JSONResponse(
|
|
status_code=404, content={"status": "error", "message": "설계값을 찾을 수 없습니다."}
|
|
)
|
|
return CompanyStandardResponse(
|
|
project_id=str(source_project_id), standard_cross_section=standard
|
|
)
|
|
|
|
|
|
@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("종횡단 상세 파일 형식이 올바르지 않습니다.")
|
|
# 배수관 측점에 세트(배관·기슭막이·보호공) 제원을 얹는다 — 횡단 카드가 그림을 그린다.
|
|
attach_culvert_sets(root, cross_sections)
|
|
return {"longitudinal": longitudinal, "cross_sections": cross_sections}
|
|
|
|
|
|
def _read_balloon_offsets(row: dict[str, Any]) -> dict[str, list[float]] | None:
|
|
"""확정 시 저장해 둔 유토곡선 balloon 위치를 꺼낸다. 없거나 형태가 깨졌으면 None."""
|
|
data = row.get("data")
|
|
if isinstance(data, (str, bytes)):
|
|
try:
|
|
data = json.loads(data)
|
|
except (ValueError, TypeError):
|
|
return None
|
|
if not isinstance(data, dict):
|
|
return None
|
|
offsets = (data.get("mass_haul") or {}).get("balloon_offsets")
|
|
if not isinstance(offsets, dict):
|
|
return None
|
|
cleaned: dict[str, list[float]] = {}
|
|
for key, value in offsets.items():
|
|
if isinstance(value, (list, tuple)) and len(value) == 2:
|
|
try:
|
|
cleaned[str(key)] = [float(value[0]), float(value[1])]
|
|
except (TypeError, ValueError):
|
|
continue
|
|
return cleaned or None
|
|
|
|
|
|
@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, balloon_offsets=_read_balloon_offsets(longitudinal))
|
|
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_grade_options(stage_params: dict[str, Any] | None) -> GradeDesignOptions | None:
|
|
"""재생성 시 종단 계획선 기준을 확정 당시 저장 파라미터(stage 2)로 재구성한다.
|
|
|
|
grade_options 없이 재생성하면 종단 계획선(profile_alignment)이 통째로 소실돼 계획
|
|
횡단도선·유토곡선·테이블이 사라진다(2026-08-06 사용자 보고 — 옛 재계산 버튼 사고의
|
|
원인). B04 파일입력→B06 자동 계산 파이프라인이 쓰는 같은 엔진 경로를 재활용한다.
|
|
"""
|
|
if not stage_params:
|
|
return None
|
|
options = stage_params.get("options") or {}
|
|
grade_class = options.get("grade_class")
|
|
if not grade_class:
|
|
return None
|
|
stored = {
|
|
key: stage_params.get(key)
|
|
for key in (
|
|
"max_grade_pct",
|
|
"min_vertical_radius_m",
|
|
"min_tangent_length_m",
|
|
"balance_segment_length_m",
|
|
"start_elevation_offset_m",
|
|
"end_elevation_offset_m",
|
|
)
|
|
if stage_params.get(key) is not None
|
|
}
|
|
# 설계속도는 B05가 저장해 둔 값을 그대로 따른다 — 없으면 종류별 기본(20).
|
|
speed = options.get("design_speed_kph")
|
|
return resolve_grade_options(
|
|
str(grade_class),
|
|
terrain_type=str(options.get("terrain_type") or "normal"),
|
|
paved=bool(options.get("paved", False)),
|
|
design_speed_kph=int(speed) if speed is not None else None,
|
|
stored=stored,
|
|
)
|
|
|
|
|
|
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"]
|
|
stage_params = route_stage.get("params") if route_stage else None
|
|
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,
|
|
stage_params,
|
|
request.cross_half_width_m,
|
|
),
|
|
# 계획선까지 함께 재계산·저장 — 없으면 계획 횡단도선·유토곡선이 사라진다.
|
|
grade_options=_regeneration_grade_options(stage_params),
|
|
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"]
|
|
# 재생성 응답도 상세 조회와 같은 배수관 세트 정보를 실어야 화면이 어긋나지 않는다.
|
|
attach_culvert_sets(project_root, result["cross_sections"])
|
|
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, bool]:
|
|
"""측점 하나의 지반 샘플·계획고·포장 제안 여부를 파일에서 읽는다 (경로 이탈 검증 포함)."""
|
|
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)
|
|
pavement_suggested = _pavement_suggestions(longitudinal).get(round(chainage_m, 3), False)
|
|
|
|
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, pavement_suggested
|
|
|
|
|
|
def _pavement_suggestions(longitudinal: dict[str, Any]) -> dict[float, bool]:
|
|
"""측점별 포장 제안(B05 solve가 법정 경사 기준으로 판정) 매핑을 만든다."""
|
|
stations = longitudinal.get("stations")
|
|
mapping: dict[float, bool] = {}
|
|
if isinstance(stations, list):
|
|
for station in stations:
|
|
suggested = station.get("pavement_suggested")
|
|
if isinstance(suggested, bool):
|
|
mapping[round(float(station.get("chainage_m", 0.0)), 3)] = suggested
|
|
return mapping
|
|
|
|
|
|
def _default_section_modes(longitudinal: dict[str, Any]) -> dict[float, str]:
|
|
"""측점별 기본 단면유형 매핑: 상단측(uphill_side)이 절토측이 되는 편절편성.
|
|
|
|
B05 solve가 자동 판정하고 사용자가 3D 램프로 바꾼 값(확정 시 정본 병합)을 그대로
|
|
소비한다. 판정 불가 측점은 매핑에서 빠지고 호출부가 좌절토로 폴백한다.
|
|
"""
|
|
stations = longitudinal.get("stations")
|
|
mapping: dict[float, str] = {}
|
|
if isinstance(stations, list):
|
|
for station in stations:
|
|
side = station.get("uphill_side")
|
|
if side in ("left", "right"):
|
|
mapping[round(float(station.get("chainage_m", 0.0)), 3)] = f"{side}_cut"
|
|
return mapping
|
|
|
|
|
|
def _attach_default_designs(
|
|
longitudinal: dict[str, Any], cross_sections: list[dict[str, Any]]
|
|
) -> None:
|
|
"""지정 설계가 없는 횡단에 기본값(리핑암 + 암반 경계 0.5m + 상단측 절토) 프리뷰를 얹는다.
|
|
|
|
detail 조회가 이미 읽어온 samples와 종단 계획선을 그대로 써서 추가 파일 I/O 없이
|
|
전 측점 프리뷰를 만든다(미저장). 계산 불가 측점은 건너뛴다.
|
|
|
|
기본 지반을 토사가 아니라 **리핑암 + 지표 아래 0.5m 암반 경계**로 두는 이유(2026-08-03
|
|
사용자 확정): 산지 절토는 대부분 표토 아래에서 암이 나오므로, 전량 토사 가정은 물량이
|
|
낙관적으로 나온다. 지표 0.5m까지 토사·그 아래 리핑암인 2단 단면이 안전한 출발값이고,
|
|
발파암이 있으면 사용자가 B06에서 측점별로 고친다.
|
|
"""
|
|
default_modes = _default_section_modes(longitudinal)
|
|
pavement = _pavement_suggestions(longitudinal)
|
|
for section in cross_sections:
|
|
if section.get("design"):
|
|
continue
|
|
try:
|
|
chainage_m = float(section.get("chainage_m", 0.0))
|
|
suggested = pavement.get(round(chainage_m, 3), False)
|
|
design = compute_cross_design(
|
|
section.get("samples", []),
|
|
design_elevation_from_longitudinal(longitudinal, chainage_m),
|
|
ground_type="ripping_rock",
|
|
section_mode=default_modes.get(round(chainage_m, 3), "left_cut"),
|
|
paved=suggested,
|
|
rock_boundary_offset_m=STANDARD_ROCK_BOUNDARY_DEFAULT_OFFSET_M,
|
|
)
|
|
design["status"] = "provisional"
|
|
design["pavement_suggested"] = suggested
|
|
section["design"] = design
|
|
except (ValueError, KeyError):
|
|
continue
|
|
|
|
|
|
#: 계획선 프리뷰 응답에 실을 설계 필드 — 유토곡선 계산(면적·지반유형·자연방토)과
|
|
#: 계획고 변경 감지에 쓰는 것만. 나머지(설계선 좌표 등)는 B06이 상세를 받을 때 온다.
|
|
_PREVIEW_DESIGN_FIELDS = (
|
|
"ground_type",
|
|
"roadbed_width_m",
|
|
"cut_area_m2",
|
|
"cut_soil_area_m2",
|
|
"cut_rock_area_m2",
|
|
"cut_rock_kind",
|
|
"fill_area_m2",
|
|
"fill_ground_slope",
|
|
"design_elevation_m",
|
|
)
|
|
|
|
|
|
def _recompute_designs_for_alignment(
|
|
longitudinal: dict[str, Any],
|
|
cross_sections: list[dict[str, Any]],
|
|
stored_designs: list[dict[str, Any]],
|
|
standard: dict[str, Any] | None,
|
|
rock_boundary_offsets: dict[str, float] | None = None,
|
|
) -> None:
|
|
"""새 계획선 기준으로 전 측점 횡단 설계를 다시 계산해 얹는다(메모리 프리뷰).
|
|
|
|
사용자가 이미 고른 지반유형·단면유형·측구·암 경계는 **그대로 유지**하고 계획고만
|
|
새 선형 값으로 바꿔 넣는다 — 계획선을 끌었다고 측점 선택이 초기화되면 안 된다.
|
|
지정이 없는 측점은 기본값(리핑암 + 암반 경계 0.5m + 상단측 절토)으로 계산한다.
|
|
`rock_boundary_offsets`(B06 세션값)가 오면 DB 저장분보다 우선한다.
|
|
"""
|
|
default_modes = _default_section_modes(longitudinal)
|
|
pavement = _pavement_suggestions(longitudinal)
|
|
stored_by_chainage = {
|
|
round(float(record["chainage_m"]), 3): (record.get("design") or {})
|
|
for record in stored_designs
|
|
}
|
|
# 세션 오프셋 키는 프론트 표기(소수 자릿수)가 제각각일 수 있어 수치로 정규화한다.
|
|
session_offsets: dict[float, float] = {}
|
|
for raw_key, offset in (rock_boundary_offsets or {}).items():
|
|
try:
|
|
session_offsets[round(float(raw_key), 3)] = float(offset)
|
|
except (TypeError, ValueError):
|
|
continue
|
|
for section in cross_sections:
|
|
chainage_m = float(section.get("chainage_m", 0.0))
|
|
key = round(chainage_m, 3)
|
|
stored = stored_by_chainage.get(key) or {}
|
|
suggested = pavement.get(key, False)
|
|
rock_offset = session_offsets.get(
|
|
key, stored.get("rock_boundary_offset_m", STANDARD_ROCK_BOUNDARY_DEFAULT_OFFSET_M)
|
|
)
|
|
try:
|
|
design = compute_cross_design(
|
|
section.get("samples", []),
|
|
design_elevation_from_longitudinal(longitudinal, chainage_m),
|
|
ground_type=str(stored.get("ground_type") or "ripping_rock"),
|
|
section_mode=str(stored.get("section_mode") or default_modes.get(key, "left_cut")),
|
|
ditch_side=stored.get("ditch_side"),
|
|
ditch_type=str(stored.get("ditch_type") or "standard"),
|
|
paved=bool(stored.get("paved", suggested)),
|
|
standard=standard,
|
|
rock_boundary_offset_m=rock_offset,
|
|
two_stage_slope=bool(stored.get("two_stage_slope", True)),
|
|
ditch_enabled=stored.get("ditch_enabled"),
|
|
)
|
|
except (ValueError, KeyError):
|
|
continue
|
|
design["status"] = "provisional"
|
|
design["pavement_suggested"] = suggested
|
|
# 표시 설정(측점 개별 반폭)은 계산과 무관 — 재계산이 지우면 안 된다(2026-08-06).
|
|
if stored.get("display_half_width_m") is not None:
|
|
design["display_half_width_m"] = stored["display_half_width_m"]
|
|
section["design"] = design
|
|
|
|
|
|
@router.post(
|
|
"/{project_id}/sections/{route_id}/cross-design/preview",
|
|
response_model=CrossDesignPreviewResponse,
|
|
)
|
|
async def preview_cross_designs(
|
|
project_id: UUID, route_id: int, request: CrossDesignPreviewRequest
|
|
) -> CrossDesignPreviewResponse | JSONResponse:
|
|
"""계획선 편집 델타로 계획선과 전 측점 횡단 설계를 다시 계산해 돌려준다(저장 없음).
|
|
|
|
B05에서 계획고를 끄는 동안 횡단 단면적·유토곡선이 함께 움직이게 하는 프리뷰 경로다.
|
|
영속화는 각 페이지의 임시저장·확정이 맡는다.
|
|
"""
|
|
pool = get_db_pool()
|
|
try:
|
|
async with pool.acquire() as connection:
|
|
longitudinal_row = await get_longitudinal_section(connection, project_id, route_id)
|
|
if not longitudinal_row:
|
|
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_row["longitudinal_file_path"]),
|
|
)
|
|
|
|
def rebuild() -> None:
|
|
alignment, profile = rebuild_alignment_profile(detail["longitudinal"], request.edits())
|
|
detail["longitudinal"]["profile_alignment"] = alignment
|
|
detail["longitudinal"]["design_profiles"] = [profile]
|
|
_recompute_designs_for_alignment(
|
|
detail["longitudinal"],
|
|
detail["cross_sections"],
|
|
designs,
|
|
request.standard_cross_section,
|
|
request.rock_boundary_offsets,
|
|
)
|
|
|
|
await asyncio.to_thread(rebuild)
|
|
# 계획고를 끄는 동안 오가는 값이라 기본은 **유토곡선이 실제로 읽는 필드만** 싣는다.
|
|
# 지반선 샘플·설계선 좌표(design_line)·선형 구조를 다 담으면 1MB가 넘어 편집이 굼떠진다
|
|
# (2026-08-03 사용자 지적). 계획선 자체는 화면이 이미 같은 규칙으로 계산해 들고 있다.
|
|
# `full_designs`(B06 진입 시 stale 일괄 재계산)일 때만 설계 전체를 1회 싣는다 —
|
|
# 측점별 순차 호출 N번보다 왕복 1번이 훨씬 싸다(2026-08-04 사용자 확인).
|
|
return CrossDesignPreviewResponse(
|
|
designs=[
|
|
{
|
|
"chainage_m": section.get("chainage_m"),
|
|
"design": section["design"]
|
|
if request.full_designs
|
|
else {
|
|
key: section["design"].get(key)
|
|
for key in _PREVIEW_DESIGN_FIELDS
|
|
if key in section["design"]
|
|
},
|
|
}
|
|
for section in detail["cross_sections"]
|
|
if section.get("design")
|
|
],
|
|
)
|
|
except FileNotFoundError as exc:
|
|
return JSONResponse(status_code=404, content={"status": "error", "message": str(exc)})
|
|
except (OSError, ValueError, json.JSONDecodeError) 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 _compute_default_designs(
|
|
project_root: Path,
|
|
longitudinal_file_path: str,
|
|
chainages: list[float],
|
|
standard: dict[str, Any] | None = None,
|
|
) -> list[tuple[float, dict[str, Any]]]:
|
|
"""미지정 측점들을 기본값(리핑암 + 암반 경계 0.5m/상단측 절토)으로 계산한 목록을 만든다.
|
|
|
|
standard가 오면(확정 요청의 패널 편집값) 그 값으로 표준단면 기하를 계산한다.
|
|
계획고 부재 등으로 계산 불가한 측점은 조용히 건너뛴다(확정을 막지 않기 위함).
|
|
"""
|
|
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"
|
|
default_modes = _default_section_modes(longitudinal)
|
|
pavement = _pavement_suggestions(longitudinal)
|
|
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
|
|
suggested = pavement.get(round(chainage_m, 3), False)
|
|
design = compute_cross_design(
|
|
samples,
|
|
design_elevation_from_longitudinal(longitudinal, chainage_m),
|
|
ground_type="ripping_rock",
|
|
section_mode=default_modes.get(round(chainage_m, 3), "left_cut"),
|
|
paved=suggested,
|
|
standard=standard,
|
|
rock_boundary_offset_m=STANDARD_ROCK_BOUNDARY_DEFAULT_OFFSET_M,
|
|
)
|
|
design["status"] = "provisional"
|
|
design["pavement_suggested"] = suggested
|
|
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, pavement_suggested = 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,
|
|
ditch_type=request.ditch_type,
|
|
paved=request.paved,
|
|
standard=request.standard_cross_section,
|
|
rock_boundary_offset_m=request.rock_boundary_offset_m,
|
|
two_stage_slope=request.two_stage_slope,
|
|
ditch_enabled=request.ditch_enabled,
|
|
)
|
|
# B06 지정 시점은 잠정치. B08 도면 확정 시 동일 엔진으로 재계산해 confirmed로 승격한다.
|
|
design["status"] = "provisional"
|
|
# 법정 근거 문구 표기용 — 사용자가 포장을 바꿔도 제안 여부는 그대로 남긴다.
|
|
design["pavement_suggested"] = pavement_suggested
|
|
async with pool.acquire() as connection:
|
|
# 표시 설정(측점 개별 반폭)은 계산 입력이 아니다 — 저장분에서 이월해 재계산이
|
|
# 지우지 않게 한다(2026-08-06).
|
|
stored_designs = await get_cross_section_designs(connection, route_id)
|
|
for record in stored_designs:
|
|
if abs(float(record["chainage_m"]) - request.chainage_m) < 0.01:
|
|
stored_design = record.get("design")
|
|
if (
|
|
isinstance(stored_design, dict)
|
|
and stored_design.get("display_half_width_m") is not None
|
|
):
|
|
design["display_half_width_m"] = stored_design["display_half_width_m"]
|
|
break
|
|
await connection.begin()
|
|
try:
|
|
updated = await update_cross_section_design(
|
|
connection,
|
|
route_id=route_id,
|
|
chainage_m=request.chainage_m,
|
|
design=design,
|
|
project_id=project_id,
|
|
)
|
|
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": "횡단 설계 계산 중 오류가 발생했습니다."},
|
|
)
|