--- title: mock_server ---

Mock streaming server — a frontend dev aid.

Boots the same FastAPI app the real streaming server uses, but backs it with :class:InProcessGpuPool wrapping a synthetic generator that emits pre-baked RGB frames. No GPU or model weights required.

Use cases:

Launch: python -m fastvideo.entrypoints.streaming.mock_server.

fastvideo.entrypoints.streaming.mock_server.ContinuationState

ContinuationState(kind: str, payload: dict[str, Any])

fastvideo.entrypoints.streaming.mock_server.GenerationRequest

GenerationRequest(prompt: str | list[str] | None = None, negative_prompt: str | None = None, inputs: InputConfig = InputConfig(), sampling: SamplingConfig = SamplingConfig(), runtime: RequestRuntimeConfig = RequestRuntimeConfig(), output: OutputConfig = OutputConfig(), stage_overrides: dict[str, Any] = dict(), state: ContinuationState | None = None, plan: GenerationPlan | None = None, extensions: dict[str, Any] = dict())

fastvideo.entrypoints.streaming.mock_server.GeneratorConfig

GeneratorConfig(model_path: str, revision: str | None = None, trust_remote_code: bool = False, engine: EngineConfig = EngineConfig(), pipeline: PipelineSelection = PipelineSelection())

fastvideo.entrypoints.streaming.mock_server.SamplingConfig

SamplingConfig(num_videos_per_prompt: int = 1, seed: int = 1024, max_sequence_length: int | None = None, num_frames: int = 125, height: int = 720, width: int = 1280, height_sr: int = 1072, width_sr: int = 1920, fps: int = 24, audio_start_in_s: float | None = None, audio_end_in_s: float | None = None, num_inference_steps: int = 50, num_inference_steps_sr: int = 50, guidance_scale: float = 1.0, batch_cfg: bool = False, guidance_scale_2: float | None = None, guidance_rescale: float = 0.0, true_cfg_scale: float | None = None, use_embedded_guidance: bool | None = None, boundary_ratio: float | None = None, sigmas: list[float] | None = None, pyramid_num_inference_steps_list: list[int] | None = None, history_sizes: list[int] | None = None, num_latent_frames_per_chunk: int = 9, keep_first_frame: bool = True, is_skip_first_chunk: bool = False, use_zero_init: bool = True, zero_steps: int = 1, is_amplify_first_chunk: bool = False)

fastvideo.entrypoints.streaming.mock_server.ServeConfig

ServeConfig(generator: GeneratorConfig, server: ServerConfig = ServerConfig(), default_request: GenerationRequest = GenerationRequest(), streaming: StreamingConfig | None = None)

Typed serve config loaded from fastvideo serve --config.

default_request is a full :class:GenerationRequest — the same type clients POST to /v1/videos. At request time the server merges it into the incoming body as the operator-pinned baseline.

Important nuance: only fields the operator explicitly wrote in the serve YAML/JSON count as defaults. Although the in-memory object is fully populated (schema defaults fill every unset field), the merge walks _fastvideo_explicit_paths — populated during parse — so unset fields are not forced onto requests. Per-request precedence:

body (client-explicit) > default_request (operator-explicit)
                       > hardcoded fallback (e.g. ``fps=24``)

See :func:fastvideo.api.compat.explicit_request_updates for the projection and entrypoints/openai/video_api.py::_build_generation_kwargs for the merge.

fastvideo.entrypoints.streaming.mock_server.StreamingConfig

StreamingConfig(session_timeout_seconds: int = 300, generation_segment_cap: int = 6, stream_mode: Literal['av_fmp4', 'legacy_jpeg'] = 'av_fmp4', warmup: WarmupConfig = WarmupConfig(), pool: GpuPoolConfig = GpuPoolConfig(), prompt: PromptEnhancerConfig = PromptEnhancerConfig(), safety: PromptSafetyConfig = PromptSafetyConfig())

fastvideo.entrypoints.streaming.mock_server.build_app

build_app(serve_config: ServeConfig, generator: _GeneratorProto | None = None, *, pool: GpuPool | None = None, session_store: SessionStore | None = None) -> FastAPI

Build the FastAPI app used by :func:run_server.

Exposed so tests can drive the WebSocket endpoint in-process via starlette.testclient.TestClient(app).websocket_connect(...).

Exactly one of generator (backed by :class:InProcessGpuPool) or pool (for the subprocess-backed production shape) must be given.

fastvideo.entrypoints.streaming.mock_server.MockGenerator

MockGenerator(sleep_ms: float = 0.0)

Generator stand-in that returns synthetic gradient frames.

Each call produces one segment worth of frames whose pixels vary by a constant derived from the request seed and segment index. Latency is configurable via sleep_ms so the caller can exercise slow- generate scenarios without spinning a GPU.

Source
@dataclass
class MockGenerator:
    """Generator stand-in that returns synthetic gradient frames.

    Each call produces one segment worth of frames whose pixels vary by
    a constant derived from the request seed and segment index. Latency
    is configurable via ``sleep_ms`` so the caller can exercise slow-
    generate scenarios without spinning a GPU.
    """

    sleep_ms: float = 0.0

    def generate(self, request: GenerationRequest) -> dict[str, Any]:
        if self.sleep_ms:
            time.sleep(self.sleep_ms / 1000.0)
        width = max(16, request.sampling.width)
        height = max(16, request.sampling.height)
        num_frames = max(1, request.sampling.num_frames)
        frames = [_gradient_frame(height, width, idx, seed=request.sampling.seed) for idx in range(num_frames)]
        state = ContinuationState(
            kind="ltx2.v1",
            payload={
                "schema_version": 1,
                "segment_index": 0,
                "source_prompt": request.prompt,
            },
        )
        return {
            "frames": frames,
            "audio_sample_rate": 24000,
            "state": state,
        }

fastvideo.entrypoints.streaming.mock_server.MockGenerator.generate

generate(request: GenerationRequest) -> dict[str, Any]
Source
def generate(self, request: GenerationRequest) -> dict[str, Any]:
    if self.sleep_ms:
        time.sleep(self.sleep_ms / 1000.0)
    width = max(16, request.sampling.width)
    height = max(16, request.sampling.height)
    num_frames = max(1, request.sampling.num_frames)
    frames = [_gradient_frame(height, width, idx, seed=request.sampling.seed) for idx in range(num_frames)]
    state = ContinuationState(
        kind="ltx2.v1",
        payload={
            "schema_version": 1,
            "segment_index": 0,
            "source_prompt": request.prompt,
        },
    )
    return {
        "frames": frames,
        "audio_sample_rate": 24000,
        "state": state,
    }

fastvideo.entrypoints.streaming.mock_server.build_mock_app

build_mock_app(*, sleep_ms: float = 0.0)

Build a FastAPI app backed by :class:MockGenerator.

Source
def build_mock_app(*, sleep_ms: float = 0.0):
    """Build a FastAPI app backed by :class:`MockGenerator`."""
    serve_config = ServeConfig(
        generator=GeneratorConfig(model_path="/models/mock"),
        streaming=StreamingConfig(
            session_timeout_seconds=120,
            generation_segment_cap=6,
        ),
    )
    serve_config.default_request.sampling = SamplingConfig(
        num_frames=24,
        height=256,
        width=256,
        fps=24,
        num_inference_steps=1,
    )
    return build_app(serve_config, MockGenerator(sleep_ms=sleep_ms))

fastvideo.entrypoints.streaming.mock_server.main

main() -> None
Source
def main() -> None:  # pragma: no cover - CLI entry
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--host", default="127.0.0.1")
    parser.add_argument("--port", type=int, default=8000)
    parser.add_argument(
        "--sleep-ms",
        type=float,
        default=0.0,
        help="Per-segment artificial latency for testing slow paths",
    )
    args = parser.parse_args()

    import uvicorn

    app = build_mock_app(sleep_ms=args.sleep_ms)
    uvicorn.run(app, host=args.host, port=args.port)