fix: correctly handle events order & allow to specify eventloops (#60)

Author: theomonnom

README.md

@@ -25,10 +25,6 @@ $ pip install livekit
 ```python
 async def main():
     room = livekit.Room()
-    # By default, autosubscribe is enabled. The participant will be subscribed to
-    # all published tracks in the room
-    await room.connect(URL, TOKEN)
-    logging.info("connected to room %s", room.name)
 
     # participants and tracks that are already available in the room
     # participant_connected and track_published events will *not* be emitted for them
@@ -41,21 +37,23 @@ async def main():
         logging.info(
             "participant connected: %s %s", participant.sid, participant.identity)
 
-    video_stream = None
+    async def receive_frames(stream: livekit.VideoStream):
+        async for frame in video_stream:
+            # received a video frame from the track, process it here
+            pass
 
     # track_subscribed is emitted whenever the local participant is subscribed to a new track
     @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:
-            nonlocal video_stream
             video_stream = livekit.VideoStream(track)
+            asyncio.ensure_future(receive_frames(video_stream))
 
-            async for frame in video_stream:
-                # received a video frame from the track, process it here
-                pass
-
-    await room.run()
+    # By default, autosubscribe is enabled. The participant will be subscribed to
+    # all published tracks in the room
+    await room.connect(URL, TOKEN)
+    logging.info("connected to room %s", room.name)
 ```
 
 ## Examples

examples/basic_room.py

@@ -6,11 +6,10 @@
 import livekit
 
 URL = 'ws://localhost:7880'
-TOKEN = 'eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJleHAiOjE5MDY2MTMyODgsImlzcyI6IkFQSVRzRWZpZFpqclFvWSIsIm5hbWUiOiJuYXRpdmUiLCJuYmYiOjE2NzI2MTMyODgsInN1YiI6Im5hdGl2ZSIsInZpZGVvIjp7InJvb20iOiJ0ZXN0Iiwicm9vbUFkbWluIjp0cnVlLCJyb29tQ3JlYXRlIjp0cnVlLCJyb29tSm9pbiI6dHJ1ZSwicm9vbUxpc3QiOnRydWV9fQ.uSNIangMRu8jZD5mnRYoCHjcsQWCrJXgHCs0aNIgBFY'
+TOKEN = 'eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJleHAiOjE5MDY2MTMyODgsImlzcyI6IkFQSVRzRWZpZFpqclFvWSIsIm5hbWUiOiJuYXRpdmUiLCJuYmYiOjE2NzI2MTMyODgsInN1YiI6Im5hdGl2ZSIsInZpZGVvIjp7InJvb20iOiJ0ZXN0Iiwicm9vbUFkbWluIjp0cnVlLCJyb29tQ3JlYXRlIjp0cnVlLCJyb29tSm9pbiI6dHJ1ZSwicm9vbUxpc3QiOnRydWV9fQ.uSNIangMRu8jZD5mnRYoCHjcsQWCrJXgHCs0aNIgBFY'  # noqa
 
 
-async def main() -> None:
-    room = livekit.Room()
+async def main(room: livekit.Room) -> None:
 
     @room.listens_to("participant_connected")
     def on_participant_connected(participant: livekit.RemoteParticipant) -> None:
@@ -47,24 +46,18 @@ def on_track_unpublished(publication: livekit.RemoteTrackPublication,
                              participant: livekit.RemoteParticipant):
         logging.info("track unpublished: %s", publication.sid)
 
-    # Keep a reference to the streams, otherwise they will be disposed
-    audio_stream = None
-    video_stream = None
-
     @room.listens_to("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:
-            nonlocal video_stream
-            video_stream = livekit.VideoStream(track)
+            _video_stream = livekit.VideoStream(track)
             # video_stream is an async iterator that yields VideoFrame
         elif track.kind == livekit.TrackKind.KIND_AUDIO:
             print("Subscribed to an Audio Track")
-            nonlocal audio_stream
-            audio_stream = livekit.AudioStream(track)
-            # audio_stream is an async iterator that yields AudioFrame 
+            _audio_stream = livekit.AudioStream(track)
+            # audio_stream is an async iterator that yields AudioFrame
 
     @room.listens_to("track_unsubscribed")
     def on_track_unsubscribed(track: livekit.Track,
@@ -120,32 +113,32 @@ def on_reconnecting() -> None:
     def on_reconnected() -> None:
         logging.info("reconnected")
 
-    try:
-        logging.info("connecting to %s", URL)
-        await room.connect(URL, TOKEN)
-        logging.info("connected to room %s", room.name)
-
-        await room.local_participant.publish_data("hello world")
+    await room.connect(URL, TOKEN)
+    logging.info("connected to room %s", room.name)
+    logging.info("participants: %s", room.participants)
 
-        logging.info("participants: %s", room.participants)
-
-        await room.run()
-    except livekit.ConnectError as e:
-        logging.error("failed to connect to the room: %s", e)
-    except asyncio.CancelledError:
-        logging.info("closing the room")
-        await room.disconnect()
+    await asyncio.sleep(2)
+    await room.local_participant.publish_data("hello world")
 
 
 if __name__ == "__main__":
     logging.basicConfig(level=logging.INFO, handlers=[
-                        logging.FileHandler("basic_room.log"), logging.StreamHandler()])
+                        logging.FileHandler("basic_room.log"),
+                        logging.StreamHandler()])
 
     loop = asyncio.get_event_loop()
-    main_task = asyncio.ensure_future(main())
+    room = livekit.Room(loop=loop)
+
+    async def cleanup():
+        await room.disconnect()
+        loop.stop()
+
+    asyncio.ensure_future(main(room))
     for signal in [SIGINT, SIGTERM]:
-        loop.add_signal_handler(signal, main_task.cancel)
+        loop.add_signal_handler(
+            signal, lambda: asyncio.ensure_future(cleanup()))
+
     try:
-        loop.run_until_complete(main_task)
+        loop.run_forever()
     finally:
         loop.close()

examples/e2ee.py

@@ -1,5 +1,4 @@
 import asyncio
-import colorsys
 import logging
 from signal import SIGINT, SIGTERM
 
@@ -10,44 +9,55 @@
 URL = 'ws://localhost:7880'
 TOKEN = 'eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJleHAiOjE5MDY2MTMyODgsImlzcyI6IkFQSVRzRWZpZFpqclFvWSIsIm5hbWUiOiJuYXRpdmUiLCJuYmYiOjE2NzI2MTMyODgsInN1YiI6Im5hdGl2ZSIsInZpZGVvIjp7InJvb20iOiJ0ZXN0Iiwicm9vbUFkbWluIjp0cnVlLCJyb29tQ3JlYXRlIjp0cnVlLCJyb29tSm9pbiI6dHJ1ZSwicm9vbUxpc3QiOnRydWV9fQ.uSNIangMRu8jZD5mnRYoCHjcsQWCrJXgHCs0aNIgBFY'  # noqa
 
+# ("livekitrocks") this is our shared key, it must match the one used by your clients
+SHARED_KEY = b"livekitrocks"
 
-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)
+async def draw_cube(source: livekit.VideoSource):
+    W, H, MID_W, MID_H = 1280, 720, 640, 360
+    cube_size = 60
+    vertices = (np.array([[-1, -1, -1], [1, -1, -1], [1, 1, -1], [-1, 1, -1],
+                [-1, -1, 1], [1, -1, 1], [1, 1, 1], [-1, 1, 1]]) * cube_size)
+    edges = [[0, 1], [1, 2], [2, 3], [3, 0], [4, 5], [5, 6],
+             [6, 7], [7, 4], [0, 4], [1, 5], [2, 6], [3, 7]]
 
-    framerate = 1 / 30
-    hue = 0.0
+    frame = livekit.ArgbFrame(livekit.VideoFormatType.FORMAT_ARGB, W, H)
+    arr = np.ctypeslib.as_array(frame.data)
+    angle = 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 = [(x * 255) for x in rgb]  # type: ignore
-
-        argb_color = np.array(rgb + [255], dtype=np.uint8)
-        arr.flat[::4] = argb_color[0]
-        arr.flat[1::4] = argb_color[1]
-        arr.flat[2::4] = argb_color[2]
-        arr.flat[3::4] = argb_color[3]
-
-        source.capture_frame(frame)
-
-        hue += framerate / 3  # 3s for a full cycle
-        if hue >= 1.0:
-            hue = 0.0
-
-        try:
-            await asyncio.sleep(framerate)
-        except asyncio.CancelledError:
-            break
-
-
-async def main():
-    room = livekit.Room()
-
+        start_time = asyncio.get_event_loop().time()
+        arr.fill(0)
+        rot = np.dot(np.array([[1, 0, 0],
+                               [0, np.cos(angle), -np.sin(angle)],
+                               [0, np.sin(angle), np.cos(angle)]]),
+                     np.array([[np.cos(angle), 0, np.sin(angle)],
+                               [0, 1, 0],
+                               [-np.sin(angle), 0, np.cos(angle)]]))
+        proj_points = [[int(pt[0] / (pt[2] / 200 + 1)), int(pt[1] / (pt[2] / 200 + 1))]
+                       for pt in np.dot(vertices, rot)]
+
+        for e in edges:
+            x1, y1, x2, y2 = *proj_points[e[0]], *proj_points[e[1]]
+            for t in np.linspace(0, 1, 100):
+                x, y = int(MID_W + (1 - t) * x1 + t *
+                           x2), int(MID_H + (1 - t) * y1 + t * y2)
+                for dx in [-1, 0, 1]:
+                    for dy in [-1, 0, 1]:
+                        if 0 <= x + dx < W and 0 <= y + dy < H:
+                            idx = (y + dy) * W * 4 + (x + dx) * 4
+                            arr[idx:idx+4] = [255, 255, 255, 255]
+
+        f = livekit.VideoFrame(
+            0, livekit.VideoRotation.VIDEO_ROTATION_0, frame.to_i420())
+        source.capture_frame(f)
+        angle += 0.02
+
+        code_duration = asyncio.get_event_loop().time() - start_time
+        await asyncio.sleep(1 / 30 - code_duration)
+
+
+async def main(room: livekit.Room):
     @room.listens_to("e2ee_state_changed")
     def on_e2ee_state_changed(participant: livekit.Participant,
                               state: livekit.EncryptionState) -> None:
@@ -56,7 +66,7 @@ def on_e2ee_state_changed(participant: livekit.Participant,
     logging.info("connecting to %s", URL)
     try:
         e2ee_options = livekit.E2EEOptions()
-        e2ee_options.key_provider_options.shared_key = b"livekitrocks"  # this is our e2ee key
+        e2ee_options.key_provider_options.shared_key = SHARED_KEY
 
         await room.connect(URL, TOKEN, options=livekit.RoomOptions(
             auto_subscribe=True,
@@ -68,33 +78,34 @@ def on_e2ee_state_changed(participant: livekit.Participant,
         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)
+    track = livekit.LocalVideoTrack.create_video_track("cube", 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")
-        source_task.cancel()
-        await source_task
-        await room.disconnect()
-
+    asyncio.ensure_future(draw_cube(source))
 
 if __name__ == "__main__":
     logging.basicConfig(level=logging.INFO, handlers=[
-                        logging.FileHandler("publish_hue.log"), logging.StreamHandler()])
+                        logging.FileHandler("e2ee.log"),
+                        logging.StreamHandler()])
 
     loop = asyncio.get_event_loop()
-    main_task = asyncio.ensure_future(main())
+    room = livekit.Room(loop=loop)
+
+    async def cleanup():
+        await room.disconnect()
+        loop.stop()
+
+    asyncio.ensure_future(main(room))
     for signal in [SIGINT, SIGTERM]:
-        loop.add_signal_handler(signal, main_task.cancel)
+        loop.add_signal_handler(
+            signal, lambda: asyncio.ensure_future(cleanup()))
+
     try:
-        loop.run_until_complete(main_task)
+        loop.run_forever()
     finally:
         loop.close()

examples/face_landmark/face_landmark.py

@@ -1,6 +1,7 @@
 import asyncio
+import logging
 import os
-import signal
+from signal import SIGINT, SIGTERM
 
 import cv2
 import mediapipe as mp
@@ -11,7 +12,7 @@
 import livekit
 
 URL = 'ws://localhost:7880'
-TOKEN = 'eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJleHAiOjE5MDY2MTMyODgsImlzcyI6IkFQSVRzRWZpZFpqclFvWSIsIm5hbWUiOiJuYXRpdmUiLCJuYmYiOjE2NzI2MTMyODgsInN1YiI6Im5hdGl2ZSIsInZpZGVvIjp7InJvb20iOiJ0ZXN0Iiwicm9vbUFkbWluIjp0cnVlLCJyb29tQ3JlYXRlIjp0cnVlLCJyb29tSm9pbiI6dHJ1ZSwicm9vbUxpc3QiOnRydWV9fQ.uSNIangMRu8jZD5mnRYoCHjcsQWCrJXgHCs0aNIgBFY'
+TOKEN = 'eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJleHAiOjE5MDY2MTMyODgsImlzcyI6IkFQSVRzRWZpZFpqclFvWSIsIm5hbWUiOiJuYXRpdmUiLCJuYmYiOjE2NzI2MTMyODgsInN1YiI6Im5hdGl2ZSIsInZpZGVvIjp7InJvb20iOiJ0ZXN0Iiwicm9vbUFkbWluIjp0cnVlLCJyb29tQ3JlYXRlIjp0cnVlLCJyb29tSm9pbiI6dHJ1ZSwicm9vbUxpc3QiOnRydWV9fQ.uSNIangMRu8jZD5mnRYoCHjcsQWCrJXgHCs0aNIgBFY'  # noqa
 
 tasks = set()
 
@@ -40,7 +41,9 @@ def draw_landmarks_on_image(rgb_image, detection_result):
         # 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
+            landmark_pb2.NormalizedLandmark(
+                x=landmark.x, y=landmark.y, z=landmark.z)
+            for landmark in face_landmarks
         ])
 
         solutions.drawing_utils.draw_landmarks(
@@ -74,7 +77,8 @@ async def frame_loop(video_stream: livekit.VideoStream) -> None:
     async for frame in video_stream:
         buffer = frame.buffer
 
-        if argb_frame is None or argb_frame.width != buffer.width or argb_frame.height != buffer.height:
+        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)
 
@@ -102,39 +106,45 @@ async def frame_loop(video_stream: livekit.VideoStream) -> None:
     cv2.destroyAllWindows()
 
 
-async def main() -> None:
-    room = livekit.Room()
-
-    loop = asyncio.get_event_loop()
-    loop.add_signal_handler(
-        signal.SIGINT, lambda: exit(0))
-
-    await room.connect(URL, TOKEN, livekit.RoomOptions(
-        # Unncomment below to enable E2EE
-        # e2ee=livekit.E2EEOptions(
-        #     key_provider_options=livekit.KeyProviderOptions(
-        #         shared_key=b"livekitrocks"
-        #     )
-        # ),
-    ))
-    print("connected to room: " + room.name)
-
+async def main(room: livekit.Room) -> None:
     video_stream = None
 
     @room.on("track_subscribed")
     def on_track_subscribed(track: livekit.Track, *_):
         if track.kind == livekit.TrackKind.KIND_VIDEO:
             nonlocal video_stream
-            # only process the first stream received
             if video_stream is not None:
+                # only process the first stream received
                 return
+
             print("subscribed to track: " + track.name)
             video_stream = livekit.VideoStream(track)
             task = asyncio.create_task(frame_loop(video_stream))
             tasks.add(task)
             task.add_done_callback(tasks.remove)
 
-    await room.run()
+    await room.connect(URL, TOKEN)
+    print("connected to room: " + room.name)
+
 
 if __name__ == "__main__":
-    asyncio.run(main())
+    logging.basicConfig(level=logging.INFO, handlers=[
+                        logging.FileHandler("face_landmark.log"),
+                        logging.StreamHandler()])
+
+    loop = asyncio.get_event_loop()
+    room = livekit.Room(loop=loop)
+
+    async def cleanup():
+        await room.disconnect()
+        loop.stop()
+
+    asyncio.ensure_future(main(room))
+    for signal in [SIGINT, SIGTERM]:
+        loop.add_signal_handler(
+            signal, lambda: asyncio.ensure_future(cleanup()))
+
+    try:
+        loop.run_forever()
+    finally:
+        loop.close()