from_buffer_copy

Author: theomonnom

livekit-rtc/livekit/rtc/_utils.py

@@ -40,11 +40,8 @@ def task_done_logger(task: asyncio.Task) -> None:
         return
 
 
-def get_address(data: memoryview) -> int:
-    """Get the address of a buffer using ctypes"""
-    nbytes = data.nbytes
-    buffer = (ctypes.c_int8 * nbytes).from_buffer(data)
-    return ctypes.addressof(buffer)
+def get_address(mv: memoryview) -> int:
+    return ctypes.addressof(ctypes.c_char.from_buffer(mv))
 
 
 T = TypeVar("T")

livekit-rtc/livekit/rtc/audio_frame.py

@@ -48,6 +48,8 @@ def __init__(
         Raises:
             ValueError: If the length of `data` is smaller than the required size.
         """
+        data = memoryview(data).cast("B")
+
         if len(data) < num_channels * samples_per_channel * ctypes.sizeof(
             ctypes.c_int16
         ):
@@ -59,9 +61,8 @@ def __init__(
             # can happen if data is bigger than needed
             raise ValueError("data length must be a multiple of sizeof(int16)")
 
-        data = memoryview(data).cast("B")
-        self._data = (ctypes.c_int16 * (len(data) // ctypes.sizeof(ctypes.c_int16)))()
-        ctypes.memmove(self._data, data, len(data))
+        n = len(data) // ctypes.sizeof(ctypes.c_int16)
+        self._data = (ctypes.c_int16 * n).from_buffer_copy(data)
 
         self._sample_rate = sample_rate
         self._num_channels = num_channels