"""
The configure.csp module contains Python classes that represent the various
aspects of CSP configuration that may be specified in a SubArrayNode.configure
command.
"""
from enum import Enum
from typing import Optional, Tuple
from pydantic import AliasChoices, Field, model_validator
from ska_tmc_cdm.messages.base import CdmObject
from ...skydirection import SkyDirection
from . import core
from .pss import PSSConfiguration
from .pst import PSTConfiguration
__all__ = [
"FSPFunctionMode",
"CSPConfiguration",
"CorrelationConfiguration",
"SubarrayConfiguration",
"CommonConfiguration",
"LowCBFConfiguration",
"MidCBFConfiguration",
"StationConfiguration",
"StnBeamConfiguration",
"VisFspConfiguration",
"VisStnBeamConfiguration",
"VisConfiguration",
"VLBIConfiguration",
"TimingBeamsConfiguration",
"ProcessingRegionConfiguration",
"BeamsConfiguration",
"SearchBeamsConfiguration",
"SearchBeam",
]
MID_CSP_SCHEMA = "https://schema.skao.int/ska-csp-configurescan/8.3"
[docs]
class FSPFunctionMode(Enum):
"""
FSPFunctionMode is an enumeration of the available FSP modes.
"""
CORR = "CORR"
PSS_BF = "PSS-BF"
PST_BF = "PST-BF"
VLBI = "VLBI"
[docs]
class SubarrayConfiguration(CdmObject):
"""
Class to hold the parameters relevant only for the current sub-array device.
:param sub-array_name: Name of the sub-array
"""
subarray_name: Optional[str] = ""
[docs]
class CommonConfiguration(CdmObject):
"""
Class to hold the CSP sub-elements.
:param config_id: CSP configuration ID
:param frequency_band: the frequency band to set
:param subarray_id: subarray_id
:param band_5_tuning: band 5 receiver to set (optional)
:param eb_id: eb_id
"""
config_id: Optional[str] = None
frequency_band: Optional[core.ReceiverBand] = None
subarray_id: Optional[int] = None
band_5_tuning: Optional[list[float]] = None
eb_id: Optional[str] = None
# The following model validator has been removed temporarily during the Band 5b down converter work and should be
# re-enabled at a later date
# @model_validator(mode="after")
# def validate_subarray_id_only_band_5(self):
# band5 = (core.ReceiverBand.BAND_5A, core.ReceiverBand.BAND_5B)
# if self.frequency_band in band5 and self.band_5_tuning is None:
# raise ValueError("Band 5 must have a band 5 tuning")
# if self.frequency_band not in band5 and self.band_5_tuning is not None:
# raise ValueError("Only Band 5 may have a band 5 tuning")
# return self
[docs]
class StnBeamConfiguration(CdmObject):
"""
Class to hold Stations Beam Configuration.
:param stn_beam_id: stn_beam_id
:param beam_id: beam_id
:param freq_ids: freq_ids
:param delay_poly: delay_poly
"""
beam_id: Optional[int]
freq_ids: list[int]
stn_beam_id: Optional[int] = None
delay_poly: Optional[str] = None
[docs]
class VisStnBeamConfiguration(CdmObject):
"""
Class to hold Vis Stations Beam Configuration.
:param stn_beam_id: stn_beam_id
:param host: host
:param port: port
:param mac: mac
:param integration_ms: integration_ms
"""
stn_beam_id: Optional[int]
integration_ms: int
host: Optional[list[Tuple[int, str]]] = None
port: Optional[list[Tuple[int, int, int]]] = None
mac: Optional[list[Tuple[int, str]]] = None
[docs]
class StationConfiguration(CdmObject):
"""
Class to hold Stations Configuration.
:param stns: stns
:param stn_beams: stn_beams
"""
stns: Optional[list[list[int]]] = None
stn_beams: Optional[list[StnBeamConfiguration]] = None
[docs]
class VisFspConfiguration(CdmObject):
"""
Class to hold Visibility(Vis) Configuration.
:param function_mode: function_mode
:param fsp_ids: fsp_ids
:param firmware: firmware
"""
function_mode: Optional[str] = None
fsp_ids: Optional[list[int]] = None
firmware: Optional[str] = None
[docs]
class VisConfiguration(CdmObject):
"""
Class to hold Vis Configuration firmware
:param fsp: fsp
:param stn_beams: stn_beams
"""
fsp: Optional[VisFspConfiguration] = None
stn_beams: Optional[list[VisStnBeamConfiguration]] = None
[docs]
class BeamsConfiguration(CdmObject):
"""
Class to hold PST Beams Configuration.
:param pst_beam_id: pst_beam_id
:param stn_beam_id: stn_beam_id
:param stn_weights: stn_weights
"""
pst_beam_id: Optional[int] = None
stn_beam_id: Optional[int] = None
stn_weights: list[float] = Field(default_factory=list)
# LowCBF Field introduced in v4.1. If omitted, MCCS target coords would be applied.
field: Optional[SkyDirection] = None
[docs]
class SearchBeam(CdmObject):
"""
Class to hold Search Beams.
:param pss_beam_id: pss_beam_id
:param stn_beam_id: stn_beam_id
:param stn_weights: stn_weights
"""
pss_beam_id: Optional[int] = None
stn_beam_id: Optional[int] = None
stn_weights: list[float] = Field(default_factory=list)
[docs]
class TimingBeamsConfiguration(CdmObject):
"""
Class to hold TimingBeams Configuration.
:param fsp: fsp
:param firmware: firmware
:param beams: beams
"""
fsp: Optional[VisFspConfiguration] = None
firmware: Optional[str] = None
beams: list[BeamsConfiguration] = Field(default_factory=list)
[docs]
class SearchBeamsConfiguration(CdmObject):
"""
Class to hold SearchBeams Configuration.
:param firmware: firmware
:param beams: beams
"""
firmware: str
beams: list[SearchBeam] = Field(default_factory=list)
[docs]
class LowCBFConfiguration(CdmObject):
"""
Class to hold Low CBF Configuration.
:param stations: stations
:param vis: vis
"""
stations: Optional[StationConfiguration] = None
vis: Optional[VisConfiguration] = None
timing_beams: Optional[TimingBeamsConfiguration] = None
search_beams: Optional[SearchBeamsConfiguration] = None
[docs]
class VLBIConfiguration(CdmObject):
pass
[docs]
class ProcessingRegionConfiguration(CdmObject):
"""
Class to hold Processing Region Configuration.
:param fsp_ids: fsp_ids
:param receptors: receptors
:param start_freq: start_freq
:param channel_width: channel_width
:param channel_count: channel_count
:param integration_factor: integration_factor
:param sdp_start_channel_id: sdp_start_channel_id
"""
fsp_ids: list[int]
receptors: list[str] = Field(default_factory=list)
start_freq: int = Field(ge=350_000_000, le=15_400_000_000)
channel_width: int = 13440
channel_count: int = Field(ge=1, le=58982, multiple_of=20)
integration_factor: int = Field(ge=1, le=10)
sdp_start_channel_id: int = Field(ge=0, le=4294901760)
class PSTBFBeamsConfiguration(CdmObject):
"""
Class to hold PST Beams Configuration.
:param timing_beam_id: timing_beam_id
:param receptors: receptors
"""
timing_beam_id: int
receptors: Optional[list[str]] = Field(default_factory=list)
class PSTBFProcessingRegionConfiguration(CdmObject):
"""
Class to hold Processing Region Configuration for PST Beamformer.
:param fsp_ids: fsp_ids
:param start_freq: start_freq
:param channel_count: channel_count
:param timing_beams: timing_beams
"""
fsp_ids: list[int] = Field(default_factory=list)
start_freq: int = Field(ge=0, le=15399982080, multiple_of=53760)
channel_count: int = Field(ge=1, le=47923)
timing_beams: list[PSTBFBeamsConfiguration] = Field(default_factory=list)
class PSTBFConfiguration(CdmObject):
"""
Class to hold PST Beamformer Configuration.
:param processing_regions: processing_regions
"""
processing_regions: list[PSTBFProcessingRegionConfiguration]
[docs]
class CorrelationConfiguration(CdmObject):
"""
Class to hold Correlation Configuration.
:param processing_regions: processing_regions
"""
processing_regions: list[ProcessingRegionConfiguration]
[docs]
class MidCBFConfiguration(CdmObject):
"""
Class to hold all FSP and VLBI configurations.
Note: frequency band offset (FBO)
Bands 1 and 2 should be specified for Stream 1 only.
Bands 5a and 5b input from the receptor consists of two data streams (1 and 2)
:param frequency_band_offset_stream1: a specified offset so that the entire observed band is shifted in Hz
:param frequency_band_offset_stream2: a specified offset so that the entire observed band is shifted in Hz
:param correlation: correlation specific parameters
:param vlbi_config: the VLBI configurations to set, it is optional
"""
frequency_band_offset_stream1: Optional[int] = Field(None, ge=-1e8, le=1e8)
frequency_band_offset_stream2: Optional[int] = Field(None, ge=-1e8, le=1e8)
correlation: Optional[CorrelationConfiguration] = None
vlbi_config: Optional[VLBIConfiguration] = Field(
default=None,
serialization_alias="vlbi",
validation_alias=AliasChoices("vlbi", "vlbi_config"),
)
pst_bf: Optional[PSTBFConfiguration] = Field(default=None)
[docs]
class CSPConfiguration(CdmObject):
"""
Class to hold all CSP configuration. 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 common: the common CSP elements to set
:param midcbf: the MID CBF configurations to set
:param lowcbf: the LOW CBF configurations to set
:param pst_config: the PST configurations to set
:param pss_config: the PSS configurations to set
"""
interface: str = MID_CSP_SCHEMA
subarray: Optional[SubarrayConfiguration] = None
common: CommonConfiguration
midcbf: Optional[MidCBFConfiguration] = None
lowcbf: Optional[LowCBFConfiguration] = None
pst_config: Optional[PSTConfiguration] = Field(
default=None,
serialization_alias="pst",
validation_alias=AliasChoices("pst", "pst_config"),
)
pss_config: Optional[PSSConfiguration] = Field(
default=None,
serialization_alias="pss",
validation_alias=AliasChoices("pss", "pss_config"),
)
@model_validator(mode="after")
def validate_interface(self):
if self.interface == MID_CSP_SCHEMA:
if self.common.subarray_id is not None:
raise ValueError(
f"subarray_id is not supported for CSP Configuration schema version {MID_CSP_SCHEMA}"
)
elif self.common.config_id is None:
raise ValueError(
f"config_id is mandatory for CSP Configuration schema version {MID_CSP_SCHEMA}"
)
return self