Media Input

A model can receive live video and audio from the client — webcam, screen share, microphone. You declare the tracks you expect, and the runtime hands you a buffer per track to read from.

Declaring input tracks

Subclass Input with one field per track, then annotate it on your model:

from reactor_runtime import Audio, Input, ReactorModel, Video


class SessionInput(Input):
    webcam: Video
    mic: Audio


class MyModel(ReactorModel):
    input: SessionInput
    fps = 24

The field names are the track identifiers your client publishes to. The runtime binds a live buffer to each one, reachable as self.input.webcam and self.input.mic. The attribute name is up to you — input is the convention, but the runtime finds the holder by its type, not its name.

Frames

Every read returns a list of InputFrame:

Field	Meaning
`data`	The payload as a NumPy array.
`pts`	Presentation timestamp in seconds, or `None` when unavailable.
`metadata`	The bytes the sender attached to this frame, or `None`. Decoding them is yours.

Video data is (height, width, 3) uint8 RGB. Audio data is (1, samples) int16 mono at 48 kHz.

frames = self.input.webcam.try_read()
if frames is not None:
    rgb = frames[0].data
    timestamp = frames[0].pts

Reading without blocking

try_read(n=1) returns n frames, or None when fewer than n have arrived. It never waits, and when it returns None it consumes nothing — the frames already buffered are still there for the next call. This is the right default for a generation loop, because it lets the model keep producing when the client’s camera stalls:

async def run(self) -> None:
    while True:
        await self.connected.wait()
        while self.connected.is_set():
            frames = self.input.webcam.try_read()
            if frames is None:
                # ✅ Nothing new yet — yield and come back
                await asyncio.sleep(0)
                continue
            result = self.pipe.forward(frames[0].data)
            await self.emit(MyOutput(main_video=result))

Reading with a wait

await read(n) parks until n frames are available, then returns them. Use it when a forward pass genuinely cannot proceed without input:

[frame] = await self.input.webcam.read(1)
result = self.pipe.forward(frame.data)

Pass timeout in seconds for a bounded wait, which raises TimeoutError if it elapses:

frames = await self.input.webcam.read(1, timeout=0.5)

An indefinite read() parks until a frame arrives or the track closes. Closing the track is what releases it, so prefer try_read() or a timeout in a loop that also needs to notice a client leaving.

`BufferClosed`

Reads raise BufferClosed once a track is closed. Catch it around the session loop:

from reactor_runtime import BufferClosed


async def run(self) -> None:
    while True:
        await self.connected.wait()
        try:
            while self.connected.is_set():
                [frame] = await self.input.webcam.read(1)
                await self.emit(MyOutput(main_video=self.pipe.forward(frame.data)))
        except BufferClosed:
            continue

Newest frames or oldest

Reads take a mode that decides which frames you get:

from reactor_runtime import ReadMode

# Newest n, discarding any older backlog (the default)
frames = self.input.webcam.try_read(1, mode=ReadMode.LATEST)

# Oldest n, in arrival order, leaving the rest queued
chunks = self.input.mic.try_read(1, mode=ReadMode.FIFO)

LATEST is right for video: if your model falls behind, you want the current frame, not a stale one. It clears the backlog so you never accumulate lag. FIFO is right for audio: dropping chunks leaves audible gaps and sample-level discontinuities, so consume them in order. A common pattern is to drain every queued chunk into a backlog each tick:

audio = []
while True:
    chunks = self.input.mic.try_read(1, mode=ReadMode.FIFO)
    if chunks is None:
        break
    audio.extend(chunks)

Each buffer holds 128 frames and evicts the oldest when full, so a model that never reads a track will not grow memory without bound.

Clearing between clients

Buffers persist across connections and across sessions, until you clear them. When a new client joins, frames from the previous one can still be queued, so clear the tracks you care about:

@connected
async def on_connect(self) -> None:
    if self._viewers == 0:
        self.input.webcam.clear()
        self.input.mic.clear()
    self._viewers += 1

clear() drops buffered frames and leaves the track open for reading.

Full example: video to video

import asyncio

from reactor_runtime import Input, InputField, Output, ReactorModel, Video, connected, event


class V2VInput(Input):
    camera: Video


class V2VOutput(Output):
    main_video: Video


class V2VModel(ReactorModel):
    input: V2VInput
    fps = 30

    def load(self, config_path: Path | None) -> None:
        self.pipe = load_style_model()
        self.style = "none"
        self._viewers = 0

    @connected
    async def on_connect(self) -> None:
        if self._viewers == 0:
            self.input.camera.clear()
            self.style = "none"
        self._viewers += 1

    @event(name="set_style", description="Style filter applied to the camera feed")
    async def set_style(
        self,
        style: str = InputField(default="none", choices=["none", "oil_paint", "sketch"]),
    ) -> None:
        self.style = style

    async def run(self) -> None:
        while True:
            await self.connected.wait()
            while self.connected.is_set():
                frames = self.input.camera.try_read()
                if frames is None:
                    await asyncio.sleep(0)
                    continue
                result = self.pipe.apply(frames[0].data, style=self.style)
                await self.emit(V2VOutput(main_video=result))

Audio

Emit audio tracks and keep them in sync with video.

The Run Loop

Emitting frames, batches, and frame rates.

Get Started

Build a Model

Media & Transport

Releases

Declaring input tracks

Frames

Reading without blocking

Reading with a wait

`BufferClosed`

Newest frames or oldest

Clearing between clients

Full example: video to video

Next

Audio

The Run Loop

​Declaring input tracks

​Frames

​Reading without blocking

​Reading with a wait

​BufferClosed

​Newest frames or oldest

​Clearing between clients

​Full example: video to video

​Next

Audio

The Run Loop

Declaring input tracks

Frames

Reading without blocking

Reading with a wait

`BufferClosed`

Newest frames or oldest

Clearing between clients

Full example: video to video

Next