AsyncDsssPool: the population's lifecycle¶

import numpy as np

from doppler.dsss import AsyncDsssPool
from doppler.telemetry import EventLog
from doppler.wfm import Gold, Synth, _SynthEngine, wfm_awgn_amplitude

SF = 1023
CHIP_RATE = 5.0e6
SPC = 2
FS = CHIP_RATE * SPC
SYM_RATE = 2700.0
TE = SF * SPC  # one code epoch, the push block
CN0_DBHZ = 47.0
DU = 6000.0  # the searcher's span: +-6 kHz
# The code-only window the searcher aligns its blocks inside: 20 symbols of
# every 270 (ten frames a second; the operating point's is 450 of 4950),
# holding the 31 whole epochs a depth of 16 needs at any chip phase -- rows
# of 305 Hz, inside the cell receivers' pull-in (the pool refuses a
# shallower searcher).
W_SYM, F_SYM, CODE_ONLY_EPOCHS = 20, 270, 31
LOST_S = 0.3  # the release interval, short enough to see
N_SLOTS = 4
EMITTERS = {  # name: (Doppler Hz, code phase at sample 0 in chips, seed)
    "A": (1500.0, 0, 1),
    "B": (-3500.0, 900, 2),
}
ON_S, OFF_S = 1.0, 3.0 * LOST_S  # A always on; B on, off, on
CODE = np.asarray(Gold().generate(SF)).astype(np.uint8)

def emitter(doppler_hz: float, chip0: int, seed: int) -> _SynthEngine:
    """One emitter: the shipped continuous DSSS at a carrier offset, clean,
    with the waveform's code-only window, its code `chip0` chips in at the
    stream's first sample (a burn-in the caller discards) -- two emitters
    must differ in code phase as well as Doppler, since one phase is one
    peak to the searcher. The engine rather than ``Synth``: the window is
    the engine's alone (#1294)."""
    syn = _SynthEngine(
        type="dsss",
        fs=FS,
        freq=doppler_hz,
        snr=100.0,  # clean: the noise is added once, at the sum
        sps=SPC,
        seed=seed,
    )
    syn.set_dsss_cont(CODE, CHIP_RATE / SYM_RATE, data="prbs")
    syn.set_dsss_window(W_SYM, F_SYM)
    if chip0:
        syn.steps(chip0 * SPC)
    return syn


def stimulus(seed: int = 7):
    """Both emitters on for ON_S, B off for OFF_S (its synth keeps running:
    coming back is not restarting), both on again; noise at the sum.
    Returns the capture and B's on-air mask per epoch."""
    a = emitter(*EMITTERS["A"])
    b = emitter(*EMITTERS["B"])
    n_on, n_off = int(ON_S * FS / TE), int(OFF_S * FS / TE)
    on_b = np.array([1] * n_on + [0] * n_off + [1] * n_on, dtype=bool)
    blocks = []
    for on in on_b:
        sa = a.steps(TE)
        sb = b.steps(TE)
        blocks.append(sa + sb if on else sa)
    x = np.concatenate(blocks)
    # C/N0 referred to fs; the Synth emits unit total power, the amplitude
    # is per component, and the sqrt(2) bridges the two conventions.
    amp = float(wfm_awgn_amplitude(CN0_DBHZ - 10.0 * np.log10(FS), 1.0))
    noise = Synth(type="noise", fs=FS, seed=seed).steps(len(x))
    return (x + amp * np.sqrt(2.0) * noise).astype(np.complex64), on_b

def owner(pool: AsyncDsssPool, slot: int) -> str | None:
    """Which emitter a slot's SEED is -- row and chip phase -- or None
    for a false alarm (a noise seed at another phase)."""
    r = pool.status(slot)
    if not r.assigned:
        return None
    for name, (f, chip0, _) in EMITTERS.items():
        truth = (r.seed_sample / SPC + chip0) % SF
        dc = abs(r.seed_chip_phase - truth)
        dc = min(dc, SF - dc)
        if abs(r.seed_doppler_hz - f) <= pool.doppler_res_hz and dc <= 1.0:
            return name
    return "false alarm"


def run(x: np.ndarray, events_path: str):
    pool = AsyncDsssPool(
        CODE,
        chip_rate=CHIP_RATE,
        symbol_rate=SYM_RATE,
        spc=SPC,
        cn0_dbhz=CN0_DBHZ,
        pfa=1e-3,
        doppler_uncertainty=DU,
        code_only_epochs=CODE_ONLY_EPOCHS,
        n_slots=N_SLOTS,
        lost_confirm_s=LOST_S,
        threads=1,
    )
    log = EventLog(events_path)
    pool.set_event_log(log)
    n_ep = len(x) // TE
    who = np.full((N_SLOTS, n_ep), "", dtype=object)  # owner per epoch
    dopp = np.full((N_SLOTS, n_ep), np.nan)  # live Doppler while tracking
    n_syms = dict.fromkeys(EMITTERS, 0)
    for k in range(n_ep):
        pool.push(x[k * TE : (k + 1) * TE])
        for i in range(N_SLOTS):
            o = owner(pool, i)
            r = pool.status(i)
            who[i, k] = o or ""
            if o in n_syms:
                n_syms[o] += len(pool.symbols(i))
            if r.assigned and r.state == 2:  # tracking
                dopp[i, k] = r.doppler_hz
    pool.set_event_log(None)
    log.close()
    with open(events_path) as fh:
        events = [json.loads(line) for line in fh]
    assert len(events) == pool.events, "every transition reached the log"
    return who, dopp, n_syms, events