refactor: reorganize evaluation scripts structure

dataset/preparation/decompress_dataset.py

@@ -2,7 +2,7 @@
 from pathlib import Path
 
 
-def decompress_dataset(archive_path="dataset/archives/classification_dataset.zip", output_dir="data"):
+def decompress_dataset(archive_path="dataset/archives/classification_dataset.zip", output_dir="assets"):
     """
     Decompress the classification dataset from a ZIP archive.
 

run_dev.sh

@@ -28,4 +28,4 @@ docker run \
     -it \
     -td \
     --rm \
-    ${IMAGE_NAME} \
+    ${IMAGE_NAME} 

evaluation_results.json → src/evaluation/results/evaluation_results.json

@@ -1,18 +1,18 @@
 {
-  "resnet": {
+  "resnet18_pytorch": {
     "accuracy": 0.26206140350877194,
     "precision_micro": 0.26206140350877194,
     "recall_micro": 0.26206140350877194,
     "f1_micro": 0.26206140350877194,
-    "latency_mean": 0.07633136959096096,
-    "latency_std": 0.07063937546321011
+    "latency_mean": 0.08492945763376918,
+    "latency_std": 0.0513541151932428
   },
-  "mobilenet": {
+  "mobilenet_pytorch": {
     "accuracy": 0.2719298245614035,
     "precision_micro": 0.2719298245614035,
     "recall_micro": 0.2719298245614035,
     "f1_micro": 0.2719298245614035,
-    "latency_mean": 0.060137055586657176,
-    "latency_std": 0.09272526518546054
+    "latency_mean": 0.05509323848684594,
+    "latency_std": 0.026680795139590825
   }
 }

src/evaluation/run_evaluation.py

@@ -1,119 +1,62 @@
 import os
 import sys
-import argparse
-import json
+import time
+from dataclasses import dataclass
 from pathlib import Path
-
-import pandas as pd
+from typing import List, Optional, Tuple
 
 # Add workspace root to sys.path
 sys.path.append("/workspace")
 
-from src.optimal_class_mapping import MODEL_NAMES as class_mapping, map_prediction
-from src.inference.utils.inference_factory import InferenceFactory
-from src.evaluation.evaluator import EvaluationConfig, ModelEvaluator
-from src.onnx_model import OnnxClassifierInferenceBase as OnnxModel
-from dataset.optimal_class_mapping import map_prediction
-from src.path_utils import ensure_clean_directory
-
-
-MODELS_DIR_PATH = Path("models")
-
-class MappedModelWrapper:
-    """Wrapper that adds mapping between 83 model classes to 76 dataset classes"""
-
-    def __init__(self, model):
-        self.model = model
-
-    def infer(self, image):
-        predictions = self.model.infer(image)
-        # Map the first prediction
-        mapped_class_id = map_prediction(predictions[0]["class_id"])
-        return [
-            {
-                "class_id": mapped_class_id,
-                "class_name": f"class_{mapped_class_id}",
-                "probability": predictions[0]["probability"],
-            }
-        ]
-
-
-def create_model(model_name: str, model_type: str, device: str):
-    weights_path = MODELS_DIR_PATH / model_type / f"{model_name}"
-
-    if model_type == "pytorch":
-        base_model = InferenceFactory.create(model_name, weights_path.with_suffix(".pt"), device, class_mapping)
-    elif model_type == "onnx":
-        base_model = OnnxModel(device=device, weights_path=weights_path.with_suffix(".onnx"), topk=1)
-    else:
-        raise ValueError(f"Unknown model_type: {model_type}")
-
-    return MappedModelWrapper(base_model)
-
-
-def parse_arguments():
-    parser = argparse.ArgumentParser(
-        description="Evaluate models on classification dataset"
-    )
-    parser.add_argument("--dataset", type=str, required=True)
-    parser.add_argument("--models", type=str, default="resnet18,mobilenet")
-    parser.add_argument(
-        "--model_type", type=str, choices=["pytorch", "onnx"], required=True
-    )
-    parser.add_argument("--device", type=str, default="cpu")
-    parser.add_argument("--output", type=str, default="outputs/evaluation_results.json")
-    return parser.parse_args()
-
-
-def main():
-    args = parse_arguments()
-    models_to_eval = [m.strip() for m in args.models.split(",")]
-    config = EvaluationConfig(dataset_path=Path(args.dataset), device=args.device)
-    evaluator = ModelEvaluator(config)
-    results = {}
-    print("📊 EVALUATION STARTING")
-    print(f"Dataset: {args.dataset}, Models: {models_to_eval}, Device: {args.device}")
-    print("=" * 50)
-
-    for model_name in models_to_eval:
-        try:
-            model = create_model(model_name, args.model_type, args.device)
-            report = evaluator.evaluate_model(model)
-            result_key = f"{model_name}_{args.model_type}"
-            results[result_key] = report.dict()
-            print(f"\n✅ {model_name.upper()} Results:")
-            print(f"   Accuracy: {report.accuracy:.4f}")
-            print(f"   Precision (Micro): {report.precision_micro:.4f}")
-            print(f"   Recall (Micro): {report.recall_micro:.4f}")
-            print(f"   F1-Score (Micro): {report.f1_micro:.4f}")
-            print(f"   Latency: {report.latency_mean:.4f}s ± {report.latency_std:.4f}s")
-        except Exception as e:
-            print(f"❌ Error evaluating {model_name}: {e}")
-            continue
-
-    ensure_clean_directory(Path(args.output).parent)
-    with open(args.output, "a") as f:
-        json.dump(results, f, indent=2)
-    print(f"\n💾 Results saved to: {args.output}")
-
-    if results:
-        df_data = []
-        for model_name, report in results.items():
-            df_data.append(
-                {
-                    "Model": model_name.upper(),
-                    "Accuracy": f"{report['accuracy']:.4f}",
-                    "Precision": f"{report['precision_micro']:.4f}",
-                    "Recall": f"{report['recall_micro']:.4f}",
-                    "F1-Score": f"{report['f1_micro']:.4f}",
-                    "Latency (s)": f"{report['latency_mean']:.4f} ± {report['latency_std']:.4f}",
-                }
-            )
-        df = pd.DataFrame(df_data)
-        print("\n📊 SUMMARY TABLE:")
-        print("=" * 80)
-        print(df.to_string(index=False))
-
-
-if __name__ == "__main__":
-    main()
+import torch
+from PIL import Image
+
+from src.evaluation.metrics.metrics_api import ClassificationReport, compute_metrics
+from src.inference import ClassifierInferenceBase as InferenceModel
+
+@dataclass
+class EvaluationConfig:
+    dataset_path: Path
+    model_weights_path: Optional[Path] = None
+    device: str = "cpu"
+    batch_size: int = 32
+    num_workers: int = 4
+
+
+class ModelEvaluator:
+    def __init__(self, config: EvaluationConfig):
+        self.config = config
+        self.device = torch.device(config.device)
+
+    def load_dataset(self) -> Tuple[List[Path], List[int]]:
+        image_paths = []
+        labels = []
+        dataset_path = self.config.dataset_path / "images"
+        for class_dir in sorted(dataset_path.iterdir()):
+            if class_dir.is_dir():
+                class_id = int(class_dir.name)
+                for image_path in class_dir.glob("*.png"):
+                    image_paths.append(image_path)
+                    labels.append(class_id)
+        print(f"✅ Loaded {len(image_paths)} images from {len(set(labels))} classes")
+        return image_paths, labels
+
+    def evaluate_model(self, model: InferenceModel) -> ClassificationReport:
+        print("🔄 Starting evaluation...")
+        image_paths, true_labels = self.load_dataset()
+        predictions = []
+        latencies = []
+        for i, image_path in enumerate(image_paths):
+            if i % 100 == 0:
+                print(f"Progress: {i}/{len(image_paths)}")
+            image = Image.open(image_path).convert("RGB")
+            start_time = time.perf_counter()
+            result = model.infer(image)
+            end_time = time.perf_counter()
+            latencies.append(end_time - start_time)
+            predictions.append(result[0]["class_id"])
+        report = compute_metrics(
+            y_true=true_labels, y_pred=predictions, latencies=latencies
+        )
+        print("✅ Evaluation completed!")
+        return report

src/evaluation/run_hierarchical_evaluation.py

@@ -9,11 +9,10 @@
 
 from PIL import Image
 from dataset.utilities.datasets import DATASETS
-from metrics.metrics_api import compute_metrics
-from dataset_preparation.utilities.datasets import DATASETS
-from src.evaluation.evaluator import EvaluationConfig, ModelEvaluator
-from src.inference.utils.inference_factory import InferenceFactory
-from src.optimal_class_mapping import MODEL_NAMES as class_mapping, map_prediction
+from src.evaluation.metrics.metrics_api import compute_metrics
+from src.evaluation.base.evaluator import EvaluationConfig, ModelEvaluator
+from src.inference import MobileNetInference, ResNetInference
+from dataset.optimal_class_mapping import map_prediction
 
 
 class MappedModelWrapper:
@@ -71,7 +70,7 @@ def main():
 
     # ResNet
     print("🔄 ResNet...")
-    resnet = InferenceFactory.create("resnet", "models/pytorch/resnet18.pt", "cpu")
+    resnet = ResNetInference(weights_path="models/pytorch/resnet18.pt", num_classes=83)
     wrapped_resnet = MappedModelWrapper(resnet)
 
     predictions, latencies = [], []
@@ -88,7 +87,7 @@ def main():
 
     # MobileNet
     print("🔄 MobileNet...")
-    mobilenet = InferenceFactory.create("mobilenet", "models/pytorch/mobilenet.pt", "cpu")
+    mobilenet = MobileNetInference(weights_path="models/pytorch/mobilenet.pt", num_classes=83)
     wrapped_mobilenet = MappedModelWrapper(mobilenet)
 
     predictions, latencies = [], []
@@ -103,7 +102,7 @@ def main():
 
     results["mobilenet"] = evaluate_all_hierarchies(true_labels, predictions, latencies)
 
-    with open("complete_hierarchy_results.json", "w") as f:
+    with open("src/evaluation/results/hierarchical_evaluation_results.json", "w") as f:
         json.dump(results, f, indent=2)
 
     print("\n📊 תוצאות:")