feat(D 예정노선): 지형 받기 서버 — 지적 뺌 · 실패 조각만 · 위치 바꿈 멀다 · 띠 · reset · 칸 상한 · 끊김 · 단계 글 · 임도망 ±5 km (PLAN 595 · 요구 463)

- 지적도를 지형 일에서 뺌(겹 GET 이 받음) · 지적 실패 조각만 다시(열쇠 failed_tiles)
- 받기 몸 center · reset — 5 km 넘음 far · 다른 TM 띠 band 물음 · reset = 옛 범위 · 병합 도엽 · 위성 버림
- 받을 칸 500 만 넘음 413 · extent max_cells · lon/lat
- 진행 기록 파일 — 서버 다시 켜짐 = interrupted(percent < 100 · message 없음) · 받은 적 없음 percent 100
- 진행 단계 글 n/5 · 받은 범위에 epsg · center 기록
- 임도망 자르기 = 기준점 ±5 km ∪ 받은 지형 · 지형 전에도 올림 · bounds · distance_m · 밖뿐 = 거리 글
- 시험 530_1 · 530_2 · 569_1 · 584_3 덧

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

D03_PlanAnalysis/D03_PlanAnalysis_Api.ts

@@ -27,11 +27,26 @@ export interface PlanBaseView {
   site: { lon: number; lat: number; address: string | null } | null;
 }
 
+/** 진행(595-3) — 끊김 = running 거짓 · percent < 100 · message 없음 · interrupted(서버가 다시 켜짐) ·
+ *  받은 적 없음 = percent 100 · message = 「n/5 …」 단계 글. */
 export interface PlanJob {
   running: boolean;
   percent: number;
   message: string | null;
   error: string | null;
+  interrupted?: boolean;
+}
+
+/** 받기 답(595-2) — 시작 · 멀다(받은 범위에서 5 km 넘음) · 다른 TM 띠 — 둘은 확인 뒤 `reset`. */
+export type TerrainStart =
+  | ({ status: "accepted" } & PlanJob)
+  | { status: "far"; distance_m: number; message: string }
+  | { status: "band"; epsg: number; center_epsg: number; message: string };
+
+/** 위치 칸 기준점(경위도 · 캐시) · 새로 받기(옛 받은 범위 버림). */
+export interface TerrainAim {
+  center?: { lon: number; lat: number } | null;
+  reset?: boolean;
 }
 
 /** 받기 전 범위(561-2) — 상자(작업 좌표계 m · 경위도 네 귀) · 받은 범위 · 도엽 · 칸 수 어림. */
@@ -45,6 +60,8 @@ export interface TerrainExtent {
   covered: boolean;
   sheets: Array<{ no: string; ring: Array<[number, number]>; stored: boolean }>;
   cells: number;
+  /** 받기 칸 상한(넘으면 받기 413). */
+  max_cells: number;
 }
 
 /** 그은 선 서버 꼴(562-2) — 마크 = 조각 경계 · 조각 없는 구역 = 연도 없음. */
@@ -89,26 +106,33 @@ async function request<T>(path: string, init: RequestInit = {}): Promise<T> {
 
 const base = (projectId: string): string => `/projects/${projectId}/plan`;
 
-/** 받기 전 범위 — `extra` = 더 덮을 점(그은 꺾임점). */
+/** 받기 전 범위 — `extra` = 더 덮을 점(그은 꺾임점) · `center` = 위치 칸 기준점(둘레 ±1 km). */
 export const fetchTerrainExtent = (
   projectId: string,
   marginM: number | null,
   extra: Vertex[],
+  center: { lon: number; lat: number } | null = null,
 ): Promise<TerrainExtent> => {
   const query = new URLSearchParams();
   if (marginM !== null) query.set("margin_m", String(marginM));
   if (extra.length) query.set("extra", JSON.stringify(extra));
+  if (center) {
+    query.set("lon", String(center.lon));
+    query.set("lat", String(center.lat));
+  }
   return request(`${base(projectId)}/terrain/extent?${query}`);
 };
 
+/** 받기 — 칸 상한 넘음 = 오류(413 글). */
 export const startTerrain = (
   projectId: string,
   marginM: number | null,
   extra: Vertex[],
-): Promise<PlanJob> =>
+  aim: TerrainAim = {},
+): Promise<TerrainStart> =>
   request(`${base(projectId)}/terrain`, {
     method: "POST",
-    body: JSON.stringify({ margin_m: marginM, extra }),
+    body: JSON.stringify({ margin_m: marginM, extra, ...aim }),
   });
 
 export const fetchPlanProgress = (projectId: string): Promise<{ terrain: PlanJob }> =>
@@ -206,12 +230,15 @@ export const fetchPlanZeroOrder = (projectId: string): Promise<PlanZeroOrder> =>
 export const fetchPlanBands = (projectId: string): Promise<{ stream: PlanBand; road: PlanBand }> =>
   request(`${base(projectId)}/layers/bands`);
 
-/** 올린 임도망도(584-3) — 받은 범위로 자른 선 + 가닥 · 길이 m · 원본 선 수 · 좌표계 · 올린 파일. */
+/** 올린 임도망도(584-3 · 595-4) — 기준점 ±5 km ∪ 받은 지형으로 자른 선 + 가닥 · 길이 m · 원본 선 수 ·
+ *  좌표계 · 잘린 상자([서, 남, 동, 북] 경위도) · 기준점 ~ 가장 가까운 임도 m · 올린 파일. */
 export interface PlanForestRoads {
   count: number;
   length_m: number;
   total: number;
   crs: string[];
+  bounds: [number, number, number, number] | null;
+  distance_m: number | null;
   source: { files: string[]; bytes: number; uploaded_at: string } | null;
   collection: GeoJsonCollection;
 }

D03_PlanAnalysis/D03_PlanAnalysis_ForestRoad.py

@@ -1,12 +1,13 @@
-"""D 노선분석 지도 올린 임도망도 겹(PLAN 584-3 · 요구 443).
+"""D 노선분석 지도 올린 임도망도 겹(PLAN 584-3 · 595-4 · 요구 443 · 463).
 
 - 출처 = 사용자가 받은 산림청 임도망도 SHP(FGIS 무료신청 · 산림빅데이터거래소 · 시군구 단위 · 선)
-- 올림 = zip 하나 또는 shp · shx · dbf(· prj · cpg) 묶음 · 합계 `MAX_BYTES` · 읽어 보고 받은 범위 안
-  선이 있어야 받음(아니면 옛 것 그대로 · 까닭 = `ValueError`)
+- 올림 = zip 하나 또는 shp · shx · dbf(· prj · cpg) 묶음 · 합계 `MAX_BYTES` · 읽어 보고 자르는 범위
+  안 선이 있어야 받음(아니면 옛 것 그대로 · 까닭 = `ValueError` + 기준점 ~ 가장 가까운 임도 거리)
+- 자르는 범위 = 기준점 ± `RADIUS_M` ∪ 받은 지형(`clip_box` · 지형 받기 전에도 기준점만으로)
 - 원본 = 프로젝트 `D03_PlanAnalysis/임도망도.zip`(낱 파일은 zip 으로 묶음) +
   `임도망도.json`(파일 이름 · 올린 때)
-- 좌표계 = prj · 없으면 EPSG:5179 · dbf 글자 = cpg · 없으면 cp949 · 받은 범위로 잘라 가닥 · 길이 →
-  파생 저장본(열쇠 = 원본 지문 · 범위)
+- 좌표계 = prj · 없으면 EPSG:5179 · dbf 글자 = cpg · 없으면 cp949 · 자르는 범위로 잘라 가닥 · 길이 ·
+  잘린 상자(`bounds` 경위도) · 기준점 거리 → 파생 저장본(열쇠 = 원본 지문 · 범위 · 기준점)
 
 계산 자리 = ③ 서버 단독.
 """
@@ -32,6 +33,7 @@ from common_util.common_util_json import atomic_write_json
 from common_util.common_util_time import utc_now_text
 
 Box = tuple[float, float, float, float]
+XY = tuple[float, float]
 
 FOLDER = "D03_PlanAnalysis"
 SOURCE_ZIP = "임도망도.zip"
@@ -41,7 +43,17 @@ DBF_ENCODING = "cp949"
 MAX_BYTES = 100 * 1024 * 1024
 SHP_PARTS = (".shp", ".shx", ".dbf", ".prj", ".cpg")
 ROAD_KIND = "plan_forest_road"
-ROAD_VERSION = 1
+ROAD_VERSION = 2  # 잘린 상자 · 기준점 거리
+RADIUS_M = 5000.0  # 기준점 둘레(요구서_463 3장 18)
+
+
+def clip_box(origin: XY, have: Box | None) -> Box:
+    """자르는 범위 = 기준점 ± `RADIUS_M` ∪ 받은 지형."""
+    x, y = origin
+    box = (x - RADIUS_M, y - RADIUS_M, x + RADIUS_M, y + RADIUS_M)
+    if have is None:
+        return box
+    return min(box[0], have[0]), min(box[1], have[1]), max(box[2], have[2]), max(box[3], have[3])
 
 
 def source_dir(project_root: Path) -> Path:
@@ -100,18 +112,22 @@ def _plain(value: Any) -> Any:
     return value if isinstance(value, (str, int, float, bool)) or value is None else str(value)
 
 
-def clip_roads(zip_path: Path, box: Box, epsg: int) -> dict[str, Any]:
-    """SHP 선 → 받은 범위로 잘라 경위도 선(원 속성 + `length_m`) · 가닥 · 길이 · 원본 선 수 ·
-    좌표계."""
+def clip_roads(zip_path: Path, box: Box, epsg: int, origin: XY | None = None) -> dict[str, Any]:
+    """SHP 선 → 범위로 잘라 경위도 선(원 속성 + `length_m`) · 가닥 · 길이 · 원본 선 수 · 좌표계 ·
+    잘린 상자(`bounds` [서, 남, 동, 북] · 없으면 None) · 기준점 ~ 가장 가까운 임도(`distance_m` ·
+    자르기 전 모든 선)."""
     from pyproj import Transformer
+    from shapely.geometry import Point, mapping
     from shapely.geometry import box as rect
-    from shapely.geometry import mapping
     from shapely.ops import transform
 
     frame = rect(*box)
+    here = Point(origin) if origin else None
     to_ll = Transformer.from_crs(f"EPSG:{epsg}", "EPSG:4326", always_xy=True).transform
     features: list[dict[str, Any]] = []
     total, length, crs = 0, 0.0, []
+    near: float | None = None
+    edges: list[tuple[float, float, float, float]] = []
     for shapes in read_shapes(zip_path):
         crs.append(shapes.crs.to_string())
         shapes = shapes[shapes.geometry.notna()]
@@ -121,19 +137,18 @@ def clip_roads(zip_path: Path, box: Box, epsg: int) -> dict[str, Any]:
         for (_, row), geometry in zip(
             shapes.drop(columns=shapes.geometry.name).iterrows(), metric.geometry
         ):
+            if here is not None:
+                gap = geometry.distance(here)
+                near = gap if near is None else min(near, gap)
             part = geometry.intersection(frame)
             if part.is_empty or part.length <= 0:
                 continue
             length += part.length
             props = {str(k): _plain(v) for k, v in row.items()}
             props["length_m"] = round(part.length, 1)
-            features.append(
-                {
-                    "type": "Feature",
-                    "geometry": mapping(transform(to_ll, part)),
-                    "properties": props,
-                }
-            )
+            lonlat = transform(to_ll, part)
+            edges.append(lonlat.bounds)
+            features.append({"type": "Feature", "geometry": mapping(lonlat), "properties": props})
     if not total:
         raise ValueError("SHP 에 선 지물이 없습니다 — 임도망도(선) 파일인지 보세요.")
     return {
@@ -141,23 +156,34 @@ def clip_roads(zip_path: Path, box: Box, epsg: int) -> dict[str, Any]:
         "length_m": round(length, 1),
         "total": total,
         "crs": sorted(set(crs)),
+        "bounds": [
+            min(e[0] for e in edges),
+            min(e[1] for e in edges),
+            max(e[2] for e in edges),
+            max(e[3] for e in edges),
+        ]
+        if edges
+        else None,
+        "distance_m": None if near is None else round(near, 1),
         "collection": {"type": "FeatureCollection", "features": features},
     }
 
 
-def road_layer(project_root: Path, box: Box, epsg: int) -> dict[str, Any]:
-    """올린 임도망도 — 원본 지문 · 범위 같으면 저장본 · 올린 적 없으면 `FileNotFoundError`."""
+def road_layer(project_root: Path, box: Box, epsg: int, origin: XY | None = None) -> dict[str, Any]:
+    """올린 임도망도 — 원본 지문 · 범위 · 기준점 같으면 저장본 · 올린 적 없으면
+    `FileNotFoundError`."""
     path = source_dir(project_root) / SOURCE_ZIP
     stamp = file_stamp(path)
     if stamp is None:
         raise FileNotFoundError(
-            "올린 임도망도가 없습니다 — 옆 올리기 단추로 시군구 SHP 를 올리세요."
+            "올린 임도망도가 없습니다 — [임도망 올리기]로 시군구 SHP 를 올리세요."
         )
-    key = derived_key(ROAD_KIND, ROAD_VERSION, stamp, int(epsg), [round(v, 1) for v in box])
+    at = [round(v, 1) for v in origin] if origin else None
+    key = derived_key(ROAD_KIND, ROAD_VERSION, stamp, int(epsg), [round(v, 1) for v in box], at)
     saved = read_derived(project_root, ROAD_KIND, key)
     if saved is not None:
         return saved["value"]
-    value = clip_roads(path, box, epsg)
+    value = clip_roads(path, box, epsg, origin)
     try:
         value["source"] = json.loads((source_dir(project_root) / SOURCE_META).read_text("utf-8"))
     except (OSError, ValueError):
@@ -167,10 +193,14 @@ def road_layer(project_root: Path, box: Box, epsg: int) -> dict[str, Any]:
 
 
 def save_roads(
-    project_root: Path, files: list[tuple[str, bytes]], box: Box, epsg: int
+    project_root: Path,
+    files: list[tuple[str, bytes]],
+    box: Box,
+    epsg: int,
+    origin: XY | None = None,
 ) -> dict[str, Any]:
-    """올림 — 묶어 읽어 보고 받은 범위 안 선이 있으면 원본을 바꿈 · 아니면 `ValueError`
-    (옛 것 그대로)."""
+    """올림 — 묶어 읽어 보고 자르는 범위 안 선이 있으면 원본을 바꿈 · 아니면 `ValueError`
+    (옛 것 그대로 · 기준점 ~ 가장 가까운 임도 거리)."""
     data = pack(files)
     folder = source_dir(project_root)
     folder.mkdir(parents=True, exist_ok=True)
@@ -178,11 +208,13 @@ def save_roads(
         temporary.write(data)
     trial = Path(temporary.name)
     try:
-        got = clip_roads(trial, box, epsg)
+        got = clip_roads(trial, box, epsg, origin)
         if not got["count"]:
+            gap = got["distance_m"]
+            where = f" — 가장 가까운 임도 {gap / 1000:.1f} km" if gap is not None else ""
             raise ValueError(
-                f"받은 범위 안에 임도가 없습니다(파일 선 {got['total']}가닥) — "
-                "사업지 시군구 파일인지 보세요."
+                f"임도가 기준점 ±{RADIUS_M / 1000:.0f} km · 받은 지형 밖뿐입니다{where}"
+                f"(파일 선 {got['total']}가닥) — 사업지 시군구 파일인지 보세요."
             )
         os.replace(trial, folder / SOURCE_ZIP)
     finally:
@@ -195,4 +227,4 @@ def save_roads(
             "uploaded_at": utc_now_text(),
         },
     )
-    return road_layer(project_root, box, epsg)
+    return road_layer(project_root, box, epsg, origin)

D03_PlanAnalysis/D03_PlanAnalysis_Layers.py

@@ -179,31 +179,37 @@ def _covers(have: Any, box: Box) -> bool:
 
 
 def ensure_cadastral(processed: Path, box: Box, epsg: int) -> dict[str, Any]:
-    """지적도 — 열쇠 상자가 덮고 실패 조각 없으면 그대로 · 아니면 상자 통째로 다시 받음."""
+    """지적도 — 열쇠 상자가 덮고 실패 조각 없으면 그대로 · 덮는데 실패 조각이 남았으면 그 조각만
+    다시(PLAN 595-1) · 못 덮으면 상자 통째로 다시 받음."""
     path, key_path = processed / CADASTRAL_FILE, processed / CADASTRAL_KEY
     key = _read_json(key_path) or {}
-    if path.is_file() and not key.get("failed") and _covers(key.get("box"), box):
+    covered = path.is_file() and _covers(key.get("box"), box)
+    if covered and not key.get("failed"):
         return {"path": path, "fetched": False, "count": key.get("count"), "failed": 0}
-    parts = tiles(box)
+    retry = key.get("failed_tiles") if covered else None  # 옛 열쇠(조각 기록 없음) = 통째로
+    old = ((_read_json(path) or {}).get("features") or []) if retry else []
+    frame = key["box"] if retry else [float(v) for v in box]
+    parts = [tuple(t) for t in retry] if retry else tiles(box)
     with ThreadPoolExecutor(max_workers=WFS_WORKERS) as pool:
         results = list(pool.map(lambda part: fetch_tile(part, epsg), parts))
-    failed = sum(bad for _, bad in results)
-    if failed == len(parts):
-        raise RuntimeError(f"지적도를 받지 못했습니다 — 브이월드 WFS 조각 {failed}개 실패")
-    features = dedup_parcels([f for got, _ in results for f in got])
+    bad = [list(part) for part, (_, n) in zip(parts, results) if n]
+    if len(bad) == len(parts) and not old:
+        raise RuntimeError(f"지적도를 받지 못했습니다 — 브이월드 WFS 조각 {len(bad)}개 실패")
+    features = dedup_parcels([*old, *(f for got, _ in results for f in got)])
     processed.mkdir(parents=True, exist_ok=True)
     atomic_write_json(path, {"type": "FeatureCollection", "features": features})
     atomic_write_json(
         key_path,
         {
-            "box": [float(v) for v in box],
+            "box": frame,
             "epsg": int(epsg),
             "count": len(features),
-            "failed": failed,
+            "failed": len(bad),
+            "failed_tiles": bad,
         },
     )
-    logger.info("D 지적도 받기: 조각 %d · 실패 %d · 필지 %d", len(parts), failed, len(features))
-    return {"path": path, "fetched": True, "count": len(features), "failed": failed}
+    logger.info("D 지적도 받기: 조각 %d · 실패 %d · 필지 %d", len(parts), len(bad), len(features))
+    return {"path": path, "fetched": True, "count": len(features), "failed": len(bad)}
 
 
 # ── 급경사 ──────────────────────────────────────────────

D03_PlanAnalysis/D03_PlanAnalysis_Router.py

@@ -11,6 +11,10 @@
 
 범위 = 기준점 둘레(`SITE_RADIUS_M`) + 지점 + 그은 점(`extra` · 지도 보기 귀) 상자 + 여유 ∪ 받은 범위
 (PLAN 591-2 · 기준점 = 처음 받는 기준일 뿐).
+위치 바꿈(PLAN 595 · 꼴 = PLAN 595 절) = 몸 `center`(위치 칸 캐시 기준점) 둘레를 더함 ·
+받은 범위에서 `FAR_M` 넘음 = 「far」 물음 · 다른 TM 띠 = 「band」 물음 → `reset` = 옛 범위 버리고
+center 둘레만 · 칸 `MAX_CELLS` 넘음 = 413 · 진행은 `models_plan/plan_job_<일>.json` 에도 적어
+서버가 다시 켜지면 「끊김」(interrupted).
 노선 정본(손 그은 선)은
 `D03_PlanAnalysis_Network`(PLAN 562-2 · 옛 추천 엔진 · 꺾임점 R 걷음).
 """
@@ -28,8 +32,9 @@ from fastapi import APIRouter, File, HTTPException, Request, Response, UploadFil
 from pydantic import BaseModel
 
 from B03_FileInput.B03_FileInput_Repository import get_project_storage_relative_path
-from common_util.common_util_crs import lonlat_to_tm
+from common_util.common_util_crs import lonlat_to_tm, tm_epsg_for_lon
 from common_util.common_util_http_cache import cached_file_response
+from common_util.common_util_json import atomic_write_json
 from common_util.common_util_storage import resolve_stored_project_path
 from config.config_db import get_db_pool
 from D03_PlanAnalysis import D03_PlanAnalysis_ForestRoad as forest_road
@@ -50,26 +55,53 @@ _TASKS: set[asyncio.Task] = set()
 KINDS = ("terrain", "analyze")
 
 
+class Center(BaseModel):
+    lon: float
+    lat: float
+
+
 class TerrainBody(BaseModel):
-    """지형 받기 — 여유(m · 빈 값 = 처음 값) · 더 덮을 점(그은 꺾임점 · 작업 좌표계 m)."""
+    """지형 받기 — 여유(m · 빈 값 = 처음 값) · 더 덮을 점(그은 꺾임점 · 작업 좌표계 m) ·
+    위치 칸 가운데(경위도 · 595) · 새로(옛 받은 범위 버림)."""
 
     margin_m: float | None = None
     extra: list[tuple[float, float]] = []
+    center: Center | None = None
+    reset: bool = False
 
 
-async def _plan_ends(project_id: UUID) -> tuple[Path, int, list[tuple[float, float]]]:
-    """(실경로, 좌표계, [지점들]) — 설계 내용 파일 · 지점 · 선이 없으면 등록 기준점 둘레 두 귀."""
+Ends = tuple[Path, int, list[tuple[float, float]], int | None]
+
+
+async def _plan_ends(
+    project_id: UUID, center: tuple[float, float] | None = None, reset: bool = False
+) -> Ends:
+    """(실경로, 좌표계, [지점들], 띠 넘음 epsg) — 설계 내용 파일 · 지점 · 선이 없으면 등록 기준점
+    둘레 두 귀 · `center`(위치 칸) = 그 둘레를 더함(`reset` = 둘레만) · 다른 TM 띠 = 확인 전이면
+    넷째 값에 그 띠(셋은 `center` 없이) · 확인 뒤(받은 지형이 그 띠) = 그 띠 · 둘레만."""
     root, plan = await store.load_plan(project_id)  # 없음 404 · 예정노선 아님 400
     site = (await network.project_info(project_id))["site"]
     epsg = network.work_epsg(plan, site)
+    meta = terrain.received_meta(root)
+    if center is None and meta.get("center") and meta.get("epsg") not in (None, epsg):
+        center = tuple(meta["center"])  # 띠 넘어 받은 뒤 [저장] 전 — 받은 지형 띠 · 가운데
+    ends = [(float(p["x"]), float(p["y"])) for p in plan["points"]]
+    if center is not None:
+        band = tm_epsg_for_lon(center[0])
+        if epsg is not None and band != epsg:
+            if not (reset or meta.get("epsg") == band):
+                return (*(await _plan_ends(project_id))[:3], band)
+            epsg, ends, reset = band, [], True  # 옛 띠 지점 · 선 = 「노선 지움」 확인으로 빠짐
+        epsg = epsg or band
+        corners = terrain.site_corners(*lonlat_to_tm(center[0], center[1], epsg))
+        return root, epsg, corners if reset else [*ends, *corners], None
     if epsg is None:
         raise HTTPException(status_code=400, detail="기준점 · 지점이 없습니다.")
-    ends = [(float(p["x"]), float(p["y"])) for p in plan["points"]]
     if site and not ends and not plan["lines"]:  # 처음 = 기준점 둘레 · 그 뒤는 받은 범위가 덮음
         ends = terrain.site_corners(*lonlat_to_tm(site["lon"], site["lat"], epsg))
     if not ends:
         raise HTTPException(status_code=400, detail="기준점 · 지점이 없습니다.")
-    return root, epsg, ends
+    return root, epsg, ends, None
 
 
 def _query_xy(extra: str | None) -> list[tuple[float, float]]:
@@ -81,15 +113,49 @@ def _query_xy(extra: str | None) -> list[tuple[float, float]]:
         raise HTTPException(status_code=400, detail="extra 꼴이 다릅니다.") from error
 
 
-def _start_job(project_id: UUID, kind: str, work: Any) -> dict[str, Any]:
+IDLE = {"running": False, "percent": 100, "message": None, "error": None, "interrupted": False}
+
+
+def _job_path(root: Path, kind: str) -> Path:
+    return terrain.models_dir_of(root) / f"plan_job_{kind}.json"
+
+
+def _note_job(path: Path | None, job: dict[str, Any]) -> None:
+    """진행 기록 — 서버가 다시 켜져도 「돌던 일」을 알게(실패해도 일은 계속)."""
+    if path is None:
+        return
+    try:
+        path.parent.mkdir(parents=True, exist_ok=True)
+        atomic_write_json(path, job)
+    except OSError as error:
+        logger.warning("D 진행 기록 실패: %s", error)
+
+
+def _saved_job(root: Path, kind: str) -> dict[str, Any]:
+    """기억에 없는 일 — 기록이 「도는 중」이면 서버가 다시 켜져 끊김(percent < 100 ·
+    message 없음)."""
+    try:
+        saved = json.loads(_job_path(root, kind).read_text(encoding="utf-8"))
+    except (OSError, ValueError):
+        return dict(IDLE)
+    if not (isinstance(saved, dict) and saved.get("running")):
+        return dict(IDLE)
+    percent = min(99, int(saved.get("percent") or 0))
+    return {**IDLE, "percent": percent, "interrupted": True}
+
+
+def _start_job(project_id: UUID, kind: str, work: Any, root: Path | None = None) -> 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}
+    job = {"running": True, "percent": 0, "message": "시작", "error": None, "interrupted": False}
     _JOBS[key] = job
+    path = _job_path(root, kind) if root is not None else None
+    _note_job(path, job)
 
     def report(percent: int, message: str) -> None:
         job.update(percent=int(percent), message=message)
+        _note_job(path, job)
 
     async def run() -> None:
         try:
@@ -100,6 +166,7 @@ def _start_job(project_id: UUID, kind: str, work: Any) -> dict[str, Any]:
             job["error"] = str(getattr(error, "detail", None) or error)
         finally:
             job["running"] = False
+            _note_job(path, job)
 
     task = asyncio.create_task(run(), name=f"D {kind} {project_id}")
     _TASKS.add(task)
@@ -133,10 +200,16 @@ async def _save_models(project_id: UUID, models: list[dict[str, Any]], epsg: int
 
 @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
+    project_id: UUID,
+    margin_m: float | None = None,
+    extra: str | None = None,
+    lon: float | None = None,
+    lat: float | None = None,
 ) -> dict[str, Any]:
-    """받기 전 범위 — 지점(+ `extra` 그은 점 `[[x, y]]`) 상자 + 여유 ∪ 받은 범위."""
-    root, epsg, ends = await _plan_ends(project_id)
+    """받기 전 범위 — 지점(+ `extra` 그은 점 `[[x, y]]` · `lon` `lat` 위치 칸 둘레) 상자 + 여유 ∪
+    받은 범위."""
+    center = (lon, lat) if lon is not None and lat is not None else None
+    root, epsg, ends, _ = await _plan_ends(project_id, center)
     view = await asyncio.to_thread(
         terrain.plan_extent, root, [*ends, *_query_xy(extra)], epsg, margin_m
     )
@@ -145,34 +218,73 @@ async def plan_terrain_extent(
 
 @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`."""
+    """사업 전체 수치지형도 · 위성 · DEM 받기(뒤에서) — 진행은 `GET …/plan/progress` ·
+    위치 칸 가운데가 멀거나 다른 띠 = 물음(`far` · `band`) · 칸 상한 넘음 = 413."""
     body = body or TerrainBody()
-    root, epsg, ends = await _plan_ends(project_id)
+    center = (body.center.lon, body.center.lat) if body.center else None
+    root, epsg, ends, band = await _plan_ends(project_id, center, body.reset)
+    if band is not None:
+        return {
+            "status": "band",
+            "epsg": epsg,
+            "center_epsg": band,
+            "message": "다른 좌표 띠로 옮깁니다 — 그은 지점 · 선을 지우고 새로 받을까요?",
+        }
+    have = None if body.reset else terrain.received_box(root, epsg)
+    if center is not None and have is not None:
+        far = terrain.distance_to_box(lonlat_to_tm(center[0], center[1], epsg), have)
+        if far > terrain.FAR_M:
+            return {
+                "status": "far",
+                "distance_m": round(far),
+                "message": f"받은 지형에서 {far / 1000:.1f} km 떨어졌습니다 — "
+                "옛 받은 범위를 버리고 새로 받을까요?",
+            }
     points = [*ends, *body.extra]
+    cells = terrain.cells_of(terrain.wanted_box(points, terrain.clamp_margin(body.margin_m), have))
+    if cells > terrain.MAX_CELLS:
+        raise HTTPException(
+            status_code=413,
+            detail=f"받을 범위가 너무 큽니다 — 칸 {cells / 1e4:.0f}만 · "
+            f"상한 {terrain.MAX_CELLS / 1e4:.0f}만(지점 · 선을 모으거나 위치를 옮겨 새로 받으세요)",
+        )
 
     async def work(report: Any) -> None:
         got = await asyncio.to_thread(
-            terrain.ensure_plan_terrain, root, points, epsg, body.margin_m, report
+            terrain.ensure_plan_terrain,
+            root,
+            points,
+            epsg,
+            body.margin_m,
+            report,
+            reset=have is None,
+            center=center,
         )
         await _save_models(project_id, got["models"], epsg)
         report(100, f"지형 받기 끝 — 새 도엽 {len(got['sheets']['downloaded'])}매")
 
-    return _start_job(project_id, "terrain", work)
+    return _start_job(project_id, "terrain", work, root)
 
 
 @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},
-    }
+    """일마다 진행 — 기억에 없으면 기록을 봄(돌던 기록 = 끊김 · 없으면 percent 100)."""
+    jobs: dict[str, Any] = {}
+    for kind in KINDS:
+        job = _JOBS.get((str(project_id), kind))
+        if job is None:
+            try:
+                job = await asyncio.to_thread(_saved_job, await _tile_root(project_id), kind)
+            except HTTPException:
+                job = IDLE
+        jobs[kind] = dict(job)
+    return {"status": "success", **jobs}
 
 
 async def _layer_root(project_id: UUID) -> tuple[Path, int, terrain.Box]:
     """(실경로, 좌표계, 받은 범위) — 지형 받기 전이면 404."""
-    root, epsg, _ = await _plan_ends(project_id)
-    box = terrain.received_box(root)
+    root, epsg, _, _ = await _plan_ends(project_id)
+    box = terrain.received_box(root, epsg)
     if box is None:
         raise HTTPException(status_code=404, detail="지형 받기 뒤 보입니다.")
     return root, epsg, box
@@ -301,12 +413,30 @@ async def plan_layer_zero_order(
     return {"status": "success", **value}
 
 
+async def _road_frame(project_id: UUID) -> tuple[Path, int, terrain.Box, tuple[float, float]]:
+    """(실경로, 좌표계, 자르는 범위, 기준점 m) — 기준점 = 위치 칸으로 받은 가운데 · 등록 기준점 ·
+    지점 가운데 차례 · 지형 받기 전에도(595-4)."""
+    root, epsg, ends, _ = await _plan_ends(project_id)
+    center = terrain.received_meta(root).get("center")
+    site = (await network.project_info(project_id))["site"]
+    if center:
+        origin = lonlat_to_tm(float(center[0]), float(center[1]), epsg)
+    elif site:
+        origin = lonlat_to_tm(site["lon"], site["lat"], epsg)
+    else:
+        xs, ys = [p[0] for p in ends], [p[1] for p in ends]
+        origin = ((min(xs) + max(xs)) / 2, (min(ys) + max(ys)) / 2)
+    box = forest_road.clip_box(origin, terrain.received_box(root, epsg))
+    return root, epsg, box, origin
+
+
 @router.get("/{project_id}/plan/layers/forest-roads", response_model=None)
 async def plan_layer_forest_roads(project_id: UUID) -> dict[str, Any]:
-    """올린 임도망도(받은 범위로 자른 선 · 가닥 · 길이) — 올린 적 없으면 404."""
-    root, epsg, box = await _layer_root(project_id)
+    """올린 임도망도(기준점 ± 5 km ∪ 받은 지형으로 자른 선 · 가닥 · 길이 · 잘린 상자 · 기준점
+    거리) — 올린 적 없으면 404."""
+    root, epsg, box, origin = await _road_frame(project_id)
     try:
-        value = await asyncio.to_thread(forest_road.road_layer, root, box, epsg)
+        value = await asyncio.to_thread(forest_road.road_layer, root, box, epsg, origin)
     except FileNotFoundError as error:
         raise HTTPException(status_code=404, detail=str(error)) from error
     return {"status": "success", **value}
@@ -317,8 +447,8 @@ async def plan_layer_forest_roads_upload(
     project_id: UUID, files: list[UploadFile] = File(...)
 ) -> dict[str, Any]:
     """임도망도 SHP 올리기(zip 하나 · shp · shx · dbf · prj 묶음 · 합계 100 MB) — 못 읽거나
-    받은 범위 안 선이 없으면 400(옛 것 그대로) · 넘치면 413."""
-    root, epsg, box = await _layer_root(project_id)
+    자르는 범위 안 선이 없으면 400(옛 것 그대로 · 가장 가까운 임도 거리) · 넘치면 413."""
+    root, epsg, box, origin = await _road_frame(project_id)
     got: list[tuple[str, bytes]] = []
     total = 0
     for upload in files:
@@ -335,7 +465,7 @@ async def plan_layer_forest_roads_upload(
     if not total:
         raise HTTPException(status_code=400, detail="빈 파일입니다.")
     try:
-        value = await asyncio.to_thread(forest_road.save_roads, root, got, box, epsg)
+        value = await asyncio.to_thread(forest_road.save_roads, root, got, box, epsg, origin)
     except ValueError as error:
         raise HTTPException(status_code=400, detail=str(error)) from error
     return {"status": "success", **value}

D03_PlanAnalysis/D03_PlanAnalysis_Terrain.py

@@ -1,13 +1,17 @@
-"""D 예정노선 지형 받기 — 사업 전체 한 벌(도엽 · 위성 · DEM) (요구 345 · 389 · PLAN 530-1 · 561-2).
+"""D 예정노선 지형 받기 — 사업 전체 한 벌(도엽 · 위성 · DEM) (요구 345 · 389 · 463 · PLAN 530-1 ·
+561-2 · 595).
 
 지형은 D 프로젝트 storage 안 `B04_PreProcess/` 배치 그대로(지형 코드 주인 = B04 · 고칠 곳 0):
 
-- 범위 = 전체 시점 · 종점(+ 그은 점) 상자 + 여유 · 받은 범위가 있으면 그 범위와 합침(줄지 않음)
+- 범위 = 전체 시점 · 종점(+ 그은 점) 상자 + 여유 · 받은 범위가 있으면 그 범위와 합침(줄지 않음) ·
+  새로(`reset` · 위치를 멀리 옮김) = 옛 받은 범위 · 병합 도엽 · 위성 범위를 버리고 새 상자만 ·
+  칸 상한 `MAX_CELLS`(넘으면 라우터가 거절)
 - 도엽 = 범위 상자에 닿는 도엽 모두 → `ensure_sheets`(없는 것만 · prune 없음)
 - 병합 = 지난번 병합 도엽 ∪ 이번 도엽(옛 도엽 유지 · 열쇠 같으면 건너뜀)
 - 위성 = 범위 도엽 도곽 ∪ 옛 위성 범위(덮으면 건너뜀)
-- 지적도 = 범위 상자 · 브이월드 WFS 1 km 조각(`D03_PlanAnalysis_Layers` · 실패해도 계속)
+- 지적도 = 지형 일에서 뺌 — 겹을 켤 때 GET 이 받음(`D03_PlanAnalysis_Layers.ensure_cadastral`)
 - DEM = 한 벌 `models_plan/`(연도는 D03 나눔 뒤에 정해짐) · 받은 범위 = `plan_terrain.json`
+  (상자 · 여유 · 좌표계 · 위치 칸 가운데)
 - 작업 좌표계 = 지점 epsg · `B03_FileInput/input/prj/plan.prj` 한 장(B04 · B05 좌표계 창구가 읽음)
 - 받기 전 범위 보기 = `plan_extent`(내려받기 없음) · 받기는 사용자가 누를 때만
 
@@ -43,7 +47,6 @@ from config.config_system_terrain import (
     SHEET_SURFACE_DEFAULT_METHOD,
     SHEET_SURFACE_MARGIN_M,
 )
-from D03_PlanAnalysis import D03_PlanAnalysis_Layers as layers
 
 logger = logging.getLogger(__name__)
 
@@ -52,6 +55,9 @@ DEFAULT_MARGIN_M = 350.0
 MAX_MARGIN_M = 5000.0
 # 처음 받는 범위 = 등록 기준점 둘레 ± 1 km(요구 453 · PLAN 591-2)
 SITE_RADIUS_M = 1000.0
+# 위치 바꿈이 받은 범위에서 이만큼 넘게 멀면 묻고 새로 · 받을 DEM 칸 상한(요구서_463 3장 4)
+FAR_M = 5000.0
+MAX_CELLS = 5_000_000
 STAGE = Path("B04_PreProcess")
 PLAN_MODELS = "models_plan"
 TERRAIN_META = "plan_terrain.json"
@@ -74,9 +80,16 @@ def clamp_margin(margin_m: float | None) -> float:
     return min(MAX_MARGIN_M, max(SHEET_SURFACE_MARGIN_M, float(margin_m)))
 
 
-def received_box(project_root: Path) -> Box | None:
-    """받은 DEM 범위 — 받은 적 없거나 읽지 못하면 None."""
-    meta = _read_json(models_dir_of(project_root) / TERRAIN_META) or {}
+def received_meta(project_root: Path) -> dict[str, Any]:
+    """받은 범위 기록(`box` · `margin_m` · `epsg` · `center`) — 없으면 빈 dict."""
+    return _read_json(models_dir_of(project_root) / TERRAIN_META) or {}
+
+
+def received_box(project_root: Path, epsg: int | None = None) -> Box | None:
+    """받은 DEM 범위 — 받은 적 없거나 읽지 못하거나 다른 좌표계(`epsg` 줄 때)면 None."""
+    meta = received_meta(project_root)
+    if epsg is not None and meta.get("epsg") not in (None, int(epsg)):
+        return None
     try:
         x0, y0, x1, y1 = (float(v) for v in meta["box"])
     except (KeyError, TypeError, ValueError):
@@ -99,6 +112,18 @@ def wanted_box(points: list[XY], margin_m: float, have: Box | None) -> Box:
     return min(box[0], have[0]), min(box[1], have[1]), max(box[2], have[2]), max(box[3], have[3])
 
 
+def cells_of(box: Box) -> int:
+    """DEM 칸 수 어림 — 받기 시간 · 메모리 가늠 · 상한 판정."""
+    return int((box[2] - box[0]) * (box[3] - box[1]) / PLAN_TERRAIN_GRID_M**2)
+
+
+def distance_to_box(xy: XY, box: Box) -> float:
+    """점 ~ 상자 거리(m · 안이면 0)."""
+    dx = max(box[0] - xy[0], 0.0, xy[0] - box[2])
+    dy = max(box[1] - xy[1], 0.0, xy[1] - box[3])
+    return math.hypot(dx, dy)
+
+
 def covers(have: Box | None, box: Box) -> bool:
     return (
         have is not None
@@ -135,7 +160,7 @@ def plan_extent(
 ) -> dict[str, Any]:
     """받기 전 범위 보기 — 내려받기 없음. `points` = 전체 시점 · 종점(+ 그은 점 · 작업 좌표계 m)."""
     margin = clamp_margin(margin_m)
-    have = received_box(project_root)
+    have = received_box(project_root, epsg)
     box = wanted_box(points, margin, have)
     store = get_project_map_sheets_dir(project_root)
     sheets = []
@@ -157,8 +182,8 @@ def plan_extent(
         "have": box_ring(have, epsg) if have else None,
         "covered": covers(have, box),
         "sheets": sheets,
-        # DEM 칸 수 어림 — 받기 시간 · 메모리 가늠
-        "cells": int((box[2] - box[0]) * (box[3] - box[1]) / PLAN_TERRAIN_GRID_M**2),
+        "cells": cells_of(box),
+        "max_cells": MAX_CELLS,
     }
 
 
@@ -181,10 +206,12 @@ def _read_json(path: Path) -> dict[str, Any] | None:
         return None
 
 
-def _merge_sheets(processed: Path, store: Path, available: list[str]) -> list[str]:
-    """병합 = 지난번 병합 도엽(zip 남은 것) ∪ 이번 도엽 · 열쇠 같으면 건너뜀."""
+def _merge_sheets(
+    processed: Path, store: Path, available: list[str], keep_old: bool = True
+) -> list[str]:
+    """병합 = 지난번 병합 도엽(zip 남은 것 · 새로면 뺌) ∪ 이번 도엽 · 열쇠 같으면 건너뜀."""
     key_path = processed / "sheet_merge.key.json"
-    old = (_read_json(key_path) or {}).get("sheets") or []
+    old = ((_read_json(key_path) or {}).get("sheets") or []) if keep_old else []
     sheets = sorted({*available, *(s for s in old if get_sheet_path(store, s) is not None)})
     key = {"sheets": sheets, "parser": PARSER_VERSION}
     if sheets and (_read_json(key_path) != key or not any(processed.glob("도엽_*.geojson"))):
@@ -193,8 +220,15 @@ def _merge_sheets(processed: Path, store: Path, available: list[str]) -> list[st
     return sheets
 
 
-def _satellite(processed: Path, prj: Path, points: list[tuple[float, float]], epsg: int) -> str:
-    """위성 = 기준 도엽 도곽 ∪ 옛 범위 · 이미 덮으면 건너뜀. 돌려줌 = skip · done · failed."""
+def _satellite(
+    processed: Path,
+    prj: Path,
+    points: list[tuple[float, float]],
+    epsg: int,
+    keep_old: bool = True,
+) -> str:
+    """위성 = 기준 도엽 도곽 ∪ 옛 범위(새로면 뺌) · 이미 덮으면 건너뜀. 돌려줌 = skip · done ·
+    failed."""
     from pyproj import Transformer
 
     to_metric = Transformer.from_crs("EPSG:4326", f"EPSG:{epsg}", always_xy=True)
@@ -207,7 +241,7 @@ def _satellite(processed: Path, prj: Path, points: list[tuple[float, float]], ep
             xs.append(x)
             ys.append(y)
     need = {"x": [min(xs), max(xs)], "y": [min(ys), max(ys)]}
-    meta = _read_json(processed / "vworld_satellite_meta.json")
+    meta = _read_json(processed / "vworld_satellite_meta.json") if keep_old else None
     try:
         old = [float(meta[k]) for k in ("x_min", "x_max", "y_min", "y_max")] if meta else None
     except (KeyError, TypeError, ValueError):
@@ -234,8 +268,11 @@ def ensure_plan_terrain(
     epsg: int,
     margin_m: float | None = None,
     on_progress: Progress | None = None,
+    reset: bool = False,
+    center: tuple[float, float] | None = None,
 ) -> dict[str, Any]:
-    """사업 전체 지형 받기 — `points` = 전체 시점 · 종점(+ 그은 점 · 작업 좌표계 m).
+    """사업 전체 지형 받기 — `points` = 전체 시점 · 종점(+ 그은 점 · 작업 좌표계 m) ·
+    `reset` = 옛 받은 범위 버리고 새로 · `center` = 위치 칸 가운데(경위도 · 기록만).
 
     돌려줌 = 도엽(받은 · 실패 · 병합) · 위성 상태 · 받은 범위 · 모델 등록용 dict 목록.
     """
@@ -250,28 +287,24 @@ def ensure_plan_terrain(
     processed.mkdir(parents=True, exist_ok=True)
     prj = write_plan_prj(project_root, epsg)
     margin = clamp_margin(margin_m)
-    box = wanted_box(points, margin, received_box(project_root))
+    have = None if reset else received_box(project_root, epsg)
+    box = wanted_box(points, margin, have)
 
-    report(5, "수치지형도 도엽 고르는 중")
+    report(5, "1/5 수치지형도 도엽 고르는 중")
     wanted = sheets_for_box(box, epsg)
     store = get_project_map_sheets_dir(project_root)
     before = {s for s in wanted if get_sheet_path(store, s) is not None}
-    report(10, f"수치지형도 도엽 {len(wanted) - len(before)}매 받는 중")
+    report(10, f"2/5 수치지형도 도엽 {len(wanted) - len(before)}매 받는 중")
     got = ensure_sheets(store, wanted)
     if not got["available"]:
         raise ValueError(f"수치지형도 도엽을 하나도 받지 못했습니다: {got['failed']}")
-    report(45, "도엽 레이어 병합 중")
-    merged = _merge_sheets(processed, store, list(got["available"]))
-    report(60, "위성 지도 받는 중")
-    satellite = _satellite(processed, prj, [(p[1], p[0]) for p in box_ring(box, epsg)], epsg)
-    report(62, "지적도 받는 중")
-    try:
-        cadastral = layers.ensure_cadastral(processed, box, epsg)["count"]
-    except Exception as exc:  # noqa: BLE001 — 지적도는 겹 · 지형 받기를 막지 않음(켤 때 다시)
-        logger.warning("D 지적도 받기 실패: %s", exc)
-        cadastral = None
+    report(45, "3/5 도엽 레이어 병합 중")
+    merged = _merge_sheets(processed, store, list(got["available"]), keep_old=have is not None)
+    report(60, "4/5 위성 지도 받는 중")
+    ring = [(p[1], p[0]) for p in box_ring(box, epsg)]
+    satellite = _satellite(processed, prj, ring, epsg, keep_old=have is not None)
 
-    report(65, "사업 지형면 만드는 중")
+    report(65, f"5/5 사업 지형면 만드는 중 — 칸 {cells_of(box) / 1e4:.0f}만 · 수 분 걸릴 수 있음")
     models_dir = models_dir_of(project_root)
     models_dir.mkdir(parents=True, exist_ok=True)
     # B 절취 = 넘긴 꼭짓점 상자 + `SHEET_SURFACE_MARGIN_M` → 그만큼 안쪽으로 넘김(여유 ≥ 그 값)
@@ -293,7 +326,11 @@ def ensure_plan_terrain(
             "사업 지형면을 만들지 못했습니다 — 범위 안 도엽 등고선 높이 확인"
             f"(병합 도엽 {len(merged)}매)"
         )
-    atomic_write_json(models_dir / TERRAIN_META, {"box": list(box), "margin_m": margin})
+    meta: dict[str, Any] = {"box": list(box), "margin_m": margin, "epsg": int(epsg)}
+    kept = None if reset else received_meta(project_root).get("center")
+    if center or kept:
+        meta["center"] = list(center or kept)
+    atomic_write_json(models_dir / TERRAIN_META, meta)
     report(100, "지형 받기 끝")
     return {
         "sheets": {
@@ -303,7 +340,6 @@ def ensure_plan_terrain(
             "merged": merged,
         },
         "satellite": satellite,
-        "cadastral": cadastral,
         "box_m": list(box),
         "models": models,
     }

resources/tester/test_530_1_plan_terrain.py

@@ -1,5 +1,6 @@
-"""D 예정노선 지형 받기(PLAN 530-1 · 561-2) — 사업 전체 한 벌 · 범위 보기 = 내려받기 없음 ·
-옛 도엽 유지 · 새 도엽만 · 범위는 줄지 않음 · 빠짐 없음.
+"""D 예정노선 지형 받기(PLAN 530-1 · 561-2 · 595) — 사업 전체 한 벌 · 범위 보기 = 내려받기 없음 ·
+옛 도엽 유지 · 새 도엽만 · 범위는 줄지 않음 · 빠짐 없음 · 지적도 뺌 · 새로(reset) = 옛 범위 버림 ·
+받은 범위 좌표계 · 칸 상한 · 거리.
 
 ./venv/Scripts/python.exe -m pytest resources/tester/test_530_1_plan_terrain.py -q
 """
@@ -54,7 +55,6 @@ def fake(tmp_path, monkeypatch):
     monkeypatch.setattr(terrain, "parse_and_merge_sheets", merge)
     monkeypatch.setattr(terrain, "download_vworld_satellite_map", satellite)
     monkeypatch.setattr(terrain, "build_sheet_surface_model", dem)
-    monkeypatch.setattr(terrain.layers, "ensure_cadastral", lambda *a: {"count": 0})  # 569-1
     return tmp_path, seen
 
 
@@ -104,6 +104,39 @@ def test_one_set_and_extend_keeps_old(fake) -> None:
     assert len(seen["merge"]) == merges and third["box_m"] == grown
 
 
+def test_reset_drops_old_box_sheets_satellite(fake) -> None:
+    """595-2 — 새로 = 옛 받은 범위 · 병합 도엽 · 위성 범위 버림 · 좌표계 · 가운데 기록 ·
+    다른 좌표계로 물으면 받은 범위 없음 · 지적도는 지형 일에 없음."""
+    root, seen = fake
+    terrain.ensure_plan_terrain(root, [A, B], EPSG, center=(127.5, 36.0))
+    old = set(seen["merge"][-1])
+    far = [(A[0] + 40000.0, A[1]), (A[0] + 41000.0, A[1] + 1000.0)]
+    got = terrain.ensure_plan_terrain(root, far, EPSG, reset=True, center=(128.0, 36.0))
+    m = terrain.DEFAULT_MARGIN_M
+    assert got["box_m"] == pytest.approx(
+        [far[0][0] - m, far[0][1] - m, far[1][0] + m, far[1][1] + m]
+    )
+    assert not old & set(seen["merge"][-1])  # 옛 도엽 병합에서 빠짐(zip 은 남음)
+    assert seen["satellite"][-1]["x"][0] > A[0] + 30000  # 옛 위성 범위와 안 합침
+    assert "cadastral" not in got
+    meta = terrain.received_meta(root)
+    assert meta["epsg"] == EPSG and meta["center"] == [128.0, 36.0]
+    assert terrain.received_box(root, EPSG) == pytest.approx(tuple(got["box_m"]))
+    assert terrain.received_box(root, 5187) is None  # 다른 띠 = 받은 범위 아님
+    # 새로 아닌 이어 받기 = 가운데 기록 그대로
+    terrain.ensure_plan_terrain(root, far, EPSG)
+    assert terrain.received_meta(root)["center"] == [128.0, 36.0]
+
+
+def test_cells_distance_and_limit(fake) -> None:
+    root, _ = fake
+    assert terrain.cells_of((0.0, 0.0, 2000.0, 2000.0)) == 1_000_000  # 2 m 격자
+    assert terrain.distance_to_box((5.0, 5.0), (0.0, 0.0, 10.0, 10.0)) == 0
+    assert terrain.distance_to_box((13.0, 14.0), (0.0, 0.0, 10.0, 10.0)) == 5
+    assert terrain.MAX_CELLS == 5_000_000 and terrain.FAR_M == 5000
+    assert terrain.plan_extent(root, [A, B], EPSG)["max_cells"] == terrain.MAX_CELLS
+
+
 def test_extent_no_download_and_covered(fake) -> None:
     root, seen = fake
     before = terrain.plan_extent(root, [A, B], EPSG)

resources/tester/test_530_2_plan_router.py

@@ -1,5 +1,6 @@
-"""D 지형 API(PLAN 530-1 · 561-2 · 591-2) — 범위 = 지점(설계 내용 파일) 상자 + 여유 · 지점 · 선이
-없으면 등록 기준점 둘레 ± 1 km · 받기 = 한 벌.
+"""D 지형 API(PLAN 530-1 · 561-2 · 591-2 · 595) — 범위 = 지점(설계 내용 파일) 상자 + 여유 · 지점 ·
+선이 없으면 등록 기준점 둘레 ± 1 km · 받기 = 한 벌 · 위치 칸 가운데 = 둘레 덧 · 멀다 · 다른 띠
+물음 · 새로(reset) · 칸 상한 413 · 서버 다시 켜짐 = 끊김.
 
 옛 분석 API 시험(GET · PUT analysis · 꺾임점 고침)은 옛 추천 엔진과 함께 걷음(PLAN 562-2) ·
 노선 정본 시험 = `test_530_2_plan_analyze.py` · 지형 셈 시험 = `test_530_1_plan_terrain.py`.
@@ -20,6 +21,8 @@ from D03_PlanAnalysis import D03_PlanAnalysis_Router as router_mod
 PID = "12953f5a-5ba1-42ad-a427-8ac857a58bc2"
 P0, P2 = (0.0, 0.0), (900.0, 400.0)
 URL = f"/api/projects/{PID}/plan/terrain"
+OX, OY = 200000.0, 500000.0  # 595 시험 = 중부 띠 안(경도 127°)
+Q0, Q2 = (OX + P0[0], OY + P0[1]), (OX + P2[0], OY + P2[1])
 
 
 class _Pool:
@@ -51,6 +54,7 @@ def client(tmp_path, monkeypatch):
     test = TestClient(app)
     test.plan = plan
     test.info = info
+    test.root = tmp_path
     return test
 
 
@@ -90,14 +94,14 @@ def test_post_runs_one_set(client, monkeypatch) -> None:
     saved: list = []
     works: list = []
 
-    def ensure(root, points, epsg, margin_m, report):
+    def ensure(root, points, epsg, margin_m, report, **kw):
         calls.append((points, epsg, margin_m))
         return {"sheets": {"downloaded": ["1"]}, "models": [{"model_file_path": "m"}]}
 
     async def save(project_id, models, epsg):
         saved.append((models, epsg))
 
-    def start(project_id, kind, work):
+    def start(project_id, kind, work, root=None):
         works.append(work)
         return {"status": "accepted", "kind": kind}
 
@@ -116,3 +120,154 @@ def test_post_runs_one_set(client, monkeypatch) -> None:
     # 옛 화면 꼴(points 몸) — 여유 처음 값 · 전체 시점 · 종점
     post({"points": []})
     assert calls[-1] == ([P0, P2], 5186, None)
+
+
+# ── 595 위치 바꿈 · 상한 · 끊김 ──
+
+
+@pytest.fixture
+def posting(client, monkeypatch):
+    """POST 받기 — 일은 바로 돌림 · ensure 부름(점 · 좌표계 · reset · center) 기록."""
+    calls: list = []
+
+    def ensure(root, points, epsg, margin_m, report, reset=False, center=None):
+        calls.append({"points": points, "epsg": epsg, "reset": reset, "center": center})
+        return {"sheets": {"downloaded": []}, "models": []}
+
+    async def save(project_id, models, epsg):
+        return None
+
+    works: list = []
+
+    def start(project_id, kind, work, root=None):
+        works.append(work)
+        return {"status": "accepted", "kind": kind}
+
+    def post(body):
+        res = client.post(URL, json=body)
+        while works:  # 일은 응답 뒤 돌림(앱 고리 밖)
+            asyncio.run(works.pop()(lambda percent, message: None))
+        return res
+
+    monkeypatch.setattr(router_mod.terrain, "ensure_plan_terrain", ensure)
+    monkeypatch.setattr(router_mod, "_save_models", save)
+    monkeypatch.setattr(router_mod, "_start_job", start)
+    client.plan["points"] = [{"x": Q0[0], "y": Q0[1]}, {"x": Q2[0], "y": Q2[1]}]
+    client.calls = calls
+    client.run_post = post
+    return client
+
+
+def _have(client, box, epsg=5186, center=None):
+    meta = {"box": list(box), "margin_m": 350.0, "epsg": epsg}
+    if center:
+        meta["center"] = list(center)
+    folder = router_mod.terrain.models_dir_of(client.root)
+    folder.mkdir(parents=True, exist_ok=True)
+    (folder / router_mod.terrain.TERRAIN_META).write_text(json.dumps(meta), "utf-8")
+
+
+def test_center_near_adds_site_ring(posting) -> None:
+    """가까운 위치 바꿈 = 지점 + 새 가운데 둘레 · 받은 범위와 합침."""
+    from common_util.common_util_crs import lonlat_to_tm, tm_to_lonlat
+
+    _have(posting, (OX - 500.0, OY - 500.0, OX + 1500.0, OY + 1000.0))
+    lon, lat = tm_to_lonlat(OX + 3000.0, OY, 5186)  # 받은 범위에서 1.5 km
+    res = posting.run_post({"center": {"lon": lon, "lat": lat}}).json()
+    assert res["status"] == "accepted"
+    call = posting.calls[-1]
+    x, y = lonlat_to_tm(lon, lat, 5186)
+    r = router_mod.terrain.SITE_RADIUS_M - router_mod.terrain.DEFAULT_MARGIN_M
+    assert call["points"][:2] == [Q0, Q2] and call["points"][2] == pytest.approx((x - r, y - r))
+    assert call["reset"] is False and call["center"] == (lon, lat)
+    # extent 에도 캐시 가운데 = 둘레가 상자에 들어감
+    got = posting.get(f"{URL}/extent", params={"lon": lon, "lat": lat}).json()
+    assert got["box_m"][2] == pytest.approx(x + router_mod.terrain.SITE_RADIUS_M)
+
+
+def test_center_far_asks_then_reset(posting) -> None:
+    """5 km 넘음 = 「far」(받지 않음) · reset = 옛 범위 버리고 새 ±1 km(지점 안 넣음)."""
+    from common_util.common_util_crs import tm_to_lonlat
+
+    _have(posting, (OX - 500.0, OY - 500.0, OX + 1500.0, OY + 1000.0))
+    lon, lat = tm_to_lonlat(OX + 8000.0, OY, 5186)  # 6.5 km
+    body = {"center": {"lon": lon, "lat": lat}}
+    far = posting.run_post(body).json()
+    assert far["status"] == "far" and far["distance_m"] == pytest.approx(6500, abs=2)
+    assert "km" in far["message"] and posting.calls == []
+    res = posting.run_post({**body, "reset": True}).json()
+    assert res["status"] == "accepted"
+    call = posting.calls[-1]
+    assert call["reset"] is True and len(call["points"]) == 2  # 둘레 두 귀만
+    assert call["points"][0][0] == pytest.approx(OX + 8000 - 650, abs=1)
+
+
+def test_center_other_band_asks_then_moves(posting) -> None:
+    """다른 TM 띠 = 「band」 · 확인(reset) 뒤 그 띠 · 그 뒤엔 받은 지형 띠를 따름."""
+    posting.plan["lines"] = [{"id": "a"}]
+    body = {"center": {"lon": 129.1017, "lat": 36.8128}}  # 용화(동부 5187)
+    ask = posting.run_post(body).json()
+    assert ask == {
+        "status": "band",
+        "epsg": 5186,
+        "center_epsg": 5187,
+        "message": ask["message"],
+    }
+    assert posting.calls == []
+    # 확인 전 extent = 가운데 무시(옛 띠 그대로)
+    assert posting.get(f"{URL}/extent", params={"lon": 129.1, "lat": 36.8}).json()["epsg"] == 5186
+    posting.run_post({**body, "reset": True})
+    call = posting.calls[-1]
+    assert call["epsg"] == 5187 and call["reset"] is True and len(call["points"]) == 2
+    # 받은 뒤([저장] 전) — 가운데 없이 물어도 받은 지형 띠 · 가운데 둘레
+    (x0, y0), (x1, y1) = call["points"]
+    box = [x0 - 400, y0 - 400, x1 + 400, y1 + 400]
+    _have(posting, box, epsg=5187, center=(129.1017, 36.8128))
+    got = posting.get(f"{URL}/extent").json()
+    assert got["epsg"] == 5187 and got["covered"] is True
+
+
+def test_cells_limit_413(posting) -> None:
+    posting.plan["points"].append(
+        {"x": OX + 6000.0, "y": OY + 6000.0}
+    )  # 6.7 × 6.7 km = 1,100 만 칸
+    res = posting.run_post({})
+    assert res.status_code == 413 and "상한 500만" in res.json()["detail"]
+    assert posting.calls == []
+
+
+def test_progress_interrupted_after_restart(client, monkeypatch) -> None:
+    """도는 기록이 남았는데 기억에 없음(서버 다시 켜짐) = 끊김 · 받은 적 없음 = percent 100."""
+    root = client.root
+
+    async def tile_root(project_id):
+        return root
+
+    monkeypatch.setattr(router_mod, "_tile_root", tile_root)
+    url = f"/api/projects/{PID}/plan/progress"
+    router_mod._JOBS.clear()
+    idle = client.get(url).json()["terrain"]
+    assert idle == {**router_mod.IDLE}
+
+    path = router_mod._job_path(root, "terrain")
+    seen = []
+
+    async def work(report):
+        report(65, "5/5 사업 지형면 만드는 중")
+        seen.append(json.loads(path.read_text("utf-8")))
+
+    async def run():
+        router_mod._start_job(PID, "terrain", work, root)
+        await asyncio.sleep(0.05)
+
+    asyncio.run(run())
+    assert seen[0]["running"] is True and seen[0]["percent"] == 65  # 도는 동안 기록
+    assert json.loads(path.read_text("utf-8"))["running"] is False  # 끝 기록
+    path.write_text(json.dumps(seen[0]), "utf-8")  # 도는 중 꺼짐
+    router_mod._JOBS.clear()  # 다시 켜짐 = 기억 빔
+    got = client.get(url).json()["terrain"]
+    assert got["running"] is False and got["interrupted"] is True
+    assert got["percent"] == 65 and got["message"] is None and got["error"] is None
+    # 끝난 기록 = 끊김 아님
+    path.write_text(json.dumps({**seen[0], "running": False}), "utf-8")
+    assert client.get(url).json()["terrain"] == {**router_mod.IDLE}

resources/tester/test_569_1_plan_layers.py

@@ -1,5 +1,5 @@
-"""D 노선분석 지도 겹(PLAN 569-1) — 도엽 겹 풀어 냄 · 지적도 1 km 조각 · 1,000 꽉 차면 넷 ·
-겹친 필지 거름 · 급경사 15° · 35° · 40° 단계 · 격자 지문 같으면 다시 안 셈.
+"""D 노선분석 지도 겹(PLAN 569-1 · 595-1) — 도엽 겹 풀어 냄 · 지적도 1 km 조각 · 1,000 꽉 차면 넷 ·
+겹친 필지 거름 · 실패 조각만 다시 · 급경사 15° · 35° · 40° 단계 · 격자 지문 같으면 다시 안 셈.
 
 ./venv/Scripts/python.exe -m pytest resources/tester/test_569_1_plan_layers.py -q
 """
@@ -127,9 +127,29 @@ def test_cadastral_failures(tmp_path, monkeypatch) -> None:
     monkeypatch.setattr(layers, "request_vworld_wfs", FakeWfs(fail=1))
     first = layers.ensure_cadastral(tmp_path, BOX, EPSG)
     assert first["failed"] == 1 and (tmp_path / layers.CADASTRAL_FILE).exists()
+    assert (
+        len(json.loads((tmp_path / layers.CADASTRAL_KEY).read_text("utf-8"))["failed_tiles"]) == 1
+    )
+    # 595-1 — 실패 조각만 다시 · 필지는 옛 것 + 그 조각 = 통째로 받은 것과 같음
     again = FakeWfs()
     monkeypatch.setattr(layers, "request_vworld_wfs", again)
-    assert layers.ensure_cadastral(tmp_path, BOX, EPSG)["fetched"] and again.calls == 15
+    retry = layers.ensure_cadastral(tmp_path, BOX, EPSG)
+    assert retry["fetched"] and again.calls == 1 and retry["failed"] == 0
+    assert retry["count"] == 15 * 2 + 1
+    assert not layers.ensure_cadastral(tmp_path, BOX, EPSG)["fetched"] and again.calls == 1
+    # 옛 열쇠(실패 조각 기록 없음) = 통째로
+    key = json.loads((tmp_path / layers.CADASTRAL_KEY).read_text("utf-8"))
+    key.update(failed=1)
+    key.pop("failed_tiles")
+    (tmp_path / layers.CADASTRAL_KEY).write_text(json.dumps(key), "utf-8")
+    layers.ensure_cadastral(tmp_path, BOX, EPSG)
+    assert again.calls == 1 + 15
+    # 다시 받기도 모두 실패 = 옛 필지 그대로 · 실패 조각 남김(오류 아님)
+    key.update(failed=1, failed_tiles=[list(layers.tiles(BOX)[0])])
+    (tmp_path / layers.CADASTRAL_KEY).write_text(json.dumps(key), "utf-8")
+    monkeypatch.setattr(layers, "request_vworld_wfs", FakeWfs(fail=99))
+    kept = layers.ensure_cadastral(tmp_path, BOX, EPSG)
+    assert kept["failed"] == 1 and kept["count"] == 15 * 2 + 1
 
 
 # ── 급경사 ──

resources/tester/test_584_3_plan_forest_road_zones.py

@@ -1,5 +1,6 @@
-"""D 노선분석 GIS 겹 더함 3(PLAN 584-3) — 올린 임도망도 SHP(zip · 낱 파일 묶음 · prj 없으면 5179 ·
-cp949 · 받은 범위로 자름 · 범위 밖 = 받지 않음 · 저장본) · 산사태취약지역 시도 표(머리 · 열 별칭 ·
+"""D 노선분석 GIS 겹 더함 3(PLAN 584-3 · 595-4) — 올린 임도망도 SHP(zip · 낱 파일 묶음 · prj 없으면
+5179 · cp949 · 기준점 ±5 km ∪ 받은 지형으로 자름 · 잘린 상자 · 기준점 거리 · 범위 밖 = 받지 않음 +
+거리 · 저장본) · 산사태취약지역 시도 표(머리 · 열 별칭 ·
 도분초 · 지번) · 점 · 지적 필지 맞춤 · 표 없는 시도 · 화면 겹 정의.
 
 ./venv/Scripts/python.exe -m pytest resources/tester/test_584_3_plan_forest_road_zones.py -q
@@ -81,7 +82,7 @@ def test_bad_uploads_keep_old(tmp_path) -> None:
         ([("a.zip", b"not a zip")], "zip 파일을 열지"),
         ([(n, d) for n, d in files if not n.endswith(".dbf")], ".dbf"),
         ([("a.zip", _zip({"readme.txt": b"x"}))], ".shp 파일이 없습니다"),
-        (_shp_files(tmp_path / "far", [LINES[2]]), "받은 범위 안에 임도가 없습니다"),
+        (_shp_files(tmp_path / "far", [LINES[2]]), "받은 지형 밖뿐입니다"),
     ]
     for upload, text in bad:
         with pytest.raises(ValueError, match=re.escape(text)):
@@ -91,6 +92,26 @@ def test_bad_uploads_keep_old(tmp_path) -> None:
     assert not list((tmp_path / fr.FOLDER).glob("*.new"))
 
 
+def test_clip_box_site_5km_and_distance(tmp_path) -> None:
+    """595-4 — 자르는 범위 = 기준점 ±5 km ∪ 받은 지형 · 잘린 상자 · 가장 가까운 임도 거리 ·
+    밖뿐 = 거절 + 거리."""
+    site = (231000.0, 450600.0)
+    frame = fr.clip_box(site, BOX)
+    assert frame == (226000.0, 445600.0, 236000.0, 455600.0)  # 5 km 상자가 받은 지형을 덮음
+    assert fr.clip_box(site, (220000.0, 450000.0, 232000.0, 451000.0))[0] == 220000.0  # ∪
+    assert fr.clip_box(site, None) == frame
+    got = fr.save_roads(tmp_path, _shp_files(tmp_path / "a", LINES), frame, EPSG, site)
+    assert (got["count"], got["total"]) == (2, 3)  # 범위 밖 한 가닥(25 km)만 빠짐
+    assert got["length_m"] == pytest.approx(2000 + 300, abs=1)  # 받은 지형 밖 반도 들어옴
+    assert got["distance_m"] == pytest.approx(100, abs=0.5)  # 450600 ~ 450500
+    w, s, e, n = got["bounds"]
+    assert (w, s) == pytest.approx(tm_to_lonlat(229000, 450100, EPSG), abs=1e-4)
+    assert (e, n) == pytest.approx(tm_to_lonlat(231500, 450500, EPSG), abs=1e-4)
+    far = _shp_files(tmp_path / "far", [LINES[2]])
+    with pytest.raises(ValueError, match=r"가장 가까운 임도 2[0-9]\.[0-9] km\(파일 선 1가닥\)"):
+        fr.save_roads(tmp_path, far, frame, EPSG, site)
+
+
 def test_no_upload_is_not_found(tmp_path) -> None:
     with pytest.raises(FileNotFoundError):
         fr.road_layer(tmp_path, BOX, EPSG)