eomsangdonandClaude Opus 5.5
4753d88983
- 분석본 = route_<연도>_<지점 열쇠>.json · GET 은 DB 지점 분석본 · 끈 지점 GET 은 그 분석 - [저장] 뒤 정본 = 끈 지점 분석본(DB 지점 따라감) - 쌓임 거르기 = 연도마다 최근 3 개 + 저장 지점 것 - 옛 꼴 route_<연도>.json 은 읽기만(공유 storage · 임시 배포 옛 판 서버가 읽음 — 옮기지 않음) - 시험 test_530_2 끈 지점 분석 · 저장 · 쌓임 Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
416 lines
18 KiB
Python
416 lines
18 KiB
Python
"""D 예정노선 2단계 분석 API (요구 345 · PLAN 530-1 · 530-2).
|
|
|
|
- `POST /api/projects/{id}/plan/terrain` — 지형 받기(뒤에서) · 끝나면 연도별 `surface_models` 행
|
|
- `POST …/plan/analyze` — 연도 구간마다 노선 추천 갈래 넷(뒤에서 · 파생 저장본)
|
|
- `GET …/plan/progress` — 두 일의 진행률
|
|
- `GET …/plan/analysis` — 저장본 + 고른 갈래 · 이은 누가 · 이은 중심선
|
|
- `PUT …/plan/analysis` — [저장] 연도별 고른 갈래(+ 꺾임점 고침) → `plan_route_years.route_json` ·
|
|
`fingerprint`
|
|
|
|
지점 정본은 `plan_route_years` 시점 · 종점 칸(529-3 몫). D03 화면에서 끈 지점(캐시 · PLAN 530-3)은
|
|
`points`(작업 좌표계 m)로 받아 그 지점으로 받기 · 분석 · 보기 · [저장] 때만 D02 규칙으로 정본에 씀.
|
|
계산 자리 = ③ 서버 단독 — [저장]은 노선을 받지 않고 갈래 이름만 받아 서버 저장본으로 다시 셈.
|
|
꺾임점 고침(543)은 마디 · 켬끔 · R 만 받아 서버가 선 · 지표를 다시 셈(`D03_PlanAnalysis_Edit`) ·
|
|
GET 은 분석 지문이 같은 저장 고침만 돌려줌.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import json
|
|
import logging
|
|
from pathlib import Path
|
|
from typing import Any
|
|
from uuid import UUID
|
|
|
|
import aiomysql
|
|
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_Edit as plan_edit
|
|
from D03_PlanAnalysis import D03_PlanAnalysis_Terrain as terrain
|
|
|
|
logger = logging.getLogger(__name__)
|
|
router = APIRouter(prefix="/api/projects", tags=["D03 Plan Analysis"])
|
|
|
|
# 진행 — (프로젝트, 일) → {running, percent, message, error} · 프로세스 메모리
|
|
_JOBS: dict[tuple[str, str], dict[str, Any]] = {}
|
|
_TASKS: set[asyncio.Task] = set()
|
|
KINDS = ("terrain", "analyze")
|
|
|
|
|
|
class PlanPoint(BaseModel):
|
|
"""끈 지점(작업 좌표계 m) — 연도 하나의 시점 · 종점."""
|
|
|
|
year: int
|
|
start: tuple[float, float]
|
|
end: tuple[float, float]
|
|
|
|
|
|
class PointsBody(BaseModel):
|
|
points: list[PlanPoint] | None = None
|
|
|
|
|
|
class PlanEditBody(BaseModel):
|
|
"""연도 하나의 꺾임점 고침 — 고친 갈래 · 그 분석 지문 · 마디(작업 좌표계 m) · 곡선 켬끔 · R."""
|
|
|
|
base_mode: str
|
|
base_fingerprint: str
|
|
nodes: list[tuple[float, float]]
|
|
curve_flags: list[bool]
|
|
radii: list[float | None]
|
|
|
|
|
|
class ChoiceRequest(PointsBody):
|
|
choices: dict[int, str] # 연도 → 갈래
|
|
edits: dict[int, PlanEditBody] | None = None # 연도 → 꺾임점 고침(없는 연도 = 엔진 노선)
|
|
|
|
|
|
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),
|
|
"saved": (seg["start"], seg["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:
|
|
await cursor.execute(
|
|
"SELECT road_type, design_speed_kph, route_start_m, route_end_m FROM projects"
|
|
" WHERE id = %s",
|
|
(str(project_id),),
|
|
)
|
|
project = await cursor.fetchone()
|
|
await cursor.execute(
|
|
"SELECT year, start_x, start_y, end_x, end_y, epsg, route_json, fingerprint "
|
|
"FROM plan_route_years WHERE project_id = %s ORDER BY year",
|
|
(str(project_id),),
|
|
)
|
|
rows = await cursor.fetchall()
|
|
if not stored or not project:
|
|
raise HTTPException(status_code=404, detail="프로젝트 저장 경로를 찾을 수 없습니다.")
|
|
if not rows:
|
|
raise HTTPException(
|
|
status_code=400, detail="연도별 지점이 없습니다 — 1단계에서 지정하세요."
|
|
)
|
|
for row in rows:
|
|
if any(row[k] is None for k in ("start_x", "start_y", "end_x", "end_y", "epsg")):
|
|
raise HTTPException(status_code=400, detail=f"{row['year']}년 지점이 비었습니다.")
|
|
if len({int(row["epsg"]) for row in rows}) != 1:
|
|
raise HTTPException(status_code=400, detail="연도마다 좌표계가 다릅니다.")
|
|
segments = [
|
|
{
|
|
"year": int(row["year"]),
|
|
"start": (float(row["start_x"]), float(row["start_y"])),
|
|
"end": (float(row["end_x"]), float(row["end_y"])),
|
|
"epsg": int(row["epsg"]),
|
|
"route_json": row["route_json"],
|
|
}
|
|
for row in rows
|
|
]
|
|
if points:
|
|
segments = _override(segments, points)
|
|
return Path(resolve_stored_project_path(stored)), project, segments
|
|
|
|
|
|
def _start_job(project_id: UUID, kind: str, work: Any) -> dict[str, Any]:
|
|
key = (str(project_id), kind)
|
|
if (_JOBS.get(key) or {}).get("running"):
|
|
raise HTTPException(status_code=409, detail="이미 돌고 있습니다.")
|
|
job = {"running": True, "percent": 0, "message": "시작", "error": None}
|
|
_JOBS[key] = job
|
|
|
|
def report(percent: int, message: str) -> None:
|
|
job.update(percent=int(percent), message=message)
|
|
|
|
async def run() -> None:
|
|
try:
|
|
await work(report)
|
|
job.update(percent=100)
|
|
except Exception as error: # noqa: BLE001 — 진행률에 까닭을 그대로 올림
|
|
logger.exception("D %s 실패: project_id=%s", kind, project_id)
|
|
job["error"] = str(getattr(error, "detail", None) or error)
|
|
finally:
|
|
job["running"] = False
|
|
|
|
task = asyncio.create_task(run(), name=f"D {kind} {project_id}")
|
|
_TASKS.add(task)
|
|
task.add_done_callback(_TASKS.discard)
|
|
return {"status": "accepted", **job}
|
|
|
|
|
|
async def _save_models(project_id: UUID, years: dict[int, list[dict[str, Any]]], epsg: int) -> None:
|
|
"""연도별 지형면 → `surface_models` 행(같은 파일 · 같은 지문이면 id 살림 · 다른 행 그대로)."""
|
|
from B04_PreProcess.B04_PreProcess_Repository import _existing_models, _upsert_surface_model
|
|
|
|
async with get_db_pool().acquire() as connection:
|
|
await connection.begin()
|
|
try:
|
|
existing = await _existing_models(connection, project_id)
|
|
for models in years.values():
|
|
for model in models:
|
|
await _upsert_surface_model(
|
|
connection,
|
|
existing,
|
|
project_id=project_id,
|
|
model=model,
|
|
source_file_id=None,
|
|
processed_cloud_id=None,
|
|
crs_epsg=epsg,
|
|
)
|
|
await connection.commit()
|
|
except Exception:
|
|
await connection.rollback()
|
|
raise
|
|
|
|
|
|
@router.post("/{project_id}/plan/terrain", response_model=None)
|
|
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, body.points if body else None)
|
|
epsg = segments[0]["epsg"]
|
|
|
|
async def work(report: Any) -> None:
|
|
got = await asyncio.to_thread(terrain.ensure_plan_terrain, root, segments, epsg, report)
|
|
await _save_models(project_id, got["years"], epsg)
|
|
report(100, f"지형 받기 끝 — 새 도엽 {len(got['sheets']['downloaded'])}매")
|
|
|
|
return _start_job(project_id, "terrain", work)
|
|
|
|
|
|
@router.post("/{project_id}/plan/analyze", response_model=None)
|
|
async def plan_analyze(project_id: UUID, body: PointsBody | None = None) -> dict[str, Any]:
|
|
"""연도 구간마다 노선 추천 갈래 넷(뒤에서 · 같은 입력이면 저장본)."""
|
|
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:
|
|
raise HTTPException(status_code=409, detail=f"지형을 먼저 받으세요: {missing}년")
|
|
|
|
async def work(report: Any) -> None:
|
|
await asyncio.to_thread(
|
|
lambda: analyze.analyze_plan(root, segments, criteria, on_progress=report)
|
|
)
|
|
|
|
return _start_job(project_id, "analyze", work)
|
|
|
|
|
|
@router.get("/{project_id}/plan/progress", response_model=None)
|
|
async def plan_progress(project_id: UUID) -> dict[str, Any]:
|
|
idle = {"running": False, "percent": 0, "message": None, "error": None}
|
|
return {
|
|
"status": "success",
|
|
**{kind: dict(_JOBS.get((str(project_id), kind)) or idle) for kind in KINDS},
|
|
}
|
|
|
|
|
|
def _saved_mode(route_json: Any) -> str | None:
|
|
try:
|
|
value = json.loads(route_json) if isinstance(route_json, str) else route_json
|
|
except ValueError:
|
|
return None
|
|
return value.get("mode") if isinstance(value, dict) else None
|
|
|
|
|
|
def _assemble(
|
|
root: Path,
|
|
project: dict[str, Any],
|
|
segments: list[dict[str, Any]],
|
|
choices: dict[int, Any],
|
|
edits: dict[int, dict[str, Any]] | None = None,
|
|
) -> tuple[list[dict[str, Any] | None], list[dict[str, Any]] | None, list[list[float]]]:
|
|
"""(연도 저장본 · 이은 누가 · 이은 중심선) — 저장본이 하나라도 없으면 None · 빈 선.
|
|
|
|
`edits` = 연도 → 고친 노선(`{mode, planned, metrics}`) — 누가 · 선만 고친 길이로(저장본은 엔진
|
|
그대로).
|
|
"""
|
|
results = [analyze.read_segment(root, seg) for seg in segments]
|
|
if any(result is None for result in results):
|
|
return results, None, []
|
|
view = analyze.with_edits(results, edits or {}) # type: ignore[arg-type]
|
|
rows = analyze.chain(view, choices, float(project["route_start_m"] or 0.0))
|
|
return results, rows, analyze.joined_line(view, rows)
|
|
|
|
|
|
def _saved_edits(
|
|
root: Path, segments: list[dict[str, Any]], choices: dict[int, Any]
|
|
) -> dict[int, dict[str, Any]]:
|
|
"""저장된 꺾임점 고침 — 저장 갈래 그대로 · 분석 지문이 지금과 같은 것만."""
|
|
out = {}
|
|
for seg in segments:
|
|
saved = plan_edit.saved_edit(seg["route_json"])
|
|
result = analyze.read_segment(root, seg)
|
|
if (
|
|
saved
|
|
and result
|
|
and saved["base_mode"] == choices.get(seg["year"])
|
|
and saved["base_fingerprint"] == result.get("fingerprint")
|
|
and saved.get("planned")
|
|
):
|
|
out[seg["year"]] = saved
|
|
return out
|
|
|
|
|
|
def _edit_view(saved: dict[str, Any]) -> dict[str, Any]:
|
|
return {key: saved[key] for key in (*plan_edit.INPUT_KEYS, "metrics")}
|
|
|
|
|
|
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, 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}
|
|
saved = await asyncio.to_thread(_saved_edits, root, segments, choices)
|
|
edits = {year: {"mode": e["base_mode"], **e} for year, e in saved.items()}
|
|
results, rows, line = await asyncio.to_thread(
|
|
_assemble, root, project, segments, choices, edits
|
|
)
|
|
return {
|
|
"status": "success",
|
|
"years": [
|
|
{
|
|
"year": seg["year"],
|
|
"start": seg["start"],
|
|
"end": seg["end"],
|
|
"analysis": result,
|
|
# 저장된 꺾임점 고침(입력 + 그때 지표) — 분석 지문이 같을 때만
|
|
"edit": _edit_view(saved[seg["year"]]) if seg["year"] in saved else None,
|
|
}
|
|
for seg, result in zip(segments, results)
|
|
],
|
|
"chain": rows,
|
|
"line": line,
|
|
"start_chainage_m": float(project["route_start_m"] or 0.0),
|
|
# 등록 목표 끝 누가(요구서 3-1) — 화면이 이은 끝 누가와 차이를 보임
|
|
"end_chainage_target_m": (
|
|
float(project["route_end_m"]) if project["route_end_m"] is not None else None
|
|
),
|
|
}
|
|
|
|
|
|
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,
|
|
session: dict[str, Any] = Depends(require_company),
|
|
) -> dict[str, Any]:
|
|
"""[저장] — 끈 지점(있으면 정본으로) · 연도별 고른 갈래를 서버 저장본으로 다시 이어 씀."""
|
|
if (request.points or request.edits) 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}")
|
|
choices = {seg["year"]: _saved_mode(seg["route_json"]) for seg in segments}
|
|
choices.update(request.choices)
|
|
results, rows, _ = await asyncio.to_thread(_assemble, root, project, segments, choices)
|
|
if rows is None:
|
|
raise HTTPException(
|
|
status_code=409, detail="분석 저장본이 없습니다 — [분석]을 먼저 하세요."
|
|
)
|
|
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}년")
|
|
edited: dict[int, dict[str, Any]] = {}
|
|
if request.edits:
|
|
saved = {seg["year"]: plan_edit.saved_edit(seg["route_json"]) for seg in segments}
|
|
asked = {year: body.model_dump() for year, body in request.edits.items()}
|
|
try:
|
|
edited = await asyncio.to_thread(
|
|
plan_edit.build_edits, root, results, rows, asked, saved
|
|
)
|
|
except plan_edit.EditError as error:
|
|
raise HTTPException(status_code=409, detail=str(error)) from error
|
|
_, rows, _ = await asyncio.to_thread(_assemble, root, project, segments, choices, edited)
|
|
if request.points:
|
|
await _store_points(project_id, segments)
|
|
async with get_db_pool().acquire() as connection:
|
|
await connection.begin()
|
|
try:
|
|
async with connection.cursor() as cursor:
|
|
for result, row in zip(results, rows):
|
|
route = edited.get(row["year"]) or result["routes"][row["mode"]]
|
|
body = {**row, **{k: route[k] for k in ("planned", "nodes", "curves")}}
|
|
if row["year"] in edited: # 꺾임점 고침 — 입력 · 지표 함께(GET 이 돌려줌)
|
|
body.update(metrics=route["metrics"], edit=route["edit"])
|
|
await cursor.execute(
|
|
"UPDATE plan_route_years SET route_json = %s, fingerprint = %s "
|
|
"WHERE project_id = %s AND year = %s",
|
|
(
|
|
json.dumps(body, ensure_ascii=False),
|
|
result["fingerprint"],
|
|
str(project_id),
|
|
row["year"],
|
|
),
|
|
)
|
|
await connection.commit()
|
|
except Exception:
|
|
await connection.rollback()
|
|
raise
|
|
return {"status": "success", "chain": rows}
|