src/preprocess.py

"""Schema-aware imputation, encoding, and scaling."""

from __future__ import annotations

from dataclasses import dataclass
from typing import Iterator

import numpy as np
import pandas as pd
from sklearn.compose import ColumnTransformer
from sklearn.impute import SimpleImputer
from sklearn.pipeline import Pipeline
from sklearn.preprocessing import OneHotEncoder, StandardScaler

from .adapters.base import CANONICAL_ID_COL, DatasetSchema


@dataclass
class PreprocessResult:
    ids: np.ndarray
    X: np.ndarray
    transformer: ColumnTransformer
    scaler: ColumnTransformer
    feature_names: list[str]
    clean_features: pd.DataFrame
    keep_mask: np.ndarray = None  # type: ignore[assignment]
    """Boolean mask over the transformer's raw output, marking finite + non-constant
    columns. Live-predict path applies this to ``transformer.transform(new_row)``
    so the new feature vector lines up with the training X exactly."""

    def __iter__(self) -> Iterator[object]:
        """Preserve the old six-value unpacking API used by tests/scripts."""
        yield self.ids
        yield self.X
        yield self.transformer
        yield self.scaler
        yield self.feature_names
        yield self.clean_features


def _legacy_schema(feature_matrix: pd.DataFrame) -> DatasetSchema:
    if CANONICAL_ID_COL not in feature_matrix.columns:
        raise ValueError("feature_matrix must contain an id_student column")
    numeric_cols = [
        col for col in feature_matrix.columns
        if col != CANONICAL_ID_COL
    ]
    return DatasetSchema(
        dataset_name="legacy",
        adapter_name="legacy",
        source_id_col=CANONICAL_ID_COL,
        feature_cols=numeric_cols,
        numeric_feature_cols=numeric_cols,
    )


def _feature_name(name: str) -> str:
    for prefix in ["num__", "cat__"]:
        if name.startswith(prefix):
            return name[len(prefix):]
    return name


def preprocess(
    feature_matrix: pd.DataFrame,
    schema: DatasetSchema | None = None,
    family_weights: dict[str, float] | None = None,
) -> PreprocessResult:
    """Return IDs, scaled feature matrix, fitted transformer, and clean features."""
    schema = schema or _legacy_schema(feature_matrix)
    if schema.id_col not in feature_matrix.columns:
        raise ValueError(f"feature_matrix must contain a {schema.id_col!r} column")

    ids = feature_matrix[schema.id_col].to_numpy()
    numeric_cols = [col for col in schema.numeric_feature_cols if col in feature_matrix.columns]
    categorical_cols = [col for col in schema.categorical_feature_cols if col in feature_matrix.columns]
    selected_cols = list(dict.fromkeys(numeric_cols + categorical_cols))
    if not selected_cols:
        raise ValueError("No clustering features are available after applying dataset schema")

    work = feature_matrix[selected_cols].copy()
    for col in numeric_cols:
        work[col] = pd.to_numeric(work[col], errors="coerce")

    transformers: list[tuple[str, Pipeline, list[str]]] = []
    if numeric_cols:
        transformers.append(
            (
                "num",
                Pipeline(
                    steps=[
                        ("imputer", SimpleImputer(strategy="median")),
                        ("scaler", StandardScaler()),
                    ]
                ),
                numeric_cols,
            )
        )
    if categorical_cols:
        transformers.append(
            (
                "cat",
                Pipeline(
                    steps=[
                        ("imputer", SimpleImputer(strategy="most_frequent")),
                        ("onehot", OneHotEncoder(handle_unknown="ignore", sparse_output=False)),
                        # Z-scoring one-hot columns is intentional: all encoded
                        # dimensions then contribute comparably to distance-based clustering.
                        ("scaler", StandardScaler()),
                    ]
                ),
                categorical_cols,
            )
        )

    transformer = ColumnTransformer(transformers=transformers, remainder="drop", verbose_feature_names_out=True)
    X = transformer.fit_transform(work)
    X = np.asarray(X, dtype=float)
    feature_names = [_feature_name(name) for name in transformer.get_feature_names_out()]

    finite_mask = np.isfinite(X).all(axis=0)
    nonzero_mask = np.nanvar(X, axis=0) > 0
    keep = finite_mask & nonzero_mask
    if not keep.any():
        raise ValueError("No non-constant features remain after preprocessing")
    X = X[:, keep]
    feature_names = [name for name, keep_col in zip(feature_names, keep) if keep_col]

    encoded = pd.DataFrame(X, columns=feature_names, index=feature_matrix.index)
    if family_weights:
        for prefix, weight in family_weights.items():
            cols = [col for col in encoded.columns if col.startswith(prefix)]
            if cols:
                encoded.loc[:, cols] *= float(weight)
        X = encoded.to_numpy()

    clean_features = work.copy()
    if numeric_cols:
        num_imputer = SimpleImputer(strategy="median")
        clean_features[numeric_cols] = num_imputer.fit_transform(work[numeric_cols])
    for col in categorical_cols:
        mode = work[col].mode(dropna=True)
        fill = mode.iloc[0] if not mode.empty else "missing"
        clean_features[col] = work[col].fillna(fill).astype(str)
    kept_source_cols = [
        col for col in selected_cols
        if col in feature_names or any(name.startswith(f"{col}_") for name in feature_names)
    ]
    clean_features = clean_features[kept_source_cols]

    return PreprocessResult(
        ids=ids,
        X=X,
        transformer=transformer,
        scaler=transformer,
        feature_names=feature_names,
        clean_features=clean_features.reset_index(drop=True),
        keep_mask=keep,
    )