Python API Reference

Python bindings for MediaX video streaming.

Installation

Python bindings are installed to /usr/local/lib/python3/dist-packages/mediax/.

Add the path before importing:

import sys
sys.path.append('/usr/local/lib/python3/dist-packages')
import mediax

Module Structure

mediax/
├── rtp                    # Core RTP streaming
├── sap                    # SAP/SDP announcement
├── rtp_uncompressed_payloader    # Uncompressed transmit
├── rtp_uncompressed_depayloader  # Uncompressed receive
├── rtp_h264_payloader     # H.264 transmit (VAAPI)
├── rtp_h264_depayloader   # H.264 receive (VAAPI)
├── rtp_h265_payloader     # H.265 transmit (VAAPI)
├── rtp_h265_depayloader   # H.265 receive (VAAPI)
├── rtp_av1_payloader      # AV1 transmit
└── rtp_av1_depayloader    # AV1 receive

Stream Information

import mediax

# Create stream info
stream_info = mediax.rtp.StreamInformation()
stream_info.session_name = "python-stream"
stream_info.hostname = "239.192.1.1"
stream_info.port = 5004
stream_info.width = 1920
stream_info.height = 1080
stream_info.framerate = 30
stream_info.encoding = mediax.rtp.ColourspaceType_kColourspaceRgb24

Colorspace Types

mediax.rtp.ColourspaceType_kColourspaceRgb24      # 24-bit RGB
mediax.rtp.ColourspaceType_kColourspaceYuv422     # YCbCr 4:2:2
mediax.rtp.ColourspaceType_kColourspaceMono8      # 8-bit grayscale
mediax.rtp.ColourspaceType_kColourspaceMono16     # 16-bit grayscale
mediax.rtp.ColourspaceType_kColourspaceH264Part10 # H.264/AVC
mediax.rtp.ColourspaceType_kColourspaceH265       # H.265/HEVC
mediax.rtp.ColourspaceType_kColourspaceAv1        # AV1
mediax.rtp.ColourspaceType_kColourspaceJpeg2000   # JPEG 2000

Uncompressed Streaming

Transmit

import sys
sys.path.append('/usr/local/lib/python3/dist-packages')
import mediax
import numpy as np

# Configure stream
stream_info = mediax.rtp.StreamInformation()
stream_info.session_name = "rgb-stream"
stream_info.hostname = "239.192.1.1"
stream_info.port = 5004
stream_info.width = 640
stream_info.height = 480
stream_info.framerate = 30
stream_info.encoding = mediax.rtp.ColourspaceType_kColourspaceRgb24

# Create payloader
payloader = mediax.rtp_uncompressed_payloader.RtpUncompressedPayloader()
payloader.SetStreamInfo(stream_info)
payloader.Open()

# Create test frame (RGB24)
frame = np.zeros((480, 640, 3), dtype=np.uint8)
frame[:, :, 0] = 255  # Red

# Transmit
payloader.Transmit(frame.tobytes(), True)

# Cleanup
payloader.Close()

Receive

import sys
sys.path.append('/usr/local/lib/python3/dist-packages')
import mediax
import numpy as np

# Configure stream
stream_info = mediax.rtp.StreamInformation()
stream_info.hostname = "239.192.1.1"
stream_info.port = 5004
stream_info.width = 640
stream_info.height = 480
stream_info.encoding = mediax.rtp.ColourspaceType_kColourspaceRgb24

# Create depayloader
depayloader = mediax.rtp_uncompressed_depayloader.RtpUncompressedDepayloader()
depayloader.SetStreamInfo(stream_info)
depayloader.Open()
depayloader.Start()

# Receive frame
buffer = bytearray(640 * 480 * 3)
if depayloader.Receive(buffer, 1000):  # 1 second timeout
    frame = np.frombuffer(buffer, dtype=np.uint8).reshape((480, 640, 3))
    print(f"Received frame: {frame.shape}")

# Cleanup
depayloader.Stop()
depayloader.Close()

H.264 Streaming

Transmit

import sys
sys.path.append('/usr/local/lib/python3/dist-packages')
import mediax
import numpy as np

stream_info = mediax.rtp.StreamInformation()
stream_info.session_name = "h264-stream"
stream_info.hostname = "239.192.1.1"
stream_info.port = 5004
stream_info.width = 1920
stream_info.height = 1080
stream_info.framerate = 30
stream_info.encoding = mediax.rtp.ColourspaceType_kColourspaceH264Part10

payloader = mediax.rtp_h264_payloader.RtpH264GstVaapiPayloader()
payloader.SetStreamInfo(stream_info)
payloader.Open()

# Send RGB frames - encoder handles compression
frame = np.zeros((1080, 1920, 3), dtype=np.uint8)
payloader.Transmit(frame.tobytes(), True)

payloader.Close()

Receive

import sys
sys.path.append('/usr/local/lib/python3/dist-packages')
import mediax
import numpy as np

stream_info = mediax.rtp.StreamInformation()
stream_info.hostname = "239.192.1.1"
stream_info.port = 5004
stream_info.width = 1920
stream_info.height = 1080
stream_info.encoding = mediax.rtp.ColourspaceType_kColourspaceH264Part10

depayloader = mediax.rtp_h264_depayloader.RtpH264GstVaapiDepayloader()
depayloader.SetStreamInfo(stream_info)
depayloader.Open()
depayloader.Start()

# Receive decoded RGB frames
buffer = bytearray(1920 * 1080 * 3)
depayloader.Receive(buffer, 1000)

depayloader.Stop()
depayloader.Close()

SAP/SDP

Announcer

import sys
sys.path.append('/usr/local/lib/python3/dist-packages')
import mediax

stream_info = mediax.rtp.StreamInformation()
stream_info.session_name = "announced-stream"
stream_info.hostname = "239.192.1.1"
stream_info.port = 5004
stream_info.width = 1920
stream_info.height = 1080
stream_info.framerate = 30
stream_info.encoding = mediax.rtp.ColourspaceType_kColourspaceRgb24

announcer = mediax.sap.SapAnnouncer()
announcer.AddSapAnnouncement(stream_info)
announcer.Start()

# Announcer runs in background...
import time
time.sleep(60)

announcer.Stop()
announcer.DeleteAllSapAnnouncements()

Listener

import sys
sys.path.append('/usr/local/lib/python3/dist-packages')
import mediax
import time

listener = mediax.sap.SapListener()
listener.Start()

# Wait for announcements
time.sleep(2)

# Get discovered streams
streams = listener.GetSapAnnouncements()
for name, info in streams.items():
    print(f"Stream: {name}")
    print(f"  Address: {info.hostname}:{info.port}")
    print(f"  Resolution: {info.width}x{info.height}")
    print(f"  Framerate: {info.framerate}")

listener.Stop()

Working with NumPy

Buffer Sizes

import numpy as np

width, height = 1920, 1080

# RGB24: 3 bytes per pixel
rgb_buffer = np.zeros((height, width, 3), dtype=np.uint8)

# YUV422: 2 bytes per pixel (UYVY packed)
yuv_buffer = np.zeros((height, width, 2), dtype=np.uint8)

# Mono8: 1 byte per pixel
mono8_buffer = np.zeros((height, width), dtype=np.uint8)

# Mono16: 2 bytes per pixel
mono16_buffer = np.zeros((height, width), dtype=np.uint16)

Frame Conversion

import cv2
import numpy as np

# Convert BGR (OpenCV) to RGB (MediaX)
bgr_frame = cv2.imread('image.png')
rgb_frame = cv2.cvtColor(bgr_frame, cv2.COLOR_BGR2RGB)

# Convert RGB to YUV422
yuv_frame = cv2.cvtColor(rgb_frame, cv2.COLOR_RGB2YUV)
# Note: Additional packing needed for UYVY format

Error Handling

import sys
sys.path.append('/usr/local/lib/python3/dist-packages')
import mediax

try:
    payloader = mediax.rtp_uncompressed_payloader.RtpUncompressedPayloader()
    stream_info = mediax.rtp.StreamInformation()
    # Configure stream_info...

    payloader.SetStreamInfo(stream_info)
    if not payloader.Open():
        print("Failed to open payloader")
        sys.exit(1)

    # ... use payloader ...

except Exception as e:
    print(f"Error: {e}")
finally:
    payloader.Close()

Complete Example

#!/usr/bin/env python3
import sys
sys.path.append('/usr/local/lib/python3/dist-packages')
import mediax
import numpy as np
import time

def main():
    # Stream configuration
    stream_info = mediax.rtp.StreamInformation()
    stream_info.session_name = "python-example"
    stream_info.hostname = "239.192.1.1"
    stream_info.port = 5004
    stream_info.width = 640
    stream_info.height = 480
    stream_info.framerate = 30
    stream_info.encoding = mediax.rtp.ColourspaceType_kColourspaceRgb24

    # Setup SAP announcer
    announcer = mediax.sap.SapAnnouncer()
    announcer.AddSapAnnouncement(stream_info)
    announcer.Start()

    # Setup payloader
    payloader = mediax.rtp_uncompressed_payloader.RtpUncompressedPayloader()
    payloader.SetStreamInfo(stream_info)
    payloader.Open()

    # Generate and transmit test pattern
    for i in range(300):  # 10 seconds at 30fps
        frame = np.zeros((480, 640, 3), dtype=np.uint8)
        # Moving color bar
        x = (i * 10) % 640
        frame[:, x:min(x+100, 640), 0] = 255  # Red bar

        payloader.Transmit(frame.tobytes(), True)
        time.sleep(1/30)  # 30 fps timing

    # Cleanup
    payloader.Close()
    announcer.Stop()
    announcer.DeleteAllSapAnnouncements()

if __name__ == "__main__":
    main()

Support