hackathon-ENGIN/engin/validation.py

"""Strict validation and normalization for inference data."""

from __future__ import annotations

from dataclasses import dataclass
from typing import Protocol

import numpy as np
import pandas as pd

from .config import FREQ_COLS, AppConfig
from .errors import InputDataError


@dataclass(frozen=True)
class ValidationResult:
    frame: pd.DataFrame
    warnings: tuple[str, ...] = ()


class FrameValidator(Protocol):
    def validate(self, frame: pd.DataFrame) -> ValidationResult: ...


class SpectrumFrameValidator:
    def __init__(self, config: AppConfig | None = None) -> None:
        self.config = config or AppConfig()

    def validate(self, frame: pd.DataFrame) -> ValidationResult:
        if not isinstance(frame, pd.DataFrame):
            raise InputDataError("Dane wejściowe nie są tabelą.")
        if frame.empty:
            raise InputDataError("Plik nie zawiera żadnych pomiarów.", code="empty_table")

        required = {"engine_id", "cylinder", "n_cylinders", *FREQ_COLS}
        missing_columns = sorted(required.difference(frame.columns))
        if missing_columns:
            raise InputDataError(
                "Brakuje wymaganych kolumn: " + ", ".join(missing_columns),
                hint="Oczekiwane są engine_id, cylinder, n_cylinders oraz mV_0...mV_20.",
                code="missing_columns",
            )

        clean = frame.copy()
        clean["engine_id"] = clean["engine_id"].astype("string").str.strip()
        if clean["engine_id"].isna().any() or clean["engine_id"].eq("").any():
            raise InputDataError("Kolumna engine_id zawiera puste wartości.", code="missing_engine_id")

        for column in ("cylinder", "n_cylinders"):
            numeric = pd.to_numeric(clean[column], errors="coerce")
            if numeric.isna().any() or not np.allclose(numeric, np.rint(numeric)):
                raise InputDataError(
                    f"Kolumna {column} musi zawierać liczby całkowite.",
                    code="invalid_identifier",
                )
            clean[column] = numeric.astype(int)

        invalid_sizes = sorted(
            set(clean["n_cylinders"]).difference(self.config.allowed_engine_sizes)
        )
        if invalid_sizes:
            raise InputDataError(
                f"Nieobsługiwane rozmiary silnika: {invalid_sizes}.",
                hint=f"Dozwolone wartości to {list(self.config.allowed_engine_sizes)}.",
                code="unsupported_engine_size",
            )

        if clean.duplicated(["engine_id", "cylinder"]).any():
            duplicated = clean.loc[
                clean.duplicated(["engine_id", "cylinder"], keep=False),
                ["engine_id", "cylinder"],
            ].head(3)
            examples = ", ".join(
                f"{row.engine_id}/C{row.cylinder}" for row in duplicated.itertuples()
            )
            raise InputDataError(
                f"Znaleziono zduplikowane cylindry: {examples}.",
                code="duplicate_cylinder",
            )

        original_spectra = clean[FREQ_COLS]
        numeric_spectra = original_spectra.apply(pd.to_numeric, errors="coerce")
        invalid_numeric = original_spectra.notna() & numeric_spectra.isna()
        if invalid_numeric.to_numpy().any():
            row_pos, col_pos = np.argwhere(invalid_numeric.to_numpy())[0]
            raise InputDataError(
                f"Wiersz {row_pos + 2}, kolumna {FREQ_COLS[col_pos]} zawiera tekst zamiast liczby.",
                code="non_numeric_spectrum",
            )
        finite_values = numeric_spectra.to_numpy(dtype=float)
        if np.isinf(finite_values).any():
            raise InputDataError("Widmo zawiera wartości nieskończone.", code="infinite_spectrum")
        if np.nanmin(finite_values) < 0:
            raise InputDataError(
                "Amplituda mV nie może być ujemna.",
                code="negative_amplitude",
            )
        clean[FREQ_COLS] = numeric_spectra

        bad_engines: list[str] = []
        for engine_id, engine in clean.groupby("engine_id", sort=False):
            declared_values = engine["n_cylinders"].unique()
            if len(declared_values) != 1:
                bad_engines.append(f"{engine_id}: niespójne n_cylinders")
                continue
            expected = int(declared_values[0])
            observed = sorted(engine["cylinder"].tolist())
            required_cylinders = list(range(1, expected + 1))
            if observed != required_cylinders:
                missing = sorted(set(required_cylinders).difference(observed))
                extra = sorted(set(observed).difference(required_cylinders))
                detail = []
                if missing:
                    detail.append(f"brak {missing}")
                if extra:
                    detail.append(f"nadmiarowe {extra}")
                bad_engines.append(f"{engine_id}: {'; '.join(detail)}")
        if bad_engines:
            raise InputDataError(
                "Niekompletne lub niespójne silniki: " + " | ".join(bad_engines[:5]),
                hint="Każdy silnik musi zawierać dokładnie cylindry 1...n_cylinders.",
                code="incomplete_engine",
            )

        missing_fraction = clean[FREQ_COLS].isna().mean(axis=1)
        excessive = missing_fraction > self.config.max_missing_fraction_per_cylinder
        if excessive.any():
            row = clean.loc[excessive, ["engine_id", "cylinder"]].iloc[0]
            percent = 100 * float(missing_fraction[excessive].iloc[0])
            raise InputDataError(
                f"{row.engine_id}/C{int(row.cylinder)} ma {percent:.0f}% brakujących pomiarów.",
                hint="Powtórz pomiar cylindra; model akceptuje maksymalnie 50% braków w wierszu.",
                code="too_many_missing_values",
            )

        warnings: list[str] = []
        total_missing_fraction = float(clean[FREQ_COLS].isna().mean().mean())
        if total_missing_fraction > 0.10:
            warnings.append(
                f"Dane zawierają {100 * total_missing_fraction:.1f}% braków; wynik może mieć niższą pewność."
            )
        return ValidationResult(clean.reset_index(drop=True), tuple(warnings))