feat(b06): 횡단도 바로 반영 — 새로 놓거나 옮긴 관 · 구조물의 횡단 카드를 [저장] 전에 보임 (273-2)

- 서버: 읽기 전용 미리 뜨기 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
This commit is contained in:
2026-10-11 20:48:32 +09:00
co-authored by Claude Opus 5.5
parent 887c89bbf8
commit dfd7427c46
14 changed files with 642 additions and 105 deletions
+107 -13
View File
@@ -27,6 +27,7 @@ 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
@@ -165,6 +166,26 @@ async def _project_paths(project_id: UUID) -> tuple[Path, dict[str, Any]] | 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:
@@ -183,20 +204,8 @@ async def _regenerate_structure_stations(
_section_options_from_stored,
drop_rows_off_stations,
)
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)
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"]))
@@ -334,3 +343,88 @@ async def create_missing_stations(project_id: UUID) -> JSONResponse:
"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})