"""
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.
"""
import astropy.units as u
from astropy.coordinates import SkyCoord
from astropy.units import Quantity
import ska_ost_senscalc.low.service as low_service
from ska_ost_senscalc.common.model import (
ContinuumCalculatorAndWeightingInput,
EnumConversion,
Weighting,
WeightingSpectralMode,
ZoomCalculatorAndWeightingInput,
)
from ska_ost_senscalc.common.service import sub_band_to_frequency_array
from ska_ost_senscalc.subarray import SubarrayStorage
from ska_ost_senscalc.utilities import Telescope
subarray_storage = SubarrayStorage(Telescope.LOW)
DEFAULT_COMMON_CALCULATE_PARAMS = {
"pointing_centre": "10:00:00 -30:00:00",
"integration_time_h": 1,
"elevation_limit": 20,
"spectral_averaging_factor": 1,
"robustness": 0,
}
DEFAULT_CONTINUUM_PARAMS = {
"freq_centre_mhz": 200,
"bandwidth_mhz": 300,
"pointing_centre": "10:00:00 -30:00:00",
"n_subbands": 1,
}
DEFAULT_ZOOM_PARAMS = {
"freq_centres_mhz": [200],
"total_bandwidths_khz": [
390.6
], # Take the default value as the total bandwidth of the narrowest zoom window allowed for all subarrays
"spectral_resolutions_hz": [
14.1285
], # Take the narrowest channel allowed in zoom mode
}
# In PSS folded pulse mode, bandwidth_mhz and spectral_resolution_hz are not exposed
# to the user. They are defined here for ease of use within the backend
PSS_BANDWIDTH_MHZ = 118.518513664 # 8192 channels * 14467.592 Hz
PSS_CHAN_WIDTH_HZ = 14467.592
DEFAULT_PSS_FOLDED_PARAMS = {
"freq_centre_mhz": 200,
"pointing_centre": "10:00:00 -30:00:00",
"integration_time_h": 1,
"elevation_limit": 20,
"dm": 0.0,
"pulse_period": 33, # Assume Crab as default (33 ms)
"intrinsic_pulse_width": 0.004, # Assume Crab as default (4 us)
"pulsar_mode": "folded_pulse",
}
DEFAULT_PSS_SINGLE_PARAMS = DEFAULT_PSS_FOLDED_PARAMS
DEFAULT_PSS_SINGLE_PARAMS["bandwidth_mhz"] = PSS_BANDWIDTH_MHZ
DEFAULT_PSS_SINGLE_PARAMS["pulsar_mode"] = "single_pulse"
# The spectral resolutions for the zoom windows are given by (781250.0 * 32/27)/(4096 * 16) multiplied by increasing powers of 2
# The allowed total bandwidths for zoom mode are then the chanel resolutions multiplied by the number of channels (1728), and converted to kHz
BANDWIDTH_PRECISION_KHZ = 1 # round to nearest 0.1 kHz, to avoid difference in floating point numbers when calculated in the front end
ALLOWED_ZOOM_TOTAL_BANDWIDTHS_KHZ = [
round(
(2 ** (N - 1) * (781250 * 32 / 27) / (4096 * 16)) * 1728 * 1e-3,
BANDWIDTH_PRECISION_KHZ,
)
for N in range(1, 9)
]
MAX_FREQUENCY_MHZ = 350
MIN_FREQUENCY_MHZ = 50
# For the subarrays not listed here, the full bandwidth is allowed defined by the limits above
MAXIMUM_BANDWIDTH_MHZ_FOR_SUBARRAY = {
"LOW_AA05_all": 75,
"LOW_AA1_all": 75,
"LOW_AA2_all": 150,
}
[docs]
def validate_and_set_defaults_for_calculate(
user_input: dict,
spectral_mode: str,
) -> ContinuumCalculatorAndWeightingInput | ZoomCalculatorAndWeightingInput:
"""
:param user_input: the parameters from the HTTP request for the /api/low/continuum/calculate or /api/low/zoom/calculate request
:param spectral_mode: the spectral mode of the calculation, either 'continuum' or 'line'
:return: A data class instance of ContinuumCalculatorAndWeightingInput or ZoomCalculatorAndWeightingInput, 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
default_params = (
DEFAULT_COMMON_CALCULATE_PARAMS | DEFAULT_CONTINUUM_PARAMS
if spectral_mode == WeightingSpectralMode.CONTINUUM
else DEFAULT_COMMON_CALCULATE_PARAMS | DEFAULT_ZOOM_PARAMS
)
user_input = default_params | user_input
# common validation
err_msgs = []
_validate_num_stations_or_subarray_configuration(user_input, err_msgs)
pointing_centre = _validate_pointing_centre(
dict(pointing_centre=user_input["pointing_centre"]), err_msgs
)
num_stations = low_service._num_stations_from_input(user_input)
# Retrieve subarray conf from user input, either from name provided or num of stations
# subarray conf set as None for custom sub-array
subarray_configuration = (
EnumConversion.to_array_configuration(
user_input["subarray_configuration"], err_msgs
)
if "subarray_configuration" in user_input
else None
)
# OpenAPI enum strings need converting to Python enum members
user_input["weighting_mode"] = EnumConversion.to_weighting(
user_input["weighting_mode"], err_msgs
)
_validate_robustness_for_robust_weighting_mode(user_input, err_msgs)
# Continuum specific validation
if spectral_mode == WeightingSpectralMode.CONTINUUM:
user_input = _validate_and_set_defaults_for_continuum(user_input, err_msgs)
# Zoom specific validation
else:
user_input = _validate_and_set_defaults_for_zoom(user_input, err_msgs)
if err_msgs:
raise ValueError(*err_msgs)
# convert the validated user_input into a ContinuumCalculatorAndWeightingInput
# data class instance to feed the calculator and weighting functions
# TODO see if we could combine these 2 data classes into one
if spectral_mode == WeightingSpectralMode.CONTINUUM:
calculator_weighting_input = ContinuumCalculatorAndWeightingInput(
freq_centre=user_input["freq_centre_hz"],
bandwidth_mhz=user_input["bandwidth_mhz"],
num_stations=num_stations,
pointing_centre=pointing_centre,
integration_time_h=user_input["integration_time_h"],
elevation_limit=user_input["elevation_limit"],
telescope=Telescope.LOW,
spectral_mode=spectral_mode,
subarray_configuration=subarray_configuration,
spectral_averaging_factor=user_input["spectral_averaging_factor"],
n_subbands=user_input["n_subbands"],
weighting_mode=user_input["weighting_mode"],
robustness=user_input["robustness"],
subband_freq_centres=user_input["subband_freq_centres_hz"],
)
if spectral_mode == WeightingSpectralMode.LINE:
calculator_weighting_input = ZoomCalculatorAndWeightingInput(
num_stations=num_stations,
integration_time_h=user_input["integration_time_h"],
elevation_limit=user_input["elevation_limit"],
spectral_averaging_factor=user_input["spectral_averaging_factor"],
spectral_resolutions_hz=user_input["spectral_resolutions_hz"],
total_bandwidths_khz=user_input["total_bandwidths_khz"],
freq_centres=user_input["freq_centres"],
pointing_centre=pointing_centre,
subarray_configuration=subarray_configuration,
weighting_mode=user_input["weighting_mode"],
robustness=user_input["robustness"],
dec=pointing_centre.dec,
telescope=Telescope.LOW,
)
return calculator_weighting_input
def _validate_and_set_defaults_for_continuum(user_input: dict, err_msgs: list) -> dict:
_validate_max_continuum_bandwidth(user_input, err_msgs)
_validate_spectral_window(
user_input["freq_centre_mhz"], user_input["bandwidth_mhz"], err_msgs
)
_validate_spectral_averaging_factor(user_input, err_msgs)
user_input["subband_freq_centres_mhz"] = _get_subband_freq_centres_for_n_subbands(
user_input, err_msgs
)
# OpenAPI spec requires frequency in MHz, weighting expects it in Hz
# weighting currently expects a list of frequencies instead of a single
# frequency value
user_input["freq_centre_hz"] = Quantity(
user_input["freq_centre_mhz"] * 1e6, unit=u.Hz
)
if "subband_freq_centres_mhz" in user_input:
user_input["subband_freq_centres_hz"] = (
[
Quantity(subband_freq_centre_hz * 1e6, unit=u.Hz)
for subband_freq_centre_hz in user_input["subband_freq_centres_mhz"]
]
if user_input["subband_freq_centres_mhz"]
else []
)
# todo add a validation test
_validate_n_subbands_for_subbands_frequencies(
user_input["subband_freq_centres_hz"],
user_input["n_subbands"],
err_msgs,
)
return user_input
def _validate_and_set_defaults_for_zoom(user_input: dict, err_msgs: list) -> dict:
_validate_zoom_parameter_combinations(user_input, err_msgs)
_validate_zoom_bandwidth_for_subarray(user_input, err_msgs)
for freq_centre_mhz, total_bandwidth_khz in zip(
user_input["freq_centres_mhz"], user_input["total_bandwidths_khz"]
):
_validate_spectral_window(freq_centre_mhz, total_bandwidth_khz * 1e-3, err_msgs)
user_input["freq_centres"] = [
Quantity(freq_centre_mhz * 1e6, unit=u.Hz)
for freq_centre_mhz in user_input["freq_centres_mhz"]
]
return user_input
[docs]
def validate_and_set_defaults_for_pss(user_input: dict) -> dict:
"""
:param user_input: the parameters from the HTTP request for the /api/low/pss/calculate request
: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
# (i.e.) if the user_input contains a key then it will not be overwritten
# by the defaults
if user_input["pulsar_mode"] == "single_pulse":
user_input = DEFAULT_PSS_SINGLE_PARAMS | user_input
else:
user_input = DEFAULT_PSS_FOLDED_PARAMS | user_input
# In the folded pulse mode, the bandwidth is not a user-definable parameter
# Reset to the default value.
user_input["bandwidth_mhz"] = PSS_BANDWIDTH_MHZ
# Insert PSS channel width
user_input["spectral_resolution_hz"] = PSS_CHAN_WIDTH_HZ
err_msgs = []
_validate_num_stations_or_subarray_configuration(user_input, err_msgs)
_validate_spectral_window(
user_input["freq_centre_mhz"], user_input["bandwidth_mhz"], err_msgs
)
_validate_pointing_centre(user_input, err_msgs)
if user_input["pulsar_mode"] == "single_pulse":
# Tests specific to single_pulse mode
if user_input["bandwidth_mhz"] > PSS_BANDWIDTH_MHZ:
err_msgs.append(
"For single-pulse calculations, bandwidth cannot be greater "
f"than {PSS_BANDWIDTH_MHZ} MHz"
)
else:
# Tests specific to folded_pulse mode
if user_input["intrinsic_pulse_width"] >= user_input["pulse_period"]:
err_msgs.append(
"Intrinsic pulse width cannot be equal to or larger than the pulse period."
)
if err_msgs:
raise ValueError(*err_msgs)
return user_input
def _validate_spectral_averaging_factor(user_input: dict, err_msgs: list):
n_continuum_channels = (
user_input["bandwidth_mhz"] * 1e3 / low_service.LOW_CONTINUUM_CHANNEL_WIDTH_KHZ
)
max_channels = int(n_continuum_channels // 2) # Floor div
valid = 1 <= user_input["spectral_averaging_factor"] <= max_channels
if not valid:
err_msgs.append(
f"The spectral averaging factor must lie between 1 and {max_channels}"
)
def _validate_robustness_for_robust_weighting_mode(user_input: dict, err_msgs: list):
if (
user_input["weighting_mode"] == Weighting.ROBUST
and "robustness" not in user_input
):
err_msgs.append("Parameter 'robustness' should be set for 'robust' weighting")
def _validate_n_subbands_for_subbands_frequencies(
subband_freq_centres_hz: list, n_subbands: int, err_msgs: list
) -> None:
if len(subband_freq_centres_hz) > 0 and len(subband_freq_centres_hz) != n_subbands:
err_msgs.append(
"When subband_freq_centres_mhz is provided, Parameter 'n_subbands' should be set and match the number of subband_freq_centres_mhz."
)
def _get_subband_freq_centres_for_n_subbands(user_input: dict, err_msgs: list):
# if the user has entered a num of subbands bigger than 1
if "n_subbands" in user_input and user_input["n_subbands"] > 1:
# if provided, return the subband_freq_centres_mhz entered by the user
if "subband_freq_centres_mhz" in user_input:
return user_input["subband_freq_centres_mhz"]
# if the user hasn't provided it, generate the subband frequencies center
else:
return sub_band_to_frequency_array(
user_input["n_subbands"],
user_input["bandwidth_mhz"],
user_input["freq_centre_mhz"],
)
# if the user has not provided num of subbands, or entered 1 for num of subbands:
# we return the subband_freq_centres_mhz entered by the user if provided
# otherwise, we return None and don't generate subband data
elif "n_subbands" not in user_input or user_input["n_subbands"] == 1:
return (
user_input["subband_freq_centres_mhz"]
if "subband_freq_centres_mhz" in user_input
else None
)
def _validate_num_stations_or_subarray_configuration(
user_input: dict, err_msgs: list
) -> None:
"""
Either num_stations or a subarray_configuration should be given by the user.
:param user_input: the parameters passed to the API.
:param err_msgs: the list of error messages to append a validation error to
"""
if ("subarray_configuration" in user_input) == ("num_stations" in user_input):
err_msgs.append(
"Only 'subarray_configuration' or 'num_stations' should be specified."
)
def _validate_spectral_window(
freq_centre: float, bandwidth_mhz: float, err_msgs: list
) -> None:
min_freq = freq_centre - bandwidth_mhz / 2
max_freq = freq_centre + bandwidth_mhz / 2
if min_freq < MIN_FREQUENCY_MHZ or max_freq > MAX_FREQUENCY_MHZ:
err_msgs.append(
"Spectral window defined by central frequency and bandwidth does"
" not lie within the 50 - 350 MHz range."
)
def _validate_pointing_centre(user_input: dict, err_msgs: list) -> SkyCoord:
try:
return SkyCoord(user_input["pointing_centre"], unit=(u.hourangle, u.deg))
except ValueError:
err_msgs.append(
"Specified pointing centre is invalid, expected format HH:MM:SS[.ss]"
" DD:MM:SS[.ss]."
)
def _validate_zoom_parameter_combinations(user_input, err_msg=None):
# Create a set with the length of each of the inputs. If they are all the same
# length then the set should have one element which is the common length
set_of_lengths = {
len(user_input.get("freq_centres_mhz", [])),
len(user_input.get("spectral_resolutions_hz", [])),
len(user_input.get("total_bandwidths_khz", [])),
}
# If they are not all the same length, or none of the values are set, raise a validation error
if len(set_of_lengths) != 1 or 0 in set_of_lengths:
msg = "Parameters 'freq_centres_mhz', 'spectral_resolutions_hz' and 'total_bandwidths_khz' must all be set together and have the same length."
if err_msg is not None:
err_msg.append(msg)
else:
raise ValueError(msg)
def _validate_max_continuum_bandwidth(user_input: dict, err_msgs: list):
"""
Validates the maximum bandwidth allowed for a continuum calculation is allowed for the given subarray configuration.
:param user_input: the parameters passed to the API.
:param err_msgs: the list of error messages to append a validation error to
"""
max_allowed_bandwidth = MAXIMUM_BANDWIDTH_MHZ_FOR_SUBARRAY.get(
user_input.get("subarray_configuration"), MAX_FREQUENCY_MHZ - MIN_FREQUENCY_MHZ
)
if user_input["bandwidth_mhz"] > max_allowed_bandwidth:
err_msgs.append(
f"Maximum bandwidth ({max_allowed_bandwidth} MHz) for this subarray has been exceeded."
)
def _validate_zoom_bandwidth_for_subarray(user_input: dict, err_msgs: list):
"""
Validates the zoom mode is allowed for the given subarray configuration, by validating the total bandwidth
:param user_input: the parameters passed to the API.
:param err_msgs: the list of error messages to append a validation error to
"""
# AA0.5 and AA1 are not supported in zoom mode so not defined in the API. AA2 only supports the larger zoom
# modes (hence taking the final 4 elements here). The later arrays support all the zoom modes, as does a custom array.
allowed_bandwidth_for_subarray = (
ALLOWED_ZOOM_TOTAL_BANDWIDTHS_KHZ
if ("subarray_configuration" not in user_input)
or (user_input.get("subarray_configuration") not in ["LOW_AA2_all"])
else ALLOWED_ZOOM_TOTAL_BANDWIDTHS_KHZ[-4:]
)
for total_bandwidth_khz in user_input["total_bandwidths_khz"]:
if (
not round(total_bandwidth_khz, BANDWIDTH_PRECISION_KHZ)
in allowed_bandwidth_for_subarray
):
err_msgs.append(
f"Bandwidth {total_bandwidth_khz} not one the of allowed"
f" values in zoom mode for {user_input.get('subarray_configuration', 'custom')} subarray configuration: {allowed_bandwidth_for_subarray}"
)