import warnings
from collections import defaultdict, deque
from copy import deepcopy
import numpy as np
from supervision.config import ORIENTED_BOX_COORDINATES
from supervision.detection.core import Detections
from supervision.detection.utils.boxes import xyxyxyxy_to_xyxy
from supervision.utils.internal import SupervisionWarnings
def _align_oriented_corners(corners: np.ndarray, reference: np.ndarray) -> np.ndarray:
"""Align OBB corner starts and winding with a reference representation."""
variants = [corners, corners[:, ::-1]]
candidates = np.stack(
[
np.roll(candidate, shift, axis=1)
for candidate in variants
for shift in range(4)
]
)
distances = np.sum((candidates - reference[None, ...]) ** 2, axis=(-2, -1))
best = np.argmin(distances, axis=0)
return np.asarray(candidates[best, np.arange(corners.shape[0])])
class DetectionsSmoother:
"""
A utility class for smoothing detections over multiple frames in video tracking.
It maintains a history of detections for each track and provides smoothed
predictions based on these histories.
!!! warning
- `DetectionsSmoother` requires the `tracker_id` for each detection. Refer to
[Roboflow Trackers](/latest/trackers/) for
information on integrating tracking into your inference pipeline.
- This class is not compatible with segmentation models.
- When detections in a frame disagree on confidence presence — some tracks
carry confidence scores and others do not — `confidence` is set to `None`
for all smoothed detections in that frame.
Example:
```pycon
>>> import numpy as np
>>> import supervision as sv
>>> smoother = sv.DetectionsSmoother(length=3)
>>> detections_1 = sv.Detections(
... xyxy=np.array([[0, 0, 10, 10]]),
... confidence=np.array([0.5]),
... tracker_id=np.array([1])
... )
>>> detections_2 = sv.Detections(
... xyxy=np.array([[2, 2, 12, 12]]),
... confidence=np.array([0.7]),
... tracker_id=np.array([1])
... )
>>> smoothed = smoother.update_with_detections(detections_1)
>>> smoothed.xyxy
array([[ 0., 0., 10., 10.]])
>>> smoothed = smoother.update_with_detections(detections_2)
>>> smoothed.xyxy
array([[ 1., 1., 11., 11.]])
>>> smoothed.confidence
array([0.6])
```
```python
import supervision as sv
from rfdetr import RFDETRMedium
video_info = sv.VideoInfo.from_video_path(video_path="")
frame_generator = sv.get_video_frames_generator(
source_path="")
model = RFDETRMedium()
tracker = sv.ByteTrack(frame_rate=video_info.fps)
smoother = sv.DetectionsSmoother()
box_annotator = sv.BoxAnnotator()
with sv.VideoSink("", video_info=video_info) as sink:
for frame in frame_generator:
detections = model.predict(frame[:, :, ::-1])
detections = tracker.update_with_detections(detections)
detections = smoother.update_with_detections(detections)
annotated_frame = box_annotator.annotate(frame.copy(), detections)
sink.write_frame(annotated_frame)
```
"""
def __init__(self, length: int = 5) -> None:
"""
Args:
length: The maximum number of frames to consider for smoothing
detections. Defaults to 5.
"""
self.tracks: defaultdict[int, deque[Detections | None]] = defaultdict(
lambda: deque(maxlen=length)
)
def reset(self) -> None:
"""
Clears the per-track detection history so the smoother can be reused
across independent streams without carrying over frames from a
previous stream. The configured window `length` is preserved.
Examples:
```pycon
>>> import numpy as np
>>> import supervision as sv
>>> smoother = sv.DetectionsSmoother()
>>> detections = sv.Detections(
... xyxy=np.array([[0, 0, 10, 10]]),
... confidence=np.array([0.5]),
... tracker_id=np.array([1])
... )
>>> _ = smoother.update_with_detections(detections)
>>> len(smoother.tracks)
1
>>> smoother.reset()
>>> len(smoother.tracks)
0
```
"""
self.tracks.clear()
def update_with_detections(self, detections: Detections) -> Detections:
"""
Updates the smoother with a new set of detections from a frame.
Args:
detections: The detections to add to the smoother.
"""
if detections.tracker_id is None:
warnings.warn(
"Smoothing skipped. DetectionsSmoother requires tracker_id. Refer to "
"https://supervision.roboflow.com/latest/trackers for more "
"information.",
category=SupervisionWarnings,
)
return detections
for detection_idx in range(len(detections)):
tracker_id_value = detections.tracker_id[detection_idx]
tracker_id = int(tracker_id_value)
self.tracks[tracker_id].append(detections.select(detection_idx))
active_ids = set(detections.tracker_id.tolist())
for track_id in self.tracks.keys():
if track_id not in active_ids:
self.tracks[track_id].append(None)
for track_id in list(self.tracks.keys()):
if all(d is None for d in self.tracks[track_id]):
del self.tracks[track_id]
current_track_ids = active_ids
return self.get_smoothed_detections(track_ids=current_track_ids)
def get_track(self, track_id: int) -> Detections | None:
"""Return the smoothed `Detections` for a single track.
Averages `xyxy` over all valid (non-`None`) frames in the track window.
`confidence` is averaged only over frames that carry it; frames with
`confidence=None` are excluded. When every valid frame carries compatible
oriented-box corners, those corners are aligned and averaged too, and
`xyxy` is derived from their resulting envelope. Mixed or incompatible
oriented-box metadata is dropped to keep the returned geometry consistent.
Returns `None` when the track is unknown or its entire window is empty.
Args:
track_id: The tracker ID whose smoothed detection to retrieve.
Returns:
Smoothed `Detections` for the track, or `None` if the track is
unknown or all frames in its window are empty.
"""
track = self.tracks.get(track_id, None)
if track is None:
return None
valid: list[Detections] = [d for d in track if d is not None]
if len(valid) == 0:
return None
ret = deepcopy(valid[0])
ret.xyxy = np.mean(np.stack([d.xyxy for d in valid], axis=0), axis=0)
# Average confidence only over frames that carry it; frames with
# confidence=None contribute nothing to the mean. Retain None when
# no frame in the window carries confidence.
confidences = [d.confidence for d in valid if d.confidence is not None]
ret.confidence = np.mean(np.array(confidences), axis=0) if confidences else None
corner_sets = [
np.asarray(d.data[ORIENTED_BOX_COORDINATES])
for d in valid
if ORIENTED_BOX_COORDINATES in d.data
]
if len(corner_sets) == len(valid) and len({c.shape for c in corner_sets}) == 1:
reference = corner_sets[0]
aligned = [
_align_oriented_corners(corners, reference) for corners in corner_sets
]
smoothed_corners = np.mean(np.stack(aligned, axis=0), axis=0)
ret.data[ORIENTED_BOX_COORDINATES] = smoothed_corners
ret.xyxy = xyxyxyxy_to_xyxy(smoothed_corners)
else:
ret.data.pop(ORIENTED_BOX_COORDINATES, None)
return ret
def get_smoothed_detections(self, track_ids: set[int] | None = None) -> Detections:
"""Return the smoothed detections for the requested active tracks.
Args:
track_ids: Optional set of track IDs to include in the output. When
provided, tracks absent from the current frame are excluded from the
emitted detections but their history stays cached.
"""
tracked_detections = []
for track_id in self.tracks:
if track_ids is not None and track_id not in track_ids:
continue
track = self.get_track(track_id)
if track is not None:
tracked_detections.append(track)
# Detections.merge requires all-or-none for optional fields.
# When tracks disagree on confidence presence, drop it from all to
# prevent ValueError inside Detections.merge (stack_or_none invariant).
if tracked_detections and any(d.confidence is None for d in tracked_detections):
for d in tracked_detections:
d.confidence = None
detections = Detections.merge(tracked_detections)
if len(detections) == 0:
detections.tracker_id = np.array([], dtype=int)
return detections