eomsangdonandClaude Opus 5.5
6f2136f592
- 측점 있음 판정 폭 0.5m → 0.005m(측점 스냅 폭) — 0.1m 옮긴 관도 측점 없음으로 셈 - 비정규 측점 병합 때 사라진 측점의 base_pvi · 편집을 지움(0.006m 안 반올림은 옮김) · 계획선 다시 셈 · DB 등급 요약 갱신 - 시험 test_192_1(검증_랩탑메인 사본 재현) · missing_stations 판정 폭 Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01F3mdb91125mPoC6tpsQx4P
295 lines
13 KiB
Python
295 lines
13 KiB
Python
"""구조물 측점이 빠진 관·시설을 **알리고, 눌러서 만든다** (계획서 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,
|
|
update_longitudinal_grade_summary,
|
|
)
|
|
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) — 측점 생성의 스냅 폭과 같음
|
|
#: (`B05_Profile_Engine_Sections_Core` · 같은 횡단 이름이 되는 0.005m 안만 한 측점).
|
|
#: 넓으면 조금 옮긴 관(769.28 → 769.38)을 「측점 있음」으로 세어 측점이 옛 자리에 남음(192-1).
|
|
STATION_MATCH_TOLERANCE_M = 0.005
|
|
|
|
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)
|
|
params = await run_with_connection(get_surface_confirmation_params, str(project_id))
|
|
surface = (params or {}).get("source_filter"), (params or {}).get("method")
|
|
# 설계 지형이 바뀌었으면(B04 재확정 · 노선 [확인] 지형 바꿈 PLAN 133-2) ①은 옛 지형 — 버림.
|
|
if snapshot is not None and (
|
|
not all(surface)
|
|
or (snapshot["filter_key"], snapshot["method"], bool(snapshot.get("smooth")))
|
|
== (*surface, bool(params.get("smooth")))
|
|
):
|
|
return snapshot
|
|
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
|
|
|
|
|
|
async def _regenerate_structure_stations(
|
|
project_id: UUID, project_root: Path, route: dict[str, Any], snapshot: dict[str, Any]
|
|
) -> int:
|
|
"""구조물 측점 전체를 노선 확정 때와 같은 조건으로 다시 떠 종단 정본에 병합한다. 뜬 수.
|
|
|
|
⚠ 종단 병합이 비정규 측점을 **통째로 교체**하므로 빠진 것만이 아니라 전체를 뜬다.
|
|
"""
|
|
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,
|
|
)
|
|
|
|
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() -> tuple[int, dict[str, Any] | None]:
|
|
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,
|
|
)
|
|
summary = None
|
|
if stations and longitudinal:
|
|
summary = _merge_irregular_into_longitudinal(
|
|
project_root, str(longitudinal["longitudinal_file_path"]), stations
|
|
)
|
|
return len(stations), summary
|
|
|
|
made, summary = await asyncio.to_thread(_regenerate)
|
|
if summary is not None:
|
|
# 옮긴 관 자리의 계획선이 다시 셈됨 — DB 등급 요약도 같이(계획선 저장과 같은 길).
|
|
await run_with_connection(
|
|
lambda connection: update_longitudinal_grade_summary(
|
|
connection, route_id=route["id"], grade_summary=summary
|
|
)
|
|
)
|
|
return made
|
|
|
|
|
|
async def ensure_structure_stations(project_id: UUID) -> int:
|
|
"""빠진 구조물 측점을 세운다 — 횡단 저장이 **행 없는 측점의 수정**을 받을 때 부름.
|
|
|
|
옮긴 관 · 구조물 측점은 B05 확정 전이면 종단 측점 · 횡단 파일 · DB 행이 없어, 그 자리
|
|
횡단 수정이 받을 곳 없이 버려진다. [측점 만들기] 단추와 같은 조건으로 세운다.
|
|
만들 수 없으면(노선 · 샘플링 조건 없음 · 빠진 것 없음) 0 — 저장을 막지 않는다.
|
|
"""
|
|
paths = await _project_paths(project_id)
|
|
if paths is None:
|
|
return 0
|
|
project_root, route = paths
|
|
missing = await asyncio.to_thread(_missing_marks, project_root, str(route["route_data_path"]))
|
|
if not missing:
|
|
return 0
|
|
snapshot = await _sampling_conditions(project_id, project_root)
|
|
if snapshot is None:
|
|
logger.warning("B06 저장: 빠진 구조물 측점 %d곳 — 샘플링 조건이 없어 못 세움", len(missing))
|
|
return 0
|
|
await _regenerate_structure_stations(project_id, project_root, route, snapshot)
|
|
return len(missing)
|
|
|
|
|
|
@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:
|
|
"""빠진 구조물 측점을 **노선 확정 때와 같은 조건으로** 만들어 종단 정본에 병합한다.
|
|
|
|
⚠ 구조물 측점 전체를 다시 뜬다 — 종단 병합이 비정규 측점을 **통째로 교체**하므로
|
|
빠진 것만 넘기면 이미 있던 구조물 측점이 지워진다.
|
|
"""
|
|
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": []})
|
|
|
|
made = await _regenerate_structure_stations(project_id, project_root, route, snapshot)
|
|
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,
|
|
}
|
|
)
|