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
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
|
"""
MAX25 sound-proxy — FreeBSD/OSS capture and playback.
Uses `sox` with OSS devices when available; falls back to raw /dev/dsp read/write.
"""
from __future__ import annotations
import os
import shutil
import subprocess
import threading
from dataclasses import dataclass
from typing import Callable, Optional
@dataclass
class OssSoundConfig:
capture: str = "/dev/dsp"
playback: str = "/dev/dsp"
sample_rate: int = 48000
channels: int = 1
class OssSoundProxy:
def __init__(self, cfg: OssSoundConfig) -> None:
self.cfg = cfg
self._rec_proc: Optional[subprocess.Popen[bytes]] = None
self._dsp_fd: Optional[int] = None
self._stop = threading.Event()
self._use_sox = shutil.which("sox") is not None
def _sox_capture_cmd(self) -> list[str]:
dev = self.cfg.capture
return [
"sox",
"-q",
"-t",
"oss",
dev,
"-r",
str(self.cfg.sample_rate),
"-c",
str(self.cfg.channels),
"-b",
"16",
"-e",
"signed-integer",
"-t",
"raw",
"-",
]
def _sox_play_cmd(self) -> list[str]:
dev = self.cfg.playback
return [
"sox",
"-q",
"-t",
"raw",
"-r",
str(self.cfg.sample_rate),
"-c",
str(self.cfg.channels),
"-b",
"16",
"-e",
"signed-integer",
"-",
"-t",
"oss",
dev,
]
def start_capture(self) -> None:
if self._use_sox:
self._rec_proc = subprocess.Popen(
self._sox_capture_cmd(),
stdout=subprocess.PIPE,
stderr=subprocess.DEVNULL,
)
return
flags = os.O_RDONLY
try:
self._dsp_fd = os.open(self.cfg.capture, flags)
except OSError as exc:
raise RuntimeError(f"OSS open {self.cfg.capture}: {exc}") from exc
def read_capture(self, nbytes: int) -> bytes:
if self._rec_proc is not None and self._rec_proc.stdout is not None:
return self._rec_proc.stdout.read(nbytes) or b""
if self._dsp_fd is not None:
try:
return os.read(self._dsp_fd, nbytes) or b""
except OSError:
return b""
return b""
def play_pcm(self, pcm: bytes) -> None:
if not pcm:
return
if self._use_sox:
subprocess.run(
self._sox_play_cmd(),
input=pcm,
check=False,
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
)
return
try:
fd = os.open(self.cfg.playback, os.O_WRONLY)
except OSError:
return
try:
os.write(fd, pcm)
finally:
os.close(fd)
def sniff_loop(
self,
chunk_symbols: int,
on_pcm: Callable[[bytes], None],
stop: Optional[threading.Event] = None,
) -> None:
stop_ev = stop or self._stop
frame_bytes = max(256, (self.cfg.sample_rate // 1200) * 2 * chunk_symbols)
self.start_capture()
try:
while not stop_ev.is_set():
chunk = self.read_capture(frame_bytes)
if not chunk:
break
on_pcm(chunk)
finally:
self.close()
def close(self) -> None:
self._stop.set()
if self._rec_proc is not None and self._rec_proc.poll() is None:
self._rec_proc.terminate()
try:
self._rec_proc.wait(timeout=1.0)
except subprocess.TimeoutExpired:
self._rec_proc.kill()
self._rec_proc = None
if self._dsp_fd is not None:
try:
os.close(self._dsp_fd)
except OSError:
pass
self._dsp_fd = None
|
