Skip to content

API reference

Auto-generated from the source docstrings. The headline entry point is StreamSession; see Python API for a task-oriented walkthrough.

Control core — player.session

streamcatcher.player.session

Headless stream control core, shared by the GUI player and the HTTP API.

:class:StreamSession owns the OpenCV capture and the optional 360 viewport and exposes programmatic controls — open, read a frame, render the current viewport, look around (pan/tilt/zoom), inspect state, close — with no window and no keyboard loop. The GUI player (:class:~streamcatcher.player.opencv_player.OpenCvPlayer) and, later, the FastAPI server both drive the same session, so the controls live in exactly one place.

cv2 is imported lazily inside :meth:StreamSession.open, so importing this module never requires OpenCV; tests inject a fake cv2.

StreamOpenError

Bases: RuntimeError

Raised when OpenCV cannot open the stream URL.

SnapshotError

Bases: RuntimeError

Raised when a snapshot frame can't be captured or written to disk.

ViewState dataclass

ViewState(
    projection, yaw_deg=None, pitch_deg=None, hfov_deg=None
)

A snapshot of the viewport orientation (None fields when not 360).

StreamSession

StreamSession(url, projection=Projection.FLAT)

Own a live capture and an optional 360 viewport, with no window.

Source code in src/streamcatcher/player/session.py
def __init__(
    self,
    url: str,
    projection: Projection = Projection.FLAT,
) -> None:
    self._url = url  # secret: embeds credentials, so it is never logged
    self._projection = Projection(projection)
    self._cap = None
    self._cv2 = None
    self._view = _build_view(self._projection)
    self._maps = None  # cached (map_x, map_y); rebuilt when the view moves

has_viewport property

has_viewport

Whether this session reprojects a look-around viewport (equirect).

open

open()

Open the stream capture with low-latency options. Raises on failure.

Source code in src/streamcatcher/player/session.py
def open(self) -> None:
    """Open the stream capture with low-latency options. Raises on failure."""
    cv2 = _load_cv2()
    os.environ.setdefault("OPENCV_FFMPEG_CAPTURE_OPTIONS", _FFMPEG_CAPTURE_OPTIONS)
    log.info("Opening live stream with OpenCV.")
    cap = cv2.VideoCapture(self._url, cv2.CAP_FFMPEG)
    if not cap.isOpened():
        cap.release()
        raise StreamOpenError(
            "Could not open the stream. Check the URL, network, and credentials."
        )
    # Keep only the freshest frame buffered so a slow render loop can't fall
    # progressively behind real time. Best-effort: honored by some backends,
    # ignored by others (FFmpeg leans on ``fflags;nobuffer`` above instead).
    cap.set(cv2.CAP_PROP_BUFFERSIZE, 1)
    self._cv2 = cv2
    self._cap = cap

close

close()

Release the capture. Safe to call more than once.

Source code in src/streamcatcher/player/session.py
def close(self) -> None:
    """Release the capture. Safe to call more than once."""
    if self._cap is not None:
        self._cap.release()
        self._cap = None
    log.info("Stream session closed.")

reconnect

reconnect()

Tear down and re-open the capture, keeping the viewport orientation.

Returns True on success, or False if the stream still can't be opened (so the caller can back off and retry). The viewport _view (yaw/pitch/hfov) is left untouched, so a look-around orientation survives a reconnect; cached remap tables are dropped in case the stream returns at a different resolution.

Source code in src/streamcatcher/player/session.py
def reconnect(self) -> bool:
    """Tear down and re-open the capture, keeping the viewport orientation.

    Returns ``True`` on success, or ``False`` if the stream still can't be
    opened (so the caller can back off and retry). The viewport ``_view``
    (yaw/pitch/hfov) is left untouched, so a look-around orientation
    survives a reconnect; cached remap tables are dropped in case the
    stream returns at a different resolution.
    """
    self.close()
    self._maps = None
    try:
        self.open()
    except StreamOpenError:
        return False
    return True

is_open

is_open()

Whether a stream capture is currently open.

Source code in src/streamcatcher/player/session.py
def is_open(self) -> bool:
    """Whether a stream capture is currently open."""
    return self._cap is not None and self._cap.isOpened()

read_frame

read_frame()

Read the next raw frame; returns the frame, or None when it ends.

Source code in src/streamcatcher/player/session.py
def read_frame(self):
    """Read the next raw frame; returns the frame, or ``None`` when it ends."""
    if self._cap is None:
        raise RuntimeError("Session is not open.")
    ok, frame = self._cap.read()
    return frame if ok else None

render

render(frame)

Return the viewport for frame — reprojected in 360, else unchanged.

Source code in src/streamcatcher/player/session.py
def render(self, frame):
    """Return the viewport for ``frame`` — reprojected in 360, else unchanged."""
    if self._view is None:
        return frame
    if self._maps is None:
        height, width = frame.shape[:2]
        self._maps = self._view.build_maps(width, height)
    map_x, map_y = self._maps
    return self._cv2.remap(frame, map_x, map_y, self._cv2.INTER_LINEAR)

grab_view

grab_view()

Read and render the next viewport frame; None when the stream ends.

Source code in src/streamcatcher/player/session.py
def grab_view(self):
    """Read and render the next viewport frame; ``None`` when the stream ends."""
    frame = self.read_frame()
    return None if frame is None else self.render(frame)

look

look(pan=0.0, tilt=0.0, zoom=0.0)

Apply pan/tilt/zoom degree deltas to the viewport (a no-op when flat).

zoom is a horizontal-FOV delta: negative narrows the view (zooms in).

Source code in src/streamcatcher/player/session.py
def look(self, pan: float = 0.0, tilt: float = 0.0, zoom: float = 0.0) -> None:
    """Apply pan/tilt/zoom degree deltas to the viewport (a no-op when flat).

    ``zoom`` is a horizontal-FOV delta: negative narrows the view (zooms in).
    """
    if self._view is None or not (pan or tilt or zoom):
        return
    if pan:
        self._view.pan(pan)
    if tilt:
        self._view.tilt(tilt)
    if zoom:
        self._view.zoom(zoom)
    self._maps = None  # view moved — rebuild maps on the next render

state

state()

Current projection and (in 360) the viewport orientation.

Source code in src/streamcatcher/player/session.py
def state(self) -> ViewState:
    """Current projection and (in 360) the viewport orientation."""
    if self._view is None:
        return ViewState(projection=self._projection.value)
    return ViewState(
        projection=self._projection.value,
        yaw_deg=self._view.yaw_deg,
        pitch_deg=self._view.pitch_deg,
        hfov_deg=self._view.hfov_deg,
    )

snapshot

snapshot(path)

Grab the current viewport and save it to path as an image file.

The saved image is the reprojected look-around viewport in 360 modes (what the viewer sees), or the raw frame when flat. Reads until a frame arrives — a just-opened decoder can return empty first — then writes it, raising :class:SnapshotError if no frame comes or the file can't be written.

Source code in src/streamcatcher/player/session.py
def snapshot(self, path: str) -> None:
    """Grab the current viewport and save it to ``path`` as an image file.

    The saved image is the reprojected look-around viewport in 360 modes
    (what the viewer sees), or the raw frame when flat. Reads until a frame
    arrives — a just-opened decoder can return empty first — then writes it,
    raising :class:`SnapshotError` if no frame comes or the file can't be
    written.
    """
    frame = None
    for _ in range(_SNAPSHOT_READ_ATTEMPTS):
        frame = self.grab_view()
        if frame is not None:
            break
    if frame is None:
        raise SnapshotError("No frame to snapshot; the stream may have ended.")
    self.write_snapshot(frame, path)

write_snapshot

write_snapshot(frame, path)

Write an already-rendered frame to path as an image file.

Creates the parent directory if needed. Raises :class:SnapshotError when OpenCV can't encode/write the file (e.g. an unknown extension or an unwritable location).

Source code in src/streamcatcher/player/session.py
def write_snapshot(self, frame, path: str) -> None:
    """Write an already-rendered ``frame`` to ``path`` as an image file.

    Creates the parent directory if needed. Raises :class:`SnapshotError`
    when OpenCV can't encode/write the file (e.g. an unknown extension or an
    unwritable location).
    """
    cv2 = self._cv2 or _load_cv2()
    directory = os.path.dirname(path)
    if directory:
        os.makedirs(directory, exist_ok=True)
    if not cv2.imwrite(path, frame):
        raise SnapshotError(f"Could not write the snapshot to {path!r}.")

Configuration — config

streamcatcher.config

Typed, secret-safe configuration for Streamcatcher.

The stream URL is treated as a secret because RTSP/RTMP URLs routinely embed credentials (rtsp://user:pass@host). It is stored as a SecretStr so it never prints in reprs/tracebacks, and :meth:Settings.secret_values exposes the plaintext only to seed the log-redaction filter (see logging_setup).

Backend

Bases: StrEnum

Playback backend that the player factory dispatches on.

Projection

Bases: StrEnum

How the player interprets each frame's geometry.

Settings

Bases: BaseSettings

Runtime settings, populated from STREAMCATCHER_* environment vars.

display_url property

display_url

The stream URL with credentials stripped — safe to log or print.

secret_values

secret_values()

Plaintext credentials from the stream URL, to seed log redaction.

Returns the embedded password and username (if any) — the sensitive parts of an rtsp://user:pass@host URL — rather than the whole URL, so the non-secret host can still be logged.

Source code in src/streamcatcher/config.py
def secret_values(self) -> list[str]:
    """Plaintext credentials from the stream URL, to seed log redaction.

    Returns the embedded password and username (if any) — the sensitive
    parts of an ``rtsp://user:pass@host`` URL — rather than the whole URL,
    so the non-secret host can still be logged.
    """
    if self.stream_url is None:
        return []
    parts = urlsplit(self.stream_url.get_secret_value())
    return [value for value in (parts.password, parts.username) if value]

strip_url_credentials

strip_url_credentials(url)

Return url with any user:pass@ userinfo removed (host/port kept).

Source code in src/streamcatcher/config.py
def strip_url_credentials(url: str) -> str:
    """Return ``url`` with any ``user:pass@`` userinfo removed (host/port kept)."""
    parts = urlsplit(url)
    if parts.username or parts.password:
        host = parts.hostname or ""
        if parts.port is not None:
            host = f"{host}:{parts.port}"
        parts = parts._replace(netloc=host)
    return urlunsplit(parts)

Reprojection — player.reprojection

streamcatcher.player.reprojection

Reprojection of 360 equirectangular streams to a flat, look-around viewport.

A 360 camera streams an equirectangular frame — a 2:1 panorama of the whole sphere. Viewed raw it looks warped. This module aims a virtual pinhole camera — an ordinary flat window — at the source, so the viewer sees an undistorted slice and can pan/tilt/zoom around it.

The lookup tables are built with NumPy only (no OpenCV), so the math is pure, deterministic, and unit-testable without a decoder or a display. The player feeds the returned map_x/map_y to cv2.remap to warp each frame.

:class:EquirectView is the virtual camera: it samples a full 360×180 equirectangular panorama for whatever slice the current yaw/pitch/hfov select.

Coordinate convention (right-handed camera space): +X right, +Y up, +Z forward. yaw pans left/right about +Y; pitch tilts up/down (positive looks up); roll rotates about +Z (used only as a fixed mounting offset). hfov is the horizontal field of view — smaller is more zoomed in. The *_offset_deg values are fixed mounting rotations added on top of the interactive yaw/pitch, so a camera mounted rotated still reads level.

EquirectView

EquirectView(
    out_width=1280,
    out_height=720,
    hfov_deg=100.0,
    yaw_deg=0.0,
    pitch_deg=0.0,
    yaw_offset_deg=0.0,
    pitch_offset_deg=0.0,
    roll_offset_deg=0.0,
)

Bases: _SphericalView

A virtual pinhole camera aimed into a full 360×180 equirectangular panorama.

Longitude spans the full 360 across the frame width and latitude the full 180 across its height, so every camera ray lands somewhere on the frame.

Source code in src/streamcatcher/player/reprojection.py
def __init__(
    self,
    out_width: int = 1280,
    out_height: int = 720,
    hfov_deg: float = 100.0,
    yaw_deg: float = 0.0,
    pitch_deg: float = 0.0,
    yaw_offset_deg: float = 0.0,
    pitch_offset_deg: float = 0.0,
    roll_offset_deg: float = 0.0,
) -> None:
    self.out_width = int(out_width)
    self.out_height = int(out_height)
    self.hfov_deg = float(hfov_deg)
    self.yaw_deg = float(yaw_deg)
    self.pitch_deg = float(pitch_deg)
    # Fixed mounting offsets, folded into every ray build (not interactive).
    self.yaw_offset_deg = float(yaw_offset_deg)
    self.pitch_offset_deg = float(pitch_offset_deg)
    self.roll_offset_deg = float(roll_offset_deg)

Reconnect — player.reconnect

streamcatcher.player.reconnect

Reconnect policy and backoff schedule — pure, no OpenCV.

When a live stream drops, the GUI player retries the connection with exponential backoff. The policy (how fast to back off, and whether to retry at all) and the delay schedule live here, separate from any window or cv2 so they stay deterministic and unit-testable. The retry loop that consumes this schedule lives in :class:~streamcatcher.player.opencv_player.OpenCvPlayer.

ReconnectPolicy dataclass

ReconnectPolicy(
    enabled=True, base_delay=1.0, factor=2.0, max_delay=30.0
)

How to retry a dropped stream.

enabled=False restores the old behaviour: exit on the first drop. Otherwise the player retries forever, waiting base_delay seconds and multiplying by factor after each failed attempt, capped at max_delay.

backoff_delays

backoff_delays(policy)

Yield reconnect wait times forever: base, base·factor, … capped at max.

Infinite by design — with the default policy this is 1, 2, 4, 8, 16, 30, 30, … — so the player retries until the stream returns or the user quits. The caller is responsible for stopping.

Source code in src/streamcatcher/player/reconnect.py
def backoff_delays(policy: ReconnectPolicy) -> Iterator[float]:
    """Yield reconnect wait times forever: ``base, base·factor, … capped at max``.

    Infinite by design — with the default policy this is
    ``1, 2, 4, 8, 16, 30, 30, …`` — so the player retries until the stream
    returns or the user quits. The caller is responsible for stopping.
    """
    delay = policy.base_delay
    while True:
        yield min(delay, policy.max_delay)
        delay *= policy.factor

Player protocol & backends

streamcatcher.player.base

The player interface: a Protocol every backend implements.

Player

Bases: Protocol

A stream player: connect, show a window, snapshot, and stop.

play

play()

Open the stream and start playback (blocks for live backends).

Source code in src/streamcatcher/player/base.py
def play(self) -> None:
    """Open the stream and start playback (blocks for live backends)."""
    ...

stop

stop()

Stop playback and release resources.

Source code in src/streamcatcher/player/base.py
def stop(self) -> None:
    """Stop playback and release resources."""
    ...

snapshot

snapshot(path)

Save a single still frame to path.

Source code in src/streamcatcher/player/base.py
def snapshot(self, path: str) -> None:
    """Save a single still frame to ``path``."""
    ...

is_playing

is_playing()

Whether playback is currently active.

Source code in src/streamcatcher/player/base.py
def is_playing(self) -> bool:
    """Whether playback is currently active."""
    ...

streamcatcher.player.opencv_player

Live stream GUI player — a thin OpenCV window over :class:StreamSession.

OpenCV's highgui creates and pumps its own native window via imshow / waitKey from a plain Python process — including on macOS — so, unlike libVLC, we own the window without an embedded native drawable or an external app. The tradeoff: cv2.VideoCapture decodes video frames only, so this backend has no audio.

All stream and viewport logic lives in :class:StreamSession; this class only adds the window, the waitKey loop, and the keyboard bindings, so the same controls are reused by the (headless) HTTP API. cv2 is imported lazily so importing this module never requires OpenCV; tests inject a fake cv2.

SnapshotError

Bases: RuntimeError

Raised when a snapshot frame can't be captured or written to disk.

StreamOpenError

Bases: RuntimeError

Raised when OpenCV cannot open the stream URL.

OpenCvPlayer

OpenCvPlayer(
    url, projection=Projection.FLAT, reconnect=None
)

Play a live RTMP/RTSP stream in an OpenCV window (video only).

With projection=Projection.EQUIRECT the stream is treated as a 360 equirectangular panorama and reprojected to a flat viewport the user can look around with W/A/S/D (tilt/pan) and +/- (zoom). Projection.FLAT (the default) shows frames unchanged.

Source code in src/streamcatcher/player/opencv_player.py
def __init__(
    self,
    url: str,
    projection: Projection = Projection.FLAT,
    reconnect: ReconnectPolicy | None = None,
) -> None:
    self._session = StreamSession(url, projection)
    self._policy = reconnect or ReconnectPolicy()
    self._window_open = False
    self._last_frame = None  # most recently rendered frame, for the 'p' snapshot
    self._last_raw = None  # raw frame behind _last_frame, to skip re-rendering it
    self._view_dirty = False  # the viewport moved — re-render even on the same frame
    self._dragging = False  # left button held for drag-to-look (360 modes)
    self._last_mouse = (0, 0)  # last cursor position while dragging

play

play()

Open the stream and show frames until the window closes or 'q' is hit.

Source code in src/streamcatcher/player/opencv_player.py
def play(self) -> None:
    """Open the stream and show frames until the window closes or 'q' is hit."""
    cv2 = _load_cv2()
    # Create the window before touching the network so a headless OpenCV
    # build fails fast with a clear message instead of connecting first and
    # then dying on the cryptic highgui ``cv2.error``.
    try:
        cv2.namedWindow(_WINDOW_TITLE, cv2.WINDOW_NORMAL)
    except cv2.error as exc:
        raise StreamOpenError(_GUI_HELP) from exc
    cv2.setMouseCallback(_WINDOW_TITLE, self._on_mouse)
    self._window_open = True
    reader = None
    try:
        self._session.open()
        if self._session.is_360:
            log.info(
                "360 viewport enabled. Look around: W/A/S/D or drag the mouse, "
                "zoom: +/-, quit: q."
            )
        log.info("Press 'p' to save a snapshot.")
        # A background thread owns the blocking reads so decode jitter can't
        # freeze the window or the look-around keys; the loop below renders
        # the freshest frame at its own cadence and stays responsive.
        reader = FrameReader(self._session)
        reader.start()
        while True:
            raw = reader.latest()
            # Re-render only when there's a new frame or the viewport moved,
            # so an idle view doesn't re-run the (costly) 360 remap each tick.
            if raw is not None and (raw is not self._last_raw or self._view_dirty):
                self._last_frame = self._session.render(raw)
                cv2.imshow(_WINDOW_TITLE, self._last_frame)
                self._last_raw = raw
                self._view_dirty = False
            key = cv2.waitKey(_WAITKEY_MS) & 0xFF
            if key == ord("q"):
                break
            self._dispatch_key(key)
            if cv2.getWindowProperty(_WINDOW_TITLE, cv2.WND_PROP_VISIBLE) < 1:
                break  # the user closed the window
            if reader.ended():
                reader.stop()
                if not self._policy.enabled:
                    log.info("Stream ended or dropped.")
                    break
                if not self._reconnect(cv2):
                    break  # the user quit while we were reconnecting
                reader = FrameReader(self._session)  # fresh capture, fresh reader
                reader.start()
    except KeyboardInterrupt:
        log.info("Interrupted — closing the stream.")
    finally:
        if reader is not None:
            reader.stop()
        self.stop()

stop

stop()

Release the capture and destroy the window.

Source code in src/streamcatcher/player/opencv_player.py
def stop(self) -> None:
    """Release the capture and destroy the window."""
    self._session.close()
    if self._window_open:
        _load_cv2().destroyWindow(_WINDOW_TITLE)
        self._window_open = False
    log.info("Live player: stopped.")

snapshot

snapshot(path)

Capture a single frame from the stream and save it to path (no window).

Opens the stream if it isn't already, grabs one rendered frame — applying the configured projection, so the still matches what the window shows — writes it, then leaves the capture as it found it. Raises :class:StreamOpenError if the stream won't open or :class:SnapshotError if no frame arrives or the file can't be written.

Source code in src/streamcatcher/player/opencv_player.py
def snapshot(self, path: str) -> None:
    """Capture a single frame from the stream and save it to ``path`` (no window).

    Opens the stream if it isn't already, grabs one rendered frame — applying
    the configured projection, so the still matches what the window
    shows — writes it, then leaves the capture as it found it. Raises
    :class:`StreamOpenError` if the stream won't open or
    :class:`SnapshotError` if no frame arrives or the file can't be written.
    """
    opened_here = not self._session.is_open()
    if opened_here:
        self._session.open()
    try:
        self._session.snapshot(path)
        log.info("Snapshot saved to %s", path)
    finally:
        if opened_here:
            self._session.close()

is_playing

is_playing()

Whether a stream capture is currently open.

Source code in src/streamcatcher/player/opencv_player.py
def is_playing(self) -> bool:
    """Whether a stream capture is currently open."""
    return self._session.is_open()

streamcatcher.player.stub_player

Offline stub player — the default backend.

Implements the :class:~streamcatcher.player.base.Player interface without any window, decoder, or network, so the app and the whole test suite run fully offline. It records what it was asked to do rather than doing it.

StubPlayer

StubPlayer(url)

A no-op player that records requests instead of touching a stream.

Source code in src/streamcatcher/player/stub_player.py
def __init__(self, url: str) -> None:
    self._url = url  # held for parity with live backends; never logged
    self._playing = False
    self.last_snapshot: str | None = None

Player factory — player.factory

streamcatcher.player.factory

Player factory — selects a backend from settings.

Defaults to the offline stub. The live OpenCV backend (:class:OpenCvPlayer) lazy-imports cv2 inside its own play method, so importing this factory never requires OpenCV to be installed — only playing a stream does.

get_player

get_player(settings)

Build the player for settings.backend using its stream URL.

Source code in src/streamcatcher/player/factory.py
def get_player(settings: Settings) -> Player:
    """Build the player for ``settings.backend`` using its stream URL."""
    if settings.stream_url is None:
        raise ValueError("No stream URL configured.")
    url = settings.stream_url.get_secret_value()

    if settings.backend is Backend.STUB:
        return StubPlayer(url)
    if settings.backend is Backend.OPENCV:
        policy = ReconnectPolicy(
            enabled=settings.reconnect_enabled,
            base_delay=settings.reconnect_base_delay,
            factor=settings.reconnect_backoff_factor,
            max_delay=settings.reconnect_max_delay,
        )
        return OpenCvPlayer(
            url,
            projection=settings.projection,
            reconnect=policy,
        )

    raise NotImplementedError(  # pragma: no cover - defensive: Backend is exhaustive
        f"Backend {settings.backend.value!r} is not supported."
    )

HTTP API models — api.models

streamcatcher.api.models

Request and response models for the HTTP control API.

These use only Pydantic (already a core dependency), so this module imports with no web stack. The stream URL is accepted on input but never returned in any response — responses expose only the opaque session id and the viewport state.

CreateSessionRequest

Bases: BaseModel

Body for POST /session.

LookRequest

Bases: BaseModel

Body for POST /session/{id}/look — pan/tilt/zoom degree deltas.

SessionStateResponse

Bases: BaseModel

The public state of a session — its id plus the viewport orientation.