eomsangdon/Aislo
1
0
Fork 0
Code Issues Pull Requests 2 Actions Packages Projects Releases Wiki Activity

feat(D03): 꺾임점 고침 [저장] 정본 — 서버가 마디로 선 · 지표 다시 셈 · 이은 누가 고친 길이 · GET 은 분석 지문 같은 고침만 · 산림청만 · 시험 (PLAN 543-3)

Browse Source 
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
eomsangdon
2026-10-26 13:32:13 +09:00
co-authored by Claude Opus 5.5
parent 7683b74524
commit 47c3b65cc4
2 changed files with 155 additions and 10 deletions
Download Patch File Download Diff File Expand all files Collapse all files
D03_PlanAnalysis/D03_PlanAnalysis_Router.py
+80 -9
View File
@@ -4,11 +4,14 @@
- `POST …/plan/analyze` — 연도 구간마다 노선 추천 갈래 넷(뒤에서 · 파생 저장본)
- `GET …/plan/progress` — 두 일의 진행률
- `GET …/plan/analysis` — 저장본 + 고른 갈래 · 이은 누가 · 이은 중심선
- `PUT …/plan/analysis` — [저장] 연도별 고른 갈래 → `plan_route_years.route_json` · `fingerprint`
- `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
@@ -31,6 +34,7 @@ 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__)
@@ -54,8 +58,19 @@ 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]]:
@@ -220,14 +235,46 @@ def _saved_mode(route_json: Any) -> str | None:
def _assemble(
root: Path, project: dict[str, Any], segments: list[dict[str, Any]], choices: dict[int, Any]
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 · 빈 선."""
"""(연도 저장본 · 이은 누가 · 이은 중심선) — 저장본이 하나라도 없으면 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, []
rows = analyze.chain(results, choices, float(project["route_start_m"] or 0.0))
return results, rows, analyze.joined_line(results, rows)
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:
@@ -247,11 +294,22 @@ async def plan_analysis(project_id: UUID, points: str | None = None) -> dict[str
"""
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)
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}
{
"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,
@@ -297,7 +355,7 @@ async def save_plan_analysis(
session: dict[str, Any] = Depends(require_company),
) -> dict[str, Any]:
"""[저장] — 끈 지점(있으면 정본으로) · 연도별 고른 갈래를 서버 저장본으로 다시 이어 씀."""
if request.points and not is_forest_service(session):
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]
@@ -313,6 +371,17 @@ async def save_plan_analysis(
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:
@@ -320,8 +389,10 @@ async def save_plan_analysis(
try:
async with connection.cursor() as cursor:
for result, row in zip(results, rows):
route = result["routes"][row["mode"]]
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",
resources/tester/test_530_2_plan_router.py
+75 -1
View File
@@ -1,5 +1,6 @@
"""D 분석 API(PLAN 530-2 · 530-3) — GET 이은 누가 · PUT [저장] = 서버 저장본으로 다시 셈 · 모르는 갈래
거절 · 끈 지점(points) = 그 지점으로 보기 · 안 이어지면 거절 · [저장] 때만 지점 정본(산림청만).
거절 · 끈 지점(points) = 그 지점으로 보기 · 안 이어지면 거절 · [저장] 때만 지점 정본(산림청만) ·
꺾임점 고침(543-3) = 서버가 마디로 선 · 지표를 다시 셈 · 누가가 고친 길이로 · GET 은 지문 같은 고침만.
./venv/Scripts/python.exe -m pytest resources/tester/test_530_2_plan_router.py -q
"""
@@ -13,6 +14,7 @@ from fastapi.testclient import TestClient
from common_util.common_util_auth import require_company
from common_util.common_util_json import atomic_write_json
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_Router as router_mod
PID = "12953f5a-5ba1-42ad-a427-8ac857a58bc2"
@@ -83,6 +85,7 @@ def client(tmp_path, monkeypatch):
"start": list(seg["start"]),
"end": list(seg["end"]),
"fingerprint": "f",
"criteria": {"min_radius_m": 12.0, "max_grade_pct": 14.0},
"routes": routes,
},
)
@@ -109,7 +112,23 @@ def client(tmp_path, monkeypatch):
stored.append([(s["year"], s["start"], s["end"]) for s in segs])
monkeypatch.setattr(router_mod, "_store_points", store_points)
# 꺾임점 고침 지표 — 지형 없이(선 길이만)
loads: list = []
monkeypatch.setattr(
analyze, "_load", lambda root, year, loader: loads.append(year) or (None, [])
)
monkeypatch.setattr(
plan_edit,
"edit_metrics",
lambda line, result, terrain, streams: {
"length_m": round(analyze.line_length(line.vertices), 2),
"max_grade_pct": 9.5,
"over_limit": {"total": 1},
},
)
test = TestClient(app)
test.segments = segments
test.loads = loads
test.log = log
test.session = session
test.stored = stored
@@ -179,3 +198,58 @@ def test_put_with_points_stores_points_forest_only(client) -> None:
moved = {"choices": {}, "points": _points((310.0, 400.0))}
client.session["email"] = "a@forest.go.kr"
assert client.put(url, json=moved).status_code == 409 and len(client.stored) == 1
def _edit(nodes, fingerprint="f") -> dict:
return {
"base_mode": "gentle",
"base_fingerprint": fingerprint,
"nodes": nodes,
"curve_flags": [True] * len(nodes),
"radii": [None] * len(nodes),
}
def test_put_edit_server_redraws_and_get_returns(client) -> None:
url = f"/api/projects/{PID}/plan/analysis"
edit = _edit([[300, 400], [600, 560], [900, 400]])
got = client.put(url, json={"choices": {}, "edits": {"2027": edit}})
assert got.status_code == 200, got.text
updates = {args[3]: json.loads(args[0]) for sql, args in client.log if sql.startswith("UPDATE")}
body = updates[2027]
line = plan_edit.edited_line(edit, 12.0)
assert len(body["planned"]) == len(line.vertices) and body["edit"]["nodes"] == edit["nodes"]
assert body["planned"][0] == [300, 400] and body["planned"][-1] == [900, 400]
assert body["metrics"]["max_grade_pct"] == 9.5 and len(body["curves"]) == 1
length = analyze.line_length(body["planned"])
assert got.json()["chain"][1]["length_m"] == round(length, 3) != 640.0
assert "edit" not in updates[2026] and client.loads == [2027]
# GET — 저장된 고침 + 고친 길이로 이은 누가 · 지문이 바뀌면 안 돌려줌
client.segments[1]["route_json"] = json.dumps(body)
view = client.get(url).json()
assert view["years"][1]["edit"]["nodes"] == edit["nodes"]
assert view["years"][1]["edit"]["metrics"]["over_limit"]["total"] == 1
assert view["chain"][1]["length_m"] == round(length, 3)
assert view["years"][1]["analysis"]["routes"]["gentle"]["metrics"]["length_m"] == 640.0
# 같은 입력 다시 [저장] = 저장된 선 · 지표 그대로(지형 안 읽음)
assert client.put(url, json={"choices": {}, "edits": {"2027": edit}}).status_code == 200
assert client.loads == [2027]
client.segments[1]["route_json"] = json.dumps(
{**body, "edit": {**body["edit"], "base_fingerprint": "old"}}
)
stale = client.get(url).json()
assert stale["years"][1]["edit"] is None and stale["chain"][1]["length_m"] == 640.0
def test_put_edit_rejects(client) -> None:
url = f"/api/projects/{PID}/plan/analysis"
off_end = _edit([[300, 400], [600, 560], [901, 400]])
stale = _edit([[300, 400], [600, 560], [900, 400]], "old")
other = {**stale, "base_fingerprint": "f", "base_mode": "gradeline"}
for bad in (off_end, stale, other):
assert client.put(url, json={"choices": {}, "edits": {"2027": bad}}).status_code == 409
client.session["email"] = "a@gmail.com"
good = _edit([[300, 400], [600, 560], [900, 400]])
assert client.put(url, json={"choices": {}, "edits": {"2027": good}}).status_code == 403
assert not [sql for sql, _ in client.log if sql.startswith("UPDATE")]