Code cleanup, README updates, interfaces for predictions and metric results

luxonis_eval/__init__.py

@@ -6,7 +6,14 @@
 __version__: Final[str] = "0.0.1"
 __semver__: Final[SemanticVersion] = SemanticVersion.parse(__version__)
 
-from .core import LuxonisEval  # noqa: F401
+from .core import (
+    EvaluationResult,
+    LuxonisEval,
+    MetricResult,
+    MetricsResult,
+    MetricValues,
+    ThroughputResult,
+)  # noqa: F401
 from .engines import *
 from .loaders import *
 from .metrics import *

luxonis_eval/__main__.py

@@ -6,6 +6,7 @@
 
 from luxonis_eval.config import EvalConfig
 from luxonis_eval.core import LuxonisEval
+from luxonis_eval.core.results import EvaluationResult
 from luxonis_eval.utils.json_utils import write_output_json
 
 app = App(
@@ -21,7 +22,7 @@ def eval_run(
     cfg: PathType | Params | EvalConfig,
     opts: Params | list[str] | tuple[str, ...] | None = None,
     output_json: str | None = None,
-) -> dict[str, Any]:
+) -> EvaluationResult:
     """Run evaluation with the given configuration."""
     # Temporary: until benchmark execution is implemented, `eval` delegates
     # to the quality-only path.
@@ -32,7 +33,7 @@ def quality_run(
     cfg: PathType | Params | EvalConfig,
     opts: Params | list[str] | tuple[str, ...] | None = None,
     output_json: str | None = None,
-) -> dict[str, Any]:
+) -> EvaluationResult:
     """Run the configured quality evaluators."""
     evaluator = LuxonisEval(cfg, opts)
     evaluator.setup()

luxonis_eval/core/__init__.py

@@ -1,3 +1,17 @@
 from .core import LuxonisEval
+from .results import (
+    EvaluationResult,
+    MetricResult,
+    MetricsResult,
+    MetricValues,
+    ThroughputResult,
+)
 
-__all__ = ["LuxonisEval"]
+__all__ = [
+    "EvaluationResult",
+    "LuxonisEval",
+    "MetricResult",
+    "MetricsResult",
+    "MetricValues",
+    "ThroughputResult",
+]

luxonis_eval/core/core.py

@@ -18,8 +18,8 @@
 from luxonis_eval.core.reporting import (
     RichProgressAdapter,
     TQDMProgressAdapter,
+    format_evaluation_result,
     get_model_name,
-    make_report_table,
 )
 from luxonis_eval.core.runtime import (
     build_metric_contexts,
@@ -36,7 +36,7 @@
 from luxonis_eval.metrics import ThroughputMetric
 from luxonis_eval.metrics.base_metric import BaseMetric
 from luxonis_eval.parsers.base_parser import BaseParser
-from luxonis_eval.parsers.yolo import clear_prediction_metadata
+from luxonis_eval.core.results import EvaluationResult
 from luxonis_eval.visualizers.base_visualizer import BaseVisualizer
 
 
@@ -123,20 +123,20 @@ def setup(self) -> None:
         self._is_setup = True
         self._is_closed = False
 
-    def evaluate(self) -> dict[str, Any]:
+    def evaluate(self) -> EvaluationResult:
         """Run the evaluation loop and return structured results."""
         self._require_setup()
         self._reset_runtime_metrics()
         return self._run_pipeline()
 
-    def _run_pipeline(self) -> dict[str, Any]:
+    def _run_pipeline(self) -> EvaluationResult:
         if self.cfg.pipeline.benchmark is not None:
             logger.warning(
                 "pipeline.benchmark is configured, but benchmark execution is not implemented yet. Running quality evaluators only."
             )
         return self._run_evaluators()
 
-    def _run_evaluators(self) -> dict[str, Any]:
+    def _run_evaluators(self) -> EvaluationResult:
         """Run the configured quality evaluator."""
         self._require_setup()
         engine_name = self.cfg.pipeline.engine.name
@@ -177,43 +177,40 @@ def _run_evaluators(self) -> dict[str, Any]:
                 )
                 parsing_elapsed = time.perf_counter() - parsing_t0
 
-                try:
-                    metric_update_t0 = time.perf_counter()
-                    for metric, metric_ctx in zip(
-                        self.metrics, self.metric_contexts, strict=True
-                    ):
-                        metric.update(
-                            predictions=predictions,
-                            target=target,
-                            **metric_ctx,
-                        )
-                    metric_update_elapsed = (
-                        time.perf_counter() - metric_update_t0
+                metric_update_t0 = time.perf_counter()
+                for metric, metric_ctx in zip(
+                    self.metrics, self.metric_contexts, strict=True
+                ):
+                    metric.update(
+                        predictions=predictions,
+                        target=target,
+                        **metric_ctx,
                     )
+                metric_update_elapsed = (
+                    time.perf_counter() - metric_update_t0
+                )
 
-                    self.throughput_metric.update(
-                        inference=inference_elapsed,
-                        parsing=parsing_elapsed,
-                        metric_update=metric_update_elapsed,
-                    )
+                self.throughput_metric.update(
+                    inference=inference_elapsed,
+                    parsing=parsing_elapsed,
+                    metric_update=metric_update_elapsed,
+                )
 
-                    active_visualizer_cfgs = [
-                        visualizer_cfg
-                        for visualizer_cfg in self.evaluator_cfg.visualizers
-                        if visualizer_cfg.active
-                    ]
-                    for visualizer, visualizer_cfg in zip(
-                        self.visualizers,
-                        active_visualizer_cfgs,
-                        strict=True,
-                    ):
-                        visualizer.visualize(
-                            predictions,
-                            self.engine.vis_frame(),
-                            **visualizer_cfg.params,
-                        )
-                finally:
-                    clear_prediction_metadata(predictions)
+                active_visualizer_cfgs = [
+                    visualizer_cfg
+                    for visualizer_cfg in self.evaluator_cfg.visualizers
+                    if visualizer_cfg.active
+                ]
+                for visualizer, visualizer_cfg in zip(
+                    self.visualizers,
+                    active_visualizer_cfgs,
+                    strict=True,
+                ):
+                    visualizer.visualize(
+                        predictions,
+                        self.engine.vis_frame(),
+                        **visualizer_cfg.params,
+                    )
                 progress.update(advance=1)
 
         metric_compute_t0 = time.perf_counter()
@@ -226,26 +223,19 @@ def _run_evaluators(self) -> dict[str, Any]:
             metric_compute=metric_compute_elapsed
         )
 
-        report = make_report_table(
-            engine_name=engine_name,
+        result = EvaluationResult(
+            evaluator_name=self.evaluator_cfg.name,
+            engine=engine_name,
             model_name=model_name,
-            tp=throughput,
-            results=results,
+            metrics=results,
+            throughput=throughput,
         )
-
         logger.warning(
             "Throughput values are end-to-end pipeline measurements and not isolated model-only benchmarks. Lower numbers than modelconverter benchmark results are expected."
         )
-        logger.info(f"\n{report}")
+        logger.info(f"\n{format_evaluation_result(result)}")
 
-        return {
-            "evaluator_name": self.evaluator_cfg.name,
-            "engine": engine_name,
-            "model_name": model_name,
-            "metrics": results,
-            "throughput": throughput,
-            "report": report,
-        }
+        return result
 
     def close(self) -> None:
         """Release owned runtime resources."""
@@ -292,27 +282,24 @@ def _sanity_check_pipeline(self) -> None:
             **self.evaluator_cfg.parser.params,
         )
 
-        try:
-            for metric, metric_ctx in zip(
-                self.metrics, self.metric_contexts, strict=True
-            ):
-                missing = set(metric.required_target_keys()) - set(target)
-                if missing:
-                    raise ValueError(
-                        "Target is missing required keys for "
-                        f"{metric.__class__.__name__}: {sorted(missing)}. "
-                        f"Got keys: {sorted(target.keys())}."
-                    )
-
-                metric.update(
-                    predictions=predictions,
-                    target=target,
-                    **metric_ctx,
+        for metric, metric_ctx in zip(
+            self.metrics, self.metric_contexts, strict=True
+        ):
+            missing = set(metric.required_target_keys()) - set(target)
+            if missing:
+                raise ValueError(
+                    "Target is missing required keys for "
+                    f"{metric.__class__.__name__}: {sorted(missing)}. "
+                    f"Got keys: {sorted(target.keys())}."
                 )
-                metric.compute()
-                metric.reset()
-        finally:
-            clear_prediction_metadata(predictions)
+
+            metric.update(
+                predictions=predictions,
+                target=target,
+                **metric_ctx,
+            )
+            metric.compute()
+            metric.reset()
 
     def _clear_runtime_fields(self) -> None:
         self.engine: BaseEngine | None = None

luxonis_eval/core/reporting.py

@@ -14,6 +14,11 @@
 from tabulate import tabulate
 from tqdm.auto import tqdm
 
+from luxonis_eval.core.results import (
+    EvaluationResult,
+    ThroughputResult,
+)
+
 
 class TQDMProgressAdapter(AbstractContextManager["TQDMProgressAdapter"]):
     def __init__(self, description: str, total: int) -> None:
@@ -79,45 +84,56 @@ def section(
     return [[centered, ""]]
 
 
-def make_report_table(
-    engine_name: str,
-    model_name: str,
-    tp: dict[str, float | int],
-    results: list[tuple[str, dict[str, Any]]],
-) -> str:
-    def format_stage(name: str) -> str:
-        ms = float(tp[f"{name}_ms_per_sample"])
-        total = float(tp["ms_per_sample"])
-        pct = (ms / total * 100.0) if total else 0.0
-        return f"{ms:5.2f} ms | {pct:4.1f}%"
+def _format_stage_latency(name: str, throughput: ThroughputResult) -> str:
+    ms = float(getattr(throughput, f"{name}_ms_per_sample"))
+    total = float(throughput.ms_per_sample)
+    pct = (ms / total * 100.0) if total else 0.0
+    return f"{ms:5.2f} ms | {pct:4.1f}%"
+
 
+def format_evaluation_result(result: EvaluationResult) -> str:
     rows: list[list[str]] = []
 
     rows += section("SETTINGS")
     rows += [
-        ["Model", model_name],
-        ["Engine", str(engine_name).upper()],
+        ["Model", result.model_name],
+        ["Engine", str(result.engine).upper()],
     ]
 
     rows += section("PERFORMANCE")
     rows += [
-        ["Throughput", f"{tp['samples_per_s']:.2f} samples/s"],
-        ["End-to-end Latency", f"{tp['ms_per_sample']:.2f} ms/sample"],
+        [
+            "Throughput",
+            f"{result.throughput.samples_per_s:.2f} samples/s",
+        ],
+        [
+            "End-to-end Latency",
+            f"{result.throughput.ms_per_sample:.2f} ms/sample",
+        ],
     ]
 
     rows += section("STAGE BREAKDOWN", line_char="-")
     rows += [
-        ["Inference", format_stage("inference")],
-        ["Parsing", format_stage("parsing")],
-        ["Metric Update", format_stage("metric_update")],
-        ["Metric Compute", format_stage("metric_compute")],
-        ["Pipeline Overhead", format_stage("overhead")],
+        ["Inference", _format_stage_latency("inference", result.throughput)],
+        ["Parsing", _format_stage_latency("parsing", result.throughput)],
+        [
+            "Metric Update",
+            _format_stage_latency("metric_update", result.throughput),
+        ],
+        [
+            "Metric Compute",
+            _format_stage_latency("metric_compute", result.throughput),
+        ],
+        [
+            "Pipeline Overhead",
+            _format_stage_latency("overhead", result.throughput),
+        ],
     ]
 
     rows += section("QUALITY")
-    for metric_name, result in results:
+    for metric_name, metric_values in result.metrics:
         rows += section(metric_name, line_char="-")
-        for k, v in result.items():
+        for k, v in metric_values.items():
             val = f"{v * 100:.2f}%" if isinstance(v, float) else str(v)
             rows.append([str(k), val])
 

luxonis_eval/core/results.py

@@ -0,0 +1,32 @@
+from dataclasses import dataclass
+
+
+MetricValues = dict[str, float]
+MetricResult = tuple[str, MetricValues]
+MetricsResult = list[MetricResult]
+
+
+@dataclass(slots=True)
+class ThroughputResult:
+    """End-to-end throughput and latency measurements."""
+
+    elapsed_s: float
+    samples: int
+    samples_per_s: float
+    ms_per_sample: float
+    overhead_ms_per_sample: float
+    inference_ms_per_sample: float
+    parsing_ms_per_sample: float
+    metric_update_ms_per_sample: float
+    metric_compute_ms_per_sample: float
+
+
+@dataclass(slots=True)
+class EvaluationResult:
+    """Structured output returned by ``LuxonisEval.evaluate()``."""
+
+    evaluator_name: str
+    engine: str
+    model_name: str
+    metrics: MetricsResult
+    throughput: ThroughputResult