fix async and unused type errors

Author: lukasIO

livekit-api/livekit/api/access_token.py

@@ -278,9 +278,9 @@ def verify(self, token: str, *, verify_signature: bool = True) -> Claims:
         return grant_claims
 
 
-def camel_to_snake(t: str):
+def camel_to_snake(t: str) -> str:
     return re.sub(r"(?<!^)(?=[A-Z])", "_", t).lower()
 
 
-def snake_to_lower_camel(t: str):
+def snake_to_lower_camel(t: str) -> str:
     return "".join(word.capitalize() if i else word for i, word in enumerate(t.split("_")))

livekit-api/livekit/api/livekit_api.py

@@ -6,7 +6,7 @@
 from .sip_service import SipService
 from .agent_dispatch_service import AgentDispatchService
 from .connector_service import ConnectorService
-from typing import Optional
+from typing import Any, Optional
 
 
 class LiveKitAPI:
@@ -96,21 +96,21 @@ def connector(self) -> ConnectorService:
         """Instance of the ConnectorService"""
         return self._connector
 
-    async def aclose(self):
+    async def aclose(self) -> None:
         """Close the API client
 
         Call this before your application exits or when the API client is no longer needed."""
         # we do not close custom sessions, that's up to the caller
-        if not self._custom_session:
+        if not self._custom_session and self._session is not None:
             await self._session.close()
 
-    async def __aenter__(self):
+    async def __aenter__(self) -> "LiveKitAPI":
         """@private
 
         Support for `async with`"""
         return self
 
-    async def __aexit__(self, exc_type, exc_val, exc_tb):
+    async def __aexit__(self, exc_type: Any, exc_val: Any, exc_tb: Any) -> None:
         """@private
 
         Support for `async with`"""

livekit-rtc/livekit/rtc/audio_stream.py

@@ -65,7 +65,7 @@ def __init__(
         num_channels: int = 1,
         frame_size_ms: int | None = None,
         noise_cancellation: Optional[NoiseCancellationOptions | FrameProcessor[AudioFrame]] = None,
-        **kwargs,
+        **kwargs: Any,
     ) -> None:
         """Initialize an `AudioStream` instance.
 
@@ -266,7 +266,7 @@ def _create_owned_stream_from_participant(
         resp = FfiClient.instance.request(req)
         return resp.audio_stream_from_participant.stream
 
-    async def _run(self):
+    async def _run(self) -> None:
         while True:
             event = await self._ffi_queue.wait_for(self._is_event)
             audio_event: proto_audio_frame.AudioStreamEvent = event.audio_stream_event

livekit-rtc/livekit/rtc/data_stream.py

@@ -66,10 +66,10 @@ def __init__(
         )
         self._queue: asyncio.Queue[proto_DataStream.Chunk | None] = asyncio.Queue()
 
-    async def _on_chunk_update(self, chunk: proto_DataStream.Chunk):
+    async def _on_chunk_update(self, chunk: proto_DataStream.Chunk) -> None:
         await self._queue.put(chunk)
 
-    async def _on_stream_close(self, trailer: proto_DataStream.Trailer):
+    async def _on_stream_close(self, trailer: proto_DataStream.Trailer) -> None:
         self.info.attributes = self.info.attributes or {}
         self.info.attributes.update(trailer.attributes)
         await self._queue.put(None)
@@ -114,10 +114,10 @@ def __init__(self, header: proto_DataStream.Header, capacity: int = 0) -> None:
         )
         self._queue: asyncio.Queue[proto_DataStream.Chunk | None] = asyncio.Queue(capacity)
 
-    async def _on_chunk_update(self, chunk: proto_DataStream.Chunk):
+    async def _on_chunk_update(self, chunk: proto_DataStream.Chunk) -> None:
         await self._queue.put(chunk)
 
-    async def _on_stream_close(self, trailer: proto_DataStream.Trailer):
+    async def _on_stream_close(self, trailer: proto_DataStream.Trailer) -> None:
         self.info.attributes = self.info.attributes or {}
         self.info.attributes.update(trailer.attributes)
         await self._queue.put(None)
@@ -166,7 +166,7 @@ def __init__(
         self._sender_identity = sender_identity or self._local_participant.identity
         self._closed = False
 
-    async def _send_header(self):
+    async def _send_header(self) -> None:
         req = proto_ffi.FfiRequest(
             send_stream_header=proto_room.SendStreamHeaderRequest(
                 header=self._header,
@@ -188,7 +188,7 @@ async def _send_header(self):
         if cb.send_stream_header.error:
             raise ConnectionError(cb.send_stream_header.error)
 
-    async def _send_chunk(self, chunk: proto_DataStream.Chunk):
+    async def _send_chunk(self, chunk: proto_DataStream.Chunk) -> None:
         if self._closed:
             raise RuntimeError(f"Cannot send chunk after stream is closed: {chunk}")
         req = proto_ffi.FfiRequest(
@@ -212,7 +212,7 @@ async def _send_chunk(self, chunk: proto_DataStream.Chunk):
         if cb.send_stream_chunk.error:
             raise ConnectionError(cb.send_stream_chunk.error)
 
-    async def _send_trailer(self, trailer: proto_DataStream.Trailer):
+    async def _send_trailer(self, trailer: proto_DataStream.Trailer) -> None:
         req = proto_ffi.FfiRequest(
             send_stream_trailer=proto_room.SendStreamTrailerRequest(
                 trailer=trailer,
@@ -233,7 +233,7 @@ async def _send_trailer(self, trailer: proto_DataStream.Trailer):
         if cb.send_stream_chunk.error:
             raise ConnectionError(cb.send_stream_trailer.error)
 
-    async def aclose(self, *, reason: str = "", attributes: Optional[Dict[str, str]] = None):
+    async def aclose(self, *, reason: str = "", attributes: Optional[Dict[str, str]] = None) -> None:
         if self._closed:
             raise RuntimeError("Stream already closed")
         self._closed = True
@@ -281,7 +281,7 @@ def __init__(
         )
         self._write_lock = asyncio.Lock()
 
-    async def write(self, text: str):
+    async def write(self, text: str) -> None:
         async with self._write_lock:
             for chunk in split_utf8(text, STREAM_CHUNK_SIZE):
                 content = chunk
@@ -333,7 +333,7 @@ def __init__(
         )
         self._write_lock = asyncio.Lock()
 
-    async def write(self, data: bytes):
+    async def write(self, data: bytes) -> None:
         async with self._write_lock:
             chunked_data = [
                 data[i : i + STREAM_CHUNK_SIZE] for i in range(0, len(data), STREAM_CHUNK_SIZE)

livekit-rtc/livekit/rtc/participant.py

@@ -49,6 +49,7 @@
 from .rpc import RpcInvocationData
 from .data_stream import (
     TextStreamWriter,
+    TextStreamInfo,
     ByteStreamWriter,
     ByteStreamInfo,
     STREAM_CHUNK_SIZE,
@@ -160,7 +161,7 @@ def __init__(
     ) -> None:
         super().__init__(owned_info)
         self._room_queue = room_queue
-        self._track_publications: dict[str, LocalTrackPublication] = {}  # type: ignore
+        self._track_publications: dict[str, LocalTrackPublication] = {}
         self._rpc_handlers: Dict[str, RpcHandler] = {}
 
     @property
@@ -327,7 +328,7 @@ async def perform_rpc(
         if cb.perform_rpc.HasField("error"):
             raise RpcError._from_proto(cb.perform_rpc.error)
 
-        return cb.perform_rpc.payload
+        return cast(str, cb.perform_rpc.payload)
 
     def register_rpc_method(
         self,
@@ -587,7 +588,7 @@ async def send_text(
         topic: str = "",
         attributes: Optional[Dict[str, str]] = None,
         reply_to_id: str | None = None,
-    ):
+    ) -> TextStreamInfo:
         total_size = len(text.encode())
         writer = await self.stream_text(
             destination_identities=destination_identities,
@@ -743,7 +744,7 @@ def __repr__(self) -> str:
 class RemoteParticipant(Participant):
     def __init__(self, owned_info: proto_participant.OwnedParticipant) -> None:
         super().__init__(owned_info)
-        self._track_publications: dict[str, RemoteTrackPublication] = {}  # type: ignore
+        self._track_publications: dict[str, RemoteTrackPublication] = {}
 
     @property
     def track_publications(self) -> Mapping[str, RemoteTrackPublication]:

livekit-rtc/livekit/rtc/room.py

@@ -470,10 +470,10 @@ def on_participant_connected(participant):
         if options.rtc_config:
             req.connect.options.rtc_config.ice_transport_type = (
                 options.rtc_config.ice_transport_type
-            )  # type: ignore
+            )
             req.connect.options.rtc_config.continual_gathering_policy = (
                 options.rtc_config.continual_gathering_policy
-            )  # type: ignore
+            )
             req.connect.options.rtc_config.ice_servers.extend(options.rtc_config.ice_servers)
 
         # subscribe before connecting so we don't miss any events
@@ -540,25 +540,25 @@ async def get_rtc_stats(self) -> RtcStats:
 
         return RtcStats(publisher_stats=publisher_stats, subscriber_stats=subscriber_stats)
 
-    def register_byte_stream_handler(self, topic: str, handler: ByteStreamHandler):
+    def register_byte_stream_handler(self, topic: str, handler: ByteStreamHandler) -> None:
         existing_handler = self._byte_stream_handlers.get(topic)
         if existing_handler is None:
             self._byte_stream_handlers[topic] = handler
         else:
             raise ValueError("byte stream handler for topic '%s' already set" % topic)
 
-    def unregister_byte_stream_handler(self, topic: str):
+    def unregister_byte_stream_handler(self, topic: str) -> None:
         if self._byte_stream_handlers.get(topic):
             self._byte_stream_handlers.pop(topic)
 
-    def register_text_stream_handler(self, topic: str, handler: TextStreamHandler):
+    def register_text_stream_handler(self, topic: str, handler: TextStreamHandler) -> None:
         existing_handler = self._text_stream_handlers.get(topic)
         if existing_handler is None:
             self._text_stream_handlers[topic] = handler
         else:
             raise ValueError("text stream handler for topic '%s' already set" % topic)
 
-    def unregister_text_stream_handler(self, topic: str):
+    def unregister_text_stream_handler(self, topic: str) -> None:
         if self._text_stream_handlers.get(topic):
             self._text_stream_handlers.pop(topic)
 
@@ -618,7 +618,7 @@ async def _listen_task(self) -> None:
         await self._drain_rpc_invocation_tasks()
         await self._drain_data_stream_tasks()
 
-    def _on_rpc_method_invocation(self, rpc_invocation: RpcMethodInvocationEvent):
+    def _on_rpc_method_invocation(self, rpc_invocation: RpcMethodInvocationEvent) -> None:
         if self._local_participant is None:
             return
 
@@ -636,7 +636,7 @@ def _on_rpc_method_invocation(self, rpc_invocation: RpcMethodInvocationEvent):
             self._rpc_invocation_tasks.add(task)
             task.add_done_callback(self._rpc_invocation_tasks.discard)
 
-    def _on_room_event(self, event: proto_room.RoomEvent):
+    def _on_room_event(self, event: proto_room.RoomEvent) -> None:
         which = event.WhichOneof("message")
         if which == "participant_connected":
             rparticipant = self._create_remote_participant(event.participant_connected.info)
@@ -905,7 +905,7 @@ def _on_room_event(self, event: proto_room.RoomEvent):
 
     def _handle_stream_header(
         self, header: proto_room.DataStream.Header, participant_identity: str
-    ):
+    ) -> None:
         stream_type = header.WhichOneof("content_header")
         if stream_type == "text_header":
             text_stream_handler = self._text_stream_handlers.get(header.topic)
@@ -935,7 +935,7 @@ def _handle_stream_header(
             logging.warning("received unknown header type, %s", stream_type)
         pass
 
-    async def _handle_stream_chunk(self, chunk: proto_room.DataStream.Chunk):
+    async def _handle_stream_chunk(self, chunk: proto_room.DataStream.Chunk) -> None:
         text_reader = self._text_stream_readers.get(chunk.stream_id)
         file_reader = self._byte_stream_readers.get(chunk.stream_id)
 
@@ -944,7 +944,7 @@ async def _handle_stream_chunk(self, chunk: proto_room.DataStream.Chunk):
         elif file_reader:
             await file_reader._on_chunk_update(chunk)
 
-    async def _handle_stream_trailer(self, trailer: proto_room.DataStream.Trailer):
+    async def _handle_stream_trailer(self, trailer: proto_room.DataStream.Trailer) -> None:
         text_reader = self._text_stream_readers.get(trailer.stream_id)
         file_reader = self._byte_stream_readers.get(trailer.stream_id)