video_stream

Author: theomonnom

examples/basic_room/room.py

@@ -11,7 +11,18 @@ async def main():
 
     @room.on("participant_connected")
     def on_participant_connected(participant: livekit.RemoteParticipant):
-        print("Participant connected: " + participant.identity)
+        print("participant connected: " + participant.identity)
+
+    @room.on("track_subscribed")
+    def on_track_subscribed(track: livekit.Track, publication: livekit.RemoteTrackPublication, participant: livekit.RemoteParticipant):
+        if track.kind == livekit.TrackKind.KIND_VIDEO:
+            video_stream = livekit.VideoStream(track)
+
+            @video_stream.on("frame_received")
+            def on_video_frame(frame: livekit.VideoFrame, buffer: livekit.VideoFrameBuffer):
+                argb = livekit.ArgbFrame(
+                    livekit.VideoFormatType.FORMAT_ABGR, buffer.width, buffer.height)
+                buffer.to_argb(argb)
 
     print("Connected to room with sid: " + room.sid)
     await room.run()

livekit/__init__.py

@@ -3,11 +3,17 @@
 
 __version__ = "0.0.1"
 
-from _proto.track_pb2 import (TrackKind, TrackSource, StreamState)
+from ._proto.video_frame_pb2 import (
+    VideoRotation, VideoFormatType, VideoFrameBufferType)
+from ._proto.track_pb2 import (TrackKind, TrackSource, StreamState)
 
 from .room import Room
 from .participant import (Participant, LocalParticipant, RemoteParticipant)
 from .track import (Track, LocalAudioTrack, LocalVideoTrack,
                     RemoteAudioTrack, RemoteVideoTrack)
 from .track_publication import (
     TrackPublication, LocalTrackPublication, RemoteTrackPublication)
+
+from .video_frame import (ArgbFrame, VideoFrame, VideoFrameBuffer, NativeVideoFrameBuffer, PlanarYuvBuffer,
+                          PlanarYuv8Buffer, PlanarYuv16Buffer, I420Buffer, I420ABuffer, I422Buffer, I010Buffer, NV12Buffer)
+from .video_stream import VideoStream

livekit/_ffi_client.py

@@ -6,28 +6,30 @@
 import pkg_resources
 import asyncio
 import threading
-import logging 
+import logging
 
 if sys.platform == "win32":
     libfile = 'livekit_ffi.dll'
 elif sys.platform == "darwin":
     libfile = 'liblivekit_ffi.dylib'
 else:
-    libfile = 'liblivekit_ffi.so' 
+    libfile = 'liblivekit_ffi.so'
 
 libpath = pkg_resources.resource_filename('livekit', libfile)
 
 ffi_lib = CDLL(libpath)
 
 # C function types
-ffi_lib.livekit_ffi_request.argtypes = [POINTER(c_ubyte), c_size_t, POINTER(POINTER(c_ubyte)), POINTER(c_size_t)]
+ffi_lib.livekit_ffi_request.argtypes = [
+    POINTER(c_ubyte), c_size_t, POINTER(POINTER(c_ubyte)), POINTER(c_size_t)]
 ffi_lib.livekit_ffi_request.restype = c_size_t
 
 ffi_lib.livekit_ffi_drop_handle.argtypes = [c_size_t]
 ffi_lib.livekit_ffi_drop_handle.restype = c_bool
 
 INVALID_HANDLE = 0
 
+
 @CFUNCTYPE(c_void_p, POINTER(c_uint8), c_size_t)
 def ffi_event_callback(data_ptr: POINTER(c_uint8), data_len: c_size_t):
     event_data = bytes(data_ptr[:data_len])
@@ -40,46 +42,51 @@ def ffi_event_callback(data_ptr: POINTER(c_uint8), data_len: c_size_t):
 
     loop.call_soon_threadsafe(dispatch_event, event)
 
+
 def dispatch_event(event: proto_ffi.FfiEvent):
     ffi_client = FfiClient()
     which = event.WhichOneof('message')
-    if which == 'connect':
-        ffi_client.emit('connect', event.connect)
-    elif which == 'room_event':
-        ffi_client.emit('room', event.room_event)
+    ffi_client.emit(which, getattr(event, which))
+
 
 class Singleton(type):
     _instances = {}
+
     def __call__(cls, *args, **kwargs):
         if cls not in cls._instances:
-            cls._instances[cls] = super(Singleton, cls).__call__(*args, **kwargs)
+            cls._instances[cls] = super(
+                Singleton, cls).__call__(*args, **kwargs)
         return cls._instances[cls]
 
+
 class FfiClient(EventEmitter, metaclass=Singleton):
     def __init__(self):
         super().__init__()
         self._lock = threading.Lock()
         self._event_loop = None
 
         req = proto_ffi.FfiRequest()
-        req.initialize.event_callback_ptr = cast(ffi_event_callback, c_void_p).value
+        req.initialize.event_callback_ptr = cast(
+            ffi_event_callback, c_void_p).value
         self.request(req)
 
     def set_event_loop(self, loop: asyncio.AbstractEventLoop):
         with self._lock:
             if self._event_loop is not None and self._event_loop != loop:
-                logging.warning("FfiClient is now using a different asyncio event_loop")
+                logging.warning(
+                    "FfiClient is now using a different asyncio event_loop")
 
             self._event_loop = loop
 
-    def request(self, req: proto_ffi.FfiRequest) -> proto_ffi.FfiResponse: 
+    def request(self, req: proto_ffi.FfiRequest) -> proto_ffi.FfiResponse:
         data = req.SerializeToString()
         data_len = len(data)
         data = (c_ubyte * data_len)(*data)
 
         resp_ptr = POINTER(c_ubyte)()
         resp_len = c_size_t()
-        handle = ffi_lib.livekit_ffi_request(data, data_len, byref(resp_ptr), byref(resp_len))
+        handle = ffi_lib.livekit_ffi_request(
+            data, data_len, byref(resp_ptr), byref(resp_len))
 
         resp_data = bytes(resp_ptr[:resp_len.value])
         resp = proto_ffi.FfiResponse()
@@ -88,6 +95,7 @@ def request(self, req: proto_ffi.FfiRequest) -> proto_ffi.FfiResponse:
         FfiHandle(handle)
         return resp
 
+
 class FfiHandle:
     handle = INVALID_HANDLE
 
@@ -96,4 +104,4 @@ def __init__(self, handle: int):
 
     def __del__(self):
         if self.handle != INVALID_HANDLE:
-            assert(ffi_lib.livekit_ffi_drop_handle(c_size_t(self.handle)))
+            assert (ffi_lib.livekit_ffi_drop_handle(c_size_t(self.handle)))

livekit/participant.py

@@ -3,7 +3,6 @@
 
 
 class Participant():
-
     def __init__(self, info: proto_participant.ParticipantInfo):
         self._info = info
         self._tracks: dict[str, TrackPublication] = {}

livekit/room.py

@@ -1,12 +1,14 @@
-from ctypes import *
 import asyncio
 from pyee.asyncio import AsyncIOEventEmitter
 from ._ffi_client import (FfiClient, FfiHandle)
-from .participant import (Participant, LocalParticipant, RemoteParticipant)
-from .track_publication import (RemoteTrackPublication, LocalTrackPublication)
 from ._proto import ffi_pb2 as proto_ffi
 from ._proto import room_pb2 as proto_room
 from ._proto import participant_pb2 as proto_participant
+from .participant import (Participant, LocalParticipant, RemoteParticipant)
+from .track_publication import (RemoteTrackPublication, LocalTrackPublication)
+from .track import (RemoteAudioTrack, RemoteVideoTrack)
+from livekit import TrackKind
+import weakref
 
 
 class ConnectError(Exception):
@@ -21,6 +23,10 @@ def __init__(self):
         self._room_info: proto_room.RoomInfo = None
         self._participants: dict[str, RemoteParticipant] = {}
 
+    def __del__(self):
+        ffi_client = FfiClient()
+        ffi_client.remove_listener('room_event', self._on_room_event)
+
     async def connect(self, url: str, token: str):
         # TODO(theomonnom): We should be more flexible about the event loop
         ffi_client = FfiClient()
@@ -37,6 +43,10 @@ async def connect(self, url: str, token: str):
 
         def on_connect_callback(cb: proto_room.ConnectCallback):
             if cb.async_id == async_id:
+                # add existing participants
+                for participant_info in cb.room.participants:
+                    self._create_remote_participant(participant_info)
+
                 future.set_result(cb)
                 ffi_client.remove_listener('connect', on_connect_callback)
 
@@ -49,7 +59,7 @@ def on_connect_callback(cb: proto_room.ConnectCallback):
             self._ffi_handle = FfiHandle(resp.room.handle.id)
             self._room_info = resp.room
             self._close_future = asyncio.Future()
-            ffi_client.add_listener('room', self._on_room_event)
+            ffi_client.add_listener('room_event', self._on_room_event)
 
     async def close(self):
         self._ffi_handle = None
@@ -74,15 +84,47 @@ def _on_room_event(self, event: proto_room.RoomEvent):
             publication = RemoteTrackPublication(
                 event.track_published.publication)
             participant._tracks[publication.sid] = publication
-            self.emit('track_published', publication)
+            self.emit('track_published', publication, participant)
         elif which == 'track_unpublished':
             participant = self._participants[event.track_unpublished.participant_sid]
+            publication = participant._tracks.pop(
+                event.track_unpublished.publication_sid)
+            self.emit('track_unpublished', publication, participant)
+        elif which == 'track_subscribed':
+            track_info = event.track_subscribed.track
+            participant = self._participants[event.track_subscribed.participant_sid]
+            publication = participant._tracks[track_info.sid]
+
+            if track_info.kind == TrackKind.KIND_VIDEO:
+                video_track = RemoteVideoTrack(
+                    track_info, weakref.ref(self), weakref.ref(participant))
+                publication._track = video_track
+                self.emit('track_subscribed', video_track,
+                          publication, participant)
+            elif track_info.kind == TrackKind.KIND_AUDIO:
+                audio_track = RemoteAudioTrack(
+                    track_info, weakref.ref(self), weakref.ref(participant))
+                publication._track = audio_track
+                self.emit('track_subscribed', audio_track,
+                          publication, participant)
+        elif which == 'track_unsubscribed':
+            participant = self._participants[event.track_unsubscribed.participant_sid]
+            publication = participant._tracks[event.track_unsubscribed.track_sid]
+            track = publication._track
+            publication._track = None
+            self.emit('track_unsubscribed', track, publication, participant)
 
     def _create_remote_participant(self, info: proto_participant.ParticipantInfo) -> RemoteParticipant:
+        if info.sid in self._participants:
+            raise Exception('participant already exists')
+
         participant = RemoteParticipant(info)
         self._participants[participant.sid] = participant
 
-        # TODO(publications)
+        #  add existing track publications
+        for publication_info in info.publications:
+            publication = RemoteTrackPublication(publication_info)
+            participant._tracks[publication.sid] = publication
 
         return participant
 

livekit/track.py

@@ -1,13 +1,24 @@
 from ._proto import track_pb2 as proto_track
-from typing import Optional
 from ._ffi_client import FfiHandle
+from typing import (Optional, TYPE_CHECKING)
+from weakref import ref
+
+if TYPE_CHECKING:
+    from livekit import (Room, Participant)
 
 
 class Track():
-    def __init__(self, info: proto_track.TrackInfo, handle: Optional[FfiHandle]):
+    def __init__(self, handle: Optional[FfiHandle], info: proto_track.TrackInfo, room: ref['Room'], participant: ref['Participant']):
         self._info = info
         self._ffi_handle = handle
 
+        # TODO(theomonnom): Simplify that and use a FfiHandleId?
+        # the weak references are needed because when we need to communicate with the
+        # ffi_server which track we are referring to, we also need to provide the room
+        # and the participant.
+        self._room = room
+        self._participant = participant
+
     @property
     def sid(self) -> str:
         return self._info.sid
@@ -33,20 +44,20 @@ def update_info(self, info: proto_track.TrackInfo):
 
 
 class LocalAudioTrack(Track):
-    def __init__(self, info: proto_track.TrackInfo, handle: FfiHandle):
-        super().__init__(info, handle)
+    def __init__(self, ffi_handle: FfiHandle, info: proto_track.TrackInfo, room: ref['Room'], participant: ref['Participant']):
+        super().__init__(ffi_handle, info, room, participant)
 
 
 class LocalVideoTrack(Track):
-    def __init__(self, info: proto_track.TrackInfo, handle: FfiHandle):
-        super().__init__(info, handle)
+    def __init__(self, ffi_handle: FfiHandle, info: proto_track.TrackInfo, room: ref['Room'], participant: ref['Participant']):
+        super().__init__(ffi_handle, info, room, participant)
 
 
 class RemoteAudioTrack(Track):
-    def __init__(self, info: proto_track.TrackInfo):
-        super().__init__(info, None)
+    def __init__(self, info: proto_track.TrackInfo, room: ref['Room'], participant: ref['Participant']):
+        super().__init__(None, info, room, participant)
 
 
 class RemoteVideoTrack(Track):
-    def __init__(self, info: proto_track.TrackInfo):
-        super().__init__(info, None)
+    def __init__(self, info: proto_track.TrackInfo, room: ref['Room'], participant: ref['Participant']):
+        super().__init__(None, info, room, participant)

livekit/track_publication.py

@@ -1,10 +1,12 @@
 from livekit._proto import track_pb2 as proto_track
 from ._proto import track_pb2 as proto_track
+from .track import Track
 
 
 class TrackPublication():
     def __init__(self, info: proto_track.TrackPublicationInfo):
         self._info = info
+        self._track: Track = None
 
     @property
     def sid(self) -> str:
@@ -23,8 +25,8 @@ def source(self) -> proto_track.TrackSource:
         return self._info.source
 
     @property
-    def simulcast(self) -> bool:
-        return self._info.simulcast
+    def simulcasted(self) -> bool:
+        return self._info.simulcasted
 
     @property
     def width(self) -> int: