import logging
from typing import List, Union
from .._common import split_interface_version
CSP_ASSIGNRESOURCES_PREFIX = "https://schema.skao.int/ska-csp-assignresources/"
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_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 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,
],
key=split_interface_version,
)
_LOGGER = logging.getLogger(__name__)
[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")