import logging
from typing import Dict, Optional, Sequence, Union
import ska_sdp_config
from realtime.receive.core.antenna_utils import load_antennas
from realtime.receive.core.baselines import Baselines
from realtime.receive.core.common import load_json_resource
from ska_telmodel.data.frontend import TMData
from realtime.receive.modules.tm.base_tm import TelescopeManager
logger = logging.getLogger(__name__)
[docs]
class SKATelescopeManager(TelescopeManager):
"""
TelescopeManager class that combines information from an AssignResources
command and an Antenna Layout to assemble the list of antennas making up
a subarray.
"""
def __init__(
self,
antenna_layout: Union[str, Sequence[Dict]],
assign_resources_command: Optional[Union[str, dict]] = None,
):
all_antennas = load_antennas(antenna_layout)
if isinstance(assign_resources_command, str):
assign_resources_command = load_json_resource(assign_resources_command)
if assign_resources_command is not None:
indexed_antennas = {antenna.label: antenna for antenna in all_antennas}
antenna_names = assign_resources_command["resources"]["receptors"]
antennas = [indexed_antennas[name] for name in antenna_names]
else:
antennas = all_antennas
logger.info(
"Build SKATelescopeModel with %d antennas: %r",
len(antennas),
antennas,
)
# TODO (rtobar): For the time being we assume Low's baseline ordering,
# which is well defined in the CBF SDP ICD. Mid's order isn't set in
# stone, and might change to match Low's.
baselines = Baselines.generate(len(antennas), autocorr=True, lower_triangular=True)
super().__init__(antennas, baselines)
[docs]
@staticmethod
def from_sdp_config(
execution_block_id: str,
sdp_config: ska_sdp_config.Config,
telmodel_key: str | None = None,
telmodel_source_uris: list[str] | None = None,
antenna_layout: str | Sequence[dict] | None = None,
):
"""
Construct a telescope manager for the given options, contacting the
SDP Configuration Database and using the SKA Telmodel package to obtain
all necessary information.
:param execution_block_id: The Execution Block ID for which a TM needs
to be created.
:param sdp_config_db: A client to the SDP Configuration DB.
:param telmodel_key: The key in the SKA Telmodel data where the antenna
layout can be read from. If not given, then `antenna_layout` must be
given.
:param telmodel_source_uris: An optional list of URIs used by the SKA
Telmodel package to read its data from.
:param antenna_layout: An antenna layout, either a a list of
dictionaries or as a URL/filename with a JSON representation of such
list. Used only if `telmodel_key` is *not* given.
"""
if not antenna_layout:
tmdata = TMData(telmodel_source_uris)
antenna_layout = tmdata.get(telmodel_key).get_dict()["receptors"]
for txn in sdp_config.txn():
execution_block = txn.get_execution_block(execution_block_id)
if not execution_block:
raise ValueError(f"No EB in SDP config with id='{execution_block_id}'")
return SKATelescopeManager(antenna_layout, execution_block)