From fa05a5f8238e9e1417235711656e5062ccc843a7 Mon Sep 17 00:00:00 2001 From: "info@mode42.com" Date: Fri, 7 Aug 2026 18:23:28 +0000 Subject: Initial push --- stacks/crdop/tools/audio-dummyd | 4 ++ stacks/crdop/tools/audio-dummyd.py | 66 +++++++++++++++++++++++++ stacks/crdop/tools/max25-signal-sniffer | 5 ++ stacks/crdop/tools/max25-signal-sniffer.py | 78 ++++++++++++++++++++++++++++++ 4 files changed, 153 insertions(+) create mode 100755 stacks/crdop/tools/audio-dummyd create mode 100644 stacks/crdop/tools/audio-dummyd.py create mode 100755 stacks/crdop/tools/max25-signal-sniffer create mode 100644 stacks/crdop/tools/max25-signal-sniffer.py (limited to 'stacks/crdop/tools') diff --git a/stacks/crdop/tools/audio-dummyd b/stacks/crdop/tools/audio-dummyd new file mode 100755 index 0000000..e64bda5 --- /dev/null +++ b/stacks/crdop/tools/audio-dummyd @@ -0,0 +1,4 @@ +#!/usr/bin/env bash +set -euo pipefail +ROOT="$(CDPATH= cd -- "$(dirname -- "$0")/../.." && pwd)" +exec python3 "${ROOT}/stacks/crdop/tools/audio-dummyd.py" "$@" diff --git a/stacks/crdop/tools/audio-dummyd.py b/stacks/crdop/tools/audio-dummyd.py new file mode 100644 index 0000000..a4bb922 --- /dev/null +++ b/stacks/crdop/tools/audio-dummyd.py @@ -0,0 +1,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()) diff --git a/stacks/crdop/tools/max25-signal-sniffer b/stacks/crdop/tools/max25-signal-sniffer new file mode 100755 index 0000000..e412997 --- /dev/null +++ b/stacks/crdop/tools/max25-signal-sniffer @@ -0,0 +1,5 @@ +#!/usr/bin/env bash +# MAX25 signal sniffer launcher +set -euo pipefail +ROOT="$(CDPATH= cd -- "$(dirname -- "$0")/../.." && pwd)" +exec python3 "${ROOT}/stacks/crdop/tools/max25-signal-sniffer.py" "$@" diff --git a/stacks/crdop/tools/max25-signal-sniffer.py b/stacks/crdop/tools/max25-signal-sniffer.py new file mode 100644 index 0000000..fed95f7 --- /dev/null +++ b/stacks/crdop/tools/max25-signal-sniffer.py @@ -0,0 +1,78 @@ +#!/usr/bin/env python3 +"""MAX25 signal sniffer — analyze Bell 202 AFSK on ALSA capture or PCM file.""" +from __future__ import annotations + +import argparse +import sys +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 sound_proxy import SoundConfig, SoundProxy # noqa: E402 + + +def read_wav_pcm(path: Path) -> bytes: + import wave + + with wave.open(str(path), "rb") as wf: + if wf.getsampwidth() != 2 or wf.getnchannels() != 1: + raise SystemExit("WAV must be mono S16_LE") + return wf.readframes(wf.getnframes()) + + +def main() -> int: + ap = argparse.ArgumentParser(description="MAX25 acoustic signal sniffer (Bell 202 1200 bd)") + ap.add_argument("-D", "--device", default="", help="ALSA capture device (hw: or plughw:)") + ap.add_argument("-r", "--rate", type=int, default=48000) + ap.add_argument("-f", "--file", type=Path, help="Read mono S16 WAV instead of ALSA") + ap.add_argument("-t", "--seconds", type=float, default=2.0, help="ALSA capture duration") + ap.add_argument("--loopback", action="store_true", help="Internal DSP loopback (no audio hw)") + ap.add_argument("--mark", action="store_true", help="Play mark calibration tone (1200 Hz)") + ap.add_argument("--space", action="store_true", help="Play space calibration tone (2200 Hz)") + args = ap.parse_args() + + cfg = SoundConfig(capture=args.device or "default", sample_rate=args.rate) + eng = AcousticEngine(sample_rate=args.rate, sound=cfg) + + if args.loopback: + rep = eng.loopback_self_test() + print(f"loopback samples={rep.samples} symbols={rep.symbols} transitions={rep.transitions}") + print(f"mark={rep.mark_ratio:.2%} space={rep.space_ratio:.2%}") + for line in rep.decode_lines: + print(line) + return 0 if rep.decode_lines else 1 + + if args.mark: + eng.play_mark_calibration(args.seconds) + print(f"played mark tone {args.seconds}s") + return 0 + if args.space: + eng.play_space_calibration(args.seconds) + print(f"played space tone {args.seconds}s") + return 0 + + if args.file: + pcm = read_wav_pcm(args.file) + else: + if not args.device: + print("error: -D hw:... or --file or --loopback required", file=sys.stderr) + return 2 + proxy = SoundProxy(cfg) + proxy.start_capture() + nbytes = int(args.rate * args.seconds) * 2 + pcm = proxy.read_capture(nbytes) + proxy.close() + + rep = eng.analyze_pcm(pcm) + print(f"sniff samples={rep.samples} symbols={rep.symbols} transitions={rep.transitions}") + print(f"mark={rep.mark_ratio:.2%} space={rep.space_ratio:.2%} frames={len(rep.frames)}") + for line in rep.decode_lines: + print(line) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) -- cgit v1.3.1