"""
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._shared import TelescopeType
from ska_oso_pdm._shared.target import FivePointParameters, PointingKind
from ska_oso_pdm.sb_definition import SBDefinition
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 LOW_SCHEMA
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.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_dishconfiguration,
convert_pointingconfiguration,
)
from .mccs import convert_mccs_allocation, convert_mccs_configuration
from .sdp import (
convert_sdpconfiguration_centralnode,
convert_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",
]
[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} ")
cdm_sdp_config = convert_sdpconfiguration_centralnode(
scheduling_block.sdp_configuration, scheduling_block.targets
)
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 = convert_mccs_allocation(
allocation=scheduling_block.mccs_allocation
)
cdm_pst_config = convert_pstconfiguration_centralnode(scheduling_block.targets)
cdm_sdp_config = convert_sdpconfiguration_centralnode(
scheduling_block.sdp_configuration, scheduling_block.targets
)
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 __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_MID:
# ... same for dish configurations..
data["dish_configurations"] = {
dish_configuration.dish_configuration_id: dish_configuration
for dish_configuration in scheduling_block.dish_configurations
}
else:
data["runtime_beam_map"] = {
v.subarray_beam_id: i + 1
for i, v in enumerate(scheduling_block.mccs_allocation.subarray_beams)
}
return data