"""Synthetic test signals shared by the test modules.""" import numpy as np from scipy.signal import butter, hilbert, lfilter FS = 32000.0 def _noise(x, snr_db, rng): p = float(np.mean(np.abs(x) ** 2)) n = np.sqrt(p / (2 * 10 ** (snr_db / 10.0))) return (x + n * (rng.standard_normal(x.size) + 1j * rng.standard_normal(x.size))).astype(np.complex64) def voice(n, fs=FS, seed=0): """Band-limited noise with a syllabic envelope -- a good speech stand-in.""" rng = np.random.default_rng(seed) b, a = butter(4, [300 / (fs / 2), 2700 / (fs / 2)], btype="band") v = lfilter(b, a, rng.standard_normal(n)) v /= max(np.abs(v).max(), 1e-9) t = np.arange(n) / fs return v * (0.4 + 0.6 * np.abs(np.sin(2 * np.pi * 1.7 * t))) def make(kind, n=64000, fs=FS, snr_db=30.0, seed=3): rng = np.random.default_rng(seed) t = np.arange(n) / fs v = voice(n, fs, seed) if kind == "nfm": msg = v + 0.15 * np.sin(2 * np.pi * 100.0 * t) x = np.exp(1j * np.cumsum(2 * np.pi * 2500 * msg / fs)) elif kind == "wfm": x = np.exp(1j * np.cumsum(2 * np.pi * 3000 * v / fs)) elif kind == "am": x = ((1 + 0.6 * v) * np.exp(2j * np.pi * 30 * t)) elif kind == "usb": x = 0.5 * hilbert(v) elif kind == "lsb": x = 0.5 * np.conj(hilbert(v)) elif kind == "carrier": x = np.exp(2j * np.pi * 137 * t) elif kind == "cw": dot = 0.08 pat = [1, 0, 1, 1, 1, 0, 0, 0, 1, 1, 1, 0, 1, 0, 0, 0, 0, 0, 0] key = np.zeros(n) i = 0 while i < n: for b in pat: m = int(dot * fs) if i + m > n: break key[i:i + m] = b i += m env = np.convolve(key, np.hanning(int(0.005 * fs)), "same") env /= max(env.max(), 1e-9) x = env * np.exp(2j * np.pi * 300 * t) elif kind.startswith("fsk"): levels = int(kind[3]) baud, dev = (1200.0, 2400.0) if levels == 2 else (4800.0, 1800.0) sp = int(fs / baud) sym = rng.integers(0, levels, n // sp + 1) lv = (sym - (levels - 1) / 2) / max(1, (levels - 1) / 2) f = np.resize(np.repeat(lv, sp) * dev, n) x = np.exp(1j * np.cumsum(2 * np.pi * f / fs)) elif kind.startswith("psk"): m = int(kind[3]) sp = int(fs / 4800.0) sym = rng.integers(0, m, n // sp + 1) x = np.exp(1j * np.resize(np.repeat(2 * np.pi * sym / m, sp), n)) elif kind == "noise": return (0.01 * (rng.standard_normal(n) + 1j * rng.standard_normal(n))).astype(np.complex64) else: raise ValueError(kind) return _noise(np.asarray(x, dtype=np.complex128), snr_db, rng) # --------------------------------------------------------------------------- # Packets. What the decoder is for: signals that carry something, rather # than random keying that only exercises the classifier. # --------------------------------------------------------------------------- def _rf(env, fs, snr_db=30.0, seed=1): """An envelope on a carrier, with noise.""" x = np.asarray(env, dtype=np.float64).astype(np.complex128) rng = np.random.default_rng(seed) p = float(np.mean(np.abs(x) ** 2)) or 1e-9 n = np.sqrt(p / (2 * 10 ** (snr_db / 10.0))) return (x + n * (rng.standard_normal(x.size) + 1j * rng.standard_normal(x.size))).astype(np.complex64) def ook_pwm(bits, fs=50_000.0, alpha=350e-6, repeats=4, gap=10e-3, sync=True, snr_db=30.0, seed=1, fixed_gap=None): """Pulse-width keying, as an EV1527 or PT2262 remote sends it. With ``fixed_gap`` the gap is constant and only the pulse varies; without it the two are complementary and the bit period is constant, which is what the common parts actually do. """ a = int(round(alpha * fs)) env = [] for _ in range(repeats): if sync: env += [1] * a + [0] * (31 * a) # the sync pulse and its gap for b in bits: if fixed_gap is None: high, low = (3 * a, a) if b == "1" else (a, 3 * a) else: high = 3 * a if b == "1" else a low = int(round(fixed_gap * fs)) env += [1] * high + [0] * low env += [0] * int(gap * fs) return _rf(env, fs, snr_db, seed) def ook_ppm(bits, fs=50_000.0, alpha=400e-6, repeats=4, gap=10e-3, snr_db=30.0, seed=2): """Pulse-distance keying: the pulse is constant, the gap carries the bit.""" a = int(round(alpha * fs)) env = [] for _ in range(repeats): for b in bits: env += [1] * a + [0] * (3 * a if b == "1" else a) env += [0] * int(gap * fs) return _rf(env, fs, snr_db, seed) def ook_manchester(bits, fs=50_000.0, baud=2000.0, repeats=3, gap=10e-3, snr_db=30.0, seed=3, preamble="10101010"): half = int(round(fs / baud / 2)) env = [] for _ in range(repeats): for b in preamble + bits: first, second = (1, 0) if b == "1" else (0, 1) # IEEE 802.3 env += [first] * half + [second] * half env += [0] * int(gap * fs) return _rf(env, fs, snr_db, seed) def fsk_nrz(bits, fs=48_000.0, baud=1200.0, dev=4500.0, snr_db=30.0, seed=4): """Two-level FSK holding each bit for one symbol period.""" sp = fs / baud n = int(round(len(bits) * sp)) idx = np.clip((np.arange(n) / sp).astype(int), 0, len(bits) - 1) level = np.array([1.0 if c == "1" else -1.0 for c in bits])[idx] rng = np.random.default_rng(seed) x = np.exp(1j * np.cumsum(2 * np.pi * level * dev / fs)) nz = np.sqrt(1.0 / (2 * 10 ** (snr_db / 10.0))) return (x + nz * (rng.standard_normal(n) + 1j * rng.standard_normal(n))).astype(np.complex64) _C4FM = {"01": 3, "00": 1, "10": -1, "11": -3} def c4fm(bits, fs=48_000.0, baud=4800.0, dev=1800.0, snr_db=30.0, seed=8): """Four-level FSK, mapped the way P25 and DMR map it.""" bits = bits[:len(bits) - len(bits) % 2] level = np.array([_C4FM[bits[i:i + 2]] / 3.0 for i in range(0, len(bits), 2)]) sp = fs / baud n = int(level.size * sp) idx = np.clip((np.arange(n) / sp).astype(int), 0, level.size - 1) rng = np.random.default_rng(seed) x = np.exp(1j * np.cumsum(2 * np.pi * level[idx] * dev / fs)) nz = np.sqrt(1.0 / (2 * 10 ** (snr_db / 10.0))) return (x + nz * (rng.standard_normal(n) + 1j * rng.standard_normal(n))).astype(np.complex64) def ax25_address(call, ssid, last=False): call = (call.upper() + " ")[:6] return bytes((ord(c) << 1) & 0xFE for c in call) + \ bytes([0x60 | ((ssid & 0x0F) << 1) | (1 if last else 0)]) def ax25_frame(source, dest, info, path=()): """A complete AX.25 frame, with its frame-check sequence.""" body = ax25_address(*dest) + ax25_address(*source, last=not path) for i, hop in enumerate(path): body += ax25_address(*hop, last=(i == len(path) - 1)) body += bytes([0x03, 0xF0]) + info.encode() crc = 0xFFFF for byte in body: crc ^= byte for _ in range(8): crc = (crc >> 1) ^ 0x8408 if crc & 1 else crc >> 1 crc ^= 0xFFFF return body + bytes([crc & 0xFF, (crc >> 8) & 0xFF]) def aprs_afsk(frames, fs=48_000.0, baud=1200.0, snr_db=30.0, seed=6, mark=1200.0, space=2200.0, dev=3000.0, flags=8): """AX.25 over Bell 202 AFSK inside an FM carrier, as APRS is sent.""" stream = "" for frame in frames: body = "01111110" * flags ones = 0 for byte in frame: for k in range(8): # least significant first bit = (byte >> k) & 1 body += "1" if bit else "0" if bit: ones += 1 if ones == 5: # stuff a zero after five body += "0" ones = 0 else: ones = 0 stream += body + "01111110" * flags level, nrzi = 1, [] for bit in stream: if bit == "0": level ^= 1 # NRZI: a zero is a change nrzi.append(level) sp = fs / baud n = int(len(nrzi) * sp) idx = np.clip((np.arange(n) / sp).astype(int), 0, len(nrzi) - 1) tone = np.where(np.array(nrzi)[idx] == 1, mark, space) audio = np.sin(np.cumsum(2 * np.pi * tone / fs)) rng = np.random.default_rng(seed) x = np.exp(1j * np.cumsum(2 * np.pi * dev * audio / fs)) nz = np.sqrt(1.0 / (2 * 10 ** (snr_db / 10.0))) return (x + nz * (rng.standard_normal(n) + 1j * rng.standard_normal(n))).astype(np.complex64)