wip

Author: theomonnom

livekit-rtc/livekit/rtc/_utils.py

@@ -17,7 +17,7 @@
 from collections import deque
 import ctypes
 import random
-from typing import Callable, Generator, Generic, List, TypeVar
+from typing import Callable, Generator, Generic, List, TypeVar, Union
 
 logger = logging.getLogger("livekit")
 
@@ -40,8 +40,25 @@ def task_done_logger(task: asyncio.Task) -> None:
         return
 
 
+def _buffer_supported_or_raise(
+    data: Union[bytes, bytearray, memoryview],
+) -> None:
+    """Validate a buffer for FFI use.
+
+    Raises clear errors for non-contiguous or sliced memoryviews.
+    """
+    if isinstance(data, memoryview):
+        if not data.contiguous:
+            raise ValueError("memoryview must be contiguous")
+        if data.nbytes != len(data.obj):
+            raise ValueError("sliced memoryviews are not supported")
+    elif not isinstance(data, (bytes, bytearray)):
+        raise TypeError(f"expected bytes, bytearray, or memoryview, got {type(data)}")
+
+
 def get_address(data) -> int:
     if isinstance(data, memoryview):
+        _buffer_supported_or_raise(data)
         if not data.readonly:
             return ctypes.addressof(ctypes.c_char.from_buffer(data))
         data = data.obj

livekit-rtc/livekit/rtc/apm.py

@@ -48,12 +48,15 @@ def process_stream(self, data: AudioFrame) -> None:
         Important:
             Audio frames must be exactly 10 ms in duration.
         """
-        bdata = data.data.cast("b")
+        if isinstance(data._data, bytes) or (
+            isinstance(data._data, memoryview) and data._data.readonly
+        ):
+            data._data = bytearray(data._data)
 
         req = proto_ffi.FfiRequest()
         req.apm_process_stream.apm_handle = self._ffi_handle.handle
-        req.apm_process_stream.data_ptr = get_address(memoryview(bdata))
-        req.apm_process_stream.size = len(bdata)
+        req.apm_process_stream.data_ptr = get_address(data._data)
+        req.apm_process_stream.size = len(data._data)
         req.apm_process_stream.sample_rate = data.sample_rate
         req.apm_process_stream.num_channels = data.num_channels
 
@@ -73,12 +76,15 @@ def process_reverse_stream(self, data: AudioFrame) -> None:
         Important:
             Audio frames must be exactly 10 ms in duration.
         """
-        bdata = data.data.cast("b")
+        if isinstance(data._data, bytes) or (
+            isinstance(data._data, memoryview) and data._data.readonly
+        ):
+            data._data = bytearray(data._data)
 
         req = proto_ffi.FfiRequest()
         req.apm_process_reverse_stream.apm_handle = self._ffi_handle.handle
-        req.apm_process_reverse_stream.data_ptr = get_address(memoryview(bdata))
-        req.apm_process_reverse_stream.size = len(bdata)
+        req.apm_process_reverse_stream.data_ptr = get_address(data._data)
+        req.apm_process_reverse_stream.size = len(data._data)
         req.apm_process_reverse_stream.sample_rate = data.sample_rate
         req.apm_process_reverse_stream.num_channels = data.num_channels
 

livekit-rtc/livekit/rtc/audio_frame.py

@@ -15,7 +15,7 @@
 import ctypes
 from ._ffi_client import FfiHandle
 from ._proto import audio_frame_pb2 as proto_audio
-from ._utils import get_address
+from ._utils import _buffer_supported_or_raise, get_address
 from typing import Any, Union
 
 
@@ -49,8 +49,7 @@ def __init__(
         Raises:
             ValueError: If the length of `data` is smaller than the required size.
         """
-        if isinstance(data, memoryview):
-            data = data.obj  # type: ignore[assignment]
+        _buffer_supported_or_raise(data)
 
         min_size = num_channels * samples_per_channel * ctypes.sizeof(ctypes.c_int16)
         data_len = len(data)

livekit-rtc/livekit/rtc/video_frame.py

@@ -18,7 +18,7 @@
 from ._proto import ffi_pb2 as proto
 from typing import List, Optional
 from ._ffi_client import FfiClient, FfiHandle
-from ._utils import get_address
+from ._utils import _buffer_supported_or_raise, get_address
 
 from typing import Any
 
@@ -48,6 +48,8 @@ def __init__(
                 (e.g., RGBA, BGRA, RGB24, etc.).
             data (Union[bytes, bytearray, memoryview]): The raw pixel data for the video frame.
         """
+        _buffer_supported_or_raise(data)
+
         self._width = width
         self._height = height
         self._type = type