Video Connector for Python

from vonage_video_connector.models import Session

# Session object properties
session.id  # str: Unique identifier for the session

from vonage_video_connector.models import Connection

# Connection object properties
connection.id             # str: Unique identifier for the connection
connection.creation_time  # datetime: When the connection was established
connection.data          # str: Connection data (encoded in the token)

from vonage_video_connector.models import Stream

# Stream object properties
stream.id          # str: Unique identifier for the stream
stream.connection  # Connection: The underlying connection that published this stream

from vonage_video_connector.models import Publisher

# Publisher object properties
publisher.stream  # Stream: The underlying stream for this publisher

from vonage_video_connector.models import Subscriber

# Subscriber object properties
subscriber.stream  # Stream: The underlying stream for this subscriber

from vonage_video_connector.models import AudioData

# AudioData object properties
audio_data.sample_buffer      # memoryview: 16-bit signed integer audio samples
audio_data.sample_rate        # int: Sample rate (8000-48000 Hz)
audio_data.number_of_channels # int: 1 (mono) or 2 (stereo)
audio_data.number_of_frames   # int: Number of audio frames

from vonage_video_connector.models import VideoFrame, VideoResolution

# VideoFrame object properties
video_frame.frame_buffer  # memoryview: 8-bit unsigned char video frame data
video_frame.resolution    # VideoResolution: Width and height in pixels
video_frame.format        # str: Video format (YUV420P, RGB24, or ARGB32)

from vonage_video_connector.models import VideoResolution

# VideoResolution object properties
resolution = VideoResolution(
    width=640,   # int: Width in pixels, defaults to 640
    height=480   # int: Height in pixels, defaults to 480
)

from vonage_video_connector.models import CaptionsData

# CaptionsData object properties
captions_data.text      # str: The caption text content
captions_data.is_final  # bool: True for final captions, False for interim/partial captions

from vonage_video_connector.models import MediaBufferStats, AudioBufferStats, VideoBufferStats

# MediaBufferStats object properties
stats.audio  # Optional[AudioBufferStats]: Audio buffer statistics
stats.video  # Optional[VideoBufferStats]: Video buffer statistics

# AudioBufferStats properties
stats.audio.duration  # timedelta: Duration of queued audio

# VideoBufferStats properties
stats.video.duration  # timedelta: Duration of queued video

from vonage_video_connector.models import SessionSettings

session_settings = SessionSettings(
    enable_migration=False,  # bool: Enable automatic session migration
    av=av_settings,          # Optional[SessionAVSettings]: Audio/video configuration
    logging=logging_settings # Optional[LoggingSettings]: Logging configuration
)

from vonage_video_connector.models import (
    SessionAVSettings,
    SessionAudioSettings,
    SessionVideoPublisherSettings,
    VideoResolution,
)

av_settings = SessionAVSettings(
    audio_publisher=SessionAudioSettings(sample_rate=48000, number_of_channels=2),
    audio_subscribers_mix=SessionAudioSettings(sample_rate=48000, number_of_channels=1),
    video_publisher=SessionVideoPublisherSettings(
        resolution=VideoResolution(width=1280, height=720),
        fps=30,
        format="YUV420P"
    )
)

from vonage_video_connector.models import SessionAudioSettings

audio_settings = SessionAudioSettings(
    sample_rate=48000,       # int: Optional, defaults to 48000
    number_of_channels=1     # int: Optional, 1 (mono) or 2 (stereo), defaults to 2
)

from vonage_video_connector.models import SessionVideoPublisherSettings, VideoResolution

video_settings = SessionVideoPublisherSettings(
    resolution=VideoResolution(width=1280, height=720),  # Required: resolution in pixels
    fps=30,              # int: Optional, 1-30, defaults to 30
    format="YUV420P"     # str: Optional, defaults to "YUV420P"
)

from vonage_video_connector.models import LoggingSettings

logging_settings = LoggingSettings(
    level="INFO"  # str: Optional, ERROR, WARN, INFO, DEBUG, or TRACE, defaults to WARN
)

from vonage_video_connector.models import PublisherSettings, PublisherAudioSettings

publisher_settings = PublisherSettings(
    name="My Application",           # str: Name for your published stream (required, min 1 char)
    has_audio=True,                  # bool: Whether to publish audio
    has_video=True,                  # bool: Whether to publish video
    enable_captions=False,           # bool: Whether to enable live captions for this stream (default: False)
    audio_settings=audio_settings    # Optional[PublisherAudioSettings]: Audio configuration
)

from vonage_video_connector.models import PublisherAudioSettings

audio_settings = PublisherAudioSettings(
    enable_stereo_mode=True,  # bool: Publish in stereo (True) or mono (False)
    enable_opus_dtx=False     # bool: Enable discontinuous transmission
)

from vonage_video_connector.models import SubscriberSettings, SubscriberVideoSettings, VideoResolution

subscriber_settings = SubscriberSettings(
    subscribe_to_audio=True,     # bool: Whether to subscribe to audio
    subscribe_to_video=True,     # bool: Whether to subscribe to video
    subscribe_to_captions=False, # bool: Whether to subscribe to live captions (default: False)
    video_settings=SubscriberVideoSettings(
        preferred_resolution=VideoResolution(width=640, height=480),
        preferred_framerate=15
    )
)

from vonage_video_connector.models import SubscriberVideoSettings, VideoResolution

video_settings = SubscriberVideoSettings(
    preferred_resolution=VideoResolution(width=640, height=480),  # Optional
    preferred_framerate=15  # Optional: Preferred FPS (1-30)
)

from vonage_video_connector import VonageVideoClient
from vonage_video_connector.models import (
    SessionSettings,
    SessionAVSettings,
    SessionAudioSettings,
    LoggingSettings,
)

# Create client instance
client = VonageVideoClient()

# Configure session settings
session_settings = SessionSettings(
    enable_migration=False,
    av=SessionAVSettings(
        audio_subscribers_mix=SessionAudioSettings(
            sample_rate=48000,
            number_of_channels=1
        )
    ),
    logging=LoggingSettings(level="INFO")
)

# Connect to session
connecting = client.connect(
    application_id="your_application_id",
    session_id="your_session_id",
    token="your_token",
    session_settings=session_settings,
    on_connected_cb=on_session_connected,
    on_error_cb=on_session_error
)

connecting = client.connect(
    application_id="your_application_id",
    session_id="your_session_id",
    token="your_token",
    session_settings=session_settings,
    on_error_cb=on_session_error,
    on_connected_cb=on_session_connected,
    on_disconnected_cb=on_session_disconnected,
    on_connection_created_cb=on_connection_created,
    on_connection_dropped_cb=on_connection_dropped,
    on_stream_received_cb=on_stream_received,
    on_stream_dropped_cb=on_stream_dropped,
    on_audio_data_cb=on_audio_data,
    on_ready_for_audio_cb=on_ready_for_audio,
    on_media_buffer_drained_cb=on_media_buffer_drained
)

success = client.disconnect()

if client.is_connected():
    print("Still connected")

from vonage_video_connector.models import (
    SessionAVSettings, 
    SessionAudioSettings, 
    SessionVideoPublisherSettings,
    VideoResolution
)

# Configure audio for publisher and subscriber mix
audio_publisher = SessionAudioSettings(
    sample_rate=48000,      # Valid: 8000, 12000, 16000, 24000, 32000, 44100, 48000
    number_of_channels=2    # 1 for mono, 2 for stereo
)

audio_subscribers_mix = SessionAudioSettings(
    sample_rate=48000,
    number_of_channels=1
)

# Configure video publisher settings
video_publisher = SessionVideoPublisherSettings(
    resolution=VideoResolution(width=1280, height=720),
    fps=30,
    format="YUV420P"  # Valid: YUV420P, RGB24, ARGB32
)

# Combine into session AV settings
av_settings = SessionAVSettings(
    audio_publisher=audio_publisher,
    audio_subscribers_mix=audio_subscribers_mix,
    video_publisher=video_publisher
)

from vonage_video_connector.models import LoggingSettings

# Configure logging level
logging_settings = LoggingSettings(
    level="DEBUG"  # Valid: ERROR, WARN, INFO, DEBUG, TRACE
)

from vonage_video_connector.models import SessionSettings

session_settings = SessionSettings(
    enable_migration=True,  # Enable automatic migration
    av=av_settings,
    logging=logging_settings
)

from vonage_video_connector.models import PublisherSettings, PublisherAudioSettings

# Configure publisher audio settings
audio_settings = PublisherAudioSettings(
    enable_stereo_mode=True,   # Publish in stereo
    enable_opus_dtx=False      # Enable discontinuous transmission for bandwidth savings
)

# Create publisher settings for audio and video
publisher_settings = PublisherSettings(
    name="AI Assistant Bot",
    has_audio=True,
    has_video=True,
    enable_captions=True,
    audio_settings=audio_settings
)

# Or audio-only publisher
audio_only_settings = PublisherSettings(
    name="Audio Bot",
    has_audio=True,
    has_video=False,
    enable_captions=False,
    audio_settings=audio_settings
)

success = client.publish(
    settings=publisher_settings,
    on_error_cb=on_publisher_error,
    on_stream_created_cb=on_stream_created,
    on_stream_destroyed_cb=on_stream_destroyed
)

import time

# Example: Wait for audio system to be ready
audio_ready = False

def on_ready_for_audio(session):
    global audio_ready
    audio_ready = True
    print("Audio system ready - can now add audio")

# Connect with the callback
client.connect(
    application_id="your_application_id",
    session_id="your_session_id",
    token="your_token",
    session_settings=session_settings,
    on_ready_for_audio_cb=on_ready_for_audio
)

# Publish
client.publish(settings=publisher_settings)

# Wait for audio to be ready before adding audio
while not audio_ready:
    time.sleep(0.01)

# Now safe to add audio
client.add_audio(audio_data)

from vonage_video_connector.models import AudioData

# Create audio data (example with 16-bit PCM samples)
audio_buffer = memoryview(your_audio_samples)  # Must be 16-bit signed integers

audio_data = AudioData(
    sample_buffer=audio_buffer,
    sample_rate=48000,
    number_of_channels=1,
    number_of_frames=960  # 20ms at 48kHz
)

# Add audio to the published stream
success = client.add_audio(audio_data)

success = client.unpublish()

if client.is_publishing():
    print("Still publishing")

from vonage_video_connector.models import SubscriberSettings, SubscriberVideoSettings, VideoResolution

def on_stream_received(session, stream):
    print(f"New stream received: {stream.id}")
    print(f"From connection: {stream.connection.id}")
    
    # Configure subscriber settings (optional)
    subscriber_settings = SubscriberSettings(
        subscribe_to_audio=True,
        subscribe_to_video=True,
        subscribe_to_captions=True,
        video_settings=SubscriberVideoSettings(
            preferred_resolution=VideoResolution(width=640, height=480),
            preferred_framerate=15
        )
    )
    
    # Subscribe to the stream
    success = client.subscribe(
        stream=stream,
        settings=subscriber_settings,
        on_error_cb=on_subscriber_error,
        on_connected_cb=on_subscriber_connected,
        on_disconnected_cb=on_subscriber_disconnected,
        on_render_frame_cb=on_render_frame,
        on_audio_data_cb=on_subscriber_audio_data,
        on_caption_text_cb=on_caption_text
    )

def on_render_frame(subscriber, video_frame):
    """Called for each subscribed stream's video frames"""
    stream_id = subscriber.stream.id
    width = video_frame.resolution.width
    height = video_frame.resolution.height
    print(f"Video frame from stream {stream_id}: {width}x{height}")
    # Process video for this specific stream

def on_audio_data(session, audio_data):
    """Called with mixed audio from all subscribed streams"""
    sample_rate = audio_data.sample_rate
    channels = audio_data.number_of_channels
    print(f"Mixed audio from all subscribers: {sample_rate}Hz, {channels} channel(s)")
    # Process the combined audio from all participants

def on_caption_text(subscriber, captions_data):
    """Called when caption text is received from a subscribed stream"""
    stream_id = subscriber.stream.id
    status = "final" if captions_data.is_final else "interim"
    print(f"Caption ({status}) from stream {stream_id}: {captions_data.text}")
    # Process interim captions for live display, final captions for storage or further processing

def on_subscriber_audio_data(subscriber, audio_data):
    """Called with individual audio from the stream subscribed to"""
    sample_rate = audio_data.sample_rate
    channels = audio_data.number_of_channels
    print(f"Individual audio from stream {subscriber.stream.id}: {sample_rate}Hz, {channels} channel(s)")
    # Process the individual audio from this stream

def on_stream_dropped(session, stream):
    print(f"Stream dropped: {stream.id}")
    
    # Unsubscribe from the stream
    success = client.unsubscribe(stream)

def on_audio_data(session, audio_data):
    """Process incoming audio data from subscribed streams"""
    
    # Access audio properties
    sample_rate = audio_data.sample_rate
    channels = audio_data.number_of_channels
    frames = audio_data.number_of_frames
    
    # Access the audio buffer (memoryview of 16-bit signed integers)
    audio_buffer = audio_data.sample_buffer
    
    print(f"Received {frames} frames at {sample_rate}Hz, {channels} channel(s)")
    
    # Convert to bytes if needed
    audio_bytes = audio_buffer.tobytes()
    
    # Process the audio (e.g., transcription, analysis, etc.)
    process_audio(audio_buffer, sample_rate, channels)
    
    # Generate response audio and add it back
    response_audio = generate_response(audio_buffer)
    if response_audio:
        client.add_audio(response_audio)

import array
import math

from vonage_video_connector.models import AudioData

# Create 16-bit PCM audio samples (example: sine wave)
sample_rate = 48000
duration_ms = 20  # 20ms frame
num_samples = int(sample_rate * duration_ms / 1000)

# Generate audio samples as 16-bit signed integers
samples = array.array('h')  # 'h' = signed short (16-bit)
for i in range(num_samples):
    # Example: generate a 440Hz sine wave
    sample = int(32767 * 0.5 * math.sin(2 * math.pi * 440 * i / sample_rate))
    samples.append(sample)

# Create AudioData object
audio_data = AudioData(
    sample_buffer=memoryview(samples),
    sample_rate=sample_rate,
    number_of_channels=1,
    number_of_frames=num_samples
)

# Add the audio
client.add_audio(audio_data)

def on_render_frame(subscriber, video_frame):
    """Process incoming video frames from subscribed streams"""
    
    # Access video properties
    width = video_frame.resolution.width
    height = video_frame.resolution.height
    format = video_frame.format
    
    # Access the video buffer (memoryview of 8-bit unsigned chars)
    frame_buffer = video_frame.frame_buffer
    
    print(f"Received {width}x{height} frame in {format} format")
    
    # Convert to bytes if needed
    frame_bytes = frame_buffer.tobytes()
    
    # Process the video frame (e.g., computer vision, recording, etc.)
    process_video(frame_buffer, width, height, format)

import array
from vonage_video_connector.models import VideoFrame, VideoResolution

# Create a video frame (example: solid color in YUV420P format)
width = 640
height = 480

# YUV420P format calculation:
# Y plane: width * height
# U plane: (width/2) * (height/2)
# V plane: (width/2) * (height/2)
y_size = width * height
uv_size = (width // 2) * (height // 2)
total_size = y_size + 2 * uv_size

# Create frame buffer as 8-bit unsigned chars
frame_data = array.array('B', [128] * total_size)  # 'B' = unsigned char (8-bit)

# Create VideoFrame object
video_frame = VideoFrame(
    frame_buffer=memoryview(frame_data),
    resolution=VideoResolution(width=width, height=height),
    format="YUV420P"
)

# Add the video frame
success = client.add_video(video_frame)

# Get current buffer statistics
stats = client.get_media_buffer_stats()

if stats.audio:
    print(f"Audio buffer duration: {stats.audio.duration.total_seconds()}s")

if stats.video:
    print(f"Video buffer duration: {stats.video.duration.total_seconds()}s")

# Clear all media buffers
success = client.clear_media_buffers()

if success:
    print("Media buffers cleared successfully")

def on_media_buffer_drained(stats):
    """Called when media buffers are drained"""
    print("Media buffers drained")
    
    if stats.audio:
        print(f"Audio buffer: {stats.audio.duration.total_seconds()}s remaining")
    
    if stats.video:
        print(f"Video buffer: {stats.video.duration.total_seconds()}s remaining")

# Get the local connection
connection = client.get_connection()

if connection:
    print(f"Connection ID: {connection.id}")
    print(f"Connection data: {connection.data}")
    print(f"Created at: {connection.creation_time}")

def on_session_error(session, error_description, error_code):
    """Handle session errors"""
    print(f"Session error: {error_description} (Code: {error_code})")

def on_session_connected(session):
    """Handle successful session connection"""
    print(f"Connected to session: {session.id}")

def on_session_disconnected(session):
    """Handle session disconnection"""
    print(f"Disconnected from session: {session.id}")

def on_ready_for_audio(session):
    """Called when the audio system is ready"""
    print("Audio system ready - can now add audio")

def on_connection_created(session, connection):
    """Handle new participant joining"""
    print(f"Participant joined: {connection.id}")
    print(f"Connection data: {connection.data}")

def on_connection_dropped(session, connection):
    """Handle participant leaving"""
    print(f"Participant left: {connection.id}")

def on_stream_received(session, stream):
    """Handle new streams from other participants"""
    print(f"Stream received: {stream.id} from connection {stream.connection.id}")
    # Decide whether to subscribe based on your application logic

def on_stream_dropped(session, stream):
    """Handle streams being removed"""
    print(f"Stream dropped: {stream.id}")

def on_publisher_error(publisher, error_description, error_code):
    """Handle publisher errors"""
    print(f"Publisher error: {error_description} (Code: {error_code})")

def on_stream_created(publisher):
    """Handle successful stream creation"""
    print(f"Published stream created: {publisher.stream.id}")

def on_stream_destroyed(publisher):
    """Handle stream destruction"""
    print(f"Published stream destroyed: {publisher.stream.id}")

def on_subscriber_error(subscriber, error_description, error_code):
    """Handle subscriber errors"""
    print(f"Subscriber error: {error_description} (Code: {error_code})")

def on_subscriber_connected(subscriber):
    """Handle successful subscription"""
    print(f"Subscribed to stream: {subscriber.stream.id}")

def on_subscriber_disconnected(subscriber):
    """Handle subscription disconnection"""
    print(f"Unsubscribed from stream: {subscriber.stream.id}")

def on_render_frame(subscriber, video_frame):
    """Handle incoming video frames"""
    width = video_frame.resolution.width
    height = video_frame.resolution.height
    print(f"Video frame: {width}x{height} in {video_frame.format} format")

def on_caption_text(subscriber, captions_data):
    """Handle incoming caption text"""
    status = "final" if captions_data.is_final else "interim"
    print(f"Caption ({status}) from stream {subscriber.stream.id}: {captions_data.text}")

from pydantic import ValidationError

from vonage_video_connector.models import PublisherSettings, SessionAudioSettings

try:
    publisher_settings = PublisherSettings(name="Bot", has_audio=False, has_video=False)
except ValidationError as e:
    # "At least one of has_audio(False) or has_video (False) must be set to true."
    print(f"Invalid publisher settings: {e}")

try:
    audio_settings = SessionAudioSettings(sample_rate=44000)
except ValidationError as e:
    # "44000 is not one of the allowed: [8000, 12000, 16000, 24000, 32000, 44100, 48000]"
    print(f"Invalid audio settings: {e}")

try:
    client.add_audio(audio_data)
except (TypeError, AttributeError) as e:
    print(f"Malformed audio data: {e}")

if not client.connect(
    application_id="your_application_id",
    session_id="your_session_id",
    token="your_token",
    session_settings=session_settings,
):
    print("Failed to start connecting - the client may already be connected")

def on_session_error(session, error_description, error_code):
    print(f"Session error: {error_description} (Code: {error_code})")

client.connect(
    application_id="your_application_id",
    session_id="your_session_id",
    token="your_token",
    on_error_cb=on_session_error,
)

try:
    # Your application logic
    connecting = client.connect(
        application_id="your_application_id",
        session_id="your_session_id",
        token="your_token",
        session_settings=session_settings,
    )
    # ... do work ...

except Exception as e:
    print(f"Application error: {e}")

finally:
    # Clean up resources
    if client.is_publishing():
        client.unpublish()

    if client.is_connected():
        client.disconnect()