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 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 (
from .mccs import convert_mccs_configuration, convert_mccsallocation
from .sdp import (

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__ = [

[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)"Setting dish : {cdm_dish.receptor_ids} ") cdm_sdp_config = convert_sdpconfiguration_centralnode( scheduling_block.sdp_configuration, scheduling_block.targets ) 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 == "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 == "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