feat(529-3): D02 지점 정본 API · 경위도 ↔ TM 점 변환
- common_util_crs: 경도로 TM 띠(5185 · 5186 · 5187) · lonlat_to_tm · tm_to_lonlat
- D02_PlanBase_Router: GET · PUT /api/projects/{id}/plan/points — 잇기(직전 종점 = 다음 시점) · 범위 밖 · 빈 연도 거절 · 산림청만 고침 · 바뀐 연도만 분석 노선 비움
- 지킴 시험 test_529_3_plan_points
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Claude Opus 5.5
@@ -0,0 +1,206 @@
|
||||
"""D 예정노선 1단계 기준 지정 — 연도별 시점 · 종점 정본 API (요구 345 · 요구서 3-2 · PLAN 529-3).
|
||||
|
||||
- `GET /api/projects/{id}/plan/points` — 사업연도 연도마다 시점 · 종점(경위도 + TM) · 빈 = null
|
||||
- `PUT …/plan/points` — 경위도를 받아 서버가 TM 을 셈 → `plan_route_years` 행
|
||||
|
||||
규칙: 직전 연도 종점 = 다음 연도 시점(서버가 맞춤 · 보낸 시점은 첫 연도만 씀) ·
|
||||
범위 밖 · 빈 연도 거절 · 좌표계 = 모든 지점 평균 경도의 TM 띠 한 벌 ·
|
||||
지점이 바뀐 연도는 분석 노선(route_json · fingerprint · confirmed_at) 비움 ·
|
||||
고침 = 산림청만(지정 회사 GET 은 공용 판정 `require_project_access` 몫).
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Any
|
||||
from uuid import UUID
|
||||
|
||||
import aiomysql
|
||||
from fastapi import APIRouter, Depends, HTTPException
|
||||
from pydantic import BaseModel, Field
|
||||
|
||||
from common_util.common_util_auth import is_forest_service, require_company
|
||||
from common_util.common_util_crs import lonlat_to_tm, tm_epsg_for_lon, tm_to_lonlat
|
||||
from config.config_db import get_db_pool
|
||||
|
||||
router = APIRouter(prefix="/api/projects", tags=["D02 Plan Base"])
|
||||
|
||||
SAME_M = 0.001 # 1 mm 안이면 같은 지점 — 되저장으로 분석 노선을 지우지 않음
|
||||
|
||||
|
||||
class LonLat(BaseModel):
|
||||
lon: float = Field(..., ge=120, le=135)
|
||||
lat: float = Field(..., ge=30, le=45)
|
||||
|
||||
|
||||
class YearPoints(BaseModel):
|
||||
year: int
|
||||
start: LonLat | None = None
|
||||
end: LonLat | None = None
|
||||
|
||||
|
||||
class PointsRequest(BaseModel):
|
||||
years: list[YearPoints]
|
||||
|
||||
|
||||
def chain_years(year_start: int, year_end: int, items: list[YearPoints]) -> list[dict[str, Any]]:
|
||||
"""연도 범위를 잇기 — [{year, start, end}] 경위도 · 시점 = 직전 종점(첫 연도만 보낸 값)."""
|
||||
by_year: dict[int, YearPoints] = {}
|
||||
for item in items:
|
||||
if not year_start <= item.year <= year_end:
|
||||
raise HTTPException(
|
||||
status_code=400,
|
||||
detail=f"{item.year}년은 사업연도({year_start}~{year_end}) 밖입니다.",
|
||||
)
|
||||
if item.year in by_year:
|
||||
raise HTTPException(status_code=400, detail=f"{item.year}년이 두 번 왔습니다.")
|
||||
by_year[item.year] = item
|
||||
rows: list[dict[str, Any]] = []
|
||||
previous: LonLat | None = None
|
||||
for year in range(year_start, year_end + 1):
|
||||
item = by_year.get(year)
|
||||
start = previous or (item.start if item else None)
|
||||
end = item.end if item else None
|
||||
if start is None or end is None:
|
||||
side = "시점" if start is None else "종점"
|
||||
raise HTTPException(status_code=400, detail=f"{year}년 {side}이 비었습니다.")
|
||||
rows.append({"year": year, "start": start, "end": end})
|
||||
previous = end
|
||||
return rows
|
||||
|
||||
|
||||
def with_tm(rows: list[dict[str, Any]]) -> tuple[int, list[dict[str, Any]]]:
|
||||
"""(epsg, 행마다 start_x · start_y · end_x · end_y 더함)."""
|
||||
lons = [p.lon for row in rows for p in (row["start"], row["end"])]
|
||||
epsg = tm_epsg_for_lon(sum(lons) / len(lons))
|
||||
out = []
|
||||
for row in rows:
|
||||
sx, sy = lonlat_to_tm(row["start"].lon, row["start"].lat, epsg)
|
||||
ex, ey = lonlat_to_tm(row["end"].lon, row["end"].lat, epsg)
|
||||
out.append({"year": row["year"], "start_x": sx, "start_y": sy, "end_x": ex, "end_y": ey})
|
||||
return epsg, out
|
||||
|
||||
|
||||
def changed(new: dict[str, Any], old: dict[str, Any] | None, epsg: int) -> bool:
|
||||
if not old or old.get("epsg") is None or int(old["epsg"]) != epsg:
|
||||
return True
|
||||
keys = ("start_x", "start_y", "end_x", "end_y")
|
||||
return any(old[k] is None or abs(float(old[k]) - new[k]) > SAME_M for k in keys)
|
||||
|
||||
|
||||
async def _load(project_id: UUID) -> tuple[dict[str, Any], dict[int, dict[str, Any]]]:
|
||||
"""(예정노선 프로젝트 칸, 연도 → 지금 행)."""
|
||||
async with (
|
||||
get_db_pool().acquire() as connection,
|
||||
connection.cursor(aiomysql.DictCursor) as cursor,
|
||||
):
|
||||
await cursor.execute(
|
||||
"SELECT project_kind, project_year, project_year_end FROM projects"
|
||||
" WHERE id = %s AND deleted_at IS NULL",
|
||||
(str(project_id),),
|
||||
)
|
||||
project = await cursor.fetchone()
|
||||
await cursor.execute(
|
||||
"SELECT year, start_x, start_y, end_x, end_y, epsg FROM plan_route_years"
|
||||
" WHERE project_id = %s",
|
||||
(str(project_id),),
|
||||
)
|
||||
rows = await cursor.fetchall()
|
||||
if not project:
|
||||
raise HTTPException(status_code=404, detail="프로젝트를 찾을 수 없습니다.")
|
||||
if project["project_kind"] != "PLAN" or project["project_year"] is None:
|
||||
raise HTTPException(status_code=400, detail="예정노선 프로젝트가 아닙니다.")
|
||||
return project, {int(row["year"]): row for row in rows}
|
||||
|
||||
|
||||
async def _store(
|
||||
project_id: UUID, epsg: int, writes: list[dict[str, Any]], drops: list[int]
|
||||
) -> None:
|
||||
"""바뀐 연도만 넣고(분석 노선 비움) · 범위 밖 연도 행은 지움 — 한 묶음."""
|
||||
async with get_db_pool().acquire() as connection:
|
||||
await connection.begin()
|
||||
try:
|
||||
async with connection.cursor() as cursor:
|
||||
for row in writes:
|
||||
await cursor.execute(
|
||||
"INSERT INTO plan_route_years"
|
||||
" (project_id, year, start_x, start_y, end_x, end_y, epsg)"
|
||||
" VALUES (%s, %s, %s, %s, %s, %s, %s)"
|
||||
" ON DUPLICATE KEY UPDATE start_x = VALUES(start_x),"
|
||||
" start_y = VALUES(start_y), end_x = VALUES(end_x), end_y = VALUES(end_y),"
|
||||
" epsg = VALUES(epsg), route_json = NULL, fingerprint = NULL,"
|
||||
" confirmed_at = NULL",
|
||||
(
|
||||
str(project_id),
|
||||
row["year"],
|
||||
row["start_x"],
|
||||
row["start_y"],
|
||||
row["end_x"],
|
||||
row["end_y"],
|
||||
epsg,
|
||||
),
|
||||
)
|
||||
for year in drops:
|
||||
await cursor.execute(
|
||||
"DELETE FROM plan_route_years WHERE project_id = %s AND year = %s",
|
||||
(str(project_id), year),
|
||||
)
|
||||
await connection.commit()
|
||||
except Exception:
|
||||
await connection.rollback()
|
||||
raise
|
||||
|
||||
|
||||
def _point(x: Any, y: Any, epsg: Any) -> dict[str, float] | None:
|
||||
if x is None or y is None or epsg is None:
|
||||
return None
|
||||
lon, lat = tm_to_lonlat(float(x), float(y), int(epsg))
|
||||
return {"lon": lon, "lat": lat, "x": float(x), "y": float(y)}
|
||||
|
||||
|
||||
def _view(project: dict[str, Any], rows: dict[int, dict[str, Any]]) -> dict[str, Any]:
|
||||
start = int(project["project_year"])
|
||||
end = int(project["project_year_end"] or start)
|
||||
years = []
|
||||
for year in range(start, end + 1):
|
||||
row = rows.get(year) or {}
|
||||
years.append(
|
||||
{
|
||||
"year": year,
|
||||
"start": _point(row.get("start_x"), row.get("start_y"), row.get("epsg")),
|
||||
"end": _point(row.get("end_x"), row.get("end_y"), row.get("epsg")),
|
||||
}
|
||||
)
|
||||
epsgs = {int(r["epsg"]) for r in rows.values() if r.get("epsg") is not None}
|
||||
return {
|
||||
"status": "success",
|
||||
"year_start": start,
|
||||
"year_end": end,
|
||||
"epsg": epsgs.pop() if len(epsgs) == 1 else None,
|
||||
"years": years,
|
||||
}
|
||||
|
||||
|
||||
@router.get("/{project_id}/plan/points", response_model=None)
|
||||
async def get_plan_points(project_id: UUID) -> dict[str, Any]:
|
||||
project, rows = await _load(project_id)
|
||||
return _view(project, rows)
|
||||
|
||||
|
||||
@router.put("/{project_id}/plan/points", response_model=None)
|
||||
async def put_plan_points(
|
||||
project_id: UUID,
|
||||
payload: PointsRequest,
|
||||
session: dict[str, Any] = Depends(require_company),
|
||||
) -> dict[str, Any]:
|
||||
if not is_forest_service(session):
|
||||
raise HTTPException(status_code=403, detail="예정노선 지점은 산림청만 고칠 수 있습니다.")
|
||||
project, old = await _load(project_id)
|
||||
start = int(project["project_year"])
|
||||
end = int(project["project_year_end"] or start)
|
||||
epsg, rows = with_tm(chain_years(start, end, payload.years))
|
||||
writes = [row for row in rows if changed(row, old.get(row["year"]), epsg)]
|
||||
drops = [year for year in old if not start <= year <= end]
|
||||
if writes or drops:
|
||||
await _store(project_id, epsg, writes, drops)
|
||||
_, now = await _load(project_id)
|
||||
return {**_view(project, now), "changed_years": [row["year"] for row in writes]}
|
||||
@@ -10,9 +10,10 @@ from __future__ import annotations
|
||||
|
||||
import logging
|
||||
import re
|
||||
from functools import lru_cache
|
||||
from pathlib import Path
|
||||
|
||||
from pyproj import CRS
|
||||
from pyproj import CRS, Transformer
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
@@ -257,3 +258,28 @@ def resolve_project_crs(
|
||||
if prj_crs:
|
||||
return prj_crs
|
||||
return f"EPSG:{db_epsg or 5186}"
|
||||
|
||||
|
||||
# --- 점 변환(경위도 ↔ Korea 2000 TM · 요구 345 · PLAN 529-3) ---
|
||||
|
||||
|
||||
def tm_epsg_for_lon(lon: float) -> int:
|
||||
"""경도로 TM 띠 — 126° 앞 서부 5185 · 128° 앞 중부 5186 · 그 밖 동부 5187(동해 5188 안 씀)."""
|
||||
return 5185 if lon < 126.0 else 5186 if lon < 128.0 else 5187
|
||||
|
||||
|
||||
@lru_cache(maxsize=16)
|
||||
def _transformer(source: int, target: int) -> Transformer:
|
||||
return Transformer.from_crs(source, target, always_xy=True)
|
||||
|
||||
|
||||
def lonlat_to_tm(lon: float, lat: float, epsg: int) -> tuple[float, float]:
|
||||
"""WGS84 경위도 → TM (x 동, y 북) m."""
|
||||
x, y = _transformer(4326, epsg).transform(lon, lat)
|
||||
return float(x), float(y)
|
||||
|
||||
|
||||
def tm_to_lonlat(x: float, y: float, epsg: int) -> tuple[float, float]:
|
||||
"""TM (x, y) m → WGS84 (경도, 위도)."""
|
||||
lon, lat = _transformer(epsg, 4326).transform(x, y)
|
||||
return float(lon), float(lat)
|
||||
|
||||
@@ -74,6 +74,7 @@ from B08_Quantity.B08_Quantity_Router_Milestone import router as b08_milestone_r
|
||||
from B09_Estimation.B09_Estimation_Router import router as b09_estimation_router
|
||||
from D01_PlanRegister.D01_PlanRegister_Router import dashboard_router as d01_dashboard_router
|
||||
from D01_PlanRegister.D01_PlanRegister_Router import router as d01_plan_register_router
|
||||
from D02_PlanBase.D02_PlanBase_Router import router as d02_plan_base_router
|
||||
from D03_PlanAnalysis.D03_PlanAnalysis_Router import router as d03_plan_analysis_router
|
||||
from E01_StabRegister.E01_StabRegister_Router import router as e01_stab_register_router
|
||||
from M01_MasterData.M01_MasterData_Router import router as m01_master_data_router
|
||||
@@ -477,6 +478,7 @@ app.include_router(b04_basins_router, dependencies=protected_with_company)
|
||||
app.include_router(tiles_router, dependencies=protected_with_company)
|
||||
app.include_router(b05_route_router, dependencies=protected_with_company)
|
||||
app.include_router(b05_route_lifecycle_router, dependencies=protected_with_company)
|
||||
app.include_router(d02_plan_base_router, dependencies=protected_with_company)
|
||||
app.include_router(d03_plan_analysis_router, dependencies=protected_with_company)
|
||||
app.include_router(b05_corridor_router, dependencies=protected_with_company)
|
||||
app.include_router(b05_route_replan_router, dependencies=protected_with_company)
|
||||
|
||||
@@ -0,0 +1,149 @@
|
||||
"""D02 기준 지정(PLAN 529-3) — 경위도 ↔ TM 변환 · 지점 정본 API(잇기 · 빈 연도 · 권한).
|
||||
|
||||
./venv/Scripts/python.exe -m pytest resources/tester/test_529_3_plan_points.py -q
|
||||
"""
|
||||
|
||||
import pytest
|
||||
from fastapi import FastAPI
|
||||
from fastapi.testclient import TestClient
|
||||
|
||||
from common_util.common_util_auth import require_company
|
||||
from common_util.common_util_crs import lonlat_to_tm, tm_epsg_for_lon, tm_to_lonlat
|
||||
from D02_PlanBase import D02_PlanBase_Router as router_mod
|
||||
|
||||
PID = "12953f5a-5ba1-42ad-a427-8ac857a58bc2"
|
||||
URL = f"/api/projects/{PID}/plan/points"
|
||||
|
||||
|
||||
# --- 좌표 변환 ---
|
||||
|
||||
|
||||
@pytest.mark.parametrize("epsg, lon0", [(5185, 125.0), (5186, 127.0), (5187, 129.0)])
|
||||
def test_origin_is_false_origin(epsg, lon0) -> None:
|
||||
x, y = lonlat_to_tm(lon0, 38.0, epsg) # 원점 = (200 000, 600 000)
|
||||
assert abs(x - 200000.0) < 0.001 and abs(y - 600000.0) < 0.001
|
||||
|
||||
|
||||
@pytest.mark.parametrize("lon, lat", [(127.3851, 36.3504), (125.7, 34.1), (129.36, 36.02)])
|
||||
def test_round_trip_within_1mm(lon, lat) -> None:
|
||||
epsg = tm_epsg_for_lon(lon)
|
||||
x, y = lonlat_to_tm(lon, lat, epsg)
|
||||
back = lonlat_to_tm(*tm_to_lonlat(x, y, epsg), epsg)
|
||||
assert abs(back[0] - x) < 0.001 and abs(back[1] - y) < 0.001
|
||||
|
||||
|
||||
def test_belt_by_longitude() -> None:
|
||||
assert [tm_epsg_for_lon(v) for v in (124.5, 125.999, 126.0, 127.999, 128.0, 131.8)] == [
|
||||
5185,
|
||||
5185,
|
||||
5186,
|
||||
5186,
|
||||
5187,
|
||||
5187,
|
||||
]
|
||||
|
||||
|
||||
# --- 지점 정본 API ---
|
||||
|
||||
|
||||
A, B, C = (
|
||||
{"lon": 127.10, "lat": 37.50},
|
||||
{"lon": 127.12, "lat": 37.51},
|
||||
{"lon": 127.15, "lat": 37.52},
|
||||
)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def api(monkeypatch):
|
||||
state = {
|
||||
"project": {"project_kind": "PLAN", "project_year": 2026, "project_year_end": 2027},
|
||||
"rows": {},
|
||||
"session": {"role": "USER", "email": "kim@forest.go.kr", "company_id": 1},
|
||||
"stores": [],
|
||||
}
|
||||
|
||||
async def load(project_id):
|
||||
return state["project"], {y: dict(r) for y, r in state["rows"].items()}
|
||||
|
||||
async def store(project_id, epsg, writes, drops):
|
||||
state["stores"].append((epsg, writes, drops))
|
||||
for row in writes:
|
||||
state["rows"][row["year"]] = {**row, "epsg": epsg}
|
||||
for year in drops:
|
||||
state["rows"].pop(year)
|
||||
|
||||
monkeypatch.setattr(router_mod, "_load", load)
|
||||
monkeypatch.setattr(router_mod, "_store", store)
|
||||
app = FastAPI()
|
||||
app.include_router(router_mod.router)
|
||||
app.dependency_overrides[require_company] = lambda: state["session"]
|
||||
client = TestClient(app)
|
||||
client.state = state
|
||||
return client
|
||||
|
||||
|
||||
def test_put_chains_years(api) -> None:
|
||||
# 2027 시점을 엉뚱하게 보내도 서버가 2026 종점으로 맞춤
|
||||
body = {"years": [{"year": 2026, "start": A, "end": B}, {"year": 2027, "start": A, "end": C}]}
|
||||
got = api.put(URL, json=body)
|
||||
assert got.status_code == 200
|
||||
data = got.json()
|
||||
assert data["epsg"] == 5186 and data["changed_years"] == [2026, 2027]
|
||||
y26, y27 = data["years"]
|
||||
assert y27["start"]["x"] == y26["end"]["x"] and y27["start"]["y"] == y26["end"]["y"]
|
||||
assert abs(y27["start"]["lon"] - B["lon"]) < 1e-8 and abs(y27["start"]["lat"] - B["lat"]) < 1e-8
|
||||
assert (y26["start"]["x"], y26["start"]["y"]) == pytest.approx(
|
||||
lonlat_to_tm(127.10, 37.50, 5186)
|
||||
)
|
||||
|
||||
|
||||
def test_put_rejects_empty_and_out_of_range(api) -> None:
|
||||
only_first = {"years": [{"year": 2026, "start": A, "end": B}]}
|
||||
got = api.put(URL, json=only_first)
|
||||
assert got.status_code == 400 and "2027년 종점" in got.json()["detail"]
|
||||
no_start = {"years": [{"year": 2026, "end": B}, {"year": 2027, "end": C}]}
|
||||
assert "2026년 시점" in api.put(URL, json=no_start).json()["detail"]
|
||||
outside = {"years": [{"year": 2026, "start": A, "end": B}, {"year": 2028, "end": C}]}
|
||||
assert api.put(URL, json=outside).status_code == 400
|
||||
assert api.state["stores"] == []
|
||||
|
||||
|
||||
def test_put_forest_only(api) -> None:
|
||||
body = {"years": [{"year": 2026, "start": A, "end": B}, {"year": 2027, "end": C}]}
|
||||
api.state["session"] = {"role": "USER", "email": "lee@company.com", "company_id": 1}
|
||||
assert api.put(URL, json=body).status_code == 403
|
||||
api.state["session"] = {"role": "SYSTEM_ADMIN", "email": "admin@x.com", "company_id": None}
|
||||
assert api.put(URL, json=body).status_code == 200
|
||||
|
||||
|
||||
def test_put_clears_only_changed_years(api) -> None:
|
||||
body = {"years": [{"year": 2026, "start": A, "end": B}, {"year": 2027, "end": C}]}
|
||||
api.put(URL, json=body)
|
||||
# 같은 지점 다시 저장 = 쓰기 없음(분석 노선 그대로) · 2027 종점만 옮기면 2027 만
|
||||
assert api.put(URL, json=body).json()["changed_years"] == []
|
||||
moved = {"years": [{"year": 2026, "start": A, "end": B}, {"year": 2027, "end": A}]}
|
||||
assert api.put(URL, json=moved).json()["changed_years"] == [2027]
|
||||
assert len(api.state["stores"]) == 2
|
||||
|
||||
|
||||
def test_get_lists_every_year_and_drops_outside(api) -> None:
|
||||
api.state["rows"][2030] = {
|
||||
"year": 2030,
|
||||
"start_x": 1,
|
||||
"start_y": 1,
|
||||
"end_x": 2,
|
||||
"end_y": 2,
|
||||
"epsg": 5186,
|
||||
}
|
||||
data = api.get(URL).json()
|
||||
assert [y["year"] for y in data["years"]] == [2026, 2027]
|
||||
assert data["years"][0]["start"] is None
|
||||
body = {"years": [{"year": 2026, "start": A, "end": B}, {"year": 2027, "end": C}]}
|
||||
api.put(URL, json=body)
|
||||
assert api.state["stores"][-1][2] == [2030] and 2030 not in api.state["rows"]
|
||||
|
||||
|
||||
def test_not_plan_project(api) -> None:
|
||||
api.state["project"] = {"project_kind": "ROAD", "project_year": 2026, "project_year_end": None}
|
||||
# _load 대역이 그대로 돌려주므로 진짜 _load 의 거절은 서버 시험(ORCA)에서
|
||||
assert api.get(URL).status_code == 200
|
||||
Reference in New Issue
Block a user