B04페이지 컨셉 완료

B03_FileInput/B03_FileInput_Engine_Analyze.py

@@ -1,7 +1,10 @@
 """B03 원본 입력 파일 메타데이터 분석."""
 
+import logging
 import math
+import re
 from pathlib import Path
+from threading import get_ident
 from typing import Any
 
 import laspy
@@ -9,6 +12,117 @@ import numpy as np
 import rasterio
 from pyproj import CRS
 
+logger = logging.getLogger(__name__)
+
+_CUSTOM_VERTICAL_AUTHORITY_PATTERN = re.compile(
+    r',?\s*AUTHORITY\["EPSG","(9995|99999)"\]',
+    re.IGNORECASE,
+)
+_CUSTOM_VERTICAL_WARNING_PATTERN = re.compile(
+    r"proj_create_from_database: crs not found: EPSG:(9995|99999)",
+    re.IGNORECASE,
+)
+
+
+class _CustomVerticalCrsWarningFilter(logging.Filter):
+    """알려진 사설 수직 CRS 경고만 B03 안내 로그로 변환한다."""
+
+    def __init__(self, source: Path) -> None:
+        super().__init__()
+        self.source = source
+        self.thread_id = get_ident()
+        self.logged_codes: set[str] = set()
+
+    def filter(self, record: logging.LogRecord) -> bool:
+        if get_ident() != self.thread_id:
+            return True
+        match = _CUSTOM_VERTICAL_WARNING_PATTERN.search(record.getMessage())
+        if match is None:
+            return True
+        code = f"EPSG:{match.group(1)}"
+        if code not in self.logged_codes:
+            logger.info(
+                "사용자 정의 수직 CRS 코드를 보존합니다: file=%s code=%s",
+                self.source.name,
+                code,
+            )
+            self.logged_codes.add(code)
+        return False
+
+
+def _component_metadata(crs: CRS) -> dict[str, Any]:
+    """CRS 구성 요소를 JSON 저장 가능한 메타데이터로 변환한다."""
+    authority = crs.to_authority()
+    epsg = crs.to_epsg()
+    return {
+        "name": crs.name,
+        "type": crs.type_name,
+        "epsg": epsg,
+        "authority": (
+            {"name": authority[0], "code": authority[1]} if authority is not None else None
+        ),
+    }
+
+
+def normalize_crs_metadata(crs: Any | None) -> dict[str, Any]:
+    """복합 CRS를 수평·수직 구성 요소로 분리해 일관된 상태를 반환한다."""
+    if crs is None:
+        return {
+            "crs": None,
+            "epsg": None,
+            "horizontal_crs": None,
+            "vertical_crs": None,
+            "crs_status": "missing_crs",
+        }
+
+    parsed = crs if isinstance(crs, CRS) else CRS.from_user_input(crs)
+    components = parsed.sub_crs_list or [parsed]
+    horizontal = next(
+        (item for item in components if item.is_projected or item.is_geographic),
+        None,
+    )
+    vertical = next((item for item in components if item.is_vertical), None)
+
+    horizontal_metadata = _component_metadata(horizontal) if horizontal is not None else None
+    vertical_metadata = _component_metadata(vertical) if vertical is not None else None
+    horizontal_epsg = horizontal_metadata["epsg"] if horizontal_metadata is not None else None
+
+    if horizontal_epsg is None:
+        status = "unknown_horizontal_crs"
+    elif vertical_metadata is not None and vertical_metadata["epsg"] is None:
+        status = "custom_vertical_crs"
+    else:
+        status = "identified"
+
+    return {
+        "crs": crs.to_string(),
+        "epsg": horizontal_epsg,
+        "horizontal_crs": horizontal_metadata,
+        "vertical_crs": vertical_metadata,
+        "crs_status": status,
+    }
+
+
+def _log_crs_status(source: Path, metadata: dict[str, Any]) -> None:
+    """정상화된 CRS 상태를 B03 도메인 로그로 남긴다."""
+    if metadata["crs_status"] == "custom_vertical_crs":
+        logger.info(
+            "사용자 정의 수직 CRS를 보존합니다: file=%s horizontal_epsg=%s vertical=%s",
+            source.name,
+            metadata["epsg"],
+            metadata["vertical_crs"]["name"],
+        )
+    elif metadata["crs_status"] == "unknown_horizontal_crs":
+        logger.warning("수평 CRS를 EPSG로 식별하지 못했습니다: file=%s", source.name)
+
+
+def _prepare_prj_wkt(text: str) -> tuple[str, list[str]]:
+    """KNGeoid24 사설 EPSG 표식만 파싱용 WKT에서 분리한다."""
+    if "KNGeoid24" not in text:
+        return text, []
+    codes = sorted({f"EPSG:{code}" for code in _CUSTOM_VERTICAL_AUTHORITY_PATTERN.findall(text)})
+    return _CUSTOM_VERTICAL_AUTHORITY_PATTERN.sub("", text), codes
+
 
 def analyze_las_metadata(path: str | Path) -> dict[str, Any]:
     """LAS/LAZ 헤더와 분류 통계를 메모리에 전체 적재하지 않고 분석한다."""
@@ -19,6 +133,8 @@ def analyze_las_metadata(path: str | Path) -> dict[str, Any]:
         dimension_names = list(point_format.dimension_names)
         point_count = int(header.point_count)
         crs = header.parse_crs()
+        crs_metadata = normalize_crs_metadata(crs)
+        _log_crs_status(source, crs_metadata)
         metadata: dict[str, Any] = {
             "file": source.name,
             "version": f"{header.version.major}.{header.version.minor}",
@@ -35,8 +151,7 @@ def analyze_las_metadata(path: str | Path) -> dict[str, Any]:
             "scale": [float(value) for value in header.scales],
             "offset": [float(value) for value in header.offsets],
             "has_crs": crs is not None,
-            "crs": crs.to_string() if crs else None,
-            "epsg": crs.to_epsg() if crs else None,
+            **crs_metadata,
             "has_classification": "classification" in dimension_names,
             "has_rgb": all(name in dimension_names for name in ("red", "green", "blue")),
             "has_intensity": "intensity" in dimension_names,
@@ -75,17 +190,21 @@ def analyze_prj_metadata(path: str | Path) -> dict[str, Any]:
         metadata["error"] = "PRJ 파일이 비어 있습니다."
         return metadata
 
+    parse_text, custom_authority_codes = _prepare_prj_wkt(text)
     try:
-        crs = CRS.from_wkt(text)
+        crs = CRS.from_wkt(parse_text)
     except Exception as exc:
         metadata["error"] = str(exc)
         return metadata
 
+    crs_metadata = normalize_crs_metadata(crs)
+    _log_crs_status(source, crs_metadata)
     metadata.update(
         {
-            "epsg": crs.to_epsg(),
+            **crs_metadata,
             "name": crs.name,
             "authority": crs.to_authority(),
+            "custom_authority_codes": custom_authority_codes,
             "is_valid": True,
         }
     )
@@ -119,28 +238,35 @@ def analyze_tfw_metadata(path: str | Path) -> dict[str, Any]:
 def analyze_tif_metadata(path: str | Path) -> dict[str, Any]:
     """TIF/GeoTIFF 데이터셋의 공간 및 밴드 메타데이터를 분석한다."""
     source = Path(path)
-    with rasterio.open(source) as dataset:
-        crs = dataset.crs
-        bounds = dataset.bounds
-        return {
-            "file": source.name,
-            "width": int(dataset.width),
-            "height": int(dataset.height),
-            "count": int(dataset.count),
-            "dtypes": list(dataset.dtypes),
-            "nodata": float(dataset.nodata) if dataset.nodata is not None else None,
-            "crs": crs.to_string() if crs else None,
-            "epsg": crs.to_epsg() if crs else None,
-            "bounds": {
-                "left": float(bounds.left),
-                "bottom": float(bounds.bottom),
-                "right": float(bounds.right),
-                "top": float(bounds.top),
-            },
-            "transform": [float(value) for value in list(dataset.transform)[:6]],
-            "resolution": [float(value) for value in dataset.res],
-            "likely_type": "dem" if dataset.count == 1 else "image",
-        }
+    rasterio_logger = logging.getLogger("rasterio._env")
+    warning_filter = _CustomVerticalCrsWarningFilter(source)
+    rasterio_logger.addFilter(warning_filter)
+    try:
+        with rasterio.open(source) as dataset:
+            crs = dataset.crs
+            crs_metadata = normalize_crs_metadata(crs)
+            _log_crs_status(source, crs_metadata)
+            bounds = dataset.bounds
+            return {
+                "file": source.name,
+                "width": int(dataset.width),
+                "height": int(dataset.height),
+                "count": int(dataset.count),
+                "dtypes": list(dataset.dtypes),
+                "nodata": float(dataset.nodata) if dataset.nodata is not None else None,
+                **crs_metadata,
+                "bounds": {
+                    "left": float(bounds.left),
+                    "bottom": float(bounds.bottom),
+                    "right": float(bounds.right),
+                    "top": float(bounds.top),
+                },
+                "transform": [float(value) for value in list(dataset.transform)[:6]],
+                "resolution": [float(value) for value in dataset.res],
+                "likely_type": "dem" if dataset.count == 1 else "image",
+            }
+    finally:
+        rasterio_logger.removeFilter(warning_filter)
 
 
 def analyze_input_metadata(path: str | Path) -> dict[str, Any]:

main.py

@@ -266,4 +266,5 @@ if __name__ == "__main__":
         host=SERVER_HOST,
         port=SERVER_PORT,
         reload=DEBUG,
+        access_log=False,
     )

scratch/test_crs_verification.py

@@ -0,0 +1,70 @@
+import sys
+import os
+from pathlib import Path
+
+# Add project root to sys.path
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+
+from B03_FileInput.B03_FileInput_Engine_Analyze import (
+    analyze_prj_metadata,
+    analyze_tif_metadata,
+    analyze_las_metadata,
+)
+
+BASE_DIR = Path("D:/02_Software_Prog/임도설계 및 견적자동화 프로그램 개발")
+PRJ_PATH = BASE_DIR / "storage/1/3/acb9170b-9ac8-49b3-82a0-51cfa32bb42d/B03_FileInput/input/prj/result.prj"
+TIF_PATH = BASE_DIR / "storage/1/3/acb9170b-9ac8-49b3-82a0-51cfa32bb42d/B03_FileInput/input/tif/result.tif"
+LAS_PATH = BASE_DIR / "storage/1/3/acb9170b-9ac8-49b3-82a0-51cfa32bb42d/B03_FileInput/input/las/cloud_merged.las"
+
+def run_tests():
+    print("--- 1. Testing PRJ Metadata Extraction ---")
+    if PRJ_PATH.exists():
+        prj_meta = analyze_prj_metadata(PRJ_PATH)
+        print("PRJ Metadata:")
+        for k, v in prj_meta.items():
+            print(f"  {k}: {v}")
+        
+        # Assertions based on PLAN.md expectations
+        assert prj_meta.get("epsg") == 5187, f"Expected horizontal EPSG to be 5187, got {prj_meta.get('epsg')}"
+        assert prj_meta.get("crs_status") == "custom_vertical_crs", f"Expected custom_vertical_crs, got {prj_meta.get('crs_status')}"
+        assert prj_meta.get("vertical_crs") is not None, "Expected vertical_crs metadata to be present"
+        assert "KNGeoid24" in prj_meta["vertical_crs"]["name"], f"Expected KNGeoid24 in vertical crs name, got {prj_meta['vertical_crs']['name']}"
+        print("PRJ test passed successfully.")
+    else:
+        print(f"PRJ file not found at {PRJ_PATH}")
+
+    print("\n--- 2. Testing TIF Metadata Extraction ---")
+    if TIF_PATH.exists():
+        tif_meta = analyze_tif_metadata(TIF_PATH)
+        print("TIF Metadata:")
+        for k, v in tif_meta.items():
+            print(f"  {k}: {v}")
+            
+        assert tif_meta.get("epsg") == 5187, f"Expected EPSG to be 5187, got {tif_meta.get('epsg')}"
+        assert tif_meta.get("crs_status") == "identified", f"Expected identified, got {tif_meta.get('crs_status')}"
+        assert tif_meta.get("vertical_crs") is None, "Expected no vertical crs for TIF file"
+        print("TIF test passed successfully.")
+    else:
+        print(f"TIF file not found at {TIF_PATH}")
+
+    print("\n--- 3. Testing LAS Metadata Extraction ---")
+    if LAS_PATH.exists():
+        las_meta = analyze_las_metadata(LAS_PATH)
+        print("LAS Metadata:")
+        for k, v in las_meta.items():
+            # Exclude large dimensions output for print readability
+            if k == "point_format":
+                print(f"  {k}: id={v.get('id')}, num_dimensions={len(v.get('dimensions', []))}")
+            else:
+                print(f"  {k}: {v}")
+                
+        assert las_meta.get("epsg") == 5187, f"Expected EPSG to be 5187, got {las_meta.get('epsg')}"
+        assert las_meta.get("crs_status") == "custom_vertical_crs", f"Expected custom_vertical_crs, got {las_meta.get('crs_status')}"
+        assert las_meta.get("vertical_crs") is not None, "Expected vertical_crs metadata to be present"
+        assert "KNGeoid24" in las_meta["vertical_crs"]["name"], f"Expected KNGeoid24 in vertical crs name, got {las_meta['vertical_crs']['name']}"
+        print("LAS test passed successfully.")
+    else:
+        print(f"LAS file not found at {LAS_PATH}")
+
+if __name__ == "__main__":
+    run_tests()