Spaces:
Sleeping
Sleeping
File size: 3,234 Bytes
160b215 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 | import numpy as np
import pandas as pd
from config import Config
class MetadataPreprocessor:
"""Fits and applies preprocessing for tabular metadata columns.
Numerical cols : median imputation → z-score standardisation (pandas/numpy)
Categorical cols: 'Unknown' imputation → one-hot encoding aligned to train
categories so unseen val categories are silently zeroed out.
Replaces sklearn SimpleImputer + StandardScaler + OneHotEncoder +
ColumnTransformer + Pipeline with plain pandas/numpy — no sklearn needed here.
"""
def __init__(self):
self._num_medians: dict = {}
self._num_means: dict = {}
self._num_stds: dict = {}
self._cat_categories: dict = {} # col -> sorted list of known categories
self._feature_cols: list = []
self.num_output_features: int = 0
def fit(self, train_df: pd.DataFrame) -> "MetadataPreprocessor":
meta = Config.get_metadata_config()
# Numerical: median for imputation, mean/std for scaling (fit on train only)
for col in meta['numerical_cols']:
s = pd.to_numeric(train_df[col], errors='coerce')
self._num_medians[col] = s.median()
filled = s.fillna(self._num_medians[col])
self._num_means[col] = float(filled.mean())
self._num_stds[col] = float(filled.std()) or 1.0 # avoid /0
# Categorical: record known categories for consistent OHE column set
for col in meta['categorical_cols']:
self._cat_categories[col] = sorted(
train_df[col].fillna('Unknown').astype(str).unique().tolist()
)
# Lock in the final column order using a single-row dry run
self._feature_cols = self._build_features(train_df.iloc[:1]).columns.tolist()
self.num_output_features = len(self._feature_cols)
print(f"Metadata preprocessor fitted. Output features: {self.num_output_features}")
return self
def transform(self, df: pd.DataFrame) -> np.ndarray:
features = self._build_features(df)
return features.reindex(columns=self._feature_cols, fill_value=0).values.astype(np.float32)
def _build_features(self, df: pd.DataFrame) -> pd.DataFrame:
meta = Config.get_metadata_config()
frames = []
# Numerical: impute then standardise
for col in meta['numerical_cols']:
s = pd.to_numeric(df[col], errors='coerce').fillna(self._num_medians.get(col, 0))
frames.append(
((s - self._num_means.get(col, 0)) / self._num_stds.get(col, 1))
.rename(col).reset_index(drop=True)
)
# Categorical: impute → OHE → align to known categories (unknowns → 0)
for col in meta['categorical_cols']:
filled = df[col].fillna('Unknown').astype(str).reset_index(drop=True)
dummies = pd.get_dummies(filled, prefix=col)
known_cols = [f"{col}_{cat}" for cat in self._cat_categories.get(col, [])]
for c in known_cols:
if c not in dummies:
dummies[c] = 0
frames.append(dummies[known_cols])
return pd.concat(frames, axis=1)
|