Stream

Streaming video reader for MGT-python.

:class:MgVideoReader is a context-manager-based iterator that yields video frames lazily using FFmpeg pipes. This avoids loading an entire video into RAM, making it suitable for long recordings.

Examples

from musicalgestures._stream import MgVideoReader with MgVideoReader("dancer.avi") as reader: ... for i, (frame, ts) in enumerate(reader): ... # frame: np.ndarray, shape (H, W, 3), dtype uint8 ... # ts: float, timestamp in seconds ... if i >= 5: ... break

MgVideoReader

MgVideoReader(filename, start=0.0, end=None, grayscale=False, scale=1.0, batch_size=1)

Context-manager that streams frames from a video file via FFmpeg.

Parameters:

Name	Type	Description	Default
filename	str | Path	Path to the video file to read.	required
start	float	Start time in seconds. Defaults to 0.	0.0
end	float | None	End time in seconds. Defaults to None (read to end of file).	None
grayscale	bool	If True, convert frames to grayscale before yielding. Default: False.	False
scale	float	Downscale factor (e.g. 0.5 → half resolution). Default: 1.0.	1.0
batch_size	int	Number of frames to read per FFmpeg read call. Larger values reduce subprocess overhead at the cost of more memory. Default: 1.	1

Yields:

Name	Type	Description
frame	ndarray	Video frame as a NumPy array, shape (H, W, 3) (BGR) or (H, W) if grayscale=True.
timestamp	float	Approximate frame timestamp in seconds.

Examples:

>>> import numpy as np
>>> # Collect every frame as a numpy array:
>>> frames = []
>>> with MgVideoReader("dancer.avi") as reader:
...     for frame, ts in reader:
...         frames.append(frame)
>>> arr = np.stack(frames)  # shape (N, H, W, 3)

Source code in musicalgestures/_stream.py

def __init__(
    self,
    filename: str | Path,
    start: float = 0.0,
    end: float | None = None,
    grayscale: bool = False,
    scale: float = 1.0,
    batch_size: int = 1,
) -> None:
    self.filename = Path(filename)
    if not self.filename.exists():
        raise FileNotFoundError(f"Video file not found: {self.filename}")
    self.start = float(start)
    self.end = end
    self.grayscale = grayscale
    self.scale = float(scale)
    self.batch_size = int(batch_size)

    self._width: int = 0
    self._height: int = 0
    self._fps: float = 0.0
    self._process: subprocess.Popen | None = None
    self._frame_index: int = 0

width property

width

Frame width in pixels (after optional scaling).

height property

height

Frame height in pixels (after optional scaling).

fps property

fps

Frames per second of the source video.

__iter__

__iter__()

Yield (frame, timestamp) pairs.

Source code in musicalgestures/_stream.py

def __iter__(self) -> Generator[tuple[np.ndarray, float], None, None]:
    """Yield ``(frame, timestamp)`` pairs."""
    if self._process is None:
        raise RuntimeError(
            "MgVideoReader must be used as a context manager: "
            "'with MgVideoReader(path) as reader:'"
        )
    channels = 1 if self.grayscale else 3
    frame_bytes = self._height * self._width * channels
    fps = self._fps

    while True:
        raw = self._process.stdout.read(frame_bytes)
        if len(raw) < frame_bytes:
            break
        frame = np.frombuffer(raw, dtype=np.uint8)
        if self.grayscale:
            frame = frame.reshape((self._height, self._width))
        else:
            frame = frame.reshape((self._height, self._width, 3))
        ts = self.start + self._frame_index / fps
        yield frame, ts
        self._frame_index += 1