Camera

Runnable example

python/examples/camera_view.py — streams for 5 s and prints fps and final-frame luminance stats.

resoio.camera.CameraClient

CameraClient(socket_path: str | None = None)

Bases: _BaseClient[CameraStub]

Async client for the Resonite IO Camera service over a UDS.

Use as an async context manager so the gRPC channel is closed deterministically. Socket resolution mirrors :class:resoio.ConnectionClient.

Source code in src/resoio/_client.py

def __init__(self, socket_path: str | None = None) -> None:
    # Defer resolution to __aenter__ so env vars patched between
    # construction and connection are honoured, and so resolution
    # errors surface at the connect site.
    self._explicit_path: str | None = socket_path
    self._channel: Channel | None = None
    self._stub: TStub | None = None
    self._resolved_path: str | None = None

stream async

stream() -> AsyncGenerator[Frame]

Stream camera frames from the server.

The capture resolution is the Display modality's responsibility; frames arrive uncapped (best-effort native fps). Raises :class:RuntimeError if called outside async with.

Source code in src/resoio/camera.py

async def stream(self) -> AsyncGenerator[Frame]:
    """Stream camera frames from the server.

    The capture resolution is the Display modality's responsibility;
    frames arrive uncapped (best-effort native fps). Raises
    :class:`RuntimeError` if called outside ``async with``.
    """
    stub = self._require_stub()
    request = CameraStreamRequest()
    async for raw in stub.stream_frames(request):
        pixels = np.frombuffer(raw.pixels, dtype=np.uint8).reshape(
            raw.height, raw.width, 4
        )
        yield Frame(
            pixels=pixels,
            unix_nanos=raw.unix_nanos,
            frame_id=raw.frame_id,
        )

shot async

shot() -> Frame

Take a single frame, then close the stream (one-shot).

Convenience wrapper over :meth:stream for the common screenshot case: open the server stream, return the first :class:Frame, and close the stream so no further frames are pulled. Raises :class:RuntimeError if called outside async with or if the server closes the stream before delivering a frame.

Source code in src/resoio/camera.py

async def shot(self) -> Frame:
    """Take a single frame, then close the stream (one-shot).

    Convenience wrapper over :meth:`stream` for the common
    screenshot case: open the server stream, return the first
    :class:`Frame`, and close the stream so no further frames are
    pulled. Raises :class:`RuntimeError` if called outside
    ``async with`` or if the server closes the stream before
    delivering a frame.
    """
    frames = self.stream()
    try:
        return await anext(frames)
    except StopAsyncIteration:
        raise RuntimeError(
            "camera stream ended before delivering a frame"
        ) from None
    finally:
        await frames.aclose()

resoio.camera.Frame dataclass

Frame(pixels: NDArray[uint8], unix_nanos: int, frame_id: int)

One decoded camera frame.

pixels is an (H, W, 4) RGBA8 view over the protobuf payload bytes (read-only; call .copy() for a writable array). Row 0 is the image top. height / width / channels are derived from pixels.shape. unix_nanos is the server-side capture timestamp in UTC nanos since the Unix epoch. frame_id is a server-side monotonic counter that restarts at 0 per stream() call.