"""
Base class for all detectors (method and feature).
Provides common interface and shared functionality.
"""
from abc import ABC, abstractmethod
from typing import Any, Dict, List, Optional
from ..configs.schemas.detectors_instances_configs import BaseAlgorithmConfig
from ..configs.video_runtime_config import SequenceRuntimeConfig
from .data import SequenceData
from .in_out import SequenceIO
[docs]class BaseDetector(ABC):
"""
Abstract base class for ALL detectors.
Defines the common interface that both method and feature detectors implement.
This enables a unified detector loop in main.py.
"""
# Each detector should have an unique algorithm type
algorithm_type: str
# Instance attributes set during initialization
data: SequenceData
io: SequenceIO
sequence_context: SequenceRuntimeConfig
detector_config: BaseAlgorithmConfig
# User-defined instance name from TOML key (set in __init__)
algorithm_instance: str
# Class attributes to be defined by subclasses
inference_config: Any
components: List[str]
# Class-level discriminator that matches an entry in ALL_DETECTORS and the
# registry key in DETECTORS_REGISTRY (e.g. "mmpose_2d", "gaze_distance").
algorithm_type: str
# Additional attributes
visualize: bool
def __init__(
self,
io: SequenceIO,
data: SequenceData,
sequence_context: SequenceRuntimeConfig,
algorithm_instance: str,
) -> None:
"""
Initialize base detector with references.
Subclasses should call super().__init__() and set inference_config.
"""
self.io = io
self.data = data
self.sequence_context = sequence_context
self.algorithm_instance = algorithm_instance
self.detector_config = sequence_context.get_detector_config(algorithm_instance)
self.visualize = getattr(self.detector_config, "visualize", False)
# Allow detector configs (e.g. MMPose 2D) to declare components per-instance.
# Falls back to the subclass's class-level `components` attribute.
config_components = getattr(self.detector_config, "components", None)
if config_components:
self.components = list(config_components)
[docs] @abstractmethod
def run(self) -> Optional[Any]:
"""
Execute the detector's main computation.
For method detectors: runs subprocess inference + post_inference()
For feature detectors: runs compute()
Returns:
Optional data for visualization (feature detectors currently return computed data)
"""
pass
[docs] @abstractmethod
def visualization(self, data: Any) -> None:
"""
Visualize detector output.
Args:
data: Output from run() or external data source
"""
pass
# -------------------------------------------------------------------------
# Shared Helper Methods
# -------------------------------------------------------------------------
@property
def predictions_mapping(self):
"""Access predictions mapping from runtime config."""
return self.sequence_context.predictions_mapping
[docs] def compute_result_folders(self) -> Dict[str, str]:
"""Compute result folders for all components."""
return {
comp: str(self.io.get_detector_output_folder(comp, self.algorithm_instance, "result"))
for comp in self.components
}
[docs] def compute_output_folders(self, requires_out_folder: bool) -> Dict[str, str]:
"""Compute extra output folders for all components."""
if requires_out_folder:
return {
comp: str(self.io.get_detector_output_folder(comp, self.algorithm_instance, "output"))
for comp in self.components
}
return {}
[docs] def compute_viz_folders(self, visualize: bool) -> Dict[str, str]:
"""Compute visualization folders for all components."""
if visualize:
return {
comp: str(self.io.get_detector_output_folder(comp, self.algorithm_instance, "visualization"))
for comp in self.components
}
return {}
def __str__(self) -> str:
config_name = type(self.inference_config).__name__
return f"Detector: {self.algorithm_instance}\n Components: {self.components}\n Config: {config_name}\n"