FPS Control Async Flow

videorate is the GStreamer element used to normalize or change a stream's frame rate. Place it before a capsfilter, then set the capsfilter to the target rate, for example video/x-raw,framerate=15/1. videorate will drop frames when lowering FPS and duplicate frames when raising FPS so the downstream pipeline receives buffers at the requested cadence.

Simple pipeline that limits a test source to 10 FPS:

gst-launch-1.0 videotestsrc is-live=true ! \
  video/x-raw,framerate=30/1 ! \
  videorate ! \
  video/x-raw,framerate=10/1 ! \
  videoconvert ! \
  autovideosink

The main idea is that FastAPI and GStreamer run in different execution contexts. The API side receives HTTP requests asynchronously. The GStreamer side owns the pipeline and runs on its own GLib thread. They communicate with commands, futures, and a thread-safe queue.

flowchart LR
    subgraph API["FastAPI / asyncio side"]
        Req["HTTP request<br/>GET /status or POST /fps"]
        Route["FastAPI route<br/>status() / set_fps()"]
        Await["await_fps_command()<br/>wait for Future"]
        Resp["HTTP JSON response"]
    end

    subgraph Bridge["Thread-safe bridge"]
        Submit["fps_controller.submit()"]
        Queue["Command queue"]
        Future["Future result"]
    end

    subgraph Gst["GStreamer / GLib thread"]
        Loop["GLib MainLoop"]
        Process["_process_commands()"]
        Handle["_handle_command()"]
        Pipeline["GStreamer pipeline<br/>capsfilter changes FPS"]
    end

    Req --> Route --> Await --> Submit --> Queue
    Submit --> Future
    Queue -->|"invoke_full()"| Loop --> Process --> Handle --> Pipeline
    Pipeline -->|"set_result() / set_exception()"| Future
    Future --> Await --> Resp

What This Means

fps_app.py should stay responsive to HTTP requests. It does not touch the GStreamer pipeline directly. When a request arrives, it creates a command and waits asynchronously for the result.

fps_control.py owns the GStreamer objects. Its controller thread runs a GLib main loop, receives queued commands, and applies changes to the pipeline from that thread.

The queue and future are the bridge:

• submit() puts a command into the queue.
• GLib.MainContext.invoke_full() wakes the GStreamer thread.
• _process_commands() handles the command.
• The command result is written back to the Future.
• FastAPI awaits that future and returns JSON to the client.

For POST /fps, the command eventually calls _set_fps(), validates the FPS, and updates the capsfilter caps to:

video/x-raw,framerate=<fps>/1

For GET /status, the command returns the current controller state.

---

Code

Pipeline

self.pipeline = Gst.parse_launch(
        "v4l2src device=/dev/video0 "
        "! video/x-raw,format=YUY2,width=640,height=480,framerate=30/1 "
        "! videorate name=rate drop-only=true "
        "! capsfilter name=fps_caps "
        "! videoconvert "
        "! fpsdisplaysink sync=false "
    )

The pipeline creates a named capsfilter:

"! videorate name=rate drop-only=true "
"! capsfilter name=fps_caps "

After Gst.parse_launch() builds the pipeline, the controller keeps a reference to that element:

self.capsfilter = self.pipeline.get_by_name("fps_caps")

When the API asks for a new FPS, _set_fps() first validates the requested value, then calls _apply_fps_caps():

def _set_fps(self, fps: int):
    fps = self._validate_fps(fps)
    if self.capsfilter is None:
        raise RuntimeError("Pipeline is not running")

    self._apply_fps_caps(fps)
    self.fps = fps
    return {"ok": True, **self._status_fields()}

_apply_fps_caps() is the part that actually edits the caps. It creates a new Gst.Caps object from the requested frame rate and writes it to the capsfilter's caps property:

def _apply_fps_caps(self, fps: int):
    caps = Gst.Caps.from_string(f"video/x-raw,framerate={fps}/1")
    self.capsfilter.set_property("caps", caps)

Because videorate is directly before the capsfilter, it adapts the incoming stream to satisfy the new downstream caps. With drop-only=true, this example is intended for reducing FPS by dropping frames rather than creating extra frames.

Demo files:

• fps_control.py contains the GStreamer controller. It creates the pipeline, owns the GLib main loop, validates FPS changes, and updates the capsfilter.
• fps_app.py contains the FastAPI application. It exposes the status and FPS endpoints, serves the HTML page, and forwards commands to the controller.
• static/fps.html contains the browser UI. It shows the current controller status and sends FPS update requests to the FastAPI server.

fps_control.py

"""
v4l2-ctl -d /dev/video0 --list-formats-ext
"""
import queue
import threading
from concurrent.futures import Future
from dataclasses import dataclass
from typing import Any

import gi

gi.require_version("Gst", "1.0")
gi.require_version("GLib", "2.0")

from gi.repository import Gst, GLib


Gst.init(None)


CMD_INIT = "init"
CMD_SHUTDOWN = "shutdown"
CMD_SET_FPS = "set_fps"
CMD_STATUS = "status"

SUPPORTED_FPS = (1, 5, 10, 20, 30)
DEFAULT_FPS = 30


@dataclass
class Command:
    name: str
    args: dict[str, Any]
    future: Future


class FpsController:
    def __init__(self):
        self.commands: queue.Queue[Command] = queue.Queue()

        self.context = GLib.MainContext()
        self.loop = GLib.MainLoop.new(self.context, False)
        self.thread = threading.Thread(target=self._thread_main, daemon=True)

        self.pipeline = None
        self.capsfilter = None
        self.fps = DEFAULT_FPS
        self.started = False

    def start(self):
        self.thread.start()
        return self.call_sync(CMD_INIT, timeout=5)

    def stop(self):
        try:
            self.call_sync(CMD_SHUTDOWN, timeout=5)
        finally:
            self._invoke(lambda: self.loop.quit() or False)
            self.thread.join(timeout=2)

    def submit(self, name: str, args: dict | None = None) -> Future:
        future = Future()
        cmd = Command(name=name, args=args or {}, future=future)
        self.commands.put(cmd)
        self._invoke(self._process_commands)
        return future

    def call_sync(self, name: str, args: dict | None = None, timeout: float = 2):
        future = self.submit(name, args)
        return future.result(timeout=timeout)

    def _invoke(self, callback):
        self.context.invoke_full(GLib.PRIORITY_DEFAULT, callback)

    def _thread_main(self):
        self.context.push_thread_default()
        try:
            self.loop.run()
        finally:
            self.context.pop_thread_default()

    def _process_commands(self):
        while True:
            try:
                cmd = self.commands.get_nowait()
            except queue.Empty:
                break

            try:
                result = self._handle_command(cmd)
                cmd.future.set_result(result)
            except Exception as exc:
                cmd.future.set_exception(exc)

        return False

    def _handle_command(self, cmd: Command):
        name = cmd.name
        args = cmd.args

        if name == CMD_INIT:
            return self._init_pipeline()
        if name == CMD_SHUTDOWN:
            return self._shutdown_pipeline()
        if name == CMD_SET_FPS:
            return self._set_fps(args["fps"])
        if name == CMD_STATUS:
            return self._status()

        raise RuntimeError(f"Unknown command: {name}")

    def _init_pipeline(self):
        if self.started:
            return {"ok": True, "already_started": True, **self._status_fields()}

        self.pipeline = Gst.parse_launch(
            "v4l2src device=/dev/video0 "
            "! video/x-raw,format=YUY2,width=640,height=480,framerate=30/1 "
            "! videorate name=rate drop-only=true "
            "! capsfilter name=fps_caps "
            "! videoconvert "
            "! fpsdisplaysink sync=false "
        )

        self.capsfilter = self.pipeline.get_by_name("fps_caps")
        if self.capsfilter is None:
            raise RuntimeError("Could not find fps capsfilter")

        self._apply_fps_caps(self.fps)

        bus = self.pipeline.get_bus()
        bus.add_signal_watch()
        bus.connect("message", self._on_bus_message)

        ret = self.pipeline.set_state(Gst.State.PLAYING)
        if ret == Gst.StateChangeReturn.FAILURE:
            raise RuntimeError("Failed to start FPS control pipeline")

        self.started = True
        return {"ok": True, **self._status_fields()}

    def _shutdown_pipeline(self):
        if self.pipeline is not None:
            self.pipeline.set_state(Gst.State.NULL)
            self.pipeline = None
            self.capsfilter = None

        self.started = False
        return {"ok": True, **self._status_fields()}

    def _set_fps(self, fps: int):
        fps = self._validate_fps(fps)
        if self.capsfilter is None:
            raise RuntimeError("Pipeline is not running")

        self._apply_fps_caps(fps)
        self.fps = fps
        return {"ok": True, **self._status_fields()}

    def _validate_fps(self, fps: int):
        if fps not in SUPPORTED_FPS:
            raise ValueError(f"FPS must be one of: {', '.join(map(str, SUPPORTED_FPS))}")
        return fps

    def _apply_fps_caps(self, fps: int):
        caps = Gst.Caps.from_string(f"video/x-raw,framerate={fps}/1")
        self.capsfilter.set_property("caps", caps)

    def _status(self):
        return {"ok": True, **self._status_fields()}

    def _status_fields(self):
        return {
            "started": self.started,
            "fps": self.fps,
            "supported_fps": list(SUPPORTED_FPS),
            "has_pipeline": self.pipeline is not None,
            "has_capsfilter": self.capsfilter is not None,
        }

    def _on_bus_message(self, bus, message):
        msg_type = message.type

        if msg_type == Gst.MessageType.ERROR:
            err, debug = message.parse_error()
            print(f"GStreamer error: {err}")
            if debug:
                print(f"GStreamer debug: {debug}")

        elif msg_type == Gst.MessageType.EOS:
            print("GStreamer EOS")

fps_app.py

import asyncio
from contextlib import asynccontextmanager
from pathlib import Path

import uvicorn
from fastapi import FastAPI, HTTPException
from fastapi.responses import FileResponse
from pydantic import BaseModel

try:
    from .fps_control import (
        CMD_SET_FPS,
        CMD_STATUS,
        FpsController,
    )
except ImportError:
    from fps_control import (
        CMD_SET_FPS,
        CMD_STATUS,
        FpsController,
    )


STATIC_DIR = Path(__file__).with_name("static")

fps_controller = FpsController()


@asynccontextmanager
async def lifespan(app: FastAPI):
    fps_controller.start()
    try:
        yield
    finally:
        fps_controller.stop()


app = FastAPI(lifespan=lifespan)


class FpsRequest(BaseModel):
    fps: int


async def await_fps_command(name: str, args: dict | None = None):
    try:
        future = fps_controller.submit(name, args)
        return await asyncio.wait_for(
            asyncio.wrap_future(future),
            timeout=3,
        )
    except asyncio.TimeoutError:
        raise HTTPException(
            status_code=504,
            detail=f"GStreamer FPS command timed out: {name}",
        )
    except ValueError as exc:
        raise HTTPException(
            status_code=400,
            detail=str(exc),
        )
    except Exception as exc:
        raise HTTPException(
            status_code=500,
            detail=str(exc),
        )


@app.get("/")
async def index():
    return FileResponse(STATIC_DIR / "fps.html")


@app.get("/status")
async def status():
    return await await_fps_command(CMD_STATUS)


@app.post("/fps")
async def set_fps(req: FpsRequest):
    return await await_fps_command(
        CMD_SET_FPS,
        {"fps": req.fps},
    )


def main() -> None:
    uvicorn.run(app, host="0.0.0.0", port=8002)


if __name__ == "__main__":
    main()

static/fps.html

<!doctype html>
<html lang="en">
<head>
  <meta charset="utf-8">
  <meta name="viewport" content="width=device-width, initial-scale=1">
  <title>BT FPS Control</title>
  <style>
    :root {
      font-family: system-ui, -apple-system, BlinkMacSystemFont, "Segoe UI", sans-serif;
      color-scheme: light dark;
    }

    body {
      margin: 0;
      padding: 24px;
      background: #101418;
      color: #edf2f7;
    }

    main {
      max-width: 720px;
      margin: 0 auto;
    }

    h1 {
      margin: 0 0 20px;
      font-size: 28px;
      letter-spacing: 0;
    }

    section {
      margin-bottom: 16px;
      padding: 16px;
      border: 1px solid #2d3748;
      border-radius: 8px;
      background: #151b22;
    }

    .fps-buttons {
      display: flex;
      flex-wrap: wrap;
      gap: 10px;
    }

    button {
      min-width: 96px;
      padding: 9px 14px;
      border: 1px solid #3182ce;
      border-radius: 6px;
      background: #2b6cb0;
      color: #edf2f7;
      cursor: pointer;
      font-size: 14px;
    }

    button.active {
      border-color: #68d391;
      background: #2f855a;
    }

    button:disabled {
      opacity: 0.5;
      cursor: not-allowed;
    }

    dl {
      display: grid;
      grid-template-columns: 140px 1fr;
      gap: 8px 12px;
      margin: 0;
    }

    dt {
      color: #a0aec0;
    }

    dd {
      margin: 0;
      word-break: break-word;
    }

    #message {
      min-height: 22px;
      color: #90cdf4;
    }

    #message.error {
      color: #feb2b2;
    }

    @media (max-width: 520px) {
      body {
        padding: 16px;
      }

      dl {
        grid-template-columns: 1fr;
      }

      button {
        width: 100%;
      }
    }
  </style>
</head>
<body>
  <main>
    <h1>BT FPS Control</h1>

    <section class="fps-buttons">
      <button data-fps="1">1 FPS</button>
      <button data-fps="5">5 FPS</button>
      <button data-fps="10">10 FPS</button>
      <button data-fps="20">20 FPS</button>
      <button data-fps="30">30 FPS</button>
    </section>

    <section>
      <dl>
        <dt>Pipeline</dt>
        <dd id="started">unknown</dd>
        <dt>Current FPS</dt>
        <dd id="fps">unknown</dd>
        <dt>Pipeline object</dt>
        <dd id="pipeline">unknown</dd>
        <dt>Caps filter</dt>
        <dd id="capsfilter">unknown</dd>
      </dl>
    </section>

    <section id="message"></section>
  </main>

  <script>
    const buttons = Array.from(document.querySelectorAll("[data-fps]"));
    const startedEl = document.getElementById("started");
    const fpsEl = document.getElementById("fps");
    const pipelineEl = document.getElementById("pipeline");
    const capsfilterEl = document.getElementById("capsfilter");
    const messageEl = document.getElementById("message");

    function yesNo(value) {
      return value ? "yes" : "no";
    }

    function setMessage(text, isError = false) {
      messageEl.textContent = text || "";
      messageEl.className = isError ? "error" : "";
    }

    async function requestJson(url, options = {}) {
      const response = await fetch(url, options);
      const body = await response.json().catch(() => ({}));
      if (!response.ok) {
        throw new Error(body.detail || `HTTP ${response.status}`);
      }
      return body;
    }

    function renderStatus(status) {
      startedEl.textContent = yesNo(status.started);
      fpsEl.textContent = `${status.fps} FPS`;
      pipelineEl.textContent = status.has_pipeline ? "ready" : "missing";
      capsfilterEl.textContent = status.has_capsfilter ? "ready" : "missing";

      for (const button of buttons) {
        const fps = Number(button.dataset.fps);
        button.disabled = !status.started || !status.has_capsfilter;
        button.classList.toggle("active", fps === status.fps);
      }
    }

    async function refreshStatus() {
      const status = await requestJson("/status");
      renderStatus(status);
      return status;
    }

    for (const button of buttons) {
      button.addEventListener("click", async () => {
        try {
          const fps = Number(button.dataset.fps);
          const status = await requestJson("/fps", {
            method: "POST",
            headers: {"content-type": "application/json"},
            body: JSON.stringify({fps}),
          });
          renderStatus(status);
          setMessage(`FPS set to ${status.fps}`);
        } catch (err) {
          setMessage(err.message, true);
          await refreshStatus().catch(() => {});
        }
      });
    }

    refreshStatus().catch((err) => setMessage(err.message, true));
    setInterval(() => {
      refreshStatus().catch((err) => setMessage(err.message, true));
    }, 2000);
  </script>
</body>
</html>