"""
This module provides semantic validation for inputs to the Sensitivity Calculator,
including checking for required values, setting default values, and domain related checks.
Syntactic validation and basic validation, for example of min/max values of numbers, is done
by Connexion and the OpenAPI spec.
"""
from ska_ost_senscalc.common.model import Weighting
from ska_ost_senscalc.mid.model import MidSpectralMode
from ska_ost_senscalc.subarray import MIDArrayConfiguration, SubarrayStorage
from ska_ost_senscalc.utilities import Telescope
subarray_storage = SubarrayStorage(Telescope.MID)
DEFAULT_CALCULATE_PARAMS = {
"pmv": 10,
"el": 45,
"alpha": 2.75,
"n_subbands": 1,
}
DEFAULT_WEIGHTING_PARAMS = {
"taper": 0.0,
}
SUBARRAY_CONFIGURATIONS_ALLOWED_FOR_ZOOM = [
MIDArrayConfiguration.MID_AASTAR_ALL,
MIDArrayConfiguration.MID_AASTAR_SKA_ONLY,
MIDArrayConfiguration.MID_AA4_ALL,
MIDArrayConfiguration.MID_AA4_MEERKAT_ONLY,
MIDArrayConfiguration.MID_AA4_SKA_ONLY,
]
MID_CONTINUUM_CHANNEL_WIDTH_KHZ = 13.44
BAND_LIMITS = {
"Band 1": [
{"type": "ska", "limits": [0.35e9, 1.05e9]},
{"type": "meerkat", "limits": [0.58e9, 1.015e9]},
{"type": "mixed", "limits": [0.58e9, 1.015e9]},
],
"Band 2": [
{"type": "ska", "limits": [0.95e9, 1.76e9]},
{"type": "meerkat", "limits": [0.95e9, 1.67e9]},
{"type": "mixed", "limits": [0.95e9, 1.67e9]},
],
"Band 3": [
{"type": "ska", "limits": [1.65e9, 3.05e9]},
{"type": "meerkat", "limits": [1.75e9, 3.05e9]},
{"type": "mixed", "limits": [1.75e9, 3.05e9]},
],
"Band 4": [{"type": "ska", "limits": [2.8e9, 5.18e9]}],
"Band 5a": [{"type": "ska", "limits": [4.6e9, 8.5e9]}],
"Band 5b": [{"type": "ska", "limits": [8.3e9, 15.4e9]}],
}
# For the subarrays not listed here, the full bandwidth is allowed defined by the limits above
MAXIMUM_BANDWIDTH_FOR_SUBARRAY = {
MIDArrayConfiguration.MID_AA05_ALL: 800e6,
MIDArrayConfiguration.MID_AA1_ALL: 800e6,
MIDArrayConfiguration.MID_AA2_ALL: 800e6,
}
[docs]
def validate_and_set_defaults_for_calculate(user_input: dict) -> dict:
"""
:param user_input: the parameters from the HTTP request to /calculate
:return: A new copy of the input dict, with defaults set for missing values
:raises: ValueError if the input data is not valid
"""
# Merge the default params and the user input into a new dict. The union operator for a dict will
# take the rightmost value, ie if the user_input contains a key then it will not be overwritten by the defaults
user_input = DEFAULT_CALCULATE_PARAMS | user_input
if ("integration_time" in user_input) == ("sensitivity" in user_input):
raise ValueError(
"Either 'sensitivity' or 'integration_time' must be specified, but not both at once."
)
_validate_array_config_and_antennas(user_input)
_validate_continuum_bandwidth_for_array_config(user_input)
_validate_spectral_window(user_input)
_validate_zoom_parameters(user_input)
_validate_subband_parameters(user_input)
return user_input
[docs]
def validate_and_set_defaults_for_weighting(user_input: dict) -> dict:
"""
:param user_input: the parameters from the HTTP request to /weighting
:return: A new copy of the input dict, with defaults set for missing values
:raises: ValueError if the input data is not valid
"""
user_input = DEFAULT_WEIGHTING_PARAMS | user_input
if (
user_input.get("frequency") is None
and user_input.get("zoom_frequencies") is None
):
raise ValueError("Parameter 'frequency' or 'zoom_frequencies' is required")
if (
user_input.get("calculator_mode") is MidSpectralMode.CONTINUUM.value
and user_input.get("frequency") is None
):
raise ValueError(
"Parameter 'frequency' must be set for 'continuum' calculator mode"
)
if (
user_input.get("calculator_mode") is MidSpectralMode.LINE.value
and user_input.get("zoom_frequencies") is None
):
raise ValueError(
"Parameter 'zoom_frequencies' must be set for 'line' calculator mode"
)
if (user_input.get("weighting") == Weighting.ROBUST.value) and user_input.get(
"robustness"
) is None:
raise ValueError("Parameter 'robustness' should be set for 'robust' weighting")
if user_input.get("weighting") not in [w.value for w in Weighting]:
raise ValueError(f"Weighting value {user_input.get('weighting')} is not valid")
if user_input.get("array_configuration") not in [
a.value for a in MIDArrayConfiguration
]:
raise ValueError(
f"Array Configuration {user_input.get('array_configuration')} is not valid"
)
return user_input
def _validate_zoom_parameters(user_input: dict) -> None:
"""
:param user_input: the parameters from the HTTP request
:raises: ValueError if the input data relevant for zoom mode is not valid
"""
if user_input.get("zoom_frequencies") or user_input.get("zoom_resolutions"):
array_configuration = user_input.get(
"array_configuration"
) # Could be none for a Custom input
if (
array_configuration
and array_configuration not in SUBARRAY_CONFIGURATIONS_ALLOWED_FOR_ZOOM
):
raise ValueError("No zoom modes are available for this array assembly.")
if user_input.get("zoom_frequencies"):
# Check that zoom_frequencies has the same length as zoom_resolutions
if not user_input.get("zoom_resolutions"):
raise ValueError(
"Parameter 'zoom_resolutions' must also be set when setting"
" 'zoom_frequencies'."
)
if len(user_input.get("zoom_frequencies")) != len(
user_input.get("zoom_resolutions")
):
raise ValueError(
"Parameters 'zoom_resolutions' and 'zoom_frequencies' must"
" have the same length."
)
if user_input.get("sensitivity"):
if user_input.get("zoom_sensitivities"):
if len(user_input.get("zoom_sensitivities")) != len(
user_input.get("zoom_resolutions")
):
raise ValueError(
"Parameters 'zoom_sensitivities' and"
" 'zoom_frequencies' must have the same length."
)
else:
# Check that zoom_sensitivities are specified if requesting a sensitivity
raise ValueError(
"Parameter 'zoom_sensitivities' must be set when setting"
" 'zoom_frequencies' and 'sensitivity'."
)
elif user_input.get("zoom_resolutions"):
raise ValueError(
"Parameter 'zoom_frequencies' must also be set when setting"
" 'zoom_resolutions'."
)
def _validate_subband_parameters(user_input: dict) -> None:
"""
:param user_input: the parameters from the HTTP request
:raises: ValueError if the input data relevant for subband calculations is not valid
"""
if not user_input.get("sensitivity"):
# Validation currently only needs to be done for the sensitivity -> integration time calculation
return
n_subbands = user_input.get("n_subbands")
subband_sensitivities = user_input.get("subband_sensitivities")
if n_subbands > 1 and not subband_sensitivities:
raise ValueError(
"Parameter 'subband_sensitivities' must be set when setting 'sensitivity' and"
"'n_subbands' is greater than 1."
)
if subband_sensitivities and n_subbands <= 1:
raise ValueError(
"Parameter 'n_subbands' must be greater than 1 when setting 'subband_sensitivities' and 'sensitivity'."
)
if (
n_subbands
and subband_sensitivities
and n_subbands != len(subband_sensitivities)
):
raise ValueError(
"Parameter 'subband_sensitivities' must have the same length as the value of 'n_subbands' for"
"'n_subbands' greater than 1."
)
def _validate_array_config_and_antennas(user_input):
"""
Validates that if the user is using a custom array (ie by giving n_ska and n_meer)
that they are not also passing an array_configuration. Also validates that both n_ska
and n_meer are given.
It does not validate that either an array_configuration or custom numbers are given, as the user
can specify neither and the default will be used.
"""
n_ska_but_not_n_meer = "n_ska" in user_input and "n_meer" not in user_input
n_meer_but_not_n_ska = "n_meer" in user_input and "n_ska" not in user_input
one_of_array_config_and_n_antennas = ("array_configuration" in user_input) is not (
"n_ska" in user_input or "n_meer" in user_input
)
if (
n_ska_but_not_n_meer
or n_meer_but_not_n_ska
or not one_of_array_config_and_n_antennas
):
raise ValueError(
"Only 'array_configuration' or the number of antennas ('n_ska' AND 'n_meer') should be specified."
)
if "array_configuration" in user_input:
user_input["array_configuration"] = MIDArrayConfiguration(
user_input["array_configuration"]
)
def _validate_continuum_bandwidth_for_array_config(user_input: dict) -> None:
"""
Validates that the continuum bandwidth is less than the maximum for the subarray.
For earlier subarrays, this is a static value.
For the later ones, the full bandwidth is allowed and this is
checked by checking the spectral window, as the limits change depending on the subarray.
"""
if (
"array_configuration" in user_input
and user_input["array_configuration"] in MAXIMUM_BANDWIDTH_FOR_SUBARRAY
):
max_continuum_bandwidth_hz = MAXIMUM_BANDWIDTH_FOR_SUBARRAY[
user_input["array_configuration"]
]
if user_input["bandwidth"] > max_continuum_bandwidth_hz:
raise ValueError(
f"Maximum bandwidth ({max_continuum_bandwidth_hz * 1e-6} MHz) for this subarray has been exceeded."
)
def _validate_spectral_window(user_input: dict) -> None:
"""
Validates that the band and array configuration combination is allowed.
Then validates that the spectral window (defined by the frequency and bandwidth)
is within the limits for the band and subarray.
"""
if "array_configuration" in user_input:
subarray = subarray_storage.load_by_label(
user_input["array_configuration"].value
)
n_ska = subarray.n_ska
n_meer = subarray.n_meer
else:
n_ska = user_input["n_ska"]
n_meer = user_input["n_meer"]
antenna_type = (
"mixed" if n_ska > 0 and n_meer > 0 else ("ska" if n_ska > 0 else "meerkat")
)
try:
if user_input["rx_band"] not in BAND_LIMITS:
raise ValueError(f"{user_input['rx_band']} not supported.")
limits = next(
filter(
lambda entry: entry["type"] == antenna_type,
BAND_LIMITS[user_input["rx_band"]],
)
)["limits"]
except StopIteration:
# This means the next function raised an error as the 'type' is not present in the band.
raise ValueError("Subarray configuration not allowed for given observing band.")
frequency = user_input["frequency"]
bandwidth = user_input["bandwidth"]
min_freq = frequency - bandwidth / 2
max_freq = frequency + bandwidth / 2
if min_freq < limits[0] or max_freq > limits[1]:
raise ValueError(
"Spectral window defined by central frequency and bandwidth does not lie within the band range."
)