260705_2

2026-07-05 21:27:23 +09:00
parent 23d907265a
commit 3abc2edba6
83 changed files with 10351 additions and 1217 deletions
@@ -0,0 +1,162 @@
+"""B03 원본 입력 파일 메타데이터 분석."""
+
+import math
+from pathlib import Path
+from typing import Any
+
+import laspy
+import numpy as np
+import rasterio
+from pyproj import CRS
+
+
+def analyze_las_metadata(path: str | Path) -> dict[str, Any]:
+    """LAS/LAZ 헤더와 분류 통계를 메모리에 전체 적재하지 않고 분석한다."""
+    source = Path(path)
+    with laspy.open(source) as las_file:
+        header = las_file.header
+        point_format = header.point_format
+        dimension_names = list(point_format.dimension_names)
+        point_count = int(header.point_count)
+        crs = header.parse_crs()
+        metadata: dict[str, Any] = {
+            "file": source.name,
+            "version": f"{header.version.major}.{header.version.minor}",
+            "point_format": {
+                "id": point_format.id,
+                "dimensions": dimension_names,
+            },
+            "point_count": point_count,
+            "bounds": {
+                "x": [float(header.mins[0]), float(header.maxs[0])],
+                "y": [float(header.mins[1]), float(header.maxs[1])],
+                "z": [float(header.mins[2]), float(header.maxs[2])],
+            },
+            "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,
+            "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,
+            "has_return_number": "return_number" in dimension_names,
+        }
+
+        if metadata["has_classification"] and point_count > 0:
+            classification_counts: dict[int, int] = {}
+            for chunk in las_file.chunk_iterator(500_000):
+                values, counts = np.unique(
+                    np.asarray(chunk.classification, dtype=np.uint8),
+                    return_counts=True,
+                )
+                for value, count in zip(values.tolist(), counts.tolist(), strict=True):
+                    classification_counts[value] = classification_counts.get(value, 0) + count
+            metadata["classification_summary"] = {
+                str(key): value for key, value in sorted(classification_counts.items())
+            }
+
+        return metadata
+
+
+def analyze_prj_metadata(path: str | Path) -> dict[str, Any]:
+    """PRJ WKT에서 좌표계 식별자와 명칭을 추출한다."""
+    source = Path(path)
+    text = source.read_text(encoding="utf-8", errors="replace").strip()
+    metadata: dict[str, Any] = {
+        "file": source.name,
+        "text_preview": text[:500],
+        "epsg": None,
+        "name": None,
+        "authority": None,
+        "is_valid": False,
+    }
+    if not text:
+        metadata["error"] = "PRJ 파일이 비어 있습니다."
+        return metadata
+
+    try:
+        crs = CRS.from_wkt(text)
+    except Exception as exc:
+        metadata["error"] = str(exc)
+        return metadata
+
+    metadata.update(
+        {
+            "epsg": crs.to_epsg(),
+            "name": crs.name,
+            "authority": crs.to_authority(),
+            "is_valid": True,
+        }
+    )
+    return metadata
+
+
+def analyze_tfw_metadata(path: str | Path) -> dict[str, Any]:
+    """TFW의 affine 변환 계수와 유효성을 분석한다."""
+    source = Path(path)
+    values = [
+        float(line.strip())
+        for line in source.read_text(encoding="utf-8", errors="replace").splitlines()
+        if line.strip()
+    ]
+    if any(not math.isfinite(value) for value in values):
+        raise ValueError("TFW 변환 계수는 유한한 숫자여야 합니다.")
+
+    return {
+        "file": source.name,
+        "values": values,
+        "pixel_size_x": values[0] if len(values) > 0 else None,
+        "rotation_y": values[1] if len(values) > 1 else None,
+        "rotation_x": values[2] if len(values) > 2 else None,
+        "pixel_size_y": values[3] if len(values) > 3 else None,
+        "origin_x": values[4] if len(values) > 4 else None,
+        "origin_y": values[5] if len(values) > 5 else None,
+        "is_valid": len(values) == 6,
+    }
+
+
+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",
+        }
+
+
+def analyze_input_metadata(path: str | Path) -> dict[str, Any]:
+    """입력 파일 확장자에 맞는 B03 메타데이터 분석 함수를 호출한다."""
+    source = Path(path)
+    extension = source.suffix.lower()
+    if extension in {".las", ".laz"}:
+        return analyze_las_metadata(source)
+    if extension == ".prj":
+        return analyze_prj_metadata(source)
+    if extension == ".tfw":
+        return analyze_tfw_metadata(source)
+    if extension in {".tif", ".tiff"}:
+        return analyze_tif_metadata(source)
+    return {
+        "file": source.name,
+        "extension": extension.lstrip("."),
+        "size_bytes": source.stat().st_size,
+    }