AsyncDsssPool: ten emitters through six receivers¶
One dsss.AsyncDsssPool runs six receivers on one
stream. Ten emitters on one Gold code come and go over six seconds, at most six
on the air at once. Each has its own code phase, its own level (within ±8 dB of
the others) and its own Doppler: a start and a rate, as from its own motion. The
Doppler stretches the chip clock as well as the carrier.
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 and seeds each idle receiver from its picks, strongest first, so two receivers never take one signal. A receiver whose emitter has left returns to idle.
What the run shows¶
- Each emitter is held by one receiver, from soon after it arrives to when it leaves. Nine of the ten are held within 50 ms of arriving. The receiver's carrier follows the emitter's Doppler ramp the whole time, and its symbol lock stays up.
- A departure frees a receiver, and a newcomer takes it. Ten emitters pass through six receivers. Receiver 0 holds A, then J. Receiver 2 holds C, then H.
- A weak emitter is taken beside a much stronger one, up to a limit. A search that skips the signals already held has to tell a second signal from the stronger ones' sidelobes, and it judges each candidate by what those sidelobes can reach at its own cell. At the floor, a one-epoch search takes a signal 14 dB below a stronger one in 90% of streams and 16 dB below in 78% (the design, §7.2). F arrives 16 dB below E and 13 dB below C: it waits for E to leave, and is taken while C is still on the air.
How it is built¶
Each emitter is wfm.Synth in continuous DSSS mode,
run through its own impairment.DopplerChannel
on a 2.2 GHz carrier, scaled to its level and placed on the stream. The noise is
source.AWGN. The truth the checks use, each emitter's
carrier and code phase over time, comes from the same channel: its Doppler law
and its group delay, delay_samples.
import numpy as np
from doppler.examples._extra import require
from doppler.impairment import DopplerChannel
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
RF_HZ = 2.2e9 # the carrier the Doppler is referred to: 1 ppm = 2200 Hz
CARRIER_UNCERTAINTY = 5000.0 # Hz, one-sided: the search range
ESN0_REF_DB = 20.0 # Es/N0 of an emitter at level 0 dB
LOST_CONFIRM_S = 0.1 # both lock detectors down this long before a release
N_RECEIVERS = 6
SEED_NOISE = 5
GOLD = np.asarray(Gold().generate(SF), np.uint8)
#: The emitters: name, level (dB, against ESN0_REF_DB), Doppler at arrival
#: (ppm) and its rate (ppm/s), code-phase offset (samples into its first
#: epoch) and the epochs it is on the air, [first, end).
EMITTERS = (
("A", +0.0, +1.20, -0.10, 0, (90, 12000)),
("B", -6.0, -1.50, +0.15, 600, (600, 6000)),
("C", +5.0, +0.30, +0.05, 1300, (1200, 9000)),
("D", -3.0, -0.40, -0.12, 1800, (2100, 18000)),
("E", +8.0, +1.80, -0.20, 300, (3000, 7500)),
("F", -8.0, -1.00, +0.08, 900, (6600, 15000)),
("G", +2.0, +0.80, +0.10, 1500, (7800, 18000)),
("H", -5.0, -1.90, +0.18, 200, (9600, 16500)),
("I", +6.0, +0.00, -0.15, 1100, (12300, 18000)),
("J", -2.0, -0.70, -0.06, 1700, (13500, 18000)),
)
N_EPOCHS = 18000 # 6 s
def channel(k: int) -> DopplerChannel:
"""Emitter ``k``'s Doppler: a clock offset and its rate, on RF_HZ."""
_, _, ppm, rate, _, _ = EMITTERS[k]
return DopplerChannel(
fs=FS, carrier_hz=RF_HZ, doppler_ppm=ppm, doppler_rate_ppm_s=rate
)
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, through its own :func:`channel`, scaled to its level and placed
on the stream at its own sample offset, plus
:class:`~doppler.source.AWGN` at the noise a level-0 emitter sees as
ESN0_REF_DB.
Returns
-------
x : NDArray[np.complex64]
:data:`N_EPOCHS` code epochs.
"""
n = N_EPOCHS * TE
x = np.zeros(n, np.complex64)
for k, (_, level, _, _, offset, (a, b)) in enumerate(EMITTERS):
s0, s1 = a * TE + offset, b * TE
# A clock running fast consumes more input than it emits: render a
# little extra so the channel never runs dry.
clean = Synth(
type="dsss",
fs=FS,
seed=11 + k, # each emitter its own data
sps=SPC,
data_code=GOLD.tobytes(),
symbol_rate=SYM_RATE,
).steps(int((s1 - s0) * 1.001) + 64)
y = np.asarray(channel(k).execute(clean), np.complex64)[: s1 - s0]
x[s0:s1] += y * np.float32(10.0 ** (level / 20.0))
amp = float(
wfm_awgn_amplitude(ESN0_REF_DB - 10.0 * np.log10(FS / SYM_RATE), 1.0)
)
x += AWGN(seed=seed, amplitude=amp).generate(n).astype(np.complex64)
return x
def truth(k: int, epochs: np.ndarray) -> tuple[np.ndarray, np.ndarray]:
"""Emitter ``k``'s carrier (Hz) and code phase (chips) at the end of
each of ``epochs``. The Doppler is linear in time; the code phase is
the nominal one plus the chips the stretched clock has gained, less the
channel's group delay (:attr:`DopplerChannel.delay_samples`)."""
_, _, ppm, rate, offset, (a, _) = EMITTERS[k]
t = (np.asarray(epochs) + 1 - a) * EPOCH_S
f = RF_HZ * 1e-6 * (ppm + rate * t)
gained = CHIP_RATE * 1e-6 * (ppm * t + rate * t * t / 2.0)
late = channel(k).delay_samples / SPC
nominal = (t * FS - offset) / SPC
return f, (nominal + gained - late) % SF
The pool is built like one receiver, plus n_receivers. carrier_freq_hz lets
each receiver's code loop follow the clock stretch its carrier implies.
cn0_dbhz is the design point, the weakest emitter's C/N0. lost_confirm_s
arms the release that frees a receiver. 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``. ``carrier_freq_hz`` tells each receiver the RF
carrier, so its code loop follows the clock stretch its carrier implies;
``cn0_dbhz`` is the design point, the weakest emitter's C/N0; and
``lost_confirm_s`` arms the release rule that frees a receiver whose
emitter left.
Returns
-------
rec : dict
``state``, ``doppler``, ``phase``, ``locked`` and ``code_locked``,
each ``(epochs, receivers)``, and the pool's release counts.
"""
weakest = min(e[1] for e in EMITTERS)
pool = AsyncDsssPool(
GOLD,
n_receivers=N_RECEIVERS,
chip_rate=CHIP_RATE,
symbol_rate=SYM_RATE,
sample_rate=FS,
cn0_dbhz=ESN0_REF_DB + weakest + 10.0 * np.log10(SYM_RATE),
doppler_uncertainty=CARRIER_UNCERTAINTY,
lost_confirm_s=LOST_CONFIRM_S,
carrier_freq_hz=RF_HZ,
)
shape = (N_EPOCHS, N_RECEIVERS)
rec = {
"state": np.zeros(shape, int),
"doppler": np.zeros(shape),
"phase": np.zeros(shape),
"locked": np.zeros(shape, int),
"code_locked": np.zeros(shape, int),
}
for e in range(N_EPOCHS):
pool.push(x[e * TE : (e + 1) * TE])
for i in range(N_RECEIVERS):
st = pool.status(i)
pool.symbols(i) # drained every push, as a consumer would
rec["state"][e, i] = st.state
rec["doppler"][e, i] = st.doppler_hz
rec["phase"][e, i] = st.chip_phase
rec["locked"][e, i] = st.locked
rec["code_locked"][e, i] = st.code_locked
rec["released"] = {
"lost": pool.released_lost,
"duplicate": pool.released_duplicate,
"ghost": pool.released_ghost,
}
return rec
The checks find the emitter each receiver holds from its carrier and code phase, and assert the claims above from the pool's own reports.
# The level gap past which a one-epoch search is not expected to take the
# weaker signal: at the floor, docs/design/async-dsss-receiver.md §7.2 takes
# it in 90% of streams at -14 dB and 78% at -16 dB, so 15 dB sits between.
MASK_DB = 15.0
SETTLE_S = 0.5 # time allowed from arrival (or unmasking) to a hold
GONE_S = LOST_CONFIRM_S + 0.2 # time allowed from departure to release
def holder_of(rec: dict) -> np.ndarray:
"""Which emitter each receiver holds at each epoch's end: the emitter
whose carrier is within 60 Hz and code phase within 3 chips of the
receiver's, or -1. A receiver still holding an emitter that has left
reads that emitter: it is matched against the emitter's extrapolated
truth for the rest of the stream, so one that is never released shows."""
out = np.full(rec["state"].shape, -1)
ep = np.arange(N_EPOCHS)
tracking = rec["state"] == 2
for k, (_, _, _, _, _, (a, _b)) in enumerate(EMITTERS):
span = ep >= a
f, ph = truth(k, ep)
for i in range(N_RECEIVERS):
d = np.abs(rec["phase"][:, i] - ph)
near = (np.abs(rec["doppler"][:, i] - f) < 60.0) & (
np.minimum(d, SF - d) < 3.0
)
out[span & tracking[:, i] & near, i] = k
return out
def unmasked_from(k: int) -> int:
"""The first epoch emitter ``k`` is on the air with no other emitter on
the air more than MASK_DB above it: from then on a search can tell it
from the others' sidelobes. A stronger one arriving after it is held
does not matter; it is the search that has the limit."""
_, level, _, _, _, (a, b) = EMITTERS[k]
for e in range(a, b):
if not any(
other - level > MASK_DB and oa <= e < ob
for _, other, _, _, _, (oa, ob) in EMITTERS
):
return e
return b
def check(rec: dict) -> dict:
"""The run's claims, asserted from the pool's own reports."""
held = holder_of(rec)
settle = int(SETTLE_S / EPOCH_S)
gone = int(GONE_S / EPOCH_S)
taken_ms = {}
for k, (name, _, _, _, _, (a, b)) in enumerate(EMITTERS):
u = unmasked_from(k)
n_holders = (held == k).sum(axis=1)
# Never two receivers on one emitter.
assert n_holders.max() <= 1, f"{name}: held twice"
# Not held while a stronger emitter masks it (F, until E leaves) ...
assert not n_holders[a:u].any(), f"{name}: held while masked"
# ... then held from SETTLE_S after it can be told from the others
assert n_holders[u:b].any(), f"{name}: never taken"
first = u + int(np.flatnonzero(n_holders[u:b])[0])
assert first - u <= settle, f"{name}: taken late"
taken_ms[name] = (first - a) * EPOCH_S * 1e3
# to its departure, but for the odd epoch a loop slips.
assert n_holders[first:b].mean() > 0.98, f"{name}: dropped"
# Symbol-locked while it is held and on the air, once the lock has
# had SETTLE_S to come up.
on = np.zeros(N_EPOCHS, bool)
on[first + settle : b] = True
locked = rec["locked"][(held == k) & on[:, None]]
assert locked.mean() > 0.99, f"{name}: lock flickers"
# Released after it leaves.
if b < N_EPOCHS:
assert not n_holders[b + gone :].any(), f"{name}: held after"
# Ten emitters passed through six receivers.
assert len(EMITTERS) > N_RECEIVERS
assert rec["released"]["lost"] >= len(EMITTERS) - N_RECEIVERS
# F is masked by E (16 dB above) until E leaves, which the
# hold-while-masked check above holds the run to, and is then held while
# C, 13 dB above it, is still on the air.
names = [e[0] for e in EMITTERS]
f_k = names.index("F")
c_end = EMITTERS[names.index("C")][5][1]
e_end = EMITTERS[names.index("E")][5][1]
assert unmasked_from(f_k) == e_end > EMITTERS[f_k][5][0]
f_first = int(np.flatnonzero((held == f_k).any(axis=1))[0])
assert f_first < c_end, "F: not held beside C"
return {"taken_ms": taken_ms, "held": held}
What you're seeing¶
- Doppler against time (top): each emitter's Doppler while it is on the air, wide and faint. Over it is the carrier of the receiver that holds it, thin, in the same colour. A thin line that runs a little past its band is a receiver still holding an emitter that has just left, until its release fires. F's band runs alone from 2.2 s until C has left.
- Symbol lock (bottom): one row per receiver, coloured by the emitter it
holds where
status().lockedis up. Light grey is a receiver assigned a signal and not yet symbol-locked: the pull-in after the controller seeds it. Hatched is a receiver that held a signal and lost symbol lock: its emitter left. Once code lock drops too, both loops coast: each is reset to its state at the last moment both locks were up and runs on at that rate, its error read but not applied, while the receiver waits outlost_confirm_sbefore the release (design §10). It does not search. A gap is an idle receiver, waiting for the controller's search to seed it.
Reproduce¶
pip install "doppler-dsp[examples]" # matplotlib + scipy, once
python -m doppler.examples.async_dsss_pool_scene out.png
The design, including the controller and the near-far limit, is in the async DSSS design, §5-§9.
Source: async_dsss_pool_scene.py.