Custom Video Track | Python

In VideoSDK, you can ingest a custom video track from a file into your meeting using a CustomVideoTrack.

This example demonstrates how to create a custom video track from MP4 file and integrate it into a VideoSDK meeting.

Requirements

• Python 3.11+
• videosdk Python library
• av library for handling video frames
• asyncio for asynchronous operations

Setup

Before running the example, ensure you have set up your environment variables for VideoSDK authentication:

VIDEOSDK_TOKEN = "<VIDEOSDK_TOKEN>"
MEETING_ID = "<MEETING_ID>"
NAME = "<NAME>"

Custom Video Track Class

The VideoFileTrack class extends CustomVideoTrack to handle video stream ingestion from an MP4 file:

import av
import time
import fractions
import asyncio
from av import VideoFrame
from vsaiortc.mediastreams import MediaStreamError
from videosdk import CustomVideoTrack, MeetingConfig, VideoSDK, Participant, MeetingEventHandler, ParticipantEventHandler

VIDEOSDK_TOKEN = "VIDEOSDK_TOKEN"
MEETING_ID = "MEETING_ID"
NAME = "NAME"
loop = asyncio.get_event_loop()

class VideoFileTrack(CustomVideoTrack):
    """
    A video track that plays a video file in a continuous loop.
    """

    def __init__(self, path: str):
        super().__init__()  # Initialize the parent class
        self.kind = "video"
        self.path = path
        self.container = av.open(path)
        self.stream = self.container.streams.video[0]
        self._start = time.time()
        self._timestamp = 0

    async def recv(self) -> VideoFrame:
        """
        Receive the next :class:`~av.video.frame.VideoFrame`.
        """
        # implementing in next step

VideoFileTrack.recv()

Purpose:

The recv method in VideoFileTrack class retrieves the next video frame from the source(file, Url, Buffer, or any).

Key Components:

• Timestamp Management: Manages the delivery of video frames based on calculated timestamps (pts) and time base (time_base).
• Packet Decoding: Iterates through packets and decodes video frames until a valid frame is obtained.
• Error Handling: Handles end-of-stream by seeking back to the beginning of the file (self.container.seek(0)) to loop the video.

Code Snippet:

    async def recv(self) -> VideoFrame:
        """
        Receive the next :class:`~av.video.frame.VideoFrame`.
        """

        # Timestamp Management
        pts, time_base = await self.next_timestamp()

        # Packet Decoding & Error Handling
        # Read next frame from the video file
        frame = None
        for packet in self.container.demux(self.stream):
            try:
                for video_frame in packet.decode():
                    frame = video_frame
                    break
                if frame:
                    break
            except EOFError:
                self.container.seek(0)
            finally:
                pass

        frame.pts = pts
        frame.time_base = time_base
        return frame

    async def next_timestamp(self) -> Tuple[int, fractions.Fraction]:
        VIDEO_PTIME = 1 / 30  # Packet time for 30fps
        VIDEO_CLOCK_RATE = 90000  # Clock rate for video
        VIDEO_TIME_BASE = fractions.Fraction(1, VIDEO_CLOCK_RATE)

        if self.readyState != "live":
            raise MediaStreamError

        if hasattr(self, "_timestamp"):
            self._timestamp += int(VIDEO_PTIME * VIDEO_CLOCK_RATE)
            wait = self._start + (self._timestamp /
                                  VIDEO_CLOCK_RATE) - time.time()
            await asyncio.sleep(max(0, wait))
        else:
            self._start = time.time()
            self._timestamp = 0
        return self._timestamp, VIDEO_TIME_BASE

Event Handlers

Define custom event handlers to handle meeting and participant events:

class MyMeetingEventHandler(MeetingEventHandler):
    def on_meeting_joined(self, data):
        print("Meeting joined")

    def on_meeting_left(self, data):
        print("Meeting left")

class MyParticipantEventHandler(ParticipantEventHandler):
    def __init__(self, participant_id: str):
        super().__init__()
        self.participant_id = participant_id

    def on_stream_enabled(self, stream):
        print(f"Participant {self.participant_id} stream enabled: {stream.kind}")

    def on_stream_disabled(self, stream):
        print(f"Participant {self.participant_id} stream disabled: {stream.kind}")

• You can learn more about MeetingEventHandler & ParticipantEventHandler class in upcoming guide.

Main Function

Configure and run the main function to initialize the meeting with the custom video track:

def main():
    # Initialize custom video track
    video_track = VideoFileTrack("./example-video.mp4")

    # Configure meeting
    meeting_config = MeetingConfig(
        meeting_id=MEETING_ID,
        name=NAME,
        mic_enabled=False,
        webcam_enabled=True,
        token=VIDEOSDK_TOKEN,
        custom_camera_video_track=video_track
    )

    # Initialize VideoSDK meeting
    meeting = VideoSDK.init_meeting(**meeting_config)

    # Add event listeners
    meeting.add_event_listener(MyMeetingEventHandler())
    meeting.local_participant.add_event_listener(MyParticipantEventHandler(participant_id=meeting.local_participant.id))

    # Join the meeting
    print("Joining the meeting...")
    meeting.join()

    print("Meeting setup complete")

if __name__ == '__main__':
    main()
    loop.run_forever()

Conclusion

This setup demonstrates how to integrate a custom video track from an MP4 file into a VideoSDK meeting, with event handling for meeting and participant interactions.

API Reference

Refer to the following API documentation for methods and events used in this guide:

• Custom Tracks
• on_stream_enabled()
• on_stream_disabled()