"""
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
from ska_ost_senscalc.low.model import LowSpectralMode, WeightingQuery
from ska_ost_senscalc.mid_utilities import Weighting
from ska_ost_senscalc.subarray import LOWArrayConfiguration
DEFAULT_CONTINUUM_PARAMS = {
"num_stations": 512,
"freq_centre": 200,
"bandwidth_mhz": 300,
"pointing_centre": "10:00:00 -30:00:00",
"duration": 1,
}
DEFAULT_ZOOM_PARAMS = {
"num_stations": 512,
"freq_centre": 200,
"total_bandwidth_khz": 24.414, # Take the default value as the total bandwidth of the narrowest zoom window
"spectral_resolution_hz": 14.1285, # Take the narrowest channel allowed in zoom mode
"pointing_centre": "10:00:00 -30:00:00",
"duration": 1,
}
# 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)
]
[docs]def validate_and_set_defaults_for_continuum(user_input: dict) -> dict:
"""
:param user_input: the parameters from the HTTP request for the /api/low/continuum/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, ie if the user_input contains a key then it will not be overwritten by the defaults
user_input = DEFAULT_CONTINUUM_PARAMS | user_input
err_msgs = []
_validate_spectral_window(
user_input["freq_centre"], user_input["bandwidth_mhz"], err_msgs
)
_validate_pointing_centre(user_input, err_msgs)
if err_msgs:
raise ValueError(*err_msgs)
return user_input
[docs]def validate_and_set_defaults_for_zoom(user_input: dict) -> dict:
"""
:param user_input: the parameters from the HTTP request for the /api/low/zoom/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, ie if the user_input contains a key then it will not be overwritten by the defaults
user_input = DEFAULT_ZOOM_PARAMS | user_input
err_msgs = []
if (
not round(user_input["total_bandwidth_khz"], BANDWIDTH_PRECISION_KHZ)
in ALLOWED_ZOOM_TOTAL_BANDWIDTHS_KHZ
):
err_msgs.append(
f"Bandwidth {user_input['total_bandwidth_khz']} not one the of allowed"
f" values in zoom mode: {ALLOWED_ZOOM_TOTAL_BANDWIDTHS_KHZ}"
)
_validate_spectral_window(
user_input["freq_centre"],
user_input["total_bandwidth_khz"] * 1e-3,
err_msgs,
)
_validate_pointing_centre(user_input, err_msgs)
if err_msgs:
raise ValueError(*err_msgs)
return user_input
[docs]def validate_weighting_params(
*, # force kw-only args
spectral_mode: str,
freq_centre: float | int,
pointing_centre: str,
subarray_configuration: str,
weighting_mode: str,
robustness: int | None,
) -> WeightingQuery:
"""
Validate arguments for a LOW weighting query, returning a typed
encapsulation of those arguments.
"""
err_msgs = []
# TODO convert this and other validation functions to accept explicit
# typed kwargs - too large a refactoring at the moment
py_pointing_centre = _validate_pointing_centre(
dict(pointing_centre=pointing_centre), err_msgs
)
# OpenAPI enum strings need converting to Python enum members
py_weighting_mode = EnumConversion.to_weighting(weighting_mode, err_msgs)
py_calculator_mode = EnumConversion.to_lowcalculatormode(spectral_mode, err_msgs)
py_subarray_configuration = EnumConversion.to_array_configuration(
subarray_configuration, err_msgs
)
if py_weighting_mode == Weighting.ROBUST and robustness is None:
err_msgs.append(
"Robustness must be provided when weighting mode is set to robust"
)
if err_msgs:
raise ValueError("; ".join(err_msgs))
# OpenAPI spec requires frequency in MHz, Beam expects it in Hz
freq_mhz = Quantity(freq_centre * 1e6, unit=u.Hz)
return WeightingQuery(
spectral_mode=py_calculator_mode,
freq_centre=freq_mhz,
pointing_centre=py_pointing_centre,
subarray_configuration=py_subarray_configuration,
weighting_mode=py_weighting_mode,
robustness=robustness,
)
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 < 50 or max_freq > 350:
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]."
)
[docs]class EnumConversion:
"""
Utility class to convert OpenAPI enumeration members to Python Enum
members.
OpenAPI enums generally map to Python enums but their name will usually
be formatted differently, as the Python convention is for enumeration
member names to be all upper case. This class decouples the OpenAPI naming
convention from that all-caps requirement.
"""
@staticmethod
def to_weighting(val: str, msgs: list[str]) -> Weighting:
return EnumConversion._convert(Weighting, val, msgs)
@staticmethod
def to_lowcalculatormode(val: str, msgs) -> LowSpectralMode:
return EnumConversion._convert(LowSpectralMode, val, msgs)
@staticmethod
def to_array_configuration(val: str, msgs) -> LOWArrayConfiguration:
return EnumConversion._convert(LOWArrayConfiguration, val, msgs)
@staticmethod
def _convert(cls, val: str, msgs: list[str]):
try:
return cls[val.upper()]
except (ValueError, KeyError):
msg = f"{val} could not be mapped to a {cls.__name__} enum member"
msgs.append(msg)