feat(640-1): 라이다 기본 고름 — 확정 implicit · 평활 끔 · 분류 덮임 조건 · implicit 기본 필터에만
- 확정 기본 표현 implicit · 평활 False · WF1 자동 짓기에 확정 표현 더함(SURFACE_AUTO_METHODS 는 dtm 그대로) - 「분류 있음」 자동 고름 = 비율 0.2 % + 10 m 칸 덮임 50 % · 덮임 저장본(classification_coverage.json · 점군 안 풂) - 분류 LAS 의 implicit 는 기본 필터에만 지음(csf 두 번 안 지음) - B04 화면 확정 없을 때 기본 표현 implicit - 지킴 시험 test_640_lidar_default Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
@@ -24,6 +24,7 @@ from config.config_system import (
|
||||
SEND_ANALYSIS_COMPLETION_EMAIL,
|
||||
SHEET_SURFACE_DEFAULT_METHOD,
|
||||
SURFACE_AUTO_METHODS,
|
||||
SURFACE_CONFIRM_DEFAULT_METHOD,
|
||||
SURFACE_CONTOUR_INTERVAL_M,
|
||||
)
|
||||
|
||||
@@ -67,7 +68,8 @@ async def trigger_wf1_analysis_and_email(
|
||||
# 자동 전처리는 기본 조합만 만든다 — 필터는 엔진이 입력 LAS를 보고 정하고
|
||||
# (분류 있으면 classification, 없으면 csf), 표현은 SURFACE_AUTO_METHODS.
|
||||
# 다른 조합은 관리자가 B04 드롭다운에서 요청할 때 만든다 (2026-09-01 사용자 확정).
|
||||
methods = list(SURFACE_AUTO_METHODS)
|
||||
# 확정 표현(implicit)도 함께 지음 — 엔진이 기본 필터에만 지음(요구 558).
|
||||
methods = list(dict.fromkeys([*SURFACE_AUTO_METHODS, SURFACE_CONFIRM_DEFAULT_METHOD]))
|
||||
params = {
|
||||
"input_file_id": str(input_file_id),
|
||||
"source_filters": None,
|
||||
@@ -176,10 +178,10 @@ async def trigger_wf1_analysis_and_email(
|
||||
else:
|
||||
# 자동 확정은 엔진이 정한 기본 필터를 그대로 따른다 — 이제 그 필터
|
||||
# 하나만 만들어 두므로 config 기본값과 어긋나면 확정할 모델이 없다.
|
||||
# 표현 · 평활은 config 확정 기본값(implicit · 평활 끔 — 요구 558).
|
||||
selection = surface_confirmation_defaults()
|
||||
if resolved_filters:
|
||||
selection["source_filter"] = resolved_filters[0]
|
||||
selection["method"] = methods[0]
|
||||
try:
|
||||
model_id = await find_surface_model_for_selection(
|
||||
connection, project_id, selection
|
||||
|
||||
@@ -279,7 +279,7 @@ def run_surface_analysis(
|
||||
_report(40, "ground_filter", "지면 필터 적용 중")
|
||||
# 자동 전처리는 목록을 넘기지 않는다 — 기본 필터는 입력 LAS를 보고 여기서 정한다.
|
||||
if not source_filters:
|
||||
source_filters = resolve_auto_source_filters(data)
|
||||
source_filters = resolve_auto_source_filters(data, processed_dir)
|
||||
logger.info("B04 자동 전처리 기본 필터: %s", ", ".join(source_filters))
|
||||
masks: dict[str, np.ndarray] = {}
|
||||
for filter_key in source_filters:
|
||||
|
||||
@@ -5,11 +5,18 @@
|
||||
쓸 수 있는지는 has_classified_ground()로 먼저 판정한다.
|
||||
"""
|
||||
|
||||
import json
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
import numpy as np
|
||||
|
||||
from config.config_system import SURFACE_CLASSIFIED_GROUND_MIN_RATIO
|
||||
from common_util.common_util_json import atomic_write_json
|
||||
from config.config_system import (
|
||||
SURFACE_CLASSIFIED_COVERAGE_CELL_M,
|
||||
SURFACE_CLASSIFIED_COVERAGE_MIN_RATIO,
|
||||
SURFACE_CLASSIFIED_GROUND_MIN_RATIO,
|
||||
)
|
||||
|
||||
# ASPRS 표준 지면 클래스 코드
|
||||
GROUND_CLASS_CODE = 2
|
||||
@@ -38,3 +45,51 @@ def has_classified_ground(
|
||||
return False
|
||||
ground = int(np.count_nonzero(classification == GROUND_CLASS_CODE))
|
||||
return ground / total >= minimum_ratio
|
||||
|
||||
|
||||
def classified_coverage(
|
||||
structured_data: dict[str, Any] | np.lib.npyio.NpzFile,
|
||||
cell_meters: float = SURFACE_CLASSIFIED_COVERAGE_CELL_M,
|
||||
) -> float:
|
||||
"""점 있는 칸 중 class 2 가 있는 칸의 비율 — 일부만 분류된 LAS 를 가려냄."""
|
||||
if "classification" not in structured_data:
|
||||
return 0.0
|
||||
classification = np.asarray(structured_data["classification"])
|
||||
xyz = structured_data["xyz"]
|
||||
bounds = np.asarray(structured_data["bounds"], dtype=np.float64)
|
||||
cell = max(float(cell_meters), 0.1)
|
||||
cols = int((bounds[0, 1] - bounds[0, 0]) // cell) + 1
|
||||
rows = int((bounds[1, 1] - bounds[1, 0]) // cell) + 1
|
||||
occupied = np.zeros((rows, cols), dtype=bool)
|
||||
ground = np.zeros((rows, cols), dtype=bool)
|
||||
for start in range(0, len(classification), 5_000_000):
|
||||
points = np.asarray(xyz[start : start + 5_000_000, :2], dtype=np.float64)
|
||||
gx = np.clip(((points[:, 0] - bounds[0, 0]) // cell).astype(np.int32), 0, cols - 1)
|
||||
gy = np.clip(((points[:, 1] - bounds[1, 0]) // cell).astype(np.int32), 0, rows - 1)
|
||||
occupied[gy, gx] = True
|
||||
is_ground = classification[start : start + len(points)] == GROUND_CLASS_CODE
|
||||
ground[gy[is_ground], gx[is_ground]] = True
|
||||
total = int(np.count_nonzero(occupied))
|
||||
return float(np.count_nonzero(ground)) / total if total else 0.0
|
||||
|
||||
|
||||
def has_classified_coverage(
|
||||
structured_data: dict[str, Any] | np.lib.npyio.NpzFile,
|
||||
processed_dir: Path | None = None,
|
||||
minimum_ratio: float = SURFACE_CLASSIFIED_COVERAGE_MIN_RATIO,
|
||||
) -> bool:
|
||||
"""분류 덮임 문턱 판정 — `processed_dir` 면 structured.npz 지문으로 저장본 재사용."""
|
||||
if processed_dir is None:
|
||||
return classified_coverage(structured_data) >= minimum_ratio
|
||||
stat = (processed_dir / "structured.npz").stat()
|
||||
stamp = [int(stat.st_size), int(stat.st_mtime_ns)]
|
||||
path = processed_dir / "classification_coverage.json"
|
||||
try:
|
||||
stored = json.loads(path.read_text(encoding="utf-8"))
|
||||
coverage = float(stored["coverage"]) if stored.get("structured") == stamp else None
|
||||
except (OSError, ValueError, KeyError, TypeError, AttributeError):
|
||||
coverage = None
|
||||
if coverage is None:
|
||||
coverage = classified_coverage(structured_data)
|
||||
atomic_write_json(path, {"structured": stamp, "coverage": coverage})
|
||||
return coverage >= minimum_ratio
|
||||
|
||||
@@ -5,12 +5,14 @@
|
||||
source_filters를 따른다.
|
||||
"""
|
||||
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
import numpy as np
|
||||
|
||||
from B04_PreProcess.B04_PreProcess_Engine_Filter_Classification import (
|
||||
filter_classification,
|
||||
has_classified_coverage,
|
||||
has_classified_ground,
|
||||
)
|
||||
from B04_PreProcess.B04_PreProcess_Engine_Filter_CSF import filter_csf
|
||||
@@ -45,14 +47,18 @@ def usable_conditional_filters(
|
||||
|
||||
def resolve_auto_source_filters(
|
||||
structured_data: dict[str, Any] | np.lib.npyio.NpzFile,
|
||||
processed_dir: Path | None = None,
|
||||
) -> list[str]:
|
||||
"""자동 전처리가 만들 필터를 입력 LAS를 보고 정한다. 첫 항목이 기본 확정값이다.
|
||||
|
||||
업체가 이미 지면을 분류해 놨으면 그것을 기본으로 쓴다 — 계산이 없고 품질도 가장
|
||||
안정적이다. **csf는 분류가 있어도 함께 만든다** — csf 처리 결과를 눈으로 검증할 수
|
||||
있어야 하기 때문이다 (2026-09-01 사용자 확정). 분류가 없으면 csf 하나만 남는다.
|
||||
「분류 있음」 = 비율 문턱 + 덮임(일부만 분류된 LAS 거름) — 자동 고름에만(요구 558).
|
||||
"""
|
||||
usable = usable_conditional_filters(structured_data)
|
||||
if usable and not has_classified_coverage(structured_data, processed_dir):
|
||||
usable = []
|
||||
return [*usable[:1], SURFACE_AUTO_FALLBACK_FILTER]
|
||||
|
||||
|
||||
|
||||
@@ -190,6 +190,8 @@ def _build_all_terrain_models(
|
||||
methods_list.remove("dtm")
|
||||
methods_list.insert(0, "dtm")
|
||||
methods = tuple(methods_list)
|
||||
# 무거운 implicit 은 기본 필터(첫 필터)에만 — 분류 LAS 의 csf 까지 두 번 짓지 않음(요구 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"]
|
||||
@@ -211,7 +213,7 @@ def _build_all_terrain_models(
|
||||
}
|
||||
started_at = time.monotonic()
|
||||
timeout = max(0, int(config.get("sync_timeout_seconds", 0)))
|
||||
total_units = max(1, len(filters) * len(methods))
|
||||
total_units = max(1, len(methods) + (len(filters) - 1) * len(later_methods))
|
||||
done_units = 0
|
||||
failures = 0
|
||||
|
||||
@@ -229,7 +231,7 @@ def _build_all_terrain_models(
|
||||
)
|
||||
filter_entry["source_point_count"] = context.source_count
|
||||
|
||||
for method in methods:
|
||||
for method in later_methods if filter_index else methods:
|
||||
stem = f"{method}_{filter_key}"
|
||||
entry = filter_entry["methods"].get(method, {})
|
||||
|
||||
|
||||
@@ -54,7 +54,7 @@ import { currentProjectId, guardProjectLeave } from "../A00_Common/b_project_swi
|
||||
const SOURCE_FILTERS = ["classification", "grid_min_z", "csf", "pmf"] as const;
|
||||
const MODEL_METHODS = ["tin", "dtm", "nurbs", "implicit", "meshfree"] as const;
|
||||
const DEFAULT_FILTER = "csf";
|
||||
const DEFAULT_METHOD = "dtm";
|
||||
const DEFAULT_METHOD = "implicit"; // 확정 기본 표현(요구 558) — 확정본이 있으면 그것이 먼저
|
||||
const ROUTE_STAGE = ROUTES.B05_PROFILE;
|
||||
// 도엽 서피스 보간 방식 버튼 순서 — 백엔드 SHEET_SURFACE_METHODS와 같은 차례로 둔다.
|
||||
const SHEET_METHOD_ORDER = ["tin_sheet", "tin", "biharmonic", "anudem", "multires", "laplace"];
|
||||
@@ -494,7 +494,7 @@ export async function renderB04Surface(root: HTMLElement): Promise<void> {
|
||||
]);
|
||||
models = modelResponse.models;
|
||||
// 확정본과 같은 조합에서 시작해야 B05와 같은 파일을 보고, 보관함도 한 벌만 쓴다.
|
||||
// 확정 이력이 없을 때만 개발 기본값(csf·dtm)으로 둔다.
|
||||
// 확정 이력이 없을 때만 개발 기본값(csf·implicit)으로 둔다.
|
||||
if (confirmed.model_id) {
|
||||
if (confirmed.source_filter) filterGroup.select.value = confirmed.source_filter;
|
||||
if (confirmed.method) methodGroup.select.value = confirmed.method;
|
||||
|
||||
@@ -63,6 +63,11 @@ SURFACE_RANSAC_SEED = int(os.getenv("SURFACE_RANSAC_SEED", "42"))
|
||||
SURFACE_CLASSIFIED_GROUND_MIN_RATIO = float(
|
||||
os.getenv("SURFACE_CLASSIFIED_GROUND_MIN_RATIO", "0.002")
|
||||
)
|
||||
# 자동 기본 고름만 — 점 있는 칸(한 변 아래 m) 중 class 2 가 있는 칸이 이 비율 이상이어야 분류 있음.
|
||||
SURFACE_CLASSIFIED_COVERAGE_CELL_M = float(os.getenv("SURFACE_CLASSIFIED_COVERAGE_CELL_M", "10.0"))
|
||||
SURFACE_CLASSIFIED_COVERAGE_MIN_RATIO = float(
|
||||
os.getenv("SURFACE_CLASSIFIED_COVERAGE_MIN_RATIO", "0.5")
|
||||
)
|
||||
# 지면점 비율이 이 값 미만이면 필터가 사실상 실패한 것으로 보고 WARNING을 남긴다.
|
||||
SURFACE_GROUND_RATIO_WARN = float(os.getenv("SURFACE_GROUND_RATIO_WARN", "0.01"))
|
||||
|
||||
@@ -166,9 +171,10 @@ SHEET_SURFACE_AUTO_METHODS = [
|
||||
|
||||
# 일반 사용자 WF1 자동 확정 기본값
|
||||
SURFACE_CONFIRM_DEFAULT_FILTER = os.getenv("SURFACE_CONFIRM_DEFAULT_FILTER", "csf")
|
||||
SURFACE_CONFIRM_DEFAULT_METHOD = os.getenv("SURFACE_CONFIRM_DEFAULT_METHOD", "dtm")
|
||||
# 확정 표현 = implicit · 평활 끔 — WF1 자동 짓기도 이 표현을 기본 필터에 지음(요구 558).
|
||||
SURFACE_CONFIRM_DEFAULT_METHOD = os.getenv("SURFACE_CONFIRM_DEFAULT_METHOD", "implicit")
|
||||
SURFACE_CONFIRM_DEFAULT_SMOOTH = (
|
||||
os.getenv("SURFACE_CONFIRM_DEFAULT_SMOOTH", "True").lower() == "true"
|
||||
os.getenv("SURFACE_CONFIRM_DEFAULT_SMOOTH", "False").lower() == "true"
|
||||
)
|
||||
|
||||
|
||||
|
||||
@@ -0,0 +1,81 @@
|
||||
"""640 라이다 기본 고름 · implicit 점 수 · 격자 칸 뽑기 · 지문 · 틈 닫기 (요구 558).
|
||||
|
||||
서버 함수만 — 작은 합성 점군 · 임시 폴더.
|
||||
"""
|
||||
|
||||
from pathlib import Path
|
||||
|
||||
import numpy as np
|
||||
|
||||
from B04_PreProcess.B04_PreProcess_Engine_Filter_Classification import (
|
||||
classified_coverage,
|
||||
has_classified_coverage,
|
||||
)
|
||||
from B04_PreProcess.B04_PreProcess_Engine_Ground import resolve_auto_source_filters
|
||||
from B04_PreProcess.B04_PreProcess_Engine_Pipeline import build_all_terrain_models
|
||||
from config import config_system as cfg
|
||||
|
||||
|
||||
def _plane(size_m: float = 60.0, spacing_m: float = 0.5) -> np.ndarray:
|
||||
axis = np.arange(0.0, size_m, spacing_m)
|
||||
x, y = np.meshgrid(axis, axis)
|
||||
z = 100.0 + 0.2 * x + 0.1 * np.sin(y / 5.0)
|
||||
return np.column_stack([x.ravel(), y.ravel(), z.ravel()])
|
||||
|
||||
|
||||
def _structured(points: np.ndarray, classification: np.ndarray | None = None) -> dict:
|
||||
data = {"xyz": points, "bounds": np.array([points.min(axis=0), points.max(axis=0)]).T}
|
||||
if classification is not None:
|
||||
data["classification"] = classification
|
||||
return data
|
||||
|
||||
|
||||
def test_확정_기본값은_implicit_평활_끔() -> None:
|
||||
assert cfg.SURFACE_CONFIRM_DEFAULT_METHOD == "implicit"
|
||||
assert cfg.SURFACE_CONFIRM_DEFAULT_SMOOTH is False
|
||||
assert cfg.SURFACE_AUTO_METHODS == ("dtm",) # D03 자체 라이다는 dtm 만 그대로
|
||||
|
||||
|
||||
def test_일부만_분류된_LAS_는_자동_기본에서_csf() -> None:
|
||||
points = _plane(100.0)
|
||||
partial = np.ones(len(points), dtype=np.uint8)
|
||||
partial[(points[:, 0] < 30) & (np.arange(len(points)) % 4 == 0)] = 2 # 비율 7 % · 덮임 30 %
|
||||
assert classified_coverage(_structured(points, partial)) < 0.5
|
||||
assert resolve_auto_source_filters(_structured(points, partial)) == ["csf"]
|
||||
|
||||
spread = np.ones(len(points), dtype=np.uint8)
|
||||
spread[::37] = 2 # 비율 2.7 % · 덮임 100 %
|
||||
assert classified_coverage(_structured(points, spread)) == 1.0
|
||||
assert resolve_auto_source_filters(_structured(points, spread)) == ["classification", "csf"]
|
||||
|
||||
|
||||
def test_덮임은_저장본을_다시_씀(tmp_path: Path) -> None:
|
||||
points = _plane(40.0)
|
||||
classes = np.full(len(points), 2, dtype=np.uint8)
|
||||
(tmp_path / "structured.npz").write_bytes(b"x")
|
||||
assert has_classified_coverage(_structured(points, classes), tmp_path)
|
||||
assert (tmp_path / "classification_coverage.json").is_file()
|
||||
|
||||
class _NoXyz(dict):
|
||||
def __getitem__(self, key): # type: ignore[no-untyped-def]
|
||||
assert key != "xyz", "저장본이 맞으면 점군을 풀지 않아야 함"
|
||||
return dict.__getitem__(self, key)
|
||||
|
||||
assert has_classified_coverage(_NoXyz(_structured(points, classes)), tmp_path)
|
||||
|
||||
|
||||
def _config(**override: object) -> dict:
|
||||
config = cfg.build_surface_model_config()
|
||||
config.update({"source_filters": ["a", "b"], "precompute": ["dtm", "implicit"]})
|
||||
config.update(override)
|
||||
return config
|
||||
|
||||
|
||||
def test_implicit_은_기본_필터에만(tmp_path: Path) -> None:
|
||||
points = _plane()
|
||||
data = _structured(points)
|
||||
masks = {"a": np.ones(len(points), bool), "b": np.arange(len(points)) % 2 == 0}
|
||||
manifest = build_all_terrain_models(data, masks, tmp_path, _config())
|
||||
assert set(manifest["source_filters"]["a"]["methods"]) == {"dtm", "implicit"}
|
||||
assert set(manifest["source_filters"]["b"]["methods"]) == {"dtm"}
|
||||
assert not (tmp_path / "implicit_b.npz").exists()
|
||||
Reference in New Issue
Block a user