1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
|
"""PipeWire/Pulse microphone capture through the local ffmpeg binary."""
from __future__ import annotations
import subprocess
from collections.abc import Iterator
import numpy as np
from mic_clipper.vad import FRAME_SAMPLES
class AudioInputUnavailable(RuntimeError):
"""The dynamic Pulse default source could not supply another frame."""
class PulseAudioInput:
"""Yield fixed 16 kHz mono float PCM frames from Pulse's `default` source."""
@staticmethod
def command() -> list[str]:
return [
"ffmpeg",
"-nostdin",
"-hide_banner",
"-loglevel",
"error",
"-f",
"pulse",
"-sample_rate",
"16000",
"-channels",
"1",
"-i",
"default",
"-ac",
"1",
"-ar",
"16000",
"-f",
"f32le",
"pipe:1",
]
def frames(self) -> Iterator[np.ndarray]:
process = subprocess.Popen(
self.command(),
stdin=subprocess.DEVNULL,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
)
assert process.stdout is not None
assert process.stderr is not None
try:
while True:
payload = process.stdout.read(FRAME_SAMPLES * np.dtype("<f4").itemsize)
if len(payload) != FRAME_SAMPLES * np.dtype("<f4").itemsize:
error = process.stderr.read(4096).decode("utf-8", errors="replace").strip()
return_code = process.wait()
detail = f": {error}" if error else ""
raise AudioInputUnavailable(
f"ffmpeg Pulse input ended with exit code {return_code}{detail}"
)
yield np.frombuffer(payload, dtype="<f4").copy()
finally:
if process.poll() is None:
process.terminate()
try:
process.wait(timeout=5)
except subprocess.TimeoutExpired:
process.kill()
process.wait()
|
