import logging
from typing import Callable, Iterable, Sequence, Union
from schema import Schema, SchemaError
from ska_telmodel._common import split_interface_version
SDP_ASSIGNRES_PREFIX = "https://schema.skao.int/ska-sdp-assignres/"
SDP_RELEASERES_PREFIX = "https://schema.skao.int/ska-sdp-releaseres/"
SDP_CONFIGURE_PREFIX = "https://schema.skao.int/ska-sdp-configure/"
SDP_SCAN_PREFIX = "https://schema.skao.int/ska-sdp-scan/"
SDP_RECVADDRS_PREFIX = "https://schema.skao.int/ska-sdp-recvaddrs/"
_LOGGER = logging.getLogger(__name__)
#: The type of a version parameter.
VERSION_TYPE = Union[int, str]
#: Call signature for schemas.
CALL_SIG = Callable[[VERSION_TYPE, bool], Schema]
#: The type af allowed prefixes.
PREFIXES_TYPE = Union[str, Sequence[str]]
_PREFIXES = [
SDP_ASSIGNRES_PREFIX,
SDP_RELEASERES_PREFIX,
SDP_CONFIGURE_PREFIX,
SDP_SCAN_PREFIX,
SDP_RECVADDRS_PREFIX,
]
SDP_INTERFACE_VERSIONS = [
(0, 0),
(0, 1),
(0, 2),
(0, 3),
(0, 4),
]
[docs]def sdp_interface_versions(prefix: str, min_ver=None, max_ver=None):
"""
Returns a list of SDP interface version URIs
:param prefix: Interface URI prefix
:param min_ver: Tuple of minimum version to return
:param max_ver: Tuple of maximum version to return
"""
sdp_vers = SDP_INTERFACE_VERSIONS
if min_ver is not None:
sdp_vers = [v for v in sdp_vers if v >= min_ver]
if max_ver is not None:
sdp_vers = [v for v in sdp_vers if v <= max_ver]
assert (
prefix[-1] == "/"
), "Please only pass prefixes ending with '/' to sdp_interface_versions!"
return [f"{prefix}{v0}.{v1}" for v0, v1 in sdp_vers]
[docs]def normalise_sdp_interface_version(
version: VERSION_TYPE,
prefix: str,
) -> str:
"""
Normalise SDP interface version.
Converts deprecated integer version number into a schema URI, where the
prefix specifies which schema to use. If the version is a string, it is
assumed to be a schema URI and it is returned unchanged.
:param version: SDP interface version
:param prefix: schema prefix
:returns: SDP interface URI
"""
if isinstance(version, int):
if 0 <= version <= 2:
version = prefix + "0.1"
elif version == 3:
version = prefix + "0.2"
else:
raise ValueError(
f"SDP interface version '{version}' not supported"
)
if version.startswith("https://schema.skatelescope.org/"):
version = "https://schema.skao.int/" + version[32:]
return version
[docs]def check_sdp_interface_version(
version: str,
allowed_prefixes: PREFIXES_TYPE = None,
) -> str:
"""
Check SDP 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: SDP interface URI
:param allowed_prefixes: allowed URI prefix(es)
:returns: version number
"""
if allowed_prefixes is None:
allowed_prefixes = _PREFIXES
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"SDP interface URI '{version}' not allowed")
[docs]class SdpVersion:
"""
Wrapper around the normalise/check functions and
stores the results.
:param version: SDP interface version
:param prefix: schema prefix
:param allowed_prefixes: allowed URI prefix(es)
:returns: version object
"""
def __init__(
self,
version: VERSION_TYPE,
prefix: str = None,
allowed_prefixes: PREFIXES_TYPE = None,
):
self.version = (
version
if prefix is None
else normalise_sdp_interface_version(version, prefix)
)
self.number = check_sdp_interface_version(
self.version, allowed_prefixes
)
def __repr__(self):
return f"{self.version}: {self.number}"
[docs]class SchemaFactory:
"""
Get the right schema for a type based on its version.
"""
def __init__(
self,
prefix: str = None,
allowed_prefixes: PREFIXES_TYPE = None,
):
"""
Construct a schema factory.
:param prefix: schema prefix
:param allowed_prefixes: allowed URI prefix(es)
"""
self._lookups = {}
self.prefix = prefix
self.allowed_prefixes = allowed_prefixes
def __call__(
self,
version: VERSION_TYPE,
strict: bool,
):
"""
Get a schema.
:param version: SDP interface version
:param strict: whether strict or not
:returns: the schema
"""
sdp_version = SdpVersion(
version, prefix=self.prefix, allowed_prefixes=self.allowed_prefixes
)
return self.get_schema(sdp_version, strict)
[docs] def register(self, version: str, func: CALL_SIG) -> None:
"""
Register a function to create the schema.
:param version: the short version number (not the URI).
:param func: function to create the schema
"""
self._lookups[version] = func
[docs] def register_all(
self,
versions: Iterable[str],
func: CALL_SIG,
) -> None:
"""
Register a function to create the schema for multiple versions.
:param versions: iterable of short version numbers (not the URIs).
:param func: function to create the schema
"""
for version in versions:
self.register(version, func)
[docs] def get_schema(self, version: SdpVersion, strict: bool) -> Schema:
"""
Get the schema for this version.
If strict, an exact match is required. Otherwise, the last minor
version matching the major version is accepted.
It is assumed that a version is of the form version.subversion.
:param version: SDP version object
:param strict: whether strict or not
:returns: the matching schema
"""
# Could consider a strategy pattern if this gets too complicated.
keys = self._lookups.keys()
if version.number in keys:
# Exact match, done.
key = version.number
elif strict:
raise SchemaError(f"{version.number} not in {keys}")
else:
_LOGGER.info(
"No exact match for %s in %s, try to match major version",
version.number,
keys,
)
major, _ = split_interface_version(version.version)
# Collect the minor versions matching the major one.
minors = []
for key in keys:
key_major, key_minor = split_interface_version("/" + key)
if major == key_major:
minors.append(key_minor)
if not minors:
raise SchemaError(
f"No matching major version for {version.number} in {keys}"
)
# Use the last matching minor version.
key = ".".join((str(major), str(sorted(minors)[-1])))
# Call the matching function.
return self._lookups[key](version.version, strict)