# -*- coding: utf-8 -*-
#
# This file is part of the SKA PST project.
#
# Distributed under the terms of the BSD 3-clause new license.
# See LICENSE for more info.
"""This module is to validate a PST configuration request."""
from __future__ import annotations
import json
from json import JSONDecodeError
from typing import Any, Callable, Dict, Tuple, TypeAlias
from jsonschema.validators import Draft7Validator as JsonSchemaValidator
from overrides import override
from schema import SchemaError
from ska_schemas.pst.schema import get_pst_config_schema
from ska_schemas.schema import csp
from ska_tango_base.commands import ArgumentValidator
from .pst_schema import pst_scan_schema_json
# A version is defined as a tuple of major and minor versions (i.e. (major, minor))
Version: TypeAlias = Tuple[int, int]
MIN_VERSION: Version = (2, 4)
def _min_version_string(version: Version) -> str:
return f"{version[0]}.{version[1]}"
PROCESSING_MODE_MIN_VERSIONS: Dict[str, Version] = {
"VOLTAGE_RECORDER": MIN_VERSION,
"FLOW_THROUGH": (2, 5),
# The following will change when we review the modes
"PULSAR_TIMING": MIN_VERSION,
# Dynamic has been rename to detected filter bank in version 3.0
"DYNAMIC_SPECTRUM": (2, 4),
"DETECTED_FILTERBANK": (3, 0),
}
[docs]class PstConfigValidator(ArgumentValidator):
"""A validator class that ensures the config request sent to PST is correct.
This extends the SKA TANGO Base class :py:class:`ArgumentValidator`. It ensures
that only the `common` and `pst` scan configuration parts.
This does not proxy the requests down to subordinate devices or the core apps
as that should be done in the component manager as that can take some time.
"""
def __init__(self: PstConfigValidator, action: Callable[[dict], None] | None = None) -> None:
"""Initialise instance of validator.
:param action: an extra action to perform after basic JSON validation, defaults to None
:type action: Callable[[dict], None] | None, optional
"""
self._action = action if action else (lambda _: None)
@override
def validate(self: PstConfigValidator, *args: Any, **kwargs: Any) -> tuple[tuple[Any, ...], dict]:
"""Validate input and return the scan request."""
try:
assert len(args) == 1, f"expected only 1 argument got {len(args)}"
config = json.loads(args[0])
version: str = config["interface"]
major, minor = csp.version.split_interface_version(version)
assert (
major,
minor,
) >= MIN_VERSION, f"version {major}.{minor} should be >= {_min_version_string(MIN_VERSION)}"
# get the CSP schema
schema = get_pst_config_schema(version=version, strict=False)
schema.validate(config)
pst_scan_config = config["pst"]["scan"]
if "observation_mode" in pst_scan_config:
processing_mode = pst_scan_config["observation_mode"]
else:
processing_mode = pst_scan_config["pst_processing_mode"]
assert processing_mode in PROCESSING_MODE_MIN_VERSIONS, (
"expected 'processing_mode' to be one of "
f"{list(PROCESSING_MODE_MIN_VERSIONS.keys())} but was '{processing_mode}'"
)
processing_mode_min_version = PROCESSING_MODE_MIN_VERSIONS[processing_mode]
assert (major, minor) >= processing_mode_min_version, (
f"version {major}.{minor} should be >= {_min_version_string(processing_mode_min_version)} "
f"when processing mode is {processing_mode}"
)
assert len(pst_scan_config["receptor_weights"]) == len(pst_scan_config["receptors"]), (
f"Number of receptors ({len(pst_scan_config['receptors'])}) does not match "
f"number of receptor weights ({len(pst_scan_config['receptor_weights'])})"
)
self._action(config)
return (), config
except KeyError as e:
message = f"Validation {e} field not in request"
raise ValueError(message) from e
except (AssertionError, SchemaError, JSONDecodeError) as e:
message = f"Validation {e}"
raise ValueError(message) from e
[docs]class PstScanValidator(ArgumentValidator):
"""A class used to validate scan requests for PST."""
@override
def validate(self: PstScanValidator, *args: Any, **kwargs: Any) -> tuple[tuple[Any, ...], dict]:
"""Validate scan command request.
A legacy of PST is that it was assumed that the request for a Scan would be just a string
version of an int (e.g. "123") but the request should be a stringified JSON request with
scan_id as a key. This validator will try to validate it against being a JSON request
and fall back to bring stringified number.
Checks that there is only one positional argument and no keyword
arguments; unpacks the positional argument from JSON into a
dictionary; and validate against the provided JSON schema.
:param args: positional args to the command
:param kwargs: keyword args to the command
:returns: validated args and kwargs
"""
assert not kwargs, "Command Scan was invoked with kwargs. JSON validation does not permit kwargs"
if args:
assert len(args) == 1, (
f"Command Scan was invoked with {len(args)} args. "
"JSON validation only permits one positional argument."
)
decoded_dict = json.loads(args[0])
if isinstance(decoded_dict, str) or isinstance(decoded_dict, int):
decoded_dict = {"scan_id": int(decoded_dict)}
else:
decoded_dict = {}
assert isinstance(
decoded_dict, dict
), f"Expected ability to convert request arg to a dict. Type is {type(decoded_dict)}"
if "interface" not in decoded_dict:
decoded_dict["interface"] = csp.version.CSP_CONFIG_VER2_4
version: str = decoded_dict["interface"]
schema = pst_scan_schema_json(version=version)
validator = JsonSchemaValidator(schema=schema)
validator.validate(decoded_dict)
return (), decoded_dict