Decorators

The Python SDK provides decorators as an idiomatic alternative to reactor.on(event, handler). Both approaches are equivalent. Use whichever you prefer.

Receiving video frames

Example

output = reactor.tracks.with_direction("recvonly").with_kind("video").one()

@output.on_frame
def render(frame):
    print(f"Frame: {frame.shape}")  # e.g. (768, 1280, 3)

    # PIL
    from PIL import Image
    Image.fromarray(frame).save("frame.png")

    # OpenCV (convert RGB to BGR first)
    import cv2
    cv2.imwrite("frame.png", cv2.cvtColor(frame, cv2.COLOR_RGB2BGR))

Use on_raw_frame() instead of on_frame() for the same frames as raw bytes, with no NumPy conversion; see Track for the full API.

`@reactor.on_message`

Registers a handler for messages from the model.

Signature

@reactor.on_message
def handler(message: Any) -> None: ...

message is the payload sent by the model (typically a dictionary). Shape depends on the model.

Example

@reactor.on_message
def on_message(message):
    if message.get("type") == "state":
        print(f"Frame: {message['data']['current_frame']}")

`@reactor.on_status`

Registers a handler for connection status changes. Supports three usage patterns.

Signature

# All status changes
@reactor.on_status
def handler(status: ReactorStatus) -> None: ...

# One specific status
@reactor.on_status(ReactorStatus.READY)
def handler(status: ReactorStatus) -> None: ...

# Multiple statuses
@reactor.on_status([ReactorStatus.CONNECTING, ReactorStatus.WAITING])
def handler(status: ReactorStatus) -> None: ...

Example

from reactor_sdk import ReactorStatus

@reactor.on_status(ReactorStatus.READY)
async def on_ready(status):
    await reactor.send_command("start", {})

@reactor.on_status([ReactorStatus.CONNECTING, ReactorStatus.WAITING])
def on_pending(status):
    print(f"Connecting: {status}")

`@reactor.on_error`

Registers a handler for errors.

Signature

@reactor.on_error
def handler(error: ReactorError) -> None: ...

error is a ReactorError with fields code, message, timestamp_ms, recoverable, and optional status, operation, retry_after_ms.

Example

@reactor.on_error
async def on_error(error: ReactorError):
    print(f"{error.operation}: [{error.code}] {error.message}")
    if error.recoverable:
        await asyncio.sleep((error.retry_after_ms or 3000) / 1000)
        await reactor.reconnect()

`@reactor.on_track`

Registers a handler for recvonly tracks becoming available. Unlike the other decorators here, it does not filter by name — check track.name yourself if you only care about one track.

Signature

@reactor.on_track
def handler(track: Track) -> None: ...

track is the Track itself — reactor.track(track.name), already resolved — not a bare name to look up before doing anything with it. Its .mid is the WebRTC media stream ID, or None if it has not been negotiated yet.

Example

@reactor.on_track
def on_track(track):
    print(f"Track available: {track.name} (mid={track.mid})")

This does not hand over decoded media — there is no MediaStreamTrack in this SDK, and the Track it hands over is a control handle (name, kind, direction, push_frame/on_frame, publish/pause/resume), not a frame source by itself. This event is for knowing when a named track becomes available at all (e.g. to enable a UI element for it); to register a handler that only fires for one track’s decoded frames, use track.on_frame instead.

Get Started

Concepts

SDK Reference

Resources

Receiving video frames

`@reactor.on_message`

`@reactor.on_status`

`@reactor.on_error`

`@reactor.on_track`

​Receiving video frames

​@reactor.on_message

​@reactor.on_status

​@reactor.on_error

​@reactor.on_track

Receiving video frames

`@reactor.on_message`

`@reactor.on_status`

`@reactor.on_error`

`@reactor.on_track`