#!/usr/bin/env python3 # Rerun of DeepSeek V4.1 Flash's analysis/deepseek41-review/adv_chain.py (its harness, unchanged), sweeping codec rate error # with and without a 40-dibit lead-in before the first frame sync. Reproduces https://www.innerpulse.net/DEEPSEEKV41FLASH/ # Run: DM1701_REPO=/path/to/DM-1701 python3 knee_rerun.py (needs gcc; writes build/ next to this script) """Adversarial live-chain test: drives the PRODUCTION firmware sources (p25_resample + p25_4fsk stream demod + p25rx) the way p25monProcessSamples does (32-sample streaming resample chunks, 768-sample windows, 512 hop), under conditions the passing suite never combines: A: streaming (per-32) resample + noise + clock drift + freq offset B: codec rate != nominal guess (resampler uses guess only) -> gate/tracker mismatch C: DC wander on top of noise D: MSB/LSB-swapped voice bits (transport blindness check) """ import ctypes as C import os import subprocess import sys from pathlib import Path ROOT = Path(os.environ.get('DM1701_REPO', '.')).resolve() LIVE = ROOT / 'opengd77-rt3s-experiments/MDUV380_firmware/application/source/p25' DEMO = ROOT / 'dm1701-p25-demo/src/p25' HERE = Path(__file__).resolve().parent sys.path.insert(0, str(ROOT / 'tests' / 'p25')) from run import ldu # noqa: E402 (independent wire builder, status every 36th) SO = Path(os.environ.get('C4FM_SIM_WORK', HERE / 'build')) / 'adv_chain.so' SO.parent.mkdir(parents=True, exist_ok=True) def build(): mbe = list((DEMO / 'mbelib').glob('*.c')) cmd = ['gcc', '-O2', '-DP25_DEMOD_WINDOW=768', '-I' + str(DEMO), '-I' + str(LIVE), '-I' + str(ROOT / 'ref-sources/dsd-master/include'), '-shared', '-fPIC', str(ROOT / 'tests/p25/host.c'), str(ROOT / 'tests/p25/trunk_host.c')] + \ [str(LIVE / n) for n in ('p25rx.c', 'p25_nid.c', 'p25_fec.c', 'p25_trunk.c', 'p25_4fsk.c', 'p25_resample.c')] + \ [str(LIVE / 'p25mon_capture.c')] + list(map(str, mbe)) + \ ['-lm', '-o', str(SO)] subprocess.run(cmd, check=True) class Timing(C.Structure): _fields_ = [('next_sample', C.c_float), ('period', C.c_float), ('initialized', C.c_int)] class RState(C.Structure): _fields_ = [('t', C.c_float), ('last', C.c_int16), ('init', C.c_int)] def setup(lib): lib.p25_channel_generate.argtypes = [C.POINTER(C.c_uint8), C.c_int, C.c_int, C.c_int, C.c_int, C.POINTER(C.c_int16), C.c_int] lib.p25_channel_generate.restype = C.c_int lib.p25_resample.argtypes = [C.POINTER(C.c_int16), C.c_int, C.c_float, C.POINTER(C.c_int16), C.c_int, C.POINTER(RState)] lib.p25_resample.restype = C.c_int lib.p25_4fsk_demodulate_stream.argtypes = [C.POINTER(C.c_int16), C.c_int, C.POINTER(C.c_uint8), C.c_int, C.c_int, C.POINTER(Timing)] lib.p25_4fsk_demodulate_stream.restype = C.c_int lib.test_feed.argtypes = [C.POINTER(C.c_uint8), C.c_int] lib.test_payload.restype = C.POINTER(C.c_uint8) lib.test_metric.argtypes = [C.c_int] lib.test_metric.restype = C.c_int32 return lib def run_case(lib, name, wire, snr, ppm, fo, front_ratio, live_ratio, chunk_in=32, dc_add=0, swap_bits=False): """front_ratio: 38.4k->codec decimation actually applied (codec truth). live_ratio: codec->38.4k ratio the firmware uses (nominal guess).""" n = len(wire) src = (C.c_uint8 * n)(*wire) ch = (C.c_int16 * (n * 8))() nch = lib.p25_channel_generate(src, n, snr, ppm, fo, ch, len(ch)) # front-end decimation to codec domain (whole-buffer linear step is what # test-4 does; streaming effects are exercised on the upsample side) rs = RState(); rs.init = 0 dec_cap = int(nch * front_ratio) + 16 dec = (C.c_int16 * dec_cap)() ndec = lib.p25_resample(ch, nch, C.c_float(front_ratio), dec, dec_cap, C.byref(rs)) buf = bytearray(dec_cap * 2) C.memmove((C.c_int16 * dec_cap).from_buffer(buf), dec, ndec * C.sizeof(C.c_int16)) dec_list = list((C.c_int16 * ndec).from_buffer_copy(bytes(buf[:ndec * 2]))) if dc_add: dec_list = [max(-32768, min(32767, s + dc_add)) for s in dec_list] # live chain: streaming upsample in 32-sample chunks rs2 = RState(); rs2.init = 0 re_list = [] pos = 0 tmp_in = (C.c_int16 * chunk_in)() tmp_out = (C.c_int16 * 256)() while pos < len(dec_list): blk = dec_list[pos:pos + chunk_in] for i, s in enumerate(blk): tmp_in[i] = s got = lib.p25_resample(tmp_in, len(blk), C.c_float(live_ratio), tmp_out, 256, C.byref(rs2)) re_list.extend(tmp_out[i] for i in range(got)) pos += len(blk) nre = len(re_list) re_arr = (C.c_int16 * nre)(*re_list) out = (C.c_uint8 * 110)() timing = Timing() lib.test_init() counts = [] off = 0 while off + 768 <= nre: window = C.cast(C.byref(re_arr, off * 2), C.POINTER(C.c_int16)) cnt = lib.p25_4fsk_demodulate_stream(window, 768, out, 110, 512, C.byref(timing)) counts.append(cnt) if swap_bits: swapped = (C.c_uint8 * cnt)(*[(out[i] ^ 1) & 3 for i in range(cnt)]) lib.test_feed(swapped, cnt) else: lib.test_feed(out, cnt) off += 512 frames = lib.test_metric(0) nids = lib.test_metric(1) got = list(lib.test_payload()[:lib.test_payload_len()]) return {'name': name, 'nch': nch, 'ndec': ndec, 'nre': nre, 'windows': len(counts), 'bad_counts': [c for c in counts if c not in (63, 64, 65)], 'frames': frames, 'nids': nids, 'got_len': len(got)} def main(): import json build() lib = setup(C.CDLL(str(SO))) wire = [] for kind in (5, 10, 5, 10): frame, payload = ldu(kind) wire += frame R24 = 24000.0 / 38400.0 UP = 38400.0 / 24000.0 errs = [-2.0, -1.5, -1.0, -0.8, -0.6, -0.4, -0.2, 0.0, 0.2, 0.4, 0.6, 0.8, 1.0, 1.5, 2.0] res = {'errors_pct': errs, 'lead0': [], 'lead40': []} for lead, key in ((0, 'lead0'), (40, 'lead40')): w = [0] * lead + wire + [0] * 80 for e in errs: r = run_case(lib, f'lead={lead} err={e}', w, 20, 0, 0, R24 * (1 + e / 100.0), UP) res[key].append({'frames': r['frames'], 'nids': r['nids']}) print(f"lead {lead:2d} error {e:+.1f}% frames {r['frames']:2d}/36 NIDs {r['nids']}/4", flush=True) json.dump(res, open(SO.parent / 'knee.json', 'w')) if __name__ == '__main__': main()