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/lib/__init__.py | 1 + stacks/crdop/lib/acoustic_engine.py | 89 ++++++++++++++ stacks/crdop/lib/afsk_demodulator.py | 56 +++++++++ stacks/crdop/lib/afsk_modulator.py | 51 ++++++++ stacks/crdop/lib/bell202_line_code.py | 50 ++++++++ stacks/crdop/lib/hdlc_codec.py | 110 +++++++++++++++++ stacks/crdop/lib/m25_host_protocol.py | 142 ++++++++++++++++++++++ stacks/crdop/lib/sound_proxy.py | 185 +++++++++++++++++++++++++++++ stacks/crdop/lib/sound_proxy_oss.py | 151 +++++++++++++++++++++++ stacks/crdop/lib/test_bell202_line_code.py | 34 ++++++ 10 files changed, 869 insertions(+) create mode 100644 stacks/crdop/lib/__init__.py create mode 100644 stacks/crdop/lib/acoustic_engine.py create mode 100644 stacks/crdop/lib/afsk_demodulator.py create mode 100644 stacks/crdop/lib/afsk_modulator.py create mode 100644 stacks/crdop/lib/bell202_line_code.py create mode 100644 stacks/crdop/lib/hdlc_codec.py create mode 100644 stacks/crdop/lib/m25_host_protocol.py create mode 100644 stacks/crdop/lib/sound_proxy.py create mode 100644 stacks/crdop/lib/sound_proxy_oss.py create mode 100644 stacks/crdop/lib/test_bell202_line_code.py (limited to 'stacks/crdop/lib') diff --git a/stacks/crdop/lib/__init__.py b/stacks/crdop/lib/__init__.py new file mode 100644 index 0000000..2137436 --- /dev/null +++ b/stacks/crdop/lib/__init__.py @@ -0,0 +1 @@ +"""MAX25-SoftModem (CRDOP) — native DSP library (Bell 202 AFSK, HDLC, ALSA).""" diff --git a/stacks/crdop/lib/acoustic_engine.py b/stacks/crdop/lib/acoustic_engine.py new file mode 100644 index 0000000..47989ec --- /dev/null +++ b/stacks/crdop/lib/acoustic_engine.py @@ -0,0 +1,89 @@ +""" +Acoustic bench engine — modulate/demodulate/sniff without RF. + +Used by audio-dummy device and max25-signal-sniffer. +""" +from __future__ import annotations + +import sys +from dataclasses import dataclass, field +from pathlib import Path +from typing import Optional + +_LIB = Path(__file__).resolve().parent +if str(_LIB) not in sys.path: + sys.path.insert(0, str(_LIB)) + +from afsk_demodulator import AfskDemodulator # noqa: E402 +from afsk_modulator import AfskModulator # noqa: E402 +from bell202_line_code import TONE_MARK, TONE_SPACE # noqa: E402 +from hdlc_codec import build_hdlc_frame, parse_hdlc_stream # noqa: E402 +from sound_proxy import SoundConfig, create_sound_proxy # noqa: E402 + +_ROOT = Path(__file__).resolve().parents[3] +_DAEMON = _ROOT / "stacks" / "daemon" +if str(_DAEMON) not in sys.path: + sys.path.insert(0, str(_DAEMON)) +from ax25_codec import ax25_build_ui, ax25_parse_ui # noqa: E402 + + +@dataclass +class SniffReport: + samples: int = 0 + symbols: int = 0 + mark_ratio: float = 0.0 + space_ratio: float = 0.0 + transitions: int = 0 + frames: list[bytes] = field(default_factory=list) + decode_lines: list[str] = field(default_factory=list) + + +class AcousticEngine: + def __init__( + self, + sample_rate: int = 48000, + baud: int = 1200, + sound: Optional[SoundConfig] = None, + ) -> None: + self.sample_rate = sample_rate + self.baud = baud + self.mod = AfskModulator(sample_rate, baud) + self.demod = AfskDemodulator(sample_rate, baud) + self.sound = sound or SoundConfig(sample_rate=sample_rate) + + def encode_ax25_ui(self, src: str, dst: str, text: str) -> bytes: + body = ax25_build_ui(src, dst, text.encode("utf-8")) + # strip FCS for KISS-style host body, then build on-air HDLC + if len(body) >= 2: + body = body[:-2] + hdlc = build_hdlc_frame(body) + return self.mod.modulate_bits(hdlc) + + def analyze_pcm(self, pcm: bytes) -> SniffReport: + rep = SniffReport(samples=len(pcm) // 2) + tones = self.demod.demodulate_pcm(pcm) + rep.symbols = len(tones) + if not tones: + return rep + marks = sum(1 for t in tones if t == TONE_MARK) + rep.mark_ratio = marks / len(tones) + rep.space_ratio = 1.0 - rep.mark_ratio + rep.transitions = sum(1 for i in range(1, len(tones)) if tones[i] != tones[i - 1]) + raw_bits = self.demod.demodulate_to_bits(pcm) + for frame in parse_hdlc_stream(raw_bits): + rep.frames.append(frame) + parsed = ax25_parse_ui(frame) + if parsed: + src, dst, info = parsed + rep.decode_lines.append(f"[AX25 UI {src}>{dst}] {info.decode('utf-8', errors='replace')}") + return rep + + def loopback_self_test(self, src: str = "TST-0", dst: str = "QST") -> SniffReport: + pcm = self.encode_ax25_ui(src, dst, "LOOP") + return self.analyze_pcm(pcm) + + def play_mark_calibration(self, seconds: float = 1.0) -> None: + create_sound_proxy(self.sound).play_pcm(self.mod.steady_tone(TONE_MARK, seconds)) + + def play_space_calibration(self, seconds: float = 1.0) -> None: + create_sound_proxy(self.sound).play_pcm(self.mod.steady_tone(TONE_SPACE, seconds)) diff --git a/stacks/crdop/lib/afsk_demodulator.py b/stacks/crdop/lib/afsk_demodulator.py new file mode 100644 index 0000000..5ba3d8b --- /dev/null +++ b/stacks/crdop/lib/afsk_demodulator.py @@ -0,0 +1,56 @@ +""" +AFSK demodulator — Bell 202 mark/space discrimination per symbol period. + +Uses per-symbol Goertzel energy at 1200 Hz and 2200 Hz (Dire Wolf class approach). +""" +from __future__ import annotations + +import math +import struct +from array import array + +from bell202_line_code import TONE_MARK, TONE_SPACE, decode_tones_to_bits + +_TWO_PI = 2.0 * math.pi + + +def _goertzel_power(samples: array, freq: float, sample_rate: int) -> float: + n = len(samples) + if n < 8: + return 0.0 + k = int(0.5 + (n * freq) / sample_rate) + w = _TWO_PI * k / n + coeff = 2.0 * math.cos(w) + s0 = s1 = s2 = 0.0 + for x in samples: + s0 = x + coeff * s1 - s2 + s2 = s1 + s1 = s0 + return s1 * s1 + s2 * s2 - coeff * s1 * s2 + + +class AfskDemodulator: + def __init__(self, sample_rate: int = 48000, baud: int = 1200) -> None: + self.sample_rate = sample_rate + self.baud = baud + self.samples_per_symbol = sample_rate // baud + + def demodulate_pcm(self, pcm: bytes) -> list[int]: + """Return mark/space tone sequence from mono S16_LE PCM.""" + count = len(pcm) // 2 + if count < self.samples_per_symbol: + return [] + samples = array("h") + samples.frombytes(pcm[: count * 2]) + tones: list[int] = [] + pos = 0 + while pos + self.samples_per_symbol <= len(samples): + window = samples[pos : pos + self.samples_per_symbol] + pos += self.samples_per_symbol + p_mark = _goertzel_power(window, 1200.0, self.sample_rate) + p_space = _goertzel_power(window, 2200.0, self.sample_rate) + tones.append(TONE_MARK if p_mark >= p_space else TONE_SPACE) + return tones + + def demodulate_to_bits(self, pcm: bytes, start_tone: int = TONE_SPACE) -> bytes: + return decode_tones_to_bits(self.demodulate_pcm(pcm), start_tone=start_tone) diff --git a/stacks/crdop/lib/afsk_modulator.py b/stacks/crdop/lib/afsk_modulator.py new file mode 100644 index 0000000..1a6bf68 --- /dev/null +++ b/stacks/crdop/lib/afsk_modulator.py @@ -0,0 +1,51 @@ +""" +Continuous-phase AFSK modulator — Bell 202 mark 1200 Hz / space 2200 Hz. +""" +from __future__ import annotations + +import math +import struct +from array import array + +from bell202_line_code import MARK_HZ, SPACE_HZ, TONE_MARK, encode_bits_to_tones + +_TWO_PI = 2.0 * math.pi + + +class AfskModulator: + def __init__(self, sample_rate: int = 48000, baud: int = 1200) -> None: + self.sample_rate = sample_rate + self.baud = baud + self._phase = 0.0 + + def _tone_freq(self, tone: int) -> float: + return MARK_HZ if tone == TONE_MARK else SPACE_HZ + + def modulate_bits(self, bits: bytes, start_tone: int = 0) -> bytes: + """Return mono S16_LE PCM for the given bit stream.""" + tones = encode_bits_to_tones(bits, start_tone=start_tone) + samples_per_symbol = self.sample_rate // self.baud + out: array[int] = array("h") + for tone in tones: + freq = self._tone_freq(tone) + step = _TWO_PI * freq / self.sample_rate + for _ in range(samples_per_symbol): + sample = int(0.7 * 32767.0 * math.sin(self._phase)) + out.append(sample) + self._phase += step + if self._phase >= _TWO_PI: + self._phase -= _TWO_PI + return struct.pack(f"<{len(out)}h", *out) + + def steady_tone(self, tone: int, duration_s: float) -> bytes: + """Calibration tone (Dire Wolf -x style mark/space hold).""" + n = int(self.sample_rate * duration_s) + freq = self._tone_freq(tone) + step = _TWO_PI * freq / self.sample_rate + out: array[int] = array("h") + for _ in range(n): + out.append(int(0.7 * 32767.0 * math.sin(self._phase))) + self._phase += step + if self._phase >= _TWO_PI: + self._phase -= _TWO_PI + return struct.pack(f"<{len(out)}h", *out) diff --git a/stacks/crdop/lib/bell202_line_code.py b/stacks/crdop/lib/bell202_line_code.py new file mode 100644 index 0000000..cd2621f --- /dev/null +++ b/stacks/crdop/lib/bell202_line_code.py @@ -0,0 +1,50 @@ +""" +Bell 202 frequency-toggle line code (1200 baud AFSK layer). + +On-air rule: bit 0 → toggle mark/space tone; bit 1 → hold current tone. +We avoid the legacy term in API names — this is Bell 202 / amateur packet radio. +""" +from __future__ import annotations + +MARK_HZ = 1200.0 +SPACE_HZ = 2200.0 +BAUD_1200 = 1200 + +TONE_MARK = 1 +TONE_SPACE = 0 + + +def encode_bits_to_tones(bits: bytes, start_tone: int = TONE_SPACE) -> list[int]: + """Map HDLC bit stream to mark/space tone sequence (one tone per bit).""" + tone = start_tone + out: list[int] = [] + for byte in bits: + for bit_i in range(7, -1, -1): + bit = (byte >> bit_i) & 1 + if bit == 0: + tone ^= 1 + out.append(tone) + return out + + +def decode_tones_to_bits(tones: list[int], start_tone: int = TONE_SPACE) -> bytes: + """Recover bit stream from mark/space tone sequence.""" + if not tones: + return b"" + prev = start_tone + bits: list[int] = [] + for tone in tones: + if tone not in (TONE_MARK, TONE_SPACE): + continue + bits.append(0 if tone != prev else 1) + prev = tone + buf = bytearray() + for i in range(0, len(bits), 8): + chunk = bits[i : i + 8] + if len(chunk) < 8: + break + val = 0 + for b in chunk: + val = (val << 1) | b + buf.append(val) + return bytes(buf) diff --git a/stacks/crdop/lib/hdlc_codec.py b/stacks/crdop/lib/hdlc_codec.py new file mode 100644 index 0000000..942168b --- /dev/null +++ b/stacks/crdop/lib/hdlc_codec.py @@ -0,0 +1,110 @@ +""" +HDLC framing for AX.25 — flags, bit-stuffing, CRC-16-CCITT. +""" +from __future__ import annotations + +import sys +from pathlib import Path + +_ROOT = Path(__file__).resolve().parents[3] +_DAEMON = _ROOT / "stacks" / "daemon" +if str(_DAEMON) not in sys.path: + sys.path.insert(0, str(_DAEMON)) + +from ax25_codec import ax25_crc # noqa: E402 + +HDLC_FLAG = 0x7E +HDLC_ESCAPE = 0x7D +HDLC_XOR = 0x20 + + +def bit_stuff(data: bytes) -> bytes: + out = bytearray() + ones = 0 + for b in data: + for bit_i in range(7, -1, -1): + bit = (b >> bit_i) & 1 + if bit: + ones += 1 + out.append(1) + if ones == 5: + out.append(0) + ones = 0 + else: + ones = 0 + out.append(0) + # pack bits to bytes MSB-first + buf = bytearray() + acc = 0 + n = 0 + for bit in out: + acc = (acc << 1) | bit + n += 1 + if n == 8: + buf.append(acc) + acc = 0 + n = 0 + if n: + buf.append(acc << (8 - n)) + return bytes(buf) + + +def bit_unstuff(data: bytes) -> bytes: + bits: list[int] = [] + for b in data: + for bit_i in range(7, -1, -1): + bits.append((b >> bit_i) & 1) + out: list[int] = [] + ones = 0 + i = 0 + while i < len(bits): + bit = bits[i] + if bit: + ones += 1 + out.append(1) + if ones == 5 and i + 1 < len(bits) and bits[i + 1] == 0: + i += 1 + ones = 0 + elif ones == 6: + break + else: + ones = 0 + out.append(0) + i += 1 + buf = bytearray() + for j in range(0, len(out), 8): + chunk = out[j : j + 8] + if len(chunk) < 8: + break + val = 0 + for bit in chunk: + val = (val << 1) | bit + buf.append(val) + return bytes(buf) + + +def build_hdlc_frame(ax25_body: bytes) -> bytes: + """AX.25 body (no FCS) → on-air HDLC bit stream as bytes (pre tone encoding).""" + crc = ax25_crc(ax25_body) + payload = ax25_body + bytes((crc & 0xFF, crc >> 8)) + stuffed = bit_stuff(payload) + return bytes([HDLC_FLAG]) + stuffed + bytes([HDLC_FLAG]) + + +def parse_hdlc_stream(raw_bits: bytes) -> list[bytes]: + """Extract AX.25 bodies (with FCS) from demodulated byte stream between flags.""" + frames: list[bytes] = [] + in_frame = False + cur = bytearray() + for b in raw_bits: + if b == HDLC_FLAG: + if in_frame and cur: + unstuffed = bit_unstuff(bytes(cur)) + if len(unstuffed) >= 2: + frames.append(unstuffed) + cur.clear() + in_frame = True + continue + if in_frame: + cur.append(b) + return frames diff --git a/stacks/crdop/lib/m25_host_protocol.py b/stacks/crdop/lib/m25_host_protocol.py new file mode 100644 index 0000000..9212336 --- /dev/null +++ b/stacks/crdop/lib/m25_host_protocol.py @@ -0,0 +1,142 @@ +""" +MAX25 native host protocol for SoftModem / audio-dummy (design freeze). + +Payload on data channel = AX.25 UI body **without** HDLC/FCS (same as KISS DATA). +Control channel = line-oriented ASCII commands (M25-family, not ARDOP FEC/ARQ). +""" +from __future__ import annotations + +import socket +import threading +from typing import Callable, Optional + +# KISS-compatible data semantics; control is text lines ending in \\n +DEFAULT_CTRL_PORT = 8515 +DEFAULT_DATA_PORT = 8516 + +CmdFn = Callable[[str], str] +DataRxFn = Callable[[bytes], None] + + +class M25SoftModemHost: + """Minimal TCP host for bench / audio-dummy (ctrl + data ports).""" + + def __init__( + self, + ctrl_port: int = DEFAULT_CTRL_PORT, + data_port: int = DEFAULT_DATA_PORT, + on_data_tx: Optional[Callable[[bytes], str]] = None, + ) -> None: + self.ctrl_port = ctrl_port + self.data_port = data_port + self._on_data_tx = on_data_tx or (lambda _b: "OK") + self._stop = threading.Event() + self._ctrl_srv: Optional[socket.socket] = None + self._data_srv: Optional[socket.socket] = None + self._threads: list[threading.Thread] = [] + self._mycall = "NOCALL-0" + self._listen = True + + def start(self) -> None: + self._ctrl_srv = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + self._ctrl_srv.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) + self._ctrl_srv.bind(("127.0.0.1", self.ctrl_port)) + self._ctrl_srv.listen(4) + self._data_srv = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + self._data_srv.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) + self._data_srv.bind(("127.0.0.1", self.data_port)) + self._data_srv.listen(4) + self._stop.clear() + for target, name in ((self._accept_ctrl, "ctrl"), (self._accept_data, "data")): + t = threading.Thread(target=target, name=f"m25-host-{name}", daemon=True) + t.start() + self._threads.append(t) + + def stop(self) -> None: + self._stop.set() + for srv in (self._ctrl_srv, self._data_srv): + if srv is not None: + try: + srv.close() + except OSError: + pass + self._ctrl_srv = None + self._data_srv = None + + def _accept_ctrl(self) -> None: + assert self._ctrl_srv is not None + while not self._stop.is_set(): + try: + self._ctrl_srv.settimeout(0.5) + conn, _ = self._ctrl_srv.accept() + except (OSError, socket.timeout): + continue + threading.Thread( + target=self._ctrl_session, + args=(conn,), + daemon=True, + ).start() + + def _ctrl_session(self, conn: socket.socket) -> None: + buf = b"" + try: + conn.settimeout(0.5) + while not self._stop.is_set(): + try: + chunk = conn.recv(4096) + except socket.timeout: + continue + if not chunk: + break + buf += chunk + while b"\n" in buf: + raw, buf = buf.split(b"\n", 1) + line = raw.decode("ascii", errors="replace").strip() + if not line: + continue + reply = self._handle_cmd(line) + conn.sendall((reply + "\n").encode("ascii")) + finally: + conn.close() + + def _handle_cmd(self, line: str) -> str: + parts = line.split() + cmd = parts[0].upper() if parts else "" + if cmd == "INITIALIZE": + return "OK" + if cmd == "PROTOCOLMODE" and len(parts) > 1 and parts[1].upper() == "KISS": + return "OK" + if cmd == "MYCALL" and len(parts) > 1: + self._mycall = parts[1].upper() + return "OK" + if cmd == "LISTEN": + self._listen = len(parts) < 2 or parts[1].upper() in ("TRUE", "1", "ON", "YES") + return "OK" + if cmd == "PING": + return "OK" + if cmd == "STATUS": + return f"STATUS ready mycall={self._mycall}" + return "ERR unknown command" + + def _accept_data(self) -> None: + assert self._data_srv is not None + while not self._stop.is_set(): + try: + self._data_srv.settimeout(0.5) + conn, _ = self._data_srv.accept() + except (OSError, socket.timeout): + continue + threading.Thread( + target=self._data_session, + args=(conn,), + daemon=True, + ).start() + + def _data_session(self, conn: socket.socket) -> None: + try: + payload = conn.recv(4096) + if payload: + reply = self._on_data_tx(payload) + conn.sendall(reply.encode("ascii", errors="replace")) + finally: + conn.close() diff --git a/stacks/crdop/lib/sound_proxy.py b/stacks/crdop/lib/sound_proxy.py new file mode 100644 index 0000000..de9b71d --- /dev/null +++ b/stacks/crdop/lib/sound_proxy.py @@ -0,0 +1,185 @@ +""" +MAX25 sound-proxy — host audio capture/playback. + +Linux/KLinux: ALSA (arecord/aplay). +FreeBSD: OSS via sound_proxy_oss (sox or /dev/dsp). +""" +from __future__ import annotations + +import os +import shutil +import subprocess +import sys +import threading +from dataclasses import dataclass +from pathlib import Path +from typing import Callable, Optional, Protocol, runtime_checkable + + +@dataclass +class SoundConfig: + capture: str = "default" + playback: str = "default" + sample_rate: int = 48000 + channels: int = 1 + period_frames: int = 256 + forbid_pulse: bool = True + backend: str = "" # alsa | oss — empty = auto from platform + + +@runtime_checkable +class SoundProxyProto(Protocol): + def start_capture(self) -> None: ... + def read_capture(self, nbytes: int) -> bytes: ... + def play_pcm(self, pcm: bytes) -> None: ... + def sniff_loop( + self, + chunk_symbols: int, + on_pcm: Callable[[bytes], None], + stop: Optional[threading.Event] = None, + ) -> None: ... + def close(self) -> None: ... + + +def _detect_backend(cfg: SoundConfig) -> str: + explicit = (cfg.backend or os.environ.get("MAX25_AUDIO_BACKEND", "")).strip().lower() + if explicit in ("alsa", "oss"): + return explicit + if sys.platform.startswith("freebsd"): + return "oss" + return "alsa" + + +def create_sound_proxy(cfg: SoundConfig) -> SoundProxyProto: + backend = _detect_backend(cfg) + if backend == "oss": + from sound_proxy_oss import OssSoundConfig, OssSoundProxy + + cap = cfg.capture if cfg.capture not in ("", "default") else "/dev/dsp" + pb = cfg.playback if cfg.playback not in ("", "default") else cap + return OssSoundProxy( + OssSoundConfig( + capture=cap, + playback=pb, + sample_rate=cfg.sample_rate, + channels=cfg.channels, + ) + ) + return AlsaSoundProxy(cfg) + + +class AlsaSoundProxy: + def __init__(self, cfg: SoundConfig) -> None: + self.cfg = cfg + self._rec_proc: Optional[subprocess.Popen[bytes]] = None + self._play_proc: Optional[subprocess.Popen[bytes]] = None + self._stop = threading.Event() + + def _alsa_env(self) -> dict[str, str]: + env = os.environ.copy() + if self.cfg.forbid_pulse: + env.pop("PULSE_SERVER", None) + env.pop("PIPEWIRE_RUNTIME_DIR", None) + env["PULSE_SERVER"] = "" + return env + + @staticmethod + def _check_device(name: str) -> None: + low = name.lower() + if low in ("default", "pulse", "pipewire") or "pulse" in low or "pipewire" in low: + raise ValueError( + f"audio device {name!r} not allowed — use hw: or plughw: (kernel ALSA)" + ) + + def start_capture(self) -> None: + if not shutil.which("arecord"): + raise RuntimeError("arecord not found — install alsa-utils") + self._check_device(self.cfg.capture) + cmd = [ + "arecord", + "-q", + "-D", + self.cfg.capture, + "-f", + "S16_LE", + "-r", + str(self.cfg.sample_rate), + "-c", + str(self.cfg.channels), + "-t", + "raw", + ] + self._rec_proc = subprocess.Popen( + cmd, + stdout=subprocess.PIPE, + stderr=subprocess.DEVNULL, + env=self._alsa_env(), + ) + + def read_capture(self, nbytes: int) -> bytes: + if self._rec_proc is None or self._rec_proc.stdout is None: + return b"" + return self._rec_proc.stdout.read(nbytes) or b"" + + def play_pcm(self, pcm: bytes) -> None: + if not pcm: + return + if not shutil.which("aplay"): + raise RuntimeError("aplay not found — install alsa-utils") + self._check_device(self.cfg.playback) + cmd = [ + "aplay", + "-q", + "-D", + self.cfg.playback, + "-f", + "S16_LE", + "-r", + str(self.cfg.sample_rate), + "-c", + str(self.cfg.channels), + "-t", + "raw", + ] + subprocess.run( + cmd, + input=pcm, + check=False, + env=self._alsa_env(), + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + ) + + 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 = (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() + for proc in (self._rec_proc, self._play_proc): + if proc is not None and proc.poll() is None: + proc.terminate() + try: + proc.wait(timeout=1.0) + except subprocess.TimeoutExpired: + proc.kill() + self._rec_proc = None + self._play_proc = None + + +# Backward-compatible alias +SoundProxy = AlsaSoundProxy diff --git a/stacks/crdop/lib/sound_proxy_oss.py b/stacks/crdop/lib/sound_proxy_oss.py new file mode 100644 index 0000000..66cf765 --- /dev/null +++ b/stacks/crdop/lib/sound_proxy_oss.py @@ -0,0 +1,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 diff --git a/stacks/crdop/lib/test_bell202_line_code.py b/stacks/crdop/lib/test_bell202_line_code.py new file mode 100644 index 0000000..3b22af6 --- /dev/null +++ b/stacks/crdop/lib/test_bell202_line_code.py @@ -0,0 +1,34 @@ +#!/usr/bin/env python3 +"""Unit tests for Bell 202 frequency-toggle line code.""" +from __future__ import annotations + +import sys +from pathlib import Path + +LIB = Path(__file__).resolve().parent +sys.path.insert(0, str(LIB)) + +from bell202_line_code import decode_tones_to_bits, encode_bits_to_tones # noqa: E402 + + +def test_roundtrip_single_byte() -> None: + bits = bytes([0x55]) # 01010101 → many toggles + tones = encode_bits_to_tones(bits, start_tone=0) + back = decode_tones_to_bits(tones, start_tone=0) + assert back == bits + + +def test_all_ones_no_toggle() -> None: + tones = encode_bits_to_tones(bytes([0xFF]), start_tone=0) + assert len(set(tones)) == 1 + + +def main() -> int: + test_roundtrip_single_byte() + test_all_ones_no_toggle() + print("OK: bell202_line_code tests") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) -- cgit v1.3.1