| """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}") |
| |
| |
| 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") |
| |
| |
| 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" |
| |
| |
| 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", |
| ) |
| |
| |
| html = f""" |
| <!DOCTYPE html> |
| <html> |
| <head> |
| <title>Myanmar Ghost Evaluation Report</title> |
| <style> |
| body {{ font-family: Arial, sans-serif; margin: 40px; }} |
| .metric {{ display: inline-block; margin: 10px; padding: 20px; |
| background: #f0f0f0; border-radius: 8px; }} |
| .metric-value {{ font-size: 32px; font-weight: bold; color: #2196F3; }} |
| .metric-label {{ font-size: 14px; color: #666; }} |
| img {{ max-width: 600px; margin: 20px 0; }} |
| </style> |
| </head> |
| <body> |
| <h1>🇲🇲 Myanmar Ghost Evaluation Report</h1> |
| <p>Generated: {datetime.now().strftime("%Y-%m-%d %H:%M:%S")}</p> |
| |
| <h2>📊 Metrics</h2> |
| <div class="metrics"> |
| <div class="metric"> |
| <div class="metric-value">{results['metrics']['accuracy']:.2%}</div> |
| <div class="metric-label">Accuracy</div> |
| </div> |
| <div class="metric"> |
| <div class="metric-value">{results['metrics']['f1_weighted']:.2%}</div> |
| <div class="metric-label">F1 (weighted)</div> |
| </div> |
| <div class="metric"> |
| <div class="metric-value">{results['metrics']['precision_weighted']:.2%}</div> |
| <div class="metric-label">Precision</div> |
| </div> |
| <div class="metric"> |
| <div class="metric-value">{results['metrics']['recall_weighted']:.2%}</div> |
| <div class="metric-label">Recall</div> |
| </div> |
| </div> |
| |
| <h2>📈 Confusion Matrix</h2> |
| <img src="confusion_matrix.png" alt="Confusion Matrix"> |
| |
| <h2>📋 Per-Class Performance</h2> |
| <table border="1" cellpadding="10"> |
| <tr> |
| <th>Class</th> |
| <th>F1 Score</th> |
| </tr> |
| """ |
| |
| for name in class_names: |
| f1_key = f"f1_{name}" |
| if f1_key in results["metrics"]: |
| html += f""" |
| <tr> |
| <td>{name}</td> |
| <td>{results['metrics'][f1_key]:.4f}</td> |
| </tr> |
| """ |
| |
| html += """ |
| </table> |
| </body> |
| </html> |
| """ |
| |
| 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 |
| |
| |
| 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) |
| |
| |
| results = evaluate_model(model_path, test_data, output_dir) |
| |
| |
| 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) |
|
|
