fixed bugs & more examples

Author: theomonnom

client-sdk-rust

@@ -1,27 +1,73 @@
 import livekit
+import logging
 import asyncio
+from signal import SIGINT, SIGTERM
 
 URL = 'ws://localhost:7880'
 TOKEN = 'eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJleHAiOjE5MDY2MTMyODgsImlzcyI6IkFQSVRzRWZpZFpqclFvWSIsIm5hbWUiOiJuYXRpdmUiLCJuYmYiOjE2NzI2MTMyODgsInN1YiI6Im5hdGl2ZSIsInZpZGVvIjp7InJvb20iOiJ0ZXN0Iiwicm9vbUFkbWluIjp0cnVlLCJyb29tQ3JlYXRlIjp0cnVlLCJyb29tSm9pbiI6dHJ1ZSwicm9vbUxpc3QiOnRydWV9fQ.uSNIangMRu8jZD5mnRYoCHjcsQWCrJXgHCs0aNIgBFY'
 
 
 async def main():
     room = livekit.Room()
-    await room.connect(URL, TOKEN)
-    print("connected to room: " + room.name)
+
+    logging.info("connecting to %s", URL)
+    try:
+        await room.connect(URL, TOKEN)
+        logging.info("connected to room %s", room.name)
+    except livekit.ConnectError as e:
+        logging.error("failed to connect to the room: %s", e)
+        return False
+
+    @room.on("participant_connected")
+    def on_participant_connected(participant: livekit.RemoteParticipant):
+        logging.info(
+            "participant connected: %s %s", participant.sid, participant.identity)
+
+    @room.on("participant_disconnected")
+    def on_participant_disconnected(participant: livekit.RemoteParticipant):
+        logging.info("participant disconnected: %s %s",
+                     participant.sid, participant.identity)
+
+    @room.on("track_published")
+    def on_track_published(publication: livekit.LocalTrackPublication, participant: livekit.RemoteParticipant):
+        logging.info("track published: %s from participant %s (%s)",
+                     publication.sid, participant.sid, participant.identity)
+
+    @room.on("track_unpublished")
+    def on_track_unpublished(publication: livekit.LocalTrackPublication, participant: livekit.RemoteParticipant):
+        logging.info("track unpublished: %s", publication.sid)
 
     @room.on("track_subscribed")
     def on_track_subscribed(track: livekit.Track, publication: livekit.RemoteTrackPublication, participant: livekit.RemoteParticipant):
+        logging.info("track subscribed: %s", publication.sid)
         if track.kind == livekit.TrackKind.KIND_VIDEO:
             video_stream = livekit.VideoStream(track)
 
             @video_stream.on("frame_received")
             def on_video_frame(frame: livekit.VideoFrame):
-                print("received video frame")
+                # e.g: Do something with the frames here :)
                 pass
 
-    await room.run()
+    @room.on("track_unsubscribed")
+    def on_track_unsubscribed(track: livekit.Track, publication: livekit.RemoteTrackPublication, participant: livekit.RemoteParticipant):
+        logging.info("track unsubscribed: %s", publication.sid)
+
+    try:
+        await room.run()
+    except asyncio.CancelledError:
+        logging.info("closing the room")
+        await room.close()
 
 
 if __name__ == "__main__":
-    asyncio.run(main())
+    logging.basicConfig(level=logging.INFO, handlers=[
+                        logging.FileHandler("basic_room.log"), logging.StreamHandler()])
+
+    loop = asyncio.get_event_loop()
+    main_task = asyncio.ensure_future(main())
+    for signal in [SIGINT, SIGTERM]:
+        loop.add_signal_handler(signal, main_task.cancel)
+    try:
+        loop.run_until_complete(main_task)
+    finally:
+        loop.close()

examples/basic_room/room.py

@@ -0,0 +1,136 @@
+import matplotlib.pyplot as plt
+import livekit
+import asyncio
+import cv2
+import numpy as np
+import os
+from queue import Queue
+import mediapipe as mp
+from mediapipe.tasks import python
+from mediapipe.tasks.python import vision
+from mediapipe import solutions
+from mediapipe.framework.formats import landmark_pb2
+import numpy as np
+
+URL = 'ws://localhost:7880'
+TOKEN = 'eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJleHAiOjE5MDY2MTMyODgsImlzcyI6IkFQSVRzRWZpZFpqclFvWSIsIm5hbWUiOiJuYXRpdmUiLCJuYmYiOjE2NzI2MTMyODgsInN1YiI6Im5hdGl2ZSIsInZpZGVvIjp7InJvb20iOiJ0ZXN0Iiwicm9vbUFkbWluIjp0cnVlLCJyb29tQ3JlYXRlIjp0cnVlLCJyb29tSm9pbiI6dHJ1ZSwicm9vbUxpc3QiOnRydWV9fQ.uSNIangMRu8jZD5mnRYoCHjcsQWCrJXgHCs0aNIgBFY'
+
+frame_queue = Queue()
+argb_frame = None
+
+# You can downlo9ad a face landmark model file from https://developers.google.com/mediapipe/solutions/vision/face_landmarker#models
+model_file = 'face_landmarker.task'
+model_path = os.path.dirname(os.path.realpath(__file__)) + '/' + model_file
+
+BaseOptions = mp.tasks.BaseOptions
+FaceLandmarker = mp.tasks.vision.FaceLandmarker
+FaceLandmarkerOptions = mp.tasks.vision.FaceLandmarkerOptions
+VisionRunningMode = mp.tasks.vision.RunningMode
+
+options = FaceLandmarkerOptions(
+    base_options=BaseOptions(model_asset_path=model_path),
+    running_mode=VisionRunningMode.VIDEO)
+
+# from https://github.com/googlesamples/mediapipe/blob/main/examples/face_landmarker/python/%5BMediaPipe_Python_Tasks%5D_Face_Landmarker.ipynb
+
+
+def draw_landmarks_on_image(rgb_image, detection_result):
+    face_landmarks_list = detection_result.face_landmarks
+
+    # Loop through the detected faces to visualize.
+    for idx in range(len(face_landmarks_list)):
+        face_landmarks = face_landmarks_list[idx]
+
+        # Draw the face landmarks.
+        face_landmarks_proto = landmark_pb2.NormalizedLandmarkList()
+        face_landmarks_proto.landmark.extend([
+            landmark_pb2.NormalizedLandmark(x=landmark.x, y=landmark.y, z=landmark.z) for landmark in face_landmarks
+        ])
+
+        solutions.drawing_utils.draw_landmarks(
+            image=rgb_image,
+            landmark_list=face_landmarks_proto,
+            connections=mp.solutions.face_mesh.FACEMESH_TESSELATION,
+            landmark_drawing_spec=None,
+            connection_drawing_spec=mp.solutions.drawing_styles
+            .get_default_face_mesh_tesselation_style())
+        solutions.drawing_utils.draw_landmarks(
+            image=rgb_image,
+            landmark_list=face_landmarks_proto,
+            connections=mp.solutions.face_mesh.FACEMESH_CONTOURS,
+            landmark_drawing_spec=None,
+            connection_drawing_spec=mp.solutions.drawing_styles
+            .get_default_face_mesh_contours_style())
+        solutions.drawing_utils.draw_landmarks(
+            image=rgb_image,
+            landmark_list=face_landmarks_proto,
+            connections=mp.solutions.face_mesh.FACEMESH_IRISES,
+            landmark_drawing_spec=None,
+            connection_drawing_spec=mp.solutions.drawing_styles
+            .get_default_face_mesh_iris_connections_style())
+
+
+async def room():
+    room = livekit.Room()
+    await room.connect(URL, TOKEN)
+    print("connected to room: " + room.name)
+
+    @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):
+                frame_queue.put(frame)
+
+    await room.run()
+
+
+def display_frames():
+    cv2.namedWindow('livekit_video', cv2.WINDOW_AUTOSIZE)
+    cv2.startWindowThread()
+
+    global argb_frame
+
+    with FaceLandmarker.create_from_options(options) as landmarker:
+        while True:
+            frame = frame_queue.get()
+            buffer = frame.buffer
+
+            if argb_frame is None or argb_frame.width != buffer.width or argb_frame.height != buffer.height:
+                argb_frame = livekit.ArgbFrame(
+                    livekit.VideoFormatType.FORMAT_ABGR, buffer.width, buffer.height)
+
+            buffer.to_argb(argb_frame)
+
+            arr = np.ctypeslib.as_array(argb_frame.data)
+            arr = arr.reshape((argb_frame.height, argb_frame.width, 4))
+            arr = cv2.cvtColor(arr, cv2.COLOR_RGBA2RGB)
+
+            mp_image = mp.Image(
+                image_format=mp.ImageFormat.SRGB, data=arr)
+
+            detection_result = landmarker.detect_for_video(
+                mp_image, frame.timestamp)
+
+            draw_landmarks_on_image(arr, detection_result)
+
+            arr = cv2.cvtColor(arr, cv2.COLOR_RGB2BGR)
+
+            cv2.imshow('livekit_video', arr)
+            if cv2.waitKey(1) & 0xFF == ord('q'):
+                break
+
+    cv2.destroyAllWindows()
+
+
+async def main():
+    loop = asyncio.get_event_loop()
+    future = loop.run_in_executor(None, asyncio.run, room())
+
+    display_frames()
+    await future
+
+if __name__ == "__main__":
+    asyncio.run(main())

examples/face_landmark/room.py

@@ -0,0 +1,80 @@
+import livekit
+import logging
+import numpy as np
+import colorsys
+import asyncio
+from signal import SIGINT, SIGTERM
+
+URL = 'ws://localhost:7880'
+TOKEN = 'eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJleHAiOjE5MDY2MTMyODgsImlzcyI6IkFQSVRzRWZpZFpqclFvWSIsIm5hbWUiOiJuYXRpdmUiLCJuYmYiOjE2NzI2MTMyODgsInN1YiI6Im5hdGl2ZSIsInZpZGVvIjp7InJvb20iOiJ0ZXN0Iiwicm9vbUFkbWluIjp0cnVlLCJyb29tQ3JlYXRlIjp0cnVlLCJyb29tSm9pbiI6dHJ1ZSwicm9vbUxpc3QiOnRydWV9fQ.uSNIangMRu8jZD5mnRYoCHjcsQWCrJXgHCs0aNIgBFY'
+
+
+async def publish_frames(source: livekit.VideoSource):
+    argb_frame = livekit.ArgbFrame(
+        livekit.VideoFormatType.FORMAT_ARGB, 1280, 720)
+
+    arr = np.ctypeslib.as_array(argb_frame.data)
+
+    framerate = 1 / 30
+    hue = 0
+
+    while True:
+        frame = livekit.VideoFrame(
+            0, livekit.VideoRotation.VIDEO_ROTATION_0, argb_frame.to_i420())
+
+        rgb = colorsys.hsv_to_rgb(hue, 1.0, 1.0)
+        rgb = [int(x * 255) for x in rgb]
+        argb_color = rgb + [255]
+
+        for i in range(0, len(arr), 4):
+            arr[i:i+4] = argb_color
+
+        source.capture_frame(frame)
+
+        hue += framerate/2  # 3s for a full cycle
+        if hue >= 1.0:
+            hue -= 1.0
+
+        await asyncio.sleep(framerate)
+
+
+async def main():
+    room = livekit.Room()
+
+    logging.info("connecting to %s", URL)
+    try:
+        await room.connect(URL, TOKEN)
+        logging.info("connected to room %s", room.name)
+    except livekit.ConnectError as e:
+        logging.error("failed to connect to the room: %s", e)
+        return False
+
+    # publish a track
+    source = livekit.VideoSource()
+    source_task = asyncio.create_task(publish_frames(source))
+
+    track = livekit.LocalVideoTrack.create_video_track("hue", source)
+    options = livekit.TrackPublishOptions()
+    options.source = livekit.TrackSource.SOURCE_CAMERA
+    publication = await room.local_participant.publish_track(track, options)
+    logging.info("published track %s", publication.sid)
+
+    try:
+        await room.run()
+    except asyncio.CancelledError:
+        logging.info("closing the room")
+        await room.close()
+
+
+if __name__ == "__main__":
+    logging.basicConfig(level=logging.INFO, handlers=[
+                        logging.FileHandler("publish_hue.log"), logging.StreamHandler()])
+
+    loop = asyncio.get_event_loop()
+    main_task = asyncio.ensure_future(main())
+    for signal in [SIGINT, SIGTERM]:
+        loop.add_signal_handler(signal, main_task.cancel)
+    try:
+        loop.run_until_complete(main_task)
+    finally:
+        loop.close()

examples/publish_hue/room.py

@@ -8,7 +8,7 @@
 from ._proto.track_pb2 import (TrackKind, TrackSource, StreamState)
 from ._proto.room_pb2 import (TrackPublishOptions)
 
-from .room import Room
+from .room import (Room, ConnectError)
 from .participant import (Participant, LocalParticipant, RemoteParticipant)
 from .track import (Track, LocalAudioTrack, LocalVideoTrack,
                     RemoteAudioTrack, RemoteVideoTrack)

livekit/__init__.py

@@ -108,5 +108,5 @@ def __init__(self, handle: int):
 
     def __del__(self):
         if self.handle != INVALID_HANDLE:
-            ffi_lib.livekit_ffi_drop_handle(
-                c_size_t(self.handle))  # TODO Assert
+            assert ffi_lib.livekit_ffi_drop_handle(
+                c_size_t(self.handle))

livekit/_ffi_client.py

@@ -19,9 +19,8 @@ def __init__(self, message: str):
 
 
 class Participant():
-    def __init__(self, info: proto_participant.ParticipantInfo, room: weakref.ref['Room']):
+    def __init__(self, info: proto_participant.ParticipantInfo):
         self._info = info
-        self._room = room
         self.tracks: dict[str, TrackPublication] = {}
 
     @property
@@ -43,7 +42,8 @@ def metadata(self) -> str:
 
 class LocalParticipant(Participant):
     def __init__(self, info: proto_participant.ParticipantInfo, room: weakref.ref['Room']):
-        super().__init__(info, room)
+        super().__init__(info)
+        self._room = room
 
     async def publish_track(self, track: Track, options: TrackPublishOptions) -> TrackPublication:
         if not isinstance(track, LocalAudioTrack) and not isinstance(track, LocalVideoTrack):
@@ -56,15 +56,15 @@ async def publish_track(self, track: Track, options: TrackPublishOptions) -> Tra
         req = proto_ffi.FfiRequest()
         req.publish_track.track_handle.id = track._ffi_handle.handle
         req.publish_track.room_handle.id = room._ffi_handle.handle
-        req.publish_track.options = options
+        req.publish_track.options.CopyFrom(options)
 
         ffi_client = FfiClient()
         resp = ffi_client.request(req)
         future = asyncio.Future()
 
         @ffi_client.on('publish_track')
         def on_publish_callback(cb: proto_room.PublishTrackCallback):
-            if cb.async_id == resp.async_id:
+            if cb.async_id == resp.publish_track.async_id:
                 future.set_result(cb)
                 ffi_client.remove_listener(
                     'publish_track', on_publish_callback)
@@ -82,5 +82,5 @@ def on_publish_callback(cb: proto_room.PublishTrackCallback):
 
 
 class RemoteParticipant(Participant):
-    def __init__(self, info: proto_participant.ParticipantInfo, room: weakref.ref['Room']):
-        super().__init__(info, room)
+    def __init__(self, info: proto_participant.ParticipantInfo):
+        super().__init__(info)