Stabilizer Class References

class StabilizerMode(Enum):
    """
    Selects the video stabilization algorithm.

    | Value | Backend | Notes |
    |:-----:|:--------|:------|
    | `StabilizerMode.ASW` | Average Sliding-Window stabilizer | **Default** |
    | `StabilizerMode.KALMAN` | Zero-latency Kalman-filter stabilizer | Real-time, no frame buffering |
    """

    ASW = "asw"  # Average Sliding-Window (default)
    KALMAN = "kalman"  # zero-latency Kalman filter

class ASWStabilizer(_StabilizerBase):
    """
    Average Sliding-Window Video Stabilizer.

    Tracks a salient feature array over a fixed-size window of past frames and
    cancels perturbations relative to a normalized box-filter-smoothed path.
    Relies on **Threaded Queue mode** for error-free & ultra-fast frame handling.
    """

    def __init__(
        self,
        smoothing_radius: int = 25,
        border_type: str = "black",
        border_size: int = 0,
        crop_n_zoom: bool = False,
        logging: bool = False,
    ):
        """
        Parameters:
            smoothing_radius (int): alter averaging window size.
            border_type (str): changes the extended border type.
            border_size (int): enables and set the value for extended border size to reduce the black borders.
            crop_n_zoom (bool): enables cropping and zooming of frames(to original size) to reduce the black borders.
            logging (bool): enables/disables logging.
        """
        super().__init__(
            border_type=border_type,
            border_size=border_size,
            crop_n_zoom=crop_n_zoom,
            logging=logging,
        )

        # bounded frame buffer (size = smoothing window)
        self.__frame_queue = deque(maxlen=smoothing_radius)

        # Bounded deque for prev_to_cur transforms [dx, dy, da].
        # A box filter of width `smoothing_radius` at the output position needs
        # transforms within a half-window of `smoothing_radius/2` around it.
        # The output frame lags the newest by `smoothing_radius - 1` frames,
        # so keeping the last `2 * smoothing_radius + 1` transforms always covers
        # the window with margin. Older entries can be dropped because the
        # `smoothed_path - path` subtraction is invariant to the absolute offset
        # of the cumulative sum. Caps memory at O(smoothing_radius) regardless
        # of stream length.
        self.__transforms = deque(maxlen=2 * smoothing_radius + 1)

        # ASW-specific state
        self.__smoothing_radius = smoothing_radius  # averaging window
        # latches True once the frame queue first fills; from then on every
        # frame emits an output. Replaces the original monotonic counter that
        # leaked memory by tracking every index forever.
        self.__buildup_complete = False
        self.__previous_gray = None  # previous gray frame
        self.__previous_keypoints = None  # previous GFTT keypoints

        # normalized box filter
        self.__box_filter = np.ones(smoothing_radius) / smoothing_radius

    def stabilize(self, frame: np.ndarray) -> np.ndarray | None:
        """
        Takes an unstabilized video frame, and returns a stabilized one
        (or `None` while the smoothing window is still filling).
        """
        if frame is None:
            return

        # save frame size for zooming
        if self._crop_n_zoom and self._frame_size is None:
            self._frame_size = frame.shape[:2]

        # initiate transformations capturing
        if not self.__frame_queue:
            # for first frame
            previous_gray = self._to_gray(frame)
            self.__previous_keypoints = self._detect_keypoints(previous_gray)
            self._frame_height, self._frame_width = frame.shape[:2]
            self.__frame_queue.append(frame)
            self.__previous_gray = previous_gray[:]
            return None

        # Latch `buildup_complete` the first time the queue is observed at
        # capacity (the upcoming append will be the first to drop an old
        # frame). Once latched, stays True: every subsequent frame emits.
        if (
            not self.__buildup_complete
            and len(self.__frame_queue) == self.__frame_queue.maxlen
        ):
            self.__buildup_complete = True

        # buffer the new frame and compute its prev->cur transform
        self.__frame_queue.append(frame)
        self.__generate_transformations()

        # still warming up — no output yet
        if not self.__buildup_complete:
            return None

        # Build path + smoothed path from the BOUNDED transform window.
        # O(smoothing_radius) work per frame regardless of stream length;
        # the original code was O(total_frames_seen) per frame.
        transforms_arr = np.asarray(self.__transforms, dtype="float32")
        path = np.cumsum(transforms_arr, axis=0)
        smoothed_path = np.copy(path)
        for i in range(3):
            smoothed_path[:, i] = self.__box_filter_convolve(
                path[:, i], window_size=self.__smoothing_radius
            )
        # deviation is translation-invariant w.r.t. absolute path offset, so
        # rebasing the cumsum to start at the current deque head is harmless.
        deviation = smoothed_path - path
        frame_transforms_smoothed = transforms_arr + deviation

        # Locate the output frame's transform row inside the bounded window.
        # After processing frame index k (0-indexed), the newest transform in
        # the deque is T_{k-1->k} at position len(deque)-1. The output frame
        # (leftmost of frame_queue after the append-and-drop) has global index
        # k - smoothing_radius + 1, so its outgoing transform sits
        # `smoothing_radius - 2` steps back from the newest, i.e. at position
        # `len(deque) - 1 - (smoothing_radius - 2)`.
        output_transform_idx = len(self.__transforms) - self.__smoothing_radius + 1

        return self.__apply_transformations(
            frame_transforms_smoothed, output_transform_idx
        )

    def __generate_transformations(self):
        """
        Generates previous-to-current transformation [dx, dy, da] for the
        latest frame in the queue and appends it to the bounded transforms
        deque (oldest is auto-dropped on overflow).
        """
        frame_gray = self._to_gray(self.__frame_queue[-1])
        self.__transforms.append(
            self._estimate_motion(
                self.__previous_gray, self.__previous_keypoints, frame_gray
            )
        )

        # refresh GFTT keypoints for next iteration
        self.__previous_keypoints = self._detect_keypoints(frame_gray)
        self.__previous_gray = frame_gray[:]

    def __box_filter_convolve(self, path, window_size):
        """
        Applies *normalized linear box filter* to path w.r.t averaging window.
        """
        path_padded = np.pad(path, (window_size, window_size), "median")
        path_smoothed = np.convolve(path_padded, self.__box_filter, mode="same")
        path_smoothed = path_smoothed[window_size:-window_size]
        assert path.shape == path_smoothed.shape
        return path_smoothed

    def __apply_transformations(self, frame_transforms_smoothed, transform_idx):
        """
        Pops the oldest frame from the queue and applies its smoothed
        transformation via the shared affine warp.
        """
        queue_frame = self.__frame_queue.popleft()

        # extracting Transformations w.r.t row in bounded smoothed window
        dx = frame_transforms_smoothed[transform_idx, 0]
        dy = frame_transforms_smoothed[transform_idx, 1]
        da = frame_transforms_smoothed[transform_idx, 2]

        return self._apply_warp(queue_frame, dx, dy, da)

    def clean(self) -> None:
        """
        Cleans ASWStabilizer resources.
        """
        if self.__frame_queue:
            logger.critical("Cleaning Resources...")
            self.__frame_queue.clear()
            self.__transforms.clear()
            # reset buildup flag so the instance can be reused cleanly
            self.__buildup_complete = False

class KalmanStabilizer(_StabilizerBase):
    """
    Zero-Latency Kalman-filter Video Stabilizer.

    Accumulates per-frame rigid motion `[dx, dy, da]` into a camera trajectory and
    smooths it online with a vectorized constant-position Kalman filter (one
    independent filter per axis). Each incoming frame is warped immediately from
    its measured trajectory onto the filtered one, so no frames are buffered and
    every `stabilize()` call returns a frame - suitable for live streams.
    """

    def __init__(
        self,
        process_noise: float = 4e-3,
        measurement_noise: float = 0.25,
        border_type: str = "black",
        border_size: int = 0,
        crop_n_zoom: bool = False,
        logging: bool = False,
    ):
        """
        Parameters:
            process_noise (float): process noise covariance `Q`. Larger = follows camera motion faster, less smoothing.
            measurement_noise (float): measurement noise covariance `R`. Larger = heavier smoothing, more lag on panning.
            border_type (str): changes the extended border type.
            border_size (int): enables and set the value for extended border size to reduce the black borders.
            crop_n_zoom (bool): enables cropping and zooming of frames(to original size) to reduce the black borders.
            logging (bool): enables/disables logging.
        """
        super().__init__(
            border_type=border_type,
            border_size=border_size,
            crop_n_zoom=crop_n_zoom,
            logging=logging,
        )
        # scalar or per-axis `[x, y, a]` covariances; broadcast against state
        self.__Q = _validate_noise("process_noise", process_noise)
        self.__R = _validate_noise("measurement_noise", measurement_noise)
        self._logging and logger.debug(
            "Setting Kalman noise: Q={}, R={}".format(self.__Q, self.__R)
        )
        self.__reset()

    def __reset(self):
        """
        Resets tracking + filter state to initial guesses.
        """
        self.__previous_gray = None  # previous gray frame
        self.__previous_keypoints = None  # previous GFTT keypoints
        self.__trajectory = np.zeros(3, dtype=np.float64)  # measured [x, y, a]
        self.__X = np.zeros(3, dtype=np.float64)  # posteriori state estimate
        self.__P = np.ones(3, dtype=np.float64)  # posteriori error covariance

    def stabilize(self, frame: np.ndarray) -> np.ndarray | None:
        """
        Takes an unstabilized video frame, and returns its stabilized version
        immediately (zero-latency).
        """
        if frame is None:
            return

        gray = self._to_gray(frame)

        if self.__previous_gray is None:
            # first frame: nothing to track against yet
            self._frame_height, self._frame_width = frame.shape[:2]
            if self._crop_n_zoom:
                self._frame_size = frame.shape[:2]
        else:
            # measurement: accumulated prev->cur motion
            self.__trajectory += self._estimate_motion(
                self.__previous_gray, self.__previous_keypoints, gray
            )
            # time update (prediction): X_ = X, P_ = P + Q
            p_prior = self.__P + self.__Q
            # measurement update (correction)
            gain = p_prior / (p_prior + self.__R)
            self.__X += gain * (self.__trajectory - self.__X)
            self.__P = (1.0 - gain) * p_prior

        self.__previous_gray = gray
        self.__previous_keypoints = self._detect_keypoints(gray)

        # warp current frame from measured onto smoothed trajectory
        dx, dy, da = self.__X - self.__trajectory
        return self._apply_warp(frame, dx, dy, da)

    def clean(self) -> None:
        """
        Cleans KalmanStabilizer resources.
        """
        super().clean()
        self.__reset()