""" ============================================================================= FUMD-AI Handover Forecaster -- self-contained preprocessing ============================================================================= Everything needed to turn a `dataset_labeled_w.csv` (the AI-ready output of the FUMD-AI preprocessing workflow's Step 6) into the exact `(n, 6, 23)` input this model was trained on -- with no dependency on the training package, on `joblib`, or on a specific scikit-learn version. All fitted parameters live in `preprocessing.json`, a plain-JSON export of the `scaler.joblib` / `label_encoders.joblib` shipped alongside it. Author(s): - Cristina Bernad (ORCID: 0000-0001-9537-415X) - Sonja Filiposka (ORCID: 0000-0003-0034-2855) - Katja Gilly (ORCID: 0000-0002-8985-0639) Funding: This work has been funded by the FUMD-AI project, an EOSC GRAVITY - Inter Project with Grant Number 25-EOSC-GRV-INTER-013. SPDX-License-Identifier: MIT """ from __future__ import annotations import json import os import numpy as np import pandas as pd HERE = os.path.dirname(os.path.abspath(__file__)) TIME_COLUMN = "t" GROUP_COLUMN = "veh_id" TARGET_COLUMN = "servingCell" CATEGORICAL_COLUMNS = ["lane"] LABEL_ONLY_COLUMNS = ["migration", "destination"] SEQUENCE_LENGTH = 6 FUTURE_STEPS = [1, 2, 3, 4, 5, 6, 7] NUM_BASE_STATIONS = 10 #: The 23 input features, in the order the scaler and the encoder LSTM expect. FEATURE_COLUMNS = [ "angle", "speed", "pos", "lane", "signals", "averageCqiDl", "distance", "measuredSinrDl", "measuredSinrUl", "rcvdSinrDl", "rlcDelayDl", "rlcPduDelayDl", "rlcThroughputDl", "servingCell", "servingCell-1", "servingCell-2", "servingCell-3", "servingCell-4", "servingCell-5", "servingCell-6", "servingCell-7", "x", "y", ] # --------------------------------------------------------------------------- # Fitted parameters # --------------------------------------------------------------------------- class Preprocessor: """ Applies the model's fitted lane encoding and standardisation. Transform-only by design: there is deliberately no `fit`. Re-fitting either the encoder or the scaler on new data would make its outputs incomparable with what the model was trained on (different lane->int codes, different z-score statistics) while still appearing to work. """ def __init__(self, params: dict): scaler = params["scaler"] self.mean = np.asarray(scaler["mean"], dtype="float64") self.scale = np.asarray(scaler["scale"], dtype="float64") self.n_features = int(scaler["n_features_in"]) if len(self.mean) != self.n_features or len(self.scale) != self.n_features: raise ValueError("preprocessing.json: scaler mean/scale length disagrees with n_features_in") self.encoders = {} for col, enc in params["label_encoders"].items(): classes = [str(c) for c in enc["classes"]] self.encoders[col] = { "index": {c: i for i, c in enumerate(classes)}, "unknown_code": int(enc["unknown_code"]), "n_classes": len(classes), } @classmethod def load(cls, path: str | None = None) -> "Preprocessor": path = path or os.path.join(HERE, "preprocessing.json") with open(path) as fh: return cls(json.load(fh)) # -- categorical ------------------------------------------------------ def encode_categoricals(self, df: pd.DataFrame) -> pd.DataFrame: """Map each categorical column's string values to their fitted codes. Values never seen at fit time map to that column's reserved unknown code instead of raising -- a road segment the training traffic simply never drove down is a legitimate input, not an error. """ out = df.copy() for col, enc in self.encoders.items(): if col not in out.columns: raise KeyError(f"expected categorical column {col!r} in the dataframe") codes = out[col].astype(str).map(enc["index"]) n_unknown = int(codes.isna().sum()) if n_unknown: print(f"encode_categoricals: {col}: {n_unknown} value(s) unseen in training " f"-> unknown code {enc['unknown_code']}") out[col] = codes.fillna(enc["unknown_code"]).astype(int) return out # -- scaling ---------------------------------------------------------- def scale_windows(self, X: np.ndarray) -> np.ndarray: """Z-score an (n, seq_len, n_features) array with the fitted statistics.""" X = np.asarray(X, dtype="float64") if X.ndim != 3 or X.shape[-1] != self.n_features: raise ValueError(f"expected (n, seq_len, {self.n_features}), got {X.shape}") flat = X.reshape(-1, self.n_features) flat = (flat - self.mean) / self.scale return flat.reshape(X.shape).astype("float32") # --------------------------------------------------------------------------- # Windowing # --------------------------------------------------------------------------- def build_sequences( df: pd.DataFrame, feature_columns: list[str] = FEATURE_COLUMNS, target_column: str = TARGET_COLUMN, sequence_length: int = SEQUENCE_LENGTH, future_steps: list[int] = FUTURE_STEPS, group_col: str = GROUP_COLUMN, time_col: str = TIME_COLUMN, label_columns: list[str] = LABEL_ONLY_COLUMNS, ): """ Build per-vehicle sliding windows, identical to how the model was trained. Returns ------- X : (n_windows, sequence_length, n_features) -- unscaled y : (n_windows, len(future_steps)) -- true serving cell per horizon extra : {label_col: (n_windows,)} -- labels at the first forecast step, for evaluation only; never model input. Windows never cross a vehicle boundary, and a vehicle with fewer than `sequence_length + max(future_steps)` rows contributes none. """ future_steps = list(future_steps) needed = sequence_length + max(future_steps) X_list, y_list = [], [] extra_lists = {c: [] for c in label_columns if c in df.columns} n_skipped = 0 has_target = target_column in df.columns for _, group in df.groupby(group_col, sort=False): group = group.sort_values(time_col).reset_index(drop=True) n = len(group) if n < needed: n_skipped += 1 continue feats = group[feature_columns].to_numpy(dtype=float) target = group[target_column].to_numpy() if has_target else None label_arrays = {c: group[c].to_numpy() for c in extra_lists} for i in range(n - needed): X_list.append(feats[i:i + sequence_length]) if has_target: y_list.append([target[i + sequence_length + s] for s in future_steps]) for c, arr in label_arrays.items(): extra_lists[c].append(arr[i + sequence_length + future_steps[0]]) if n_skipped: print(f"build_sequences: skipped {n_skipped} vehicle(s) with < {needed} rows") X = np.asarray(X_list, dtype=float) y = np.asarray(y_list, dtype=float) if has_target else None extra = {c: np.asarray(v) for c, v in extra_lists.items()} print(f"build_sequences: {X.shape[0]} windows, X {X.shape}") return X, y, extra def prepare( df: pd.DataFrame, preprocessor: Preprocessor | None = None, **kwargs, ): """Encode -> window -> scale, in one call. Returns `(X_scaled, y, extra)`.""" preprocessor = preprocessor or Preprocessor.load() df = df.sort_values([GROUP_COLUMN, TIME_COLUMN]).reset_index(drop=True) df = preprocessor.encode_categoricals(df) X, y, extra = build_sequences(df, **kwargs) return preprocessor.scale_windows(X), y, extra