Source code for ska_tmc_cdm.messages.subarray_node.configure.csp

The configure.csp module contains Python classes that represent the various
aspects of CSP configuration that may be specified in a SubArrayNode.configure
import enum
from typing import List, Tuple, Optional

from . import core

__all__ = ["CSPConfiguration", "FSPConfiguration", "FSPFunctionMode",
           "CBFConfiguration", "SubarrayConfiguration",

[docs]class FSPFunctionMode(enum.Enum): """ FSPFunctionMode is an enumeration of the available FSP modes. """ CORR = "CORR" PSS_BF = "PSS-BF" PST_BF = "PST-BF" VLBI = "VLBI"
[docs]class FSPConfiguration: """ FSPConfiguration defines the configuration for a CSP Frequency Slice Processor. """ # pylint: disable=too-many-arguments def __init__( self, fsp_id: int, function_mode: FSPFunctionMode, frequency_slice_id: int, integration_factor: int, zoom_factor: int, channel_averaging_map: List[Tuple] = None, output_link_map: List[Tuple] = None, channel_offset: int = None, zoom_window_tuning: int = None ): """ Create a new FSPConfiguration. Channel averaging map is an optional list of 20 x (int,int) tuples. :param fsp_id: FSP configuration ID [1..27] :param function_mode: FSP function mode :param frequency_slice_id: frequency slicer ID [1..26] :param zoom_factor: zoom factor [0..6] :param integration_factor: integration factor [1..10] :param channel_averaging_map: Optional channel averaging map :param output_link_map: Optional output link map :param channel_offset: Optional FSP channel offset :param zoom_window_tuning: Optional zoom window tuning :raises ValueError: Invalid parameter values entered """ if not 1 <= fsp_id <= 27: raise ValueError("FSP ID must be in range 1..27. Got {}".format(fsp_id)) self.fsp_id = fsp_id self.function_mode = function_mode if not 1 <= frequency_slice_id <= 26: msg = "Frequency slice ID must be in range 1..26. Got {}" "".format( frequency_slice_id ) raise ValueError(msg) self.frequency_slice_id = frequency_slice_id if not 0 <= zoom_factor <= 6: msg = "Zoom factor must be in range 0..6. Got {}".format( zoom_factor ) raise ValueError(msg) self.zoom_factor = zoom_factor if not 1 <= integration_factor <= 10: msg = "Integration factor must in range 1..10. Got {}" "".format( integration_factor ) raise ValueError(msg) self.integration_factor = integration_factor if channel_averaging_map and len(channel_averaging_map) > 20: msg = ( "Number of tuples in channel averaging map must be 20 or fewer." f"Got {len(channel_averaging_map)}" ) raise ValueError(msg) self.channel_averaging_map = channel_averaging_map # TODO: update enforcements for output_link_map self.output_link_map = output_link_map self.channel_offset = channel_offset self.zoom_window_tuning = zoom_window_tuning def __eq__(self, other): if not isinstance(other, FSPConfiguration): return False return ( self.fsp_id == other.fsp_id and self.function_mode == other.function_mode and self.frequency_slice_id == other.frequency_slice_id and self.zoom_factor == other.zoom_factor and self.integration_factor == other.integration_factor and self.channel_averaging_map == other.channel_averaging_map and self.output_link_map == other.output_link_map and self.channel_offset == other.channel_offset and self.zoom_window_tuning == other.zoom_window_tuning )
[docs]class SubarrayConfiguration: """ Class to hold the parameters relevant only for the current sub-array device. """ def __init__(self, subarray_name: str): """ Create sub-array device configuration. :param sub-array_name: Name of the sub-array """ self.subarray_name = subarray_name def __eq__(self, other): if not isinstance(other, SubarrayConfiguration): return False return ( self.subarray_name == other.subarray_name )
[docs]class CommonConfiguration: """ Class to hold the CSP sub-elements. """ def __init__( self, config_id: str, frequency_band: core.ReceiverBand, subarray_id: int = None, band_5_tuning: Optional[List[float]] = None ): """ Create a new CSPConfiguration. :param config_id: CSP configuration ID :param frequency_band: the frequency band to set :param subarray_id: an ID of sub-array device :param band_5_tuning: band 5 receiver to set (optional) """ self.config_id = config_id self.frequency_band = frequency_band self.subarray_id = subarray_id self.band_5_tuning = band_5_tuning def __eq__(self, other): if not isinstance(other, CommonConfiguration): return False return ( self.config_id == other.config_id and self.frequency_band == other.frequency_band and self.subarray_id == other.subarray_id and self.band_5_tuning == other.band_5_tuning )
class VLBIConfiguration: pass
[docs]class CBFConfiguration: """ Class to hold all FSP and VLBI configurations. """ def __init__( self, fsp_configs: List[FSPConfiguration], vlbi_config: VLBIConfiguration = None, ): """ Create a new CBFConfiguration. :param fsp_configs: the FSP configurations to set :param vlbi_config: the VLBI configurations to set, it is optional """ self.fsp_configs = fsp_configs self.vlbi_config = vlbi_config def __eq__(self, other): if not isinstance(other, CBFConfiguration): return False return ( self.fsp_configs == other.fsp_configs and self.vlbi_config == other.vlbi_config )
class PSTConfiguration: pass class PSSConfiguration: pass
[docs]class CSPConfiguration: """ Class to hold all CSP configuration. """ def __init__( self, interface: str = None, subarray_config: SubarrayConfiguration = None, common_config: CommonConfiguration = None, cbf_config: CBFConfiguration = None, pst_config: PSTConfiguration = None, pss_config: PSSConfiguration = None ): """ Create a new CSPConfiguration, In order to support backward compatibility, We have kept old attributes as it is and added support of new attributes as per ADR-18 :param interface: url string to determine JsonSchema version :param subarray_config: Sub-array configuration to set :param common_config: the common CSP elemenets to set :param cbf_config: the CBF configurations to set :param pst_config: the PST configurations to set :param pss_config: the PSS configurations to set """ self.interface = interface self.subarray_config = subarray_config self.common_config = common_config self.cbf_config = cbf_config self.pst_config = pst_config self.pss_config = pss_config def __eq__(self, other): if not isinstance(other, CSPConfiguration): return False return ( self.interface == other.interface and self.subarray_config == other.subarray_config and self.common_config == other.common_config and self.cbf_config == other.cbf_config and self.pst_config == other.pst_config and self.pss_config == other.pss_config )