AidStream Python API

AidStream Python API Reference

Overview

AidStream provides Python bindings (via pybind11) that allow developers to call AidStream SDK core functionality from Python. The Python binding module is named aidstreamgst.

warning

The Python bindings currently expose only a subset of the C++ API. Some advanced features (such as AidStreamEngine, Splice API, rolling recording, etc.) are not yet available in Python.

Installation and Import

import aidstreamgst

Enumerations

LogLevel

Member	Description
LogLevel.SINFO	Informational log level
LogLevel.SWARNING	Warning log level
LogLevel.SERROR	Error log level
LogLevel.SDEBUG	Debug log level
LogLevel.SOFF	Logging disabled

StreamType

warning

The Python bindings expose only 3 of the 5 StreamType values. RTMPSINK and MP4 types are currently unsupported.

Member	Description
StreamType.EMPTY	No output (fakesink)
StreamType.WAYLANDSINK	Wayland display output
StreamType.RTSPSINK	RTSP streaming output

Image Class

The Image class encapsulates frame data and metadata received from the GStreamer pipeline callback.

Property	Type	Description
width	int	Image width
height	int	Image height
size	int	Buffer size occupied by the image data (bytes)
fps	float	Frame rate (C++ type is double; in Python it appears as float)
pipe_status	int	Pipeline status

Method	Description
to_bytes()	Copies the current frame data into a Python bytes object (thread-safe)
as_memoryview()	Returns a zero-copy memoryview of the current frame data (does not hold the underlying memory; only safe within the callback)

def my_callback(img):
    # Access frame dimensions and frame rate
    print(f"width: {img.width}, height: {img.height}, fps: {img.fps}")

    # Get image data as bytes
    raw_bytes = img.to_bytes()

    # Or get a zero-copy memory view (only valid inside the callback)
    mem = img.as_memoryview()

    return 0

mat_image Class

mat_image wraps an OpenCV cv::Mat object for Python. It supports the buffer protocol, so it can be directly converted to a numpy array.

Method	Description
is_empty()	Returns True if the image is empty

import numpy as np

def my_callback(img):
    mat = aidstreamgst.get_data(img.height, img.width, img.to_bytes())
    if mat.is_empty():
        print("Empty image")
        return -1
    arr = np.array(mat, copy=False)
    # Process arr...
    return 0

API Functions

set_log_level

Sets the log level.

Parameter	Type	Description
log_level	LogLevel	Log level enum value

aidstreamgst.set_log_level(aidstreamgst.LogLevel.SINFO)

clogger

Creates log files and sets the log path and filename prefix.

Parameter	Type	Description
destinationPath	str	Log file path and prefix, e.g., "./aidclog_aidstream_"

aidstreamgst.clogger("./aidclog_aidstream_")

start_stream

Starts a stream by stream_id. This function is blocking. The callback is GIL-safe.

Parameter	Type	Description
stream_id	str	Stream ID defined in the aidstream-gst.conf configuration file
cb	Callable[[Image], int]	Callback function that receives an Image object and returns an int (0 for success)

def my_callback(img):
    print(f"Received frame: {img.width}x{img.height} at {img.fps}fps")
    return 0

# Start the stream (blocking call)
aidstreamgst.start_stream("stream1", my_callback)

shutdown

Stops the pipeline and releases resources. Equivalent to gstreamer_pipeline_shutdown() in C++.

Parameter	Type	Default	Description
stream_id	str	"all_stream"	The stream ID to stop. Pass a specific stream ID to stop a single stream, or "all_stream" to stop all streams.

# Stop all streams
aidstreamgst.shutdown()

# Stop a specific stream
aidstreamgst.shutdown("stream1")

get_data

Converts image data into an OpenCV Mat object.

Parameter	Type	Description
height	int	Image height
width	int	Image width
data	bytes	Raw image data (obtained via img.to_bytes())

mat = aidstreamgst.get_data(img.height, img.width, img.to_bytes())

get_stream_status

Returns the current running status of the stream.

Return Value	Description
1	Stream is running
0	Stream is idle
-1	Uninitialized

status = aidstreamgst.get_stream_status()
print(f"Stream status: {status}")

Complete Example

import aidstreamgst
import signal

running = True

def my_callback(img):
    """Callback invoked for each video frame."""
    print(f"Frame: {img.width}x{img.height}, fps: {img.fps:.1f}")
    # Get image bytes
    img_bytes = img.to_bytes()
    return 0

def main():
    # Set up logging
    aidstreamgst.clogger("./aidclog_aidstream_")
    aidstreamgst.set_log_level(aidstreamgst.LogLevel.SINFO)

    # Start stream using stream ID from the configuration file
    aidstreamgst.start_stream("1", my_callback)

if __name__ == "__main__":
    main()