Cleanup

examples/audio-only_client.py

@@ -23,11 +23,9 @@
 # Works on MacOS. Does NOT work on RPi 3B+ (I cannot figure out why. Help will
 # be much appreciated)
 
+import pyaudio
 import pymumble.pymumble_py3 as pymumble_py3
 from pymumble.pymumble_py3.callbacks import PYMUMBLE_CLBK_SOUNDRECEIVED as PCS
-import subprocess as sp
-from time import sleep
-import pyaudio
 
 # Connection details for mumble server. Harded code for now, will have to be
 # command line arguments eventually
@@ -44,12 +42,14 @@
 RATE = 48000  # pymumble soundchunk.pcm is 48000Hz
 
 p = pyaudio.PyAudio()
-stream = p.open(format=FORMAT,
-                channels=CHANNELS,
-                rate=RATE,
-                input=True,  # enable both talk
-                output=True,  # and listen
-                frames_per_buffer=CHUNK)
+stream = p.open(
+    format=FORMAT,
+    channels=CHANNELS,
+    rate=RATE,
+    input=True,  # enable both talk
+    output=True,  # and listen
+    frames_per_buffer=CHUNK,
+)
 
 
 # mumble client set up

examples/echobot.py

@@ -2,8 +2,9 @@
 # This bot sends any sound it receives back to where it has come from.
 # WARNING! Don't put two bots in the same place!
 
-import pymumble_py3
 import time
+
+import pymumble_py3
 from pymumble_py3.callbacks import PYMUMBLE_CLBK_SOUNDRECEIVED as PCS
 
 pwd = ""  # password

examples/talking_bot.py

@@ -2,10 +2,12 @@
 # This bot reads standard input and converts them to speech via espeak and
 # sends them to server(of course after converting the wave format to s32le)
 # A blank line to exit.
-import pymumble_py3
 import subprocess as sp
+
+import pymumble_py3
+
 try:
-    import readline # optional
+    import readline  # optional
 except ImportError:
     pass
 
@@ -17,13 +19,14 @@
 mumble.start()
 s = " "
 while s:
-    s = input(") ") 
+    s = input(") ")
     # converting text to speech
-    command = ["espeak","--stdout", s]
+    command = ["espeak", "--stdout", s]
     wave_file = sp.Popen(command, stdout=sp.PIPE).stdout
     # converting the wave speech to pcm
-    command = ["ffmpeg", "-i", "-", "-ac", "1", "-f", "s32le","-"] 
-    sound = sp.Popen(command, stdout=sp.PIPE, stderr=sp.DEVNULL,
-    stdin=wave_file).stdout.read()
+    command = ["ffmpeg", "-i", "-", "-ac", "1", "-f", "s32le", "-"]
+    sound = sp.Popen(
+        command, stdout=sp.PIPE, stderr=sp.DEVNULL, stdin=wave_file
+    ).stdout.read()
     # sending speech to server
     mumble.sound_output.add_sound(sound)

pymumble_py3/blobs.py

@@ -1,42 +1,45 @@
 # -*- coding: utf-8 -*-
 import struct
+import typing
 
 from .constants import *
+from .mumble import Mumble
 from .mumble_pb2 import RequestBlob
 
 
-class Blobs(dict):
+class Blobs(typing.Dict[bytes, typing.Any]):
     """
     Manage the Blob library
     """
-    def __init__(self, mumble_object):
+
+    def __init__(self, mumble_object: Mumble):
         self.mumble_object = mumble_object
-        
-    def get_user_comment(self, hash):
+
+    def get_user_comment(self, hash: bytes) -> None:
         """Request the comment of a user"""
         if hash in self:
             return
         request = RequestBlob()
         request.session_comment.extend(struct.unpack("!5I", hash))
-        
+
         self.mumble_object.send_message(PYMUMBLE_MSG_TYPES_REQUESTBLOB, request)
-    
-    def get_user_texture(self, hash):
+
+    def get_user_texture(self, hash: bytes) -> None:
         """Request the image of a user"""
         if hash in self:
             return
 
         request = RequestBlob()
         request.session_texture.extend(struct.unpack("!5I", hash))
-        
+
         self.mumble_object.send_message(PYMUMBLE_MSG_TYPES_REQUESTBLOB, request)
-    
-    def get_channel_description(self, hash):
+
+    def get_channel_description(self, hash: bytes) -> None:
         """Request the description/comment of a channel"""
         if hash in self:
             return
 
         request = RequestBlob()
         request.channel_description.extend(struct.unpack("!5I", hash))
-        
+
         self.mumble_object.send_message(PYMUMBLE_MSG_TYPES_REQUESTBLOB, request)

pymumble_py3/callbacks.py

@@ -1,91 +1,102 @@
 # -*- coding: utf-8 -*-
 
-from .errors import UnknownCallbackError
-from .constants import *
 import threading
+import typing
+
+from .constants import *
+from .errors import UnknownCallbackError
+
+_Callback = typing.Callable[..., typing.Any]
+_Callbacks = typing.List[_Callback]
 
 
-class CallBacks(dict):
+class CallBacks(typing.Dict[str, typing.Optional[_Callbacks]]):
     """
     Define the callbacks that can be registered by the application.
     Multiple functions can be assigned to a callback using "add_callback"
 
     The call is done from within the pymumble loop thread, it's important to
     keep processing short to avoid delays on audio transmission
     """
-    def __init__(self):
-        self.update({
-            PYMUMBLE_CLBK_CONNECTED: None,  # Connection succeeded
-            PYMUMBLE_CLBK_CHANNELCREATED: None,  # send the created channel object as parameter
-            PYMUMBLE_CLBK_CHANNELUPDATED: None,  # send the updated channel object and a dict with all the modified fields as parameter
-            PYMUMBLE_CLBK_CHANNELREMOVED: None,  # send the removed channel object as parameter
-            PYMUMBLE_CLBK_USERCREATED: None,  # send the added user object as parameter
-            PYMUMBLE_CLBK_USERUPDATED: None,  # send the updated user object and a dict with all the modified fields as parameter
-            PYMUMBLE_CLBK_USERREMOVED: None,  # send the removed user object and the mumble message as parameter
-            PYMUMBLE_CLBK_SOUNDRECEIVED: None,  # send the user object that received the sound and the SoundChunk object itself
-            PYMUMBLE_CLBK_TEXTMESSAGERECEIVED: None,  # Send the received message
-            PYMUMBLE_CLBK_CONTEXTACTIONRECEIVED: None,  # Send the contextaction message
-        })
-
-    def set_callback(self, callback, dest):
+
+    def __init__(self) -> None:
+        self.update(
+            {
+                PYMUMBLE_CLBK_CONNECTED: None,  # Connection succeeded
+                PYMUMBLE_CLBK_CHANNELCREATED: None,  # send the created channel object as parameter
+                PYMUMBLE_CLBK_CHANNELUPDATED: None,  # send the updated channel object and a dict with all the modified fields as parameter
+                PYMUMBLE_CLBK_CHANNELREMOVED: None,  # send the removed channel object as parameter
+                PYMUMBLE_CLBK_USERCREATED: None,  # send the added user object as parameter
+                PYMUMBLE_CLBK_USERUPDATED: None,  # send the updated user object and a dict with all the modified fields as parameter
+                PYMUMBLE_CLBK_USERREMOVED: None,  # send the removed user object and the mumble message as parameter
+                PYMUMBLE_CLBK_SOUNDRECEIVED: None,  # send the user object that received the sound and the SoundChunk object itself
+                PYMUMBLE_CLBK_TEXTMESSAGERECEIVED: None,  # Send the received message
+                PYMUMBLE_CLBK_CONTEXTACTIONRECEIVED: None,  # Send the contextaction message
+            }
+        )
+
+    def set_callback(self, callback: str, dest: _Callback) -> None:
         """Define the function to call for a specific callback.  Suppress any existing callback function"""
         if callback not in self:
-            raise UnknownCallbackError("Callback \"%s\" does not exists." % callback)
-        
+            raise UnknownCallbackError('Callback "%s" does not exists.' % callback)
+
         self[callback] = [dest]
-        
-    def add_callback(self, callback, dest):
+
+    def add_callback(self, callback: str, dest: _Callback) -> None:
         """Add the function to call for a specific callback."""
         if callback not in self:
-            raise UnknownCallbackError("Callback \"%s\" does not exists." % callback)
-        
+            raise UnknownCallbackError('Callback "%s" does not exists.' % callback)
+
         if self[callback] is None:
             self[callback] = list()
-        self[callback].append(dest)
-        
-    def get_callback(self, callback):
+        typing.cast(_Callbacks, self[callback]).append(dest)
+
+    def get_callback(self, callback: str) -> typing.Optional[_Callbacks]:
         """Get the functions assigned to a callback as a list. Return None if no callback defined"""
         if callback not in self:
-            raise UnknownCallbackError("Callback \"%s\" does not exists." % callback)
-        
+            raise UnknownCallbackError('Callback "%s" does not exists.' % callback)
+
         return self[callback]
-    
-    def remove_callback(self, callback, dest):
+
+    def remove_callback(self, callback: str, dest: _Callback) -> None:
         """Remove a specific function from a specific callback.  Function object must be the one added before."""
         if callback not in self:
-            raise UnknownCallbackError("Callback \"%s\" does not exists." % callback)
-
-        if self[callback] is None or dest not in self[callback]:
-            raise UnknownCallbackError("Function not registered for callback \"%s\"." % callback)
-
-        self[callback].remove(dest)
-        if len(self[callback]) == 0:
-            self[callback] = None
-
-    def reset_callback(self, callback):
+            raise UnknownCallbackError('Callback "%s" does not exists.' % callback)
+
+        callbacks = self[callback]
+        if callbacks is None or dest not in callbacks:
+            raise UnknownCallbackError(
+                'Function not registered for callback "%s".' % callback
+            )
+
+        callbacks.remove(dest)
+        if len(callbacks) == 0:
+            callbacks = None
+
+    def reset_callback(self, callback: str) -> None:
         """remove functions for a defined callback"""
         if callback not in self:
-            raise UnknownCallbackError("Callback \"%s\" does not exists." % callback)
-        
+            raise UnknownCallbackError('Callback "%s" does not exists.' % callback)
+
         self[callback] = None
-        
-    def call_callback(self, callback, *pos_parameters):
+
+    def call_callback(self, callback: str, *pos_parameters: typing.Any) -> None:
         """Call all the registered function for a specific callback."""
         if callback not in self:
-            raise UnknownCallbackError("Callback \"%s\" does not exists." % callback)
-        
+            raise UnknownCallbackError('Callback "%s" does not exists.' % callback)
+
         if self[callback]:
-            for func in self[callback]:
+            for func in typing.cast(_Callbacks, self[callback]):
                 if callback is PYMUMBLE_CLBK_TEXTMESSAGERECEIVED:
                     thr = threading.Thread(target=func, args=pos_parameters)
                     thr.start()
                 else:
                     func(*pos_parameters)
-    
-    def __call__(self, callback, *pos_parameters):
+
+    def __call__(self, callback: str, *pos_parameters: typing.Any) -> None:
         """shortcut to be able to call the dict element as a function"""
         self.call_callback(callback, *pos_parameters)
-    
-    def get_callbacks_list(self):
+
+    def get_callbacks_list(self) -> typing.List[str]:
         """Get a list of all callbacks"""
         return list(self.keys())