IDKHowToCodeFr/tinyml-backend

1import os2import io3import pandas as pd4from typing import Optional5 6class RetrainResult:7    def __init__(self, success: bool, message: str):8        self.success = success9        self.message = message10 11class MLOpsEngine:12    """13    Deep Module encapsulating the MLOps Lifecycle:14    - Maintains the active ensemble model.15    - Handles schema validation and data ingestion natively with pandas.16    - Orchestrates model retraining and hot-swapping.17    """18    def __init__(self, dataset_path: str):19        self.dataset_path = dataset_path20        self.active_ensemble = None21        self._load_active_models()22 23    def _load_active_models(self):24        try:25            from ensemble import EnsembleModel26            self.active_ensemble = EnsembleModel()27        except Exception as e:28            print(f"MLOps: Model loading failed (likely untrained): {e}")29 30    def get_ensemble(self):31        return self.active_ensemble32 33    def ingest_batch_sync(self, csv_bytes: bytes) -> RetrainResult:34        try:35            new_df = pd.read_csv(io.BytesIO(csv_bytes))36            new_df.columns = [c.strip() for c in new_df.columns]37            38            if os.path.exists(self.dataset_path):39                existing_df = pd.read_csv(self.dataset_path, encoding='utf-8')40                existing_df.columns = [c.strip() for c in existing_df.columns]41                42                # Check for schema mismatch before concatenation43                missing = set(existing_df.columns) - set(new_df.columns)44                if missing:45                    return RetrainResult(False, f"Schema mismatch. Missing columns: {list(missing)}")46                47                # Ponytail shrink: pandas join="inner" drops mismatched extra columns cleanly48                combined_df = pd.concat([existing_df, new_df], join="inner", ignore_index=True)49            else:50                combined_df = new_df51                52            os.makedirs(os.path.dirname(self.dataset_path), exist_ok=True)53            combined_df.to_csv(self.dataset_path, index=False, encoding='utf-8')54            55            # Re-train models synchronously56            from models import train_models57            train_models()58            59            # Hot-swap the live models60            self._load_active_models()61            62            return RetrainResult(True, f"Dataset updated (now {len(combined_df)} records) and ensemble hot-swapped!")63        except Exception as e:64            return RetrainResult(False, f"Retraining failed: {str(e)}")65