"""
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.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.common.validation import (
validate_n_subbands_for_subbands_frequencies,
validate_pointing_centre,
)
from ska_ost_senscalc.subarray import SubarrayStorage
from ska_ost_senscalc.utilities import Telescope
subarray_storage = SubarrayStorage(Telescope.LOW)
DEFAULT_COMMON_CALCULATE_PARAMS = {
"integration_time_h": 1,
"elevation_limit": 20,
"spectral_averaging_factor": 1,
"robustness": 0,
}
DEFAULT_CONTINUUM_PARAMS = {
"freq_centre_mhz": 200,
# Keep the default window inside the new LOW observing band: the maximum valid
# total span when centred at 200 MHz is 2 * (MAX_FREQUENCY_MHZ - 200).
"bandwidth_mhz": 299.21875,
"n_subbands": 1,
}
DEFAULT_ZOOM_PARAMS = {
"freq_centres_mhz": [200],
"total_bandwidths_khz": [
904.2244
], # 4000 channels at the default spectral resolution
"spectral_resolutions_hz": [
226.0561
], # Narrowest channel width of the zoom modes supported by all subarrays
}
# 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 total bandwidth of a zoom window can take any value, as long as the number of channels implied by
# the total bandwidth and spectral resolution lies between MIN_ZOOM_CHANNELS and MAX_ZOOM_CHANNELS.
# At least 2 channels are needed for a valid spectral averaging factor.
MIN_ZOOM_CHANNELS = 2
MAX_ZOOM_CHANNELS = 55296
MAX_FREQUENCY_MHZ = 349.609375
MIN_FREQUENCY_MHZ = 49.609375
# 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(
user_input.get("reference_coordinate") or user_input.get("pointing_centre"),
err_msgs,
)
num_stations = (
low_service._num_stations_from_input(user_input)
if ("subarray_configuration" in user_input or "num_stations" in user_input)
else None
)
# 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,
field_name="subband_freq_centres_mhz",
)
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(user_input, err_msgs)
_validate_spectral_averaging_factor_for_zoom(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.get("reference_coordinate") or user_input.get("pointing_centre"),
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 _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"
f" not lie within the {MIN_FREQUENCY_MHZ} - {MAX_FREQUENCY_MHZ} MHz range."
)
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(user_input: dict, err_msgs: list):
"""
Validates the total bandwidth of each zoom window against its spectral resolution
(channel width). The bandwidth can take any value that implies between MIN_ZOOM_CHANNELS
and MAX_ZOOM_CHANNELS channels at that channel width.
:param user_input: the parameters passed to the API.
:param err_msgs: the list of error messages to append a validation error to
"""
for spectral_resolution_hz, total_bandwidth_khz in zip(
user_input["spectral_resolutions_hz"], user_input["total_bandwidths_khz"]
):
min_bandwidth_khz = spectral_resolution_hz * MIN_ZOOM_CHANNELS * 1e-3
max_bandwidth_khz = spectral_resolution_hz * MAX_ZOOM_CHANNELS * 1e-3
if not (min_bandwidth_khz <= total_bandwidth_khz <= max_bandwidth_khz):
err_msgs.append(
f"Bandwidth {total_bandwidth_khz} kHz must be between"
f" {round(min_bandwidth_khz, 3)} kHz and {round(max_bandwidth_khz, 3)} kHz"
f" ({MIN_ZOOM_CHANNELS} to {MAX_ZOOM_CHANNELS} channels at a spectral"
f" resolution of {spectral_resolution_hz} Hz)."
)
def _validate_spectral_averaging_factor_for_zoom(user_input: dict, err_msgs: list):
"""
Validates the spectral averaging factor does not exceed half the number of channels of the
zoom window with the fewest channels.
:param user_input: the parameters passed to the API.
:param err_msgs: the list of error messages to append a validation error to
"""
n_channels_per_window = [
(total_bandwidth_khz * 1e3) / spectral_resolution_hz
for total_bandwidth_khz, spectral_resolution_hz in zip(
user_input["total_bandwidths_khz"], user_input["spectral_resolutions_hz"]
)
]
if not n_channels_per_window:
return
max_averaging_factor = int(min(n_channels_per_window) // 2)
valid = 1 <= user_input["spectral_averaging_factor"] <= max_averaging_factor
if not valid:
err_msgs.append(
f"The spectral averaging factor must lie between 1 and {max_averaging_factor}"
)
[docs]
def validate_and_set_defaults_for_sefd_lookup(user_input: dict) -> dict:
"""
Validate and convert parameters for SEFD lookup endpoint.
Ensures all required parameters are present, are numbers, and within allowed ranges.
Returns a dict with all parameters converted to the appropriate types.
Raises ValueError if any check fails.
"""
required = [
"start_freq_mhz",
"stop_freq_mhz",
"step_freq_mhz",
"az",
"el",
"start_lst",
"end_lst",
]
errors = []
out = {}
for key in required:
if key not in user_input:
errors.append(f"Missing required parameter: {key}")
else:
try:
out[key] = float(user_input[key])
except Exception:
errors.append(f"Parameter {key} must be a number")
if "num_stations" not in user_input:
errors.append("Missing required parameter: num_stations")
else:
try:
out["num_stations"] = int(user_input["num_stations"])
except Exception:
errors.append("num_stations must be an integer")
# Range checks
if "start_freq_mhz" in out and not (
MIN_FREQUENCY_MHZ <= out["start_freq_mhz"] <= MAX_FREQUENCY_MHZ
):
errors.append(
f"start_freq_mhz must be between {MIN_FREQUENCY_MHZ} and {MAX_FREQUENCY_MHZ}"
)
if "stop_freq_mhz" in out and not (
MIN_FREQUENCY_MHZ <= out["stop_freq_mhz"] <= MAX_FREQUENCY_MHZ
):
errors.append(
f"stop_freq_mhz must be between {MIN_FREQUENCY_MHZ} and {MAX_FREQUENCY_MHZ}"
)
if "step_freq_mhz" in out and not (0.1 <= out["step_freq_mhz"] <= 300):
errors.append("step_freq_mhz must be between 0.1 and 300")
if (
"start_freq_mhz" in out
and "stop_freq_mhz" in out
and out["start_freq_mhz"] > out["stop_freq_mhz"]
):
errors.append("start_freq_mhz must be less than or equal to stop_freq_mhz")
if "az" in out and not (0 <= out["az"] <= 360):
errors.append("az must be between 0 and 360")
if "el" in out and not (0 <= out["el"] <= 90):
errors.append("el must be between 0 and 90")
if "start_lst" in out and not (0 <= out["start_lst"] <= 24):
errors.append("start_lst must be between 0 and 24")
if "end_lst" in out and not (0 <= out["end_lst"] <= 24):
errors.append("end_lst must be between 0 and 24")
if "num_stations" in out and not (1 <= out["num_stations"] <= 512):
errors.append("num_stations must be between 1 and 512")
if errors:
raise ValueError("; ".join(errors))
return out