"""
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 List, Mapping
from astropy import units as units
from astropy.coordinates import SkyCoord
from ska_oso_pdm.entities.common.target import HorizontalCoordinates, Target
from ska_oso_pdm.entities.mccs.mccs_allocation import MCCSAllocation, SubarrayBeamID
from ska_oso_pdm.entities.mccs.subarray_beam_configuration import (
SubarrayBeamConfiguration,
)
from ska_oso_pdm.entities.mccs.target_beam_configuration import TargetBeamConfiguration
from ska_tmc_cdm.messages.central_node.mccs import MCCSAllocate as cdm_MCCSAllocate
from ska_tmc_cdm.messages.subarray_node.configure.mccs import (
MCCSConfiguration as cdm_MCCSConfiguration,
)
from ska_tmc_cdm.messages.subarray_node.configure.mccs import (
StnConfiguration as cdm_MCCS_StnConfiguration,
)
from ska_tmc_cdm.messages.subarray_node.configure.mccs import (
SubarrayBeamConfiguration as cdm_MCCS_SubarrayBeamConfiguration,
)
from ska_tmc_cdm.messages.subarray_node.configure.mccs import (
SubarrayBeamTarget as cdm_MCCS_SubarrayBeamTarget,
)
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__ = [
"convert_mccsallocation",
"convert_mccs_configuration",
]
[docs]def convert_mccsallocation(
mccsallocation: MCCSAllocation,
subarray_beam_map: Mapping[SubarrayBeamID, int],
union: bool = False,
) -> List[cdm_MCCSAllocate]:
"""
Convert a PDM Low MCCSAllocation to a list of CDM MCCSAllocate instances,
one CDM instance per beam.
At the time of writing (PI10), MCCS requires subarray beams to be
allocated one at a time, that is, one subarray beam per allocation
request. This behaviour can be toggled by setting the union argument. When
union is set to True, this function returns a list of CDM instances, each
instance narrowed to configure a single beam.
:param mccsallocation: The PDM MCCSAllocation
:param subarray_beam_map: mapping of offline beam IDs to online beam IDs
:param union: True to create one multi-beam request, False to create n
per-beam requests
:return: equivalent CDM instances
"""
if not isinstance(mccsallocation, MCCSAllocation):
raise TypeError(f"Expected PDM MCCSAllocation, got {type(mccsallocation)}")
def to_runtime_ids(pdm_ids: List[SubarrayBeamID]):
return [subarray_beam_map[i] for i in pdm_ids]
if union:
return [
cdm_MCCSAllocate(
station_ids=mccsallocation.station_ids,
channel_blocks=mccsallocation.channel_blocks,
subarray_beam_ids=to_runtime_ids(mccsallocation.subarray_beam_ids),
)
]
else:
return [
cdm_MCCSAllocate(
station_ids=[i],
channel_blocks=[j],
subarray_beam_ids=to_runtime_ids([k]),
)
for i, j, k in zip(
mccsallocation.station_ids,
mccsallocation.channel_blocks,
mccsallocation.subarray_beam_ids,
)
]
[docs]def convert_mccs_configuration(
mccs_allocation: MCCSAllocation,
target_beam_configurations: List[TargetBeamConfiguration],
targets: List[Target],
subarray_beam_configurations: List[SubarrayBeamConfiguration],
subarray_beam_map: Mapping[SubarrayBeamID, int],
) -> cdm_MCCSConfiguration:
"""
Convert PDM Low SB TargetBeamConfiguration list to a CDM MCCSConfiguration.
Other SB elements required are the Target list and
SubarrayBeamConfiguration list, which are referenced by the
TargetBeamConfigurations.
The MCCSAllocation is also needed for its list of station_ids.
:param mccs_allocation: The PDM MCCSAllocation
:param target_beam_configurations: The PDM TargetBeamConfiguration list
:param targets: The PDM Target list
:param subarray_beam_configurations: The PDM SubarrayBeamConfiguration list
:param subarray_beam_map: mapping of offline beam IDs to online beam IDs
:return: the required CDM MCCSConfiguration
"""
if not isinstance(mccs_allocation, MCCSAllocation):
raise TypeError(f"Expected PDM MCCSAllocation, got {type(mccs_allocation)}")
if not isinstance(target_beam_configurations, list):
raise TypeError(f"Expected List, got {type(target_beam_configurations)}")
if not isinstance(targets, list):
raise TypeError(f"Expected List, got {type(targets)}")
if not isinstance(subarray_beam_configurations, list):
raise TypeError(f"Expected List, got {type(subarray_beam_configurations)}")
# station ids are a list of lists, assuming here that the full list
# of station ids is just the union of the tuples. This needs clarification.
station_ids = {
station_id
for inner_list in mccs_allocation.station_ids
for station_id in inner_list
}
station_configs = [cdm_MCCS_StnConfiguration(i) for i in station_ids]
# create some convenience mappings of object IDs to objects. The target
# and subarray beam configurations referenced by the target beam config
# will be retrieved from these dicts, naturally raising a KeyError if the
# references cannot be found.
targets_by_id = {target.target_id: target for target in targets}
beam_configs_by_id = {
o.subarray_beam_configuration_id: o for o in subarray_beam_configurations
}
subarray_beam_configs = [
convert_mccs_subarraybeamconfiguration(
mccs_allocation=mccs_allocation,
target=targets_by_id[tbc.target],
subarray_beam_configuration=beam_configs_by_id[
tbc.subarray_beam_configuration
],
subarray_beam_map=subarray_beam_map,
)
for tbc in target_beam_configurations
]
return cdm_MCCSConfiguration(
station_configs=station_configs,
subarray_beam_configs=subarray_beam_configs,
)
def convert_mccs_subarraybeamconfiguration(
mccs_allocation: MCCSAllocation,
target: Target,
subarray_beam_configuration: SubarrayBeamConfiguration,
subarray_beam_map: Mapping[SubarrayBeamID, int],
) -> cdm_MCCS_SubarrayBeamConfiguration:
"""
Convert PDM Low SB elements to a CDM MCCSSubarrayBeamConfiguration.
"""
if not isinstance(mccs_allocation, MCCSAllocation):
raise TypeError(f"Expected PDM MCCSAllocation, got {type(mccs_allocation)}")
if not isinstance(target, Target):
raise TypeError(f"Expected PDM Target, got {type(target)}")
if not isinstance(subarray_beam_configuration, SubarrayBeamConfiguration):
raise TypeError(
"Expected PDM Low SubarrayBeamConfiguration, "
f"got {type(subarray_beam_configuration)}"
)
pdm_beam_id = subarray_beam_configuration.subarray_beam_id
cdm_beam_id = subarray_beam_map[pdm_beam_id]
# map PDM subarray beam ID to station IDs for that beam. This assumes that
# only stations that contribute to the targeted subarray beam are to be
# configured.
beam_id_to_stations = {
j: k
for j, k in zip(mccs_allocation.subarray_beam_ids, mccs_allocation.station_ids)
}
stations_for_beam = beam_id_to_stations[pdm_beam_id]
return cdm_MCCS_SubarrayBeamConfiguration(
subarray_beam_id=cdm_beam_id,
station_ids=stations_for_beam,
channels=subarray_beam_configuration.channels,
update_rate=subarray_beam_configuration.update_rate,
target=convert_subarraybeamtarget(target),
antenna_weights=subarray_beam_configuration.antenna_weights,
phase_centre=subarray_beam_configuration.phase_centre,
)
def convert_subarraybeamtarget(target: Target) -> cdm_MCCS_SubarrayBeamTarget:
"""
Convert a PDM Target to a CDM MCCS SubarrayBeamTarget.
This is for use in LOW, where the only acceptable target frame is 'altaz'.
"""
if not isinstance(target, Target):
raise TypeError(f"Expected PDM Target, got {type(target)}")
if not isinstance(target.reference_coordinate, HorizontalCoordinates):
raise NotImplementedError(
f"Expected PDM HorizontalCoordinates, got {type(target.reference_coordinate)}"
)
name = target.target_id
coord: SkyCoord = target.reference_coordinate.coord
frame = coord.frame.name
az = coord.az.to(units.deg).value
el = coord.alt.to(units.deg).value
return cdm_MCCS_SubarrayBeamTarget(
az=az, el=el, target_name=name, reference_frame=frame
)