Files
Aislo/B04_PreProcess/B04_PreProcess_Engine_Ground.py
T
eomsangdonandClaude Opus 5 e588c73aa7 feat(B04): 분류가 있어도 csf를 함께 만들어 저장한다
csf 처리 결과를 눈으로 검증할 수 있어야 한다(사용자 지시). 지면분류가 있는 LAS는
classification 하나만 만들고 끝내면 csf가 제대로 도는지 확인할 방법이 없다.

resolve_auto_source_filters 가 이제 목록을 돌려준다 — 첫 항목이 기본 확정값이다.
  분류 있음: ['classification', 'csf']   기본 확정 classification
  분류 없음: ['csf']                      기본 확정 csf

서피스 생성 로직은 두 경우가 같다. 갈리는 것은 기본 확정값 지정뿐이다.

실측: cloud_merged.las(미분류) -> ['csf'], 용화.las(class 2 2.33%) ->
['classification', 'csf'].

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
2026-09-01 13:01:00 +09:00

94 lines
3.6 KiB
Python

"""B04 지면 필터 오케스트레이션.
구조화된 포인트클라우드(structured.npz)에 대해 grid_min_z/csf/pmf/ransac
필터를 실행하여 지면 마스크 딕셔너리를 만든다. 필터 선택은 config의
source_filters를 따른다.
"""
from typing import Any
import numpy as np
from B04_PreProcess.B04_PreProcess_Engine_Filter_Classification import (
filter_classification,
has_classified_ground,
)
from B04_PreProcess.B04_PreProcess_Engine_Filter_CSF import filter_csf
from B04_PreProcess.B04_PreProcess_Engine_Filter_Grid import filter_grid_min_z
from B04_PreProcess.B04_PreProcess_Engine_Filter_PMF import filter_pmf
from B04_PreProcess.B04_PreProcess_Engine_Filter_RANSAC import filter_ransac
from config.config_system import SURFACE_AUTO_FALLBACK_FILTER
# 필터 키 → 함수 매핑
_FILTERS = {
"grid_min_z": filter_grid_min_z,
"csf": filter_csf,
"pmf": filter_pmf,
"ransac": filter_ransac,
"classification": filter_classification,
}
# LAS에 지면분류가 실제로 들어 있을 때만 쓰는 필터 (미분류 LAS에서는 마스크가 빈다)
_CONDITIONAL_FILTERS = {"classification": has_classified_ground}
def available_filters() -> tuple[str, ...]:
return tuple(_FILTERS.keys())
def usable_conditional_filters(
structured_data: dict[str, Any] | np.lib.npyio.NpzFile,
) -> list[str]:
"""이 LAS에서 실제로 쓸 수 있는 조건부 필터를 고른다 (드롭다운 노출용)."""
return [key for key, is_usable in _CONDITIONAL_FILTERS.items() if is_usable(structured_data)]
def resolve_auto_source_filters(
structured_data: dict[str, Any] | np.lib.npyio.NpzFile,
) -> list[str]:
"""자동 전처리가 만들 필터를 입력 LAS를 보고 정한다. 첫 항목이 기본 확정값이다.
업체가 이미 지면을 분류해 놨으면 그것을 기본으로 쓴다 — 계산이 없고 품질도 가장
안정적이다. **csf는 분류가 있어도 함께 만든다** — csf 처리 결과를 눈으로 검증할 수
있어야 하기 때문이다 (2026-09-01 사용자 확정). 분류가 없으면 csf 하나만 남는다.
"""
usable = usable_conditional_filters(structured_data)
return [*usable[:1], SURFACE_AUTO_FALLBACK_FILTER]
def run_ground_filter(
filter_key: str, structured_data: dict[str, Any] | np.lib.npyio.NpzFile
) -> np.ndarray:
"""단일 지면 필터를 실행해 불리언 마스크를 반환한다."""
if filter_key not in _FILTERS:
raise ValueError(f"알 수 없는 지면 필터입니다: {filter_key}")
return np.asarray(_FILTERS[filter_key](structured_data), dtype=bool)
def build_ground_masks(
structured_data: dict[str, Any] | np.lib.npyio.NpzFile,
filter_keys: tuple[str, ...] | list[str],
) -> dict[str, np.ndarray]:
"""지정한 필터들을 실행해 {filter_key: mask} 딕셔너리를 만든다."""
masks: dict[str, np.ndarray] = {}
for filter_key in filter_keys:
masks[filter_key] = run_ground_filter(filter_key, structured_data)
return masks
def summarize_masks(
structured_data: dict[str, Any] | np.lib.npyio.NpzFile,
masks: dict[str, np.ndarray],
) -> dict[str, dict[str, Any]]:
"""각 필터 마스크의 지면 포인트 수·비율 요약을 만든다."""
total = int(len(structured_data["xyz"]))
summary: dict[str, dict[str, Any]] = {}
for filter_key, mask in masks.items():
ground = int(np.count_nonzero(mask))
summary[filter_key] = {
"ground_point_count": ground,
"total_point_count": total,
"ground_ratio": round(ground / total, 4) if total else 0.0,
}
return summary