"""
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 dataclasses import dataclass
from typing import Optional
from ska_oso_pdm import SBDefinition, TelescopeType
from ska_oso_pdm.sb_definition import (
FivePointParameters,
PointedMosaicParameters,
PointingPatternParameters,
ScanDefinition,
SDPConfiguration,
SinglePointParameters,
Target,
)
from ska_oso_pdm.sb_definition.mccs.mccs_allocation import (
MCCSAllocation as pdm_MCCSAllocation,
)
from ska_tmc_cdm.messages.central_node.assign_resources import (
AssignResourcesRequest as cdm_AssignResourcesRequest,
)
from ska_tmc_cdm.messages.central_node.csp import (
CSPConfiguration as cdm_central_node_CSPConfiguration,
)
from ska_tmc_cdm.messages.subarray_node.configure import (
ConfigureRequest as cdm_ConfigureRequest,
)
from ska_tmc_cdm.messages.subarray_node.configure import (
PointingConfiguration as cdm_PointingConfiguration,
)
from ska_tmc_cdm.messages.subarray_node.configure.core import Target as cdm_Target
from ska_tmc_cdm.messages.subarray_node.configure.receptorgroup import (
FixedTrajectory as cdm_FixedTrajectory,
)
from ska_tmc_cdm.messages.subarray_node.configure.receptorgroup import (
Projection,
ReceptorGroup,
)
from ska_tmc_cdm.messages.subarray_node.configure.tmc import (
TMCConfiguration as cdm_TMCConfiguration,
)
from .common import convert_tmcconfiguration
from .csp import convert_cspconfiguration, convert_pstconfiguration_centralnode
from .dish import (
convert_dishallocation,
convert_pointingconfiguration,
create_dishconfiguration,
)
from .mccs import convert_mccs_configuration, create_mccs_allocation
from .sdp import (
convert_sdpconfiguration_centralnode,
convert_sdpconfiguration_subarraynode,
create_sdpconfiguration_centralnode,
create_sdpconfiguration_subarraynode,
)
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_cdm_configure_request_from_scheduling_block",
"create_cdm_assign_resources_request_from_scheduling_block",
]
MID_CONFIGURE_SCHEMA = "https://schema.skao.int/ska-tmc-configure/4.2"
LOW_CONFIGURE_SCHEMA = "https://schema.skao.int/ska-low-tmc-configure/4.1"
@dataclass
class ScanData:
scan_definition: ScanDefinition
target: Target
csp_configuration: cdm_central_node_CSPConfiguration
mccs_allocation: Optional[pdm_MCCSAllocation]
sdp_configuration: Optional[SDPConfiguration]
[docs]
def create_cdm_assign_resources_request_from_scheduling_block(
subarray_id: int,
scheduling_block: SBDefinition,
) -> cdm_AssignResourcesRequest:
"""
creates a list of CDM AssignResourcesRequest object from a Scheduling Block
"""
if scheduling_block.telescope == TelescopeType.SKA_MID:
cdm_allocation_request = create_mid_assign_resources_request(
subarray_id, scheduling_block
)
else:
cdm_allocation_request = create_low_assign_resources_request(
subarray_id, scheduling_block
)
return cdm_allocation_request
def create_mid_assign_resources_request(
subarray_id: int, scheduling_block: SBDefinition
) -> cdm_AssignResourcesRequest:
"""
function to create an SKA MID assign resources request from a scheduling block
"""
cdm_dish = convert_dishallocation(scheduling_block.dish_allocations)
LOG.info(f"Setting dish : {cdm_dish.receptor_ids} ")
if scheduling_block.sdp_configuration:
cdm_sdp_config = convert_sdpconfiguration_centralnode(
scheduling_block.sdp_configuration, scheduling_block.targets
)
else:
cdm_sdp_config = create_sdpconfiguration_centralnode(scheduling_block)
LOG.info(
f"Setting SDP configuration for EB: {cdm_sdp_config.execution_block.eb_id} "
)
cdm_assign_resources_request = cdm_AssignResourcesRequest(
subarray_id=subarray_id,
dish_allocation=cdm_dish,
sdp_config=cdm_sdp_config,
)
return cdm_assign_resources_request
def create_low_assign_resources_request(
subarray_id: int,
scheduling_block: SBDefinition,
) -> cdm_AssignResourcesRequest:
"""
function to create an SKA LOW assign resources request from a scheduling block
"""
cdm_allocation = create_mccs_allocation(scheduling_block)
cdm_pst_config = convert_pstconfiguration_centralnode(scheduling_block.targets)
if scheduling_block.sdp_configuration:
cdm_sdp_config = convert_sdpconfiguration_centralnode(
scheduling_block.sdp_configuration, scheduling_block.targets
)
else:
cdm_sdp_config = create_sdpconfiguration_centralnode(scheduling_block)
cdm_assign_resources_request = cdm_AssignResourcesRequest(
subarray_id=subarray_id,
mccs=cdm_allocation,
sdp_config=cdm_sdp_config,
csp=cdm_central_node_CSPConfiguration(pst=cdm_pst_config),
)
return cdm_assign_resources_request
def create_low_single_point_configure_request(
id_instance_map: ScanData,
) -> cdm_ConfigureRequest:
"""
creates a CDM ConfigureRequest object for SKA LOW telescope with SINGLE_POINT pointing kind
"""
cdm_config = cdm_ConfigureRequest(
interface=LOW_CONFIGURE_SCHEMA,
mccs=convert_mccs_configuration(
allocation=id_instance_map.mccs_allocation,
lowcbf=id_instance_map.csp_configuration.lowcbf,
target=id_instance_map.target,
),
csp=convert_cspconfiguration(
TelescopeType.SKA_LOW,
pdm_config=id_instance_map.csp_configuration,
pdm_mccs_allocation=id_instance_map.mccs_allocation,
target=id_instance_map.target,
),
)
cdm_config.tmc = convert_tmcconfiguration(id_instance_map.scan_definition)
if id_instance_map.sdp_configuration:
cdm_config.sdp = convert_sdpconfiguration_subarraynode(
id_instance_map.scan_definition
)
else:
cdm_config.sdp = create_sdpconfiguration_subarraynode(
id_instance_map.scan_definition, id_instance_map.target
)
return cdm_config
def create_mid_single_point_configure_request(
id_instance_map: ScanData,
) -> cdm_ConfigureRequest:
"""
creates a CDM ConfigureRequest object for SKA MID telescope with SINGLE_POINT pointing kind
"""
cdm_config = cdm_ConfigureRequest(
interface=MID_CONFIGURE_SCHEMA,
dish=create_dishconfiguration(id_instance_map.csp_configuration.midcbf),
pointing=convert_pointingconfiguration(
target=id_instance_map.target,
correction=id_instance_map.scan_definition.pointing_correction,
),
csp=convert_cspconfiguration(
telescope=TelescopeType.SKA_MID,
pdm_config=id_instance_map.csp_configuration,
pdm_mccs_allocation=id_instance_map.mccs_allocation,
),
)
cdm_config.tmc = convert_tmcconfiguration(id_instance_map.scan_definition)
if id_instance_map.sdp_configuration:
cdm_config.sdp = convert_sdpconfiguration_subarraynode(
id_instance_map.scan_definition
)
else:
cdm_config.sdp = create_sdpconfiguration_subarraynode(
id_instance_map.scan_definition, id_instance_map.target
)
return cdm_config
def create_five_point_mid_configure_request(
id_instance_map: ScanData,
) -> list[cdm_ConfigureRequest]:
"""
creates a dictionary mapping PDM scan definition ids to a list of CDM ConfigureRequest
objects for SKA MID telescope with FIVE_POINT pointing kind
"""
initial_cdm_config = create_mid_single_point_configure_request(
id_instance_map=id_instance_map,
)
ca_ie_offset_pattern = [(0, 1), (0, -1), (1, 0), (-1, 0)]
pattern_params = __get_pointing_pattern(id_instance_map.target)
offset = pattern_params.offset_arcsec
partial_cdm_config = [
cdm_ConfigureRequest(
interface=initial_cdm_config.interface,
tmc=cdm_TMCConfiguration(partial_configuration=True),
pointing=cdm_PointingConfiguration(
ca_offset_arcsec=ca_ie_offset[0] * offset,
ie_offset_arcsec=ca_ie_offset[1] * offset,
target=cdm_Target(
ca_offset_arcsec=ca_ie_offset[0] * offset,
ie_offset_arcsec=ca_ie_offset[1] * offset,
),
),
)
for ca_ie_offset in ca_ie_offset_pattern
]
configure_requests = [initial_cdm_config] + partial_cdm_config
return configure_requests
def create_mosaic_mid_configure_request(
id_instance_map: ScanData,
) -> list[cdm_ConfigureRequest]:
"""
creates a dictionary mapping PDM scan definition ids to a list of CDM ConfigureRequest
objects for SKA MID telescope with MOSAIC pointing kind
"""
pattern_params = __get_pointing_pattern(id_instance_map.target)
offsets = pattern_params.offsets
initial_cdm_config = create_mid_single_point_configure_request(
id_instance_map=id_instance_map,
)
partial_cdm_config = [
cdm_ConfigureRequest(
interface=initial_cdm_config.interface,
tmc=cdm_TMCConfiguration(
partial_configuration=True,
scan_duration=id_instance_map.scan_definition.scan_duration_ms,
),
pointing=cdm_PointingConfiguration(
groups=[
ReceptorGroup(
trajectory=cdm_FixedTrajectory(
attrs=cdm_FixedTrajectory.Attrs(x=offset.x, y=offset.y)
),
projection=Projection(),
)
]
),
)
for offset in offsets[1:]
]
configure_requests = [initial_cdm_config] + partial_cdm_config
return configure_requests
def __get_pointing_pattern(target: Target) -> PointingPatternParameters:
# As it currently stands, the pointing_pattern.parameters of a target is a list of any/every different
# pattern attempted. This function extracts the parameters of the 'active' pointing pattern
for pattern in target.pointing_pattern.parameters:
if pattern.kind == target.pointing_pattern.active:
return pattern
def __filter_scheduling_block(scheduling_block: SBDefinition) -> dict:
"""
private function to create a dictionary of the necessary information required to create the CDM
ConfigureRequests
"""
# Scan sequence is an ordered list of ScanDefinition identifiers. These
# are string IDs, not the ScanDefinition instances themselves.
# We need the ScanDefinition with matching ID. We could inspect each
# ScanDefinition and return the one with matching ID, or we could do
# as we do here, creating a look-up table and retrieving by key.
# The advantage of this is that we can create the table outside
# the loop, therefore creating it once rather than once per iteration.
data = {
"scan_definitions": {
scan_definition.scan_definition_id: scan_definition
for scan_definition in scheduling_block.scan_definitions
},
# Similarly we will need a look-up table for the Targets as
# the scan definitions contain only the Target IDs
"targets": {target.target_id: target for target in scheduling_block.targets},
"csp_configurations": {
csp_configuration.config_id: csp_configuration
for csp_configuration in scheduling_block.csp_configurations
},
}
if scheduling_block.telescope == TelescopeType.SKA_LOW:
data["runtime_beam_map"] = {
v.subarray_beam_id: i + 1
for i, v in enumerate(scheduling_block.mccs_allocation.subarray_beams)
}
return data
def __create_id_instance_map(
data: dict, scan_definition_id: str, scheduling_block: SBDefinition
) -> ScanData:
"""
private function to filter the data dictionary for a specific scan definition
"""
scan_definition = data["scan_definitions"][scan_definition_id]
csp_configuration = data["csp_configurations"][
scan_definition.csp_configuration_ref
]
return ScanData(
scan_definition=data["scan_definitions"][scan_definition_id],
target=data["targets"][scan_definition.target_ref],
csp_configuration=csp_configuration,
mccs_allocation=scheduling_block.mccs_allocation,
sdp_configuration=scheduling_block.sdp_configuration,
)