feat: Add getitrack video I/O, annotators, and detector adapters

libraries/getitrack/src/getitrack/adapters/__init__.py

@@ -0,0 +1,13 @@
+# Copyright (C) 2026 Intel Corporation
+# SPDX-License-Identifier: Apache-2.0
+"""Adapters between third-party detector frameworks and getitrack types.
+
+Each adapter is a `DetectionAdapter` subclass wrapping one detector
+instance. Framework imports stay lazy or duck-typed, so no adapter adds
+a hard dependency to getitrack.
+"""
+
+from getitrack.adapters.base import DetectionAdapter
+from getitrack.adapters.geti import GetiAdapter
+
+__all__ = ["DetectionAdapter", "GetiAdapter"]

libraries/getitrack/src/getitrack/adapters/base.py

@@ -0,0 +1,48 @@
+# Copyright (C) 2026 Intel Corporation
+# SPDX-License-Identifier: Apache-2.0
+"""Adapter interface between detector frameworks and getitrack.
+
+A `DetectionAdapter` owns one detector instance and translates between
+raw BGR frames and getitrack `Detections`, so trackers and pipelines
+consume every framework through the same two members: `detect` and
+`class_names`.
+"""
+
+from __future__ import annotations
+
+from abc import ABC, abstractmethod
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    import numpy as np
+
+    from getitrack.core.detection import Detections
+
+
+class DetectionAdapter(ABC):
+    """Wraps one detector behind a framework-agnostic interface.
+
+    Concrete adapters keep their framework imports lazy or duck-typed so
+    getitrack stays installable without the framework.
+    """
+
+    @abstractmethod
+    def detect(self, frame_bgr: np.ndarray, frame_id: int) -> Detections:
+        """Run the detector on one BGR frame.
+
+        Args:
+            frame_bgr: ``(H, W, 3)`` uint8 frame in BGR order.
+            frame_id: Frame index to stamp on the returned `Detections`.
+
+        Returns:
+            `Detections` in original frame coordinates.
+        """
+
+    @property
+    def class_names(self) -> dict[int, str] | None:
+        """Class-id-to-name mapping carried by the detector, if any.
+
+        Returns None when the framework does not expose meaningful names,
+        in which case callers supply their own table
+        """
+        return None

libraries/getitrack/src/getitrack/adapters/geti.py

@@ -0,0 +1,158 @@
+# Copyright (C) 2026 Intel Corporation
+# SPDX-License-Identifier: Apache-2.0
+"""Adapter for getitune detection models.
+
+getitune and torch are imported lazily inside the methods that need them, so
+getitrack imports without either installed. `GetiAdapter.to_detections` is
+duck-typed and works on any prediction-batch-shaped object.
+"""
+
+from __future__ import annotations
+
+import re
+from typing import Any
+
+import cv2
+import numpy as np
+
+from getitrack.adapters.base import DetectionAdapter
+from getitrack.core.detection import Detections
+
+# Matches getitune placeholder class names like "label_0", "label_1".
+_PLACEHOLDER_NAME = re.compile(r"^label_\d+$")
+
+
+class GetiAdapter(DetectionAdapter):
+    """Runs a getitune detection model on raw BGR frames.
+
+    Wraps the preprocess, ``predict_step``, and postprocess round trip
+    of a getitune Lightning detection model (RF-DETR, YOLOX, ...) for
+    inference outside the getitune Trainer.
+
+    Example:
+        >>> adapter = GetiAdapter(model, device="cuda")
+        >>> detections = adapter.detect(frame, frame_id=0)
+    """
+
+    def __init__(self, model: Any, device: str = "cpu") -> None:  # noqa: ANN401
+        """Wrap a getitune detection model.
+
+        Args:
+            model: getitune detection model in eval mode, exposing
+                ``data_input_params``, ``predict_step``, and ``label_info``.
+            device: Torch device the model lives on.
+        """
+        self.model = model
+        self.device = device
+
+    @property
+    def class_names(self) -> dict[int, str] | None:
+        """Class names read from the model's ``label_info``.
+
+        getitune models trained or loaded against a dataset carry real
+        names; models built from a bare class count carry generated
+        ``label_N`` placeholders, for which this returns None so callers
+        fall back to their own table.
+        """
+        names = getattr(getattr(self.model, "label_info", None), "label_names", None)
+        if not names or all(_PLACEHOLDER_NAME.match(name) for name in names):
+            return None
+        return dict(enumerate(names))
+
+    def detect(self, frame_bgr: np.ndarray, frame_id: int) -> Detections:
+        """Run one BGR frame through the model.
+
+        Composes `preprocess`, the model's ``predict_step``, and
+        `to_detections` into the full frame-to-detections round trip.
+        Requires getitune and torch.
+
+        Args:
+            frame_bgr: ``(H, W, 3)`` uint8 frame in BGR order.
+            frame_id: Frame index to stamp on the returned `Detections`.
+
+        Returns:
+            `Detections` for the frame, in original frame coordinates.
+        """
+        import torch
+
+        with torch.no_grad():
+            preds = self.model.predict_step(self.preprocess(frame_bgr), batch_idx=0)
+        return self.to_detections(preds, frame_id=frame_id)
+
+    def preprocess(self, frame_bgr: np.ndarray) -> Any:  # noqa: ANN401
+        """Preprocess a BGR frame into a getitune ``SampleBatch``.
+
+        Applies the resize and normalization described by the model's
+        ``data_input_params`` and sets ``scale_factor`` so predicted
+        boxes map back to the original frame coordinates. Requires
+        getitune and torch.
+
+        Args:
+            frame_bgr: ``(H, W, 3)`` uint8 frame in BGR order, e.g. from
+                ``cv2.VideoCapture``.
+
+        Returns:
+            A single-image getitune ``SampleBatch`` on ``self.device``.
+        """
+        import torch
+        from getitune.data.entity import ImageInfo, SampleBatch
+
+        inp_h, inp_w = self.model.data_input_params.input_size
+        mean = self.model.data_input_params.mean
+        std = self.model.data_input_params.std
+        ori_h, ori_w = frame_bgr.shape[:2]
+
+        rgb = cv2.cvtColor(frame_bgr, cv2.COLOR_BGR2RGB)
+        tensor = torch.from_numpy(cv2.resize(rgb, (inp_w, inp_h))).permute(2, 0, 1).float()
+        # Mean values below 1.0 indicate the model expects 0-1 normalized pixels
+        # (e.g. RF-DETR); otherwise it expects the raw 0-255 range (e.g. YOLOX).
+        if all(m < 1.0 for m in mean):
+            tensor = tensor / 255.0
+        tensor = (tensor - torch.tensor(mean).view(3, 1, 1)) / torch.tensor(std).view(3, 1, 1)
+
+        img_info = ImageInfo(
+            img_idx=0,
+            img_shape=(inp_h, inp_w),
+            ori_shape=(ori_h, ori_w),
+            scale_factor=(inp_h / ori_h, inp_w / ori_w),
+        )
+        return SampleBatch(images=tensor.unsqueeze(0).to(self.device), imgs_info=[img_info])
+
+    @staticmethod
+    def to_detections(batch: Any, frame_id: int, image_index: int = 0) -> Detections:  # noqa: ANN401
+        """Convert one image's predictions from a getitune ``PredictionBatch``.
+
+        Duck-typed: works without getitune or torch installed, on any
+        object with per-image ``bboxes``, ``scores``, and ``labels`` lists.
+
+        Args:
+            batch: A getitune ``PredictionBatch`` with per-image ``bboxes``,
+                ``scores``, and ``labels`` lists (torch tensors or arrays).
+            frame_id: Frame index to stamp on the returned `Detections`.
+            image_index: Which image of the batch to convert.
+
+        Returns:
+            `Detections` with float32 xyxy boxes, float32 scores, and int64
+            class ids.
+
+        Raises:
+            ValueError: If the batch is missing bboxes, scores, or labels.
+        """
+        if batch.bboxes is None or batch.scores is None or batch.labels is None:
+            msg = "PredictionBatch must carry bboxes, scores, and labels"
+            raise ValueError(msg)
+        return Detections(
+            bboxes=_to_numpy(batch.bboxes[image_index]).reshape(-1, 4).astype(np.float32),
+            scores=_to_numpy(batch.scores[image_index]).reshape(-1).astype(np.float32),
+            class_ids=_to_numpy(batch.labels[image_index]).reshape(-1).astype(np.int64),
+            frame_id=frame_id,
+        )
+
+
+def _to_numpy(value: Any) -> np.ndarray:  # noqa: ANN401
+    """Convert a torch tensor (possibly on an accelerator) or array-like to numpy."""
+    if hasattr(value, "cpu"):
+        value = value.cpu()
+    if hasattr(value, "numpy"):
+        value = value.numpy()
+    return np.asarray(value)

libraries/getitrack/src/getitrack/io/__init__.py

@@ -0,0 +1,7 @@
+# Copyright (C) 2026 Intel Corporation
+# SPDX-License-Identifier: Apache-2.0
+"""Video input and output built on OpenCV."""
+
+from getitrack.io.video import VideoReader, VideoWriter
+
+__all__ = ["VideoReader", "VideoWriter"]

libraries/getitrack/src/getitrack/io/video.py

@@ -0,0 +1,162 @@
+# Copyright (C) 2026 Intel Corporation
+# SPDX-License-Identifier: Apache-2.0
+"""OpenCV-backed video reader and writer.
+
+`VideoReader` iterates BGR uint8 frames from a video file. `VideoWriter`
+writes BGR uint8 frames to a video file, creating parent directories on
+demand. Both are context managers and release their OpenCV handles on close.
+"""
+
+from __future__ import annotations
+
+from pathlib import Path
+from typing import TYPE_CHECKING
+
+import cv2
+
+if TYPE_CHECKING:
+    from collections.abc import Iterator
+    from types import TracebackType
+
+    import numpy as np
+
+_FRAME_CHANNELS = 3
+
+
+class VideoReader:
+    """Sequential frame reader over a video file.
+
+    Iterating yields BGR uint8 arrays of shape ``(height, width, 3)`` in
+    decode order. Metadata (fps, frame size, frame count) is exposed as
+    properties read from the container header.
+
+    Example:
+        >>> with VideoReader("input.mp4") as reader:
+        ...     for frame in reader:
+        ...         process(frame)
+    """
+
+    def __init__(self, path: str | Path) -> None:
+        self.path = Path(path)
+        if not self.path.is_file():
+            msg = f"video file not found: {self.path}"
+            raise FileNotFoundError(msg)
+        self._cap = cv2.VideoCapture(str(self.path))
+        if not self._cap.isOpened():
+            msg = f"OpenCV could not open video: {self.path}"
+            raise ValueError(msg)
+
+    @property
+    def fps(self) -> float:
+        """Frames per second reported by the container."""
+        return float(self._cap.get(cv2.CAP_PROP_FPS))
+
+    @property
+    def width(self) -> int:
+        """Frame width in pixels."""
+        return int(self._cap.get(cv2.CAP_PROP_FRAME_WIDTH))
+
+    @property
+    def height(self) -> int:
+        """Frame height in pixels."""
+        return int(self._cap.get(cv2.CAP_PROP_FRAME_HEIGHT))
+
+    @property
+    def frame_count(self) -> int:
+        """Number of frames reported by the container header.
+
+        Some containers report an estimate; the true count is the number
+        of frames actually yielded by iteration.
+        """
+        return int(self._cap.get(cv2.CAP_PROP_FRAME_COUNT))
+
+    def __iter__(self) -> Iterator[np.ndarray]:
+        while True:
+            ok, frame = self._cap.read()
+            if not ok:
+                return
+            yield frame
+
+    def close(self) -> None:
+        """Release the underlying OpenCV capture handle."""
+        self._cap.release()
+
+    def __enter__(self) -> VideoReader:
+        return self
+
+    def __exit__(
+        self,
+        exc_type: type[BaseException] | None,
+        exc: BaseException | None,
+        tb: TracebackType | None,
+    ) -> None:
+        self.close()
+
+
+class VideoWriter:
+    """Video writer for annotated tracking output.
+
+    Frames must be BGR uint8 arrays whose size matches ``frame_size``.
+    Parent directories of ``path`` are created automatically.
+
+    Example:
+        >>> with VideoWriter("out.mp4", fps=30.0, frame_size=(640, 480)) as writer:
+        ...     writer.write(frame)
+    """
+
+    def __init__(
+        self,
+        path: str | Path,
+        fps: float,
+        frame_size: tuple[int, int],
+        codec: str = "mp4v",
+    ) -> None:
+        """Open a video file for writing.
+
+        Args:
+            path: Destination video path.
+            fps: Output frame rate.
+            frame_size: ``(width, height)`` of every frame.
+            codec: FourCC codec identifier.
+        """
+        self.path = Path(path)
+        self.path.parent.mkdir(parents=True, exist_ok=True)
+        self._frame_size = frame_size
+        fourcc = cv2.VideoWriter.fourcc(*codec)
+        self._writer = cv2.VideoWriter(str(self.path), fourcc, fps, frame_size)
+        if not self._writer.isOpened():
+            msg = f"OpenCV could not open video for writing: {self.path} (codec '{codec}')"
+            raise ValueError(msg)
+        self.frames_written = 0
+
+    def write(self, frame: np.ndarray) -> None:
+        """Append one BGR uint8 frame.
+
+        Args:
+            frame: ``(height, width, 3)`` uint8 array matching the
+                ``frame_size`` given at construction.
+
+        Raises:
+            ValueError: If the frame shape does not match ``frame_size``.
+        """
+        expected = (self._frame_size[1], self._frame_size[0], _FRAME_CHANNELS)
+        if frame.shape != expected:
+            msg = f"frame shape {frame.shape} does not match expected {expected}"
+            raise ValueError(msg)
+        self._writer.write(frame)
+        self.frames_written += 1
+
+    def close(self) -> None:
+        """Finalise the container and release the OpenCV writer handle."""
+        self._writer.release()
+
+    def __enter__(self) -> VideoWriter:
+        return self
+
+    def __exit__(
+        self,
+        exc_type: type[BaseException] | None,
+        exc: BaseException | None,
+        tb: TracebackType | None,
+    ) -> None:
+        self.close()