207 lines
9 KiB
Python
207 lines
9 KiB
Python
#!/usr/bin/env python3
|
|
|
|
import os
|
|
from os import path
|
|
import subprocess
|
|
import json
|
|
from collections import namedtuple, defaultdict
|
|
from tqdm import tqdm
|
|
import uuid
|
|
import multiprocessing
|
|
import sqlite3
|
|
import time
|
|
from urllib.parse import urlparse
|
|
import functools
|
|
import tempfile
|
|
import itertools
|
|
|
|
import numpy as np
|
|
np.set_printoptions(linewidth=240)
|
|
|
|
from dsss_demod_test_waveform_gen import load_noise_meas_params, load_noise_synth_params,\
|
|
mains_noise_measured, mains_noise_synthetic, modulate as dsss_modulate
|
|
|
|
|
|
def build_test_binary(nbits, thf, decimation, symbols, cachedir):
|
|
build_id = str(uuid.uuid4())
|
|
builddir = path.join(cachedir, build_id)
|
|
os.mkdir(builddir)
|
|
|
|
cwd = path.join(path.dirname(__file__), '..')
|
|
|
|
env = os.environ.copy()
|
|
env['BUILDDIR'] = path.abspath(builddir)
|
|
env['DSSS_GOLD_CODE_NBITS'] = str(nbits)
|
|
env['DSSS_DECIMATION'] = str(decimation)
|
|
env['DSSS_THRESHOLD_FACTOR'] = str(thf)
|
|
env['DSSS_WAVELET_WIDTH'] = str(0.73 * decimation)
|
|
env['DSSS_WAVELET_LUT_SIZE'] = str(10 * decimation)
|
|
env['TRANSMISSION_SYMBOLS'] = str(symbols)
|
|
|
|
with open(path.join(builddir, 'make_stdout.txt'), 'w') as stdout,\
|
|
open(path.join(builddir, 'make_stderr.txt'), 'w') as stderr:
|
|
subprocess.run(['make', 'clean', os.path.abspath(path.join(builddir, 'tools/dsss_demod_test'))],
|
|
env=env, cwd=cwd, check=True, stdout=stdout, stderr=stderr)
|
|
|
|
return build_id
|
|
|
|
@functools.lru_cache()
|
|
def load_noise_gen(url):
|
|
schema, refpath = url.split('://')
|
|
if not path.isabs(refpath):
|
|
refpath = path.abspath(path.join(path.dirname(__file__), refpath))
|
|
|
|
if schema == 'meas':
|
|
return mains_noise_measured, load_noise_meas_params(refpath)
|
|
elif schema == 'synth':
|
|
return mains_noise_synthetic, load_noise_synth_params(refpath)
|
|
else:
|
|
raise ValueError('Invalid schema', schema)
|
|
|
|
def sequence_matcher(test_data, decoded, max_shift=3):
|
|
match_result = []
|
|
for shift in range(-max_shift, max_shift):
|
|
failures = -shift if shift < 0 else 0 # we're skipping the first $shift symbols
|
|
a = test_data if shift > 0 else test_data[-shift:]
|
|
b = decoded if shift < 0 else decoded[shift:]
|
|
for i, (ref, found) in enumerate(itertools.zip_longest(a, b)):
|
|
if ref is None: # end of signal
|
|
break
|
|
if ref != found:
|
|
failures += 1
|
|
match_result.append(failures)
|
|
failures = min(match_result)
|
|
return failures/len(test_data)
|
|
|
|
ResultParams = namedtuple('ResultParams', ['nbits', 'thf', 'decimation', 'symbols', 'seed', 'amplitude', 'background'])
|
|
|
|
def run_test(seed, amplitude_spec, background, nbits, decimation, symbols, thfs, lookup_binary, cachedir):
|
|
noise_gen, noise_params = load_noise_gen(background)
|
|
|
|
test_data = np.random.RandomState(seed=seed).randint(0, 2 * (2**nbits), symbols)
|
|
|
|
signal = np.repeat(dsss_modulate(test_data, nbits) * 2.0 - 1, decimation)
|
|
# We're re-using the seed here. This is not a problem.
|
|
noise = noise_gen(seed, len(signal), *noise_params)
|
|
|
|
amplitudes = amplitude_spec[0] * 10 ** np.linspace(0, amplitude_spec[1], amplitude_spec[2])
|
|
output = []
|
|
for amp in amplitudes:
|
|
with tempfile.NamedTemporaryFile(dir=cachedir) as f:
|
|
waveform = signal*amp + noise
|
|
f.write(waveform.astype('float').tobytes())
|
|
f.flush()
|
|
|
|
for thf in thfs:
|
|
cmdline = [lookup_binary(nbits, thf, decimation, symbols), f.name]
|
|
proc = subprocess.Popen(cmdline, stdout=subprocess.PIPE, text=True)
|
|
stdout, _stderr = proc.communicate()
|
|
if proc.returncode != 0:
|
|
raise SystemError(f'Subprocess signalled error: {proc.returncode=}')
|
|
|
|
lines = stdout.splitlines()
|
|
matched = [ l.partition('[')[2].partition(']')[0]
|
|
for l in lines if l.strip().startswith('data sequence received:') ]
|
|
matched = [ [ int(elem) for elem in l.split(',') ] for l in matched ]
|
|
|
|
ser = min(sequence_matcher(test_data, match) for match in matched) if matched else None
|
|
rpars = ResultParams(nbits, thf, decimation, symbols, seed, amp, background)
|
|
output.append((rpars, ser))
|
|
print(f'ran {rpars} {ser=} {" ".join(cmdline)}')
|
|
return output
|
|
|
|
def parallel_generator(db, table, columns, builder, param_list, desc, context={}, params_mapper=lambda *args: args):
|
|
with multiprocessing.Pool(multiprocessing.cpu_count()) as pool:
|
|
with db as conn:
|
|
jobs = []
|
|
for params in param_list:
|
|
found_res = conn.execute(
|
|
f'SELECT result FROM {table} WHERE ({",".join(columns)}) = ({",".join("?"*len(columns))})',
|
|
params_mapper(*params)).fetchone()
|
|
|
|
if found_res:
|
|
yield params, json.loads(*found_res)
|
|
|
|
else:
|
|
jobs.append((params, pool.apply_async(builder, params, context)))
|
|
|
|
pool.close()
|
|
print('Using', len(param_list) - len(jobs), 'cached jobs', flush=True)
|
|
with tqdm(total=len(jobs), desc=desc) as tq:
|
|
for params, res in jobs:
|
|
tq.update(1)
|
|
result = res.get()
|
|
with db as conn:
|
|
conn.execute(f'INSERT INTO {table} VALUES ({"?,"*len(params)}?,?)',
|
|
(*params_mapper(*params), json.dumps(result), timestamp()))
|
|
yield params, result
|
|
pool.join()
|
|
|
|
if __name__ == '__main__':
|
|
import argparse
|
|
parser = argparse.ArgumentParser()
|
|
parser.add_argument('-d', '--dump', help='Write results to JSON file')
|
|
parser.add_argument('-c', '--cachedir', default='dsss_test_cache', help='Directory to store build output and data in')
|
|
args = parser.parse_args()
|
|
|
|
DecoderParams = namedtuple('DecoderParams', ['nbits', 'thf', 'decimation', 'symbols'])
|
|
dec_paramses = [ DecoderParams(nbits=nbits, thf=thf, decimation=decimation, symbols=20)
|
|
for nbits in [5, 6]
|
|
for thf in [4.5, 4.0, 5.0]
|
|
for decimation in [10, 5, 22] ]
|
|
# dec_paramses = [ DecoderParams(nbits=nbits, thf=thf, decimation=decimation, symbols=100)
|
|
# for nbits in [5, 6, 7, 8]
|
|
# for thf in [1.5, 2.0, 2.5, 3.0, 3.5, 4.0, 4.5, 5.0, 5.5, 6.0, 6.5, 7.0, 7.5, 8.0, 8.5, 9.0, 9.5, 10.0]
|
|
# for decimation in [1, 2, 3, 4, 5, 6, 9, 10, 11, 12, 16, 22, 30, 40, 50] ]
|
|
|
|
build_cache_dir = path.join(args.cachedir, 'builds')
|
|
data_cache_dir = path.join(args.cachedir, 'data')
|
|
os.makedirs(build_cache_dir, exist_ok=True)
|
|
os.makedirs(data_cache_dir, exist_ok=True)
|
|
|
|
build_db = sqlite3.connect(path.join(args.cachedir, 'build_db.sqlite3'))
|
|
build_db.execute('CREATE TABLE IF NOT EXISTS builds (nbits, thf, decimation, symbols, result, timestamp)')
|
|
timestamp = lambda: int(time.time()*1000)
|
|
|
|
builds = dict(parallel_generator(build_db, table='builds', columns=['nbits', 'thf', 'decimation', 'symbols'],
|
|
builder=build_test_binary, param_list=dec_paramses, desc='Building decoders',
|
|
context=dict(cachedir=build_cache_dir)))
|
|
print('Done building decoders.')
|
|
|
|
GeneratorParams = namedtuple('GeneratorParams', ['seed', 'amplitude_spec', 'background'])
|
|
gen_params = [ GeneratorParams(rep, (5e-3, 1, 5), background)
|
|
#GeneratorParams(rep, (0.05e-3, 3.5, 50), background)
|
|
for rep in range(30)
|
|
for background in ['meas://fmeas_export_ocxo_2day.bin', 'synth://grid_freq_psd_spl_108pt.json'] ]
|
|
|
|
data_db = sqlite3.connect(path.join(args.cachedir, 'data_db.sqlite3'))
|
|
data_db.execute('CREATE TABLE IF NOT EXISTS waveforms'
|
|
'(seed, amplitude_spec, background, nbits, decimation, symbols, thresholds, result, timestamp)')
|
|
|
|
dec_param_groups = defaultdict(lambda: [])
|
|
for nbits, thf, decimation, symbols in dec_paramses:
|
|
dec_param_groups[(nbits, decimation, symbols)].append(thf)
|
|
waveform_params = [ (*gp, *dp, thfs) for gp in gen_params for dp, thfs in dec_param_groups.items() ]
|
|
print(f'Generated {len(waveform_params)} parameter sets')
|
|
|
|
def lookup_binary(*params):
|
|
return path.join(build_cache_dir, builds[tuple(params)], 'tools/dsss_demod_test')
|
|
|
|
def params_mapper(seed, amplitude_spec, background, nbits, decimation, symbols, thresholds):
|
|
amplitude_spec = ','.join(str(x) for x in amplitude_spec)
|
|
thresholds = ','.join(str(x) for x in thresholds)
|
|
return seed, amplitude_spec, background, nbits, decimation, symbols, thresholds
|
|
|
|
results = []
|
|
for _params, chunk in parallel_generator(data_db, 'waveforms',
|
|
['seed', 'amplitude_spec', 'background', 'nbits', 'decimation', 'symbols', 'thresholds'],
|
|
params_mapper=params_mapper,
|
|
builder=run_test,
|
|
param_list=waveform_params, desc='Generating waveforms',
|
|
context=dict(cachedir=data_cache_dir, lookup_binary=lookup_binary)):
|
|
results += chunk
|
|
|
|
if args.dump:
|
|
with open(args.dump, 'w') as f:
|
|
json.dump(results, f)
|
|
|