Update 0.3.10 (#9)

- Fixed a bug that made the class inoperable after calling the
`abort(...)` function
- Fixed the issue of audio distortion at the beginning of playback
- Added the `precallback` argument to the `CallbackSoundDeviceStreamer`
class (this is the method that is called before the internal callback
method)
- Fixed error call `abort(...)` error
- Minor typification and attribute fixes

Author: peachboy0

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "seaplayer-audio"
-version = "0.3.7"
+version = "0.3.10"
 description = "A library for async/sync playback audio."
 repository = "https://github.com/romanin-rf/seaplayer-audio"
 authors = ["Romanin <60302782+romanin-rf@users.noreply.github.com>"]

seaplayer_audio/queues.py

@@ -1,11 +1,11 @@
 import asyncio
-from asyncio import AbstractEventLoop
+from queue import Queue
+from asyncio import AbstractEventLoop, Queue as AsyncQueue
 from numpy import ndarray, vstack as npvstack, zeros as npzeros
 from sounddevice import OutputStream
 from typing_extensions import Any, Optional, NoReturn, Callable, deprecated
 from .._types import AudioSamplerate, AudioChannels, AudioDType
 from ..base import AsyncSoundDeviceStreamerBase, SoundDeviceStreamerBase, StreamerState
-from ..queues import AsyncQueue, Queue
 
 # ! Main Class
 class CallbackSoundDeviceStreamer(SoundDeviceStreamerBase):
@@ -18,7 +18,7 @@ def __init__(
         dtype: Optional[AudioDType]=None,
         closefd: bool=True,
         device: Optional[int]=None,
-        precallback: Optional[Callable[[int], bool]]=None
+        precallback: Optional[Callable[[int], Any]]=None
     ) -> None:
         super().__init__(samplerate, channels, dtype, closefd, device)
         self.queue: Queue[ndarray] = Queue(1)
@@ -33,21 +33,20 @@ def __init__(
         self.buffer: Optional[ndarray] = None
 
     def __callback__(self, outdata: ndarray, frames: int, time, status):
-        if not self.precallback(frames):
-            return
+        self.precallback(frames)
         if self.buffer is None:
             try:
-                d = self.queue.get()
+                d = self.queue.get_nowait()
             except:
                 return
             wdata = d[:frames]
             self.buffer = d[frames:]
         elif len(self.buffer) >= frames:
             wdata = self.buffer[:frames]
             self.buffer = self.buffer[frames:]
-        elif (len(self.buffer) < frames) and (not self.queue.qsize() == 0):
+        elif (len(self.buffer) < frames) and (self.queue.qsize() >= 1):
             try:
-                d = self.queue.get()
+                d = self.queue.get_nowait()
             except:
                 return
             wdata = self.buffer.copy()
@@ -59,10 +58,10 @@ def __callback__(self, outdata: ndarray, frames: int, time, status):
             wdata = self.buffer.copy()
             self.buffer = None
             needed = frames - len(wdata)
-            wdata = npvstack([wdata, npzeros((needed, 2), dtype=outdata.dtype)])
+            wdata = npvstack([wdata, npzeros((needed, self.channels), dtype=outdata.dtype)])
         else:
             self.buffer = None
-            return
+            wdata = npzeros((frames, self.channels), dtype=outdata.dtype)
         outdata[:] = wdata
 
     def is_busy(self) -> bool:
@@ -82,14 +81,14 @@ def stop(self) -> None:
         if StreamerState.RUNNING in self.state:
             self.state &= ~StreamerState.RUNNING
             self.state |= StreamerState.LOCKED
-            try: self.queue.abort()
-            except: pass
+            try: self.queue.task_done()
+            except ValueError: pass
             self.stream.stop()
 
     def abort(self):
         self.state |= StreamerState.LOCKED
-        try: self.queue.abort()
-        except: pass
+        try: self.queue.task_done()
+        except ValueError: pass
         self.stream.abort()
         self.state &= ~StreamerState.LOCKED
 
@@ -125,21 +124,20 @@ def __init__(
         self.buffer: Optional[ndarray] = None
 
     def __callback__(self, outdata: ndarray, frames: int, time, status):
-        if not self.precallback(frames):
-            return
+        self.precallback(frames)
         if self.buffer is None:
             try:
-                d = asyncio.run_coroutine_threadsafe(self.queue.get(), self.loop).result()
+                d = asyncio.run_coroutine_threadsafe(self.queue.get_nowait(), self.loop).result()
             except:
                 return
             wdata = d[:frames]
             self.buffer = d[frames:]
         elif len(self.buffer) >= frames:
             wdata = self.buffer[:frames]
             self.buffer = self.buffer[frames:]
-        elif (len(self.buffer) < frames) and (not self.queue.qsize() == 0):
+        elif (len(self.buffer) < frames) and (self.queue.qsize() >= 1):
             try:
-                d = asyncio.run_coroutine_threadsafe(self.queue.get(), self.loop).result()
+                d = asyncio.run_coroutine_threadsafe(self.queue.get_nowait(), self.loop).result()
             except:
                 return
             wdata = self.buffer.copy()
@@ -151,10 +149,10 @@ def __callback__(self, outdata: ndarray, frames: int, time, status):
             wdata = self.buffer.copy()
             self.buffer = None
             needed = frames - len(wdata)
-            wdata = npvstack([wdata, npzeros((needed, 2), dtype=outdata.dtype)])
+            wdata = npvstack([wdata, npzeros((needed, self.channels), dtype=outdata.dtype)])
         else:
             self.buffer = None
-            return
+            wdata = npzeros((frames, self.channels), dtype=outdata.dtype)
         outdata[:] = wdata
 
     async def is_busy(self) -> bool:
@@ -174,14 +172,14 @@ async def stop(self) -> None:
         if StreamerState.RUNNING in self.state:
             self.state &= ~StreamerState.RUNNING
             self.state |= StreamerState.LOCKED
-            try: await self.queue.abort()
-            except: pass
+            try: self.queue.task_done()
+            except ValueError: pass
             self.stream.stop()
 
     async def abort(self):
         self.state |= StreamerState.LOCKED
-        try: await self.queue.abort()
-        except: pass
+        try: self.queue.task_done()
+        except ValueError: pass
         self.stream.abort()
         self.state &= ~StreamerState.LOCKED
 

seaplayer_audio/streamers/callbackstreamersnd.py

@@ -3,12 +3,12 @@
 import asyncio
 import numpy as np
 import sounddevice as sd
+from queue import Queue
 from threading import Thread
-from asyncio import AbstractEventLoop
+from asyncio import AbstractEventLoop, Queue as AsyncQueue
 from typing_extensions import Optional
 from .._types import AudioSamplerate, AudioChannels, AudioDType
 from ..base import SoundDeviceStreamerBase, AsyncSoundDeviceStreamerBase, StreamerState
-from ..queues import AsyncQueue, Queue
 
 # ^ Thread Streamer
 
@@ -63,16 +63,16 @@ def stop(self) -> None:
         if StreamerState.RUNNING in self.state:
             self.state &= ~StreamerState.RUNNING
             self.state |= StreamerState.LOCKED
-            try: self.queue.abort()
-            except: pass
+            try: self.queue.task_done()
+            except ValueError: pass
             self.stream.stop()
             while StreamerState.STARTED in self.state:
                 time.sleep(0.01)
 
     def abort(self):
         self.state |= StreamerState.LOCKED
-        try: self.queue.abort()
-        except: pass
+        try: self.queue.task_done()
+        except ValueError: pass
         self.stream.abort()
         self.state &= ~StreamerState.LOCKED
 
@@ -135,16 +135,16 @@ async def stop(self) -> None:
         if StreamerState.RUNNING in self.state:
             self.state &= ~StreamerState.RUNNING
             self.state |= StreamerState.LOCKED
-            try: self.queue.abort()
-            except: pass
+            try: self.queue.task_done()
+            except ValueError: pass
             self.stream.stop()
             while (StreamerState.STARTED in self.state) or (not self.task.done()):
                 await asyncio.sleep(0.01)
 
     async def abort(self):
         self.state |= StreamerState.LOCKED
-        try: self.queue.abort()
-        except: pass
+        try: self.queue.task_done()
+        except ValueError: pass
         self.stream.abort()
         self.state &= ~StreamerState.LOCKED