ska_pst.testutils.udp_gen

Submodule for UDP generator classes.

This package provides Python classes for generating UDP packet streams. These classes should only be used where there is no upstream UDP data being sent to the PST system under test.

class ska_pst.testutils.udp_gen.GaussianNoiseConfig(*, normal_dist_mean: float = 0.0, normal_dist_stddev: float = 10.0, normal_dist_red_stddev: Optional[float] = None)

Config data class for the GaussianNoise data generator.

normal_dist_mean: float = 0.0

The mean for the normal distribution.

The default value is 0.0.

normal_dist_red_stddev: float | None = None

The standard deviation of a red noise process.

If not set, or set to 0.0, there will be no red noise applied.

normal_dist_stddev: float = 10.0

The standard deviation of normal distribution.

The default value is 10.0.

class ska_pst.testutils.udp_gen.SineWaveConfig(*, sinusoid_freq: float)

Config data class for the Sine wave data generator.

sinusoid_freq: float

The frequency of the sine wave.

class ska_pst.testutils.udp_gen.SquareWaveConfig(*, cal_off_intensity: Optional[float] = None, cal_on_intensity: Optional[float] = None, cal_on_pol_0_intensity: Optional[float] = None, cal_on_pol_1_intensity: Optional[float] = None, cal_on_chan_0_intensity: Optional[float] = None, cal_on_chan_n_intensity: Optional[float] = None, cal_on_pol_0_chan_0_intensity: Optional[float] = None, cal_on_pol_0_chan_n_intensity: Optional[float] = None, cal_on_pol_1_chan_0_intensity: Optional[float] = None, cal_on_pol_1_chan_n_intensity: Optional[float] = None, cal_duty_cycle: Optional[float] = None, calfreq: Optional[float] = None)

Config data class for the SquareWave data generator.

If any one of the above CHAN_0 intensities is specified, then the matching CHAN_N intensity must also be specified. Each CHAN_0,CHAN_N pair defines an intensity gradient that will be applied to all frequency channels. If any intensity (in any polarization or frequency channel) is multiply defined, then the intensity configuration parameters that appear later in the above list will override any configuration set by parameters listed earlier.

cal_duty_cycle: float | None = None

The fraction of period in the on-pulse state.

If not set the default value is 0.5.

cal_off_intensity: float | None = None

The off-pulse intensity for all polarizations and frequency channels.

cal_on_chan_0_intensity: float | None = None

The on-pulse intensity for all polarizations at frequency channel zero.

cal_on_chan_n_intensity: float | None = None

The on-pulse intensity for all polarizations at the number of frequency channels.

cal_on_intensity: float | None = None

The on-pulse intensity for all polarizations and frequency channels.

cal_on_pol_0_chan_0_intensity: float | None = None

The on-pulse intensity for polarization 0 at frequency channel zero.

cal_on_pol_0_chan_n_intensity: float | None = None

The on-pulse intensity for polarization 0 at the number of frequency channels.

cal_on_pol_0_intensity: float | None = None

The on-pulse intensity for polarization 0 and all frequency channels.

cal_on_pol_1_chan_0_intensity: float | None = None

The on-pulse intensity for polarization 1 at frequency channel zero.

cal_on_pol_1_chan_n_intensity: float | None = None

The on-pulse intensity for polarization 1 at the number of frequency channels

cal_on_pol_1_intensity: float | None = None

The on-pulse intensity for polarization 1 and all frequency channels.

calfreq: float | None = None

The frequency of square wave (inverse of period) in Hz

class ska_pst.testutils.udp_gen.UdpDataGenerator(environment: dict, scanlen: int, udpgen_extra_args: Optional[List[str]] = None, logger: Optional[Logger] = None)

Class used to abstract away generating of UDP data.

This class should be used to simulate UDP data is sent to PST when there is no upstream CBF data.

This class is configured with an environment that can then generate a config file for the ska_pst_recv_udpgen command.

Creating an instance of this does not do anything, the call to generate_udp_data is needed. Which will in turn run a background process to generate the data without blocking the client. If the client wants to wait for the data to have been completely sent they should call wait_for_end_of_data.

abort() → None

Abort sending data if its running.

generate_udp_data() → None

Generate UDP data.

This will launch a background thread that will handle an external process that does the work for sending udp data.

is_aborting() → bool

Check if UDP generator is aborting.

is_generating() → bool

Check if UDP generator is generating data.

is_starting() → bool

Check if UDP generator is starting up.

The generator has been requested to start generating but hasn’t started to generate the data. This is different to is_generating as there is background processing happening and is used to avoid calling the generate method twice.

is_stopped() → bool

Check if UDP generator has stopped generating data.

wait_for(predicate: Callable[[...], bool], timeout: Optional[float] = None) → None

Wait until a given condition is met.

This method waits for changes of the state of the data generator and then checks the predicate to see if it should return.

This method can take an optional timeout to avoid blocking indefinitely.

Parameters

• predicate – the predicate to wait against.

• timeout – the amount of time to wait for the condition to be met else this returns.

wait_for_end_of_data() → None

Wait until all the data has been sent.

ska_pst.testutils.udp_gen.create_udp_data_generator(scan_resources: dict, scan_id: int, scanlen: int, channel_block_configuration: dict, data_generator: Optional[str] = None, generator_params: Optional[Union[SineWaveConfig, GaussianNoiseConfig, SquareWaveConfig]] = None, udpgen_extra_args: Optional[List[str]] = None, logger: Optional[Logger] = None, **kwargs: Any) → UdpDataGenerator

Create instance of a UdpDataGenerator.

This is a utility method to help with creating an instance of a UdpDataGenerator object. This method should be preferred over calling the constructor directly as this will validate extra parameters needed for different types of data generators.

This method handles the following types of generators:

• Random - the default, just sends uniform random data.

• GaussianNoise - sends normally distributed data given as mean and standard deviation, along with optional red noise parameters.

• Sine - a sine wave generator with a given frequency.

• SquareWave - sends a square wave signal with configurable duty cycle, calfrequency

Parameters

• scan_resources (dict) – parameters relating to the resources returned from calling ScanConfigGenerator.calculate_udp_gen_resources()

• scan_id (int) – the Scan ID

• scanlen (int) – the scan length in seconds.

• channel_block_configuration (dict) – details about where to send UDP data to. This can be retrieved from the BEAM.MGMT TANGO device’s channelBlockConfiguration after a scan has been configured.

• data_generator (str | None, optional) – the name of the data generate, defaults to None. Valid values are listed above.

• generator_params (SineWaveConfig | GaussianNoiseConfig | SquareWaveConfig | None, optional) – generator specific parameters, defaults to None. These are type safe generator parameters that should be preferred rather than using the kwargs of the function.

• udpgen_extra_args (List[str] | None, optional) – extra parameters that should be sent to the ska_pst_recv_udpgen executable, defaults to None

• logger (logging.Logger | None, optional) – the logger to within the generator, defaults to None

Returns

an instance of the UdpDataGenerator

Return type

UdpDataGenerator