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.")

abandon

abandon()

Drop the capture without releasing it, for when a read may be in flight.

Releasing a capture another thread is still blocked inside is undefined behaviour in OpenCV's native layer and can crash the process. When a background reader won't stop (see :meth:FrameReader.stop), the caller abandons the capture instead: the session forgets it, so a later :meth:close or :meth:reconnect is safe, and the operating system reclaims the socket and decoder when the process exits.

Source code in src/streamcatcher/player/session.py

def abandon(self) -> None:
    """Drop the capture *without* releasing it, for when a read may be in flight.

    Releasing a capture another thread is still blocked inside is undefined
    behaviour in OpenCV's native layer and can crash the process. When a
    background reader won't stop (see :meth:`FrameReader.stop`), the caller
    abandons the capture instead: the session forgets it, so a later
    :meth:`close` or :meth:`reconnect` is safe, and the operating system
    reclaims the socket and decoder when the process exits.
    """
    if self._cap is not None:
        log.warning("Abandoning the stream capture without releasing it (read in flight).")
    self._cap = None
    self._maps = None

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()

capture_fps

capture_fps()

The capture's reported frame rate, or None when it doesn't report one.

Live streams often report 0 here, so callers should treat None as "unknown" and fall back to a default.

Source code in src/streamcatcher/player/session.py

def capture_fps(self) -> float | None:
    """The capture's reported frame rate, or ``None`` when it doesn't report one.

    Live streams often report ``0`` here, so callers should treat ``None`` as
    "unknown" and fall back to a default.
    """
    if self._cap is None:
        return None
    fps = self._cap.get(self._cv2.CAP_PROP_FPS)
    return fps if fps and fps > 0 else None

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}.")

split_orientations

split_orientations(
    size=DEFAULT_SIZE, hfov_deg=DEFAULT_HFOV_DEG
)

Grab one raw frame and split it into four cardinal views.

Reads the raw equirectangular frame (never the look-around viewport) — retrying while a just-opened decoder returns empty — then reprojects it into {"front", "right", "back", "left"} images. Raises :class:OrientationError if no frame arrives.

Source code in src/streamcatcher/player/session.py

def split_orientations(
    self, size: int = DEFAULT_SIZE, hfov_deg: float = DEFAULT_HFOV_DEG
) -> dict:
    """Grab one raw frame and split it into four cardinal views.

    Reads the raw equirectangular frame (never the look-around viewport) —
    retrying while a just-opened decoder returns empty — then reprojects it
    into ``{"front", "right", "back", "left"}`` images. Raises
    :class:`OrientationError` if no frame arrives.
    """
    frame = None
    for _ in range(_SNAPSHOT_READ_ATTEMPTS):
        frame = self.read_frame()
        if frame is not None:
            break
    if frame is None:
        raise OrientationError("No frame to split; the stream may have ended.")
    cv2 = self._cv2 or _load_cv2()
    return split_equirect(frame, cv2, size=size, hfov_deg=hfov_deg)

write_orientations

write_orientations(views, out_dir, ext='.jpg')

Write each cardinal view to out_dir as <name><ext>.

Creates out_dir if needed. Returns {name: path}. Raises :class:OrientationError if any file can't be written.

Source code in src/streamcatcher/player/session.py

def write_orientations(self, views: dict, out_dir: str, ext: str = ".jpg") -> dict:
    """Write each cardinal view to ``out_dir`` as ``<name><ext>``.

    Creates ``out_dir`` if needed. Returns ``{name: path}``. Raises
    :class:`OrientationError` if any file can't be written.
    """
    cv2 = self._cv2 or _load_cv2()
    os.makedirs(out_dir, exist_ok=True)
    paths: dict = {}
    for name, image in views.items():
        path = os.path.join(out_dir, f"{name}{ext}")
        if not cv2.imwrite(path, image):
            raise OrientationError(f"Could not write the orientation image to {path!r}.")
        paths[name] = path
    return paths

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.

RecordMode

Bases: StrEnum

How streamcatcher play --record writes the stream to disk.

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]

validate_stream_url

validate_stream_url(url)

Return url unchanged if it is a supported stream URL, else raise.

Guards the boundary where a URL becomes a capture: OpenCV will happily open a local file path or an http:// address, so an unvalidated URL lets a caller read arbitrary files or reach internal hosts through the server. A netloc is required as well as the scheme, so rtsp:// alone is rejected.

Source code in src/streamcatcher/config.py

def validate_stream_url(url: str) -> str:
    """Return ``url`` unchanged if it is a supported stream URL, else raise.

    Guards the boundary where a URL becomes a capture: OpenCV will happily open
    a local file path or an ``http://`` address, so an unvalidated URL lets a
    caller read arbitrary files or reach internal hosts through the server. A
    netloc is required as well as the scheme, so ``rtsp://`` alone is rejected.
    """
    parts = urlsplit(url)
    if parts.scheme.lower() not in STREAM_URL_SCHEMES or not parts.netloc:
        supported = " or ".join(STREAM_URL_PREFIXES)
        raise ValueError(f"Stream URL must start with {supported} and include a host.")
    return url

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``."""
    ...

save_orientations

save_orientations(out_dir)

Split a 360 frame into four cardinal views saved under out_dir.

Source code in src/streamcatcher/player/base.py

def save_orientations(self, out_dir: str) -> None:
    """Split a 360 frame into four cardinal views saved under ``out_dir``."""
    ...

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.

OrientationError

Bases: RuntimeError

Raised when the four-orientation split can't be produced or written.

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,
    recorder=None,
    record_duration=None,
    orientation_size=DEFAULT_SIZE,
    orientation_hfov_deg=DEFAULT_HFOV_DEG,
)

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,
    recorder: Recorder | None = None,
    record_duration: float | None = None,
    orientation_size: int = DEFAULT_SIZE,
    orientation_hfov_deg: float = DEFAULT_HFOV_DEG,
) -> None:
    self._session = StreamSession(url, projection)
    self._policy = reconnect or ReconnectPolicy()
    self._recorder = recorder  # None unless --record was requested
    self._recording = False  # the recorder has been started (lazily, on frame 1)
    # Optional recording length cap (seconds). The deadline is a monotonic
    # timestamp set when recording actually starts (the first frame), so the
    # limit measures recorded time, not time spent waiting for a frame.
    self._record_duration = record_duration
    self._record_deadline: float | None = None
    self._orientation_size = orientation_size
    self._orientation_hfov_deg = orientation_hfov_deg
    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()
            if raw is not None:
                new_frame = raw is not self._last_raw
                # 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 new_frame or self._view_dirty:
                    self._last_frame = self._session.render(raw)
                    cv2.imshow(_WINDOW_TITLE, self._last_frame)
                    self._view_dirty = False
                if new_frame:
                    # Record the raw frame (the full panorama in 360), not the
                    # rendered viewport, so a recording doesn't follow the look.
                    self._record(raw)
                    self._last_raw = raw
            if self._record_duration_reached():
                log.info("Reached the recording duration limit — stopping.")
                break
            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():
                self._release_reader(reader)
                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:
            self._release_reader(reader)
        self._stop_recording()
        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()

save_orientations

save_orientations(out_dir)

Split one 360 frame into four cardinal views and save them (no window).

Opens the stream if needed, grabs one raw equirectangular frame, splits it into front/right/back/left images, writes them into out_dir as front.jpg/right.jpg/back.jpg/left.jpg, then leaves the capture as it found it. Raises :class:StreamOpenError if the stream won't open or :class:OrientationError if no frame arrives or a file can't be written.

Source code in src/streamcatcher/player/opencv_player.py

def save_orientations(self, out_dir: str) -> None:
    """Split one 360 frame into four cardinal views and save them (no window).

    Opens the stream if needed, grabs one raw equirectangular frame, splits
    it into front/right/back/left images, writes them into ``out_dir`` as
    ``front.jpg``/``right.jpg``/``back.jpg``/``left.jpg``, then leaves the
    capture as it found it. Raises :class:`StreamOpenError` if the stream
    won't open or :class:`OrientationError` if no frame arrives or a file
    can't be written.
    """
    opened_here = not self._session.is_open()
    if opened_here:
        self._session.open()
    try:
        views = self._session.split_orientations(
            size=self._orientation_size, hfov_deg=self._orientation_hfov_deg
        )
        paths = self._session.write_orientations(views, out_dir)
        log.info("Saved %d orientation views to %s", len(paths), out_dir)
    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
    self.last_orientations: 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, record_path=None)

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

When record_path is set, the OpenCV backend also records the stream to that file (see settings.record_mode). Recording requires the OpenCV backend — it is not supported for the offline stub.

Source code in src/streamcatcher/player/factory.py

def get_player(settings: Settings, record_path: str | None = None) -> Player:
    """Build the player for ``settings.backend`` using its stream URL.

    When ``record_path`` is set, the OpenCV backend also records the stream to
    that file (see ``settings.record_mode``). Recording requires the OpenCV
    backend — it is not supported for the offline stub.
    """
    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:
        if record_path is not None:
            raise ValueError(
                "Recording needs the 'opencv' backend; the stub backend cannot record."
            )
        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,
        )
        recorder = (
            build_recorder(settings.record_mode, record_path, settings, url)
            if record_path is not None
            else None
        )
        return OpenCvPlayer(
            url,
            projection=settings.projection,
            reconnect=policy,
            recorder=recorder,
            record_duration=settings.record_duration if record_path is not None else None,
            orientation_size=settings.orientation_size,
            orientation_hfov_deg=settings.orientation_hfov_deg,
        )

    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.