left token
Hey guys! I'm trying to use the library called sound device to make, sort of a music player where you can stream out multiple tracks potentially, so I need to implement a pause/ resume feature. My code currently plays the song (a numpy array of samples) using sounddevice in a coroutine.
import sys
from threading import Event
import asyncio
import sounddevice as sd
import numpy as np
from pedalboard import Pedalboard
from pedalboard.io import AudioFile
import aioconsole
filname = "session.mp3"
with AudioFile(filname) as f:
samples = f.read(f.frames)
sample_rate = f.samplerate
current_frame = 0
data = samples.T
print(data.shape)
playing = "b"
pause_event = asyncio.Event()
def callback(outdata, frames, time, status):
global current_frame, playing
if playing == "a":
print("you just hit pause")
return
if status:
print(status)
chunksize = min(len(data) - current_frame, frames)
outdata[:chunksize, :] = data[current_frame:current_frame + chunksize]
if chunksize < frames:
outdata[chunksize:] = 0
raise sd.CallbackStop()
current_frame += chunksize
async def get_steam_reader(pipe) -> asyncio.StreamReader:
loop = asyncio.get_event_loop()
reader = asyncio.StreamReader(loop=loop)
protocol = asyncio.StreamReaderProtocol(reader)
await loop.connect_read_pipe(lambda: protocol, pipe)
return reader
async def main():
event = asyncio.Event()
reader = await get_steam_reader(sys.stdin)
playing = await reader.readline()
stream = sd.OutputStream(
samplerate=sample_rate, channels=data.shape[1],
callback=callback, finished_callback=event.set)
with stream:
await event.wait()
if __name__ == "__main__":
coroutine = asyncio.run(main())
Right now when I hit either a or b, it just plays the song and never pauses.
bold sinew
