Source code for

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

import ska_tmc_cdm.messages.subarray_node.configure.csp as cdm
from astropy.coordinates import SkyCoord
from import EquatorialCoordinates, Target
from import (
    DishAllocation as pdm_DishAllocation,
from import (
    DishConfiguration as pdm_DishConfiguration,
from import (
    ReceiverBand as pdm_ReceiverBand,
from ska_tmc_cdm.messages.central_node.common import (
    DishAllocation as cdm_DishAllocation,
from ska_tmc_cdm.messages.subarray_node.configure import (
    DishConfiguration as cdm_DishConfiguration,
from ska_tmc_cdm.messages.subarray_node.configure import (
    PointingConfiguration as cdm_PointingConfiguration,
from ska_tmc_cdm.messages.subarray_node.configure.core import (
    ReceiverBand as cdm_ReceiverBand,
from ska_tmc_cdm.messages.subarray_node.configure.core import Target as cdm_Target

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

def convert_frequency_band(
    pdm_val: pdm_ReceiverBand,
) -> cdm.core.ReceiverBand:
    Converts a PDM receiver band to a CDM receiver band.

    :param pdm_val: PDM receiver band to convert
    :return: CDM receiver band
    return cdm.core.ReceiverBand[]

[docs]def convert_pointingconfiguration( target: Target, ) -> cdm_PointingConfiguration: """ Convert a PDM Target to the equivalent TMC configuration """ if not isinstance(target, Target): raise TypeError(f"Expected PDM Target, got {type(target)}") if not isinstance(target.reference_coordinate, EquatorialCoordinates): raise NotImplementedError( f"Expected PDM EquatorialCoordinate, got {type(target.reference_coordinate)}" ) coord: SkyCoord = target.reference_coordinate.coord raw_ra = coord.ra.value raw_dec = coord.dec.value radec_units = (, frame = name = target.target_id return cdm_PointingConfiguration( target=cdm_Target( ra=raw_ra, dec=raw_dec, unit=radec_units, reference_frame=frame, target_name=name, ) )
[docs]def convert_dishconfiguration( dish_configuration: pdm_DishConfiguration, ) -> cdm_DishConfiguration: """ Convert a PDM Dish configuration to a CDM Dish Configuration """ if not isinstance(dish_configuration, pdm_DishConfiguration): raise TypeError( f"Expected PDM DishConfiguration, got {type(dish_configuration)}" ) pdm_rx = dish_configuration.receiver_band return cdm_DishConfiguration(receiver_band=cdm_ReceiverBand(pdm_rx.value))
[docs]def convert_dishallocation(dish_allocation: pdm_DishAllocation) -> cdm_DishAllocation: """ Convert a PDM DishAllocation to the equivalent CDM DishAllocation. """ return cdm_DishAllocation(receptor_ids=dish_allocation.receptor_ids)