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
|
#!/usr/bin/env python3
"""Audio dummy daemon — M25 host TCP + acoustic DSP bench for max25d."""
from __future__ import annotations
import argparse
import sys
import threading
from pathlib import Path
ROOT = Path(__file__).resolve().parents[3]
LIB = ROOT / "stacks" / "crdop" / "lib"
sys.path.insert(0, str(LIB))
from acoustic_engine import AcousticEngine # noqa: E402
from m25_host_protocol import DEFAULT_CTRL_PORT, DEFAULT_DATA_PORT, M25SoftModemHost # noqa: E402
from sound_proxy import SoundConfig # noqa: E402
_DAEMON = ROOT / "stacks" / "daemon"
sys.path.insert(0, str(_DAEMON))
from ax25_codec import ax25_build_ui # noqa: E402
def main() -> int:
ap = argparse.ArgumentParser(description="MAX25 audio-dummy device daemon")
ap.add_argument("--ctrl-port", type=int, default=DEFAULT_CTRL_PORT)
ap.add_argument("--data-port", type=int, default=DEFAULT_DATA_PORT)
ap.add_argument("-D", "--capture", default="")
ap.add_argument("-P", "--playback", default="")
ap.add_argument("-r", "--rate", type=int, default=48000)
args = ap.parse_args()
sound = SoundConfig(
capture=args.capture or "default",
playback=args.playback or args.capture or "default",
sample_rate=args.rate,
)
eng = AcousticEngine(sample_rate=args.rate, sound=sound)
rx_lock = threading.Lock()
last_rx: list[str] = []
def on_tx(payload: bytes) -> str:
if len(payload) < 16:
return "ERR short payload"
pcm = eng.mod.modulate_bits(payload)
eng.sound.playback = sound.playback
from sound_proxy import create_sound_proxy
create_sound_proxy(sound).play_pcm(pcm)
return "OK"
host = M25SoftModemHost(args.ctrl_port, args.data_port, on_data_tx=on_tx)
host.start()
print(
f"audio-dummyd ctrl=:{args.ctrl_port} data=:{args.data_port} rate={args.rate}",
flush=True,
)
try:
while True:
threading.Event().wait(3600.0)
except KeyboardInterrupt:
host.stop()
return 0
if __name__ == "__main__":
raise SystemExit(main())
|
