Video & Audio Tracks

Real-time media is streamed as it is produced, frame by frame, in both directions. It travels on named streams called tracks. Your model declares the tracks it sends on an Output class and the tracks it receives on a MediaInput class. The runtime handles the encoding and the transport at both ends; you deal in NumPy arrays. The examples on this page are a video-to-video model in the shape of SANA-Streaming: the client publishes its camera, the model restyles it according to a prompt, and the restyled video comes back.

Output tracks

Subclass Output with one field per track. The field name is the track name a client subscribes to, and the annotation says what it carries:

from reactor_runtime import Audio, Output, Video


class SanaStreamingOutput(Output):
    main_video: Video

A step returns an instance of this class, and the runtime puts each field on its track. Every declared track has to be present on every Output. If a step has no audio for an audio track, send an empty array rather than leaving the field out. A video frame is an RGB array of shape (height, width, 3) with dtype uint8. A step that produces several frames returns them as one (frames, height, width, 3) array; the runtime splits the batch and paces it. Audio is int16 of shape (1, samples) at 48 kHz, mono; the Audio section below covers rates and sync.

Several tracks

A model can send more than one stream, and each client picks the ones it wants. Take a two-player world model where each player gets their own point of view. The Output declares one track per player, and every step fills both:

class ArenaOutput(Output):
    # ✅ One track per point of view
    player_one: Video
    player_two: Video
    # ✅ Shared by both clients
    ambience: Audio


class Arena(ReactorApp):
    def generate(self, input: ArenaState) -> ArenaOutput:
        views = self.world.step(input.moves)  # both cameras
        return ArenaOutput(
            player_one=views[0],
            player_two=views[1],
            ambience=self.world.sound(),
        )

The two clients connect to the same session and see the same world from different places. Player one’s app subscribes to player_one, player two’s to player_two, and both take ambience. Tracks returned in one Output are kept in step with each other, so the two views and the sound line up.

Frame rate

generate() hands frames over one chunk at a time, and a chunk starts playing the moment it lands. While it plays, the runtime is already asking for the next one. Generation runs one chunk ahead of what the client is playing, and the chunks play in the order they were made:

A time axis from 0 to 3 seconds. On top, generate() returns chunk 1, chunk 2, and chunk 3 back to back. Below, each chunk's 24 frames start playing the moment that chunk lands, so playout runs exactly one chunk behind generation. — Each chunk starts playing when it lands and the next one is generated meanwhile. By default a chunk plays for as long as it took to make.

To play at a fixed rate instead, declare fps on the class:

class SanaStreaming(ReactorApp):
    fps = 24

Now every chunk plays at 24 frames per second regardless of how long it took. A model that runs faster than that is held back: returning the next Output waits while frames already handed over are still playing. That is what stops a fast model from running ahead of the client, and it is why you never need to sleep in generate(). Two controls sit on self.output, and both reach every connected client:

self.output.fps = 30   # re-pace what is queued, and what follows
self.output.flush()    # drop what is queued and cut playout

Call flush() when the scene changes, so nothing from the old scene plays after the cut. A recording keeps running across the cut.

Frame metadata

A frame can carry a small tag. Wrap the array in TrackPayload and pass metadata: a mapping travels as JSON, bytes travel as they are. On a batch, pass one value for every frame or a list with one value per frame.

tag = {"chunk": result.chunk_index}
tags = [tag] * len(result.frames)
video = TrackPayload(result.frames, metadata=tags)
return SanaStreamingOutput(main_video=video)

The client reads the tag next to the frame it came with. A video-to-video model can copy the tag it received on an input frame onto the frame it produced from it, which lets the client pair the two without a side channel.

Input tracks

Subclass MediaInput with one field per track you expect the client to publish, and annotate it on your model. The runtime finds it by its type, so the attribute name is up to you; media is the convention:

from reactor_runtime import MediaInput, ReactorApp, Video

class SanaStreamingMedia(MediaInput):
    webcam: Video

class SanaStreaming(ReactorApp):
    media: SanaStreamingMedia

The runtime negotiates each declared track with the client and buffers the frames that arrive, reachable as self.media.webcam. Each track holds the last 128 frames and drops the oldest when full, so a model that falls behind does not grow memory without bound. Input is the old name for MediaInput. It still imports, with a deprecation warning.

Several tracks

A client can publish more than one stream too. A robot policy driven from three cameras declares one track per camera and reads them all in process_input():

class RobotMedia(MediaInput):
    # ✅ One track per camera; the robot publishes all three
    head: Video
    left_wrist: Video
    right_wrist: Video


class Policy(ReactorApp):
    media: RobotMedia

    async def process_input(self) -> PolicyInput:
        # ✅ The newest frame from each camera
        head = self.media.head.try_read(1)
        left = self.media.left_wrist.try_read(1)
        right = self.media.right_wrist.try_read(1)

        # ✅ Wait until every camera has delivered at least once
        if head is None or left is None or right is None:
            raise ApplicationError("waiting for the cameras")

        return PolicyInput(
            head=head[0].data,
            left_wrist=left[0].data,
            right_wrist=right[0].data,
            task=self.state.task,
        )

Each track is its own buffer, so the cameras do not have to arrive in lockstep. Reading the newest frame from each gives the policy the most recent view from every angle, whatever order the frames came in.

Reading frames

Read tracks in process_input() with try_read(n). It returns the n frames you asked for, or None if fewer than that have arrived, and it never waits. Too few frames is a reason to skip the step, so the model runs only with a full input:

async def process_input(self) -> SanaStreamingInput:
    # ✅ Take the 4 newest camera frames
    frames = self.media.webcam.try_read(4)

    # ✅ Not enough yet: skip this step and try again shortly
    if frames is None:
        raise ApplicationError("waiting for 4 webcam frames")

    # ✅ Stack them into one array for the model
    batch = np.stack([f.data for f in frames])
    prompt = self.state.prompt
    return SanaStreamingInput(prompt=prompt, frames=batch)

Each frame arrives as an InputFrame, whose data holds the array: (height, width, 3) uint8 for video, (1, samples) int16 for audio. A frame also carries the bytes the client tagged it with and, when the sender supplied one, a presentation timestamp. Reads take a mode. ReadMode.LATEST, the default, returns the newest frames and drops the backlog, which is what video wants: if the model fell behind, it should catch up to now rather than restyle stale frames. ReadMode.FIFO returns the oldest in order and leaves the rest queued, which is what audio wants, since dropped samples are audible gaps.

frames = self.media.webcam.try_read(4)  # newest 4
mic = self.media.mic
chunks = mic.try_read(1, mode=ReadMode.FIFO)  # oldest 1

await read(n) waits until n frames are available instead of returning None. Inside process_input() a wait holds the step and no command runs until a frame arrives, so prefer try_read() there and keep read() for a hand-written run(). When the last client leaves or the session ends, every input buffer is reset, so the next client starts from empty tracks.

Inbound media reference

Every field on a frame, both read calls with their arguments, and the rest of the buffer surface.

Audio

Audio tracks work like video tracks. Declare them on the Output, return samples alongside the frames, and the runtime encodes and transports them:

class MyOutput(Output):
    main_video: Video
    main_audio: Audio


def generate(self, input: MyState) -> MyOutput:
    video = self.pipe.forward(prompt=input.prompt)
    audio = self.pipe.generate_audio()
    return MyOutput(
        main_video=video,  # (H, W, 3) uint8
        main_audio=audio,  # (1, N) int16
    )

Samples are int16, shaped (1, samples) for mono, at 48 kHz. The runtime encodes them as Opus, whose native rate is 48 kHz, so audio already at that rate needs no resampling. When a step has no audio, send an empty (1, 0) array rather than leaving the field out.

Keeping audio and video together

When one Output carries both tracks, the runtime keeps them aligned: if the video is a batch, the audio is spread across those frames. So each Output should carry the audio that belongs to its video. At 30 fps one frame is 48000 / 30 = 1600 samples, so a batch of 3 frames carries about 4,800 samples. A frame’s worth of video with a second’s worth of audio drifts, however good the transport.

Other sample rates

Live audio always goes out at 48 kHz, and the runtime never resamples for you. If your model produces audio at another rate, resample to 48 kHz before it goes on the Output, or playback runs at the wrong pitch. If you would rather emit at the native rate, subclass Audio and declare it:

class Audio16k(Audio):
    sample_rate = 16_000


class MyOutput(Output):
    main_video: Video
    main_audio: Audio16k

This does not change what goes out live, which stays 48 kHz. It matters for recording: the recorder encodes the audio at the rate you declare, which is what keeps a recorded clip’s pitch and duration right.

Two producers

A model whose audio and video come from separate producers has to pair them before emitting. That is a loop which blocks on input, so it keeps a hand-written run() (see Controlling the loop). Buffer whichever stream runs ahead, and emit only when both have material for the same span:

async def run(self) -> None:
    while True:
        await self.connected.wait()
        audio_pending: dict[int, np.ndarray] = {}
        video_pending: dict[int, np.ndarray] = {}
        next_block = 0

        while self.connected.is_set():
            kind, index, data = await self._next_block()
            if kind == "audio":
                audio_pending[index] = data
            else:
                video_pending[index] = data

            # ✅ Emit only once both tracks have the same block
            ready = lambda i: i in audio_pending and i in video_pending
            while ready(next_block):
                await self.emit(
                    MyOutput(
                        main_video=video_pending.pop(next_block),
                        main_audio=audio_pending.pop(next_block),
                    )
                )
                next_block += 1

Pairing by block index rather than by arrival order is what keeps the two tracks locked when one producer is briefly slower than the other.

Reading audio from the client

Inbound audio arrives as (1, samples) int16 at 48 kHz and is read like video, but in FIFO order so no samples are dropped:

chunks = self.media.mic.try_read(1, mode=ReadMode.FIFO)

A common pattern is to drain everything queued each step:

audio = []
while (chunks := self.media.mic.try_read(1, mode=ReadMode.FIFO)):
    audio.extend(chunks)

A video-to-video model

Everything above, in one class. The client publishes webcam and subscribes to main_video. Each step takes the four newest camera frames and the current prompt, restyles them, and sends them back tagged with the step they came from.

sana_streaming.py

class SanaStreamingState(InputState):
    ...  # prompt and strength, as on Managing State


# ✅ What the client sends: one video track named "webcam"
class SanaStreamingMedia(MediaInput):
    webcam: Video


# ✅ What the client receives: one video track named "main_video"
class SanaStreamingOutput(Output):
    main_video: Video


# ✅ Optional typing: what process_input() hands to generate()
@dataclass(frozen=True)
class SanaStreamingInput:
    prompt: str
    frames: np.ndarray  # (4, H, W, 3) uint8


class SanaStreaming(ReactorApp):
    state: SanaStreamingState
    media: SanaStreamingMedia

    def load(self, config_path: Path | None) -> None:
        weights = get_weights_path()
        self.pipe = SanaStreamingPipeline.from_pretrained(weights)
        self.chunk = 0

    async def process_input(self) -> SanaStreamingInput:
        # ✅ Take the 4 newest camera frames
        frames = self.media.webcam.try_read(4)

        # ✅ Not enough yet: skip this step and try again shortly
        if frames is None:
            raise ApplicationError("waiting for 4 webcam frames")

        # ✅ Stack them into one array for the model
        batch = np.stack([f.data for f in frames])
        prompt = self.state.prompt
        return SanaStreamingInput(prompt=prompt, frames=batch)

    def generate(self, input: SanaStreamingInput) -> np.ndarray:
        return self.pipe.restyle(
            input.frames,
            prompt=input.prompt,
            strength=self.state.strength,
        )

    async def process_output(
        self, outcome: StepOutcome
    ) -> SanaStreamingOutput | None:
        if outcome.error is not None:
            raise outcome.error

        # ✅ Tag every frame with the step it came from
        self.chunk += 1
        tag = {"chunk": self.chunk}
        video = TrackPayload(outcome.result, metadata=[tag] * 4)
        return SanaStreamingOutput(main_video=video)

Before the first camera frame lands, every step is skipped and generate() never runs. Once frames flow, the model keeps up with the camera by design: LATEST reads always start from the newest frames, so a slow step costs a few dropped camera frames rather than growing lag.

Session Recording

Record sessions and let clients capture clips.

The Step Loop

How process_input() and process_output() fit around generate().

Get Started

CLI & Platform

Build a Model

Media & Transport

Releases

Video & Audio Tracks

Output tracks

Several tracks

Frame rate

Frame metadata

Input tracks

Several tracks

Reading frames

Inbound media reference

Audio

Keeping audio and video together

Other sample rates

Two producers

Reading audio from the client

A video-to-video model

Next

Session Recording

The Step Loop

​Output tracks

​Several tracks

​Frame rate

​Frame metadata

​Input tracks

​Several tracks

​Reading frames

Inbound media reference

​Audio

​Keeping audio and video together

​Other sample rates

​Two producers

​Reading audio from the client

​A video-to-video model

​Next

Session Recording

The Step Loop

Output tracks

Several tracks

Frame rate

Frame metadata

Input tracks

Several tracks

Reading frames

Audio

Keeping audio and video together

Other sample rates

Two producers

Reading audio from the client

A video-to-video model

Next