feat(635-2): 받은 세트 문맥 GET plan-context · 원본 꼭짓점 GET route/plan/source · 다시 추적 전 조밀화 = 체인 초기선 거리 0(요구 546 · 547)

- 세트 연결 = 638-1 그대로(B 하나 = 세트 하나 · 열쇠만 세트)
- 새 D04_PlanConfirm_Context — 같은 예정노선 세트 모두 · 내 세트 · 연도 · 임도 좌표계로 pyproj 변환 · 업체 이름 없음 · 보기 전용
- 새 B05_Profile_Router_PlanSource — 원본 꼭짓점 = 범위 + 지표면 트림 · 조밀화만 뺌(Router_Replan 안 늘림 · 덧칸 대신 새 길)
- RouteFollow 추적 전 체인과 같은 3.6 m 조밀화(브레인 허락 · 옛 촘촘 입력은 그대로 = 626-2 같음)
- 실측 검증_sub3: 원본 64 점 → follow = 초기선 181 점 거리 0 · 파일 안 바뀜

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

B05_Profile/B05_Profile_Router_PlanSource.py

@@ -0,0 +1,84 @@
+"""받은 예정노선 원본 꼭짓점(635-2 · 요구 546) — B05 「예정노선 수정」이 고칠 선.
+
+계산 자리 ③ 서버 단독 · 읽기만(파일 안 만듦). 원본 = 올린(받은) 노선 파일 꼭짓점을 사업지 좌표로 ·
+사용 범위(`projects.route_start_m · route_end_m`) · 지표면 트림까지 — 체인 예정노선
+(`load_design_route`)과 같은 차례에서 **조밀화만 뺌**. 고친 꼭짓점은 `POST /route/follow` 가
+같은 간격(`spacing_m`)으로 조밀화한 뒤 추적 → 안 고친 원본 = 체인 초기선과 거리 0(`test_635_2`).
+
+  GET /api/projects/{id}/route/plan/source
+    → {status, expected_source: [[x, y], …], spacing_m}   (원본 없음 = 빈 목록)
+    404 프로젝트 없음
+"""
+
+from __future__ import annotations
+
+import asyncio
+from pathlib import Path
+from typing import Any
+from uuid import UUID
+
+from fastapi import APIRouter
+from fastapi.responses import JSONResponse
+
+from B05_Profile.B05_Profile_Router_Replan import _PROJECT_PATH_MISSING, _project_paths
+from common_util.common_util_route_geometry import load_design_route, trim_route_to_surface
+from common_util.common_util_surface_confirmation import get_surface_confirmation_params
+from common_util.common_util_surface_sampler import build_surface_sampler
+from config.config_db import get_db_pool
+from config.config_system import (
+    ROUTE_DIRECT_LINK_CELL_FACTOR,
+    ROUTE_GRID_RES_M,
+    ROUTE_PLANNED_DENSIFY_SAFETY,
+)
+
+router = APIRouter(prefix="/api/projects", tags=["B05 Plan Source"])
+
+#: 체인 예정노선 조밀화 간격(`load_design_route` 와 같은 값)
+SPACING_M = ROUTE_DIRECT_LINK_CELL_FACTOR * ROUTE_GRID_RES_M * ROUTE_PLANNED_DENSIFY_SAFETY
+
+
+def source_vertices(
+    root: Path, surface: dict[str, Any] | None, route_range: tuple | None
+) -> list[list[float]]:
+    """원본 꼭짓점 — 사업지 좌표 · 범위 · 지표면 트림(지표면을 못 열면 트림 없이) · 조밀화 없음."""
+    planned = load_design_route(root, None, route_range)
+    if planned is None:
+        return []
+    points = [(v.x, v.y) for v in planned.vertices]
+    if surface:
+        try:
+            sampler = build_surface_sampler(
+                root / "B04_PreProcess" / "models",
+                str(surface["source_filter"]),
+                str(surface["method"]),
+                bool(surface["smooth"]),
+            )
+        except (FileNotFoundError, KeyError, OSError):
+            sampler = None
+        if sampler is not None:
+            points = trim_route_to_surface(points, sampler)
+    return [[float(x), float(y)] for x, y in points] if len(points) >= 2 else []
+
+
+@router.get("/{project_id}/route/plan/source", response_model=None)
+async def read_plan_source(project_id: UUID) -> dict[str, Any] | JSONResponse:
+    paths = await _project_paths(project_id)
+    if paths is None:
+        return JSONResponse(status_code=404, content=_PROJECT_PATH_MISSING)
+    async with get_db_pool().acquire() as connection:
+        surface = await get_surface_confirmation_params(connection, str(project_id))
+        async with connection.cursor() as cursor:
+            await cursor.execute(
+                "SELECT route_start_m, route_end_m FROM projects WHERE id = %s",
+                (str(project_id),),
+            )
+            row = await cursor.fetchone()
+    vertices = await asyncio.to_thread(
+        source_vertices, paths[0], surface, (row[0], row[1]) if row else None
+    )
+    return {
+        "status": "success",
+        "project_id": str(project_id),
+        "expected_source": vertices,
+        "spacing_m": SPACING_M,
+    }

B05_Profile/B05_Profile_Router_RouteFollow.py

@@ -1,9 +1,11 @@
 """예정노선 다시 추적(626-2 · 요구 530) — 고친 예정노선 → 평면선(노드 · 곡선 · 편집표) · 저장 0.
 
 계산 자리 ③ 서버 단독 — 체인 초기 폴리라인(`_ensure_planned_initial`)과 같은 함수
-`followed_table`(지형 안 씀 · 순수 기하). 화면은 받은 노드를 편집기 한 걸음으로 싣고 페이지 [저장]
-길로 보냄(고친 예정노선 = 초안 `expected` → `expected_route_edit.csv`). `/route/replan` ·
-`as_initial` · `expected_route.csv` · 초기값 · 추천 저장본은 안 건드림.
+`followed_table`(지형 안 씀 · 순수 기하) · 추적 전 체인 예정노선과 같은 간격 조밀화(635-2 —
+성긴 원본 꼭짓점 = 체인 초기선과 같은 선 · 촘촘한 입력은 그대로).
+화면은 받은 노드를 편집기 한 걸음으로 싣고 페이지 [저장] 길로 보냄(고친 예정노선 = 초안 `expected`
+→ `expected_route_edit.csv`). `/route/replan` · `as_initial` · `expected_route.csv` · 초기값 ·
+추천 저장본은 안 건드림.
 
   POST /api/projects/{id}/route/follow
     본문 {points: [[x, y], …], crs: "tm" | "lonlat"}   ("lonlat" = [경도, 위도])
@@ -30,7 +32,13 @@ from B05_Profile.B05_Profile_Router_Replan import (
     router,
 )
 from common_util.common_util_crs import lonlat_to_tm
+from common_util.common_util_route_geometry import densify_route
 from config.config_db import get_db_pool
+from config.config_system import (
+    ROUTE_DIRECT_LINK_CELL_FACTOR,
+    ROUTE_GRID_RES_M,
+    ROUTE_PLANNED_DENSIFY_SAFETY,
+)
 
 
 class RouteFollowRequest(BaseModel):
@@ -40,7 +48,8 @@ class RouteFollowRequest(BaseModel):
 
 def follow_route_line(points: list[tuple[float, float]], radius_m: float) -> dict[str, Any]:
     """점 묶음(TM) → 예정노선 추적 평면선 — 파일 꼴(csv 4자리)과 같은 값."""
-    result, table = followed_table(points, radius_m)
+    spacing = ROUTE_DIRECT_LINK_CELL_FACTOR * ROUTE_GRID_RES_M * ROUTE_PLANNED_DENSIFY_SAFETY
+    result, table = followed_table(densify_route(points, spacing), radius_m)
     return {
         "planned": [[round(x, 4), round(y, 4)] for x, y in result.vertices],
         "nodes": [node.as_dict() for node in result.nodes],

D04_PlanConfirm/D04_PlanConfirm_Context.py

@@ -0,0 +1,81 @@
+"""임도 프로젝트가 받은 예정노선 문맥(635-2 · 요구 546 · 547) — 같은 예정노선의 세트 모두 · 내 세트.
+
+계산 자리 ③ 서버 단독 · 보기 전용(파일 · DB 안 씀) — 예정노선 설계 내용 파일을 그때 읽어
+(`D03_PlanAnalysis_Store.plan_state`) 조각 꺾은선을 임도 프로젝트 좌표계로 바꿈(pyproj ·
+예정노선 = GRS80 TM · 임도 = 사업지 좌표계 `resolve_project_crs`). 업체 이름 · 메일은 안 실음.
+
+  GET /api/projects/{id}/plan-context
+    → {status, linked: False} | {status, linked: True, plan_name, epsg(임도), mine, year,
+       sets: [{key, year, no, mine, parts, branch, length_m, lines: [[[x, y], …] 조각마다]}]}
+"""
+
+from __future__ import annotations
+
+from pathlib import Path
+from typing import Any
+from uuid import UUID
+
+from fastapi import APIRouter
+from pyproj import CRS, Transformer
+
+from common_util.common_util_crs import resolve_project_crs
+from common_util.common_util_storage import resolve_stored_project_path
+from D03_PlanAnalysis import D03_PlanAnalysis_Store as store
+from D04_PlanConfirm.D04_PlanConfirm_Link import _fetch_one
+
+router = APIRouter(prefix="/api/projects", tags=["D04 Plan Context"])
+
+
+def context_sets(
+    state: dict[str, Any], mine: dict[str, Any] | None, target: str
+) -> list[dict[str, Any]]:
+    """세트마다 조각 꺾은선(임도 좌표) — 세트 · 조각 차례 그대로."""
+    source = f"EPSG:{int(state['plan']['epsg'])}"
+    move = Transformer.from_crs(source, target, always_xy=True)
+    pieces = {p["id"]: p for p in state["pieces"]}
+    out = []
+    for unit in state["units"]:
+        lines = [
+            [list(move.transform(x, y)) for x, y in pieces[pid]["planned"]]
+            for pid in unit["pieces"]
+        ]
+        out.append(
+            {
+                "key": unit["id"],
+                "year": unit["year"],
+                "no": unit["set"],
+                "mine": mine is not None and unit["id"] == mine["id"],
+                "parts": unit["parts"],
+                "branch": unit["branch"],
+                "length_m": unit["length_m"],
+                "lines": lines,
+            }
+        )
+    return out
+
+
+@router.get("/{project_id}/plan-context", response_model=None)
+async def plan_context(project_id: UUID) -> dict[str, Any]:
+    """받은 세트 · 같은 예정노선의 다른 세트(다른 해 · 같은 해 다른 업체) ·
+    연결 없으면 linked False."""
+    row = await _fetch_one(
+        "SELECT a.project_id, a.piece_id, s.name AS plan_name FROM plan_route_assignees a"
+        " JOIN projects s ON s.id = a.project_id AND s.deleted_at IS NULL"
+        " WHERE a.linked_project_id = %s ORDER BY a.id LIMIT 1",
+        (str(project_id),),
+    )
+    if not row:
+        return {"status": "success", "linked": False}
+    road = await store.project_row(project_id)
+    target = resolve_project_crs(Path(resolve_stored_project_path(road["storage_path"])))
+    state = await store.plan_state(str(row["project_id"]))
+    mine = store.find_unit(state["units"], row["piece_id"])
+    return {
+        "status": "success",
+        "linked": True,
+        "plan_name": row["plan_name"],
+        "epsg": CRS.from_user_input(target).to_epsg(),  # WKT 원문이면 못 찾을 때 None
+        "mine": mine and mine["id"],
+        "year": mine and mine["year"],
+        "sets": context_sets(state, mine, target) if state["plan"]["epsg"] else [],
+    }

main.py

@@ -47,6 +47,7 @@ from B04_PreProcess.B04_PreProcess_Router_Watershed import router as b04_watersh
 from B05_Profile.B05_Profile_Router import router as b05_route_router
 from B05_Profile.B05_Profile_Router_Corridor import router as b05_corridor_router
 from B05_Profile.B05_Profile_Router_Lifecycle import router as b05_route_lifecycle_router
+from B05_Profile.B05_Profile_Router_PlanSource import router as b05_plan_source_router
 from B05_Profile.B05_Profile_Router_Replan import router as b05_route_replan_router
 from B05_Profile.B05_Profile_Router_Terrain import router as b05_route_terrain_router
 from B05_Profile.B05_Profile_Structures_Router import router as b05_structures_router
@@ -79,6 +80,7 @@ from D03_PlanAnalysis.D03_PlanAnalysis_Lidar import router as d03_plan_lidar_rou
 from D03_PlanAnalysis.D03_PlanAnalysis_Network import router as d03_plan_network_router
 from D03_PlanAnalysis.D03_PlanAnalysis_Router import router as d03_plan_analysis_router
 from D03_PlanAnalysis.D03_PlanAnalysis_Router_Tiles import router as d03_plan_tiles_router
+from D04_PlanConfirm.D04_PlanConfirm_Context import router as d04_plan_context_router
 from D04_PlanConfirm.D04_PlanConfirm_Link import dashboard_router as d04_link_dashboard_router
 from D04_PlanConfirm.D04_PlanConfirm_Link import router as d04_plan_link_router
 from D04_PlanConfirm.D04_PlanConfirm_Router import router as d04_plan_confirm_router
@@ -501,6 +503,7 @@ app.include_router(d03_plan_network_router, dependencies=protected_with_company)
 app.include_router(d03_plan_lidar_router, dependencies=protected_with_company)
 app.include_router(d04_plan_confirm_router, dependencies=protected_with_company)
 app.include_router(d04_plan_link_router, dependencies=protected_with_company)
+app.include_router(d04_plan_context_router, dependencies=protected_with_company)
 app.include_router(e02_stab_ground_router, dependencies=protected_with_company)
 app.include_router(e04_stab_3d_router, dependencies=protected_with_company)
 app.include_router(e05_stab_report_router, dependencies=protected_with_company)
@@ -511,6 +514,7 @@ app.include_router(e03_stab_prob_router, dependencies=protected_with_company)
 app.include_router(e06_stab_rock_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)
+app.include_router(b05_plan_source_router, dependencies=protected_with_company)
 app.include_router(b05_route_terrain_router, dependencies=protected_with_company)
 app.include_router(b05_structures_router, dependencies=protected_with_company)
 app.include_router(b06_section_router, dependencies=protected_with_company)

resources/tester/test_635_2_plan_context.py

@@ -0,0 +1,183 @@
+# -*- coding: utf-8 -*-
+"""635-2 받은 세트 문맥 · 원본 꼭짓점 · 다시 추적 거리 0(요구 546 · 547).
+
+- 원본 꼭짓점(조밀화 전 · 범위 · 지표면 트림) → `/route/follow` = 체인 초기선과 같은 선(거리 0) —
+  follow 가 체인 예정노선과 같은 간격으로 조밀화한 뒤 추적(촘촘한 옛 입력은 그대로)
+- `GET /route/plan/source` = 원본 꼭짓점 · 조밀화 없음 · 트림 · 범위 넘김
+- `GET /plan-context` = 같은 예정노선 세트 모두 · 내 세트 · 연도 · 임도 좌표계 · 업체 이름 없음 ·
+  연결 없으면 linked False · 파일 안 씀
+"""
+
+from __future__ import annotations
+
+import asyncio
+import math
+import sys
+from contextlib import asynccontextmanager
+from pathlib import Path
+from types import SimpleNamespace
+from uuid import uuid4
+
+from pyproj import Transformer
+
+ROOT = Path(__file__).resolve().parents[2]
+if str(ROOT) not in sys.path:
+    sys.path.insert(0, str(ROOT))
+
+from helper_plan_file import plan_of, use_plan  # noqa: E402
+
+from B05_Profile import B05_Profile_Router_PlanSource as source  # noqa: E402
+from B05_Profile import B05_Profile_Router_Replan as replan  # noqa: E402
+from B05_Profile import B05_Profile_Router_RouteFollow as follow  # noqa: E402
+from common_util.common_util_route_geometry import (  # noqa: E402
+    densify_route,
+    expected_route_csv_path,
+    planned_route_initial_path,
+    write_route_csv,
+)
+from D04_PlanConfirm import D04_PlanConfirm_Context as context  # noqa: E402
+
+RADIUS_M = 12.0
+# 성긴 원본 꼭짓점(17 m 안팎 · 굽은 산길) — 사업지 TM
+SPARSE = [(200000.0 + 17 * i, 450000.0 + 30 * math.sin(i / 3)) for i in range(40)]
+
+
+def _rows(path: Path) -> list[list[float]]:
+    return replan._vertices_of(path)
+
+
+def test_sparse_vertices_follow_equals_chain_initial(tmp_path: Path) -> None:
+    """체인 = 원본 → 조밀화(3.6 m) → 예정노선 파일 → 초기 폴리라인 · follow(원본) = 같은 선."""
+    dense = densify_route(SPARSE, source.SPACING_M)
+    assert len(dense) > 3 * len(SPARSE)
+    write_route_csv(expected_route_csv_path(tmp_path), [{"x": x, "y": y} for x, y in dense])
+    assert replan._ensure_planned_initial(tmp_path, RADIUS_M)
+    initial = _rows(planned_route_initial_path(tmp_path))
+    line = follow.follow_route_line(SPARSE, RADIUS_M)["planned"]
+    assert len(line) == len(initial)
+    assert max(math.dist(a, b) for a, b in zip(line, initial, strict=True)) < 1e-3
+    # 촘촘한 옛 입력(체인 파일 점) = 조밀화가 아무것도 안 끼움 · 같은 선
+    same = [tuple(p) for p in replan._vertices_of(expected_route_csv_path(tmp_path))]
+    assert densify_route(same, source.SPACING_M) == same
+    assert follow.follow_route_line(same, RADIUS_M)["planned"] == initial
+
+
+def test_plan_source_skips_densify_keeps_trim(tmp_path: Path, monkeypatch) -> None:
+    seen = {}
+
+    def load(root, surface, route_range):
+        seen["args"] = (root, surface, route_range)
+        return SimpleNamespace(vertices=[SimpleNamespace(x=x, y=y) for x, y in SPARSE])
+
+    def trim(points, sampler):
+        return points[2:-2]
+
+    monkeypatch.setattr(source, "load_design_route", load)
+    monkeypatch.setattr(source, "build_surface_sampler", lambda *a: object())
+    monkeypatch.setattr(source, "trim_route_to_surface", trim)
+    surface = {"source_filter": "classification", "method": "dtm", "smooth": True}
+    got = source.source_vertices(tmp_path, surface, (10.0, 500.0))
+    assert seen["args"] == (tmp_path, None, (10.0, 500.0))  # 지표면 인자 없이 = 조밀화 안 함
+    assert got == [list(p) for p in SPARSE[2:-2]]
+
+    def broken(*a):
+        raise FileNotFoundError("모델 없음")
+
+    monkeypatch.setattr(source, "build_surface_sampler", broken)
+    assert len(source.source_vertices(tmp_path, surface, None)) == len(SPARSE)  # 트림 없이
+    monkeypatch.setattr(source, "load_design_route", lambda *a: None)
+    assert source.source_vertices(tmp_path, surface, None) == []
+
+    # 끝점 — 경로 없음 404 · 있으면 칸 · 간격
+    async def paths(_pid):
+        return tmp_path, "s"
+
+    class _Cursor:
+        async def __aenter__(self):
+            return self
+
+        async def __aexit__(self, *a):
+            return False
+
+        async def execute(self, *a):
+            pass
+
+        async def fetchone(self):
+            return (None, None)
+
+    class _Conn:
+        def cursor(self):
+            return _Cursor()
+
+    class _Pool:
+        @asynccontextmanager
+        async def acquire(self):
+            yield _Conn()
+
+    async def confirmed(*a):
+        return surface
+
+    monkeypatch.setattr(source, "_project_paths", paths)
+    monkeypatch.setattr(source, "get_db_pool", lambda: _Pool())
+    monkeypatch.setattr(source, "get_surface_confirmation_params", confirmed)
+    monkeypatch.setattr(source, "load_design_route", load)
+    monkeypatch.setattr(source, "build_surface_sampler", lambda *a: object())
+    body = asyncio.run(source.read_plan_source(uuid4()))
+    assert body["expected_source"] == [list(p) for p in SPARSE[2:-2]]
+    assert body["spacing_m"] == source.SPACING_M == 3.6
+
+    async def none(_pid):
+        return None
+
+    monkeypatch.setattr(source, "_project_paths", none)
+    assert asyncio.run(source.read_plan_source(uuid4())).status_code == 404
+
+
+MAIN = {
+    "id": "main",
+    "vertices": [[200000.0, 450000.0], [200100.0, 450000.0], [200200.0, 450000.0]],
+}
+GAP = {"id": "g", "vertices": [[200300.0, 450000.0], [200400.0, 450000.0]]}
+
+
+def test_plan_context_sets_mine_crs(tmp_path: Path, monkeypatch) -> None:
+    plan = plan_of(
+        [MAIN, GAP],
+        [{"line": "main", "at_m": 100.0}],
+        [
+            {"id": "a", "line": "main", "from_m": 0, "to_m": 100, "year": 2027, "set": 1},
+            {"id": "b", "line": "main", "from_m": 100, "to_m": 200, "year": 2027, "set": 2},
+            {"id": "c", "line": "g", "from_m": 0, "to_m": 100, "year": 2028, "set": 1},
+        ],
+    )
+    use_plan(monkeypatch, plan)
+    rows = {"link": {"project_id": "D", "piece_id": "b", "plan_name": "예정"}}
+
+    async def fetch(sql, args):
+        assert "a.linked_project_id" in sql and "company" not in sql
+        return rows["link"]
+
+    async def road(_pid):
+        return {"storage_path": "x"}
+
+    monkeypatch.setattr(context, "_fetch_one", fetch)
+    monkeypatch.setattr(context.store, "project_row", road)
+    monkeypatch.setattr(context, "resolve_stored_project_path", lambda s: str(tmp_path))
+    monkeypatch.setattr(context, "resolve_project_crs", lambda root: "EPSG:5174")
+    out = asyncio.run(context.plan_context(uuid4()))
+    assert (out["linked"], out["mine"], out["year"], out["epsg"]) == (True, "2027-2", 2027, 5174)
+    assert [(s["key"], s["mine"], s["year"], len(s["lines"])) for s in out["sets"]] == [
+        ("2027-1", False, 2027, 1),
+        ("2027-2", True, 2027, 1),
+        ("2028-1", False, 2028, 1),
+    ]
+    move = Transformer.from_crs("EPSG:5186", "EPSG:5174", always_xy=True)
+    want = [list(move.transform(*p)) for p in ([200100.0, 450000.0], [200200.0, 450000.0])]
+    assert out["sets"][1]["lines"][0] == want  # 내 세트 = 임도 좌표(Bessel)
+    assert "업체" not in str(out) and "email" not in str(out)
+    # 옛 업체 열쇠(조각 id) · 세트 열쇠 = 같은 답
+    rows["link"] = {**rows["link"], "piece_id": "2027-2"}
+    assert asyncio.run(context.plan_context(uuid4()))["mine"] == "2027-2"
+    rows["link"] = None
+    assert asyncio.run(context.plan_context(uuid4())) == {"status": "success", "linked": False}
+    assert not any(tmp_path.iterdir())  # 파일 안 씀