ska_pst.common

This package provides common classes used by the different submodules of ska-pst.

class ska_pst.common.CbfPstConfig(*, udp_format: str, tsamp: float, udp_nchan: int, udp_nsamp: int, wt_nsamp: int, nbit: int, oversampling_ratio: ~typing.Union[~typing.Tuple[int, int], ~typing.List[int]], max_nchan: int, rf_freq_minimum_mhz: float, rf_freq_maximum_mhz: float, rf_bw_mhz: float, chan_separation_khz: float, channel_0_start_freq_mhz: float, ring_buffer_config: ~ska_pst.common.cbf_pst_config.RingBufferConfig, channelisation_stages: ~typing.List[~ska_pst.common.cbf_pst_config.ChannelisationStage] = <factory>, ndim: int = 2, npol: int = 2, wt_nbit: int = 16)[source]

A data class used to represent the CBF / PST frequency band constants.

This class also includes utility methods that can be used to calculate values like nchan, and bytes_per_sec given other values.

apply_nchan_limits(nchan: int, min_nchan: Optional[int] = None) → int[source]

Apply limits to the nchan value.

This method ensures that the following conditions:

self.udp_nchan <= nchan <= self.max_nchan

nchan % self.udp_nchan == 0

If min_nchan is set, then the lower bound of nchan is min_nchan but the output value is still a multiple of self.udp_nchan

Parameters

nchan (int) – the proposed NCHAN value that needs to have limits applied.
min_nchan (int | None, optional) – the requested minimum number of channels, defaults to None. If not set then self.udp_nchan is used.

Returns

a value that meets the channel limits conditions.

Return type

int

bandwidth_mhz(nchan: Optional[int] = None) → float[source]

Get the bandwidth given the number of channels, in MHz, rounded to the nearest hertz.

If nchan == None then this value will return the maximum bandwidth based on the max_nchan property.

Parameters: nchan (int | None, optional) – the number of channels, defaults to the max_nchan value of the configuration.
Returns: the calculated bandwidth, in MHz, given the configuration.
Return type: float

calculate_channel_range(bandwidth_mhz: float, centre_freq_mhz: float) → Tuple[int, int][source]

Calculate the start and end channel given a centre frequency and bandwidth.

The total number of channels is calculated from the bandwidth, tsamp and oversampling ratio of the frequency band configuration.

The start channel will be a multiple of the number of channels in a UDP packet, relative to the minimum valid channel.

Parameters

bandwidth_mhz (float) – the required bandwidth of the RF signal, in MHz.
centre_freq_mhz (float) – the centre frequency of the RF signal, in MHz.

Returns

a tuple representing the range of the channels, inclusive of the start channel but exclusive of the end channel: [start_chan, end_chan)

Return type

Tuple[int, int]

centre_freq_mhz(start_chan: int = 0, nchan: Optional[int] = None) → float[source]

Get the centre frequency, in MHz, based on the start channel and the number of channels.

The centre frequency is rounded to the nearest hertz.

Parameters

start_chan (int, optional) – the start channel, defaults to 0
nchan (int | None, optional) – the number of channels, defaults to the max_nchan value of the configuration.

Returns

the calculated centre frequency given the channel separation, bottom frequency, starting channel, and number of channels.

Return type

float

chan_separation_khz: float: The frequency separation between the start and end of a channel, in kHz.

property chan_separation_mhz: float

Get the channel frequency separation, in MHz.

This is a derived property that converts the chan_separation_khz, which is in kHz, to MHz.

Returns: the channel frequency separation, in MHz.
Return type: float

channel_0_start_freq_mhz: float: The effective start frequency of channel 0, in MHz.

channel_centre_freq_mhz(chan: int = 0) → float[source]

Get the channel centre frequency, in MHz.

Parameters: chan – the PST fine channel to get the centre frequency for.
Returns: the calculated centre frequency for the given channel.
Return type: float

channelisation_stages: List[ChannelisationStage]: The configuration used in the channelisation of the RF signal by the CBF.

data_rate(nchan: Optional[int] = None) → float[source]

Get the data rate, in bytes/second, for the given number of channels.

If nchan == None then this value will return the maximum data rate based on the max_nchan property.

Parameters: nchan (int | None, optional) – the number of channels, defaults to the max_nchan value of the configuration.
Returns: the calculated number of bytes per second.
Return type: float

max_nchan: int: The maximum number of channels for the frequency band.

property max_valid_chan: int: Get the maximum valid channel number for the frequency band.

property min_valid_chan: int: Get the minimum valid channel number for the frequency band.

nbit: int

The number of bits per sample.

This value is the number of bits used for a value within a given dimension. For complex valued data the total number of bits used for the complex value is 2 * nbit.

nchan_for_bandwidth(bandwidth_mhz: float, **kwargs: Any) → int[source]

Get the number of channels closest to the given bandwidth.

This determines the number of channels the given bandwidth, in MHz, would be equivalent to given the tsamp and oversampling_ratio.

The output number of channels is given as a multiple of udp_nchan.

Parameters: bandwidth_mhz (float) – the desired bandwidth, in MHz.
Returns: the closest number of channels, as a multiple of udp_nchan that span the given bandwidth.
Return type: int

nchan_for_data_rate(data_rate: float) → int[source]

Calculate the number of channels given a input data rate.

The calculation of the input bytes per seconds is given by:

data_rate = nchan * npol * nbit * ndim / 8 / (tsamp / 1e6)

However, this method ensures the output nchan is not greater than the max_nchan value but is also a multiple of udp_nchan.

Parameters: data_rate (float) – the requested input data rate, in bytes per second.
Returns: the number of channels needed to get to a data rate close to data_rate
Return type: int

ndim: int = 2

The number of dimensions of the input data.

If the value is 1, then the input data is real valued. If the value is 2 then the input data is complex valued. The default is 2 (i.e. complex valued data).

npol: int = 2

The number of polarisations of the input data.

The default is 2.

property num_channelisation_stages: int: Get the number of stages of channelisation for the frequency band.

oversampling_ratio: Union[Tuple[int, int], List[int]]

The oversampling ratio for the frequency band.

This is represented as a tuple like (4, 3) or a list with only 2 elements.

pst_channel_number_for_freq(freq_mhz: float, centre_freq_mhz: float, bandwidth_mhz: float, strict: bool = False) → int[source]

Get the PST channel number for the given frequency.

The PST signal processing pipelines count the PST channels based on the input RF signal, with channel 0 being centre_freq_mhz - bandwidth_mhz / 2 with a total number of channels (nchan) being bandwidth_mhz / chan_separation_mhz.

This method is used to find the channel number within the RF signal that has a given frequency in MHz. If the frequency is outside of the given RF signal range then the channel number is clipped to be in the exclusive range of [0, nchan).

Parameters

freq_mhz (float) – the frequency to get the PST channel number for in MHz
centre_freq_mhz (float) – the centre frequency of the RF signal, in MHz.
bandwidth_mhz (float) – the required bandwidth of the RF signal, in MHz.
strict (bool) – test that the channel number resides within the band, default False

Returns

the PST channel number

Return type

int

rf_bw_mhz: float: The bandwidth of the band, in MHz.

rf_freq_maximum_mhz: float: The top end of the frequency band, in MHz.

rf_freq_minimum_mhz: float: The bottom end of the frequency band, in MHz.

ring_buffer_config: RingBufferConfig: Configuration specific for setting up ring buffers.

tsamp: float: The time per sample, in microseconds.

udp_format: str: The UDP format for the given frequency band.

udp_nchan: int: The number of channels in an UDP packet.

udp_nsamp: int

The number of samples per channel in an UDP packet.

For each of the udp_nchan channels in an UDP packet there are this many samples.

wt_nbit: int = 16

The number of bits per weight.

The default is 16.

wt_nsamp: int

The number of samples per weight per channel in an UDP packet.

This value should be the same as udp_nsamp

class ska_pst.common.ChannelisationStage(*, num_frequency_channels: int, oversampling_ratio: Union[Tuple[int, int], List[int]])[source]

A data class used to store the configuration for a CBF channelisation stage.

num_frequency_channels: int: Number of channels used in the channelisation stage.

oversampling_ratio: Union[Tuple[int, int], List[int]]

The oversampling ratio for the channelisation stage.

This is represented as a tuple like (4, 3) or a list with only 2 elements.

class ska_pst.common.FrequencyBandConfig(*, receiver_config: ReceiverConfig, cbf_pst_config: CbfPstConfig)[source]

A data class used to model specific configuration for a given frequency band.

cbf_pst_config: CbfPstConfig: Configuration specific to the CBF/PST interface for a given frequency band.

receiver_config: ReceiverConfig: Configuration specific to the receiver for a given frequency band.

class ska_pst.common.NormalSampleStats(*, mean: float = 0.0, variance: float = 1.0, ndat: Optional[int] = None)[source]

A data class used to calculate sample statistics of a normal distribution.

This class can be used to get the mean, standard deviation / variance, median and median absolute deviation (MAD) of a known sample from a given population which has given mean and variance.

mean: float = 0.0

The population mean.

Defaults to 0.0.

ndat: int | None = None

The number of samples in the sample distribution.

Defaults to None to allow delaying or reusing the stats to finally get the sample values. This must be set before calling any of the sample_* properties off this statistics class.

property population_mad: float

Get the theoretical median absolute deviation (MAD) of the population.

This is calculated from the stddev by a scalar factor of around 1.4826. See [https://en.wikipedia.org/wiki/Median_absolute_deviation#Relation_to_standard_deviation] for more details.

Returns: the theoretical median absolute deviation (MAD) of the population.
Return type: float

property population_stddev: float

Get the population standard deviation.

This returns sqrt(self.variance).

Returns: the population standard deviation.
Return type: float

property sample_mad_estimate: ValueErrorEstimate

Get the sample’s MAD error estimate.

Returns: the sample’s MAD error estimate.
Return type: ValueErrorEstimate

property sample_mad_stddev: float

Get the standard deviation of the sample MAD.

This is defined as sqrt(pi/2) * pop_stddev / sqrt(ndat)

Returns: the standard deviation of the sample MAD.
Return type: float

property sample_mean_estimate: ValueErrorEstimate

Get the sample’s mean error estimate.

Returns: the sample’s mean error estimate.
Return type: ValueErrorEstimate

property sample_mean_stddev: float

Get the standard deviation of the sample mean.

This is defined as sqrt(variance / ndat).

Returns: the standard deviation of the sample mean.
Return type: float

property sample_median_estimate: ValueErrorEstimate

Get the sample’s median error estimate.

Returns: the sample’s median error estimate.
Return type: ValueErrorEstimate

property sample_median_stddev: float

Get the standard deviation of the sample median.

This is defined as sqrt(pi/2) * (sample's mean stddev)

Returns: the standard deviation of the sample median.
Return type: float

property sample_variance_estimate: ValueErrorEstimate

Get the sample’s variance error estimate.

Returns: the sample’s variance error estimate.
Return type: ValueErrorEstimate

property sample_variance_stddev: float

Get the standard deviation of the sample variance.

This is defined as sqrt(2.0 * (ndat - 1)) * variance / ndat

Returns: the standard deviation of the sample variance.
Return type: float

variance: float = 1.0

The population variance.

Defaults to 1.0

class ska_pst.common.ReceiverConfig(*, receiver_id: str, feed_polarization: str, feed_handedness: int, feed_angle: float, feed_tracking_mode: str, feed_position_angle: float)[source]

A data class used to model specific configuration of a receiver.

feed_angle: float

The feed angle of the receiver.

Feed angle of the E-vector of the receiver for an equal in-phase response from the A(X) and B(Y) probes, measured in the direction of increasing feed angle or position angle (clockwise when looking down on a prime focus receiver).

feed_handedness: int

The handedness of the feed of the receiver.

For value of +1 for XYZ forming RH set with Z in the direction of propagation. Looking up into the feed of a prime-focus receiver or at the sky).

For FD_HAND = +1, the rotation from A (or X) to B (or Y) is counter clockwise or in the direction of increasing Feed Angle (FA) or Position Angle (PA).

For circular feeds, FD_HAND = +1 for IEEE LCP on the A (or X) probe.

feed_polarization: str: The native polarisation of the feed for the receiver.

feed_position_angle: float

The requested angle of feed reference.

For feed_mode = ‘FA’ this is respect to the telescope’s reference frame (feed_angle = 0), and for feed_mode = ‘CPA’ this is with respect to the celestial north (parallactic angle = 0) or with respect to the Galactic north for coordinate_mode = ‘GALACTIC’.

feed_tracking_mode: str

The tracking mode for the feed of the receiver.

FA - constant feed angle and that the feed stays fixed with respect to the
telescope’s reference frame.
CPA - the feed rotates to maintain a constant phase angle (i.e. it tracks
the variation of the parallactic angle.). When the coordinate mode is GALACTIC, PA is with respect to Galactic north and similarly for coordinate mode ECLIPTIC then PA is with respect to ecliptic north.
SPA - the feed angle is held fixed at an angle such that the requested PA is
obtained at the mid-point of the observation.
TPA - is only relevant for scan observations - the feed is rotated to maintain a
constant angle with respect to the scan direction.

receiver_id: str

The name/id of the receiver.

This value should match the receiver_id in the CSP/PST scan configuration.

class ska_pst.common.RingBufferConfig(*, num_buffers: int, packets_per_buffer: int)[source]

A data class used to store ring buffer (RB) configuration for a given frequency band.

An element within the RB is configured to be large enough to capture all the channels of the subband for packets_per_buffer number of packets.

num_buffers: int: Number of buffers withing a ring buffer.

packets_per_buffer: int: The number of UDP packets per buffer that a channel needs be recorded for.

class ska_pst.common.TelescopeConfig(*, name: str, frequency_bands: ~typing.Dict[str, ~ska_pst.common.telescope_configuration.FrequencyBandConfig] = <factory>)[source]

A data class used to model telescope specific configuration used by PST.

frequency_bands: Dict[str, FrequencyBandConfig]

Configuration specific to the frequency bands for a given telescope.

For SKALow there is only 1 frequency band, “low”, while SKAMid has multiple frequency bands. The configuration that should be used for a given scan request will depend on what telescope the scan is for and the frequency band if the telescope is SKAMid.

get_cbf_pst_config(frequency_band: Optional[str] = None, **kwargs: Any) → CbfPstConfig[source]

Get the configuration specific for a frequency band.

This will return the configuration that is specific to a frequency band. The standard for SKA is that if the frequency_band is not set or is “low” then it corresponds to the Low telescope, which has only one band. Frequency bands of 1, 2, 3, 4, 5a, or 5b will return specific configuration.

Parameters: frequency_band (str | None, optional) – the frequency band to get configuration for, defaults to None
Returns: configuration for the frequency band.
Return type: CbfPstConfig

name: str

The name of the telescope.

This should be either SKALow or SKAMid.

class ska_pst.common.TelescopeFacilityEnum(value)[source]

Enum representing the different telescope facilities within SKAO.

Low = 1: Used to present that the functionality is for the SKA-Low facility.

Mid = 2: Used to present that the functionality is for the SKA-Mid facility.

static from_telescope(telescope: str) → TelescopeFacilityEnum[source]

Get the enum value based on telescope string.

The telescope parameter must be either “SKALow” or “SKAMid”.

Parameters: telescope (str) – the name of the telescope to get the enum for.
Returns: the enum value based on telescope string.
Return type: TelescopeFacilityEnum

property telescope: str

Get the SKA telescope that the facility enum represents.

Returns: the SKA telescope that the facility enum represents.
Return type: str

class ska_pst.common.ValueErrorEstimate(*, val: float = 0.0, variance: float = 1.0)[source]

A data class used to get the error of a value with a known value and variance.

This class implements some Python magic dunder methods to make it easier to do manipulation and propagation of the errors, such as ** (power), / (division), * (multiplication).

property error: float

Get the error of the value.

This is the standard deviation in the error and is sqrt(variance).

Returns: the error of the value.
Return type: float

inverse() → ValueErrorEstimate[source]

Get the inverse of the value and variance.

This is the programmatic way of doing 1/val. This will return a new value and it’s error estimate.

Returns: the inverse of the value and variance.
Return type: ValueErrorEstimate

sqrt() → ValueErrorEstimate[source]

Get the square root of the value and the variance.

Returns: the square root of the value and the variance.
Return type: ValueErrorEstimate

val: float = 0.0: The value that has an associated variance in it’s error.

variance: float = 1.0: The variance in the value.

ska_pst.common.convert_csp_config_to_pst_config(telescope_config: TelescopeConfig, csp_configure_scan_request: dict) → dict[source]

Convert a CSP configure scan request into a PST configure scan request.

This method flattens the configuration by selection the common and pst/scan paths of the CSP request and putting them all into one dictionary.

It takes the frequency_band value in the common section of the request and gets the frequency band configuration from the telescope_config to augment the request. For the version 3.0+ of the PST schema a lot of these parameters have been removed from the CSP/PST request but the values are needed within the PST LMC to send down to CORE applications.

Parameters

telescope_config (TelescopeConfig) – the telescope configuration that PST BEAM is running in.
csp_configure_scan_request (dict) – the CSP configure scan request as Python dict.

Returns

a dictionary used internally by PST to handling configuring of the PST BEAM.

Return type

dict

ska_pst.common.get_telescope_config(telescope: str) → TelescopeConfig[source]

Get the telescope configuration for a given telescope.

Parameters: telescope (str) – the name of the telescope.
Returns: the configuration specific to the telescope.
Return type: TelescopeConfig

ska_pst.common.get_udp_nsamp_for_format(udp_format: str) → int[source]

Get the UDP_NSAMP for a given UDP Format.

NOTE: This method is only used by the WeightsFileReader both in the SEND Python code and the ska_pst.testutils package. In both cases the value should get this value based off the telescope and frequency band rather than calling this method.

Returns: the UDP_NSAMP for the given UDP format.
Return type: int
Raises: AssertionError – if the UDP format is unknown.

ska_pst.common.round_(value: float, precision: int = 16) → float[source]

Round a value to a given precision.

Parameters

value (float) – the value to round to a given precision
precision (int, optional) – the number of digits of precision, defaults to 16

Returns

the value rounded to a given precision.

Return type

float