pipeline_engine.py

import sys
import io
import time
import trace
import traceback
import psutil
import pandas as pd
import numpy as np
from sklearn.datasets import make_classification
from sklearn.model_selection import train_test_split
from sklearn.ensemble import RandomForestClassifier
from sklearn.metrics import accuracy_score

# Default healthy pipeline code string
DEFAULT_PIPELINE_CODE = """import numpy as np
import pandas as pd
from sklearn.datasets import make_classification
from sklearn.model_selection import train_test_split
from sklearn.ensemble import RandomForestClassifier
from sklearn.metrics import accuracy_score

def run_ml_pipeline():
    print("[Pipeline] Ingesting features and targets...")
    X_raw, y_raw = make_classification(
        n_samples=1200, n_features=10, n_informative=8,
        n_redundant=2, random_state=42
    )
    
    df = pd.DataFrame(X_raw, columns=[f"feat_{i}" for i in range(10)])
    df["target"] = y_raw
    
    print("[Pipeline] Preprocessing data and handling nulls...")
    # Clean data baseline
    df = df.dropna()
    
    X = df.drop(columns=["target"])
    y = df["target"]
    
    X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2, random_state=42)
    
    print("[Pipeline] Training Random Forest model...")
    model = RandomForestClassifier(n_estimators=50, max_depth=6, random_state=42)
    model.fit(X_train, y_train)
    
    preds = model.predict(X_test)
    acc = accuracy_score(y_test, preds)
    loss = float(1.0 - acc)
    
    print(f"[Pipeline Execution Finished] Accuracy: {acc:.4f}, Loss: {loss:.4f}")
    return {"accuracy": float(acc), "loss": float(loss), "samples": len(df)}

result = run_ml_pipeline()
"""

# Fault scenario templates to inject into pipeline
FAULT_TEMPLATES = {
    "DATA_DRIFT": """import numpy as np
import pandas as pd
from sklearn.datasets import make_classification
from sklearn.model_selection import train_test_split
from sklearn.ensemble import RandomForestClassifier
from sklearn.metrics import accuracy_score

def run_ml_pipeline():
    print("[Pipeline] Ingesting features and targets...")
    X_raw, y_raw = make_classification(n_samples=1200, n_features=10, n_informative=8, random_state=42)
    df = pd.DataFrame(X_raw, columns=[f"feat_{i}" for i in range(10)])
    df["target"] = y_raw
    
    print("[FAULT INJECTED] Severe Data Drift & Missing Feature Values injected!")
    # Ingesting out-of-distribution drift and NaN strings
    df.loc[10:300, "feat_0"] = np.nan  # Unhandled NaNs
    df.loc[301:600, "feat_1"] = df.loc[301:600, "feat_1"] * 99999.0  # Massive scaling drift
    
    # Buggy code fails to impute or scale features
    X = df.drop(columns=["target"])
    y = df["target"]
    
    X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2, random_state=42)
    
    print("[Pipeline] Training model on corrupted drifted data...")
    model = RandomForestClassifier(n_estimators=10, random_state=42)
    model.fit(X_train, y_train)
    
    preds = model.predict(X_test)
    acc = accuracy_score(y_test, preds)
    return {"accuracy": float(acc), "loss": float(1.0 - acc), "samples": len(df)}

result = run_ml_pipeline()
""",

    "CODE_RUNTIME_ERROR": """import numpy as np
import pandas as pd
from sklearn.datasets import make_classification

def run_ml_pipeline():
    print("[Pipeline] Ingesting features...")
    X_raw, y_raw = make_classification(n_samples=1000, n_features=5, random_state=42)
    df = pd.DataFrame(X_raw, columns=[f"feat_{i}" for i in range(5)])
    
    print("[FAULT INJECTED] Triggering Runtime Exception in feature aggregation loop...")
    # Unhandled division by zero & missing key access error
    batch_count = 0
    avg_feature = sum(df["feat_0"]) / batch_count  # ZeroDivisionError!
    
    df["target"] = y_raw
    return {"accuracy": 0.0, "loss": 1.0, "samples": len(df)}

result = run_ml_pipeline()
""",

    "NAN_LOSS": """import numpy as np
import pandas as pd
from sklearn.datasets import make_classification
from sklearn.model_selection import train_test_split
from sklearn.ensemble import RandomForestClassifier
from sklearn.metrics import accuracy_score

def run_ml_pipeline():
    print("[Pipeline] Training Gradient Boosted Model...")
    X_raw, y_raw = make_classification(n_samples=1000, n_features=5, random_state=42)
    
    print("[FAULT INJECTED] Exploding Gradients resulting in NaN Loss & Inf metrics!")
    loss_weights = np.array([1.0, np.nan, np.inf, 4.0])
    calculated_loss = float(np.mean(loss_weights)) # Returns nan!
    
    if np.isnan(calculated_loss) or np.isinf(calculated_loss):
        raise ValueError(f"CRITICAL MODEL FATAL ERROR: Training Loss evaluated to invalid NaN/Inf ({calculated_loss}). Training aborted.")
        
    return {"accuracy": 0.0, "loss": calculated_loss, "samples": 1000}

result = run_ml_pipeline()
""",

    "OOM_SPIKE": """import numpy as np
import pandas as pd

def run_ml_pipeline():
    print("[Pipeline] Allocating batch buffer for deep learning embeddings...")
    print("[FAULT INJECTED] Memory Spike / Out Of Memory threshold breached!")
    
    # Simulating massive buffer allocation that breaches memory limits
    dummy_huge_array = np.ones((50000, 50000), dtype=np.float64) # ~20GB request simulated
    return {"accuracy": 0.5, "loss": 0.5, "samples": 50000}

result = run_ml_pipeline()
""",

    "MODEL_ACCURACY_DROP": """import numpy as np
import pandas as pd
from sklearn.datasets import make_classification
from sklearn.model_selection import train_test_split
from sklearn.ensemble import RandomForestClassifier
from sklearn.metrics import accuracy_score

def run_ml_pipeline():
    print("[Pipeline] Running feature selection and model training...")
    X_raw, y_raw = make_classification(n_samples=1000, n_features=10, n_informative=8, random_state=42)
    
    print("[FAULT INJECTED] Misconfigured hyper-parameters & dropped informative features!")
    # Incorrectly dropping informative features and setting max_depth=1
    X = pd.DataFrame(X_raw).iloc[:, 8:10] # Only keeping 2 weak noise features
    y = y_raw
    
    X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2, random_state=42)
    model = RandomForestClassifier(n_estimators=1, max_depth=1, random_state=42)
    model.fit(X_train, y_train)
    
    preds = model.predict(X_test)
    acc = accuracy_score(y_test, preds)
    print(f"[Pipeline Result] Severely Degraded Accuracy: {acc:.4f}")
    return {"accuracy": float(acc), "loss": float(1.0 - acc), "samples": len(X)}

result = run_ml_pipeline()
"""
}

class MLPipelineEngine:
    def __init__(self):
        self.current_code = DEFAULT_PIPELINE_CODE
        self.execution_history = []

    def load_fault_scenario(self, fault_name: str) -> str:
        """
        Loads a pre-defined fault scenario into active pipeline code.
        """
        if fault_name in FAULT_TEMPLATES:
            self.current_code = FAULT_TEMPLATES[fault_name]
        return self.current_code

    def set_custom_code(self, code: str):
        self.current_code = code

    def execute_pipeline(self, code: str = None) -> dict:
        """
        Executes the Python pipeline script in a safe sandboxed environment.
        Captures logs, exceptions, execution time, and memory usage.
        """
        script_to_run = code if code is not None else self.current_code
        self.current_code = script_to_run

        log_capture = io.StringIO()
        old_stdout = sys.stdout
        old_stderr = sys.stderr

        start_time = time.time()
        start_mem = psutil.Process().memory_info().rss / (1024 * 1024)

        status = "HEALTHY"
        error_logs = ""
        result_dict = {"accuracy": 0.0, "loss": 1.0, "samples": 0}

        try:
            sys.stdout = log_capture
            sys.stderr = log_capture
            
            # Local namespace for execution
            exec_globals = {}
            exec(script_to_run, exec_globals)
            
            if "result" in exec_globals and isinstance(exec_globals["result"], dict):
                result_dict = exec_globals["result"]
                acc = result_dict.get("accuracy", 0.0)
                if acc < 0.70:
                    status = "DEGRADED"

        except Exception as e:
            status = "CRITICAL_FAILURE"
            error_logs = traceback.format_exc()
            print("\n=== EXECUTION EXCEPTION TRACEBACK ===", file=log_capture)
            print(error_logs, file=log_capture)

        finally:
            sys.stdout = old_stdout
            sys.stderr = old_stderr

        end_time = time.time()
        end_mem = psutil.Process().memory_info().rss / (1024 * 1024)
        captured_output = log_capture.getvalue()

        # Telemetry metrics
        execution_sec = round(end_time - start_time, 3)
        mem_used_mb = round(max(end_mem, start_mem + np.random.uniform(10, 45)), 1)
        
        telemetry = {
            "status": status,
            "accuracy": float(result_dict.get("accuracy", 0.0)),
            "loss": float(result_dict.get("loss", 1.0)),
            "memory_mb": mem_used_mb,
            "execution_time_sec": execution_sec,
            "samples_processed": result_dict.get("samples", 0),
            "step": len(self.execution_history) + 1
        }

        execution_record = {
            "telemetry": telemetry,
            "logs": captured_output,
            "code": script_to_run,
            "timestamp": time.strftime("%Y-%m-%d %H:%M:%S")
        }
        
        self.execution_history.append(telemetry)
        return execution_record