eomsangdonandClaude Opus 5.5
f17db59ab6
- 지형 받기 — 연도 구간마다 models_<연도> 대신 한 벌 models_plan · 범위 = 전체 시점 · 종점(+ 그은 점) 상자 + 여유 ∪ 받은 범위(줄지 않음) · 받은 범위 plan_terrain.json
- 도엽 = 범위 상자에 닿는 도엽 모두(3×3 이웃 대신) · 위성 = 그 도엽 도곽
- 새 GET …/plan/terrain/extent — 상자 네 귀 경위도 · 도엽 도곽 · 칸 수 어림 · 받은 범위 · covered(더 받기 판정) · 내려받기 없음
- POST …/plan/terrain 몸 = {margin_m, extra} · 여유 처음 값 350 m(옛 DEM 폭) · 바닥 300 m · 실패 글에 병합 도엽 매수
- 전체 시점 · 종점 = 임시로 첫 연도 시점 · 끝 연도 종점(_plan_ends · 560-2 받은 뒤 바꿈)
- 시험 530_1 다시 씀 · 530_2_plan_router 범위 · 받기 둘 더함 · 530_2_plan_analyze 옛 연도 DEM 기대 한 줄 고침(한 벌로 바뀜)
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
453 lines
20 KiB
Python
453 lines
20 KiB
Python
"""D 예정노선 2단계 분석 API (요구 345 · PLAN 530-1 · 530-2).
|
||
|
||
- `GET /api/projects/{id}/plan/terrain/extent` — 받기 전 범위(상자 · 도엽 · 칸 수 · 받은 범위)
|
||
- `POST …/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 TerrainBody(BaseModel):
|
||
"""지형 받기 — 여유(m · 빈 값 = 처음 값) · 더 덮을 점(그은 꺾임점 · 작업 좌표계 m)."""
|
||
|
||
margin_m: float | None = None
|
||
extra: list[tuple[float, float]] = []
|
||
|
||
|
||
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
|
||
|
||
|
||
async def _plan_ends(project_id: UUID) -> tuple[Path, int, list[tuple[float, float]]]:
|
||
"""(실경로, 좌표계, [전체 시점, 전체 종점]) — 지금은 첫 연도 시점 · 끝 연도 종점."""
|
||
root, _, segments = await _plan_inputs(project_id)
|
||
return root, segments[0]["epsg"], [segments[0]["start"], segments[-1]["end"]]
|
||
|
||
|
||
def _query_xy(extra: str | None) -> list[tuple[float, float]]:
|
||
if not extra:
|
||
return []
|
||
try:
|
||
return [(float(x), float(y)) for x, y in json.loads(extra)]
|
||
except (ValueError, TypeError) as error:
|
||
raise HTTPException(status_code=400, detail="extra 꼴이 다릅니다.") from error
|
||
|
||
|
||
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, models: 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 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.get("/{project_id}/plan/terrain/extent", response_model=None)
|
||
async def plan_terrain_extent(
|
||
project_id: UUID, margin_m: float | None = None, extra: str | None = None
|
||
) -> dict[str, Any]:
|
||
"""받기 전 범위 — 전체 시점 · 종점(+ `extra` 그은 점 `[[x, y]]`) 상자 + 여유 ∪ 받은 범위."""
|
||
root, epsg, ends = await _plan_ends(project_id)
|
||
view = await asyncio.to_thread(
|
||
terrain.plan_extent, root, [*ends, *_query_xy(extra)], epsg, margin_m
|
||
)
|
||
return {"status": "success", **view}
|
||
|
||
|
||
@router.post("/{project_id}/plan/terrain", response_model=None)
|
||
async def plan_terrain(project_id: UUID, body: TerrainBody | None = None) -> dict[str, Any]:
|
||
"""사업 전체 수치지형도 · 위성 · DEM 받기(뒤에서) — 진행은 `GET …/plan/progress`."""
|
||
body = body or TerrainBody()
|
||
root, epsg, ends = await _plan_ends(project_id)
|
||
points = [*ends, *body.extra]
|
||
|
||
async def work(report: Any) -> None:
|
||
got = await asyncio.to_thread(
|
||
terrain.ensure_plan_terrain, root, points, epsg, body.margin_m, report
|
||
)
|
||
await _save_models(project_id, got["models"], 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}
|