Aislo/B04_PreProcess/B04_PreProcess_Schema.py

"""B04 지표면 분석 요청·응답 검증 모델."""

from typing import Any

from pydantic import BaseModel, ConfigDict, Field

from config.config_system import (
    SURFACE_MODEL_PRECOMPUTE,
    SURFACE_MODEL_SOURCE_FILTERS,
)

# classification은 LAS가 지면분류를 싣고 있을 때 엔진이 자동으로 붙이지만,
# 재계산 요청에서 이름으로 지정하는 것도 허용한다.
_ALLOWED_FILTERS = set(SURFACE_MODEL_SOURCE_FILTERS) | {
    "ransac",
    "classification",
    "classification_refine",
    "ptd_refine",
}
_ALLOWED_METHODS = set(SURFACE_MODEL_PRECOMPUTE)


class SurfaceAnalyzeRequest(BaseModel):
    """지표면 분석 실행 요청."""

    model_config = ConfigDict(extra="forbid")

    input_file_id: int = Field(gt=0, description="구조화 대상 원본 LAS input_files.id")
    source_filters: list[str] | None = Field(
        default=None, description="실행할 지면 필터 (미지정 시 config 기본값)"
    )
    methods: list[str] | None = Field(
        default=None, description="생성할 지표면 표현 (미지정 시 config 기본값)"
    )
    force: bool = Field(default=False, description="캐시 무시 후 강제 재계산")

    def resolved_filters(self) -> list[str]:
        filters = self.source_filters or list(SURFACE_MODEL_SOURCE_FILTERS)
        invalid = [f for f in filters if f not in _ALLOWED_FILTERS]
        if invalid:
            raise ValueError(f"허용되지 않은 지면 필터입니다: {invalid}")
        return filters

    def resolved_methods(self) -> list[str]:
        methods = self.methods or list(SURFACE_MODEL_PRECOMPUTE)
        invalid = [m for m in methods if m not in _ALLOWED_METHODS]
        if invalid:
            raise ValueError(f"허용되지 않은 지표면 표현입니다: {invalid}")
        return methods


class SurfaceConfirmRequest(BaseModel):
    """사용자가 선택한 지표면 모델 확정 요청."""

    model_config = ConfigDict(extra="forbid")

    model_id: int = Field(gt=0, description="확정할 surface_models.id")
    smooth: bool | None = Field(default=None, description="확정 시 적용한 스무딩 여부")
    contour_interval_m: float | None = Field(
        default=None,
        gt=0,
        description="확정 시 적용한 등고선 간격(m)",
    )


class SurfaceConfirmResponse(BaseModel):
    """지표면 모델 확정 결과."""

    status: str = "success"
    project_id: str
    model_id: int
    confirmed: bool = True


class SurfaceModelSummary(BaseModel):
    """저장된 지표면 모델 요약."""

    id: int
    model_type: str
    status: str
    resolution_m: float | None = None
    model_file_path: str | None = None
    generation_params: dict[str, Any] | None = None
    created_at: str | None = None


class SurfaceInputFileSummary(BaseModel):
    """WF1 분석 대상으로 선택 가능한 LAS/LAZ 입력 파일 요약."""

    id: int
    file_type: str
    original_filename: str
    raw_file_path: str
    file_size_mb: float | None = None
    crs_epsg: int | None = None
    status: str | None = None
    created_at: str | None = None


class SurfaceInputFileListResponse(BaseModel):
    """프로젝트 지표면 분석 입력 파일 목록 응답."""

    status: str = "success"
    project_id: str
    files: list[SurfaceInputFileSummary]


class SurfacePointCloudSampleResponse(BaseModel):
    """B04 3D 미리보기용 포인트클라우드 샘플."""

    status: str = "success"
    project_id: str
    point_count: int
    sampled_count: int
    bounds: dict[str, float]
    points: list[list[float]]
    rgb: list[list[int]] | None = None


class SurfaceConfirmedResponse(BaseModel):
    """확정 지표면 요약 — 화면 진입 판정·준비화면·B05가 공통으로 쓰는 단일 출처.

    포인트 배열 없이 확정값과 지형 가장자리만 담아 수 KB로 유지한다.
    signature는 확정 구성이 바뀌었는지 프론트가 한 줄로 비교하기 위한 값이다.
    """

    status: str = "success"
    project_id: str
    model_id: int | None = None
    source_filter: str | None = None
    method: str | None = None
    smooth: bool | None = None
    contour_interval_m: float | None = None
    signature: str
    point_count: int | None = None
    bounds: dict[str, float] | None = None
    # 계획노선(B03 CSV)의 평면 범위. 지도 초기 화면을 도로 기준으로 맞출 때 쓴다.
    route_bounds: dict[str, float] | None = None


class SurfaceGroundStatsResponse(BaseModel):
    """필터별 지면 포인트 통계 응답."""

    status: str = "success"
    project_id: str
    filters: dict[str, dict[str, Any]]


class SurfaceAnalyzeResponse(BaseModel):
    """지표면 분석 실행 결과."""

    status: str = "success"
    project_id: str
    ground_summary: dict[str, Any]
    manifest_status: str
    surface_model_ids: list[int]


class SurfaceModelListResponse(BaseModel):
    """프로젝트 지표면 모델 목록 응답."""

    status: str = "success"
    project_id: str
    models: list[SurfaceModelSummary]