draco.util.testing

draco test utils.

Functions

mock_freq_data(freq, ntime, delaycut[, ...])

Make mock delay data with a constant delay spectrum up to a specified cut.

Classes

`DummyTask`()	Produce an empty data stream for testing.
`RandomFreqData`()	Generate a random sidereal stream with structure in delay.

class draco.util.testing.DummyTask[source]

Bases: SingleTask

Produce an empty data stream for testing.

total_len

Length of output data stream. Default: 1.

Type:: int

tag

What to use as a tag for the produced data.

Type:: str

Initialize pipeline task.

May be overridden with no arguments. Will be called after any config.Property attributes are set and after ‘input’ and ‘requires’ keys are set up.

process()[source]

Produce an empty stream and pass on.

Returns:: cont – Empty data stream.
Return type:: subclass of memh5.BasicCont

class draco.util.testing.RandomFreqData[source]

Bases: RandomTask

Generate a random sidereal stream with structure in delay.

num_realisation: How many to generate in subsequent process calls.

num_correlated: The number of correlated realisations output per cycle.

num_ra: The number of RA samples in the output.

num_base: The number of baselines in the output.

freq_start, freq_end: The start and end frequencies.

num_freq: The number of frequency channels.

delay_cut: The maximum delay in the data in us.

noise: The RMS noise level.

Generate and set a new random seed for this task.

This will generate a new random seed on rank 0, broadcast it to all ranks, and set the _local_seed attribute accordingly.

Warning

Generating the seed is a collective operation if the seed is not set, and so all ranks must participate in the call to __init__.

next() → SiderealStream | list[SiderealStream][source]

Generate correlated sidereal streams.

Returns:: Either a single stream (if num_correlated=None), or a list of correlated streams.
Return type:: streams

draco.util.testing.mock_freq_data(freq: ndarray, ntime: int, delaycut: float, ndata: int | None = None, noise: float = 0.0, bad_freq: ndarray | None = None, rng: Generator | None = None) → tuple[ndarray, ndarray][source]

Make mock delay data with a constant delay spectrum up to a specified cut.

Parameters:

freq – Frequencies of each channel (in MHz).
ntime – Number of independent time samples.
delaycut – Cutoff in us.
ndata – Number of correlated data sets. If not set (i.e. None) then do no add a dataset axis.
noise – RMS noise level in the data.
bad_freq – A list of bad frequencies to mask out.
rng – The random number generator to use.

Returns:

data – The 2D/3D data array [dataset, freq, time]. If ndata is None then the dataset axis is dropped.
weights – The 2D weights data [freq, time].