143 lines
6.1 KiB
Python
143 lines
6.1 KiB
Python
"""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 AppConfig, FREQ_COLS
|
|
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))
|