> For the complete documentation index, see [llms.txt](https://developer.fotokite.com/llms.txt). Markdown versions of documentation pages are available by appending `.md` to page URLs; this page is available as [Markdown](https://developer.fotokite.com/code-examples/video-streams.md).

# Video Streams

This section describes how to work with the system’s video streams. The streams use the RTSP protocol and can be converted to other formats as needed. Tools such as GStreamer or FFmpeg can efficiently pipe the source stream into different formats.

For more details, see [Video streams](/fotokite-api/video-streams.md) or the [GitHub repository](https://github.com/fotokite/fotokite-api/tree/main/example/).

### List video streams

Lists all active video streams. This data is generally stable and changes only when a stream’s status changes (e.g., on/off).

```python
def list_videostreams(access_token: str = "") -> VideoStreamMessage:
    """Fetches the list of available video streams.

    Args:
        access_token: Bearer token for API authentication. Defaults to "".

    Returns:
        Parsed JSON response containing video stream information if successful.
        An empty dictionary if an error occurs.

    Raises:
        requests.HTTPError: If the response status code is not 200.
        Exception: For any other errors during the request.
    """
    try:
        response: requests.Response = requests.get(
            f"{BASE_REST_API_URL}/videostreams",
            headers={"Authorization": f"Bearer {access_token}"},
        )
        if response.status_code == 200:
            return cast(VideoStreamMessage, response.json())
        else:
            response.raise_for_status()
    except Exception as e:
        logging.error(f"Error fetching video stream info: {e}")

    return {}
```

### Change zoom level

Certain aspects of the video streams can be adjusted, such as zoom level and thermal color palette. These adjustments are stream-specific, and not all commands apply to all streams—for example, the zoom command applies only to the zoom video stream. Ensure that the correct stream identifier is used as specified above.

The example demonstrates how to change the zoom level.

```python
def zoom(
    access_token: str = "", stream_id: str = "zoom", zoom_level: float = 1.0
) -> bool:
    """Adjusts the zoom level of a specified video stream.

    Args:
        access_token: Bearer token for API authentication.
        stream_id: Identifier of the video stream to zoom. Defaults to "zoom".
        zoom_level: Desired zoom level. Defaults to 1.0.

    Returns:
        True if the zoom level was successfully changed, False otherwise.

    Raises:
        requests.HTTPError: If the API request fails with a non-200 status code.
    """
    try:
        # Tries to retrieve handoff control before sending a command. If handoff retrieval fails then it will raise an exception.
        retrieve_handoff(access_token)
        
        response: requests.Response = requests.post(
            f"{BASE_REST_API_URL}/videostreams/{stream_id}/control/zoom",
            json={"zoom_level": zoom_level},
            headers={"Authorization": f"Bearer {access_token}"},
        )
        if response.status_code == 200:
            logging.info("Zoom level changed to %s", zoom_level)
            return True
        else:
            response.raise_for_status()
    except Exception as e:
        logging.error(f"Error changing zoom level: {e}")

    return False
```

The example demonstrates how to change the thermal color palette of the thermal video stream.

```python
def set_palette(
    access_token: str = "", stream_id: str = "thermal", palette: str = "Rainbow"
) -> bool:
    """Sets the thermal color palette for a specified video stream.

    Args:
        access_token: The access token for authentication. Defaults to an empty string.
        stream_id: The ID of the video stream to modify. Defaults to "thermal".
        palette: The name of the thermal color palette to set. Defaults to "Rainbow".

    Returns:
        True if the palette was successfully changed, False otherwise.

    Raises:
        requests.HTTPError: If the HTTP request fails with a non-200 status code.
    """
    try:
        # Tries to retrieve handoff control before sending a command. If handoff retrieval fails then it will raise an exception.
        retrieve_handoff(access_token)
        
        response: requests.Response = requests.post(
            f"{BASE_REST_API_URL}/videostreams/{stream_id}/control/palette",
            json={"thermal_color_palette": palette},
            headers={"Authorization": f"Bearer {access_token}"},
        )
        if response.status_code == 200:
            logging.info("Palette changed to %s", palette)
            return True
        else:
            response.raise_for_status()
    except Exception as e:
        logging.error(f"Error changing zoom level: {e}")

    return False
```

### GStreamer - HLS

This example demonstrates how to consume thermal and color RTSP video streams, create a GStreamer pipeline, and serve the resulting HLS streams over HTTP.

```python
import http.server
import logging
import os
import signal
import socketserver
import subprocess
import sys
import threading
from typing import Any, List, Tuple

logging.basicConfig(
    level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s"
)

""" CORS support for browsers to access HLS streams """


class CORSRequestHandler(http.server.SimpleHTTPRequestHandler):
    def end_headers(self) -> None:
        self.send_header("Access-Control-Allow-Origin", "*")
        self.send_header("Access-Control-Allow-Methods", "GET, OPTIONS")
        self.send_header(
            "Access-Control-Allow-Headers", "Range, Content-Type, Origin, Accept"
        )
        super().end_headers()

    def do_OPTIONS(self) -> None:
        self.send_response(200, "ok")
        self.end_headers()


class ReusableTCPServer(socketserver.TCPServer):
    allow_reuse_address = True


def start_gstreamer(rtsp_url: str, hls_dir: str) -> subprocess.Popen[bytes]:
    """Starts a GStreamer pipeline to stream RTSP video and convert it to HLS segments.

    Args:
        rtsp_url: The RTSP URL of the video stream source.
        hls_dir: The directory where HLS segments and playlist will be stored.

    Returns:
        The process running the GStreamer pipeline.
    """
    playlist_path: str = os.path.join(hls_dir, "playlist.m3u8")

    if not os.path.exists(hls_dir):
        os.makedirs(hls_dir)

    cmd: List[str] = [
        "gst-launch-1.0",
        "rtspsrc",
        f"location={rtsp_url}",
        "latency=0",
        "!",
        "rtph264depay",
        "!",
        "h264parse",
        "!",
        "mpegtsmux",
        "!",
        f"hlssink",
        f"location={hls_dir}/segment%05d.ts",
        f"playlist-location={playlist_path}",
        "target-duration=2",
        "max-files=5",
    ]

    process = subprocess.Popen(
        cmd, stdout=sys.stdout, stderr=sys.stderr, preexec_fn=os.setsid
    )
    logging.info(f"Started GStreamer for {rtsp_url}")
    return process


def start_http_server(
    hls_dir: str, port: int
) -> Tuple[threading.Thread, socketserver.TCPServer]:
    """Starts an HTTP server to serve files from the specified HLS directory with CORS support.

    Args:
        hls_dir: Path to the directory containing HLS files to be served.
        port: Port number on which the HTTP server will listen.

    Returns:
        A tuple containing the thread running the server and the server instance itself.
    """
    os.chdir(hls_dir)

    httpd: ReusableTCPServer = ReusableTCPServer(("", port), CORSRequestHandler)

    def serve() -> None:
        logging.info(f"Serving HLS on http://localhost:{port}/playlist.m3u8")
        try:
            httpd.serve_forever()
        except Exception as e:
            logging.error(f"HTTP server error: {e}")

    thread = threading.Thread(target=serve, daemon=True)
    thread.start()
    return thread, httpd


def main() -> None:
    """Starts GStreamer processes and HTTP servers for multiple RTSP streams, manages their lifecycles, and handles graceful shutdown on SIGINT or SIGTERM.

    The function:
    - Configures multiple RTSP streams and corresponding HLS output directories and HTTP ports.
    - Launches GStreamer processes to convert RTSP streams to HLS.
    - Starts HTTP servers to serve the HLS streams.
    - Registers signal handlers to gracefully terminate all processes and servers on shutdown signals.
    - Waits for all server threads to finish before exiting.

    """

    # Mapping of RTSP streams to HLS directories and ports
    configs: List[Tuple[str, str, str, int]] = [
        ("Thermal", "rtsp://192.168.2.100:5012/video", "/tmp/hlsThermal", 8082),
        ("Color", "rtsp://192.168.2.100:5010/video", "/tmp/hlsColor", 8083),
    ]

    processes: List[subprocess.Popen[bytes]] = []
    servers: List[socketserver.TCPServer] = []
    threads: List[threading.Thread] = []

    # Start GStreamer processes and HTTP servers for each RTSP stream
    for _, rtsp_url, hls_dir, port in configs:
        proc = start_gstreamer(rtsp_url, hls_dir)
        thread, httpd = start_http_server(hls_dir, port)
        processes.append(proc)
        servers.append(httpd)
        threads.append(thread)

    # Handle graceful shutdown on SIGINT or SIGTERM
    def shutdown(signum: int, frame: Any) -> None:
        logging.info("Received shutdown signal, stopping everything...")

        # Stop GStreamer processes
        for proc in processes:
            try:
                pgid: int = os.getpgid(proc.pid)
                os.killpg(pgid, signal.SIGKILL)
                logging.info(f"Stopped GStreamer process group {pgid}")
            except Exception as e:
                logging.error(f"Error stopping GStreamer process: {e}")

        # Stop HTTP servers
        for httpd in servers:
            try:
                httpd.shutdown()
                logging.info("HTTP server shut down")
            except Exception as e:
                logging.error(f"Error shutting down server: {e}")

        sys.exit(0)

    signal.signal(signal.SIGINT, shutdown)
    signal.signal(signal.SIGTERM, shutdown)

    # Wait for threads to finish (they will stop after shutdown)
    try:
        for thread in threads:
            thread.join()
    except KeyboardInterrupt:
        shutdown(signal.SIGINT, None)


if __name__ == "__main__":
    main()
```

### FFMPEG - HLS

A similar example using FFmpeg for the pipeline can be found in the [GitHub repository](https://github.com/fotokite/fotokite-api/blob/main/example/python/fotokite_api/streamer/ffmpeg.py).


---

# Agent Instructions
This documentation is published with GitBook. GitBook is the documentation platform designed so that both humans and AI agents can read, navigate, and reason over technical content effectively. Learn more at gitbook.com.

## Querying This Documentation
If you need additional information that is not directly available in this page, you can query the documentation dynamically by asking a question.

Perform an HTTP GET request on the current page URL with the `ask` query parameter, and the optional `goal` query parameter:

```
GET https://developer.fotokite.com/code-examples/video-streams.md?ask=<question>&goal=<endgoal>
```

`ask` is the immediate question: it should be specific, self-contained, and written in natural language.
`goal` is optional and describes the broader end goal you are ultimately trying to accomplish on behalf of the user. GitBook uses it to tailor the answer towards what is most useful for that goal.

The response will contain a direct answer to the question and relevant excerpts and sources from the documentation.

Use this mechanism when the answer is not explicitly present in the current page, you need clarification or additional context, or you want to retrieve related documentation sections.