[McByte part 2] Feat/mcbyte mask manager

src/trackers/core/mcbyte/mask_manager.py

@@ -0,0 +1,72 @@
+# ------------------------------------------------------------------------
+# Trackers
+# Copyright (c) 2026 Roboflow. All Rights Reserved.
+# Licensed under the Apache License, Version 2.0 [see LICENSE for details]
+# ------------------------------------------------------------------------
+
+from __future__ import annotations
+
+import numpy as np
+
+from trackers.core.mcbyte.masks.base import (
+    MaskGenerator,
+    MaskOutput,
+    MaskPropagator,
+    TrackletSnapshot,
+)
+
+
+class MaskManager:
+    """Manage McByte mask generation and propagation.
+
+    The manager follows the original McByte timing: masks for the current frame
+    are prepared before association, but they are initialized/updated from tracker
+    outputs produced on the previous frame.
+    """
+
+    def __init__(
+        self,
+        mask_generator: MaskGenerator,
+        mask_propagator: MaskPropagator | None = None,
+    ) -> None:
+        self.mask_generator = mask_generator
+        self.mask_propagator = mask_propagator
+        self._initialized = False
+
+    def reset(self) -> None:
+        self._initialized = False
+        if self.mask_propagator is not None:
+            self.mask_propagator.reset()
+
+    def get_updated_masks(
+        self,
+        frame: np.ndarray,
+        previous_frame: np.ndarray | None,
+        previous_tracklets: list[TrackletSnapshot],
+    ) -> MaskOutput | None:
+        """Return masks for the current frame.
+
+        No masks are returned until at least one previous frame and previous
+        tracker output are available.
+
+        If a propagator is configured, masks are initialized from the previous
+        frame and propagated to the current frame. If propagation is unavailable
+        or fails, ``None`` is returned to avoid using stale or misaligned masks.
+        """
+        if previous_frame is None or len(previous_tracklets) == 0:
+            return None
+
+        if self.mask_propagator is None:
+            return None
+
+        if not self._initialized:
+            mask_output = self.mask_generator.generate(previous_frame, previous_tracklets)
+            self.mask_propagator.initialize(previous_frame, mask_output)
+            self._initialized = True
+
+        propagated_output = self.mask_propagator.propagate(frame)
+        if propagated_output is not None:
+            return propagated_output
+
+        self._initialized = False
+        return None

src/trackers/core/mcbyte/masks/__init__.py

@@ -0,0 +1,19 @@
+# ------------------------------------------------------------------------
+# Trackers
+# Copyright (c) 2026 Roboflow. All Rights Reserved.
+# Licensed under the Apache License, Version 2.0 [see LICENSE for details]
+# ------------------------------------------------------------------------
+
+from trackers.core.mcbyte.masks.base import (
+    MaskGenerator,
+    MaskOutput,
+    MaskPropagator,
+    TrackletSnapshot,
+)
+
+__all__ = [
+    "MaskGenerator",
+    "MaskOutput",
+    "MaskPropagator",
+    "TrackletSnapshot",
+]

src/trackers/core/mcbyte/masks/base.py

@@ -0,0 +1,64 @@
+# ------------------------------------------------------------------------
+# Trackers
+# Copyright (c) 2026 Roboflow. All Rights Reserved.
+# Licensed under the Apache License, Version 2.0 [see LICENSE for details]
+# ------------------------------------------------------------------------
+
+from __future__ import annotations
+
+from abc import ABC, abstractmethod
+from dataclasses import dataclass
+
+import numpy as np
+
+
+@dataclass(frozen=True)
+class TrackletSnapshot:
+    """Minimal tracker state needed by mask components."""
+
+    tracker_id: int
+    xyxy: np.ndarray
+
+
+@dataclass(frozen=True)
+class MaskOutput:
+    """Mask information produced before McByte association."""
+
+    masks: np.ndarray | None
+    tracklet_mask_dict: dict[int, int]
+    mask_avg_prob_dict: dict[int, float] | None = None
+
+
+class MaskGenerator(ABC):
+    """Generate masks from tracklet boxes."""
+
+    @abstractmethod
+    def generate(
+        self,
+        frame: np.ndarray,
+        tracklets: list[TrackletSnapshot],
+    ) -> MaskOutput:
+        """Generate masks for the given tracklet snapshots."""
+
+
+class MaskPropagator(ABC):
+    """Propagate masks from one frame to the next."""
+
+    @abstractmethod
+    def reset(self) -> None:
+        """Reset propagation state."""
+
+    @abstractmethod
+    def initialize(
+        self,
+        frame: np.ndarray,
+        mask_output: MaskOutput,
+    ) -> None:
+        """Initialize propagation state."""
+
+    @abstractmethod
+    def propagate(
+        self,
+        frame: np.ndarray,
+    ) -> MaskOutput | None:
+        """Propagate masks to the current frame."""

src/trackers/core/mcbyte/masks/dummy.py

@@ -0,0 +1,84 @@
+# ------------------------------------------------------------------------
+# Trackers
+# Copyright (c) 2026 Roboflow. All Rights Reserved.
+# Licensed under the Apache License, Version 2.0 [see LICENSE for details]
+# ------------------------------------------------------------------------
+
+from __future__ import annotations
+
+import numpy as np
+
+from trackers.core.mcbyte.masks.base import (
+    MaskGenerator,
+    MaskOutput,
+    MaskPropagator,
+    TrackletSnapshot,
+)
+
+
+class DummyBoxMaskGenerator(MaskGenerator):
+    """Generate rectangular binary masks from tracklet bounding boxes."""
+
+    def generate(
+        self,
+        frame: np.ndarray,
+        tracklets: list[TrackletSnapshot],
+    ) -> MaskOutput:
+        height, width = frame.shape[:2]
+        masks = np.zeros((len(tracklets), height, width), dtype=bool)
+        tracklet_mask_dict: dict[int, int] = {}
+
+        for mask_index, tracklet in enumerate(tracklets):
+            x1, y1, x2, y2 = tracklet.xyxy.astype(int)
+
+            x1 = int(np.clip(x1, 0, width))
+            x2 = int(np.clip(x2, 0, width))
+            y1 = int(np.clip(y1, 0, height))
+            y2 = int(np.clip(y2, 0, height))
+
+            masks[mask_index, y1:y2, x1:x2] = True
+            tracklet_mask_dict[tracklet.tracker_id] = mask_index
+
+        return MaskOutput(
+            masks=masks,
+            tracklet_mask_dict=tracklet_mask_dict,
+            mask_avg_prob_dict=None,
+        )
+
+
+class DummyIdentityMaskPropagator(MaskPropagator):
+    """Return the last initialized mask output unchanged."""
+
+    def __init__(self) -> None:
+        self._mask_output: MaskOutput | None = None
+
+    def reset(self) -> None:
+        self._mask_output = None
+
+    def initialize(
+        self,
+        frame: np.ndarray,
+        mask_output: MaskOutput,
+    ) -> None:
+        self._mask_output = MaskOutput(
+            masks=None if mask_output.masks is None else mask_output.masks.copy(),
+            tracklet_mask_dict=mask_output.tracklet_mask_dict.copy(),
+            mask_avg_prob_dict=(
+                None if mask_output.mask_avg_prob_dict is None else mask_output.mask_avg_prob_dict.copy()
+            ),
+        )
+
+    def propagate(
+        self,
+        frame: np.ndarray,
+    ) -> MaskOutput | None:
+        if self._mask_output is None:
+            return None
+
+        return MaskOutput(
+            masks=None if self._mask_output.masks is None else self._mask_output.masks.copy(),
+            tracklet_mask_dict=self._mask_output.tracklet_mask_dict.copy(),
+            mask_avg_prob_dict=(
+                None if self._mask_output.mask_avg_prob_dict is None else self._mask_output.mask_avg_prob_dict.copy()
+            ),
+        )