from typing import List, Tuple, Union
from .._common import split_interface_version
CSP_ASSIGNRESOURCES_PREFIX = "https://schema.skao.int/ska-csp-assignresources/"
CSP_CONFIGSCAN_PREFIX = "https://schema.skao.int/ska-csp-configurescan/"
CSP_CONFIG_PREFIX = "https://schema.skao.int/ska-csp-configure/"
CSP_SCAN_PREFIX = "https://schema.skao.int/ska-csp-scan/"
CSP_ENDSCAN_PREFIX = "https://schema.skao.int/ska-csp-endscan/"
CSP_RELEASERESOURCES_PREFIX = (
"https://schema.skao.int/ska-csp-releaseresources/"
)
CSP_DELAYMODEL_PREFIX = "https://schema.skao.int/ska-csp-delaymodel/"
CSP_MID_DELAYMODEL_PREFIX = "https://schema.skao.int/ska-mid-csp-delaymodel/"
CSP_LOW_DELAYMODEL_PREFIX = "https://schema.skao.int/ska-low-csp-delaymodel/"
_ALLOWED_URI_PREFIXES = [
CSP_ASSIGNRESOURCES_PREFIX,
CSP_CONFIG_PREFIX,
CSP_CONFIGSCAN_PREFIX,
CSP_SCAN_PREFIX,
CSP_ENDSCAN_PREFIX,
CSP_RELEASERESOURCES_PREFIX,
CSP_DELAYMODEL_PREFIX,
CSP_MID_DELAYMODEL_PREFIX,
CSP_LOW_DELAYMODEL_PREFIX,
]
CSP_CONFIG_VER0 = CSP_CONFIG_PREFIX + "0"
# ADR-3 Configuring and Scanning,
# ADR-4 Link map CSP configuration
CSP_CONFIG_VER0_0 = CSP_CONFIG_PREFIX + "0.0"
# ADR-10 Revise receive addresses exchange
CSP_CONFIG_VER0_1 = CSP_CONFIG_PREFIX + "0.1"
# ADR-18 Change in CSP configuration
CSP_CONFIG_VER1 = CSP_CONFIG_PREFIX + "1"
CSP_CONFIG_VER1_0 = CSP_CONFIG_PREFIX + "1.0"
# ADR-35 change in CSP Configuration
CSP_CONFIG_VER2_0 = CSP_CONFIG_PREFIX + "2.0"
CSP_CONFIG_VER2 = CSP_CONFIG_PREFIX + "2"
CSP_CONFIG_VER2_1 = CSP_CONFIG_PREFIX + "2.1"
CSP_CONFIG_VER2_2 = CSP_CONFIG_PREFIX + "2.2"
CSP_CONFIG_VER2_3 = CSP_CONFIG_PREFIX + "2.3"
CSP_CONFIG_VER2_4 = CSP_CONFIG_PREFIX + "2.4"
CSP_CONFIG_VER2_5 = CSP_CONFIG_PREFIX + "2.5"
CSP_CONFIG_VER2_6 = CSP_CONFIG_PREFIX + "2.6"
CSP_CONFIG_VER3 = CSP_CONFIGSCAN_PREFIX + "3"
CSP_CONFIG_VER3_0 = CSP_CONFIGSCAN_PREFIX + "3.0"
CSP_CONFIG_VER4 = CSP_CONFIGSCAN_PREFIX + "4"
CSP_CONFIG_VER4_0 = CSP_CONFIGSCAN_PREFIX + "4.0"
CSP_CONFIG_VER4_1 = CSP_CONFIGSCAN_PREFIX + "4.1"
CSP_CONFIG_VER5_0 = CSP_CONFIGSCAN_PREFIX + "5.0"
CSP_CONFIG_VER6_0 = CSP_CONFIGSCAN_PREFIX + "6.0"
CSP_CONFIG_VER6_1 = CSP_CONFIGSCAN_PREFIX + "6.1"
CSP_CONFIG_VER7_0 = CSP_CONFIGSCAN_PREFIX + "7.0"
CSP_CONFIG_VER8_0 = CSP_CONFIGSCAN_PREFIX + "8.0"
CSP_CONFIG_VER8_1 = CSP_CONFIGSCAN_PREFIX + "8.1"
# CSP configuration versions, chronologically sorted
CSP_CONFIG_VERSIONS = sorted(
[
CSP_CONFIG_VER0_0,
CSP_CONFIG_VER0_1,
CSP_CONFIG_VER1_0,
CSP_CONFIG_VER2_0,
CSP_CONFIG_VER2_1,
CSP_CONFIG_VER2_2,
CSP_CONFIG_VER2_3,
CSP_CONFIG_VER2_4,
CSP_CONFIG_VER2_5,
CSP_CONFIG_VER2_6,
CSP_CONFIG_VER3_0,
CSP_CONFIG_VER4_0,
CSP_CONFIG_VER4_1,
CSP_CONFIG_VER5_0,
CSP_CONFIG_VER6_0,
CSP_CONFIG_VER6_1,
CSP_CONFIG_VER7_0,
CSP_CONFIG_VER8_0,
CSP_CONFIG_VER8_1,
],
key=split_interface_version,
)
MAX_MID_PSS_BEAMS = 1125
[docs]
def csp_config_versions(min_ver=None, max_ver=None):
"""
Returns a list of CSP configuration interface version URIs
:param min_ver: Tuple of minimum version to return
:param max_ver: Tuple of maximum version to return
"""
csp_vers = CSP_CONFIG_VERSIONS
if min_ver is not None:
csp_vers = [
v for v in csp_vers if split_interface_version(v) >= min_ver
]
if max_ver is not None:
csp_vers = [
v for v in csp_vers if split_interface_version(v) <= max_ver
]
return csp_vers
[docs]
def normalize_csp_config_version(
csp_interface_version: Union[int, str], csp_config: dict = None
):
"""Provides a standard interface version for configure
:param csp_interface_version: External guess at the interface version
:param csp_config: Example configuration to derive version from
:returns: Canonical URI of interface version
"""
# Get from CSP configuration, if available
if csp_config is not None and "interface" in csp_config:
csp_interface_version = csp_config["interface"]
# If interface prefix is "https://schema.skatelescope.org/" update
# with new interface value
if csp_interface_version.startswith("https://schema.skatelescope.org/"):
csp_interface_version = (
"https://schema.skao.int/" + csp_interface_version[32:]
)
return csp_interface_version
[docs]
def check_csp_interface_version(
version: str,
allowed_prefixes: Union[str, List[str]] = _ALLOWED_URI_PREFIXES,
) -> str:
"""
Check CSP interface version.
Checks that the interface URI has one of the allowed prefixes. If it does,
the version number is returned. If not, a ValueError exception is raised.
:param version: CSP interface URI
:param allowed_prefixes: allowed URI prefix(es)
:returns: version number
"""
if not isinstance(allowed_prefixes, list):
allowed_prefixes = [allowed_prefixes]
# Valid?
for prefix in allowed_prefixes:
if version.startswith(prefix):
number = version[len(prefix) :]
return number
raise ValueError(f"CSP interface URI '{version}' not allowed")
[docs]
def get_csp_config_subsystem_version(
version: str, subsystem: str
) -> Tuple[str, str]:
"""
Using the CompatibilityMap evaluates the subsystem
command interface number.
:param version: the CSP command interface URI
:param subsystem: the name of the subsystem
:return: A Tuple containing the string subsystem_version
and the subsystem_uri.
"""
subsystem_version = "0.0"
version_num = check_csp_interface_version(
version, [CSP_CONFIG_PREFIX, CSP_CONFIGSCAN_PREFIX]
)
if version_num < "3.0":
if subsystem == "pss":
(major, minor) = split_interface_version(version)
if (major, minor) < (2, 1):
subsystem_version = "0.0"
else:
subsystem_version = "0.1"
else:
subsystem_version = version_num
else:
if version_num in ConfigCompatibilityMap.keys():
subsystem_version = ConfigCompatibilityMap[version_num][subsystem]
else:
raise ValueError("Unknown major schema version")
subsystem_uri = get_config_subsystem_uri(version, subsystem)
subsystem_uri = subsystem_uri + subsystem_version
return subsystem_version, subsystem_uri
[docs]
def get_config_subsystem_uri(version: str, subsystem: str) -> str:
"""Return the desired CSP subsystem version URI related to the input
CSP configure command version"""
subsystem_uri = ""
if subsystem == "pss":
subsystem_uri = "https://schema.skao.int/ska-pss-configure/"
elif subsystem == "pst":
subsystem_uri = "https://schema.skao.int/ska-pst-configure/"
elif subsystem == "midcbf":
subsystem_uri = "/".join(version.rsplit("/", 1)[:-1]) + "/"
else:
raise ValueError(
f"Subsystem {subsystem} name not available at csp level"
)
return subsystem_uri
ConfigCompatibilityMap = {
"3.0": {
"midcbf": "3.0",
"pss": "0.1",
"pst": "2.5",
}, # PST and PSS not supported by CBF
"4.0": {
"midcbf": "4.0",
"pss": "0.1",
"pst": "2.5",
}, # PST and PSS not supported by CBF
"4.1": {
"midcbf": "4.1",
"pss": "0.1",
"pst": "2.5",
}, # PST and PSS not supported by CBF
"5.0": {
"midcbf": "5.0",
"pss": "0.1",
"pst": "2.5",
}, # PSS not supported by CBF
"6.0": {
"midcbf": "6.0",
"pss": "1.0",
"pst": "2.5",
}, # PSS not supported by CBF
"6.1": {
"midcbf": "6.0",
"pss": "1.1",
"pst": "2.5",
}, # PSS not supported by CBF
"7.0": {
"midcbf": "6.0",
"pss": "1.1",
"pst": "3.0",
}, # PSS not supported by CBF
"8.0": {
"midcbf": "6.0",
"pss": "1.2",
"pst": "3.0",
}, # PSS not supported by CBF
"8.1": {
"midcbf": "6.0",
"pss": "1.2",
"pst": "3.0",
}, # PSS not supported by CBF
}