fix remaining type errors

Author: lukasIO

livekit-rtc/livekit/rtc/_ffi_client.py

@@ -24,7 +24,7 @@
 import platform
 import atexit
 import threading
-from typing import Generic, List, Optional, TypeVar
+from typing import Any, Generic, List, Optional, TypeVar
 
 from ._proto import ffi_pb2 as proto_ffi
 from ._utils import Queue, classproperty
@@ -34,7 +34,7 @@
 atexit.register(_resource_files.close)
 
 
-def get_ffi_lib():
+def get_ffi_lib() -> ctypes.CDLL:
     # allow to override the lib path using an env var
     libpath = os.environ.get("LIVEKIT_LIB_PATH", "").strip()
     if libpath:
@@ -92,7 +92,7 @@ def __init__(self, handle: int) -> None:
         self.handle = handle
         self._disposed = False
 
-    def __del__(self):
+    def __del__(self) -> None:
         self.dispose()
 
     @property
@@ -142,7 +142,7 @@ def unsubscribe(self, queue: Queue[T]) -> None:
                     break
 
 
-@ctypes.CFUNCTYPE(None, ctypes.POINTER(ctypes.c_uint8), ctypes.c_size_t)
+@ctypes.CFUNCTYPE(None, ctypes.POINTER(ctypes.c_uint8), ctypes.c_size_t)  # type: ignore[untyped-decorator]
 def ffi_event_callback(
     data_ptr: ctypes.POINTER(ctypes.c_uint8),  # type: ignore
     data_len: ctypes.c_size_t,
@@ -219,7 +219,7 @@ def __init__(self) -> None:
         )
 
         @atexit.register
-        def _dispose_lk_ffi():
+        def _dispose_lk_ffi() -> None:
             ffi_lib.livekit_ffi_dispose()
 
     @property

livekit-rtc/livekit/rtc/_utils.py

@@ -17,16 +17,16 @@
 from collections import deque
 import ctypes
 import random
-from typing import Callable, Generator, Generic, List, TypeVar
+from typing import Any, Callable, Generator, Generic, List, TypeVar
 
 logger = logging.getLogger("livekit")
 
 
 class classproperty(object):
-    def __init__(self, f):
-        self.f = classmethod(f)
+    def __init__(self, f: Callable[..., Any]) -> None:
+        self.f: Any = classmethod(f)
 
-    def __get__(self, *a):
+    def __get__(self, *a: Any) -> Any:
         return self.f.__get__(*a)()
 
 
@@ -118,7 +118,7 @@ async def join(self) -> None:
 _base62_characters = "0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ"
 
 
-def generate_random_base62(length=12):
+def generate_random_base62(length: int = 12) -> str:
     """
     Generate a random base62 encoded string of a specified length.
 

livekit-rtc/livekit/rtc/e2ee.py

@@ -13,7 +13,7 @@
 # limitations under the License.
 
 from dataclasses import dataclass, field
-from typing import List, Optional
+from typing import List, Optional, cast
 
 from ._ffi_client import FfiClient
 from ._proto import e2ee_pb2 as proto_e2ee
@@ -84,7 +84,7 @@ def export_shared_key(self, key_index: int) -> bytes:
         req.e2ee.get_shared_key.key_index = key_index
         resp = FfiClient.instance.request(req)
         key = resp.e2ee.get_shared_key.key
-        return key
+        return cast(bytes, key)
 
     def ratchet_shared_key(self, key_index: int) -> bytes:
         """Ratchets the shared encryption key to a new key.
@@ -107,7 +107,7 @@ def ratchet_shared_key(self, key_index: int) -> bytes:
         resp = FfiClient.instance.request(req)
 
         new_key = resp.e2ee.ratchet_shared_key.new_key
-        return new_key
+        return cast(bytes, new_key)
 
     def set_key(self, participant_identity: str, key: bytes, key_index: int) -> None:
         """Sets the encryption key for a specific participant.
@@ -152,7 +152,7 @@ def export_key(self, participant_identity: str, key_index: int) -> bytes:
         req.e2ee.get_key.key_index = key_index
         resp = FfiClient.instance.request(req)
         key = resp.e2ee.get_key.key
-        return key
+        return cast(bytes, key)
 
     def ratchet_key(self, participant_identity: str, key_index: int) -> bytes:
         """Ratchets the encryption key for a specific participant to a new key.
@@ -176,7 +176,7 @@ def ratchet_key(self, participant_identity: str, key_index: int) -> bytes:
 
         resp = FfiClient.instance.request(req)
         new_key = resp.e2ee.ratchet_key.new_key
-        return new_key
+        return cast(bytes, new_key)
 
 
 class FrameCryptor:

livekit-rtc/livekit/rtc/jupyter.py

@@ -50,7 +50,7 @@ def room_html(url: str, token: str, *, width: str, height: str) -> HTML:
             f'srcdoc="{escaped_content}"></iframe>'
         )
 
-    return HTML(html_text)
+    return HTML(html_text)  # type: ignore[no-untyped-call]
 
 
 def display_room(url: str, token: str, *, width: str = "100%", height: str = "110px") -> None:
@@ -65,4 +65,4 @@ def display_room(url: str, token: str, *, width: str = "100%", height: str = "11
         The rendered HTML will include the provided `url` and `token` in plain text.
         Avoid using sensitive tokens in public notebooks (e.g., tokens with long expiration times).
     """
-    display(room_html(url, token, width=width, height=height))
+    display(room_html(url, token, width=width, height=height))  # type: ignore[no-untyped-call]

livekit-rtc/livekit/rtc/media_devices.py

@@ -22,7 +22,7 @@
 import threading
 
 if TYPE_CHECKING:
-    import sounddevice as sd  # type: ignore[import-not-found, import-untyped]
+    import sounddevice as sd  # type: ignore[import-not-found]
 
 from . import AudioSource
 from .audio_frame import AudioFrame
@@ -163,7 +163,7 @@ def __init__(
         output_device: Optional[int] = None,
         delay_estimator: Optional[_APMDelayEstimator] = None,
     ) -> None:
-        import sounddevice as sd  # type: ignore[import-not-found, import-untyped]
+        import sounddevice as sd
 
         self._sample_rate = sample_rate
         self._num_channels = num_channels
@@ -303,16 +303,17 @@ async def start(self) -> None:
         if self._mixer is None:
             self._mixer = AudioMixer(sample_rate=self._sample_rate, num_channels=self._num_channels)
 
-        async def _playback_loop():
+        async def _playback_loop() -> None:
             """Internal playback loop that consumes frames from the mixer."""
             self._running = True
             self._stream.start()
             try:
-                async for frame in self._mixer:
-                    if not self._running:
-                        break
-                    # Append raw PCM bytes for callback consumption
-                    self._buffer.extend(frame.data.tobytes())
+                if self._mixer is not None:
+                    async for frame in self._mixer:
+                        if not self._running:
+                            break
+                        # Append raw PCM bytes for callback consumption
+                        self._buffer.extend(frame.data.tobytes())
             finally:
                 self._running = False
                 try:
@@ -402,7 +403,7 @@ def list_input_devices(self) -> list[dict[str, Any]]:
         Returns a list of dictionaries with the `sounddevice` metadata and an
         added `index` key corresponding to the device index.
         """
-        import sounddevice as sd  # type: ignore[import-not-found, import-untyped]
+        import sounddevice as sd
 
         devices = sd.query_devices()
         result: list[dict[str, Any]] = []
@@ -413,7 +414,7 @@ def list_input_devices(self) -> list[dict[str, Any]]:
 
     def list_output_devices(self) -> list[dict[str, Any]]:
         """List available output devices with indices."""
-        import sounddevice as sd  # type: ignore[import-not-found, import-untyped]
+        import sounddevice as sd
 
         devices = sd.query_devices()
         result: list[dict[str, Any]] = []
@@ -424,14 +425,14 @@ def list_output_devices(self) -> list[dict[str, Any]]:
 
     def default_input_device(self) -> Optional[int]:
         """Return the default input device index (or None)."""
-        import sounddevice as sd  # type: ignore[import-not-found, import-untyped]
+        import sounddevice as sd
 
         dev = sd.default.device
         return dev[0] if isinstance(dev, (list, tuple)) else None
 
     def default_output_device(self) -> Optional[int]:
         """Return the default output device index (or None)."""
-        import sounddevice as sd  # type: ignore[import-not-found, import-untyped]
+        import sounddevice as sd
 
         dev = sd.default.device
         return dev[1] if isinstance(dev, (list, tuple)) else None
@@ -471,7 +472,7 @@ def open_input(
         Returns:
             InputCapture: Holder with `source`, `apm`, and `aclose()`.
         """
-        import sounddevice as sd  # type: ignore[import-not-found, import-untyped]
+        import sounddevice as sd
 
         loop = self._loop
         source = AudioSource(self._in_sr, self._channels, loop=loop)

pyproject.toml

@@ -71,6 +71,6 @@ exclude = "(build|setup\\.py|livekit-rtc/rust-sdks.*)"
 namespace_packages = true
 explicit_package_bases = true
 mypy_path = "livekit-protocol:livekit-api:livekit-rtc"
-strict = false # TODO re-enable strict checking
+strict = true
 disallow_any_generics = false
 plugins = ["pydantic.mypy"]