Spaces:
Runtime error
Runtime error
| import os | |
| import io | |
| import threading | |
| from dataclasses import dataclass | |
| import pandas as pd | |
| from typing import Optional | |
| class RetrainResult: | |
| success: bool | |
| message: str | |
| CANONICAL_COLUMNS = [ | |
| 'Patient Number', | |
| 'Heart Rate (bpm)', | |
| 'SpO2 Level (%)', | |
| 'Systolic Blood Pressure (mmHg)', | |
| 'Diastolic Blood Pressure (mmHg)', | |
| 'Body Temperature (°C)', | |
| 'Fall Detection', | |
| 'Predicted Disease', | |
| 'Data Accuracy (%)', | |
| 'Heart Rate Alert', | |
| 'SpO2 Level Alert', | |
| 'Blood Pressure Alert', | |
| 'Temperature Alert' | |
| ] | |
| # Case-insensitive alias map for telemetry and label columns | |
| COLUMN_ALIASES = { | |
| 'heart rate (bpm)': 'Heart Rate (bpm)', | |
| 'heart rate': 'Heart Rate (bpm)', | |
| 'heart_rate': 'Heart Rate (bpm)', | |
| 'heartrate': 'Heart Rate (bpm)', | |
| 'hr': 'Heart Rate (bpm)', | |
| 'spo2 level (%)': 'SpO2 Level (%)', | |
| 'spo2 level': 'SpO2 Level (%)', | |
| 'spo2_level': 'SpO2 Level (%)', | |
| 'spo2': 'SpO2 Level (%)', | |
| 'oxygen': 'SpO2 Level (%)', | |
| 'o2_saturation': 'SpO2 Level (%)', | |
| 'systolic blood pressure (mmhg)': 'Systolic Blood Pressure (mmHg)', | |
| 'systolic blood pressure': 'Systolic Blood Pressure (mmHg)', | |
| 'systolic_bp': 'Systolic Blood Pressure (mmHg)', | |
| 'systolic bp': 'Systolic Blood Pressure (mmHg)', | |
| 'sys_bp': 'Systolic Blood Pressure (mmHg)', | |
| 'systolic': 'Systolic Blood Pressure (mmHg)', | |
| 'diastolic blood pressure (mmhg)': 'Diastolic Blood Pressure (mmHg)', | |
| 'diastolic blood pressure': 'Diastolic Blood Pressure (mmHg)', | |
| 'diastolic_bp': 'Diastolic Blood Pressure (mmHg)', | |
| 'diastolic bp': 'Diastolic Blood Pressure (mmHg)', | |
| 'dia_bp': 'Diastolic Blood Pressure (mmHg)', | |
| 'diastolic': 'Diastolic Blood Pressure (mmHg)', | |
| 'body temperature (°c)': 'Body Temperature (°C)', | |
| 'body temperature': 'Body Temperature (°C)', | |
| 'body_temp': 'Body Temperature (°C)', | |
| 'temperature': 'Body Temperature (°C)', | |
| 'temp': 'Body Temperature (°C)', | |
| 'predicted disease': 'Predicted Disease', | |
| 'predicted_disease': 'Predicted Disease', | |
| 'disease': 'Predicted Disease', | |
| 'condition': 'Predicted Disease', | |
| 'diagnosis': 'Predicted Disease', | |
| 'label': 'Predicted Disease', | |
| } | |
| class MLOpsEngine: | |
| """ | |
| Encapsulates the MLOps Lifecycle: | |
| - Normalizes and sanitizes telemetry schemas with alias resolution. | |
| - Handles data ingestion and schema validation. | |
| - Orchestrates model retraining and hot-swapping. | |
| """ | |
| def __init__(self, dataset_path: str): | |
| self.dataset_path = dataset_path | |
| self.active_ensemble = None | |
| self._lock = threading.Lock() | |
| self._load_active_models() | |
| def _load_active_models(self): | |
| try: | |
| from ensemble import EnsembleModel | |
| self.active_ensemble = EnsembleModel() | |
| except Exception as e: | |
| print(f"MLOps: Model loading failed (likely untrained): {e}") | |
| def get_ensemble(self): | |
| return self.active_ensemble | |
| def normalize_batch(self, df: pd.DataFrame) -> pd.DataFrame: | |
| """Standardizes column names, patches encoding issues, and fills missing non-essential metadata.""" | |
| # Clean column names | |
| cleaned_cols = {} | |
| for col in df.columns: | |
| normalized_key = str(col).strip().lower().replace('\ufffd', '°') | |
| matched_canonical = COLUMN_ALIASES.get(normalized_key, col) | |
| cleaned_cols[col] = matched_canonical | |
| df = df.rename(columns=cleaned_cols) | |
| # Check required continuous physiological signals | |
| required_signals = [ | |
| 'Heart Rate (bpm)', | |
| 'SpO2 Level (%)', | |
| 'Systolic Blood Pressure (mmHg)', | |
| 'Diastolic Blood Pressure (mmHg)', | |
| 'Body Temperature (°C)' | |
| ] | |
| missing_signals = [s for s in required_signals if s not in df.columns] | |
| if missing_signals: | |
| raise ValueError(f"Missing required physiological signals: {', '.join(missing_signals)}") | |
| # Ensure target label exists; default to 'Healthy' if not specified | |
| if 'Predicted Disease' not in df.columns: | |
| df['Predicted Disease'] = 'Healthy' | |
| # Auto-populate optional canonical fields with consistent values | |
| if 'Patient Number' not in df.columns: | |
| df['Patient Number'] = range(1, len(df) + 1) | |
| if 'Data Accuracy (%)' not in df.columns: | |
| df['Data Accuracy (%)'] = 100 | |
| if 'Fall Detection' not in df.columns: | |
| df['Fall Detection'] = 'No' | |
| if 'Heart Rate Alert' not in df.columns: | |
| df['Heart Rate Alert'] = 'NORMAL' | |
| if 'SpO2 Level Alert' not in df.columns: | |
| df['SpO2 Level Alert'] = 'NORMAL' | |
| if 'Blood Pressure Alert' not in df.columns: | |
| df['Blood Pressure Alert'] = 'NORMAL' | |
| if 'Temperature Alert' not in df.columns: | |
| df['Temperature Alert'] = 'NORMAL' | |
| # Ensure numeric types for continuous signals | |
| for sig in required_signals: | |
| df[sig] = pd.to_numeric(df[sig], errors='coerce') | |
| # Drop rows where all signals are NaN | |
| df = df.dropna(subset=required_signals, how='all') | |
| # Align columns to canonical ordering | |
| return df[[c for c in CANONICAL_COLUMNS if c in df.columns]] | |
| def ingest_batch_sync(self, csv_bytes: bytes) -> RetrainResult: | |
| with self._lock: | |
| try: | |
| raw_df = pd.read_csv(io.BytesIO(csv_bytes), encoding_errors='replace') | |
| new_df = self.normalize_batch(raw_df) | |
| if len(new_df) == 0: | |
| return RetrainResult(False, "Uploaded CSV contains no valid telemetry rows.") | |
| if os.path.exists(self.dataset_path): | |
| existing_df = pd.read_csv(self.dataset_path, encoding='utf-8', encoding_errors='replace') | |
| existing_df.columns = [c.replace('\ufffd', '°') for c in existing_df.columns] | |
| # Adjust Patient Number to maintain continuous increment | |
| if 'Patient Number' in existing_df.columns: | |
| start_id = int(existing_df['Patient Number'].max() or 0) + 1 | |
| new_df['Patient Number'] = range(start_id, start_id + len(new_df)) | |
| combined_df = pd.concat([existing_df, new_df], join="inner", ignore_index=True) | |
| else: | |
| combined_df = new_df | |
| os.makedirs(os.path.dirname(self.dataset_path), exist_ok=True) | |
| combined_df.to_csv(self.dataset_path, index=False, encoding='utf-8') | |
| # Re-train models synchronously | |
| from models import train_models | |
| train_models() | |
| # Hot-swap the live models | |
| self._load_active_models() | |
| return RetrainResult(True, f"Dataset successfully updated ({len(combined_df)} total observations) and ensemble hot-swapped.") | |
| except Exception as e: | |
| return RetrainResult(False, f"Ingestion failed: {str(e)}") | |