54 lines
1.7 KiB
Python
54 lines
1.7 KiB
Python
from queue import Empty, Queue
|
|
from shlex import split
|
|
from subprocess import PIPE, Popen
|
|
from sys import stderr
|
|
from threading import Event, Thread
|
|
|
|
# FFPLAY_CMD = "ffplay -f f32le -ar 44100 -i -"
|
|
|
|
|
|
class AplaySink:
|
|
def __init__(self, sample_rate: int) -> None:
|
|
self.cmd = f"aplay -t raw -f FLOAT_LE -r {sample_rate}"
|
|
self.process = Popen(split(self.cmd), stdin=PIPE)
|
|
self.queue = Queue()
|
|
self.stop_event = Event()
|
|
self.feeder_thred = Thread(
|
|
target=self._consume, name="aplay-feeder", daemon=False
|
|
)
|
|
print(f"Starting {self.feeder_thred.name} thread...", file=stderr)
|
|
self.feeder_thred.start()
|
|
|
|
def submit(self, samples: bytes):
|
|
self.queue.put(samples)
|
|
|
|
def close(self):
|
|
self.stop_event.set()
|
|
self.feeder_thred.join()
|
|
if self.process.stdin is not None:
|
|
self.process.stdin.close()
|
|
self.process.wait()
|
|
print(f"Stopped {self.feeder_thred.name} thread...", file=stderr)
|
|
|
|
def __enter__(self):
|
|
return self
|
|
|
|
def __exit__(self, exc_type, exc_value, traceback):
|
|
self.close()
|
|
|
|
def _consume(self, timeout=0.1):
|
|
while not self.stop_event.is_set():
|
|
samples = []
|
|
try:
|
|
samples = self.queue.get(timeout=timeout)
|
|
except Empty:
|
|
pass
|
|
if not samples:
|
|
continue
|
|
if self.process.poll() is not None:
|
|
raise RuntimeError("aplay process has terminated")
|
|
if self.process.stdin is None:
|
|
raise RuntimeError("stdin is None")
|
|
self.process.stdin.write(samples)
|
|
self.process.stdin.flush()
|