"""
The pdm_transforms module contains code to transform Project Data Model (PDM)
entities to Configuration Data Model (CDM) entities. The pdm_transforms code
is called by observing scripts to convert the PDM Scheduling Block to the
equivalent CDM configurations, which are then sent to TMC devices to control
the telescope.
"""
import logging
from ska_oso_pdm import SBDefinition
from ska_oso_pdm.sb_definition import (
AltAzCoordinates,
CSPConfiguration,
GalacticCoordinates,
ICRSCoordinates,
Target,
)
from ska_oso_pdm.sb_definition.mccs.mccs_allocation import (
SubarrayBeamConfiguration as pdm_SubarrayBeamConfiguration,
)
from ska_tmc_cdm.messages.central_node.mccs import (
ApertureConfiguration as cdm_ApertureConfiguration,
)
from ska_tmc_cdm.messages.central_node.mccs import MCCSAllocate as cdm_MCCSAllocate
from ska_tmc_cdm.messages.central_node.mccs import (
SubArrayBeamsConfiguration as cdm_SubArrayBeamsConfiguration,
)
from ska_tmc_cdm.messages.subarray_node.configure.mccs import (
MCCSConfiguration as cdm_MCCSConfiguration,
)
from ska_tmc_cdm.messages.subarray_node.configure.mccs import (
SubarrayBeamAperatures as cdm_SubarrayBeamAperatures,
)
from ska_tmc_cdm.messages.subarray_node.configure.mccs import (
SubarrayBeamConfiguration as cdm_SubarrayBeamConfiguration,
)
from ska_tmc_cdm.messages.subarray_node.configure.mccs import (
SubarrayBeamSkyCoordinates as cdm_SubarrayBeamSkyCoordinates,
)
from ska_oso_scripting.pdm_transforms.common import (
csp_configuration_for_subarray_beam_scan,
get_aperture_id,
get_station_id_counts,
target_for_subarray_beam_scan,
)
from ska_oso_scripting.pdm_transforms.csp import _calculate_logical_bands
CDM_MCC_SCHEMA_URI = "https://schema.skao.int/ska-low-mccs-controller-allocate/3.0"
LOG = logging.getLogger(__name__)
FORMAT = "%(asctime)-15s %(message)s"
logging.basicConfig(level=logging.INFO, format=FORMAT)
# Not every function in this module should be called externally
__all__ = [
"create_mccs_allocation",
"create_mccs_configuration",
"get_allocation_apertures",
"get_sky_coordinates_based_on_reference_coordinates",
]
[docs]
def get_allocation_apertures(
subarray_beam: pdm_SubarrayBeamConfiguration,
) -> list[cdm_ApertureConfiguration]:
"""
Get the apertures from one of the subArray beam of MCC Allocation
"""
station_id_counts = get_station_id_counts(subarray_beam)
apertures = [
cdm_ApertureConfiguration(
station_id=aperture.station_id,
aperture_id=get_aperture_id(
aperture.station_id,
aperture.substation_id,
station_id_counts,
),
)
for aperture in subarray_beam.apertures
]
return apertures
[docs]
def create_mccs_allocation(sbd: SBDefinition) -> cdm_MCCSAllocate:
"""
Creates an CDM MCCSAllocate instance from the relevant parts of the SBDefinition.
"""
cdm_subarray_beams = [
cdm_SubArrayBeamsConfiguration(
subarray_beam_id=subarray_beam.subarray_beam_id,
number_of_channels=_max_number_of_channels(
_all_csp_configurations_for_subarray_beam(subarray_beam, sbd=sbd)
),
apertures=get_allocation_apertures(subarray_beam),
)
for subarray_beam in sbd.mccs_allocation.subarray_beams
]
cdm_allocation = cdm_MCCSAllocate(
interface=CDM_MCC_SCHEMA_URI,
subarray_beams=cdm_subarray_beams,
)
return cdm_allocation
[docs]
def get_sky_coordinates_based_on_reference_coordinates(
target: Target,
) -> cdm_SubarrayBeamSkyCoordinates:
"""
Get the sky coordinates based on target reference_coordinates from the SB
"""
if not isinstance(target, Target):
raise TypeError(f"Expected PDM Target, got {type(target)}")
ref_coord = target.reference_coordinate
match ref_coord:
case AltAzCoordinates():
beam_coords = cdm_SubarrayBeamSkyCoordinates(
reference_frame="topocentric",
c1=ref_coord.az,
c2=ref_coord.el,
)
case GalacticCoordinates():
beam_coords = cdm_SubarrayBeamSkyCoordinates(
reference_frame="galactic",
c1=ref_coord.l,
c2=ref_coord.b,
)
case ICRSCoordinates():
beam_coords = cdm_SubarrayBeamSkyCoordinates(
reference_frame="ICRS",
c1=ref_coord.to_sky_coord().ra.degree,
c2=ref_coord.to_sky_coord().dec.degree,
)
case _:
raise TypeError(f"reference_coordinate {ref_coord}, is not supported")
return beam_coords
def get_config_apertures(
subarray_beam: pdm_SubarrayBeamConfiguration,
) -> list[cdm_SubarrayBeamAperatures]:
"""
Get the apertures from one of the subArray beam of MCC Allocation
"""
station_id_counts = get_station_id_counts(subarray_beam)
apertures = [
cdm_SubarrayBeamAperatures(
weighting_key_ref=aperture.weighting_key,
aperture_id=get_aperture_id(
aperture.station_id,
aperture.substation_id,
station_id_counts,
),
)
for aperture in subarray_beam.apertures
]
return apertures
[docs]
def create_mccs_configuration(
scan_number: int, sbd: SBDefinition
) -> cdm_MCCSConfiguration:
"""
Create a CDM MCCSConfiguration for the given scan in the Low SBDefinition.
"""
mccs_configuration = cdm_MCCSConfiguration(
subarray_beam_configs=[
cdm_SubarrayBeamConfiguration(
update_rate=0.0,
logical_bands=_calculate_logical_bands(
csp_configuration_for_subarray_beam_scan(
subarray_beam, scan_number, sbd
).lowcbf
),
apertures=get_config_apertures(subarray_beam),
sky_coordinates=get_sky_coordinates_based_on_reference_coordinates(
target_for_subarray_beam_scan(subarray_beam, scan_number, sbd)
),
subarray_beam_id=subarray_beam.subarray_beam_id,
)
for subarray_beam in sbd.mccs_allocation.subarray_beams
],
)
return mccs_configuration
def _max_number_of_channels(csp_configurations: list[CSPConfiguration]) -> int:
# We want to assign resources using the maximum number of channels of all the CSP configurations.
# This function will sum up the number of channels in each CSP configuration (by summing the
# individual SPWs) then will return the maximum of these values
return max(
[
sum(
[
correlation_spw.number_of_channels
for correlation_spw in csp_configuration.lowcbf.correlation_spws
]
)
for csp_configuration in csp_configurations
]
)
def _all_csp_configurations_for_subarray_beam(
subarray_beam: pdm_SubarrayBeamConfiguration, sbd: SBDefinition
) -> list[CSPConfiguration]:
csp_configuration_refs = [
scan_definition.csp_configuration_ref
for scan_definition in subarray_beam.scan_sequence
]
return [
csp_configuration
for csp_configuration in sbd.csp_configurations
if csp_configuration.config_id in csp_configuration_refs
]