From aa17159f7ea61c67e1142004bc1eed95466c3eee Mon Sep 17 00:00:00 2001 From: Username Date: Tue, 24 Feb 2026 16:45:45 +0100 Subject: [PATCH] client: add return annotations and cache users/channels properties --- src/tuimble/client.py | 52 ++++++++++++++++++++++++++++++------------- 1 file changed, 36 insertions(+), 16 deletions(-) diff --git a/src/tuimble/client.py b/src/tuimble/client.py index aab854a..6f75e17 100644 --- a/src/tuimble/client.py +++ b/src/tuimble/client.py @@ -74,6 +74,10 @@ class MumbleClient: self._mumble = None self._connected = False self._dispatcher: Callable | None = None + self._users_cache: dict[int, User] = {} + self._channels_cache: dict[int, Channel] = {} + self._users_dirty: bool = True + self._channels_dirty: bool = True # Application callbacks (fired via dispatcher) self.on_connected = None @@ -83,14 +87,14 @@ class MumbleClient: self.on_channel_update = None # () self.on_sound_received = None # (user, pcm_data) - def set_dispatcher(self, fn: Callable): + def set_dispatcher(self, fn: Callable) -> None: """Set a function to marshal callbacks into the host event loop. Typically Textual's ``call_from_thread``. """ self._dispatcher = fn - def _dispatch(self, callback, *args): + def _dispatch(self, callback, *args) -> None: """Call *callback* via the dispatcher, or directly if none is set.""" if callback is None: return @@ -113,6 +117,8 @@ class MumbleClient: def users(self) -> dict[int, User]: if not self._mumble: return {} + if not self._users_dirty: + return self._users_cache result = {} for sid, u in self._mumble.users.items(): result[sid] = User( @@ -124,12 +130,16 @@ class MumbleClient: self_mute=u.get("self_mute", False), self_deaf=u.get("self_deaf", False), ) + self._users_cache = result + self._users_dirty = False return result @property def channels(self) -> dict[int, Channel]: if not self._mumble: return {} + if not self._channels_dirty: + return self._channels_cache result = {} for cid, ch in self._mumble.channels.items(): result[cid] = Channel( @@ -138,6 +148,8 @@ class MumbleClient: parent_id=ch.get("parent", 0), description=ch.get("description", ""), ) + self._channels_cache = result + self._channels_dirty = False return result @property @@ -151,7 +163,7 @@ class MumbleClient: # -- connection ---------------------------------------------------------- - def connect(self): + def connect(self) -> None: """Connect to the Mumble server (blocking). Raises: @@ -204,7 +216,7 @@ class MumbleClient: self._host, self._port, self._username, ) - def disconnect(self): + def disconnect(self) -> None: """Disconnect from the server.""" if self._mumble: try: @@ -212,9 +224,11 @@ class MumbleClient: except Exception: pass self._connected = False + self._users_dirty = True + self._channels_dirty = True log.info("disconnected") - def reconnect(self): + def reconnect(self) -> None: """Disconnect and reconnect to the same server. Raises: @@ -226,7 +240,7 @@ class MumbleClient: # -- actions ------------------------------------------------------------- - def send_text(self, message: str): + def send_text(self, message: str) -> None: """Send a text message to the current channel.""" if self._mumble and self._connected: try: @@ -237,12 +251,12 @@ class MumbleClient: return ch.send_text_message(message) - def send_audio(self, pcm_data: bytes): + def send_audio(self, pcm_data: bytes) -> None: """Send PCM audio to the server (pymumble encodes to Opus).""" if self._mumble and self._connected: self._mumble.sound_output.add_sound(pcm_data) - def join_channel(self, channel_id: int): + def join_channel(self, channel_id: int) -> None: """Move to a different channel. Raises: @@ -254,7 +268,7 @@ class MumbleClient: raise ValueError(f"channel {channel_id} not found") ch.move_in() - def set_self_deaf(self, deaf: bool): + def set_self_deaf(self, deaf: bool) -> None: """Toggle self-deafen on the server.""" if self._mumble and self._connected: if deaf: @@ -264,7 +278,7 @@ class MumbleClient: # -- pymumble callbacks (run on pymumble thread) ------------------------- - def _register_callbacks(self): + def _register_callbacks(self) -> None: import pymumble_py3.constants as const cb = self._mumble.callbacks @@ -279,25 +293,31 @@ class MumbleClient: cb.set_callback(const.PYMUMBLE_CLBK_CHANNELUPDATED, self._on_channel_event) cb.set_callback(const.PYMUMBLE_CLBK_CHANNELREMOVED, self._on_channel_event) - def _on_connected(self): + def _on_connected(self) -> None: self._connected = True + self._users_dirty = True + self._channels_dirty = True self._dispatch(self.on_connected) - def _on_disconnected(self): + def _on_disconnected(self) -> None: self._connected = False + self._users_dirty = True + self._channels_dirty = True self._dispatch(self.on_disconnected) - def _on_text_message(self, message): + def _on_text_message(self, message) -> None: users = self._mumble.users actor = message.actor name = users[actor]["name"] if actor in users else "?" self._dispatch(self.on_text_message, name, message.message) - def _on_sound_received(self, user, sound_chunk): + def _on_sound_received(self, user, sound_chunk) -> None: self._dispatch(self.on_sound_received, user, sound_chunk.pcm) - def _on_user_event(self, *_args): + def _on_user_event(self, *_args) -> None: + self._users_dirty = True self._dispatch(self.on_user_update) - def _on_channel_event(self, *_args): + def _on_channel_event(self, *_args) -> None: + self._channels_dirty = True self._dispatch(self.on_channel_update)