Aislo/B04_PreProcess/B04_PreProcess_Engine_Pipeline.py

"""B04 지표면 모델 파이프라인 오케스트레이터.

세 지면 필터(grid_min_z/csf/pmf)와 다섯 표현(TIN/DTM/NURBS/implicit/meshfree)의
캐시를 만들고 manifest.json을 관리한다. 캐시 유효성 검증, 스무딩/등고선 연동,
동일 출력 폴더의 중복 실행 취소를 포함한다.
"""

import json
import logging
import threading
import time
from pathlib import Path
from typing import Any, Callable

import numpy as np

from B04_PreProcess.B04_PreProcess_Engine_Contour import (
    CONTOUR_EXTRACTOR_VERSION,
    extract_contours,
)
from B04_PreProcess.B04_PreProcess_Engine_ModelBuild import BUILDERS
from B04_PreProcess.B04_PreProcess_Engine_ModelContext import (
    MODEL_VERSION,
    TerrainContext,
    bounds_dict,
    config_signature,
)
from B04_PreProcess.B04_PreProcess_Engine_Smooth import (
    compute_smoothing_signature,
    run_smoothing,
)
from common_util.common_util_atomic import atomic_write_bytes
from common_util.common_util_json import atomic_write_json

# 진행률 콜백: (overall_percent, detail_message)
ProgressReporter = Callable[[int, str], None]
logger = logging.getLogger(__name__)

# 같은 프로세스에서 동일 프로젝트 계산 요청이 겹치면 두 번째 요청을 즉시 취소.
_ACTIVE_TERRAIN_BUILDS: set[str] = set()
_ACTIVE_TERRAIN_BUILDS_GUARD = threading.Lock()


def _write_json_file(path: Path, value: dict[str, Any]) -> None:
    atomic_write_json(path, value)


def _cache_contours(
    output_dir: Path,
    stem: str,
    filter_key: str,
    method: str,
    representation: str,
    config: dict[str, Any],
    bounds_info: dict[str, Any],
    metadata: dict[str, Any],
) -> None:
    """빌드 완료 직후 기본 간격 등고선을 사전 추출·캐싱한다 (원본 + 스무딩)."""
    interval = float(config.get("contour_interval_meters", 1.0))
    target_grid_m = float(config.get("contour_grid_resolution_meters", 1.0))
    model_path = output_dir / f"{stem}.npz"

    if model_path.exists():
        contours = extract_contours(
            model_path,
            representation=representation,
            interval=interval,
            target_grid_m=target_grid_m,
            scene_center=None,
        )
        payload = {
            "extractor_version": CONTOUR_EXTRACTOR_VERSION,
            "project_id": output_dir.parent.name,
            "source_filter": filter_key,
            "method": method,
            "interval": interval,
            "bounds": bounds_info,
            "contours": contours,
        }
        atomic_write_bytes(
            output_dir / f"contour_{filter_key}_{method}_{interval}m.json",
            json.dumps(payload, ensure_ascii=False).encode("utf-8"),
        )

    smooth_model_path = output_dir / f"{stem}_smooth.npz"
    smooth_meta = metadata.get("smooth", {})
    if smooth_model_path.exists() and smooth_meta.get("status") == "completed":
        smooth_rep = "regular_grid" if method == "dtm" else "triangular_mesh"
        smooth_contours = extract_contours(
            smooth_model_path,
            representation=smooth_rep,
            interval=interval,
            target_grid_m=target_grid_m,
            scene_center=None,
        )
        payload = {
            "extractor_version": CONTOUR_EXTRACTOR_VERSION,
            "project_id": output_dir.parent.name,
            "source_filter": filter_key,
            "method": method,
            "interval": interval,
            "bounds": bounds_info,
            "contours": smooth_contours,
        }
        atomic_write_bytes(
            output_dir / f"contour_{filter_key}_{method}_smooth_{interval}m.json",
            json.dumps(payload, ensure_ascii=False).encode("utf-8"),
        )


def _cache_contours_safely(
    output_dir: Path,
    stem: str,
    filter_key: str,
    method: str,
    representation: str,
    config: dict[str, Any],
    bounds_info: dict[str, Any],
    metadata: dict[str, Any],
) -> None:
    """등고선 사전 캐시 실패를 기록하되 모델 빌드는 유지한다."""
    try:
        _cache_contours(
            output_dir,
            stem,
            filter_key,
            method,
            representation,
            config,
            bounds_info,
            metadata,
        )
    except Exception as exc:
        logger.warning(
            "등고선 사전 캐시 실패: filter=%s, method=%s, error=%s",
            filter_key,
            method,
            exc,
        )


_REPRESENTATIONS = {
    "meshfree": "meshfree_surfels",
    "dtm": "regular_grid",
    "tin": "triangular_mesh",
    "nurbs": "bspline_surface",
    "implicit": "local_rbf_height_field",
}


def _cache_is_valid(
    output_dir: Path, stem: str, method: str, entry: dict[str, Any], config: dict[str, Any]
) -> bool:
    """디스크의 결과 파일과 스무딩 메타데이터가 유효한지 검사한다."""
    ext = "ply" if method == "meshfree" else "glb"
    files_exist = (output_dir / f"{stem}_preview.{ext}").exists() and (
        output_dir / f"{stem}.npz"
    ).exists()
    if not files_exist:
        return False
    if method in config.get("smoothing_methods", ("dtm", "tin")):
        smooth_entry = entry.get("smooth", {})
        smooth_exist = (output_dir / f"{stem}_smooth.npz").exists() and (
            output_dir / f"{stem}_smooth_preview.glb"
        ).exists()
        if (
            not smooth_exist
            or smooth_entry.get("status") != "completed"
            or smooth_entry.get("smoothing_signature") != compute_smoothing_signature(config)
        ):
            return False
    return True


def _build_all_terrain_models(
    structured_data: dict[str, np.ndarray] | np.lib.npyio.NpzFile,
    ground_masks: dict[str, np.ndarray],
    output_dir: Path,
    config: dict[str, Any],
    *,
    force: bool = False,
    progress: ProgressReporter | None = None,
) -> dict[str, Any]:
    """세 지면 필터와 다섯 표현의 캐시를 만들고 manifest를 반환한다."""
    output_dir.mkdir(parents=True, exist_ok=True)
    manifest_path = output_dir / "manifest.json"
    filters = tuple(key for key in config["source_filters"] if key in ground_masks)
    methods_list = [key for key in config["precompute"] if key in BUILDERS]
    if "dtm" in methods_list:
        methods_list.remove("dtm")
        methods_list.insert(0, "dtm")
    methods = tuple(methods_list)
    signature = config_signature(config)
    bounds = np.asarray(structured_data["bounds"], dtype=np.float64)
    xyz = structured_data["xyz"]

    existing: dict[str, Any] = {}
    if manifest_path.exists() and not force:
        try:
            existing = json.loads(manifest_path.read_text(encoding="utf-8"))
        except (json.JSONDecodeError, OSError):
            existing = {}
    if existing.get("config_signature") != signature:
        existing = {}
    manifest: dict[str, Any] = existing or {
        "version": MODEL_VERSION,
        "config_signature": signature,
        "bounds": bounds_dict(bounds),
        "source_filters": {},
        "started_at_unix": time.time(),
    }
    started_at = time.monotonic()
    timeout = max(0, int(config.get("sync_timeout_seconds", 0)))
    total_units = max(1, len(filters) * len(methods))
    done_units = 0
    failures = 0

    def _report(detail: str) -> None:
        if progress:
            progress(int(100 * done_units / total_units), detail)

    for filter_index, filter_key in enumerate(filters):
        mask = np.asarray(ground_masks[filter_key], dtype=bool)
        if len(mask) != len(xyz):
            raise ValueError(f"{filter_key} 마스크 길이가 XYZ 데이터와 다릅니다.")
        context = TerrainContext(xyz=xyz, mask=mask, bounds=bounds, config=config)
        filter_entry = manifest["source_filters"].setdefault(
            filter_key, {"source_point_count": context.source_count, "methods": {}}
        )
        filter_entry["source_point_count"] = context.source_count

        for method in methods:
            stem = f"{method}_{filter_key}"
            entry = filter_entry["methods"].get(method, {})

            if not force and _cache_is_valid(output_dir, stem, method, entry, config):
                if entry.get("status") != "completed":
                    entry.update(
                        {
                            "status": "completed",
                            "representation": _REPRESENTATIONS[method],
                            "model_file": f"{stem}.npz",
                            "preview_file": f"{stem}_preview."
                            + ("ply" if method == "meshfree" else "glb"),
                            "preview_media_type": "application/octet-stream"
                            if method == "meshfree"
                            else "model/gltf-binary",
                            "error": None,
                        }
                    )
                    filter_entry["methods"][method] = entry
                    _write_json_file(manifest_path, manifest)
                interval = float(config.get("contour_interval_meters", 1.0))
                contour_path = output_dir / f"contour_{filter_key}_{method}_{interval}m.json"
                needs_contours = not contour_path.exists()
                if method in config.get("smoothing_methods", ("dtm", "tin")):
                    smooth_contour_path = (
                        output_dir / f"contour_{filter_key}_{method}_smooth_{interval}m.json"
                    )
                    needs_contours = needs_contours or not smooth_contour_path.exists()
                if needs_contours:
                    _cache_contours_safely(
                        output_dir,
                        stem,
                        filter_key,
                        method,
                        entry.get("representation", _REPRESENTATIONS[method]),
                        config,
                        manifest.get("bounds", {}),
                        entry,
                    )
                done_units += 1
                _report(f"{filter_key}-{method} 캐시 재사용")
                continue

            if timeout and time.monotonic() - started_at >= timeout:
                failures += 1
                filter_entry["methods"][method] = {
                    "status": "failed",
                    "error": f"동기 계산 제한시간 {timeout}초를 초과했습니다.",
                }
                _write_json_file(manifest_path, manifest)
                done_units += 1
                _report(f"{filter_key}-{method} 시간 초과")
                continue

            method_started = time.monotonic()
            filter_entry["methods"][method] = {"status": "running", "error": None}
            _write_json_file(manifest_path, manifest)
            # 모델을 새로 만들기 전에 구세대 등고선 캐시를 제거해 모델-등고선 세대 불일치를 막는다.
            for stale_contour in output_dir.glob(f"contour_{filter_key}_{method}_*.json"):
                stale_contour.unlink(missing_ok=True)
            try:
                metadata = BUILDERS[method](
                    context,
                    output_dir,
                    stem,
                    lambda value: _report(f"{filter_key}-{method} {value}%"),
                )
                metadata.update(
                    {
                        "status": "completed",
                        "duration_seconds": round(time.monotonic() - method_started, 3),
                        "error": None,
                    }
                )
                if method in config.get("smoothing_methods", ("dtm", "tin")):
                    original_model_path = output_dir / f"{stem}.npz"
                    if original_model_path.exists():
                        try:
                            smooth_meta = run_smoothing(
                                method, context, output_dir, stem, original_model_path
                            )
                            smooth_meta["status"] = "completed"
                            metadata["smooth"] = smooth_meta
                        except Exception as smooth_exc:
                            metadata["smooth"] = {"status": "failed", "error": str(smooth_exc)}

                filter_entry["methods"][method] = metadata
                _cache_contours_safely(
                    output_dir,
                    stem,
                    filter_key,
                    method,
                    metadata.get("representation", "regular_grid"),
                    config,
                    manifest.get("bounds", {}),
                    metadata,
                )
            except Exception as exc:
                failures += 1
                filter_entry["methods"][method] = {
                    "status": "failed",
                    "duration_seconds": round(time.monotonic() - method_started, 3),
                    "error": str(exc),
                }
            done_units += 1
            _report(f"{filter_key}-{method} 완료")
            _write_json_file(manifest_path, manifest)

        context.clear_caches()

    manifest["status"] = "completed" if failures == 0 else "completed_with_errors"
    manifest["completed_at_unix"] = time.time()
    manifest["duration_seconds"] = round(time.monotonic() - started_at, 3)
    manifest["failure_count"] = failures
    _write_json_file(manifest_path, manifest)
    return manifest


def build_all_terrain_models(
    structured_data: dict[str, np.ndarray] | np.lib.npyio.NpzFile,
    ground_masks: dict[str, np.ndarray],
    output_dir: Path,
    config: dict[str, Any],
    *,
    force: bool = False,
    progress: ProgressReporter | None = None,
) -> dict[str, Any]:
    """동일 출력 폴더의 중복 실행을 즉시 취소하고 실제 빌드를 한 번만 수행한다."""
    output_dir = Path(output_dir)
    build_key = str(output_dir.resolve())
    with _ACTIVE_TERRAIN_BUILDS_GUARD:
        if build_key in _ACTIVE_TERRAIN_BUILDS:
            manifest_path = output_dir / "manifest.json"
            try:
                current = json.loads(manifest_path.read_text(encoding="utf-8"))
            except (OSError, json.JSONDecodeError):
                current = {"status": "running", "source_filters": {}}
            response = dict(current)
            response["request_status"] = "cancelled_already_running"
            response["message"] = (
                "동일 프로젝트의 지표면 모델 계산이 이미 진행 중이어서 요청을 취소했습니다."
            )
            return response
        _ACTIVE_TERRAIN_BUILDS.add(build_key)

    try:
        return _build_all_terrain_models(
            structured_data, ground_masks, output_dir, config, force=force, progress=progress
        )
    finally:
        with _ACTIVE_TERRAIN_BUILDS_GUARD:
            _ACTIVE_TERRAIN_BUILDS.discard(build_key)