Fix Pyright errors

.github/workflows/static-analyzers.yml

@@ -0,0 +1,30 @@
+name: Run static analysis tools
+on: [push, pull_request]
+jobs:
+  static-analysis:
+    runs-on: ubuntu-latest
+    steps:
+      - name: Set up Python
+        uses: actions/setup-python@v5
+        with:
+          python-version: "3"
+      - name: Double-check Python version
+        run: |
+          python --version
+      - name: Clone Git repository
+        uses: actions/checkout@v4
+      - name: Install sounddevice module
+        run: |
+          python -m pip install .
+      - name: Install Pyright and NumPy
+        run: |
+          python -m pip install pyright numpy
+      - name: Check sounddevice module with Pyright
+        run: |
+          python -m pyright sounddevice.py
+      - name: Install dependencies for examples
+        run: |
+          python -m pip install ffmpeg-python matplotlib soundfile
+      - name: Check examples with Pyright
+        run: |
+          python -m pyright examples/*.py

examples/play_long_file.py

@@ -96,7 +96,7 @@ def callback(outdata, frames, time, status):
             callback=callback, finished_callback=event.set)
         with stream:
             timeout = args.blocksize * args.buffersize / f.samplerate
-            while len(data):
+            while len(data):  # type: ignore
                 data = f.read(args.blocksize)
                 q.put(data, timeout=timeout)
             event.wait()  # Wait until playback is finished

examples/play_long_file_raw.py

@@ -87,7 +87,7 @@ def callback(outdata, frames, time, status):
             callback=callback, finished_callback=event.set)
         with stream:
             timeout = args.blocksize * args.buffersize / f.samplerate
-            while data:
+            while data:  # type: ignore
                 data = f.buffer_read(args.blocksize, dtype='float32')
                 q.put(data, timeout=timeout)
             event.wait()  # Wait until playback is finished

examples/rec_gui.py

@@ -75,8 +75,6 @@ def validate(self):
 
 class RecGui(tk.Tk):
 
-    stream = None
-
     def __init__(self):
         super().__init__()
 
@@ -122,7 +120,7 @@ def __init__(self):
         self.update_gui()
 
     def create_stream(self, device=None):
-        if self.stream is not None:
+        with contextlib.suppress(AttributeError):
             self.stream.close()
         self.stream = sd.InputStream(
             device=device, channels=1, callback=self.audio_callback)
@@ -205,7 +203,7 @@ def on_settings(self, *args):
     def init_buttons(self):
         self.rec_button['text'] = 'record'
         self.rec_button['command'] = self.on_rec
-        if self.stream:
+        if hasattr(self, 'stream'):
             self.rec_button['state'] = 'normal'
         self.settings_button['state'] = 'normal'
 

examples/spectrogram.py

@@ -32,7 +32,7 @@ def int_or_str(text):
     print(sd.query_devices())
     parser.exit(0)
 parser = argparse.ArgumentParser(
-    description=__doc__ + '\n\nSupported keys:' + usage_line,
+    description=__doc__ + '\n\nSupported keys:' + usage_line,  # type: ignore
     formatter_class=argparse.RawDescriptionHelpFormatter,
     parents=[parser])
 parser.add_argument(

sounddevice.py

@@ -69,7 +69,7 @@
             break
     else:
         raise OSError('PortAudio library not found')
-    _lib = _ffi.dlopen(_libname)
+    _lib: ... = _ffi.dlopen(_libname)
 except OSError:
     if _platform.system() == 'Darwin':
         _libname = 'libportaudio.dylib'
@@ -83,9 +83,9 @@
     import _sounddevice_data
     _libname = _os.path.join(
         next(iter(_sounddevice_data.__path__)), 'portaudio-binaries', _libname)
-    _lib = _ffi.dlopen(_libname)
+    _lib: ... = _ffi.dlopen(_libname)
 
-_sampleformats = {
+_sampleformats: ... = {
     'float32': _lib.paFloat32,
     'int32': _lib.paInt32,
     'int24': _lib.paInt24,
@@ -453,7 +453,7 @@ def get_stream():
         raise RuntimeError('play()/rec()/playrec() was not called yet')
 
 
-def query_devices(device=None, kind=None):
+def query_devices(device=None, kind=None) -> ...:
     """Return information about available devices.
 
     Information and capabilities of PortAudio devices.
@@ -571,7 +571,7 @@ def query_devices(device=None, kind=None):
     if not info:
         raise PortAudioError(f'Error querying device {device}')
     assert info.structVersion == 2
-    name_bytes = _ffi.string(info.name)
+    name_bytes = _ffi_string(info.name)
     try:
         # We don't know beforehand if DirectSound and MME device names use
         # 'utf-8' or 'mbcs' encoding.  Let's try 'utf-8' first, because it more
@@ -606,7 +606,7 @@ def query_devices(device=None, kind=None):
     return device_dict
 
 
-def query_hostapis(index=None):
+def query_hostapis(index=None) -> ...:
     """Return information about available host APIs.
 
     Parameters
@@ -651,7 +651,7 @@ def query_hostapis(index=None):
         raise PortAudioError(f'Error querying host API {index}')
     assert info.structVersion == 1
     return {
-        'name': _ffi.string(info.name).decode(),
+        'name': _ffi_string(info.name).decode(),
         'devices': [_lib.Pa_HostApiDeviceIndexToDeviceIndex(index, i)
                     for i in range(info.deviceCount)],
         'default_input_device': info.defaultInputDevice,
@@ -721,7 +721,7 @@ def get_portaudio_version():
         (1899, 'PortAudio V19-devel (built Feb 15 2014 23:28:00)')
 
     """
-    return _lib.Pa_GetVersion(), _ffi.string(_lib.Pa_GetVersionText()).decode()
+    return _lib.Pa_GetVersion(), _ffi_string(_lib.Pa_GetVersionText()).decode()
 
 
 class _StreamBase:
@@ -1204,7 +1204,7 @@ def _raw_read(self, frames):
         """
         channels, _ = _split(self._channels)
         samplesize, _ = _split(self._samplesize)
-        data = _ffi.new('signed char[]', channels * samplesize * frames)
+        data = _ffi.new('signed char[]', channels * samplesize * frames)  # type: ignore
         err = _lib.Pa_ReadStream(self._ptr, data, frames)
         if err == _lib.paInputOverflowed:
             overflowed = True
@@ -1302,7 +1302,7 @@ def _raw_write(self, data):
             pass  # input is not a buffer
         _, samplesize = _split(self._samplesize)
         _, channels = _split(self._channels)
-        samples, remainder = divmod(len(data), samplesize)
+        samples, remainder = divmod(len(data), samplesize)  # type: ignore
         if remainder:
             raise ValueError('len(data) not divisible by samplesize')
         frames, remainder = divmod(samples, channels)
@@ -2052,7 +2052,7 @@ class _InputOutputPair:
     _indexmapping = {'input': 0, 'output': 1}
 
     def __init__(self, parent, default_attr):
-        self._pair = [None, None]
+        self._pair: ... = [None, None]
         self._parent = parent
         self._default_attr = default_attr
 
@@ -2113,7 +2113,7 @@ class default:
     _pairs = 'device', 'channels', 'dtype', 'latency', 'extra_settings'
     # The class attributes listed in _pairs are only provided here for static
     # analysis tools and for the docs.  They're overwritten in __init__().
-    device = None, None
+    device: ... = (None, None)
     """Index or query string of default input/output device.
 
     If not overwritten, this is queried from PortAudio.
@@ -2123,15 +2123,17 @@ class default:
     `default`, `query_devices()`, the *device* argument of `Stream`
 
     """
-    channels = _default_channels = None, None
+    _default_channels = None, None
+    channels: ... = _default_channels
     """Default number of input/output channels.
 
     See Also
     --------
     `default`, `query_devices()`, the *channels* argument of `Stream`
 
     """
-    dtype = _default_dtype = 'float32', 'float32'
+    _default_dtype = 'float32', 'float32'
+    dtype: ... = _default_dtype
     """Default data type used for input/output samples.
 
     The types ``'float32'``, ``'int32'``, ``'int16'``, ``'int8'`` and
@@ -2147,25 +2149,27 @@ class default:
     `default`, `numpy:numpy.dtype`, the *dtype* argument of `Stream`
 
     """
-    latency = _default_latency = 'high', 'high'
+    _default_latency = 'high', 'high'
+    latency: ... = _default_latency
     """See the *latency* argument of `Stream`."""
-    extra_settings = _default_extra_settings = None, None
+    _default_extra_settings = None, None
+    extra_settings: ... = _default_extra_settings
     """Host-API-specific input/output settings.
 
     See Also
     --------
     AsioSettings, CoreAudioSettings, WasapiSettings
 
     """
-    samplerate = None
+    samplerate: ... = None
     """Sampling frequency in Hertz (= frames per second).
 
     See Also
     --------
     `default`, `query_devices()`
 
     """
-    blocksize = _lib.paFramesPerBufferUnspecified
+    blocksize: ... = _lib.paFramesPerBufferUnspecified
     """See the *blocksize* argument of `Stream`."""
     clip_off = False
     """Disable clipping.
@@ -2225,10 +2229,12 @@ def _default_device(self):
                 _lib.Pa_GetDefaultOutputDevice())
 
     @property
-    def hostapi(self):
+    def hostapi(self):  # type: ignore
         """Index of the default host API (read-only)."""
         return _check(_lib.Pa_GetDefaultHostApi())
 
+    hostapi: int
+
     def reset(self):
         """Reset all attributes to their "factory default"."""
         vars(self).clear()
@@ -2237,7 +2243,8 @@ def reset(self):
 
 if not hasattr(_ffi, 'I_AM_FAKE'):
     # This object shadows the 'default' class, except when building the docs.
-    default = default()
+    _default_instance: ... = default()
+    default = _default_instance
 
 
 class PortAudioError(Exception):
@@ -2507,13 +2514,10 @@ def __init__(self, exclusive=False, auto_convert=False, explicit_sample_format=F
 class _CallbackContext:
     """Helper class for reuse in play()/rec()/playrec() callbacks."""
 
-    blocksize = None
-    data = None
-    out = None
     frame = 0
+    frames: int
     input_channels = output_channels = None
     input_dtype = output_dtype = None
-    input_mapping = output_mapping = None
     silent_channels = None
 
     def __init__(self, loop=False):
@@ -2557,7 +2561,7 @@ def check_data(self, data, mapping, device):
         if len(mapping) + len(silent_channels) != channels:
             raise ValueError('each channel may only appear once in mapping')
 
-        self.data = data
+        self.data: np.typing.NDArray = data
         self.output_channels = channels
         self.output_dtype = dtype
         self.output_mapping = mapping
@@ -2641,8 +2645,8 @@ def callback_exit(self):
     def finished_callback(self):
         self.event.set()
         # Drop temporary audio buffers to free memory
-        self.data = None
-        self.out = None
+        del self.data
+        del self.out
         # Drop CFFI objects to avoid reference cycles
         self.stream._callback = None
         self.stream._finished_callback = None
@@ -2675,6 +2679,10 @@ def wait(self, ignore_errors=True):
         return self.status if self.status else None
 
 
+def _ffi_string(cdata) -> bytes:
+    return _ffi.string(cdata)  # type: ignore
+
+
 def _remove_self(d):
     """Return a copy of d without the 'self' entry."""
     d = d.copy()
@@ -2711,7 +2719,7 @@ def _check_dtype(dtype):
 
 
 def _get_stream_parameters(kind, device, channels, dtype, latency,
-                           extra_settings, samplerate):
+                           extra_settings: ..., samplerate):
     """Get parameters for one direction (input or output) of a stream."""
     assert kind in ('input', 'output')
     if device is None:
@@ -2749,7 +2757,7 @@ def _get_stream_parameters(kind, device, channels, dtype, latency,
         latency = info['default_' + latency + '_' + kind + '_latency']
     if samplerate is None:
         samplerate = info['default_samplerate']
-    parameters = _ffi.new('PaStreamParameters*', (
+    parameters: ... = _ffi.new('PaStreamParameters*', (
         device, channels, sampleformat, latency,
         extra_settings._streaminfo if extra_settings else _ffi.NULL))
     return parameters, dtype, samplesize, samplerate
@@ -2805,7 +2813,7 @@ def _check(err, msg=''):
     if err >= 0:
         return err
 
-    errormsg = _ffi.string(_lib.Pa_GetErrorText(err)).decode()
+    errormsg = _ffi_string(_lib.Pa_GetErrorText(err)).decode()
     if msg:
         errormsg = f'{msg}: {errormsg}'
 
@@ -2816,7 +2824,7 @@ def _check(err, msg=''):
         # in scenarios where multiple APIs are being used simultaneously.
         info = _lib.Pa_GetLastHostErrorInfo()
         host_api = _lib.Pa_HostApiTypeIdToHostApiIndex(info.hostApiType)
-        hosterror_text = _ffi.string(info.errorText).decode()
+        hosterror_text = _ffi_string(info.errorText).decode()
         hosterror_info = host_api, info.errorCode, hosterror_text
         raise PortAudioError(errormsg, err, hosterror_info)