AsyncDsssPool: emitters that come and go¶

One dsss.AsyncDsssPool runs three receivers on one stream. Four emitters on one Gold code reach it at different times, code phases and carriers. A, B and C arrive one after another, B leaves, and D arrives after it.

The pool's controller runs one search over the whole carrier range while any receiver is idle. It leaves out the code phases the receivers hold, so the search never takes a signal already held, and seeds each idle receiver from its picks, strongest first, so no two receivers take one signal. When a receiver's emitter leaves, its release rule fires, the receiver returns to idle, and the search seeds it again.

What the run shows¶

One receiver per emitter. While A, B and C are on the air, each is held by exactly one receiver, and no two receivers hold one signal.
A departure frees a receiver. After B leaves, its receiver is released once both of its lock detectors have stayed down for lost_confirm_s, and it returns to idle for the controller's search to seed again.
The freed receiver takes the newcomer. D is taken by the receiver B freed, without disturbing the two that track A and C.

How it is built¶

Each emitter is wfm.Synth in continuous DSSS mode, placed on the stream at its own sample offset and carrier, with source.AWGN at the Es/N0 every emitter shares.

import numpy as np

from doppler.examples._extra import require
from doppler.source import AWGN
from doppler.wfm import Gold, Synth, wfm_awgn_amplitude

SF = 1023  # Gold code period, chips (CCSDS)
CHIP_RATE = 3.069e6  # chips per second
SPC = 2  # samples per chip
FS = CHIP_RATE * SPC  # sample rate, 6.138 MSa/s
SYM_RATE = 2700.0  # BPSK symbols per second, asynchronous to the code
TE = SF * SPC  # one code epoch, samples: one push
EPOCH_S = TE / FS  # seconds per push
CARRIER_UNCERTAINTY = 5000.0  # Hz, one-sided: the search range
ESN0_DB = 10.0  # Es/N0 of each emitter at the input
CN0_DBHZ = ESN0_DB + 10.0 * np.log10(SYM_RATE)  # C/N0, dB-Hz
LOST_CONFIRM_S = 0.1  # both lock detectors down this long before a release
N_RECEIVERS = 3
SEED_NOISE = 5

GOLD = np.asarray(Gold().generate(SF), np.uint8)

#: The emitters: name, carrier (Hz), code-phase offset (samples into its
#: first epoch, so each sits hundreds of chips from the others), and the
#: epochs it is on the air, [first, end).
EMITTERS = (
    ("A", 1500.0, 0, (90, 9600)),
    ("B", -2200.0, 600, (900, 4500)),
    ("C", 700.0, 1300, (1800, 9600)),
    ("D", -1000.0, 1800, (5700, 9600)),
)
N_EPOCHS = 9600  # 3.2 s


def make_stream(seed: int = SEED_NOISE) -> np.ndarray:
    """The capture: :data:`EMITTERS` on one Gold code, each from
    :class:`~doppler.wfm.Synth` in continuous DSSS mode with its own PRBS
    seed, placed on the stream at its own sample offset and carrier, plus
    :class:`~doppler.source.AWGN` at the Es/N0 every emitter shares.

    Returns
    -------
    x : NDArray[np.complex64]
        :data:`N_EPOCHS` code epochs.
    """
    n = N_EPOCHS * TE
    x = np.zeros(n, np.complex64)
    for k, (_, freq, offset, (a, b)) in enumerate(EMITTERS):
        s0, s1 = a * TE + offset, b * TE
        x[s0:s1] += np.asarray(
            Synth(
                type="dsss",
                fs=FS,
                seed=11 + k,  # each emitter its own data
                sps=SPC,
                data_code=GOLD.tobytes(),
                symbol_rate=SYM_RATE,
                freq=freq,
            ).steps(s1 - s0),
            np.complex64,
        )
    amp = float(
        wfm_awgn_amplitude(ESN0_DB - 10.0 * np.log10(FS / SYM_RATE), 1.0)
    )
    x += AWGN(seed=seed, amplitude=amp).generate(n).astype(np.complex64)
    return x


def emitter_phase(k: int, sample: int) -> float:
    """Emitter ``k``'s code phase, chips, at ``sample``: its code starts at
    chip 0 on its first sample."""
    _, _, offset, (a, _) = EMITTERS[k]
    return ((sample - (a * TE + offset)) / SPC) % SF

The pool is built like one receiver, plus n_receivers. lost_confirm_s arms each receiver's release rule. The pool is fed one code epoch per push, and the script reads every receiver's status() after each one.

def run(x: np.ndarray) -> dict:
    """Push the capture one code epoch at a time and record every
    receiver's status after each push.

    The pool is built like one :class:`~doppler.dsss.AsyncDsssReceiver`,
    plus ``n_receivers``. ``lost_confirm_s`` arms each receiver's release
    rule, which is what frees a receiver whose emitter left.

    Returns
    -------
    rec : dict
        ``state``, ``phase`` and ``locked``, each ``(epochs, receivers)``;
        ``symbols``, each receiver's last 600 symbols; and the pool's
        ``released_lost`` and ``released_duplicate`` counts.
    """
    pool = AsyncDsssPool(
        GOLD,
        n_receivers=N_RECEIVERS,
        chip_rate=CHIP_RATE,
        symbol_rate=SYM_RATE,
        sample_rate=FS,
        cn0_dbhz=CN0_DBHZ,
        doppler_uncertainty=CARRIER_UNCERTAINTY,
        lost_confirm_s=LOST_CONFIRM_S,
    )
    shape = (N_EPOCHS, N_RECEIVERS)
    state = np.zeros(shape, int)
    phase = np.zeros(shape)
    locked = np.zeros(shape, int)
    tail = [[] for _ in range(N_RECEIVERS)]
    rx1_t, rx1_s = [], []  # every symbol receiver 1 emits, and when
    for e in range(N_EPOCHS):
        pool.push(x[e * TE : (e + 1) * TE])
        for i in range(N_RECEIVERS):
            st = pool.status(i)
            state[e, i], phase[e, i], locked[e, i] = (
                st.state,
                st.chip_phase,
                st.locked,
            )
            sym = pool.symbols(i)
            tail[i] = (tail[i] + list(sym))[-600:]
            if i == 1 and len(sym):
                # Spread over the push that emitted them: a symbol's time
                # to within one code epoch.
                rx1_t.extend(
                    e * EPOCH_S + EPOCH_S * np.arange(len(sym)) / len(sym)
                )
                rx1_s.extend(sym)
    return {
        "state": state,
        "phase": phase,
        "locked": locked,
        "symbols": [np.asarray(t) for t in tail],
        "rx1_t": np.asarray(rx1_t),
        "rx1_s": np.asarray(rx1_s),
        "released_lost": pool.released_lost,
        "released_duplicate": pool.released_duplicate,
    }

The checks identify the emitter each receiver holds by its code phase, and assert the claims above from the pool's own reports.

def holder_of(rec: dict) -> np.ndarray:
    """Which emitter each receiver holds at each epoch's end: the index of
    the emitter within 1.5 chips of the receiver's code phase, or -1
    (searching). A receiver still holding an emitter that has left -- until
    its release fires -- reads that emitter."""
    out = np.full(rec["state"].shape, -1)
    for e in range(N_EPOCHS):
        end = (e + 1) * TE
        for i in range(N_RECEIVERS):
            if rec["state"][e, i] not in (1, 2):
                continue
            for k, (_, _, _, (a, _b)) in enumerate(EMITTERS):
                d = abs(rec["phase"][e, i] - emitter_phase(k, end))
                if e >= a and min(d, SF - d) <= 1.5:
                    out[e, i] = k
    return out


def check(rec: dict) -> dict:
    """The run's claims, asserted from the pool's own reports.

    Settling margins: an emitter is counted from 0.5 s after it arrives,
    and a departed one from ``lost_confirm_s`` plus 0.2 s after it leaves.
    """
    held = holder_of(rec)
    settle = int(0.5 / EPOCH_S)
    gone = int((LOST_CONFIRM_S + 0.2) / EPOCH_S)
    arrive_ms = {}
    for k, (name, _, _, (a, b)) in enumerate(EMITTERS):
        span = held[a + settle : b]
        n_holders = (span == k).sum(axis=1)
        assert (n_holders == 1).all(), (
            f"{name}: held by {set(n_holders)} receivers while on the air"
        )
        first = np.flatnonzero((held[a:b] == k).any(axis=1))[0]
        arrive_ms[name] = first * EPOCH_S * 1e3
        if b < N_EPOCHS:  # left: released within the confirm time
            assert not (held[b + gone :] == k).any(), (
                f"{name} still held after it left"
            )
            last = np.flatnonzero((held == k).any(axis=1))[-1]
            release_ms = (last + 1 - b) * EPOCH_S * 1e3
    # A departure freed a receiver, and the newcomer took it.
    assert rec["released_lost"] >= 1
    b_rx = int(np.flatnonzero((held == 1).any(axis=0))[0])
    d_rx = int(np.flatnonzero((held == 3).any(axis=0))[0])
    assert b_rx == d_rx, "D was taken by the receiver B freed"
    # Every receiver holding a signal at the end has symbol lock.
    for i in range(N_RECEIVERS):
        assert rec["locked"][-1, i] == 1, f"receiver {i} not symbol-locked"
    return {"arrive_ms": arrive_ms, "b_rx": b_rx, "release_ms": release_ms}

What you're seeing¶

What each receiver holds (top): the emitters on the air, then one row per receiver, coloured by the emitter it holds. Grey is refining after a hand-off. The hatched span is receiver 1 still holding B after B left, until its release fires.
Receiver 1 over time (middle): B's symbols, then the noise it emits after B leaves, while its symbol lock is down and before the release, then nothing while it is idle, then D's symbols. The symbol lock, status().locked, says when the symbols are a signal's.
Each receiver's last 600 symbols (bottom): A, D and C, one BPSK constellation each.

Reproduce¶

pip install "doppler-dsp[examples]"   # matplotlib + scipy, once
python -m doppler.examples.async_dsss_pool_demo out.png

The design is in the async DSSS design, §5-§9.

Source: async_dsss_pool_demo.py.