Skip to content

A Message Over DSSS, Received and Deframed

A message is split into 48-bit chunks, one chunk per frame, and each frame is spread into a DSSS burst. DsssBurstReceiver finds every burst; the same FrameDesc the transmitter was given then checks each frame and returns its chunk. The message comes back from the capture alone — there is no hand-slicing of bit positions and no second copy of the frame layout.

What it shows

The run is exactly the frames. The message's bits are the segment's data, the description [sync | data:48 | crc16] says how each frame is made, and the short last chunk is padded from fill. No num_samples is given: the run length is the frames', derived.

The receiver is built from the description. The sync word and the frame length are the description's, so the receiver finds all six bursts at their exact samples, and FrameDesc.check passes every frame's CRC-16.

The message comes back byte for byte. FrameDesc.deframe hands back each corrected frame; the payload is a slice found by name. Concatenated, the chunks reproduce the message, and the spare bits of the last frame are the fill.

The check is a real detector: one flipped payload bit fails it.

The script asserts all of this and exits non-zero if any claim breaks. Its C twin is native/examples/data_source_link_demo.c.

The code

import math

import numpy as np

from doppler.cvt import bytes_to_bin
from doppler.dsss import DsssBurstReceiver
from doppler.wfm import (
    STAGE_CRC16,
    Composer,
    FrameDesc,
    Segment,
    field_bits,
)

CHUNK, REPS, SPC, CHIP_RATE = 48, 5, 2, 1.0e6
acq_code, data_code = field_bits("pn:255:8:1"), field_bits("pn:31:5:3")

message = b"a message, one chunk per DSSS burst"
octets = np.frombuffer(message, np.uint8)
bits = np.zeros(8 * octets.size, np.uint8)
bytes_to_bin(octets, bits, 0)  # MSB first, as a file is read
fill = np.array([0, 1], np.uint8)  # pads the short last frame
frames = math.ceil(bits.size / CHUNK)  # one burst per chunk

# The frame, once: a sync word, one chunk of the source, a CRC-16 over it.
d = FrameDesc()
d.add_field("sync", field_bits("0000011001010"))  # Barker-13, inverted
d.add_data("payload", CHUNK)
d.add_derived("crc", 16)
d.add_stage_over(STAGE_CRC16, "payload", "crc")
d.build()
NB = d.nbits

receiver = DsssBurstReceiver(
    acq_code=acq_code,
    data_code=data_code,
    frame=d,  # the sync word and the frame length are the description's
    reps=REPS,
    spc=SPC,
    chip_rate=CHIP_RATE,
    cn0_dbhz=60.0,
    doppler_uncertainty=0.0,
    pfa=1e-3,
    pd=0.9,
    carrier_hz=0.0,
    max_rate=0.0,
    est_segments=10,
)
seg = Segment(
    type="dsss",
    fs=CHIP_RATE * SPC,
    snr=12.0,
    snr_mode="esno",
    seed=1,
    sps=SPC,
    acq_code=acq_code,
    acq_reps=REPS,
    data_code=data_code,
    frame=d,
    data=bits,  # the message: its bits, one chunk per burst
    fill=fill,
    off_samples=receiver.min_gap,  # dead air after the train
)
x = np.asarray(Composer([seg]).compose())

out = np.asarray(receiver.push(x))
events = receiver.events()

off = d.field_off(d.field_index("payload"))
chunks = []
for k in range(frames):
    frame = out[k * NB : (k + 1) * NB]
    verdict = d.check(frame)
    assert verdict.passed == verdict.checked == 1, f"frame {k}: CRC failed"
    chunks.append(np.asarray(d.deframe(frame))[off : off + CHUNK])
got = np.concatenate(chunks)