Vision

Dense-Armor sees a video the way it reads a robot's joints: one frame at a time, in fixed memory, with no pretrained model.

The pages, in order:

1. Frames: every source gives the same Frame with the image and the timestamp.
2. Features: the oriented gradients, the intensity moments, the colour.
3. Flow: Lucas–Kanade in a cell, the pyramid, dense flow, the motion mask.
4. Reduce: the random projection, the incremental PCA, the whole chain.
5. Motion: the motion energy, the global shift, the scene changes, the sliding window.
6. Tracking: the moving object tracker, the video to signal adapters, the joints to video fusion.
7. Kinematics: landmarks, the cleaner, the joint angles, the registration.
8. Quality: the frame quality, the event frame, the false colour, the overlay, the benchmark.
9. Patches: patch memory and the RAD benchmark.

The ideas, and where they come from

Every technique on this page has been on someone's desk for decades, and each one answers a question that is easy to ask and surprisingly hard to answer well.

The histogram of oriented gradients was invented in 2005 by Dalal and Triggs, who wanted to find people walking in photographs. Their insight was that a person is recognisable not by the brightness of their clothes but by the directions of the edges on their body. It remains a standard way to describe an image cheaply.

The optical flow came from Lucas and Kanade in 1981. They wanted to know how much a patch of pixels moved from one frame to the next, and they realised that a single pixel is not enough to tell direction, but a small neighbourhood is. The global alternative, Horn and Schunck in the same year, adds a smoothness term over the whole image and solves one large system; Dense-Armor uses Lucas–Kanade because it is local and its cost per cell does not grow with the frame.

The Johnson and Lindenstrauss lemma is from 1984 and it is a statement about a question a pure mathematician would ask: how many dimensions do you really need to keep the distances between a set of points? The answer is surprisingly few, and a random projection is enough to keep them. This page uses that.

Principal component analysis is even older, from Pearson in 1901, and the incremental version used here is a form of Oja's rule from 1982 that learns the answer as data streams in. Krasulina, in 1969, arrived at almost the same rule from a different direction, and the modern analysis in Balsubramani, Dasgupta and Freund (2013) proves how quickly it converges. What all these ideas share is a willingness to throw information away: a frame becomes a hundred numbers, a hundred numbers become four, four become one anomaly score. The trick is doing it without losing what matters, and four decades of mathematics went into making that possible.

API reference

streams

Frame streams for the vision module.

A stream is an iterator of :class:Frame objects, one image per sample. A frame carries a numpy image (float in [0, 1] when grayscale) and a timestamp t in seconds. Three streams cover the three sources a robot or a developer has:

• :class:CameraStream reads from a camera through OpenCV. OpenCV is an optional dependency (extra [vision]) and imported lazily, so the rest of the vision pipeline works without it.
• :class:ImageFolderStream reads a folder of image files in sorted order (Pillow).
• :class:ArrayStream takes an array of frames already in memory, used by tests and by anyone with the frames at hand.

Every stream yields the same object, :class:Frame, so the rest of the pipeline is source-agnostic.

Frame dataclass

Frame(array: ndarray, t: float, dropped: int = 0)

One image frame with its timestamp.

Attributes:

Name	Type	Description
array	ndarray	H x W float array in [0, 1] for grayscale, or H x W x 3 for color; or H x W x C for a channel count different from 3.
t	float	timestamp in seconds.
dropped	int	number of frames lost between the previous yielded frame and this one. 0 means no loss. Only the streaming sources that can lose frames (a network camera, a reader slower than the source) set this; in-memory and file sources leave it at 0.

to_dict

to_dict() -> dict

Return a dict with the image and the timestamp.

Returns:

Type	Description
dict	A dict {"frame": array, "t": t}.

Source code in dense_armor/utility/vision/streams.py

def to_dict(self) -> dict:
    """Return a dict with the image and the timestamp.

    Returns:
        A dict ``{"frame": array, "t": t}``.
    """
    return {"frame": self.array, "t": self.t}

ArrayStream

ArrayStream(
    frames: ndarray,
    fps: float = 30.0,
    size: tuple[int, int] | None = None,
    gray: bool = True,
    dtype: str = "float32",
)

Iterate over a numpy array of frames.

Parameters:

Name	Type	Description	Default
frames	ndarray	array of shape (n, H, W) or (n, H, W, C).	required
fps	float	frame rate used to set t = i / fps, in Hz. Must be positive.	30.0
size	tuple[int, int] | None	(width, height) to resize each frame to, or None to keep the original size.	None
gray	bool	convert color frames to grayscale.	True
dtype	str	target dtype of the output array. "float32" (default) scales integer inputs to [0, 1] and leaves floats untouched; "uint8" keeps the 8-bit scale; "uint16" keeps the 16-bit scale (a 16-bit sensor readout loses no value). When dtype is "uint8" or "uint16" the output array carries that dtype and gray conversion is done on the integer values.	'float32'

Raises:

Type	Description
ValueError	if fps <= 0 or dtype is not supported.

Examples:

>>> import numpy as np
>>> frames = np.zeros((3, 4, 4), dtype=np.float32)
>>> s = ArrayStream(frames, fps=10.0)
>>> len(s)
3
>>> f = next(iter(s))
>>> f.array.shape, round(f.t, 3)
((4, 4), 0.0)
>>> frames16 = np.array([[[0, 1000], [60000, 65535]]], dtype=np.uint16)
>>> s16 = ArrayStream(frames16, dtype="uint16")
>>> next(iter(s16)).array.tolist()
[[0, 1000], [60000, 65535]]

Source code in dense_armor/utility/vision/streams.py

def __init__(
    self,
    frames: np.ndarray,
    fps: float = 30.0,
    size: tuple[int, int] | None = None,
    gray: bool = True,
    dtype: str = "float32",
) -> None:
    if fps <= 0:
        raise ValueError(f"fps must be > 0, got {fps}")
    if dtype not in self._DTYPES:
        raise ValueError(
            f"dtype must be one of {self._DTYPES}, got {dtype!r}"
        )
    self.frames = np.asarray(frames)
    self.fps = float(fps)
    self.size = size
    self.gray = gray
    self.dtype = dtype

ImageFolderStream

ImageFolderStream(
    path: str | Path | Sequence[str | Path],
    fps: float | None = None,
    size: tuple[int, int] | None = None,
    gray: bool = True,
)

Iterate over image files in a folder.

Parameters:

Name	Type	Description	Default
path	str | Path | Sequence[str | Path]	folder path, or a sequence of file paths.	required
fps	float | None	if given, t = i / fps; otherwise t = float(i).	None
size	tuple[int, int] | None	(width, height) to resize each frame to, or None.	None
gray	bool	convert to grayscale.	True

Raises:

Type	Description
ValueError	if the folder contains no image file.

Examples:

>>> import tempfile, pathlib
>>> import numpy as np
>>> from PIL import Image
>>> d = tempfile.mkdtemp()
>>> for i in range(2):
...     _ = Image.fromarray(
...         (np.zeros((4, 4), dtype=np.uint8) + i)
...     ).save(pathlib.Path(d) / f"{i:03d}.png")
>>> s = ImageFolderStream(d, fps=2.0)
>>> len(s)
2
>>> round(next(iter(s)).t, 3)
0.0

Source code in dense_armor/utility/vision/streams.py

def __init__(
    self,
    path: str | Path | Sequence[str | Path],
    fps: float | None = None,
    size: tuple[int, int] | None = None,
    gray: bool = True,
) -> None:
    if isinstance(path, (str, Path)):
        files = sorted(
            q for q in Path(path).iterdir() if q.suffix.lower() in self._SUFFIXES
        )
    else:
        files = [Path(q) for q in path]
    if not files:
        raise ValueError("ImageFolderStream: no image files found")
    self.files = files
    self.fps = fps
    self.size = size
    self.gray = gray

CameraStream

CameraStream(
    device: int = 0,
    fps: float = 30.0,
    size: tuple[int, int] | None = (160, 120),
    gray: bool = True,
    max_frames: int | None = None,
)

Iterate over camera frames through OpenCV.

OpenCV is imported lazily inside __init__. If it is not installed, the constructor raises :class:ImportError with a clear message. OpenCV delivers color frames in BGR order; they are turned to RGB before the grayscale conversion.

Parameters:

Name	Type	Description	Default
device	int	camera index (default 0).	0
fps	float	requested frame rate, in Hz, passed to the camera. Must be positive.	30.0
size	tuple[int, int] | None	(width, height) to resize each frame to, or None.	(160, 120)
gray	bool	convert to grayscale.	True
max_frames	int | None	stop after this many frames (None = infinite).	None

Raises:

Type	Description
ImportError	if OpenCV is not installed.
ValueError	if fps <= 0.

Source code in dense_armor/utility/vision/streams.py

def __init__(
    self,
    device: int = 0,
    fps: float = 30.0,
    size: tuple[int, int] | None = (160, 120),
    gray: bool = True,
    max_frames: int | None = None,
) -> None:
    try:
        import cv2
    except ImportError as exc:
        raise ImportError(
            "CameraStream requires OpenCV. Install it with "
            "`pip install dense-armor[vision]`."
        ) from exc
    if fps <= 0:
        raise ValueError(f"fps must be > 0, got {fps}")
    self._cv2 = cv2
    self.device = device
    self.fps = float(fps)
    self.size = size
    self.gray = gray
    self.max_frames = max_frames

VideoStream

VideoStream(
    fps: float = 30.0,
    size: tuple[int, int] | None = None,
    gray: bool = True,
    dtype: str = "float32",
    start: int = 0,
    stop: int | None = None,
    stride: int = 1,
)

Common interface of the frame sources.

Every video source in the library is an iterator of :class:Frame and exposes the same knobs: an optional resize, an optional grayscale conversion, a target dtype, a start and stop index, a stride to read one frame every k. The base class does not read anything; it declares the interface that :class:ArrayStream, :class:ImageFolderStream, :class:CameraStream, :class:VideoFileStream and :class:CubeStream implement.

Parameters:

Name	Type	Description	Default
fps	float	frame rate used when the source has no timestamp of its own. Must be positive.	30.0
size	tuple[int, int] | None	(width, height) to resize each frame to, or None.	None
gray	bool	convert color frames to grayscale.	True
dtype	str	target dtype of the output array; see :class:ArrayStream.	'float32'
start	int	first frame index to yield.	0
stop	int | None	last frame index to yield, exclusive, or None.	None
stride	int	yield one frame every stride, starting at start. Must be >= 1.	1

Raises:

Type	Description
ValueError	on a bad fps, start, stop or stride.

Source code in dense_armor/utility/vision/streams.py

def __init__(
    self,
    fps: float = 30.0,
    size: tuple[int, int] | None = None,
    gray: bool = True,
    dtype: str = "float32",
    start: int = 0,
    stop: int | None = None,
    stride: int = 1,
) -> None:
    if fps <= 0:
        raise ValueError(f"fps must be > 0, got {fps}")
    if stride < 1:
        raise ValueError(f"stride must be >= 1, got {stride}")
    if start < 0:
        raise ValueError(f"start must be >= 0, got {start}")
    if stop is not None and stop < start:
        raise ValueError(
            f"stop ({stop}) must be >= start ({start})"
        )
    self.fps = float(fps)
    self.size = size
    self.gray = gray
    self.dtype = dtype
    self.start = int(start)
    self.stop = stop
    self.stride = int(stride)

VideoFileStream

VideoFileStream(
    source: str | Path | Sequence[str | Path],
    fps: float = 30.0,
    size: tuple[int, int] | None = None,
    gray: bool = True,
    dtype: str = "float32",
    start: int = 0,
    stop: int | None = None,
    stride: int = 1,
    max_frames: int | None = None,
    rotate: str | int | float | bool = "auto",
    follow: bool = False,
    wait_s: float = 0.05,
    max_waits: int | None = 200,
)

Bases: VideoStream

Read a video from a file, an URL, a glob of files or a network stream.

OpenCV is imported lazily inside __init__, so a session without it can still import the module and use the numpy-only streams. The timestamp of each frame is the container's own clock, read from the decoder in milliseconds and converted to seconds; it is not frame_index / fps, so a video whose frame rate varies with time keeps the true time of every frame. A frame that the reader skips because the analysis is slower than the source is counted in :attr:Frame.dropped of the next yielded frame.

Parameters:

Name	Type	Description	Default
source	str | Path | Sequence[str | Path]	a file path, an URL (file://, http://, rtsp://), a glob (shots/*.mp4) or a sequence of paths. When a glob or a sequence matches more than one file, the frames of the files are concatenated in order and the time of each frame is the container time of the file it comes from plus the offset accumulated from the previous files.	required
fps	float	fallback rate used only if the container does not report its own frame rate; it is never used to set t.	30.0
size	tuple[int, int] | None	(width, height) to resize each frame to, or None.	None
gray	bool	convert color frames to grayscale.	True
dtype	str	target dtype; see :class:ArrayStream.	'float32'
start	int	first frame index.	0
stop	int | None	last frame index, exclusive.	None
stride	int	one frame every this many.	1
max_frames	int | None	stop after this many yielded frames (None = all).	None
rotate	str | int | float | bool	"auto" (default) reads the rotation from the container metadata when present; False ignores it; a number in {0, 90, 180, 270} forces that rotation in degrees counterclockwise.	'auto'
follow	bool	if True, on a source that keeps growing (a file being written by another process, a live stream) the reader waits and retries instead of stopping at the first read failure; a False value stops at the first EOF.	False
wait_s	float	seconds to sleep between retries when follow is True.	0.05
max_waits	int | None	maximum number of consecutive retries before giving up (None = forever).	200

Raises:

Type	Description
ImportError	if OpenCV is not installed.
FileNotFoundError	if source is a path that does not exist.
ValueError	on bad parameters.

Examples:

>>> import tempfile, pathlib, math
>>> import numpy as np
>>> import cv2
>>> d = pathlib.Path(tempfile.mkdtemp()) / "clip.mp4"
>>> w = cv2.VideoWriter(
...     str(d), cv2.VideoWriter_fourcc(*"mp4v"), 10.0, (8, 8)
... )
>>> for i in range(5):
...     frame = np.full((8, 8, 3), i * 50, dtype=np.uint8)
...     _ = w.write(frame)
>>> w.release()
>>> s = VideoFileStream(str(d))
>>> frames = list(s)
>>> len(frames)
5
>>> round(frames[0].t, 3)
0.0
>>> round(frames[-1].t, 3)
0.4

Source code in dense_armor/utility/vision/streams.py

def __init__(
    self,
    source: str | Path | Sequence[str | Path],
    fps: float = 30.0,
    size: tuple[int, int] | None = None,
    gray: bool = True,
    dtype: str = "float32",
    start: int = 0,
    stop: int | None = None,
    stride: int = 1,
    max_frames: int | None = None,
    rotate: str | int | float | bool = "auto",
    follow: bool = False,
    wait_s: float = 0.05,
    max_waits: int | None = 200,
) -> None:
    super().__init__(fps, size, gray, dtype, start, stop, stride)
    try:
        import cv2
    except ImportError as exc:
        raise ImportError(
            "VideoFileStream requires OpenCV. Install it with "
            "`pip install dense-armor[vision]`."
        ) from exc
    self._cv2 = cv2
    self.source = source
    self.max_frames = max_frames
    self.rotate = rotate
    self.follow = follow
    self.wait_s = float(wait_s)
    self.max_waits = max_waits
    self._paths = self._resolve(source)

CubeStream

CubeStream(
    cube: ndarray,
    fps: float = 30.0,
    size: tuple[int, int] | None = None,
    gray: bool = False,
    dtype: str = "float32",
    start: int = 0,
    stop: int | None = None,
    stride: int = 1,
    max_frames: int | None = None,
)

Bases: VideoStream

Iterate over a multi-channel cube (T, H, W, C).

A cube is the format of a thermal sequence (one channel), a multispectral or hyperspectral stack (many channels), a depth map (one channel of floats) and, in general, of any video whose channel count is not three. The time of the frame i is i / fps; the cube carries no clock of its own.

Parameters:

Name	Type	Description	Default
cube	ndarray	array of shape (T, H, W) or (T, H, W, C).	required
fps	float	frame rate; sets t = i / fps.	30.0
size	tuple[int, int] | None	(width, height) to resize each frame to, or None.	None
gray	bool	reduce a 3-channel frame to grayscale; a cube with a channel count different from 3 is left untouched.	False
dtype	str	target dtype; "uint16" keeps a 16-bit thermal readout without loss.	'float32'
start	int	first frame index.	0
stop	int | None	last frame index, exclusive.	None
stride	int	one frame every this many.	1
max_frames	int | None	stop after this many yielded frames.	None

Raises:

Type	Description
ValueError	on a cube whose first dimension is not the time, or on bad parameters.

Examples:

>>> import numpy as np
>>> cube = np.arange(2 * 3 * 4, dtype=np.uint16).reshape(2, 3, 4)
>>> s = CubeStream(cube, fps=2.0, dtype="uint16", gray=False)
>>> frames = list(s)
>>> len(frames)
2
>>> frames[0].array.dtype
dtype('uint16')
>>> round(frames[1].t, 3)
0.5

Source code in dense_armor/utility/vision/streams.py

def __init__(
    self,
    cube: np.ndarray,
    fps: float = 30.0,
    size: tuple[int, int] | None = None,
    gray: bool = False,
    dtype: str = "float32",
    start: int = 0,
    stop: int | None = None,
    stride: int = 1,
    max_frames: int | None = None,
) -> None:
    super().__init__(fps, size, gray, dtype, start, stop, stride)
    a = np.asarray(cube)
    if a.ndim not in (3, 4):
        raise ValueError(
            f"cube must have shape (T, H, W) or (T, H, W, C), "
            f"got {a.shape}"
        )
    self.cube = a
    self.max_frames = max_frames

features

Per-frame features: oriented gradients, moments, and short-range flow.

A single frame on its own carries little motion information, but three families of descriptors are cheap, local, and enough to tell a moving square from a still one:

• a histogram of oriented gradients (HOG) on a fixed grid of cells. The image is divided into cells rectangles; inside each cell the gradient orientation of every pixel votes into one of n_bins angular bins, weighted by the gradient magnitude. The cell histogram is normalised to unit L2 norm, so a change in brightness does not change the descriptor. The construction is the one in Huang and Huang (2017), section 2.
• three global moments of the intensity: mean, standard deviation (the contrast), and the centroid of the intensity, which is where the "mass" of the image is. With gray=False the same two moments are computed for each colour channel separately, and three extra pairs of keys appear.
• short-range optical flow against the previous frame, on a coarser grid: in every flow cell the classic Lucas and Kanade (1981) least squares is solved, as written in Ziani (2025), equations 4 and 5. With flow_levels > 1 the flow is estimated coarse-to-fine on a pyramid (each level a 2x2 average of the one below; Ziani uses Gaussian pyramids) with bilinear interpolation between levels, the scheme of Ziani (2025), section 3.2, applied here to Lucas-Kanade.

The first frame has no previous, so its flow block is zero. A frame that is empty, has a wrong shape, or contains a non-finite value is skipped and counted in n_missing.

FrameFeatures

FrameFeatures(
    n_bins: int = 9,
    cells: tuple[int, int] = (3, 4),
    flow_cells: tuple[int, int] = (2, 3),
    flow_levels: int = 1,
    gray: bool = True,
    eps: float = 1e-08,
)

Bases: Transformer

Turn one frame into a fixed-length dict of float features.

Keys of the returned dict (all float):

• hog_{cy}_{cx}_{b}: the L2-normalised HOG histogram of cell (cy, cx), one value per angular bin b. There are cells[0] * cells[1] cells and n_bins bins per cell.
• mean, contrast: the mean and the standard deviation of the luminance.
• centroid_x, centroid_y: the coordinates of the luminance centroid, normalised in [0, 1].
• flow_u_{fy}_{fx}, flow_v_{fy}_{fx}: the mean optical flow components inside the coarse flow cell (fy, fx).
• flow_mag_{fy}_{fx}: the magnitude of the same mean flow.
• if gray=False, six extra keys mean_R, mean_G, mean_B, contrast_R, contrast_G, contrast_B for the per-channel moments.

Parameters:

Name	Type	Description	Default
n_bins	int	number of angular bins in each cell. Default 9, the value used in Huang and Huang (2017).	9
cells	tuple[int, int]	(rows, cols) of the HOG grid. Default (3, 4).	(3, 4)
flow_cells	tuple[int, int]	(rows, cols) of the flow grid. Default (2, 3).	(2, 3)
flow_levels	int	number of pyramid levels for the optical flow. Default 1: a single level, the same behaviour as without a pyramid. n > 1 builds a pyramid of 2x2 averages with n levels, solves Lucas-Kanade at the coarsest level and refines the flow down to the finest level with bilinear interpolation, the coarse-to-fine scheme of Ziani (2025), section 3.2.	1
gray	bool	True (default) takes 2D images and computes the moments on the luminance. False takes (H, W, 3) images, adds the per-channel moments, and keeps the HOG and the flow on the luminance, so the other keys do not change.	True
eps	float	numerical guard for the normalisation and the least squares.	1e-08

Raises:

Type	Description
ValueError	if n_bins < 1, if a grid size is not positive, or if flow_levels < 1.

Examples:

>>> import numpy as np
>>> from dense_armor.utility.vision.features import FrameFeatures
>>> ff = FrameFeatures()
>>> frame = np.zeros((6, 8), dtype=np.float32)
>>> out = ff.transform_one(frame)
>>> out["mean"], out["contrast"]
(0.0, 0.0)

Source code in dense_armor/utility/vision/features.py

def __init__(
    self,
    n_bins: int = 9,
    cells: tuple[int, int] = (3, 4),
    flow_cells: tuple[int, int] = (2, 3),
    flow_levels: int = 1,
    gray: bool = True,
    eps: float = 1e-8,
) -> None:
    if n_bins < 1:
        raise ValueError(f"n_bins must be >= 1, got {n_bins}")
    if cells[0] < 1 or cells[1] < 1:
        raise ValueError(f"cells must be positive, got {cells}")
    if flow_cells[0] < 1 or flow_cells[1] < 1:
        raise ValueError(f"flow_cells must be positive, got {flow_cells}")
    if flow_levels < 1:
        raise ValueError(f"flow_levels must be >= 1, got {flow_levels}")
    self.n_bins = n_bins
    self.cells = cells
    self.flow_cells = flow_cells
    self.flow_levels = flow_levels
    self.gray = gray
    self.eps = eps
    self._prev: np.ndarray | None = None
    self.n_missing_ = 0

learn_one

learn_one(
    x: Any, y: Any = None, t: float | None = None
) -> FrameFeatures

Store the current frame so the next one can compute flow.

Source code in dense_armor/utility/vision/features.py

def learn_one(
    self, x: Any, y: Any = None, t: float | None = None
) -> "FrameFeatures":
    """Store the current frame so the next one can compute flow."""
    self._time_step(t)
    img = _as_array(x)
    if not self._valid(img):
        self.n_missing_ += 1
        return self
    assert img is not None
    self._prev = _to_gray(img).copy()
    return self

transform_one

transform_one(
    x: Any, t: float | None = None
) -> dict[str, float]

Return the feature dict of one frame.

Parameters:

Name	Type	Description	Default
x	Any	a Frame, a dict with a frame key, or a 2D array (or 3D with gray=False).	required
t	float | None	unused, kept for the interface.	None

Returns:

Type	Description
dict[str, float]	A flat dict of float features.

Source code in dense_armor/utility/vision/features.py

def transform_one(self, x: Any, t: float | None = None) -> dict[str, float]:
    """Return the feature dict of one frame.

    Args:
        x: a Frame, a dict with a ``frame`` key, or a 2D array
            (or 3D with ``gray=False``).
        t: unused, kept for the interface.

    Returns:
        A flat dict of float features.
    """
    img = _as_array(x)
    if not self._valid(img):
        self.n_missing_ += 1
        return self._empty()
    assert img is not None
    lum = _to_gray(img)
    out: dict[str, float] = {}
    out.update(self._hog(lum))
    out.update(self._moments(lum))
    if not self.gray:
        out.update(self._color_moments(img if img.ndim == 3 else np.stack([lum] * 3, axis=-1)))
    if self._prev is None or self._prev.shape != lum.shape:
        out.update(self._flow(lum, lum))
    else:
        out.update(self._flow(lum, self._prev))
    return out

reduce

Dimensionality reduction for vision features.

Two transformers, both in the spirit of the library: they read a dict of floats and return a smaller dict of floats, one sample at a time.

• :class:RandomProjection draws a fixed random matrix once, in the constructor, and multiplies every incoming vector by it. The Johnson and Lindenstrauss (1984) lemma guarantees that the pairwise distances are preserved up to a factor (1 + eps) with high probability, provided the output dimension k is large enough. The version used here follows Ghojogh et al. (2021), section 2.2: the elements of the matrix are i.i.d. Gaussian with mean zero and variance 1 / k, so the expected squared norm of a projected vector equals the squared norm of the original one. The distortion bound is equation 21 of that paper, with delta = 2 exp(-(eps^2 - eps^3) k / 4).

• :class:IncrementalPCA learns the top k directions of the data as they stream in, with Oja's rule generalised to k components. The update follows the one-component form in Balsubramani, Dasgupta, Freund (2013), equation (1),

  v_n = v_{n-1} + gamma_n (X_n X_n^T - (v_{n-1}^T X_n X_n^T v_{n-1}) I_d) v_{n-1},

For k > 1 the matrix V (d x k) is updated with the same step, V + gamma_n z (V^T z)^T, and re-orthonormalised with a QR decomposition after each sample: a stochastic subspace (power) iteration. The QR makes the correction term of Oja's rule redundant, so the multi-component case is this iteration, not a separate rule from the paper. The running mean is removed from the input first. The step size is gamma_n = c / n with c chosen by the user or by the default rule. The paper analyses the rate of convergence: the potential 1 - (v . v*)^2 / |v|^2 goes to zero as O(1 / n) when c is at least 1 / (2 (lambda_1 - lambda_2)).

RandomProjection

RandomProjection(
    k: int,
    seed: int | None = None,
    input_dim: int | None = None,
)

Bases: Transformer

Fixed Gaussian random projection to k dimensions.

Parameters:

Name	Type	Description	Default
k	int	output dimension. Must be positive.	required
seed	int | None	seed of the random generator, so a run is reproducible.	None
input_dim	int | None	number of input features. If None (default), the matrix is built on the first call, using the keys of the first dict in sorted order.	None

Raises:

Type	Description
ValueError	if k <= 0.

Examples:

>>> import numpy as np
>>> from dense_armor.utility.vision.reduce import RandomProjection
>>> rp = RandomProjection(k=3, seed=0)
>>> x = {f"f{i}": float(i) for i in range(6)}
>>> out = rp.transform_one(x)
>>> len(out)
3
>>> sorted(out)
['p0', 'p1', 'p2']

Source code in dense_armor/utility/vision/reduce.py

def __init__(
    self,
    k: int,
    seed: int | None = None,
    input_dim: int | None = None,
) -> None:
    if k <= 0:
        raise ValueError(f"k must be > 0, got {k}")
    self.k = k
    self.seed = seed
    self.input_dim = input_dim
    self._rng = np.random.default_rng(seed)
    self._W: np.ndarray | None = None
    self._keys: list[str] | None = None
    self.n_missing_ = 0

learn_one

learn_one(
    x: dict, y: Any = None, t: float | None = None
) -> RandomProjection

No learning: the projection is fixed at construction.

Source code in dense_armor/utility/vision/reduce.py

def learn_one(
    self, x: dict, y: Any = None, t: float | None = None
) -> "RandomProjection":
    """No learning: the projection is fixed at construction."""
    self._time_step(t)
    if self._W is None:
        self._ensure(x)
    return self

transform_one

transform_one(
    x: dict, t: float | None = None
) -> dict[str, float]

Project one feature dict onto the random matrix.

Parameters:

Name	Type	Description	Default
x	dict	dict of float features.	required
t	float | None	unused, kept for the interface.	None

Returns:

Type	Description
dict[str, float]	A dict with keys p0, ..., p{k-1}.

Source code in dense_armor/utility/vision/reduce.py

def transform_one(self, x: dict, t: float | None = None) -> dict[str, float]:
    """Project one feature dict onto the random matrix.

    Args:
        x: dict of float features.
        t: unused, kept for the interface.

    Returns:
        A dict with keys ``p0, ..., p{k-1}``.
    """
    if self._W is None:
        self._ensure(x)
    try:
        v = self._to_vec(x)
    except (KeyError, ValueError, TypeError):
        self.n_missing_ += 1
        return {f"p{j}": 0.0 for j in range(self.k)}
    if not np.isfinite(v).all():
        self.n_missing_ += 1
        return {f"p{j}": 0.0 for j in range(self.k)}
    W = self._W
    assert W is not None
    proj = W.T @ v
    return {f"p{j}": float(proj[j]) for j in range(self.k)}

IncrementalPCA

IncrementalPCA(
    k: int, lr: float | None = None, eps: float = 1e-09
)

Bases: Transformer

Online PCA: Oja's rule for one component, stochastic subspace iteration with QR for several.

Parameters:

Name	Type	Description	Default
k	int	number of components.	required
lr	float | None	if given, the constant c in gamma_n = c / n. If None (default), uses c = 1.0. See the class docstring for the rule the paper suggests.	None
eps	float	numerical guard.	1e-09

Raises:

Type	Description
ValueError	if k <= 0 or lr <= 0.

Examples:

>>> import numpy as np
>>> from dense_armor.utility.vision.reduce import IncrementalPCA
>>> rng = np.random.default_rng(0)
>>> pca = IncrementalPCA(k=1)
>>> for _ in range(200):
...     _ = pca.learn_one({"a": float(rng.normal(1, 0.1)),
...                        "b": float(rng.normal(0, 1))})
>>> out = pca.transform_one({"a": 1.0, "b": 0.0})
>>> len(out)
1

Source code in dense_armor/utility/vision/reduce.py

def __init__(
    self,
    k: int,
    lr: float | None = None,
    eps: float = 1e-9,
) -> None:
    if k <= 0:
        raise ValueError(f"k must be > 0, got {k}")
    if lr is not None and lr <= 0:
        raise ValueError(f"lr must be > 0, got {lr}")
    self.k = k
    self.lr = lr
    self.eps = eps
    self._V: np.ndarray | None = None
    self._mean: np.ndarray | None = None
    self._keys: list[str] | None = None
    self._n = 0
    self.n_missing_ = 0

learn_one

learn_one(
    x: dict, y: Any = None, t: float | None = None
) -> IncrementalPCA

Update the components with one sample.

Parameters:

Name	Type	Description	Default
x	dict	dict of float features.	required
y	Any	unused.	None
t	float | None	unused.	None

Returns:

Type	Description
IncrementalPCA	self.

Source code in dense_armor/utility/vision/reduce.py

def learn_one(
    self, x: dict, y: Any = None, t: float | None = None
) -> "IncrementalPCA":
    """Update the components with one sample.

    Args:
        x: dict of float features.
        y: unused.
        t: unused.

    Returns:
        ``self``.
    """
    self._time_step(t)
    if self._V is None:
        self._ensure(x)
    try:
        v = self._to_vec(x)
    except (KeyError, ValueError, TypeError):
        self.n_missing_ += 1
        return self
    if not np.isfinite(v).all():
        self.n_missing_ += 1
        return self
    self._n += 1
    n = self._n
    mean = self._mean
    V = self._V
    assert mean is not None and V is not None
    mean = mean + (v - mean) / n
    self._mean = mean
    z = v - mean
    c = 1.0 if self.lr is None else float(self.lr)
    gamma = c / n
    a = V.T @ z
    update = np.outer(z, a)
    V_new = V + gamma * update
    q, _ = np.linalg.qr(V_new)
    self._V = q[:, : self.k]
    return self

transform_one

transform_one(
    x: dict, t: float | None = None
) -> dict[str, float]

Project one feature dict onto the current components.

Parameters:

Name	Type	Description	Default
x	dict	dict of float features.	required
t	float | None	unused, kept for the interface.	None

Returns:

Type	Description
dict[str, float]	A dict with keys pc0, ..., pc{k-1}.

Source code in dense_armor/utility/vision/reduce.py

def transform_one(self, x: dict, t: float | None = None) -> dict[str, float]:
    """Project one feature dict onto the current components.

    Args:
        x: dict of float features.
        t: unused, kept for the interface.

    Returns:
        A dict with keys ``pc0, ..., pc{k-1}``.
    """
    if self._V is None:
        self._ensure(x)
    try:
        v = self._to_vec(x)
    except (KeyError, ValueError, TypeError):
        self.n_missing_ += 1
        return {f"pc{j}": 0.0 for j in range(self.k)}
    if not np.isfinite(v).all():
        self.n_missing_ += 1
        return {f"pc{j}": 0.0 for j in range(self.k)}
    mean = self._mean
    V = self._V
    assert mean is not None and V is not None
    z = v - mean
    proj = V.T @ z
    return {f"pc{j}": float(proj[j]) for j in range(self.k)}

---

Details

• Everything on this page is computed by the library in numpy; the camera driver (OpenCV) is the only optional dependency, extra [vision], imported only by CameraStream.
• FrameFeatures keys: hog_{cy}_{cx}_{b} (default 3×4 cells, 9 bins), mean, contrast, centroid_x, centroid_y, flow_u_*, flow_v_*, flow_mag_* (default 2×3 flow cells). The first frame has zero flow; an empty or non-finite frame is skipped and counted in n_missing.
• RandomProjection uses a dense Gaussian matrix with variance 1 / k (the matrix of equation 21). Sparse projections (Achlioptas 2003, Li et al. 2006) are named in the same survey but not defined there; they are not implemented yet.
• IncrementalPCA with several components updates all of them with the same step and re-orthonormalises with a QR decomposition after each sample (a stochastic subspace iteration); the paper analyses the one-component rule.
• Pipelines built with | train every step: each intermediate step first transforms the sample, then learns from it, so FrameFeatures keeps the previous frame for the flow.
• flow_levels: the pyramid is built by 2×2 averaging; Ziani (2025, section 3.2) builds Gaussian pyramids and uses the coarse-to-fine scheme for Horn–Schunck; the library applies the same scheme to Lucas–Kanade. flow_levels=1 gives exactly the single-level flow.
• gray=False: keys mean_R, mean_G, mean_B, contrast_R, contrast_G, contrast_B are always present; for a grey input they repeat the luminance moments.
• Real robot images, RAD (Chang et al. 2024; MIT licence): 13 objects photographed by a Franka arm from many viewpoints, normal and defective. benchmarks/vision_rad.py learns FrameFeatures | IncrementalPCA(k=8) | OnlineRobustMahalanobis on the normal training views and scores each test image (160×120, grey, same parameters for every category, no tuning). Image-level AUROC: binderclip 0.531, bowl 0.461, box 0.428, can 0.599, charger 0.508, cup1 0.443, cup2 0.488, gluebottle 0.492, phonecase 0.458, rubberduck 0.552, spoon 0.569, spraybottle 0.464, tennisball 0.473; mean 0.497, chance level. The paper reports 0.833 for PatchCore and 0.803 for EfficientAD (p. 10), both built on pretrained networks. Global features (gradients, moments) notice changes of the scene; they do not separate a small defect from a change of viewpoint. Each category has only 9–27 normal test images, so the per-category values are uncertain; the mean is the number to read. Latency 4.5 ms p50 per image (score only).
• Sources:
• Chang, X. et al. (2024). RAD: A Realistic Multi-View Benchmark for Pose-Agnostic Anomaly Detection. arXiv:2410.00713. The robot images of the Details.
• Huang, C., Huang, J. (2017). A fast HOG descriptor using lookup table and integral image. arXiv:1703.06256. The HOG construction used in section 2.
• Dalal, N., Triggs, B. (2005). Histograms of oriented gradients for human detection. In IEEE Conference on Computer Vision and Pattern Recognition. The original HOG paper, cited in the opening.
• Ziani, H. (2025). Investigating optical flow computation: from local methods to a multiresolution Horn–Schunck implementation. arXiv:2511.16535. The Lucas–Kanade least squares of section 4, the bilinear interpolation of section 4b, and the coarse-to-fine scheme.
• Lucas, B. D., Kanade, T. (1981). An iterative image registration technique with an application to stereo vision. In Proceedings of the 7th International Joint Conference on Artificial Intelligence. The original Lucas–Kanade algorithm.
• Horn, B. K., Schunck, B. G. (1981). Determining optical flow. Artificial Intelligence 17(1-3), 185-203. The global method mentioned at the end of section 4.
• Johnson, W. B., Lindenstrauss, J. (1984). Extensions of Lipschitz mappings into a Hilbert space. Contemporary Mathematics 26. The distance-preservation lemma of section 5.
• Achlioptas, D. (2003). Database-friendly random projections: Johnson–Lindenstrauss with binary coins. Journal of Computer and System Sciences 66(4), 671-687. The sparse projection matrix named in the Details.
• Li, P., Hastie, T. J., Church, K. W. (2006). Very sparse random projections. In Proceedings of the 12th ACM SIGKDD International Conference on Knowledge Discovery and Data Mining. The sparser alternative named in the Details.
• Ghojogh, B., Ghodsi, A., Karray, F., Crowley, M. (2021). Johnson–Lindenstrauss lemma, linear and nonlinear random projections, random Fourier features, and random kitchen sinks: tutorial and survey. arXiv:2108.04172. The bound of section 5, equation 21.
• Pearson, K. (1901). On lines and planes of closest fit to systems of points in space. Philosophical Magazine 2(11), 559-572. The origin of principal component analysis.
• Krasulina, T. P. (1969). A method of stochastic approximation for the determination of the least eigenvalue of a symmetrical matrix. USSR Computational Mathematics and Mathematical Physics 9(6), 189-195. One of the two classical one-component rules.
• Oja, E. (1982). Simplified neuron model as a principal component analyzer. Journal of Mathematical Biology 15(3), 267-273. The other classical one-component rule.
• Balsubramani, A., Dasgupta, S., Freund, Y. (2013). The fast convergence of incremental PCA. In Advances in Neural Information Processing Systems (NIPS), arXiv:1501.03796. The convergence rate of section 6, equation 1.
• Guillot, A. et al. (2026). Online robust covariance and outlier detection. arXiv:2601.03957. The robust Mahalanobis detector used in the last step of section 7.
• Cardot, H., Cénac, P., Zitt, P.-A. (2013). Efficient and fast estimation of the geometric median in Hilbert spaces with an averaged stochastic gradient algorithm. Bernoulli 19(1), 18-43. The geometric median update of section 7.
• Cardot, H., Godichon-Baggioni, A. (2017). Fast estimation of the median covariation matrix with application to online robust principal components analysis. TEST 26(3), 461-480. The median covariation matrix update of section 7.