Aislo/B04_PreProcess/B04_PreProcess_Engine_Pipeline.py

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

지면 필터(업체 분류 · 점진 TIN · 둘의 다듬기 판)와 표현(DTM/implicit)의
캐시를 만들고 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,
    contour_cache_is_stale,
    extract_contours,
)
from B04_PreProcess.B04_PreProcess_Engine_Ground import base_filter
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_Quality import quality_after_build
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_contours = extract_contours(
            smooth_model_path,
            representation="regular_grid",
            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 = {
    "dtm": "regular_grid",
    "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:
    """디스크의 결과 파일과 스무딩 메타데이터가 유효한지 검사한다.

    manifest 항목이 완료여야 함 — 설정 지문이 바뀌면 manifest 가 비므로 파일만 남은
    옛 결과(평활 없는 implicit)를 다시 짓는다(요구 558).
    """
    if entry.get("status") != "completed":
        return False
    files_exist = (output_dir / f"{stem}_preview.glb").exists() and (
        output_dir / f"{stem}.npz"
    ).exists()
    if not files_exist:
        return False
    if method in config.get("smoothing_methods", ("dtm",)):
        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)
    # dtm 은 늘 맨 앞 — 같은 필터 외곽 · implicit 굳힌 격자 · 등고선 자르기의 바탕이라
    # 어느 표현을 지어도 낡았으면(manifest 항목 없음 · 지문 다름) 같이 다시 지음(754-1).
    methods = ("dtm", *(key for key in config["precompute"] if key in BUILDERS and key != "dtm"))
    # 무거운 implicit 은 기본 필터(첫 필터)에만 — 둘째 필터엔 안 지음(요구 558).
    later_methods = tuple(method for method in methods if method != "implicit")
    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(methods) + (len(filters) - 1) * len(later_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 데이터와 다릅니다.")
        # 다듬기 판은 바탕 필터 지면점으로 외곽을 그림 — 다듬기가 지운 자리 = 외곽 안 구멍(766-2)
        outer = ground_masks.get(base_filter(filter_key) or "")
        context = TerrainContext(xyz=xyz, mask=mask, bounds=bounds, config=config, outer_mask=outer)
        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 later_methods if filter_index else 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):
                interval = float(config.get("contour_interval_meters", 1.0))
                contour_path = output_dir / f"contour_{filter_key}_{method}_{interval}m.json"
                needs_contours = contour_cache_is_stale(contour_path, output_dir / f"{stem}.npz")
                if method in config.get("smoothing_methods", ("dtm",)):
                    smooth_contour_path = (
                        output_dir / f"contour_{filter_key}_{method}_smooth_{interval}m.json"
                    )
                    needs_contours = needs_contours or contour_cache_is_stale(
                        smooth_contour_path, output_dir / f"{stem}_smooth.npz"
                    )
                if needs_contours:
                    _cache_contours_safely(
                        output_dir,
                        stem,
                        filter_key,
                        method,
                        entry.get("representation", _REPRESENTATIONS[method]),
                        config,
                        manifest.get("bounds", {}),
                        entry,
                    )
                quality_after_build(output_dir, filter_key, method, context, fresh=False)
                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",)):
                    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,
                )
                quality_after_build(output_dir, filter_key, method, context, fresh=True)
            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)