ska_pst.testutils.scan_config

Submodule for Scan configuration and validation of channel block config after configure scan.

class ska_pst.testutils.scan_config.ChannelBlockValidator(encoded_json: str, logger: Optional[Logger] = None)[source]

Class that can be used to read and validate channel block configuration.

is_empty() → bool[source]: Check the the channel block configuration is empty.

unpack(encoded_json: str) → Dict[Any, Any][source]: Unpack the json into a dict.

validate() → None[source]: Test if channel block configuration is valid.

class ska_pst.testutils.scan_config.FlowThroughConfig(rescale_timescale: float = 0.0, periodic_update: bool = False, algorithm: str = 'MedianMAD', channels_out: Optional[Tuple[int, int]] = None, polarisations: Optional[str] = None, num_bits_out: Optional[int] = None, requantisation_scale: float = 1.0)[source]

Data class representing the Flow Through configuration used in testing.

algorithm: str = 'MedianMAD': The algorithm used to determine the scales and offsets when rescaling complex voltage data.

channels_out: Optional[Tuple[int, int]] = None

The selected channel as an exclusive range.

To select all the channels use (-1, -1) which the scan config generator will use as a sentinel value for all channels

num_bits_out: int | None = None: The number of bits to use when performing digitisation.

periodic_update: bool = False: An indicator for whether to recalculate the rescale statistics periodically.

polarisations: str | None = None

The polarisations to select in ChanPolSelect.

Default is we test a random selection of “A”, “B” or “Both”

requantisation_scale: float = 1.0: The scale to apply to values before performing digitisation.

rescale_timescale: float = 0.0

The timescale needed to calculate rescale stats, in seconds.

This value is how long in time to sample data before calculating the rescale statistics. If periodic_update is true, then this is also the period of how often the rescale stats are recalculated.

If this value is set to 0.0, then PST will use the smallest chunk of data available to it to perform the statistics calculation.

Default: 0.0

property use_robust_statistics: bool

Get the use robust statistics property.

If algorithm property is MedianMAD then this will return True.

Returns: the use robust statistics property.
Return type: bool

class ska_pst.testutils.scan_config.RfConfig(*, nchan: 'int', centre_frequency_mhz: 'float', bandwidth_mhz: 'float', start_chan: 'int', cbf_pst_config: 'CbfPstConfig')[source]

as_scan_config() → dict[source]

property bandwidth_hz: float: Get frequency in Hz.

bandwidth_mhz: float: The bandwidth of the CBF data, in MHz.

cbf_pst_config: CbfPstConfig: The CBF/PST configuration, based on the frequency band, that the RF configuration is for.

property centre_frequency_hz: float: Get frequency in Hz.

centre_frequency_mhz: float: The centre frequency of the CBF data, in MHz.

property data_rate: float: Get the data rate for the given configuration, in bytes/sec.

property end_chan: int

Get the end channel that this config covers.

This is equivalent to start_chan + nchan - 1.

property end_frequency_hz: float: Get the end frequency that this config is for, in Hz.

property end_frequency_mhz: float: Get the end frequency that this config is for, in MHz.

nchan: int: The number of channels of the data.

start_chan: int: The channel that the data from the CBF is starting from.

property start_frequency_hz: float: Get the starting frequency of that this config is for, in Hz.

property start_frequency_mhz: float: Get the starting frequency of that this config is for, in MHz.

class ska_pst.testutils.scan_config.ScanConfigGenerator(beam_id: int, telescope: str, facility: TelescopeFacilityEnum, frequency_band: Optional[str] = None, max_scan_length: float = 30.0, pst_config_version: str = ska_telmodel.pst.version.PST_CONFIG_VER3_0, processing_mode: ska_control_model.PstProcessingMode = ska_control_model.PstProcessingMode.VOLTAGE_RECORDER, max_data_generated_size: Optional[int] = None, logger: Optional[Logger] = None, **kwargs: Any)[source]

Utility class to generate Scan configuration.

add_processing_mode_config_value(key: str, value: Any) → None[source]: Set a configuration specific for the given processing mode.

calculate_udp_gen_resources(overrides: dict = {}) → dict[source]

Calculate the resources for PST for current configuration.

This is used to send requests to the UDP generator and it needs to know from the configuration the bandwidth, nbits, etc to provide a valid configuration file to be used by the UDP generator.

property cbf_pst_config: CbfPstConfig

Get the CBF/PST config for the telescope and frequency band.

This is a utility property to avoid duplication of code and the need to get the configuration all the time just to get cbf_pst_config needed for generating a scan configuration.

Returns: the CBF/PST config for the telescope and frequency band.
Return type: CbfPstConfig

property curr_config: dict: Get the current scan configuration.

property curr_config_id: str

property curr_config_json: str: Get the current scan configuration as JSON string.

property facility: TelescopeFacilityEnum: Get the current facility.

property frequency_band: str: Get current frequency band.

property frequency_band_config: FrequencyBandConfig

Get the current frequency band configuration for the current telescope and frequency band.

This is a utility property to avoid duplication of code and the need to get the configuration all the time just to get frequency band config needed for generating a scan configuration.

Returns: the current frequency band configuration for the current telescope and frequency band.
Return type: FrequencyBandConfig

generate(eb_id: Optional[str] = None, overrides: dict = {}, max_data_rate: Optional[float] = None, **kwargs: Any) → dict[source]

Generate a configuration.

This is the public method that should be used by test fixtures to generate the configuration. The overrides parameter can be used to generate an invalid configuration.

generate_json() → str[source]

Generate a configuration and return it as a json string.

This is equivalent of doing:

scan_config = scan_config_generator.generate()
scan_config_str = json.dumps(scan_config)

override_config(key: str, value: Any) → None[source]: Override a specific config value.

property processing_mode: ska_control_model.PstProcessingMode: Get current configured processing mode.

property replay_config: dict | None

reset_replay_config() → None[source]

property telescope: str

Get the current telescope.

There is no setter for this as it depends on the facility

Returns: the current telescope.
Return type: str

property telescope_config: TelescopeConfig

Get the configuration of the current telescope.

This is a utility property to avoid duplication of code and the need to get the configuration all the time just to get telescope config needed for generating a scan configuration.

Returns: the configuration of the current telescope.
Return type: TelescopeConfig

class ska_pst.testutils.scan_config.ScanIdFactory[source]

Utility class to generate random scan ids that are unique upon each call.

Instances of this class should be at a session level scope rather than being created for every scan. This will allow it to keep track of previously generated scan ids.

generate_scan_id() → int[source]: Generate next scan id.

ska_pst.testutils.scan_config.create_default_scan_config_generator(beam_id: int = 1, frequency_band: str = 'low', max_scan_length: float = 30.0, max_data_generated_size: Optional[int] = None, logger: Optional[Logger] = None, **kwargs: Any) → ScanConfigGenerator[source]

Create instance of a ScanConfigGenerator using default values.

Use this method if wanting to test using a default configuration. The implementation of this uses the CSP-PST v2.5 JSON <https://developer.skao.int/projects/ska-telmodel/en/latest/schemas/pst/ska-pst-configure-2.5.html>_ for the default values.

Parameters

beam_id (int, optional) – the PST Beam ID to use, defaults to 1
frequency_band (str, optional) – the frequency band, defaults to “low”.
max_scan_length (float, optional) – the maximum length, in seconds, that the scan will be for, defaults to 30.0.
max_data_generated_size (int | None, optional) – the maximum amount of data, in bytes, that the scan should generate, defaults to None. This should be set when performing tests in an automated environment that has limited storage capacity.
logger (logging.Logger | None, optional) – the logger to use when logging, defaults to None. The ScanConfigGenerator only uses the logger for debug messages.

Returns

an instance of the scan config generator.

Return type

ScanConfigGenerator

ska_pst.testutils.scan_config.create_fixed_scan_config_generator(scan_config: dict) → ScanConfigGenerator[source]: Create instance of ScanConfigGenerator that replays provided scan configuration.

ska_pst.testutils.scan_config.generate_eb_id() → str[source]: Generate a new execution block id.

ska_pst.testutils.scan_config.generate_rf_config(cbf_pst_config: CbfPstConfig, max_data_rate: Optional[float] = None, min_nchan: Optional[int] = None, **kwargs: Any) → RfConfig[source]

Generate a random RF configuration that is valid for a test.

This generates a random RF configuration that will ensure the number of channels, frequency and bandwidth are consistent with the TSAMP value for the given frequency band.

Parameters

cbf_pst_config (CbfPstConfig) – the CBF/PST configuration for a given frequency band
max_data_rate (float | None, optional) – the maximum number of bytes per second to generated config for, defaults to None.
min_nchan (int | None, optional) – the minimum number of input PST channels to use, defaults to None.

Returns

a random RF configuration

Return type

RfConfig

ska_pst.testutils.scan_config.generate_rfi_frequency_mask_config(rf_config: RfConfig, min_rfi_masks: int = 0, max_rfi_masks: int = 4, percent_masked: float = 5.0, **kwargs: Any) → dict[source]

Generate a random RFI Mask configuration.

Generates a RFI frequency mask that would cover percent_masked with a maximum of max_rfi_masks contiguous masks.

Parameters

rf_config (RfConfig) – the current RF configuration to generate the mask range from.
min_rfi_masks (int, optional) – the minimum number of masks ranges to use, defaults to 0
max_rfi_masks (int, optional) – the maximum number of masks ranges to use, defaults to 4
percent_masked (float, optional) – the percentage of the RF band to mask as RFI, defaults to 5.0

Returns

a dictionary with values of num_rfi_frequency_masks and an array of rfi_frequency_masks.

Return type

dict