feat(530-3): D03 예정노선 분석 화면 — 평면 · 지점 끌기 · 더 받기 · 갈래 카드 · 측점 · 지반 횡단 · 3D
- 평면 = 위성 + 도엽 등고선 + 다른 갈래(옅게) + 고른 노선 + 측점(누가 = 등록 시작부터) + 지점 끌기 겹 - 끈 지점 = 캐시 · 받기 · 분석 · 보기는 points 로 실어 보냄 · [저장] 때만 D02 규칙으로 정본 - 연도별 갈래 넷 카드(연장 · 최대 기울기 · 위반 수) · 고름 = 캐시 · [저장] = 서버 저장본으로 다시 셈 - 측점 누름 = 하 패널 지반선 횡단(sectionsGround · buildCrossSvg 지반만 인자 · 기본 그대로) - 3D 모달 = 연도 DEM + 노선 + 측점 막대(코리도 · 구조물 안 부름) - 화면 계산 logic 짝(pickMode · chainRows · joinedLine ↔ 파이썬) 거울 시험 Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Claude Opus 5.5
@@ -6,8 +6,9 @@
|
||||
- `GET …/plan/analysis` — 저장본 + 고른 갈래 · 이은 누가 · 이은 중심선
|
||||
- `PUT …/plan/analysis` — [저장] 연도별 고른 갈래 → `plan_route_years.route_json` · `fingerprint`
|
||||
|
||||
지점 정본은 `plan_route_years` 시점 · 종점 칸(529-3 몫)을 읽기만 함. 계산 자리 = ③ 서버 단독 —
|
||||
[저장]은 브라우저 값을 받지 않고 갈래 이름만 받아 서버 저장본으로 다시 셈.
|
||||
지점 정본은 `plan_route_years` 시점 · 종점 칸(529-3 몫). D03 화면에서 끈 지점(캐시 · PLAN 530-3)은
|
||||
`points`(작업 좌표계 m)로 받아 그 지점으로 받기 · 분석 · 보기 · [저장] 때만 D02 규칙으로 정본에 씀.
|
||||
계산 자리 = ③ 서버 단독 — [저장]은 노선을 받지 않고 갈래 이름만 받아 서버 저장본으로 다시 셈.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
@@ -20,12 +21,15 @@ from typing import Any
|
||||
from uuid import UUID
|
||||
|
||||
import aiomysql
|
||||
from fastapi import APIRouter, HTTPException
|
||||
from fastapi import APIRouter, Depends, HTTPException
|
||||
from pydantic import BaseModel
|
||||
|
||||
from B03_FileInput.B03_FileInput_Repository import get_project_storage_relative_path
|
||||
from common_util.common_util_auth import is_forest_service, require_company
|
||||
from common_util.common_util_crs import tm_to_lonlat
|
||||
from common_util.common_util_storage import resolve_stored_project_path
|
||||
from config.config_db import get_db_pool
|
||||
from D02_PlanBase import D02_PlanBase_Router as base
|
||||
from D03_PlanAnalysis import D03_PlanAnalysis_Analyze as analyze
|
||||
from D03_PlanAnalysis import D03_PlanAnalysis_Terrain as terrain
|
||||
|
||||
@@ -38,12 +42,43 @@ _TASKS: set[asyncio.Task] = set()
|
||||
KINDS = ("terrain", "analyze")
|
||||
|
||||
|
||||
class ChoiceRequest(BaseModel):
|
||||
class PlanPoint(BaseModel):
|
||||
"""끈 지점(작업 좌표계 m) — 연도 하나의 시점 · 종점."""
|
||||
|
||||
year: int
|
||||
start: tuple[float, float]
|
||||
end: tuple[float, float]
|
||||
|
||||
|
||||
class PointsBody(BaseModel):
|
||||
points: list[PlanPoint] | None = None
|
||||
|
||||
|
||||
class ChoiceRequest(PointsBody):
|
||||
choices: dict[int, str] # 연도 → 갈래
|
||||
|
||||
|
||||
async def _plan_inputs(project_id: UUID) -> tuple[Path, dict[str, Any], list[dict[str, Any]]]:
|
||||
"""(프로젝트 실경로, 프로젝트 칸, 연도 구간) — 지점이 빈 연도 · 좌표계 섞임은 거절."""
|
||||
def _override(segments: list[dict[str, Any]], points: list[PlanPoint]) -> list[dict[str, Any]]:
|
||||
"""연도 구간 지점을 끈 지점으로 — 연도 같음 · 직전 종점 = 다음 시점(1 mm) 아니면 거절."""
|
||||
by_year = {p.year: p for p in points}
|
||||
if set(by_year) != {seg["year"] for seg in segments}:
|
||||
raise HTTPException(status_code=400, detail="끈 지점의 연도가 사업연도와 다릅니다.")
|
||||
out = [
|
||||
{**seg, "start": tuple(by_year[seg["year"]].start), "end": tuple(by_year[seg["year"]].end)}
|
||||
for seg in segments
|
||||
]
|
||||
for prev, seg in zip(out, out[1:]):
|
||||
if max(abs(a - b) for a, b in zip(prev["end"], seg["start"])) > base.SAME_M:
|
||||
raise HTTPException(
|
||||
status_code=400, detail=f"{seg['year']}년 시점이 직전 종점과 다릅니다."
|
||||
)
|
||||
return out
|
||||
|
||||
|
||||
async def _plan_inputs(
|
||||
project_id: UUID, points: list[PlanPoint] | None = None
|
||||
) -> tuple[Path, dict[str, Any], list[dict[str, Any]]]:
|
||||
"""(실경로, 프로젝트 칸, 연도 구간) — 빈 연도 · 좌표계 섞임 거절 · `points` = 끈 지점."""
|
||||
async with get_db_pool().acquire() as connection:
|
||||
stored = await get_project_storage_relative_path(connection, project_id)
|
||||
async with connection.cursor(aiomysql.DictCursor) as cursor:
|
||||
@@ -79,6 +114,8 @@ async def _plan_inputs(project_id: UUID) -> tuple[Path, dict[str, Any], list[dic
|
||||
}
|
||||
for row in rows
|
||||
]
|
||||
if points:
|
||||
segments = _override(segments, points)
|
||||
return Path(resolve_stored_project_path(stored)), project, segments
|
||||
|
||||
|
||||
@@ -134,9 +171,9 @@ async def _save_models(project_id: UUID, years: dict[int, list[dict[str, Any]]],
|
||||
|
||||
|
||||
@router.post("/{project_id}/plan/terrain", response_model=None)
|
||||
async def plan_terrain(project_id: UUID) -> dict[str, Any]:
|
||||
async def plan_terrain(project_id: UUID, body: PointsBody | None = None) -> dict[str, Any]:
|
||||
"""지점 기준 수치지형도 · 위성 · 연도별 DEM 받기(뒤에서) — 진행은 `GET …/plan/progress`."""
|
||||
root, _, segments = await _plan_inputs(project_id)
|
||||
root, _, segments = await _plan_inputs(project_id, body.points if body else None)
|
||||
epsg = segments[0]["epsg"]
|
||||
|
||||
async def work(report: Any) -> None:
|
||||
@@ -148,9 +185,9 @@ async def plan_terrain(project_id: UUID) -> dict[str, Any]:
|
||||
|
||||
|
||||
@router.post("/{project_id}/plan/analyze", response_model=None)
|
||||
async def plan_analyze(project_id: UUID) -> dict[str, Any]:
|
||||
async def plan_analyze(project_id: UUID, body: PointsBody | None = None) -> dict[str, Any]:
|
||||
"""연도 구간마다 노선 추천 갈래 넷(뒤에서 · 같은 입력이면 저장본)."""
|
||||
root, project, segments = await _plan_inputs(project_id)
|
||||
root, project, segments = await _plan_inputs(project_id, body.points if body else None)
|
||||
criteria = analyze.plan_criteria(project["road_type"], project["design_speed_kph"])
|
||||
missing = [s["year"] for s in segments if not terrain.models_dir_of(root, s["year"]).is_dir()]
|
||||
if missing:
|
||||
@@ -192,10 +229,22 @@ def _assemble(
|
||||
return results, rows, analyze.joined_line(results, rows)
|
||||
|
||||
|
||||
def _query_points(points: str | None) -> list[PlanPoint] | None:
|
||||
if not points:
|
||||
return None
|
||||
try:
|
||||
return [PlanPoint(**item) for item in json.loads(points)]
|
||||
except (ValueError, TypeError) as error:
|
||||
raise HTTPException(status_code=400, detail="points 꼴이 다릅니다.") from error
|
||||
|
||||
|
||||
@router.get("/{project_id}/plan/analysis", response_model=None)
|
||||
async def plan_analysis(project_id: UUID) -> dict[str, Any]:
|
||||
"""분석 저장본(지점이 지금과 같은 것만) + 고른 갈래(저장값 · 없으면 첫 성공 갈래)."""
|
||||
root, project, segments = await _plan_inputs(project_id)
|
||||
async def plan_analysis(project_id: UUID, points: str | None = None) -> dict[str, Any]:
|
||||
"""분석 저장본(지점이 지금과 같은 것만) + 고른 갈래(저장값 · 없으면 첫 성공 갈래).
|
||||
|
||||
`points` = 끈 지점 JSON(`[{year, start, end}]` · 작업 좌표계 m) — 그 지점의 저장본을 봄.
|
||||
"""
|
||||
root, project, segments = await _plan_inputs(project_id, _query_points(points))
|
||||
choices = {seg["year"]: _saved_mode(seg["route_json"]) for seg in segments}
|
||||
results, rows, line = await asyncio.to_thread(_assemble, root, project, segments, choices)
|
||||
return {
|
||||
@@ -206,13 +255,46 @@ async def plan_analysis(project_id: UUID) -> dict[str, Any]:
|
||||
],
|
||||
"chain": rows,
|
||||
"line": line,
|
||||
"start_chainage_m": float(project["route_start_m"] or 0.0),
|
||||
}
|
||||
|
||||
|
||||
async def _store_points(project_id: UUID, segments: list[dict[str, Any]]) -> None:
|
||||
"""끈 지점 → D02 정본 규칙(경위도 · 잇기 · 바뀐 연도만 · 그 연도 노선 비움)."""
|
||||
epsg = segments[0]["epsg"]
|
||||
|
||||
def lonlat(xy: tuple[float, float]) -> base.LonLat:
|
||||
lon, lat = tm_to_lonlat(xy[0], xy[1], epsg)
|
||||
return base.LonLat(lon=lon, lat=lat)
|
||||
|
||||
project, old = await base._load(project_id)
|
||||
start = int(project["project_year"])
|
||||
end = int(project["project_year_end"] or start)
|
||||
items = [
|
||||
base.YearPoints(year=s["year"], start=lonlat(s["start"]), end=lonlat(s["end"]))
|
||||
for s in segments
|
||||
]
|
||||
new_epsg, rows = base.with_tm(base.chain_years(start, end, items))
|
||||
if new_epsg != epsg:
|
||||
raise HTTPException(
|
||||
status_code=409,
|
||||
detail="끈 지점이 다른 좌표계 띠로 넘어갔습니다 — 1단계에서 지정하세요.",
|
||||
)
|
||||
writes = [row for row in rows if base.changed(row, old.get(row["year"]), epsg)]
|
||||
if writes:
|
||||
await base._store(project_id, epsg, writes, [])
|
||||
|
||||
|
||||
@router.put("/{project_id}/plan/analysis", response_model=None)
|
||||
async def save_plan_analysis(project_id: UUID, request: ChoiceRequest) -> dict[str, Any]:
|
||||
"""[저장] — 연도별 고른 갈래를 서버 저장본으로 다시 이어 `plan_route_years` 에 씀."""
|
||||
root, project, segments = await _plan_inputs(project_id)
|
||||
async def save_plan_analysis(
|
||||
project_id: UUID,
|
||||
request: ChoiceRequest,
|
||||
session: dict[str, Any] = Depends(require_company),
|
||||
) -> dict[str, Any]:
|
||||
"""[저장] — 끈 지점(있으면 정본으로) · 연도별 고른 갈래를 서버 저장본으로 다시 이어 씀."""
|
||||
if request.points and not is_forest_service(session):
|
||||
raise HTTPException(status_code=403, detail="예정노선 지점은 산림청만 고칠 수 있습니다.")
|
||||
root, project, segments = await _plan_inputs(project_id, request.points)
|
||||
bad = [m for m in request.choices.values() if m not in analyze.PLAN_MODES]
|
||||
if bad:
|
||||
raise HTTPException(status_code=400, detail=f"모르는 갈래: {bad}")
|
||||
@@ -226,6 +308,8 @@ async def save_plan_analysis(project_id: UUID, request: ChoiceRequest) -> dict[s
|
||||
lost = [r["year"] for r in rows if choices.get(r["year"]) not in (None, r["mode"])]
|
||||
if lost:
|
||||
raise HTTPException(status_code=409, detail=f"고른 갈래에 노선이 없습니다: {lost}년")
|
||||
if request.points:
|
||||
await _store_points(project_id, segments)
|
||||
async with get_db_pool().acquire() as connection:
|
||||
await connection.begin()
|
||||
try:
|
||||
|
||||
Reference in New Issue
Block a user