Source code for ska_oso_scripting.functions.pdm_transforms.wrapper
"""
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 typing import Dict, List
from ska_oso_pdm.entities.common.sb_definition import SBDefinition, TelescopeType
from ska_oso_pdm.entities.common.target import FivePointParameters
from ska_tmc_cdm.messages.central_node.assign_resources import (
AssignResourcesRequest as cdm_AssignResourcesRequest,
)
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
from .dish import (
convert_dishallocation,
convert_dishconfiguration,
convert_pointingconfiguration,
)
from .mccs import convert_mccs_configuration, convert_mccsallocation
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,
) -> List[cdm_AssignResourcesRequest]:
"""
creates a list of CDM AssignResourcesRequest object from a Scheduling Block
"""
if scheduling_block.telescope == TelescopeType.MID:
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_allocation_request = [
cdm_AssignResourcesRequest(
subarray_id=subarray_id,
dish_allocation=cdm_dish,
sdp_config=cdm_sdp_config,
)
]
else:
data = __filter_scheduling_block(scheduling_block)
cdm_allocations = convert_mccsallocation(
mccsallocation=scheduling_block.mccs_allocation,
subarray_beam_map=data["runtime_beam_map"],
union=False,
)
cdm_allocation_request = [
cdm_AssignResourcesRequest(subarray_id=subarray_id, mccs=cdm_allocation)
for cdm_allocation in cdm_allocations
]
return cdm_allocation_request
[docs]def create_cdm_configure_request_from_scheduling_block(
scheduling_block: SBDefinition,
) -> Dict:
"""
creates a dictionary mapping PDM scan definition ids to a list of CDM ConfigureRequest
objects
"""
data = __filter_scheduling_block(scheduling_block)
configure_request = {}
for scan_definition_id in data["scan_definitions"]:
# The target is referenced by ID in the scan definition
scan_definition = data["scan_definitions"][scan_definition_id]
target = data["targets"][scan_definition.target_id]
csp_configuration = data["csp_configurations"][
scan_definition.csp_configuration_id
]
# The initial CDM ConfigureRequest is identical to a SinglePoint request. As we currently only support
# SinglePoint or FivePoint Scans for now, this request is created outside the factory loop. As we extend
# support to include RasterScans etc., the following code will likely have to be embedded or a separate
# create_ function that deals with scans (rather than pointings) will be written.
if scheduling_block.telescope == TelescopeType.MID:
if scan_definition.dish_configuration_id in data["dish_configurations"]:
dish_configuration = data["dish_configurations"][
scan_definition.dish_configuration_id
]
initial_cdm_config: cdm_ConfigureRequest = cdm_ConfigureRequest(
dish=convert_dishconfiguration(dish_configuration),
pointing=convert_pointingconfiguration(target),
csp=convert_cspconfiguration(
csp_configuration, dish_configuration.receiver_band
),
)
else:
initial_cdm_config: cdm_ConfigureRequest = cdm_ConfigureRequest(
mccs=convert_mccs_configuration(
mccs_allocation=scheduling_block.mccs_allocation,
target_beam_configurations=scheduling_block.target_beam_configurations,
targets=scheduling_block.targets,
subarray_beam_configurations=scheduling_block.subarray_beam_configurations,
subarray_beam_map=data["runtime_beam_map"],
),
csp=convert_cspconfiguration(csp_configuration),
)
initial_cdm_config.tmc = convert_tmcconfiguration(scan_definition)
initial_cdm_config.sdp = convert_sdpconfiguration_subarraynode(scan_definition)
if (
scheduling_block.telescope == TelescopeType.MID
and target.pointing_pattern.active == "SinglePointParameters"
) or (scheduling_block.telescope == TelescopeType.LOW):
# TODO: create a factory that can be extended when support for more "Point" type patterns:
configure_request[scan_definition_id] = [initial_cdm_config]
elif (
scheduling_block.telescope == TelescopeType.MID
and target.pointing_pattern.active == "FivePointParameters"
):
ca_ie_offset_pattern = [(0, 1), (0, -1), (1, 0), (-1, 0)]
# the SB stores the pattern parameters of all patterns attempted - not just active
# extracting the information we need
offset = next(
p.offset_arcsec
for p in target.pointing_pattern.parameters
if type(p) is FivePointParameters
)
partial_cdm_config = [
cdm_ConfigureRequest(
interface=initial_cdm_config.interface,
tmc=cdm_TMCConfiguration(partial_configuration=True),
pointing=cdm_PointingConfiguration(
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
configure_request[scan_definition_id] = configure_requests
return configure_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.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: i + 1
for i, v in enumerate(scheduling_block.mccs_allocation.subarray_beam_ids)
}
return data