Merge remote-tracking branch 'origin/feat/las-free-sheet-surface' into feat/B07-cad-block-library

common_util/common_util_drainage_pipes.py

@@ -5,9 +5,17 @@
 읽고 쓴다 — 관리자 화면에서 옮긴 관이 사용자 화면에서 다르게 보이면 안 되기 때문이다
 (2026-08-01 사용자 지시).
 
-위치는 좌표가 아니라 **누가거리(chainage_m)** 로 저장한다. 지면 필터나 지표면 모델을 바꾸면
+위치는 **누가거리(chainage_m) + 좌표(x, y)** 로 저장한다. 지면 필터나 지표면 모델을 바꾸면
 종단 Z가 달라지지만 관이 놓인 자리는 그대로여야 하고, 그때는 세부유역만 다시 나누면 된다.
-노선 자체가 바뀌면(`route_signature` 불일치) 기준이 사라지므로 전량 버리고 다시 만든다.
+
+좌표를 같이 남기는 이유(2026-08-30 사용자 지적 — "결국 노선 위에 위치해야 한다"): 관 자리를
+정한 선(계획노선 CSV)과 화면에 그려지는 선(B05 최적 경로)은 **같은 자리를 지나면서 연장이
+다르다**(실측 350.11m vs 354.83m). 누가거리만 남기면 읽는 쪽이 쥔 선에 따라 같은 값이 3~4m
+미끄러져 관이 선 옆에 떨어진 것처럼 보인다. 좌표를 남겨 두면 어느 선으로 읽든 그 좌표를
+투영해 **항상 선 위에** 앉힐 수 있다.
+
+노선이 바뀌면(`route_signature` 불일치) 좌표가 있는 저장분은 새 노선에 투영해 이월하고,
+좌표가 없는 구 저장분만 버린다.
 """
 
 from __future__ import annotations
@@ -19,9 +27,11 @@ from dataclasses import dataclass
 from pathlib import Path
 from typing import Any
 
+from shapely.geometry import LineString, Point
+
 from B04_PreProcess.B04_PreProcess_Engine_Watershed_Export import drainage_dir
 from common_util.common_util_json import atomic_write_json
-from common_util.common_util_route_geometry import RouteVertex
+from common_util.common_util_route_geometry import RouteVertex, interpolate_vertex
 from config.config_system import (
     DRAINAGE_DETAIL_FILENAME,
     DRAINAGE_EDITS_DIRNAME,
@@ -72,6 +82,9 @@ class PipePoint:
     start_m: float | None = None
     end_m: float | None = None
     options: dict[str, Any] | None = None
+    # 관이 실제로 놓인 자리(사업지 CRS, m). 노선이 바뀌어도 이 자리는 그대로다.
+    x: float | None = None
+    y: float | None = None
 
     def as_dict(self) -> dict[str, Any]:
         # 구 형식 저장분이 확장 필드 없이 그대로 다시 저장되도록 기본값은 생략한다.
@@ -86,6 +99,9 @@ class PipePoint:
             payload["end_m"] = round(float(self.end_m), 2)
         if self.options:
             payload["options"] = self.options
+        if self.x is not None and self.y is not None:
+            payload["x"] = round(float(self.x), 3)
+            payload["y"] = round(float(self.y), 3)
         return payload
 
 
@@ -113,9 +129,59 @@ def route_signature(vertices: list[RouteVertex]) -> str:
     return f"{len(vertices)}-{digest.hexdigest()[:16]}"
 
 
-def load_pipe_points(stored_path: str, signature: str) -> list[PipePoint] | None:
-    """저장된 관 지점을 읽는다. 파일이 없거나 노선이 바뀌었으면 None(= 다시 만들어야 함)."""
-    path = pipe_points_path(stored_path)
+def fill_pipe_coordinates(points: list[PipePoint], vertices: list[RouteVertex]) -> list[PipePoint]:
+    """좌표가 비어 있는 관에 그 누가거리의 노선 좌표를 채운다(제자리 수정)."""
+    if not vertices:
+        return points
+    for point in points:
+        if point.x is None or point.y is None:
+            x, y, _ = interpolate_vertex(vertices, float(point.chainage_m))
+            point.x, point.y = float(x), float(y)
+    return points
+
+
+def project_pipe_points(points: list[PipePoint], vertices: list[RouteVertex]) -> list[PipePoint]:
+    """저장된 좌표를 주어진 노선에 투영해 누가거리를 다시 매긴다.
+
+    관이 놓인 **자리**는 좌표가 정본이고 누가거리는 그 자리를 읽는 선에 종속된 값이다.
+    앞뒤 구간(start_m·end_m)은 기준점이 옮겨간 만큼 같이 민다 — 구간 길이는 시설 치수라
+    노선이 바뀌어도 변하지 않는다.
+    """
+    if not vertices:
+        return points
+    line = LineString([(vertex.x, vertex.y) for vertex in vertices])
+    if line.length <= 0:
+        return points
+    for point in points:
+        if point.x is None or point.y is None:
+            continue
+        moved = float(line.project(Point(point.x, point.y)))
+        shift = moved - float(point.chainage_m)
+        point.chainage_m = moved
+        if point.start_m is not None:
+            point.start_m = float(point.start_m) + shift
+        if point.end_m is not None:
+            point.end_m = float(point.end_m) + shift
+    points.sort(key=lambda item: item.chainage_m)
+    return points
+
+
+def load_pipe_points(
+    stored_path: str, signature: str, vertices: list[RouteVertex] | None = None
+) -> list[PipePoint] | None:
+    """저장된 관 지점을 읽는다. 파일이 없거나 이월할 수 없으면 None(= 다시 만들어야 함).
+
+    노선 지문이 다르면 예전에는 전량 버렸다. 저장분에 좌표가 있으면 `vertices`(읽는 쪽이
+    쓰는 노선)에 투영해 이월한다 — 같은 자리를 지나면서 연장만 다른 선끼리 관이 통째로
+    사라지던 것을 막는다(2026-08-30 사용자 지적).
+    """
+    return load_pipe_points_file(pipe_points_path(stored_path), signature, vertices)
+
+
+def load_pipe_points_file(
+    path: Path, signature: str, vertices: list[RouteVertex] | None = None
+) -> list[PipePoint] | None:
+    """`load_pipe_points`와 같되 파일 경로로 직접 읽는다 (B05는 프로젝트 루트를 쥔다)."""
     if not path.exists():
         return None
     try:
@@ -124,11 +190,19 @@ def load_pipe_points(stored_path: str, signature: str) -> list[PipePoint] | None
     except (OSError, json.JSONDecodeError):
         logger.warning("배수유역: 관 지점 파일을 읽지 못했습니다 (%s).", path)
         return None
+    points = parse_pipe_points(document.get("points"))
     stored_signature = str(document.get("route_signature") or "")
-    if stored_signature != signature:
-        logger.info("배수유역: 노선이 바뀌어 저장된 관 지점을 버립니다 (%s).", path.name)
-        return None
-    return parse_pipe_points(document.get("points"))
+    if stored_signature == signature:
+        return points
+    if vertices and points and all(p.x is not None and p.y is not None for p in points):
+        logger.info(
+            "배수유역: 노선이 바뀌어 관 지점 %d건을 좌표로 이월합니다 (%s).",
+            len(points),
+            path.name,
+        )
+        return project_pipe_points(points, vertices)
+    logger.info("배수유역: 노선이 바뀌어 저장된 관 지점을 버립니다 (%s).", path.name)
+    return None
 
 
 def _parse_span(item: dict[str, Any], chainage: float) -> tuple[float | None, float | None]:
@@ -187,6 +261,7 @@ def parse_pipe_points(values: Any) -> list[PipePoint]:
         facility = str(item.get("facility") or PIPE_FACILITY_PIPE)
         start, end = _parse_span(item, float(chainage))
         options = item.get("options")
+        raw_x, raw_y = item.get("x"), item.get("y")
         points.append(
             PipePoint(
                 chainage_m=float(chainage),
@@ -194,6 +269,8 @@ def parse_pipe_points(values: Any) -> list[PipePoint]:
                 facility=facility if facility in _KNOWN_FACILITIES else PIPE_FACILITY_PIPE,
                 start_m=start,
                 end_m=end,
+                x=float(raw_x) if isinstance(raw_x, (int, float)) else None,
+                y=float(raw_y) if isinstance(raw_y, (int, float)) else None,
                 options=(
                     _migrate_protection(dict(options))
                     if isinstance(options, dict) and options
@@ -227,8 +304,19 @@ def carry_facility_attributes(base: list[PipePoint], reference: list[PipePoint])
     return base
 
 
-def save_pipe_points(stored_path: str, signature: str, points: list[PipePoint]) -> int:
-    """관 지점을 정본 파일에 쓴다. 저장된 개수를 돌려준다."""
+def save_pipe_points(
+    stored_path: str,
+    signature: str,
+    points: list[PipePoint],
+    vertices: list[RouteVertex] | None = None,
+) -> int:
+    """관 지점을 정본 파일에 쓴다. 저장된 개수를 돌려준다.
+
+    `vertices`를 주면 좌표가 빈 관을 그 노선 위 좌표로 채워 둔다 — 다음에 다른 선으로
+    읽어도 그 자리에 되놓을 수 있다.
+    """
+    if vertices:
+        fill_pipe_coordinates(points, vertices)
     path = pipe_points_path(stored_path)
     path.parent.mkdir(parents=True, exist_ok=True)
     atomic_write_json(

common_util/common_util_storage.py

@@ -60,8 +60,9 @@ def resolve_temp_batch_path(user_id: int, batch_id: str, *, create: bool = True)
     if not batch_id or any(sep in batch_id for sep in ("/", "\\", "..")):
         raise ValueError("임시 보관함 묶음 식별자가 올바르지 않습니다.")
 
-    temp_root = os.path.abspath(os.path.join(STORAGE_BASE_DIR, TEMP_UPLOAD_DIR_NAME))
-    path = os.path.abspath(os.path.join(temp_root, str(user_id), batch_id))
+    # 실경로로 맞춘다 — 저장 엔진이 `Path.resolve()`를 쓰므로 기준이 같아야 한다.
+    temp_root = os.path.realpath(os.path.join(STORAGE_BASE_DIR, TEMP_UPLOAD_DIR_NAME))
+    path = os.path.realpath(os.path.join(temp_root, str(user_id), batch_id))
     if os.path.commonpath((temp_root, path)) != temp_root or path == temp_root:
         raise ValueError("임시 보관함 경로가 보관함 루트를 벗어났습니다.")
     if create:
@@ -70,8 +71,8 @@ def resolve_temp_batch_path(user_id: int, batch_id: str, *, create: bool = True)
 
 
 def temp_upload_root() -> str:
-    """임시 보관함 루트(`storage/tmp`) 절대 경로."""
-    return os.path.abspath(os.path.join(STORAGE_BASE_DIR, TEMP_UPLOAD_DIR_NAME))
+    """임시 보관함 루트(`storage/tmp`) 실경로."""
+    return os.path.realpath(os.path.join(STORAGE_BASE_DIR, TEMP_UPLOAD_DIR_NAME))
 
 
 def resolve_project_root_for_delete(relative_path: str, project_id: str) -> str:
@@ -100,15 +101,20 @@ def resolve_project_root_for_delete(relative_path: str, project_id: str) -> str:
 
 
 def resolve_stored_project_path(relative_path: str) -> str:
-    """DB의 storage 기준 상대 경로를 검증해 실제 프로젝트 경로로 변환한다."""
+    """DB의 storage 기준 상대 경로를 검증해 실제 프로젝트 경로로 변환한다.
+
+    실경로(`realpath`)로 돌려준다. `storage/`가 심볼릭 링크·정션일 수 있고(워크트리를
+    나눠 쓰면 실제로 그렇다), 저장 엔진 쪽은 `Path.resolve()`로 링크를 따라간다. 두
+    경로의 기준이 다르면 `chunk_path.relative_to(project_root)`가 터진다.
+    """
     normalized = PurePosixPath(relative_path.replace("\\", "/"))
     if normalized.is_absolute() or ".." in normalized.parts:
         raise ValueError("프로젝트 저장 경로는 안전한 상대 경로여야 합니다.")
     if not normalized.parts or normalized.parts[0] != "storage":
         raise ValueError("프로젝트 저장 경로는 storage/로 시작해야 합니다.")
 
-    storage_root = os.path.abspath(STORAGE_BASE_DIR)
-    path = os.path.abspath(os.path.join(storage_root, *normalized.parts[1:]))
+    storage_root = os.path.realpath(STORAGE_BASE_DIR)
+    path = os.path.realpath(os.path.join(storage_root, *normalized.parts[1:]))
     if os.path.commonpath((storage_root, path)) != storage_root or path == storage_root:
         raise ValueError("프로젝트 저장 경로가 저장소 루트를 벗어났습니다.")
     os.makedirs(path, exist_ok=True)