"""End-to-end model evaluation pipeline.""" import argparse import json import logging import sys from datetime import datetime from pathlib import Path sys.path.insert(0, str(Path(__file__).parent.parent.parent)) from src.utils.logger import setup_logger from src.utils.visualization import ( plot_confusion_matrix, plot_label_distribution, ) logger = setup_logger("evaluation_pipeline") def evaluate_model(model_path: str, test_data: str, output_dir: str) -> dict: """Evaluate a trained model.""" logger.info(f"Evaluating model: {model_path}") # Run evaluation import subprocess result = subprocess.run([ "python", "src/models/evaluate.py", "--model_path", model_path, "--data_path", test_data, "--output_path", f"{output_dir}/results.json", ], capture_output=True, text=True) logger.info(result.stdout) if result.returncode != 0: logger.error(result.stderr) raise RuntimeError("Evaluation failed") # Load results with open(f"{output_dir}/results.json", "r") as f: results = json.load(f) return results def generate_report(results: dict, output_dir: str) -> str: """Generate evaluation report with visualizations.""" logger.info("Generating evaluation report...") report_path = f"{output_dir}/report.html" # Create visualizations cm = results["confusion_matrix"] class_names = results["class_names"] plot_confusion_matrix( cm, class_names, title="Sentiment Classification Confusion Matrix", output_path=f"{output_dir}/confusion_matrix.png", ) # Create HTML report html = f""" Myanmar Ghost Evaluation Report

🇲🇲 Myanmar Ghost Evaluation Report

Generated: {datetime.now().strftime("%Y-%m-%d %H:%M:%S")}

📊 Metrics

{results['metrics']['accuracy']:.2%}

Accuracy

{results['metrics']['f1_weighted']:.2%}

F1 (weighted)

{results['metrics']['precision_weighted']:.2%}

Precision

{results['metrics']['recall_weighted']:.2%}

Recall

📈 Confusion Matrix

📋 Per-Class Performance

""" for name in class_names: f1_key = f"f1_{name}" if f1_key in results["metrics"]: html += f""" """ html += """

Class	F1 Score
{name}	{results['metrics'][f1_key]:.4f}

""" with open(report_path, "w", encoding="utf-8") as f: f.write(html) logger.info(f"Report saved to {report_path}") return report_path def compare_models(model_results: list, output_dir: str) -> dict: """Compare multiple model evaluations.""" logger.info(f"Comparing {len(model_results)} models...") comparison = { "models": [], "best_model": None, "best_f1": 0, } for result in model_results: model_name = result.get("model_name", "unknown") metrics = result.get("metrics", {}) comparison["models"].append({ "name": model_name, "accuracy": metrics.get("accuracy", 0), "f1_weighted": metrics.get("f1_weighted", 0), "f1_macro": metrics.get("f1_macro", 0), }) if metrics.get("f1_weighted", 0) > comparison["best_f1"]: comparison["best_f1"] = metrics.get("f1_weighted", 0) comparison["best_model"] = model_name # Save comparison with open(f"{output_dir}/model_comparison.json", "w") as f: json.dump(comparison, f, indent=2) logger.info(f"Best model: {comparison['best_model']} (F1: {comparison['best_f1']:.4f})") return comparison def run_evaluation_pipeline( model_path: str, test_data: str, output_dir: str = "outputs/evaluation", ) -> dict: """Run full evaluation pipeline.""" Path(output_dir).mkdir(parents=True, exist_ok=True) # Evaluate model results = evaluate_model(model_path, test_data, output_dir) # Generate report report_path = generate_report(results, output_dir) return { "results": results, "report": report_path, } if __name__ == "__main__": parser = argparse.ArgumentParser() parser.add_argument("--model_path", type=str, required=True) parser.add_argument("--test_data", type=str, required=True) parser.add_argument("--output_dir", type=str, default="outputs/evaluation") args = parser.parse_args() run_evaluation_pipeline(args.model_path, args.test_data, args.output_dir)