Extract and attach context once at runtime once per run and remove ambiguous kwargs

README.md

@@ -498,7 +498,9 @@ The framework consumes the returned `ModelSpec` to configure loader preprocessin
 
 Subclass [`BaseParser`](luxonis_eval/parsers/base_parser.py) and implement the single abstract method:
 
-- **`parse(raw_output, **kwargs)`** - Convert raw engine output into a structured prediction format
+- **`parse(output)`** - Convert raw engine output into a structured prediction format
+
+Parser configuration belongs in the parser itself, and LuxonisEval provides the remaining runtime information during setup.
 
 The parser bridges the gap between model-specific tensor layouts and the standardized prediction fields that downstream metrics expect. The built-in parsers populate [`Prediction`](luxonis_eval/parsers/predictions.py) as follows:
 
@@ -515,7 +517,7 @@ Subclass [`BaseMetric`](luxonis_eval/metrics/base_metric.py) and implement the f
 
 - **`required_target_keys()`** - Declare which annotation keys the metric requires
 - **`reset()`** - Reset internal state such as counters or accumulators
-- **`update(predictions, target, **kwargs)`** - Update the metric state for one sample
+- **`update(predictions, target)`** - Update the metric state for one sample
 - **`compute()`** - Return the final metric values
 
 > [!IMPORTANT]

luxonis_eval/core/context.py

@@ -0,0 +1,32 @@
+from collections.abc import Callable
+from dataclasses import dataclass
+
+import numpy as np
+
+from luxonis_eval.engines.base_engine import ModelSpec
+
+TargetConverter = Callable[
+    [np.ndarray, int, int],
+    tuple[np.ndarray, np.ndarray],
+]
+
+
+@dataclass(frozen=True, slots=True)
+class EvalContext:
+    """Static runtime metadata shared across one evaluation run."""
+
+    model_spec: ModelSpec
+    class_map: dict[int, str]
+    target_class_map: dict[int, str]
+    class_index_map: dict[int, int] | None
+    category_ids: tuple[int, ...]
+    target_background_index: int | None
+    target_converter: TargetConverter
+
+    @property
+    def width(self) -> int:
+        return self.model_spec.width
+
+    @property
+    def height(self) -> int:
+        return self.model_spec.height

luxonis_eval/core/core.py

@@ -8,6 +8,7 @@
 from luxonis_ml.typing import Params, PathType
 
 from luxonis_eval.config import EvalConfig, EvaluatorConfig
+from luxonis_eval.core.context import EvalContext
 from luxonis_eval.core.factories import (
     create_engine,
     create_loader,
@@ -22,7 +23,7 @@
     get_model_name,
 )
 from luxonis_eval.core.runtime import (
-    build_metric_contexts,
+    build_eval_context,
     normalize_target,
     resolve_class_mapping,
     select_evaluator_outputs,
@@ -103,13 +104,13 @@ def setup(self) -> None:
                 loader_params=self.cfg.pipeline.loader.params,
                 loader_task_name=self.loader_task_name,
             )
-            self.metric_contexts = build_metric_contexts(
-                self.evaluator_cfg,
+            self.eval_context = build_eval_context(
                 model_spec=self.model_spec,
                 ldf_class_map=self.ldf_class_map,
                 class_map=self.class_map,
                 class_index_map=self.class_index_map,
             )
+            self._attach_eval_context()
             self._sanity_check_pipeline()
         except Exception:
             try:
@@ -147,7 +148,6 @@ def _run_evaluators(self) -> EvaluationResult:
         assert self.parser is not None
         assert self.throughput_metric is not None
         assert self.evaluator_cfg is not None
-        assert self.model_spec is not None
 
         with self._progress(
             f"Running {engine_name.upper()} inference ({model_name})...",
@@ -170,21 +170,15 @@ def _run_evaluators(self) -> EvaluationResult:
                     select_evaluator_outputs(
                         raw_output,
                         self.evaluator_cfg.outputs,
-                    ),
-                    model_spec=self.model_spec,
-                    class_map=self.class_map,
-                    **self.evaluator_cfg.parser.params,
+                    )
                 )
                 parsing_elapsed = time.perf_counter() - parsing_t0
 
                 metric_update_t0 = time.perf_counter()
-                for metric, metric_ctx in zip(
-                    self.metrics, self.metric_contexts, strict=True
-                ):
+                for metric in self.metrics:
                     metric.update(
                         predictions=predictions,
                         target=target,
-                        **metric_ctx,
                     )
                 metric_update_elapsed = (
                     time.perf_counter() - metric_update_t0
@@ -196,21 +190,8 @@ def _run_evaluators(self) -> EvaluationResult:
                     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,
-                    )
+                for visualizer in self.visualizers:
+                    visualizer.visualize(predictions, self.engine.vis_frame())
                 progress.update(advance=1)
 
         metric_compute_t0 = time.perf_counter()
@@ -258,7 +239,6 @@ def _sanity_check_pipeline(self) -> None:
         assert self.engine is not None
         assert self.parser is not None
         assert self.evaluator_cfg is not None
-        assert self.model_spec is not None
 
         if len(self.loader) == 0:
             raise ValueError(
@@ -276,15 +256,10 @@ def _sanity_check_pipeline(self) -> None:
         )
         raw_output = self.engine.infer_once(img)
         predictions = self.parser.parse(
-            select_evaluator_outputs(raw_output, self.evaluator_cfg.outputs),
-            model_spec=self.model_spec,
-            class_map=self.class_map,
-            **self.evaluator_cfg.parser.params,
+            select_evaluator_outputs(raw_output, self.evaluator_cfg.outputs)
         )
 
-        for metric, metric_ctx in zip(
-            self.metrics, self.metric_contexts, strict=True
-        ):
+        for metric in self.metrics:
             missing = set(metric.required_target_keys()) - set(target)
             if missing:
                 raise ValueError(
@@ -296,7 +271,6 @@ def _sanity_check_pipeline(self) -> None:
             metric.update(
                 predictions=predictions,
                 target=target,
-                **metric_ctx,
             )
             metric.compute()
             metric.reset()
@@ -310,13 +284,27 @@ def _clear_runtime_fields(self) -> None:
         self.visualizers: list[BaseVisualizer] = []
         self.evaluator_cfg: EvaluatorConfig | None = None
 
+        self.eval_context: EvalContext | None = None
         self.model_spec: ModelSpec | None = None
         self.loader_task_name: str | None = None
 
         self.ldf_class_map: dict[int, str] = {}
         self.class_map: dict[int, str] = {}
         self.class_index_map: dict[int, int] | None = None
-        self.metric_contexts: list[dict[str, Any]] = []
+
+    def _attach_eval_context(self) -> None:
+        if self.eval_context is None:
+            raise RuntimeError(
+                "Evaluation context is unavailable before setup."
+            )
+        if self.parser is None:
+            raise RuntimeError("Parser is unavailable before setup.")
+
+        self.parser.attach_context(self.eval_context)
+        for metric in self.metrics:
+            metric.attach_context(self.eval_context)
+        for visualizer in self.visualizers:
+            visualizer.attach_context(self.eval_context)
 
     def _require_setup(self) -> None:
         if not self._is_setup:

luxonis_eval/core/runtime.py

@@ -1,14 +1,13 @@
 import json
 from importlib.resources import files
-from typing import Any
 
 import numpy as np
 from loguru import logger
 from luxonis_ml.data.loaders import LuxonisLoader
 from luxonis_ml.data.utils import split_task
 from luxonis_ml.typing import Params
 
-from luxonis_eval.config import EvaluatorConfig
+from luxonis_eval.core.context import EvalContext
 from luxonis_eval.engines.base_engine import ModelSpec
 from luxonis_eval.engines.io import EngineOutput
 from luxonis_eval.loaders.base_loader import BaseEvalLoader
@@ -40,46 +39,22 @@ def resolve_class_mapping(
     return loader.get_class_mapping(**loader_params)
 
 
-def build_metric_contexts(
-    evaluator_cfg: EvaluatorConfig,
+def build_eval_context(
     model_spec: ModelSpec,
     ldf_class_map: dict[int, str],
     class_map: dict[int, str],
     class_index_map: dict[int, int] | None,
-) -> list[dict[str, Any]]:
-    return [
-        get_metric_ctx(
-            base_ctx=metric_cfg.params,
-            width=model_spec.width,
-            height=model_spec.height,
-            ldf_class_map=ldf_class_map,
-            class_map=class_map,
-            class_index_map=class_index_map,
-        )
-        for metric_cfg in evaluator_cfg.metrics
-    ]
-
-
-def get_metric_ctx(base_ctx: dict[str, Any], **kwargs: Any) -> dict[str, Any]:
-    class_index_map = kwargs.get("class_index_map")
-    class_map = kwargs.get("class_map") or {}
-    ldf_class_map = kwargs.get("ldf_class_map") or {}
-    width = kwargs.get("width", -1)
-    height = kwargs.get("height", -1)
-
+) -> EvalContext:
     ldf_name_to_idx = {v: k for k, v in ldf_class_map.items()}
-
-    return {
-        **base_ctx,
-        "class_map": class_map,
-        "class_index_map": class_index_map,
-        "width": width,
-        "height": height,
-        "category_ids": sorted(class_map.keys()),
-        "target_converter": normalized_xywh_to_coco_xywh,
-        "target_bg": ldf_name_to_idx.get("background"),
-        "target_class_map": ldf_class_map,
-    }
+    return EvalContext(
+        model_spec=model_spec,
+        class_map=class_map,
+        target_class_map=ldf_class_map,
+        class_index_map=class_index_map,
+        category_ids=tuple(sorted(class_map.keys())),
+        target_background_index=ldf_name_to_idx.get("background"),
+        target_converter=normalized_xywh_to_coco_xywh,
+    )
 
 
 def select_evaluator_outputs(

luxonis_eval/metrics/base_metric.py

@@ -4,6 +4,7 @@
 import numpy as np
 from luxonis_ml.utils.registry import AutoRegisterMeta
 
+from luxonis_eval.core.context import EvalContext
 from luxonis_eval.parsers.predictions import Prediction
 from luxonis_eval.registry import METRICS_REGISTRY
 
@@ -23,8 +24,23 @@ def __init__(self, **kwargs: Any) -> None:
         **kwargs : Any
             Metric basic configuration.
         """
+        del kwargs
+        self._context: EvalContext | None = None
         self.reset()
 
+    def attach_context(self, context: EvalContext) -> None:
+        """Attach evaluation runtime metadata after setup."""
+        self._context = context
+
+    def require_context(self) -> EvalContext:
+        """Return the attached evaluation context."""
+        if self._context is None:
+            raise RuntimeError(
+                f"{type(self).__name__} is missing evaluation context. "
+                "Call attach_context() during setup before update()."
+            )
+        return self._context
+
     @abstractmethod
     def required_target_keys(self) -> list[str]:
         """Return the ground-truth keys required by the metric."""
@@ -40,7 +56,6 @@ def update(
         self,
         predictions: Prediction,
         target: dict[str, np.ndarray],
-        **kwargs: Any,
     ) -> None:
         """Update the metric with predictions and ground truths.
 
@@ -50,8 +65,6 @@ def update(
             Structured model predictions.
         target : dict[str, np.ndarray]
             Ground-truth data.
-        **kwargs : Any
-            Additional context.
         """
         ...
 

luxonis_eval/metrics/bbox_map.py

@@ -46,7 +46,6 @@ def update(
         self,
         predictions: Prediction,
         target: dict[str, np.ndarray],
-        **kwargs: Any,
     ) -> None:
         """Update internal metric state.
 
@@ -56,22 +55,17 @@ def update(
             Structured detection predictions.
         target : dict[str, np.ndarray]
             Ground-truth data.
-        **kwargs : Any
-            Additional context.
         """
+        context = self.require_context()
         detections = predictions.require_detections()
         target_boxes = target[self.required_target_keys()[0]]
-        width = int(kwargs["width"])
-        height = int(kwargs["height"])
-
-        class_map: dict[int, str] = kwargs.get("class_map", {})
-        category_ids: Sequence[int] | None = kwargs.get("category_ids")
-        class_index_map = kwargs.get("class_index_map")
-        target_converter = kwargs.get("target_converter")
-        if target_converter is None:
-            raise ValueError(
-                "BboxMeanAveragePrecision requires target_converter in ctx."
-            )
+        width = context.width
+        height = context.height
+
+        class_map = context.class_map
+        category_ids: Sequence[int] = context.category_ids
+        class_index_map = context.class_index_map
+        target_converter = context.target_converter
 
         self._store.init_categories_once(
             class_map=class_map, category_ids=category_ids

luxonis_eval/metrics/dice_coef.py

@@ -64,7 +64,6 @@ def update(
         self,
         predictions: Prediction,
         target: dict[str, np.ndarray],
-        **kwargs: Any,
     ) -> None:
         """Update internal metric state.
 
@@ -74,17 +73,16 @@ def update(
             Structured segmentation predictions.
         target : dict[str, np.ndarray]
             Ground-truth labels.
-        **kwargs : Any
-            Additional context.
         """
+        context = self.require_context()
         if self.target_class_map is None:
-            self.target_class_map = kwargs.get("target_class_map", {})
+            self.target_class_map = context.target_class_map
         prepared = prepare_segmentation_metric_inputs(
             predictions,
             target,
             include_background=self.include_background,
-            target_bg=kwargs.get("target_bg"),
-            class_index_map=kwargs.get("class_index_map"),
+            target_bg=context.target_background_index,
+            class_index_map=context.class_index_map,
         )
         self.metric.update(prepared.pred_tensor, prepared.target_tensor)