eomsangdonandClaude Opus 5.5
dfd7427c46
- 서버: 읽기 전용 미리 뜨기 POST /section/preview-stations — 로그인 · 회사 · 프로젝트 권한 그대로 · 한 번에 60곳 · 파일 · DB 안 씀 · 저장과 같은 길(sample_structure_sections · compute_default_designs crosses) - 화면: 초안이 심는 자리 중 횡단 없는 곳만 묻고 한 번 뜬 자리는 기억(B06_Section_Section_Drafts) · 카드 이름 · 관 그림 얹음 · 늦은 옛 응답 무시 · 배수 재계산이 관경을 정하면 카드 다시 얹음 · B06 관 옮김도 카드 다시 - [저장]은 지금처럼 서버가 같은 계산으로 정본에 씀 — 미리 본 카드와 같은 값 - 알림 글: 횡단도는 바로 보이고 세부 배수유역만 저장 뒤 다시 셈 - 시험: test_273_2 새로 · 177-3 헬퍼에 revealPick 자리 채움 Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01LLCk2y5xE8YGVgjbTntWsY
431 lines
19 KiB
Python
431 lines
19 KiB
Python
"""구조물 측점이 빠진 관·시설을 **알리고, 눌러서 만든다** (계획서 3-14 ㉯).
|
|
|
|
무엇이 문제였나
|
|
측점을 만드는 자리는 **B05 노선 [확정] 한 곳뿐**이다. 관을 저장하는
|
|
`PUT /drainage/pipe-points` 는 관 파일과 유역만 쓰고 측점을 다시 만들지 않는다.
|
|
⇒ **관을 나중에 놓거나 옮기면 그 측점이 안 생긴다.** 그 관은 횡단도에도 안 서고
|
|
수량·금액에서 통째로 빠지는데 **아무 말도 안 나온다**(실측: 배수관 넷).
|
|
|
|
왜 이 방식인가 (세 갈래 중 ㉯)
|
|
㉮ 관 저장 뒤 바로 만들기 — 그 엔드포인트에 노선·지표면 인자가 없어 끌어와야 함
|
|
㉯ **알리고 [측점 만들기] 단추** — 누를 때만 돌아 비용이 적고 **왜 값이 없는지가 보임**
|
|
㉰ 그대로 두기 — 조용히 빠지는 것이 문제라 적어도 알림은 있어야 함
|
|
|
|
⚠ **지어내지 않는 것** — 지표 샘플링 조건(어느 DTM·어느 방법)이 없으면 만들지 않는다.
|
|
조건이 다르면 그 측점만 다른 지표에서 뽑혀 **옆 측점과 지반고가 어긋난다.** 조건은
|
|
노선 확정 때 남긴 `B06_Section/sampling.json` 에서 읽고, 없으면 사유를 내고 막는다.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import logging
|
|
import math
|
|
from pathlib import Path
|
|
from typing import Any
|
|
from uuid import UUID
|
|
|
|
from fastapi import APIRouter
|
|
from fastapi.responses import JSONResponse
|
|
from pydantic import BaseModel, ConfigDict, Field
|
|
|
|
from B03_FileInput.B03_FileInput_Repository import get_project_storage_relative_path
|
|
from B05_Profile.B05_Profile_Engine_Grade_Alignment import SAME_STATION_SNAP_M
|
|
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) — 같은 측점 폭(213-1 · 반올림 차이 0.006m) 한 벌(219-1).
|
|
#: 넓으면 조금 옮긴 관(769.28 → 769.38)을 「측점 있음」으로 세어 측점이 옛 자리에 남음(192-1).
|
|
STATION_MATCH_TOLERANCE_M = SAME_STATION_SNAP_M
|
|
|
|
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 _planted(project_root: Path, route_data_path: str) -> list[tuple[float, str]]:
|
|
"""관 · 구조물 정본이 심는 구조물 측점 가운데 **노선 안**(세울 수 있는 것)만."""
|
|
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)
|
|
extras = resolve_extra_stations(project_root, _load_pipe_points(project_root, polyline))
|
|
# 노선 밖(0 미만 · 끝 넘음) = 측점을 못 세움 — 판정에서 뺌(219-1 · 길이 모르면 끝은 안 봄)
|
|
end_m = sum(math.dist(a[:2], b[:2]) for a, b in zip(polyline, polyline[1:]))
|
|
return [
|
|
(float(chainage), label)
|
|
for chainage, label in extras
|
|
if float(chainage) >= -STATION_MATCH_TOLERANCE_M
|
|
and (end_m <= 0 or float(chainage) <= end_m + STATION_MATCH_TOLERANCE_M)
|
|
]
|
|
|
|
|
|
def _missing_marks(project_root: Path, route_data_path: str) -> list[dict[str, Any]]:
|
|
"""구조물 측점 가운데 **종단 정본에 행이 없는 것**만. 없으면 빈 목록."""
|
|
planted = _planted(project_root, route_data_path)
|
|
if not planted:
|
|
return []
|
|
existing = _station_chainages(project_root)
|
|
missing = [
|
|
{"chainage_m": round(value, 3), "label": label}
|
|
for value, label in planted
|
|
if all(abs(value - other) > STATION_MATCH_TOLERANCE_M for other in existing)
|
|
]
|
|
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
|
|
|
|
|
|
def _stale_marks(project_root: Path, route_data_path: str) -> int:
|
|
"""종단 정본의 구조물 측점과 **지금 관 · 구조물**이 어긋난 수 — 옮긴 관의 옛 자리 · 구조물이
|
|
떠난 격자 자리 · 구조물이 새로 얹혀야 할 격자 자리(272). 0 이면 맞음."""
|
|
import json
|
|
|
|
planted = {round(value, 2) for value, _label in _planted(project_root, route_data_path)}
|
|
standing: set[float] = set()
|
|
for path in sorted((project_root / "B06_Section" / "longitudinal").glob("*.json")):
|
|
try:
|
|
data = json.loads(path.read_text(encoding="utf-8"))
|
|
except (OSError, ValueError):
|
|
continue
|
|
for station in data.get("stations") or []:
|
|
if station.get("kind") == "irregular" or station.get("structure"):
|
|
standing.add(round(float(station.get("chainage_m") or 0.0), 2))
|
|
return len(standing ^ planted)
|
|
|
|
|
|
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 _generation_inputs(
|
|
project_id: UUID, route: dict[str, Any], snapshot: dict[str, Any]
|
|
) -> tuple[Any, int | None, dict[str, Any] | None]:
|
|
"""측점을 뜰 때 쓰는 저장분 — (횡단 옵션, 좌표계 EPSG, 종단 행) · 만들기 · 미리 뜨기 한 벌."""
|
|
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
|
|
|
|
return await run_with_connection(_load)
|
|
|
|
|
|
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,
|
|
drop_rows_off_stations,
|
|
)
|
|
|
|
stored_options, crs_epsg, longitudinal = await _generation_inputs(project_id, route, snapshot)
|
|
|
|
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 longitudinal: # 빈 목록이어도 맞춤 — 옛 구조물 측점을 지움(272)
|
|
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 longitudinal:
|
|
await run_with_connection(
|
|
drop_rows_off_stations,
|
|
route["id"],
|
|
project_root,
|
|
str(longitudinal["longitudinal_file_path"]),
|
|
)
|
|
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"]))
|
|
# 빠진 것만이 아니라 낡은 것(옮긴 관 옛 자리 · 떠난 격자 자리)도 맞춤 — 두 벌로 남지 않게(272)
|
|
stale = await asyncio.to_thread(_stale_marks, project_root, str(route["route_data_path"]))
|
|
if not missing and not stale:
|
|
return 0
|
|
snapshot = await _sampling_conditions(project_id, project_root)
|
|
if snapshot is None:
|
|
logger.warning(
|
|
"B06 저장: 구조물 측점 %d곳을 못 맞춤 — 샘플링 조건 없음", len(missing) or stale
|
|
)
|
|
return 0
|
|
await _regenerate_structure_stations(project_id, project_root, route, snapshot)
|
|
return len(missing) or stale
|
|
|
|
|
|
@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,
|
|
}
|
|
)
|
|
|
|
|
|
#: 한 번에 미리 뜨는 자리 수 상한 — 화면 초안 한 벌이면 충분 · 큰 요청으로 서버를 붙잡지 않게.
|
|
PREVIEW_MAX_STATIONS = 60
|
|
|
|
|
|
class PreviewStationsRequest(BaseModel):
|
|
"""미리 뜰 자리(누가거리 m) — 노선 밖 · 숫자 아닌 값은 서버가 버림."""
|
|
|
|
model_config = ConfigDict(extra="forbid")
|
|
|
|
chainages: list[float] = Field(default_factory=list, max_length=PREVIEW_MAX_STATIONS)
|
|
|
|
|
|
@router.post("/{project_id}/section/preview-stations")
|
|
async def preview_structure_sections(
|
|
project_id: UUID, request: PreviewStationsRequest
|
|
) -> JSONResponse:
|
|
"""초안 자리(놓거나 옮긴 관 · 구조물)의 횡단을 **미리 떠서 돌려만 준다**(273-2 · 요구 168).
|
|
|
|
파일 · DB 에 아무것도 안 쓴다 — 화면이 캐시에 얹어 B06 카드를 [저장] 전에 보이고, [저장] 때
|
|
서버가 같은 길(`sample_structure_sections` · `compute_default_designs`)로 다시 만들어 정본에 씀.
|
|
샘플링 조건은 측점 만들기와 같다(없으면 지어내지 않고 409).
|
|
"""
|
|
from B05_Profile.B05_Profile_Engine_Sections import sample_structure_sections
|
|
from B05_Profile.B05_Profile_Router_Confirm import _section_options_from_stored
|
|
from B06_Section.B06_Section_Router_Design import compute_default_designs
|
|
|
|
chainages = sorted({round(c, 3) for c in request.chainages if math.isfinite(c) and c >= 0})
|
|
if not chainages:
|
|
return JSONResponse(content={"status": "success", "cross_sections": []})
|
|
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}
|
|
)
|
|
stored_options, crs_epsg, longitudinal = await _generation_inputs(
|
|
project_id, route, snapshot
|
|
)
|
|
if not longitudinal:
|
|
return JSONResponse(content={"status": "success", "cross_sections": []})
|
|
|
|
def _preview() -> list[dict[str, Any]]:
|
|
_stations, crosses = sample_structure_sections(
|
|
project_root,
|
|
str(route["route_data_path"]),
|
|
str(snapshot["filter_key"]),
|
|
str(snapshot["method"]),
|
|
bool(snapshot.get("smooth")),
|
|
extra_stations=tuple((chainage, "") for chainage in chainages),
|
|
options=_section_options_from_stored(stored_options),
|
|
crs=f"EPSG:{crs_epsg}" if crs_epsg is not None else None,
|
|
)
|
|
wanted = set(chainages)
|
|
picked = {
|
|
round(float(cross["chainage_m"]), 3): cross
|
|
for cross in crosses
|
|
if round(float(cross["chainage_m"]), 3) in wanted
|
|
}
|
|
designs = compute_default_designs(
|
|
project_root,
|
|
str(longitudinal["longitudinal_file_path"]),
|
|
list(picked),
|
|
crosses=picked,
|
|
)
|
|
for chainage, design in designs:
|
|
picked[round(chainage, 3)]["design"] = design
|
|
return list(picked.values())
|
|
|
|
sections = await asyncio.to_thread(_preview)
|
|
except Exception:
|
|
logger.exception("B06 미리 뜨기 실패: project_id=%s", project_id)
|
|
return JSONResponse(
|
|
status_code=500,
|
|
content={"status": "error", "message": "횡단을 미리 뜨지 못했습니다."},
|
|
)
|
|
return JSONResponse(content={"status": "success", "cross_sections": sections})
|