"""
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 dataclasses import asdict
import astropy.units as u
import numpy
from astropy.coordinates import SkyCoord
from ska_ost_senscalc.common.model import (
CalculatorInputPSS,
ContinuumRequest,
PssRequestMid,
PulsarSamplingTime,
Weighting,
ZoomRequest,
ZoomRequestPrepared,
)
from ska_ost_senscalc.common.service import sub_band_to_frequency_array
from ska_ost_senscalc.mid.calculator import DEFAULT_ALPHA, DEFAULT_EL, DEFAULT_PWV
from ska_ost_senscalc.subarray import MIDArrayConfiguration, SubarrayStorage
from ska_ost_senscalc.utilities import Telescope
subarray_storage = SubarrayStorage(Telescope.MID)
DEFAULT_CALCULATE_PARAMS = {
"pwv": DEFAULT_PWV,
"el": DEFAULT_EL.value,
"alpha": DEFAULT_ALPHA,
"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,
MIDArrayConfiguration.MID_inner_r2km_aa4,
MIDArrayConfiguration.MID_inner_r2km_aastar,
MIDArrayConfiguration.MID_inner_r20km_aa4,
MIDArrayConfiguration.MID_inner_r20km_aastar,
MIDArrayConfiguration.MID_inner_r125m_aa4,
MIDArrayConfiguration.MID_inner_r125m_aastar,
MIDArrayConfiguration.MID_inner_r500m_aa4,
MIDArrayConfiguration.MID_inner_r500m_aastar,
]
# ToDo: Update the following two with a more precise value when known
MID_CONTINUUM_CHANNEL_WIDTH_KHZ = 13.44
PSS_CHAN_WIDTH_HZ = 107.5e3
PSS_MAX_BANDWIDTH_HZ = 300e6
PSS_DEFAULT_DM = 0.0
PSS_DEFAULT_PULSE_PERIOD = 33.0 # Assume crab as default (33 ms)
PSS_DEFAULT_PULSE_WIDTH = 0.004 # Assume crab as default (4 us)
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_continuum(
params: ContinuumRequest,
) -> ContinuumRequest:
"""
Validate arguments for a MID Continuum query, returning a typed
encapsulation of those arguments.
"""
params.telescope = Telescope.MID
err_msgs = []
params.pointing_centre = _validate_pointing_centre(params.pointing_centre, err_msgs)
params.weighting_mode = Weighting(params.weighting_mode)
if params.subarray_configuration:
# TODO: handle number of antennas while weighting requires subarray config
params.subarray_configuration = MIDArrayConfiguration(
params.subarray_configuration
)
# TODO: add validation test for all modes
_validate_n_subbands_for_subbands_frequencies(
params.subband_freq_centres_hz, params.n_subbands, err_msgs
)
params.subband_freq_centres_hz = _get_subband_freq_centres_for_n_subbands(
params, err_msgs
)
params.subband_freq_centres_hz = [
subband_freq_centre_hz * u.Hz
for subband_freq_centre_hz in params.subband_freq_centres_hz
]
params.freq_centre = params.freq_centre_hz * u.Hz
params.taper = params.taper * u.arcsec
if (
params.integration_time_s is not None
and params.supplied_sensitivity is not None
):
msg = "Either 'supplied_sensitivity' or 'integration_time_s' must be specified, they are mutually exclusive."
raise ValueError(msg)
d_params = asdict(params)
_validate_array_config_and_antennas_check_none(d_params)
_validate_continuum_bandwidth_for_array_config(d_params)
_validate_spectral_window(params)
_validate_subband_parameters(d_params)
if err_msgs:
raise ValueError("; ".join(err_msgs))
return params
[docs]
def validate_and_set_defaults_for_zoom(params: ZoomRequest) -> ZoomRequestPrepared:
"""
:param user_input: the parameters from the HTTP request to /zoom/calculate
:return: A new copy of the ZoomRequestPrepared, with defaults set for missing values
:raises: ValueError if the input data is not valid
"""
if (params.integration_time_s is None) == (params.supplied_sensitivities is None):
raise ValueError(
"Either 'supplied_sensitivities' or 'integration_time_s' must be specified, they are mutually exclusive."
)
_validate_array_config_and_antennas(params)
_validate_zoom_parameters(params)
_validate_spectral_window(params)
validated = validate_and_convert_zoom_weighting_params(params)
return validated
[docs]
def validate_and_set_defaults_for_pss(kwargs: PssRequestMid) -> CalculatorInputPSS:
"""
Validate arguments for a MID PSS query and set appropriate defaults.
Note that PSS does not have to perform an extensive validation as continuum.
This is because PSS internally calls the continuum mode and then
transforms the continuum sensitivity to get PSS sensitivity.
So, we will only do minimal validation here.
"""
err_msgs = []
# API has already checked that
# 1. the specified subarray is valid for PSS
# 2. the bandwidth, if specified, is less than or equal to 300 MHz
# Here, check only for the PSS mode-specific inputs. All other validation
# will be performed by the continuum mode.
if kwargs.pulsar_mode == "folded_pulse":
# Ensure that when in folded-pulse mode, bandwidth is always 300 MHz
if not numpy.isclose(
kwargs.bandwidth_hz, PSS_MAX_BANDWIDTH_HZ, rtol=0, atol=1e-6
):
err_msgs.append("In folded-pulse mode, bandwidth must be 300 MHz.")
# kwargs.integration_time_s = kwargs.integration_time_s * u.s
# If the pulsar_mode is set to single_pulse, edit the integration time so that
# sensitivity is estimated not for the entire duration of the observation but
# for a single time resolution element.
if kwargs.pulsar_mode == "single_pulse":
kwargs.integration_time_s = PulsarSamplingTime.LOW_PSS.value.to(u.s).value
kwargs.pointing_centre = _validate_pointing_centre(kwargs.pointing_centre, err_msgs)
if err_msgs:
raise ValueError("; ".join(err_msgs))
return kwargs
[docs]
def validate_and_convert_zoom_weighting_params(
params: ZoomRequest,
) -> ZoomRequestPrepared:
"""
TODO the validation for weighting is different to the other calculations, in that it converts
the input to an object with astropy qualities, etc. We should unify the approaches, along with
handling defaults properly and consistently
Validate arguments for a MID weighting query, returning a typed
encapsulation of those arguments.
"""
err_msgs = []
pointing_centre = _validate_pointing_centre(params.pointing_centre, err_msgs)
if params.weighting_mode == Weighting.ROBUST and params.robustness is None:
# TODO: might not be needed in fact as we demand robustness in openapi and ZoomRequest
err_msgs.append("Parameter 'robustness' should be set for 'robust' weighting")
if err_msgs:
raise ValueError("; ".join(err_msgs))
freq_centres = [
u.Quantity(freq_centre_hz, unit=u.Hz)
for freq_centre_hz in params.freq_centres_hz
]
new = ZoomRequestPrepared(
**asdict(params),
telescope=Telescope.MID,
)
new.pointing_centre = pointing_centre
new.freq_centres = freq_centres
new.taper = params.taper * u.arcsec
new.weighting_mode = Weighting(params.weighting_mode)
return new
def _validate_pointing_centre(pointing_centre: str, err_msgs: list) -> SkyCoord:
try:
return SkyCoord(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]."
)
return None
def _validate_zoom_parameters(params: ZoomRequest) -> None:
"""
:param params: the parameters from the HTTP request
:raises: ValueError if the input data relevant for zoom mode is not valid
"""
# 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(params.freq_centres_hz),
len(params.spectral_resolutions_hz),
len(params.total_bandwidths_hz),
}
# 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:
raise ValueError(
"Parameters 'freq_centres_hz', 'spectral_resolutions_hz' and 'total_bandwidths_hz' must all be set together and have the same length."
)
if params.supplied_sensitivities:
if len(params.supplied_sensitivities) != next(iter(set_of_lengths)):
raise ValueError(
"Parameter 'supplied_sensitivities' must be set to calculate an integration time for the zoom window. It should have the same length as 'freq_centres_hz', 'spectral_resolutions_hz' and 'total_bandwidths_hz'."
)
if params.freq_centres_hz or params.spectral_resolutions_hz:
array_configuration = (
params.subarray_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 params.freq_centres_hz:
# Check that freq_centres_hz has the same length as spectral_resolutions_hz
if not params.spectral_resolutions_hz:
raise ValueError(
"Parameter 'spectral_resolutions_hz' must also be set when setting"
" 'freq_centres_hz'."
)
if len(params.freq_centres_hz) != len(params.spectral_resolutions_hz):
raise ValueError(
"Parameters 'spectral_resolutions_hz' and 'freq_centres_hz' must"
" have the same length."
)
elif params.spectral_resolutions_hz:
raise ValueError(
"Parameter 'freq_centres_hz' must also be set when setting"
" 'spectral_resolutions_hz'."
)
def _validate_subband_parameters(user_input: dict, err_msgs=None) -> 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("supplied_sensitivity"):
# Validation currently only needs to be done for the sensitivity -> integration time calculation
return
n_subbands = user_input.get("n_subbands", 1)
subband_supplied_sensitivities = user_input.get("subband_supplied_sensitivities")
if subband_supplied_sensitivities and n_subbands <= 1:
msg = "Parameter 'n_subbands' must be greater than 1 when setting 'subband_supplied_sensitivities' and 'supplied_sensitivity'."
raise ValueError(msg)
if (
n_subbands
and subband_supplied_sensitivities
and n_subbands != len(subband_supplied_sensitivities)
):
msg = "Parameter 'subband_supplied_sensitivities' must have the same length as the value of 'n_subbands' for 'n_subbands' greater than 1."
raise ValueError(msg)
if user_input["subarray_configuration"] is None:
if n_subbands > 1 and not subband_supplied_sensitivities:
raise ValueError(
"Parameter 'subband_supplied_sensitivities' must be set when setting 'supplied_sensitivity' and 'n_subbands' is greater than 1 for a custom subarray."
)
def _validate_array_config_and_antennas(params: ZoomRequest):
"""
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 = params.n_ska is not None and params.n_meer is None
n_meer_but_not_n_ska = params.n_meer is not None and params.n_ska is None
one_of_array_config_and_n_antennas = (
params.subarray_configuration is not None
) != (params.n_ska is not None or params.n_meer is not None)
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 params.subarray_configuration is not None:
params.subarray_configuration = MIDArrayConfiguration(
params.subarray_configuration
)
# TODO: rename to reflect the meaning of the function
def _validate_array_config_and_antennas_check_none(user_input, err_msgs=None):
"""
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 = (
user_input["n_ska"] is not None and user_input["n_meer"] is None
)
n_meer_but_not_n_ska = (
user_input["n_meer"] is not None and user_input["n_ska"] is None
)
one_of_array_config_and_n_antennas = (
user_input["subarray_configuration"] is not None
) != (user_input["n_ska"] is not None or user_input["n_meer"] is not None)
if (
n_ska_but_not_n_meer
or n_meer_but_not_n_ska
or not one_of_array_config_and_n_antennas
):
err_msg = "Only 'array_configuration' or the number of antennas ('n_ska' AND 'n_meer') should be specified."
if err_msgs is not None:
err_msgs.append(err_msg)
else:
raise ValueError(err_msg)
def _validate_continuum_bandwidth_for_array_config(
user_input: dict, err_msgs=None
) -> 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 (
"subarray_configuration" in user_input
and user_input["subarray_configuration"] in MAXIMUM_BANDWIDTH_FOR_SUBARRAY
):
max_continuum_bandwidth_hz = MAXIMUM_BANDWIDTH_FOR_SUBARRAY[
user_input["subarray_configuration"]
]
if user_input["bandwidth_hz"] > max_continuum_bandwidth_hz:
err_msg = f"Maximum bandwidth ({max_continuum_bandwidth_hz * 1e-6} MHz) for this subarray has been exceeded."
if err_msgs is not None:
err_msgs.append(err_msg)
raise ValueError(err_msg)
def _validate_continuum_bandwidth_for_array_config_check_none(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 (
"subarray_configuration" in user_input
and user_input["subarray_configuration"] in MAXIMUM_BANDWIDTH_FOR_SUBARRAY
):
max_continuum_bandwidth_hz = MAXIMUM_BANDWIDTH_FOR_SUBARRAY[
user_input["subarray_configuration"]
]
if user_input["bandwidth_hz"] > 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(
params: ZoomRequest | ContinuumRequest | PssRequestMid, err_msgs: list = None
) -> None:
"""
For continuum input or each zoom window within a zoom input,
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 params.subarray_configuration:
subarray = subarray_storage.load_by_label(params.subarray_configuration.value)
n_ska = subarray.n_ska
n_meer = subarray.n_meer
else:
n_ska = params.n_ska
n_meer = params.n_meer
antenna_type = (
"mixed" if n_ska > 0 and n_meer > 0 else ("ska" if n_ska > 0 else "meerkat")
)
try:
if (b := params.rx_band) not in BAND_LIMITS:
raise ValueError(f"{b} not supported.")
limits = next(
filter(
lambda entry: entry["type"] == antenna_type,
BAND_LIMITS[b],
)
)["limits"]
except StopIteration:
# This means the next function raised an error as the 'type' is not present in the band.
msg = "Subarray configuration not allowed for given observing band."
if type(err_msgs) == list:
err_msgs.append(msg)
else:
raise ValueError(msg)
def validate_within_band(freq_centre_hz, bandwidth_hz):
min_freq = freq_centre_hz - bandwidth_hz / 2
max_freq = freq_centre_hz + bandwidth_hz / 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."
)
# This function is used to validate both the continuum and zoom inputs
# This block has to handle various parameters
if hasattr(params, "freq_centre_hz"):
validate_within_band(params.freq_centre_hz, params.bandwidth_hz)
elif hasattr(params, "freq_centres_hz"):
[
validate_within_band(freq_centre_hz, total_bandwidth_hz)
for freq_centre_hz, total_bandwidth_hz in zip(
params.freq_centres_hz, params.total_bandwidths_hz
)
]
else:
raise ValueError(
"Neither freq_centre_hz nor freq_centres_hz was present in the request parameters"
)
raise ValueError("Not supported validation parameters")
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 is provided, Parameter 'n_subbands' should be set and match the number of subband_freq_centres."
)
def _get_subband_freq_centres_for_n_subbands(
user_input: ContinuumRequest, err_msgs: list
):
# if the user has entered a num of subbands bigger than 1
if user_input.n_subbands > 1:
# if provided, return the subband_freq_centres_hz entered by the user
if user_input.subband_freq_centres_hz:
return user_input.subband_freq_centres_hz
# 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_hz,
user_input.freq_centre_hz,
)
# 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 user_input.n_subbands == 1:
return (
user_input.subband_freq_centres_hz
if user_input.subband_freq_centres_hz
else []
)