dada_utils

Tools for turning MeerKAT SPEAD packet captures into DADA voltage files and total-intensity SIGPROC filterbanks:

tool	what it does
distribute_new	SPEAD capture files (plain or zstd-seekable) → TAFTP DADA files, reader-parallel over MPI
run_meerkat.py	driver: submits one distribute_new Slurm job per bridge on the meerkat storage servers
toy_beamformer	TAFTP DADA voltages → total-intensity filterbank (boresight, 32-bit float or 8-bit)
merge_filterbanks	per-bridge filterbanks → one wide-band filterbank, zeros where a bridge is missing

Build

module load gcc/12.3.0 openmpi/4.1.5    # or OpenMPI/4.1.5-GCC-12.3.0
make                                     # builds all three binaries

distribute_new needs MPI, libconfig and libzstd; xxHash is vendored (include/xxhash.h). toy_beamformer and merge_filterbanks are plain C with no dependencies (make toy_beamformer, make merge_filterbanks to build them alone, no MPI needed).

distribute_new: SPEAD captures → DADA

Reader-parallel rewrite of the original distribute: each DADA file is assembled completely in memory and written with large sequential writes only (the original wrote sparse files at random offsets, which fragments the filesystem).

                 shared in-memory window (window_files DADA buffers)
 reader rank 1 ──┐   ┌───────────┬───────────┬───────────┬───────────┐
 reader rank 2 ──┼──▶│ round w   │ round w+1 │ round w+2 │ round w+3 │
 reader rank 3 ──┘   └─────┬─────┴───────────┴───────────┴───────────┘
   (scatter, race-free)    │ complete, or all readers past its end
                           ▼
                 writer rank 0: gap-fill → sequential write → rename

• All ranks run on one node and share the window via MPI_Win_allocate_shared; rank 0 writes, ranks 1..N-1 read the capture files striped over them in time order (with -n 1 the single rank does both).
• Every SPEAD payload has a unique preallocated position in the window, so the data path needs no locks. Packets beyond the window go to a per-reader overflow pool (-P MiB); a full pool applies backpressure.
• A buffer is flushed when complete or when every reader's watermark has passed its end; missing payloads are gap-filled (below), and the file is written to <file>.part and atomically renamed.

Output is byte-identical to the original code wherever the original is deterministic, with one intentional fix: FILE_NUMBER counts files (the original always wrote 0). Dropped relative to the original: the ncurses UI and gain calibration.

Usage

mpirun -n <1 + readers> ./distribute_new -c bridge_XX.conf

8–12 readers (≈ one per rx stream) is a good starting point; more readers do not increase memory (they share the window). Shared memory is window_files × timestamp_per_dada_file × ts_size, e.g. 4 × 8192 × 3.9 MB ≈ 128 GB for a full 62-antenna 4k-mode bridge.

Useful flags (-h for all): -W/-P override window files/pool MiB, -q disables the progress bar, -C <file> checkpoint/resume, -d dry run, -r <file> inspects a single capture file, -g <policy> gap fill, -M writes a .mask file per DADA file, -F <file> reads a capture-file list instead of scanning a directory.

Config file

libconfig format; examples/bridge_02.conf is a real production config. Fields:

sourcedata = {
    source_dir = "..."            # directory of capture files (or use -F)
    tag = "NGC6388"               # capture filename tag: <tag>_<bridge>_<rx>_<cnt>
    bridge = 2
    start_timestamp = 131062160162816.0   # sample-clock window to extract
    end_timestamp   = 131264174620672.0
}
destdata = {
    dest_dir = "..."              # output directory for .dada files
    checkpoint_file = "..."       # used with -C
    timestamp_per_dada_file = 8192
    dada_files = 12               # number of output files
    tag = "NGC6388"
    date = "2025-11-21-14:35:27"  # goes into UTC_START
}
observation = {                   # copied into the DADA headers
    synctime = 1763659172.0
    source = "J1909-3744"
    ra = "19:09:47.43"  dec = "-37:44:14.5"
    obs_id = "20251121-0005"
    bandwidth = 856000000.0  channel_mode = 4096
    central_frequency = 1283895507.8125
    receiver = "L-band"  nchan_per_bridge = 64  n_feng = 62
}
distribution = {                  # optional tuning
    window_files = 4              # DADA buffers held in RAM
    pool_max_mb = 1024            # per-reader overflow pool
    packets_per_read = 8192       # read chunk (packets)
    write_chunk_mb = 64           # write chunk
}

File lists and disk-striped captures (-F)

-F <file> takes a newline-separated list of capture file paths instead of scanning a single source directory (empty lines, # comments and paths not matching <tag>_<bridge> are ignored, so a raw per-host find inventory can be passed as is). This matters on the meerkat storage servers, where each capture stream is striped over the two local disk arrays (/data/inner and /data/outer) as time-overlapping counter files — a single-directory copy sees only half the packets. Readers merge their files by timestamp, so overlapping counter files are handled correctly.

Gap filling (-g)

Payload positions that never received a packet are filled with data statistically indistinguishable from the surroundings (zeros would bias power levels, so zero fill is deliberately not offered). All policies are deterministic, so re-runs produce identical files. The donor is always the same (fengine, channel, offset) stream at another timestamp of the same file.

policy	donor timestamp	notes
random (default)	randomized expanding-distance search, replicating the original code's statistics	duplicates scattered, not adjacent
nearest	nearest captured timestamp	cheapest; a long dropout becomes repeated identical blocks — a spectral comb at ~816 Hz
pseudo_noise	nearest captured timestamp, 256 time samples permuted	exact per-payload power, no repeated blocks

-M dumps each DADA file's payload-presence bitmap as <file>.dada.mask so downstream code can tell real payloads from filled ones.

Running on the meerkat storage servers (run_meerkat.py)

Every bridge lives wholly on one meerkat host, so no cross-node MPI is needed: run_meerkat.py reads one HOSTNAME.txt file inventory per host, maps each requested bridge to the host that owns its files, and submits one single-node MPI job per bridge with sbatch -p meerkat -w HOSTNAME (dry run by default, --submit to launch; --cores sets ranks per job; Slurm's memory accounting throttles concurrent jobs per host):

./run_meerkat.py --lists '/aphid/temp/compact/meerkat*.txt' \
    --template examples/bridge_02.conf --outdir /aphid/temp/compact/pointing1 \
    --bridges 02,03,10-13 --cores 13 --submit

It writes <outdir>/brXX/{files.txt,bridge_XX.conf,bridge_XX.sbatch} per bridge and puts the DADA files in <outdir>/dada_brXX/. tests/production_example.sbatch shows a hand-written single-bridge job instead.

toy_beamformer: DADA voltages → total-intensity filterbank

./toy_beamformer [-b 8] [-C] -o out.fil dada_brXX/*.dada

Sums the intensity of all polarisations of all antennas per (time, channel) cell — the boresight beam, no delays applied — and writes a SIGPROC filterbank (TF order, channels in descending frequency, 32-bit float by default). All metadata (band, tsamp, MJD, source, coordinates) is taken from the DADA headers; input files are sorted by OBS_OFFSET and must be contiguous. -C sums antenna voltages coherently per polarisation before squaring instead.

-b 8 writes 8-bit unsigned samples: a first pass samples every 16th block for per-channel mean/std, then encodes (I - mean)/std × 8 + 64 clamped to 0–255 (SIGPROC stores no scale factors, so the flattened bandpass is baked in; ~8 σ of range below the mean, ~24 σ above). One 118 s bridge (12 × 33 GB DADA) → 1.6 GB at 8 bit, 6.3 GB at 32 bit.

examples/toybf8_generic.sbatch runs one bridge per Slurm job (sbatch -J toybf8-br02 toybf8_generic.sbatch 02).

merge_filterbanks: per-bridge filterbanks → full band

./merge_filterbanks -o all.fil [-n 4096] [-F <fch1 MHz>] br*.fil

Slots each per-bridge filterbank into one wide-band file by its fch1, writing zeros for channels no input covers (absent/failed bridges). Inputs must agree in nbits, foff, tsamp, tstart and sample count. The output band top defaults to the highest input fch1, so pass -F if the highest-frequency bridge itself is missing (1711.791015625 MHz is the L-band top for 4096 channels).

Note that the bridge number in the filename is not the bridge's position in frequency (the mapping is scrambled); merging goes by each file's fch1 header, which is derived from the data itself.

examples/merge_all.sbatch merges a whole pointing, guarding against incomplete inputs, typically submitted with --dependency=afterany:<beamformer job ids>.

Tests

cd tests && sbatch validation.sbatch     # or: bash run_validation.sh

run_validation.sh generates synthetic SPEAD data (generate_fake_spead.py: correct packet format, deterministic payloads, optional gaps/duplicates/zstd-seekable compression) and checks every output payload against the generator's manifest (verify_dada.py, an oracle independent of the code under test). If OLD_REPO points at a checkout of the original meerkat_data_distribution code, the original is also built and the outputs are additionally required to be byte-identical to it.

test	dataset	check
T1	clean	single rank matches reference
T2	clean	5 ranks, byte-identical
T3	clean	window=1 file + 1 MiB pool (stress)
T4	zstd	compressed input
T5	duplicates	cross-stream retransmits deduplicated, verified against the oracle
T6	2% drops	every present payload correct, drops gap-filled
T7	short file	non-divisible time range
T8	clean	resume from checkpoint round 2
T9	clean	run-to-run determinism
T10	striped	time-overlapping counter files via -F, junk list lines ignored

License

GPL v2.0, as the original meerkat_data_distribution code this derives from.