Documentation Index
Fetch the complete documentation index at: https://docs.reactor.inc/llms.txt
Use this file to discover all available pages before exploring further.
The Python SDK provides decorators as an idiomatic alternative to reactor.on(event, handler). Both approaches are equivalent. Use whichever you prefer.
@reactor.on_frame
Registers a callback to receive video frames as NumPy arrays.
@reactor.on_frame
def handler(frame: NDArray[np.uint8]) -> None: ...
frame is a NumPy array with shape (H, W, 3), dtype uint8, in RGB color order.
@reactor.on_frame
def on_frame(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))
@reactor.on_message
Registers a handler for messages from the model.
@reactor.on_message
def handler(message: Any) -> None: ...
message is the payload sent by the model (typically a dictionary). Shape depends on the model.
@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.
# 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: ...
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.
@reactor.on_error
def handler(error: ReactorError) -> None: ...
error is a ReactorError with fields code, message, timestamp, recoverable, component, and optional retry_after.
@reactor.on_error
async def on_error(error: ReactorError):
print(f"[{error.component}:{error.code}] {error.message}")
if error.recoverable:
await asyncio.sleep(error.retry_after or 3)
await reactor.reconnect()
@reactor.on_track
Registers a handler for a named media track received from the model.
@reactor.on_track(name: str)
def handler(track: MediaStreamTrack) -> None: ...
The track name to listen for (e.g., "main_video").
@reactor.on_track("main_video")
def on_main_video(track):
print(f"Received track: {track.kind}")
For most use cases, @reactor.on_frame is simpler since it delivers decoded NumPy frames directly. Use @reactor.on_track when you need access to the raw MediaStreamTrack.
