Create a demo

A demo implements IApplication and ISession from flashdreams.api_v2. Reuse an app from the demo gallery when it already supports your interaction; then add a model adapter.

This tutorial builds the Minimal T2V app (minimal_t2v), which sends ImGui prompts to a model loop. All source below is included from the repository.

New application files

apps/minimal_t2v/
  pyproject.toml
  minimal_t2v/
    __init__.py
    application.py
    session.py
    model_loop.py
    ui_loop.py
  tests/test_application.py

docs/source/demos/minimal_t2v.md # Add the app page with controls, arguments, and a link to its model's quick start.
docs/source/demos/.nav.yml # Add `Minimal T2V: minimal_t2v.md` to the sidebar.
docs/source/demos/index.md # Add a Minimal T2V gallery card linking to `minimal_t2v/`.

Package

Define a new demo package and export its application slug.

apps/minimal_t2v/pyproject.toml:

# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
# SPDX-License-Identifier: Apache-2.0

[build-system]
requires = ["setuptools>=69", "wheel"]
build-backend = "setuptools.build_meta"

[project]
name = "flashdreams-minimal-t2v"
version = "0.1.0"
description = "Reference implementation for the Create a demo tutorial"
requires-python = ">=3.10"
dependencies = ["flashdreams[local-window]"]

[tool.uv.sources]
flashdreams = { workspace = true }

[tool.setuptools.packages.find]
include = ["minimal_t2v*"]

apps/minimal_t2v/minimal_t2v/__init__.py:

# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
# SPDX-License-Identifier: Apache-2.0

"""Reference application for the Create a demo tutorial."""

from .application import MinimalT2VApplication

__all__ = ["MinimalT2VApplication"]

IApplication

Implement an IApplication to be implemented by every model that desires to integrate with this demo:

__init__ is your constructor/contract to any implementor of this demo.
- pipeline_config is the model's pipeline configuration.
- desc is our session description - more on this later.
- total_blocks is the model's total number of blocks to generate for the entire session.
init() parses arguments & is where we load/download our model.
session_desc() is how our ISession collects a session description from our IApplication.
- SessionDesc
create_session() is called every time a session is made, passing IApplication data to ISession

close() is called when the application is shutting down.

# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
# SPDX-License-Identifier: Apache-2.0

"""Minimal text-to-video application with lazy model initialization."""

import argparse
from collections.abc import Sequence
from typing import Any

from flashdreams.api_v2.application import IApplication
from flashdreams.runtime_v2.session_desc import SessionDesc
from flashdreams.runtime_v2.video_tensor import VideoTensorLayout

from .session import MinimalT2VSession


class MinimalT2VApplication(IApplication):
    """Tutorial app configured by a model adapter.

    The pipeline must accept text and latent dimensions in ``initialize_cache``
    and return ``tchw`` video from ``generate``, as Wan T2V pipelines do.
    """

    def __init__(
        self, pipeline_config: Any, desc: SessionDesc, total_blocks: int = 8
    ) -> None:
        """Keep model configuration without loading weights."""
        if desc.output_layout is not VideoTensorLayout.tchw:
            raise ValueError("Minimal T2V requires tchw output.")
        if total_blocks <= 0:
            raise ValueError("total_blocks must be positive.")
        self.pipeline_config = pipeline_config
        self.desc = desc
        self.total_blocks = total_blocks
        self.pipeline: Any = None
        self.prompt = ""

    def init(self, commandline_args: Sequence[str]) -> None:
        """Parse the initial prompt and load the shared pipeline."""
        parser = argparse.ArgumentParser(description="Generate video from text.")
        parser.add_argument("--default-prompt", default="")
        parser.add_argument("--device", default="cuda")
        args = parser.parse_args(commandline_args)
        self.prompt = args.default_prompt.strip()
        self.pipeline = self.pipeline_config.setup().to(args.device).eval()

    def session_desc(self) -> SessionDesc:
        """Return the model adapter's output description."""
        return self.desc

    def create_session(self, session_desc: SessionDesc) -> MinimalT2VSession:
        """Create a session at the model's supported size and layout."""
        if self.pipeline is None:
            raise RuntimeError("Call init before create_session.")
        if (
            session_desc.video_width != self.desc.video_width
            or session_desc.video_height != self.desc.video_height
            or session_desc.output_layout != self.desc.output_layout
            or session_desc.frames_per_second_for_step
            != self.desc.frames_per_second_for_step
        ):
            raise ValueError("The session must match the model's output description.")
        ratio = self.pipeline.decoder.spatial_compression_ratio
        if session_desc.video_width % ratio or session_desc.video_height % ratio:
            raise ValueError(
                "Video dimensions must be divisible by decoder compression."
            )
        return MinimalT2VSession(
            self.pipeline, session_desc, self.prompt, self.total_blocks
        )

    def close(self) -> None:
        """Release the shared pipeline reference."""
        self.pipeline = None

Session

ISession owns the two main loops driving our application session (running in separate threads):

IModelLoop: Drives model inference/logic.
IUILoop: Drives UI/user-interaction.

These are set with register_*_loop() methods part of ISession. These must be called during ISession.init().

# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
# SPDX-License-Identifier: Apache-2.0

"""Session wiring for the Minimal T2V model and UI loops."""

from typing import Any

from flashdreams.api_v2.session import ISession
from flashdreams.runtime_v2.session_desc import SessionDesc

from .model_loop import TextToVideoLoop, TextToVideoState
from .ui_loop import MinimalT2VUILoop, MinimalT2VUIState


class MinimalT2VSession(ISession):
    """One interactive tutorial session sharing the application's pipeline."""

    def __init__(
        self, pipeline: Any, desc: SessionDesc, prompt: str, total_blocks: int
    ) -> None:
        """Keep session inputs until the runtime initializes the loops."""
        self.pipeline = pipeline
        self.desc = desc
        self.prompt = prompt
        self.total_blocks = total_blocks

    @property
    def session_desc(self) -> SessionDesc:
        """Return the description used to configure the window."""
        return self.desc

    def init(self) -> None:
        """Register independent model and UI state."""
        model_loop = self.register_model_loop(
            TextToVideoLoop,
            state=TextToVideoState(
                self.pipeline, self.desc, self.total_blocks, self.prompt
            ),
        )
        self.register_ui_loop(
            MinimalT2VUILoop,
            state=MinimalT2VUIState(model_loop, self.prompt),
            width=self.desc.video_width,
            height=self.desc.video_height,
        )

Model loop

IModelLoop governs all model logic and will be where we run our model inference:

MinimalT2VModelState is the type of ILoop.state for our IModelLoop.
step() is called for each loop of our IModelLoop.
- pipeline.initialize_cache() is called to initialize our model's cache.
- pipeline.generate() is called to generate a chunk of frames given our current model's state (pipeline).
- pipeline.finalize() is called to finalize the generated chunk & produce metrics.
reset() is called when the session is reset.
close() is called when the session is closed.

# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
# SPDX-License-Identifier: Apache-2.0

"""Prompt and autoregressive cache owned by the tutorial model loop."""

from dataclasses import dataclass
from typing import Any

from flashdreams.api_v2.loop import IModelLoop
from flashdreams.runtime_v2.session_desc import SessionDesc
from flashdreams.runtime_v2.step_result import StepResult
from flashdreams.runtime_v2.user_input_events import UserInputEvents


@dataclass
class TextToVideoState:
    """Mutable rollout state, accessed on the model thread."""

    pipeline: Any
    """Text-conditioned pipeline shared across sessions."""
    desc: SessionDesc
    """Pixel dimensions and frame layout requested by the runtime."""
    total_blocks: int
    """Maximum blocks the selected model supports per prompt."""
    prompt: str = ""
    """Current prompt; empty text pauses generation."""
    cache: Any = None
    """Per-prompt cache, created lazily on the model thread."""
    index: int = 0
    """Next autoregressive block, independent of the runtime step index."""

    def set_prompt(self, prompt: str) -> None:
        """Start a fresh rollout or pause generation for empty text."""
        self.prompt = prompt.strip()
        self.cache = None
        self.index = 0


class TextToVideoLoop(IModelLoop[TextToVideoState]):
    """Generate bounded rollouts while waiting for further UI prompts."""

    def step(self, step_index: int, events: UserInputEvents) -> list[StepResult]:
        """Generate and finalize one block, or wait for another prompt."""
        state = self.state
        if not state.prompt or state.index >= state.total_blocks:
            return []
        if state.cache is None:
            ratio = state.pipeline.decoder.spatial_compression_ratio
            state.cache = state.pipeline.initialize_cache(
                text=[state.prompt],
                height=state.desc.video_height // ratio,
                width=state.desc.video_width // ratio,
            )
        frames = state.pipeline.generate(
            autoregressive_index=state.index, cache=state.cache
        )
        metrics = state.pipeline.finalize(
            autoregressive_index=state.index, cache=state.cache
        )
        state.index += 1
        return [
            StepResult(
                step_index=step_index,
                output=frames.detach(),
                frame_count=int(frames.shape[0]),
                output_layout=state.desc.output_layout,
                metrics=metrics,
            )
        ]

    def reset(self) -> None:
        """Restart the current prompt after a runtime reset."""
        self.state.set_prompt(self.state.prompt)

    def close(self) -> None:
        """Release the per-prompt cache."""
        self.state.cache = None

UI loop

IUILoop governs all UI logic and will be where we handle user-interaction:

MinimalT2VUIState is the type of ILoop.state for our IUILoop.
invoke_async(self.state.model_loop, lambda...) is called when the user submits a new prompt (via imgui button) to submit a lambda to run on our IModelLoop thread before the next new IModelLoop.step().
step_ui(...) is called for each loop of our IUILoop. It contains imgui state when IUILoop inherits from ImGuiUILoop.

# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
# SPDX-License-Identifier: Apache-2.0

"""ImGui prompt controls sending messages to the tutorial model loop."""

from dataclasses import dataclass
from typing import TYPE_CHECKING, Any

from torch import Tensor

from flashdreams.api_v2.loop import IModelLoop, invoke_async
from flashdreams.runtime_v2.imgui_ui_loop import ImGuiUILoop
from flashdreams.runtime_v2.user_input_events import UserInputEvents

if TYPE_CHECKING:
    from .model_loop import TextToVideoState


@dataclass
class MinimalT2VUIState:
    """Editable prompt owned by the UI thread."""

    model_loop: IModelLoop["TextToVideoState"]
    """Destination of prompt update messages."""
    text: str = ""
    """Text currently displayed in the prompt field."""


class MinimalT2VUILoop(ImGuiUILoop[MinimalT2VUIState]):
    """Send prompt updates without directly mutating model state."""

    def step_ui(
        self, imgui: Any, step_index: int, events: UserInputEvents
    ) -> Tensor | None:
        """Draw prompt controls over the latest generated frame."""
        imgui.begin("Minimal T2V")
        try:
            _, self.state.text = imgui.input_text("Prompt", self.state.text)
            if imgui.button("Send"):
                invoke_async(
                    self.state.model_loop,
                    lambda state, text=self.state.text: state.set_prompt(text),
                )
        finally:
            imgui.end()
        return self.presented_model_frame()