katgpucbf.dsim.signal module

Synthesis of simulated signals.

katgpucbf.dsim.signal.CHUNK_SIZE = 1048576: Dask chunk size for sampling signals (must be a multiple of 8)

class katgpucbf.dsim.signal.CW(amplitude: float, frequency: float)[source]

Bases: Periodic

Continuous wave.

To make the resulting signal periodic, the frequency is adjusted during sampling so that the sampled result can be looped.

class katgpucbf.dsim.signal.Comb(amplitude: float, frequency: float)[source]

Bases: Periodic

Signal with periodic impulses.

To make the resulting signal periodic, the frequency is adjusted during sampling so that the sampled result can be looped.

class katgpucbf.dsim.signal.CombinedSignal(a: Signal, b: Signal, combine: Callable[[Array, Array], Array], op_name: str)[source]

Bases: Signal

Signal built by combining two other signals.

Parameters:

a – Input signals
b – Input signals
combine (collections.abc.Callable[[dask.array.core.Array, dask.array.core.Array], dask.array.core.Array]) – Operator to combine two arrays
op_name (str) – Symbol for the operator

a: Signal

b: Signal

combine: Callable[[Array, Array], Array]

op_name: str

sample(n: int, sample_rate: float) → Array[source]

Sample the signal at regular intervals.

The returned values should be scaled to the range (-1, 1).

Note

Calling this method with two different values of n may yield results that are not consistent with each other.

Parameters:

n – Number of samples to generate
sample_rate – Frequency of samples (Hz)

Returns:

Dask array of samples, float32. The chunk size must be CHUNK_SIZE.

Return type:

samples

class katgpucbf.dsim.signal.Constant(value: float)[source]

Bases: Signal

Fixed value.

sample(n: int, sample_rate: float) → Array[source]

Sample the signal at regular intervals.

The returned values should be scaled to the range (-1, 1).

Note

Calling this method with two different values of n may yield results that are not consistent with each other.

Parameters:

n – Number of samples to generate
sample_rate – Frequency of samples (Hz)

Returns:

Dask array of samples, float32. The chunk size must be CHUNK_SIZE.

Return type:

samples

value: float

class katgpucbf.dsim.signal.Delay(signal: Signal, delay: int)[source]

Bases: Signal

Delay another signal by an integer number of samples.

Parameters:

signal (katgpucbf.dsim.signal.Signal) – Underlying signal to delay
delay (int) – Number of samples to delay the signal (may be negative)

delay: int

sample(n: int, sample_rate: float) → Array[source]

Sample the signal at regular intervals.

The returned values should be scaled to the range (-1, 1).

Note

Calling this method with two different values of n may yield results that are not consistent with each other.

Parameters:

n – Number of samples to generate
sample_rate – Frequency of samples (Hz)

Returns:

Dask array of samples, float32. The chunk size must be CHUNK_SIZE.

Return type:

samples

signal: Signal

class katgpucbf.dsim.signal.MultiCW(m: int, amplitude0: float, amplitude_step: float, frequency0: float, frequency_step: float)[source]

Bases: Signal

Sum of multiple continuous waves.

The amplitudes and frequencies are both arithmetic progressions. The frequencies are adjusted during sampling so that the sampled result can be looped.

amplitude0: float

amplitude_step: float

frequency0: float

frequency_step: float

m: int

sample(n: int, sample_rate: float) → Array[source]

Sample the signal at regular intervals.

The returned values should be scaled to the range (-1, 1).

Note

Calling this method with two different values of n may yield results that are not consistent with each other.

Parameters:

n – Number of samples to generate
sample_rate – Frequency of samples (Hz)

Returns:

Dask array of samples, float32. The chunk size must be CHUNK_SIZE.

Return type:

samples

class katgpucbf.dsim.signal.Nodither(signal: Signal)[source]

Bases: Signal

Mark a signal expression as not needing dither.

Parameters:: signal (katgpucbf.dsim.signal.Signal) – Underlying signal

sample(n: int, sample_rate: float) → Array[source]

Sample the signal at regular intervals.

The returned values should be scaled to the range (-1, 1).

Note

Calling this method with two different values of n may yield results that are not consistent with each other.

Parameters:

n – Number of samples to generate
sample_rate – Frequency of samples (Hz)

Returns:

Dask array of samples, float32. The chunk size must be CHUNK_SIZE.

Return type:

samples

signal: Signal

property terminal: bool: Prevent this signal from being used in expressions.

class katgpucbf.dsim.signal.Periodic(amplitude: float, frequency: float)[source]

Bases: Signal

Base class for periodic signals.

The frequency is adjusted during sampling so that the sampled result can be looped.

amplitude: float

frequency: float

sample(n: int, sample_rate: float) → Array[source]

Sample the signal at regular intervals.

The returned values should be scaled to the range (-1, 1).

Note

Calling this method with two different values of n may yield results that are not consistent with each other.

Parameters:

n – Number of samples to generate
sample_rate – Frequency of samples (Hz)

Returns:

Dask array of samples, float32. The chunk size must be CHUNK_SIZE.

Return type:

samples

class katgpucbf.dsim.signal.Random(entropy: int | None = None)[source]

Bases: Signal

Base class for randomly-generated signals.

This base class is only suitable when the samples at different times are independent. The derived class must implement _sample_chunk().

entropy: int: entropy used to populate a np.random.SeedSequence

sample(n: int, sample_rate: float) → Array[source]

Sample the signal at regular intervals.

The returned values should be scaled to the range (-1, 1).

Note

Calling this method with two different values of n may yield results that are not consistent with each other.

Parameters:

n – Number of samples to generate
sample_rate – Frequency of samples (Hz)

Returns:

Dask array of samples, float32. The chunk size must be CHUNK_SIZE.

Return type:

samples

class katgpucbf.dsim.signal.Signal[source]

Bases: ABC

Abstract base class for signals.

An instance is simply a real-valued function of time, for a single polarisation.

abstractmethod sample(n: int, sample_rate: float) → Array[source]

Sample the signal at regular intervals.

The returned values should be scaled to the range (-1, 1).

Note

Calling this method with two different values of n may yield results that are not consistent with each other.

Parameters:

n – Number of samples to generate
sample_rate – Frequency of samples (Hz)

Returns:

Dask array of samples, float32. The chunk size must be CHUNK_SIZE.

Return type:

samples

property terminal: bool

Indicate whether the signal is terminal.

Terminal signals cannot be combined into larger expressions, because they contain information about how to handle their postprocessing.

class katgpucbf.dsim.signal.SignalService(arrays: Sequence[DataArray], sample_bits: int, dither_seed: int | None = None)[source]

Bases: object

Compute signals in a separate process.

The provided arrays must be backed by SharedArray, and each must have an xarray attribute called "shared_array" which holds the backing SharedArray.

Parameters:

arrays – All the arrays that might be passed to sample().
sample_bits – Number of bits per sample for all queries.
dither_seed – Seed used to generate a fixed dither.

async sample(signals: Sequence[Signal], timestamp: int, period: int | None, sample_rate: float, out: DataArray, out_saturated: DataArray | None = None, saturation_group: int = 1) → None[source]

Perform signal sampling in the remote process.

out and out_saturated must each be one of the arrays passed to the constructor. Only the first n samples will be populated (and this will be taken as the period).

async stop() → None[source]: Shut down the process.

exception katgpucbf.dsim.signal.TerminalError(signal: Signal)[source]

Bases: TypeError

Indicate that a terminal signal has been used in an expression.

class katgpucbf.dsim.signal.WGN(std: float, entropy: int | None = None)[source]

Bases: Random

White Gaussian Noise signal.

Each sample in time is an independent Gaussian random variable with zero mean and a given standard deviation.

In practice, the signal has a period equal to the value of n given to sample(), which could lead to undesirable correlations.

Parameters:

std (float) – Standard deviation of the samples
entropy (int) – If provided, used to seed the random number generator

std: float = 1.0: standard deviation

katgpucbf.dsim.signal.format_signals(signals: Sequence[Signal]) → str[source]

Inverse of parse_signals().

Currently object identity is not preserved, so if a simple signal is re-used multiple times (e.g., shared across output signals), it will be repeated in the output. This is subject to change.

katgpucbf.dsim.signal.make_dither(n_pols: int, n: int, entropy: int | None = None) → DataArray[source]

Create a set of dither signals to use with quantise().

The returned array has pol and data axes, and is backed by a Dask array.

The implementation currently uses a uniform distribution, but that is subject to change.

katgpucbf.dsim.signal.packbits(data: Array, bits: int) → Array[source]

Pack integers into bytes.

The least-significant bits bits of each integer in data is collected together in big-endian order, and returned as a sequence of bytes. The total number of bits must form a whole number of bytes.

If the chunks in data are not be aligned on byte boundaries then a slower path is used.

katgpucbf.dsim.signal.parse_signals(prog: str) → list[Signal][source]

Generate a set of signals from a domain-specific language.

See Signal specification for a description of the language.

katgpucbf.dsim.signal.quantise(data: Array, bits: int, dither: Array) → Array[source]

Convert floating-point data to fixed-point.

Parameters:

data – Array of values, nominally in the range -1 to 1 (values outside the range are clamped).
bits – Total number of bits per output sample (including the sign bit). The input values are scaled by \(2^{bits-1} - 1\).
dither – Values to add to the data after scaling.

katgpucbf.dsim.signal.sample(signals: Sequence[Signal], timestamp: int, period: int | None, sample_rate: float, sample_bits: int, out: DataArray, out_saturated: DataArray | None = None, saturation_group: int = 1, *, dither: bool | DataArray = True, dither_seed: int | None = None) → None[source]

Sample, quantise and pack a set of signals.

The number of samples to generate is determined from the output array.

Parameters:

signals – Signals to sample, one per polarisation
timestamp – Timestamp for the first element to return. The signal is rotated by this amount.
period – Number of samples after which to repeat. This must divide into the total number of samples to generate. If not specified, uses the total number of samples.
sample_rate – Passed to Signal.sample()
sample_bits – Passed to quantise() and packbits()
out – Output array, with a dimension called pol (which must match the number of signals). The other dimensions are flattened.
out_saturated – Output array, with the same shape as out, into which saturation counts are written.
saturation_group – Samples are taken in contiguous groups of this size and each element of out_saturated is a saturation count for one group. This must divide into the total number of samples.
dither – If true (default), add uniform random values in the range [-0.5, 0.5) after scaling to reduce artefacts. It may also be a xr.DataArray with axes called pol (which must match the number of signals) and data (which must have length at least equal to period).

katgpucbf.dsim.signal.saturation_counts(data: Array, saturation_value) → Array[source]

Return an array indicating counts of saturated elements of data.

The count is taken along each row of data.

Elements are considered saturated if they exceed saturation_value in absolute value.