Data tracks support

examples/data_tracks/publisher.py

@@ -0,0 +1,70 @@
+import os
+import logging
+import asyncio
+import time
+from signal import SIGINT, SIGTERM
+from livekit import rtc
+
+# Set the following environment variables with your own values
+TOKEN = os.environ.get("LIVEKIT_TOKEN")
+URL = os.environ.get("LIVEKIT_URL")
+
+
+async def read_sensor() -> bytes:
+    # Dynamically read some sensor data...
+    return bytes([0xFA] * 256)
+
+
+async def push_frames(track: rtc.LocalDataTrack):
+    while True:
+        logging.info("Pushing frame")
+        data = await read_sensor()
+        try:
+            frame = rtc.DataTrackFrame(payload=data, user_timestamp=int(time.time() * 1000))
+            track.try_push(frame)
+        except rtc.PushFrameError as e:
+            logging.error("Failed to push frame: %s", e)
+        await asyncio.sleep(0.5)
+
+
+async def main(room: rtc.Room):
+    logging.basicConfig(level=logging.INFO)
+    logger = logging.getLogger(__name__)
+
+    await room.connect(URL, TOKEN)
+    logger.info("connected to room %s", room.name)
+
+    track = await room.local_participant.publish_data_track(name="my_sensor_data")
+    await push_frames(track)
+
+
+if __name__ == "__main__":
+    logging.basicConfig(
+        level=logging.INFO,
+        handlers=[
+            logging.FileHandler("publisher.log"),
+            logging.StreamHandler(),
+        ],
+    )
+
+    loop = asyncio.get_event_loop()
+    room = rtc.Room(loop=loop)
+
+    main_task = asyncio.ensure_future(main(room))
+
+    async def cleanup():
+        main_task.cancel()
+        try:
+            await main_task
+        except asyncio.CancelledError:
+            pass
+        await room.disconnect()
+        loop.stop()
+
+    for signal in [SIGINT, SIGTERM]:
+        loop.add_signal_handler(signal, lambda: asyncio.ensure_future(cleanup()))
+
+    try:
+        loop.run_forever()
+    finally:
+        loop.close()

examples/data_tracks/subscriber.py

@@ -0,0 +1,69 @@
+import os
+import logging
+import asyncio
+import time
+from signal import SIGINT, SIGTERM
+from livekit import rtc
+
+# Set the following environment variables with your own values
+TOKEN = os.environ.get("LIVEKIT_TOKEN")
+URL = os.environ.get("LIVEKIT_URL")
+
+
+async def subscribe(track: rtc.RemoteDataTrack):
+    logging.info(
+        "Subscribing to '%s' published by '%s'",
+        track.info.name,
+        track.publisher_identity,
+    )
+    try:
+        async for frame in track.subscribe():
+            logging.info("Received frame (%d bytes)", len(frame.payload))
+
+            if frame.user_timestamp is not None:
+                latency = (int(time.time() * 1000) - frame.user_timestamp) / 1000.0
+                logging.info("Latency: %.3f s", latency)
+    except rtc.SubscribeDataTrackError as e:
+        logging.error("Failed to subscribe to '%s': %s", track.info.name, e.message)
+
+
+async def main(room: rtc.Room):
+    logging.basicConfig(level=logging.INFO)
+    logger = logging.getLogger(__name__)
+
+    active_tasks = []
+
+    @room.on("data_track_published")
+    def on_data_track_published(track: rtc.RemoteDataTrack):
+        task = asyncio.create_task(subscribe(track))
+        active_tasks.append(task)
+        task.add_done_callback(lambda _: active_tasks.remove(task))
+
+    await room.connect(URL, TOKEN)
+    logger.info("connected to room %s", room.name)
+
+
+if __name__ == "__main__":
+    logging.basicConfig(
+        level=logging.INFO,
+        handlers=[
+            logging.FileHandler("subscriber.log"),
+            logging.StreamHandler(),
+        ],
+    )
+
+    loop = asyncio.get_event_loop()
+    room = rtc.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()

livekit-rtc/generate_proto.sh

@@ -31,6 +31,7 @@ protoc \
     $FFI_PROTOCOL/participant.proto \
     $FFI_PROTOCOL/room.proto \
     $FFI_PROTOCOL/track.proto \
+    $FFI_PROTOCOL/data_track.proto \
     $FFI_PROTOCOL/video_frame.proto \
     $FFI_PROTOCOL/e2ee.proto \
     $FFI_PROTOCOL/stats.proto \
@@ -41,5 +42,5 @@ protoc \
 touch -a "$FFI_OUT_PYTHON/__init__.py"
 
 for f in "$FFI_OUT_PYTHON"/*.py "$FFI_OUT_PYTHON"/*.pyi; do
-    perl -i -pe 's|^(import (audio_frame_pb2\|ffi_pb2\|handle_pb2\|participant_pb2\|room_pb2\|track_pb2\|video_frame_pb2\|e2ee_pb2\|stats_pb2\|rpc_pb2\|track_publication_pb2\|data_stream_pb2))|from . $1|g' "$f"
+    perl -i -pe 's|^(import (audio_frame_pb2\|ffi_pb2\|handle_pb2\|participant_pb2\|room_pb2\|track_pb2\|video_frame_pb2\|e2ee_pb2\|stats_pb2\|rpc_pb2\|track_publication_pb2\|data_stream_pb2\|data_track_pb2))|from . $1|g' "$f"
 done

livekit-rtc/livekit/rtc/__init__.py

@@ -108,6 +108,15 @@
     ByteStreamWriter,
     ByteStreamReader,
 )
+from .data_track import (
+    LocalDataTrack,
+    RemoteDataTrack,
+    DataTrackStream,
+    DataTrackFrame,
+    DataTrackInfo,
+    PushFrameError,
+    SubscribeDataTrackError,
+)
 from .frame_processor import FrameProcessor
 
 __all__ = [
@@ -186,6 +195,13 @@
     "ByteStreamWriter",
     "AudioProcessingModule",
     "FrameProcessor",
+    "LocalDataTrack",
+    "RemoteDataTrack",
+    "DataTrackStream",
+    "DataTrackFrame",
+    "DataTrackInfo",
+    "PushFrameError",
+    "SubscribeDataTrackError",
     "__version__",
 ]