diff --git a/B03_FileInput/B03_FileInput_Engine_Analyze.py b/B03_FileInput/B03_FileInput_Engine_Analyze.py index 19c195d1..682b711b 100644 --- a/B03_FileInput/B03_FileInput_Engine_Analyze.py +++ b/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]: diff --git a/main.py b/main.py index a5b9a852..37e03222 100644 --- a/main.py +++ b/main.py @@ -266,4 +266,5 @@ if __name__ == "__main__": host=SERVER_HOST, port=SERVER_PORT, reload=DEBUG, + access_log=False, ) diff --git a/scratch/test_crs_verification.py b/scratch/test_crs_verification.py new file mode 100644 index 00000000..6fdb188f --- /dev/null +++ b/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()