"""
PSS configuration examples.
"""
import copy
from ska_telmodel._common import lookup_example
from . import cheetah_examples
from .version import pss_configure_uri
PSS_COMMON_0_0 = {
"common": {
"config_id": "sbi-mvp01-20200325-00001-science_A",
"subarray_id": 1,
"frequency_band": "low",
"eb_id": "eb-x449-20231105-34696",
},
}
def get_pss_cheetah_configure_example(version: str, *args):
return lookup_example(
cheetah_examples, version, "get_pss_cheetah_config_example"
)
[docs]
def get_pss_config_scan_0_1_example(version: str) -> dict:
"""Get the example directly from pss cheetah example."""
cheetah_version = version.replace("pss", "pss-cheetah")
pss_config_0_1 = get_pss_cheetah_configure_example(cheetah_version)
return copy.deepcopy(pss_config_0_1)
def get_pss_config_scan_1_0_example(version: str) -> dict:
pss_config_1_0 = get_pss_config_scan_0_1_example(version)
return copy.deepcopy(pss_config_1_0)
def get_pss_config_scan_1_1_example(version: str) -> dict:
pss_config_1_1 = get_pss_config_scan_0_1_example(version)
return copy.deepcopy(pss_config_1_1)
def get_pss_config_scan_1_2_example(version: str) -> dict:
# PSS LMC version 1.2 use cheetah version 1.1
cheetah_version = version
cheetah_version = cheetah_version.replace("2", "1")
pss_config_1_2 = copy.deepcopy(
get_pss_config_scan_0_1_example(cheetah_version)
)
pss_config_1_2 = _update_pss_configure_example(pss_config_1_2)
pss_config_1_2["interface"] = version
return copy.deepcopy(pss_config_1_2)
def _update_pss_configure_example(
pss_base_example: dict, active_beam: int = 2, cheetah_number: int = 1
) -> dict:
"""Update the PSS CONFIGURE example to adapt the cheetah schema
to the PSS subarray one.
:param pss_base_example: dictionary PSS example to be modified
:param active_beam: number of beams that have to be marked as active
:param cheetah_number: number of Cheetah in the example
"""
beam_per_cheetah = 3
if (
active_beam <= 0
or cheetah_number <= 0
or active_beam > beam_per_cheetah * cheetah_number
):
raise ValueError(
f"The desired active_beam {active_beam} and/or "
f"the desired cheetah_number {cheetah_number} is not allowed"
)
# add COMMON schema to the PSS input schema
pss_base_example.update(copy.deepcopy(PSS_COMMON_0_0))
# add transaction id
pss_base_example["transaction_id"] = "txn-....-00001"
# add cheetah beams inside cheetah attribute
pss_beams = pss_base_example.pop("beams")
cheetah = {"cheetah_id": 1, "beams": copy.deepcopy(pss_beams)}
cheetah_list = []
for cheetah_counter in range(cheetah_number):
cheetah["cheetah_id"] = cheetah_counter + 1
for beam_counter, beam in enumerate(cheetah["beams"]):
# each cheetah contains 3 beams
if (
beam_counter + (cheetah_counter * beam_per_cheetah)
>= active_beam
):
beam["beam"]["active"] = False
else:
beam["beam"]["active"] = True
for sink_el in beam["beam"]["sinks"]["channels"]["sps_events"][
"sink"
]:
# rename sink id
_rename_cheetah_ids(sink_el, "sink_id")
# add source flag (only one should be active at the same time)
beam["beam"]["source"]["sigproc"]["active"] = True
beam["beam"]["source"]["udp_low"]["active"] = False
# remove listen attribute from UDP_LOW source
beam["beam"]["source"]["udp_low"].__delitem__("listen")
# rename sispccl_files id
spccl_el = beam["beam"]["sinks"]["sink_configs"]["spccl_files"]
_rename_cheetah_ids(spccl_el, "sink_id")
spccl_sigproc_el = beam["beam"]["sinks"]["sink_configs"][
"spccl_sigproc_files"
]
_rename_cheetah_ids(spccl_sigproc_el, "sink_id")
# rename Beam id
_rename_cheetah_ids(beam["beam"], "beam_id")
cheetah_list.append(copy.deepcopy(cheetah))
# cheetah i.e. [{ "cheetah_id": 1, "beams":pss_beams},]
pss_base_example["cheetah"] = cheetah_list
# remove id (pipeline) moved inside cheetah attribute
del pss_base_example["id"]
return copy.deepcopy(pss_base_example)
def _rename_cheetah_ids(schema: dict, new_name: str) -> None:
"""Update dictionary renaming id element with a new name.
:param schema: dictionary that contains the id attribute to be modified
:param new_name: attribute new name
"""
attr = schema["id"]
schema[new_name] = attr
del schema["id"]