"""구조물 측점이 빠진 관·시설을 **알리고, 눌러서 만든다** (계획서 3-14 ㉯). 무엇이 문제였나 측점을 만드는 자리는 **B05 노선 [확정] 한 곳뿐**이다. 관을 저장하는 `PUT /drainage/pipe-points` 는 관 파일과 유역만 쓰고 측점을 다시 만들지 않는다. ⇒ **관을 나중에 놓거나 옮기면 그 측점이 안 생긴다.** 그 관은 횡단도에도 안 서고 수량·금액에서 통째로 빠지는데 **아무 말도 안 나온다**(실측: 배수관 넷). 왜 이 방식인가 (세 갈래 중 ㉯) ㉮ 관 저장 뒤 바로 만들기 — 그 엔드포인트에 노선·지표면 인자가 없어 끌어와야 함 ㉯ **알리고 [측점 만들기] 단추** — 누를 때만 돌아 비용이 적고 **왜 값이 없는지가 보임** ㉰ 그대로 두기 — 조용히 빠지는 것이 문제라 적어도 알림은 있어야 함 ⚠ **지어내지 않는 것** — 지표 샘플링 조건(어느 DTM·어느 방법)이 없으면 만들지 않는다. 조건이 다르면 그 측점만 다른 지표에서 뽑혀 **옆 측점과 지반고가 어긋난다.** 조건은 노선 확정 때 남긴 `B06_Section/sampling.json` 에서 읽고, 없으면 사유를 내고 막는다. """ from __future__ import annotations import asyncio import logging 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_Repository import get_latest_route, get_surface_crs_epsg from B05_Profile.B05_Profile_Router_Confirm import load_sampling_snapshot from common_util.common_util_storage import resolve_stored_project_path from config.config_db import run_with_connection logger = logging.getLogger(__name__) router = APIRouter(prefix="/api/projects", tags=["B06 Section"]) #: 「그 자리에 측점이 있다」고 볼 거리(m). #: ⚠⚠ **0.5m 다 — 0.05m 가 아니다**(2026-09-09 실측으로 뒤집힌 자리). #: 측점을 만들 때 **정수 미터가 같은 격자 측점이 있으면 그리로 스냅**한다 #: (`B05_Profile_Engine_Sections_Core` 의 파일명 가드). 그래서 관 440.241 은 측점 440.0 #: 위에 서고, 그 측점은 **구조물 이름표까지 달고 있다**(`structure`). #: 0.05m 로 보면 그런 자리를 「측점 없음」으로 잘못 세어 **있는 측점을 또 만들라고 한다.** #: 스냅 폭이 「정수 미터 반올림」이므로 최대 어긋남은 0.5m 다. STATION_MATCH_TOLERANCE_M = 0.5 SNAPSHOT_MISSING = ( "지표 샘플링 조건을 찾을 수 없어 측점을 만들 수 없음 — 1단계(지표 확정)를 마친 뒤" " 다시 눌러야 함. ⚠ 조건을 지어내면 그 측점만 다른 지표에서 뽑혀 옆 측점과 지반고가 어긋남" ) async def _sampling_conditions(project_id: UUID, project_root: Path) -> dict[str, Any] | None: """이 프로젝트가 쓰는 지표 샘플링 조건. 둘 다 **기록된 값**이고 지어내지 않는다. ① 노선 [확정] 때 남긴 `B06_Section/sampling.json` — **그때 실제로 쓴 조건**이라 1순위. ② 없으면 1단계(지표 확정) 저장값 — B06 화면 `context` 가 쓰는 그 값이라 같은 조건이다. 옛 프로젝트는 ①이 없으므로 이 길이 없으면 단추가 영영 안 돈다. """ from common_util.common_util_surface_confirmation import get_surface_confirmation_params snapshot = load_sampling_snapshot(project_root) if snapshot is not None: return snapshot params = await run_with_connection(get_surface_confirmation_params, str(project_id)) if not params or not params.get("source_filter") or not params.get("method"): return None return { "filter_key": params["source_filter"], "method": params["method"], "smooth": bool(params.get("smooth")), "surface_model_id": None, "source": "stage1", } def _missing_marks(project_root: Path, route_data_path: str) -> list[dict[str, Any]]: """구조물 측점 가운데 **종단 정본에 행이 없는 것**만. 없으면 빈 목록.""" from B05_Profile.B05_Profile_Engine_Sections import ( _load_pipe_points, _load_route_polyline, resolve_extra_stations, ) polyline = _load_route_polyline(project_root, route_data_path) pipes = _load_pipe_points(project_root, polyline) extras = resolve_extra_stations(project_root, pipes) if not extras: return [] existing = _station_chainages(project_root) missing = [] for chainage, label in extras: value = float(chainage) if any(abs(value - other) <= STATION_MATCH_TOLERANCE_M for other in existing): continue missing.append({"chainage_m": round(value, 3), "label": label}) return sorted(missing, key=lambda item: item["chainage_m"]) def _station_chainages(project_root: Path) -> list[float]: """종단 정본에 실제로 서 있는 측점 누가거리. 파일이 없으면 빈 목록.""" import json folder = project_root / "B06_Section" / "longitudinal" values: list[float] = [] for path in sorted(folder.glob("*.json")): try: data = json.loads(path.read_text(encoding="utf-8")) except (OSError, ValueError): continue for station in data.get("stations") or []: chainage = station.get("chainage_m") if isinstance(chainage, (int, float)): values.append(float(chainage)) return values async def _project_paths(project_id: UUID) -> tuple[Path, dict[str, Any]] | None: async def _load(connection): stored = await get_project_storage_relative_path(connection, project_id) route = await get_latest_route(connection, project_id) return stored, route stored, route = await run_with_connection(_load) if not stored or not route: return None return Path(resolve_stored_project_path(stored)), route @router.get("/{project_id}/section/missing-stations") async def get_missing_stations(project_id: UUID) -> JSONResponse: """측점이 없는 구조물 목록 — 화면이 「측점 없는 관 N개」를 띄우는 데 쓴다.""" try: paths = await _project_paths(project_id) if paths is None: return JSONResponse(content={"status": "success", "missing": [], "can_create": False}) project_root, route = paths missing = await asyncio.to_thread( _missing_marks, project_root, str(route["route_data_path"]) ) snapshot = await _sampling_conditions(project_id, project_root) except Exception: logger.exception("B06 측점 점검 실패: project_id=%s", project_id) return JSONResponse( status_code=500, content={"status": "error", "message": "측점을 점검하지 못했습니다."}, ) return JSONResponse( content={ "status": "success", "missing": missing, "can_create": bool(missing) and snapshot is not None, "reason": "" if snapshot is not None else SNAPSHOT_MISSING, } ) @router.post("/{project_id}/section/missing-stations") async def create_missing_stations(project_id: UUID) -> JSONResponse: """빠진 구조물 측점을 **노선 확정 때와 같은 조건으로** 만들어 종단 정본에 병합한다. ⚠ 구조물 측점 전체를 다시 뜬다 — 종단 병합이 비정규 측점을 **통째로 교체**하므로 빠진 것만 넘기면 이미 있던 구조물 측점이 지워진다. """ from B05_Profile.B05_Profile_Engine_Sections import ( _load_pipe_points, _load_route_polyline, generate_irregular_sections, resolve_extra_stations, ) from B05_Profile.B05_Profile_Router_Confirm import ( _merge_irregular_into_longitudinal, _section_options_from_stored, ) from B06_Section.B06_Section_Repository import ( get_latest_section_options, get_longitudinal_section, ) try: paths = await _project_paths(project_id) if paths is None: return JSONResponse( status_code=404, content={"status": "error", "message": "확정된 노선이 없습니다."}, ) project_root, route = paths snapshot = await _sampling_conditions(project_id, project_root) if snapshot is None: return JSONResponse( status_code=409, content={"status": "error", "message": SNAPSHOT_MISSING}, ) missing = await asyncio.to_thread( _missing_marks, project_root, str(route["route_data_path"]) ) if not missing: return JSONResponse(content={"status": "success", "created": 0, "missing": []}) async def _load(connection): options = await get_latest_section_options(connection, project_id) crs_epsg = await get_surface_crs_epsg( connection, project_id, snapshot.get("surface_model_id") ) longitudinal = await get_longitudinal_section(connection, project_id, route["id"]) return options, crs_epsg, longitudinal stored_options, crs_epsg, longitudinal = await run_with_connection(_load) def _regenerate() -> int: polyline = _load_route_polyline(project_root, str(route["route_data_path"])) pipes = _load_pipe_points(project_root, polyline) extras = resolve_extra_stations(project_root, pipes) stations = generate_irregular_sections( project_root, str(route["route_data_path"]), str(snapshot["filter_key"]), str(snapshot["method"]), bool(snapshot.get("smooth")), extra_stations=extras, options=_section_options_from_stored(stored_options), crs=f"EPSG:{crs_epsg}" if crs_epsg is not None else None, ) if stations and longitudinal: _merge_irregular_into_longitudinal( project_root, str(longitudinal["longitudinal_file_path"]), stations ) return len(stations) made = await asyncio.to_thread(_regenerate) except Exception: logger.exception("B06 측점 만들기 실패: project_id=%s", project_id) return JSONResponse( status_code=500, content={"status": "error", "message": "측점을 만들지 못했습니다."}, ) return JSONResponse( content={ "status": "success", # 새로 선 것만 세어 낸다 — 다시 뜬 총수(`made`)와 다르다. "created": len(missing), "regenerated": made, "missing": missing, } )