Aislo/B03_FileInput/B03_FileInput_Engine_Analyze.py

"""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,
    }