from datetime import datetime import json from pathlib import Path import dagshub from loguru import logger from mlflow.tracking import MlflowClient import numpy as np import pandas as pd import typer from turing import config from turing.modeling.model_selector import get_best_model_by_tag from turing.monitoring.baseline_manager import extract_baseline_statistics from turing.monitoring.drift_detector import DriftDetector from turing.monitoring.feedback_manager import load_feedback_for_language from turing.monitoring.synthetic_data_generator import SyntheticDataGenerator app = typer.Typer() def load_training_data(dataset_name: str, language: str): """ Load training data for a specific programming language. Args: dataset_name: Dataset name (e.g., 'clean-k5000') language: Programming language (java, python, pharo) Returns: Tuple of (texts, labels) as lists """ dataset_path = config.INTERIM_DATA_DIR / "features" / dataset_name if not dataset_path.exists(): raise FileNotFoundError(f"Dataset path not found: {dataset_path}") train_file = None for file in dataset_path.rglob(f"{language}_train*.csv"): train_file = file break if not train_file: raise FileNotFoundError(f"Training file not found for {language} in {dataset_path}") logger.info(f"Loading training data from: {train_file}") df = pd.read_csv(train_file) X_train = df[config.INPUT_COLUMN].tolist() if isinstance(df[config.LABEL_COLUMN].iloc[0], str): y_train = np.array([eval(label) for label in df[config.LABEL_COLUMN]]) else: y_train = df[config.LABEL_COLUMN].values logger.success(f"Loaded {len(X_train)} training samples for {language}") return X_train, y_train def print_drift_report(drift_results: dict, drift_type: str, report_lines: list | None = None): """ Format and display drift detection results for a specific drift type. Args: drift_results: Dictionary with drift detection metrics and alerts drift_type: Name of drift type tested (e.g., 'none', 'text_length_short') report_lines: Optional list to collect formatted report lines """ def log_and_collect(msg: str): logger.info(msg) if report_lines is not None: report_lines.append(msg) log_and_collect(f"\n{'=' * 60}") log_and_collect(f"DRIFT DETECTION REPORT - {drift_type.upper()}") log_and_collect(f"{'=' * 60}") for metric_name, result in drift_results.items(): if metric_name == "overall": continue p_value = result.get("p_value", result.get("check_result", {}).get("passed", None)) statistic = result.get("statistic", None) drifted = result.get("drifted", False) alert = result.get("alert", False) if alert: status = "ALERT" elif drifted: status = "DRIFT" else: status = "OK" log_and_collect(f"\n{metric_name.upper()}") log_and_collect(f" Status: {status}") if p_value is not None: log_and_collect(f" P-value: {p_value:.6f}") if statistic is not None: log_and_collect(f" Statistic: {statistic:.6f}") log_and_collect(f" Drift detected: {drifted}") log_and_collect(f" Critical alert: {alert}") log_and_collect(f" Method: {result.get('method', 'unknown')}") overall = drift_results.get("overall", {}) overall_drifted = overall.get("drifted", False) overall_alert = overall.get("alert", False) drift_count = overall.get("num_drifts", 0) log_and_collect(f"\n{'=' * 60}") log_and_collect("OVERALL SUMMARY") log_and_collect(f" Drift detected: {overall_drifted}") log_and_collect(f" Critical alert: {overall_alert}") log_and_collect(f" Number of drifted metrics: {drift_count}") log_and_collect(f" Methods used: {overall.get('methods', [])}") log_and_collect(f"{'=' * 60}\n") def save_drift_report( language: str, dataset_name: str, baseline_stats: dict, test_results: dict, report_text: str, ): """ Save drift detection report to TXT and JSON files. Args: language: Programming language tested dataset_name: Name of dataset used baseline_stats: Baseline statistics dictionary test_results: Dictionary with test results for each drift type report_text: Formatted report text """ def convert_numpy_types(obj): if isinstance(obj, dict): return {k: convert_numpy_types(v) for k, v in obj.items()} elif isinstance(obj, list): return [convert_numpy_types(item) for item in obj] elif isinstance(obj, np.bool_): return bool(obj) elif isinstance(obj, np.integer): return int(obj) elif isinstance(obj, np.floating): return float(obj) elif isinstance(obj, np.ndarray): return obj.tolist() else: return obj monitoring_dir = config.REPORTS_DIR / "monitoring" monitoring_dir.mkdir(parents=True, exist_ok=True) report_file = monitoring_dir / f"drift_report_{language}.txt" with open(report_file, "w") as f: f.write("DRIFT DETECTION REPORT\n") f.write(f"Language: {language}\n") f.write(f"Dataset: {dataset_name}\n") f.write(f"Timestamp: {datetime.now().isoformat()}\n") f.write(f"P-value threshold: {config.DRIFT_P_VALUE_THRESHOLD}\n") f.write(f"Alert threshold: {config.DRIFT_ALERT_THRESHOLD}\n") f.write("\n" + "=" * 80 + "\n\n") f.write("BASELINE STATISTICS\n") f.write( f" Text length: mean={baseline_stats['text_length_mean']:.2f}, std={baseline_stats['text_length_std']:.2f}\n" ) f.write( f" Word count: mean={baseline_stats['word_count_mean']:.2f}, std={baseline_stats['word_count_std']:.2f}\n" ) f.write(f" Label counts: {baseline_stats['label_counts']}\n") f.write(f" Number of samples: {baseline_stats['num_samples']}\n") f.write("\n" + "=" * 80 + "\n\n") f.write(report_text) json_file = monitoring_dir / f"drift_report_{language}.json" report_data = { "language": language, "dataset": dataset_name, "timestamp": datetime.now().isoformat(), "config": { "p_value_threshold": config.DRIFT_P_VALUE_THRESHOLD, "alert_threshold": config.DRIFT_ALERT_THRESHOLD, }, "baseline": { "text_length_mean": baseline_stats["text_length_mean"], "text_length_std": baseline_stats["text_length_std"], "word_count_mean": baseline_stats["word_count_mean"], "word_count_std": baseline_stats["word_count_std"], "label_counts": baseline_stats["label_counts"], "num_samples": baseline_stats["num_samples"], "n_labels": baseline_stats["n_labels"], }, "test_results": convert_numpy_types(test_results), } with open(json_file, "w") as f: json.dump(report_data, f, indent=2) logger.success("Report saved to:") logger.info(f" Text: {report_file}") logger.info(f" JSON: {json_file}") @app.command() def verify( language: str = typer.Option("java", help="Language to test (java, python, pharo)"), repo_owner: str = typer.Option("se4ai2526-uniba", help="DagsHub repository owner"), repo_name: str = typer.Option("Turing", help="DagsHub repository name"), n_samples: int = typer.Option(100, help="Number of samples for synthetic data generation"), use_feedback: bool = typer.Option(False, help="Include user feedback rows in drift analysis"), feedback_path: Path = typer.Option( config.PROJ_ROOT / "turing" / "monitoring" / "feedback" / "feedback_data.csv", help="Path to user feedback CSV", ), ): """ Verify drift detection on best model's training dataset. """ import os logger.info("Starting drift detection verification...") logger.info("Configuration:") logger.info(f" Language: {language}") logger.info(f" P-value threshold: {config.DRIFT_P_VALUE_THRESHOLD}") logger.info(f" Alert threshold: {config.DRIFT_ALERT_THRESHOLD}") logger.info(f" Baseline cache: {config.BASELINE_CACHE_DIR}") # Setup DagsHub credentials from environment variables dagshub_username = os.getenv("DAGSHUB_USERNAME") dagshub_token = os.getenv("DAGSHUB_TOKEN") mlflow_uri = os.getenv("MLFLOW_TRACKING_URI") is_ci_environment = os.getenv("CI") or os.getenv("GITHUB_ACTIONS") if dagshub_username and dagshub_token and mlflow_uri: # Use environment credentials for non-interactive mode (GitHub Actions) logger.info("Using DagsHub credentials from environment variables") os.environ["MLFLOW_TRACKING_USERNAME"] = dagshub_username os.environ["MLFLOW_TRACKING_PASSWORD"] = dagshub_token # Don't call dagshub.init() - credentials are already set via environment logger.info("Skipping dagshub.init() - using environment credentials directly") elif is_ci_environment: # In CI without credentials, skip OAuth and log warning logger.warning("CI environment detected but credentials not found. Proceeding without dagshub.init()") else: # Interactive mode - try to initialize with OAuth logger.info("Initializing DagsHub interactively") try: dagshub.init(repo_owner=repo_owner, repo_name=repo_name, mlflow=True) except Exception as e: logger.warning(f"DagsHub initialization failed: {e}") logger.info(f"\n[1/6] Searching for best model for {language}...") best_model_info = get_best_model_by_tag(language=language) if not best_model_info: logger.error(f"No best model found for {language}") return run_id = best_model_info["run_id"] logger.info(f"\n[2/6] Retrieving dataset information from MLflow run {run_id}...") client = MlflowClient() run = client.get_run(run_id) dataset_name = run.data.tags.get("dataset_name", None) if not dataset_name: logger.error("Dataset name not found in run tags") return logger.success(f"Found dataset: {dataset_name}") logger.info("\n[3/6] Loading training data...") try: X_train, y_train = load_training_data(dataset_name, language) y_train = np.asarray(y_train) # Ensure y_train is np.ndarray except Exception as e: logger.error(f"Failed to load training data: {e}") return logger.info("\n[4/6] Extracting baseline statistics...") baseline_stats = extract_baseline_statistics(X_train, y_train, language) logger.success("Baseline extracted:") logger.info( f" Text length: mean={baseline_stats['text_length_mean']:.2f}, std={baseline_stats['text_length_std']:.2f}" ) logger.info( f" Word count: mean={baseline_stats['word_count_mean']:.2f}, std={baseline_stats['word_count_std']:.2f}" ) logger.info(f" Label counts: {baseline_stats['label_counts']}") logger.info("\n[5/6] Initializing drift detection components...") drift_detector = DriftDetector() synthetic_generator = SyntheticDataGenerator(seed=42) feedback_texts, feedback_labels = [], np.array([]) if use_feedback: try: feedback_texts, feedback_labels = load_feedback_for_language(feedback_path, language) except Exception as e: logger.warning(f"Feedback load skipped: {e}") logger.info("\n[6/6] Testing drift detection on different data types...\n") test_cases = [ ("NORMAL DATA (no drift expected)", "none"), ("SHORT TEXT DRIFT", "text_length_short"), ("LONG TEXT DRIFT", "text_length_long"), ("CORRUPTED VOCABULARY DRIFT", "corrupted_vocab"), ("CLASS IMBALANCE DRIFT", "class_imbalance"), ] if use_feedback and len(feedback_texts) > 0: test_cases.append(("USER FEEDBACK", "feedback")) all_test_results = {} all_report_lines = [] for test_name, drift_type in test_cases: logger.info(f"\n{'#' * 60}") logger.info(f"Test: {test_name}") logger.info(f"{'#' * 60}") if drift_type == "feedback": production_texts = feedback_texts production_labels = feedback_labels else: production_texts, production_labels = synthetic_generator.generate_synthetic_batch( reference_texts=X_train, reference_labels=y_train, drift_type=drift_type, batch_size=n_samples, ) drift_results = drift_detector.detect_all_drifts( production_texts=production_texts, production_labels=production_labels, reference_texts=X_train, reference_labels=y_train, ) all_test_results[drift_type] = drift_results print_drift_report(drift_results, drift_type, report_lines=all_report_lines) logger.info("\nSaving drift detection report...") report_text = "\n".join(all_report_lines) save_drift_report( language=language, dataset_name=dataset_name, baseline_stats=baseline_stats, test_results=all_test_results, report_text=report_text, ) logger.success("\nDrift detection verification completed!") if __name__ == "__main__": app()