import copy
from ska_telmodel.csp.low_version import (
get_csp_subsystem_version,
lowcsp_assignresources_uri,
lowcsp_configure_uri,
lowcsp_releaseresources_uri,
lowcsp_scan_uri,
)
from .._common import split_interface_version
from ..lowcbf import get_lowcbf_configurescan_example
from ..pss.example import (
get_pss_cheetah_configure_example,
get_pss_configure_example,
)
from ..pst.examples import (
get_pst_config_example,
)
LOW_CSP_ASSIGNRESOURCES_2_0 = {
"interface": "https://schema.skao.int/ska-low-csp-assignresources/2.0",
"common": {"subarray_id": 1},
"lowcbf": {
"resources": [
{
"device": "fsp_01",
"shared": True,
"fw_image": "pst",
"fw_mode": "unused",
},
{
"device": "p4_01",
"shared": True,
"fw_image": "p4.bin",
"fw_mode": "p4",
},
]
},
"pss": {"beams_id": [1, 2, 3]},
"pst": {"beams_id": [1]},
}
LOW_CSP_ASSIGNRESOURCES_3_0 = {
"interface": "https://schema.skao.int/ska-low-csp-assignresources/3.0",
"common": {"subarray_id": 1},
"lowcbf": {},
"pss": {"beams_id": [1, 2, 3]},
"pst": {"beams_id": [1]},
}
LOW_CSP_ASSIGNRESOURCES_3_2 = {
"interface": "https://schema.skao.int/ska-low-csp-assignresources/3.2",
"transaction_id": "txn-....-00001",
"common": {"subarray_id": 1},
"lowcbf": {},
"pss": {"beams_id": [1, 2, 3]},
"pst": {"beams_id": [1]},
}
LOW_CSP_ASSIGNRESOURCES_6_0 = {
"interface": "https://schema.skao.int/ska-low-csp-assignresources/6.0",
"transaction_id": "txn-....-00001",
"common": {"subarray_id": 1},
"lowcbf": {},
"pst": {"pst_beam_ids": [1, 2, 3]},
"pss": {"pss_beam_ids": [1]},
}
LOW_CSP_CONFIGURE_2_0 = {
"interface": "https://schema.skao.int/ska-low-csp-configure/2.0",
"subarray": {"subarray_name": "science period 23"},
"common": {
"config_id": "sbi-mvp01-20200325-00001-science_A",
"subarray_id": 1,
"eb_id": "eb-x449-20231105-34696",
},
"lowcbf": {
"stations": {
"stns": [[1, 0], [2, 0], [3, 0], [4, 0]],
"stn_beams": [
{
"beam_id": 1,
"freq_ids": [64, 65, 66, 67, 68, 69, 70, 71],
"boresight_dly_poly": (
"tango://delays.skao.int/low/stn-beam/1"
),
}
],
},
},
"pss": {},
"pst": {"beams": []},
}
LOW_CSP_CONFIGURE_3_0 = {
"interface": "https://schema.skao.int/ska-low-csp-configure/3.0",
"subarray": {"subarray_name": "science period 23"},
"common": {
"config_id": "sbi-mvp01-20200325-00001-science_A",
"subarray_id": 1,
"eb_id": "eb-x449-20231105-34696",
},
"lowcbf": {
"stations": {
"stns": [[1, 1], [2, 1], [3, 1], [4, 1], [5, 1], [6, 1]],
"stn_beams": [
{
"beam_id": 1,
"freq_ids": [400],
"delay_poly": "tango://delays.skao.int/low/stn-beam/1",
}
],
},
"vis": {
"fsp": {"firmware": "vis", "fsp_ids": [1]},
"stn_beams": [
{
"stn_beam_id": 1,
"host": [[0, "192.168.0.1"]],
"port": [[0, 9000, 1]],
"mac": [[0, "02-03-04-0a-0b-0c"]],
"integration_ms": 849,
}
],
},
},
"pss": {},
"pst": {"beams": []},
}
LOW_CSP_CONFIGURE_3_1 = {
"interface": "https://schema.skao.int/ska-low-csp-configure/3.1",
"subarray": {"subarray_name": "science period 23"},
"common": {
"config_id": "sbi-mvp01-20200325-00001-science_A",
"subarray_id": 1,
"eb_id": "eb-x449-20231105-34696",
},
"lowcbf": {
"stations": {
"stns": [[1, 1], [2, 1], [3, 1], [4, 1], [5, 1], [6, 1]],
"stn_beams": [
{
"beam_id": 1,
"freq_ids": [400],
"delay_poly": "tango://delays.skao.int/low/stn-beam/1",
}
],
},
"vis": {
"fsp": {"firmware": "vis", "fsp_ids": [1]},
"stn_beams": [
{
"stn_beam_id": 1,
"host": [[0, "192.168.0.1"]],
"port": [[0, 9000, 1]],
"mac": [[0, "02-03-04-0a-0b-0c"]],
"integration_ms": 849,
}
],
},
},
"pss": {},
"pst": {"beams": []},
}
LOW_CSP_CONFIGURE_3_2 = {
"interface": "https://schema.skao.int/ska-low-csp-configure/3.2",
"transaction_id": "txn-....-00001",
"subarray": {"subarray_name": "science period 23"},
"common": {
"config_id": "sbi-mvp01-20200325-00001-science_A",
"subarray_id": 1,
"eb_id": "eb-x449-20231105-34696",
},
"lowcbf": {
"stations": {
"stns": [[1, 1], [2, 1], [3, 1], [4, 1], [5, 1], [6, 1]],
"stn_beams": [
{
"beam_id": 1,
"freq_ids": [400],
"delay_poly": "tango://delays.skao.int/low/stn-beam/1",
}
],
},
"vis": {
"fsp": {"firmware": "vis", "fsp_ids": [1]},
"stn_beams": [
{
"stn_beam_id": 1,
"host": [[0, "192.168.0.1"]],
"port": [[0, 9000, 1]],
"mac": [[0, "02-03-04-0a-0b-0c"]],
"integration_ms": 849,
}
],
},
},
}
LOW_CSP_CONFIGURE_4_0 = {
"interface": "https://schema.skao.int/ska-low-csp-configure/4.0",
"subarray": {"subarray_name": "science period 23"},
"common": {
"config_id": "sbi-mvp01-20200325-00001-science_A",
"subarray_id": 1,
"eb_id": "eb-x449-20231105-34696",
},
"lowcbf": {
"stations": {
"stns": [[1, 1], [2, 1], [3, 1], [4, 1], [5, 1], [6, 1]],
"stn_beams": [
{
"stn_beam_id": 1,
"freq_ids": [400],
"delay_poly": "tango://delays.skao.int/low/stn-beam/1",
}
],
},
"vis": {
"fsp": {"function_mode": "vis", "fsp_ids": [1]},
"stn_beams": [
{
"stn_beam_id": 1,
"host": [[0, "192.168.0.1"]],
"port": [[0, 9000, 1]],
"mac": [[0, "02-03-04-0a-0b-0c"]],
"integration_ms": 849,
}
],
},
},
"pss": {},
"pst": {"beams": []},
}
LOW_CSP_CONFIGURE_5_0 = {
"interface": "https://schema.skao.int/ska-low-csp-configure/5.0",
"transaction_id": "txn-....-00001",
"subarray": {"subarray_name": "science period 23"},
"common": {
"config_id": "sbi-mvp01-20200325-00001-science_A",
"subarray_id": 1,
"eb_id": "eb-x449-20231105-34696",
},
}
LOW_CSP_RELEASERESOURCES_2_0 = {
"interface": "https://schema.skao.int/ska-low-csp-releaseresources/2.0",
"common": {"subarray_id": 1},
"lowcbf": {
"resources": [
{"device": "fsp_01"},
]
},
"pst": {"beams_id": [1]},
}
LOW_CSP_RELEASERESOURCES_3_0 = {
"interface": "https://schema.skao.int/ska-low-csp-releaseresources/3.0",
"common": {"subarray_id": 1},
"lowcbf": {},
"pst": {"beams_id": [1]},
}
LOW_CSP_RELEASERESOURCES_3_2 = {
"interface": "https://schema.skao.int/ska-low-csp-releaseresources/3.2",
"transaction_id": "txn-....-00001",
"common": {"subarray_id": 1},
"lowcbf": {},
"pst": {"beams_id": [1]},
}
LOW_CSP_RELEASERESOURCES_6_0 = {
"interface": "https://schema.skao.int/ska-low-csp-releaseresources/6.0",
"transaction_id": "txn-....-00001",
"common": {"subarray_id": 1},
"lowcbf": {},
"pst": {"pst_beam_ids": [1]},
"pss": {"pss_beam_ids": [1]},
}
LOW_CSP_SCAN_2_0 = {
"interface": "https://schema.skao.int/ska-low-csp-scan/2.0",
"common": {"subarray_id": 1},
"lowcbf": {"scan_id": 987654321},
}
LOW_CSP_SCAN_3_2 = {
"interface": "https://schema.skao.int/ska-low-csp-scan/3.2",
"transaction_id": "txn-....-00001",
"common": {"subarray_id": 1},
"lowcbf": {"scan_id": 987654321},
}
LOW_CSP_SCAN_4_0 = {
"interface": "https://schema.skao.int/ska-low-csp-scan/4.0",
"transaction_id": "txn-....-00001",
"common": {"subarray_id": 1},
"scan_id": 987654321,
}
# Functions returning copies of the appropriate version of LOW CSP JSON strings
[docs]
def get_low_csp_assignresources_example(version: str):
"""Get appropriate version example of LOW CSP assign resources string.
:param version: Version URI of configuration format
"""
if version.startswith(lowcsp_assignresources_uri(2, 0)):
return copy.deepcopy(LOW_CSP_ASSIGNRESOURCES_2_0)
elif version.startswith(lowcsp_assignresources_uri(3, 0)):
return copy.deepcopy(LOW_CSP_ASSIGNRESOURCES_3_0)
elif version.startswith(lowcsp_assignresources_uri(3, 1)):
return copy.deepcopy(LOW_CSP_ASSIGNRESOURCES_3_0)
elif version.startswith(lowcsp_assignresources_uri(3, 2)):
return copy.deepcopy(LOW_CSP_ASSIGNRESOURCES_3_2)
elif version.startswith(lowcsp_assignresources_uri(4, 0)):
example = update_interface(LOW_CSP_ASSIGNRESOURCES_3_2, "4.0")
return copy.deepcopy(example)
elif version.startswith(lowcsp_assignresources_uri(5, 0)):
example = update_interface(LOW_CSP_ASSIGNRESOURCES_3_2, "5.0")
return copy.deepcopy(example)
elif version.startswith(lowcsp_assignresources_uri(6, 0)):
return copy.deepcopy(LOW_CSP_ASSIGNRESOURCES_6_0)
elif version.startswith(lowcsp_assignresources_uri(7, 0)):
example = update_interface(LOW_CSP_ASSIGNRESOURCES_6_0, "7.0")
return copy.deepcopy(example)
elif version.startswith(lowcsp_assignresources_uri(7, 1)):
example = update_interface(LOW_CSP_ASSIGNRESOURCES_6_0, "7.1")
return copy.deepcopy(example)
elif version.startswith(lowcsp_assignresources_uri(7, 2)):
example = update_interface(LOW_CSP_ASSIGNRESOURCES_6_0, "7.2")
return copy.deepcopy(example)
raise ValueError(f"Could not generate example for schema {version})!")
[docs]
def update_interface(json_example: dict, desired_version: str):
"""Update the version of the example interface."""
old_interface = json_example["interface"]
subsystem_uri = "/".join(old_interface.rsplit("/", 1)[:-1]) + "/"
new_interface = subsystem_uri + desired_version
json_example["interface"] = new_interface
return json_example
[docs]
def update_pss_config_example(json_example):
"""Remove PSS example attributes that are not needed in the csp example."""
if "pss" in json_example:
if "interface" in json_example["pss"]:
del json_example["pss"]["interface"]
if "common" in json_example["pss"]:
del json_example["pss"]["common"]
[docs]
def update_config_cbf_keys(csp_version, json_example, subsystem, data) -> dict:
"""Add/Remove CBF example attributes that differ between base CSP
example."""
if "pst" in subsystem:
if csp_version == (2, 0):
json_example["lowcbf"]["timing_beams"] = {
"beams": [
{
"pst_beam_id": 13,
"stn_beam_id": 1,
"offset_dly_poly": "url",
"stn_weights": [0.9, 1.0, 1.0, 0.9],
"jones": "url",
"rfi_enable": [True, True, True],
"rfi_static_chans": [1, 206, 997],
"rfi_dynamic_chans": [242, 1342],
"rfi_weighted": 0.87,
}
]
}
elif csp_version > (2, 0):
if "destinations" in data["beams"][0]:
data["beams"][0].pop("destinations", None)
json_example["lowcbf"]["timing_beams"] = data
elif "pss" in subsystem:
for beam in data.get("beams", []):
if "destinations" in beam:
beam.pop("destinations")
json_example["lowcbf"]["search_beams"] = data
[docs]
def update_config_pst_keys(csp_version, json_example, pst_example) -> dict:
"""Add/Remove PST schema attributes that differ between PST and CSP
schema."""
# remove common part and interface from PST example
pst_beam_example = pst_example["pst"]["scan"]
# remove "timing_beam_id" attributes
del pst_beam_example["timing_beam_id"]
# Add one or more PST beam according to the CBF schema
pst_beam_section = []
pst_beam_dict = {}
for beam in json_example["lowcbf"]["timing_beams"]["beams"]:
# aligns CBF beam_ids with the one in the PST
pst_beam_dict = {
"beam_id": beam["pst_beam_id"],
"scan": pst_beam_example,
}
pst_beam_section.append(pst_beam_dict)
# add list of pst beams
json_example["pst"] = {"beams": pst_beam_section}
[docs]
def get_low_csp_releaseresources_example(version: str):
"""Get appropriate version example of LOW CSP release resources string.
:param version: Version URI of configuration format
"""
if version.startswith(lowcsp_releaseresources_uri(2, 0)):
return copy.deepcopy(LOW_CSP_RELEASERESOURCES_2_0)
elif version.startswith(lowcsp_releaseresources_uri(3, 0)):
return copy.deepcopy(LOW_CSP_RELEASERESOURCES_3_0)
elif version.startswith(lowcsp_releaseresources_uri(3, 1)):
return copy.deepcopy(LOW_CSP_RELEASERESOURCES_3_0)
elif version.startswith(lowcsp_releaseresources_uri(3, 2)):
return copy.deepcopy(LOW_CSP_RELEASERESOURCES_3_2)
elif version.startswith(lowcsp_releaseresources_uri(4, 0)):
example = update_interface(LOW_CSP_RELEASERESOURCES_3_2, "4.0")
return copy.deepcopy(example)
elif version.startswith(lowcsp_releaseresources_uri(5, 0)):
example = update_interface(LOW_CSP_RELEASERESOURCES_3_2, "5.0")
return copy.deepcopy(example)
elif version.startswith(lowcsp_releaseresources_uri(6, 0)):
return copy.deepcopy(LOW_CSP_RELEASERESOURCES_6_0)
elif version.startswith(lowcsp_releaseresources_uri(7, 0)):
example = update_interface(LOW_CSP_RELEASERESOURCES_6_0, "7.0")
return copy.deepcopy(example)
elif version.startswith(lowcsp_releaseresources_uri(7, 1)):
example = update_interface(LOW_CSP_RELEASERESOURCES_6_0, "7.1")
return copy.deepcopy(example)
elif version.startswith(lowcsp_releaseresources_uri(7, 2)):
example = update_interface(LOW_CSP_RELEASERESOURCES_6_0, "7.2")
return copy.deepcopy(example)
raise ValueError(f"Could not generate example for schema {version})!")
[docs]
def get_low_csp_scan_example(version: str):
"""Get appropriate version example of LOW CSP scan string.
:param version: Version URI of configuration format
"""
if version.startswith(lowcsp_scan_uri(2, 0)):
return copy.deepcopy(LOW_CSP_SCAN_2_0)
elif version.startswith(lowcsp_scan_uri(3, 0)):
return copy.deepcopy(LOW_CSP_SCAN_2_0)
elif version.startswith(lowcsp_scan_uri(3, 1)):
return copy.deepcopy(LOW_CSP_SCAN_2_0)
elif version.startswith(lowcsp_scan_uri(3, 2)):
return copy.deepcopy(LOW_CSP_SCAN_3_2)
elif version.startswith(lowcsp_scan_uri(4, 0)):
return copy.deepcopy(LOW_CSP_SCAN_4_0)
elif version.startswith(lowcsp_scan_uri(5, 0)):
example = update_interface(LOW_CSP_SCAN_4_0, "5.0")
return copy.deepcopy(example)
elif version.startswith(lowcsp_scan_uri(6, 0)):
example = update_interface(LOW_CSP_SCAN_4_0, "6.0")
return copy.deepcopy(example)
elif version.startswith(lowcsp_scan_uri(7, 0)):
example = update_interface(LOW_CSP_SCAN_4_0, "7.0")
return copy.deepcopy(example)
elif version.startswith(lowcsp_scan_uri(7, 1)):
example = update_interface(LOW_CSP_SCAN_4_0, "7.1")
return copy.deepcopy(example)
elif version.startswith(lowcsp_scan_uri(7, 2)):
example = update_interface(LOW_CSP_SCAN_4_0, "7.2")
return copy.deepcopy(example)
raise ValueError(f"Could not generate example for schema {version})!")