"""
These functions map to the API paths, with the returned value being the API response
Connexion maps the function name to the operationId in the OpenAPI document path
"""
import logging
from http import HTTPStatus
from typing import List, TypedDict
from ska_ost_senscalc.common.api import ResponseTuple, error_handler
from ska_ost_senscalc.common.model import (
ContinuumSensitivityCalculateResponse,
WeightingSpectralMode,
)
from ska_ost_senscalc.low.service import (
PSSSensitivityResponse,
ZoomSensitivityResponse,
convert_continuum_input_and_calculate,
convert_pss_input_and_calculate,
convert_zoom_input_and_calculate,
get_subarray_response,
lookup_a_over_t_service,
lookup_stokes_i_sefd_service,
)
from ska_ost_senscalc.low.validation import (
validate_and_set_defaults_for_calculate,
validate_and_set_defaults_for_pss,
validate_and_set_defaults_for_sefd_lookup,
)
[docs]
class SefdLookupItem(TypedDict):
freq_mhz: float
sefd: float
SefdLookupResponse = List[SefdLookupItem]
[docs]
class AOverTLookupItem(TypedDict):
freq_mhz: float
a_over_t: float
AOverTLookupResponse = List[AOverTLookupItem]
LOGGER = logging.getLogger("senscalc")
[docs]
@error_handler
def continuum_calculate_deprecated(
**kwargs,
) -> ResponseTuple[ContinuumSensitivityCalculateResponse]:
"""
HTTP GET request handler for /api/low/continuum/calculate endpoint.
Combines the request parameters for xcalculator and weighting calculations and returns a dictionary
with the sensitivities, the weighting continuum and spectral results, and the transformed results which
combine sensitivity with the weighting results for the respective spectral mode into a format easy to use
for the front-end.
:param kwargs: the HTTP parameters
:return: a tuple of the response body (which is either a ContinuumCalculateResponse or an ErrorResponse)
and HTTP status, which Connexion will wrap into a Response
"""
validated_and_updated_params = validate_and_set_defaults_for_calculate(
user_input=kwargs, spectral_mode=WeightingSpectralMode.CONTINUUM
)
response = convert_continuum_input_and_calculate(validated_and_updated_params)
return (
response,
HTTPStatus.OK,
)
[docs]
@error_handler
def continuum_calculate(
**kwargs,
) -> ResponseTuple[List[ContinuumSensitivityCalculateResponse]]:
"""
HTTP POST request handler for /api/low/continuum/calculate endpoint.
Combines the request parameters for calculator and weighting calculations and returns a
list of dictionaries, each with the sensitivities, the weighting continuum and spectral
results, and the transformed results which combine sensitivity with the weighting results
for the respective spectral mode into a format easy to use for the front-end - one entry
per reference_coordinate item supplied in the request, in the same order.
:param kwargs: the request body
:return: a tuple of the response body (a list of ContinuumCalculateResponse, or an
ErrorResponse) and HTTP status, which Connexion will wrap into a Response
"""
user_input = kwargs["body"]
reference_coordinates = user_input.pop("reference_coordinates", None)
if reference_coordinates is not None:
per_item_inputs = [
{**user_input, "reference_coordinate": reference_coordinate}
for reference_coordinate in reference_coordinates
]
else:
per_item_inputs = [user_input]
responses = []
for item_input in per_item_inputs:
validated_and_updated_params = validate_and_set_defaults_for_calculate(
user_input=item_input, spectral_mode=WeightingSpectralMode.CONTINUUM
)
responses.append(
convert_continuum_input_and_calculate(validated_and_updated_params)
)
return (
responses,
HTTPStatus.OK,
)
[docs]
@error_handler
def subarrays():
"""
Function that GET requests to the /api/low/subarrays are routed to.
Returns a response containing a list of available subarrays
"""
return (
get_subarray_response(),
HTTPStatus.OK,
)
[docs]
@error_handler
def zoom_calculate_deprecated(
**kwargs,
) -> ResponseTuple[ZoomSensitivityResponse]:
"""
HTTP GET request handler for /api/low/zoom/calculate endpoint. Combines the request parameters from
calculate and weighting calls and creates a dictionary with
the responses from the original combined requests, so it can be used to calculate the transformed results for low
zoom. It then combines and transforms the responses into the results returned to the UI, which is returned along the weighting
response and the warnings from the original calculate response.
:param kwargs: the HTTP parameters
:return: a tuple of the response body (which is either a :class:`ska_ost_senscalc.low.service.ZoomSensitivityResponse`
or an :class:`ErrorResponse`) and HTTP status, which Connexion will wrap into a Response
"""
validated = validate_and_set_defaults_for_calculate(
user_input=kwargs, spectral_mode=WeightingSpectralMode.LINE
)
response = convert_zoom_input_and_calculate(validated)
return response, HTTPStatus.OK
[docs]
@error_handler
def zoom_calculate(
**kwargs,
) -> ResponseTuple[List[ContinuumSensitivityCalculateResponse]]:
"""
HTTP POST request handler for /api/low/zoom/calculate endpoint.
Combines the request parameters for calculator and weighting calculations and returns a
list of dictionaries, each with the sensitivities, the weighting continuum and spectral
results, and the transformed results which combine sensitivity with the weighting results
for the respective spectral mode into a format easy to use for the front-end - one entry
per reference_coordinate item supplied in the request, in the same order.
:param kwargs: the request body
:return: a tuple of the response body (a list of ContinuumCalculateResponse, or an
ErrorResponse) and HTTP status, which Connexion will wrap into a Response
"""
user_input = kwargs["body"]
reference_coordinates = user_input.pop("reference_coordinates", None)
if reference_coordinates is not None:
per_item_inputs = [
{**user_input, "reference_coordinate": reference_coordinate}
for reference_coordinate in reference_coordinates
]
else:
per_item_inputs = [user_input]
responses = []
for item_input in per_item_inputs:
validated_and_updated_params = validate_and_set_defaults_for_calculate(
user_input=item_input, spectral_mode=WeightingSpectralMode.LINE
)
responses.append(convert_zoom_input_and_calculate(validated_and_updated_params))
return (
responses,
HTTPStatus.OK,
)
[docs]
@error_handler
def pss_calculate(**kwargs) -> ResponseTuple[PSSSensitivityResponse]:
"""
Function which HTTP GET requests to /api/low/pss/calculate are routed to.
:param kwargs: the HTTP parameters
:return: a tuple of the response body (which is either a :class:`ska_ost_senscalc.low.service.SensitivityResponse`
or an :class:`ErrorResponse`) and HTTP status, which Connexion will wrap into a Response
"""
validated_params = validate_and_set_defaults_for_pss(kwargs)
return (
convert_pss_input_and_calculate(validated_params),
HTTPStatus.OK,
)
[docs]
@error_handler
def sefd_lookup(**kwargs) -> ResponseTuple[SefdLookupResponse]:
"""
Function which HTTP GET requests to /api/low/sefd/lookup are routed to.
:param kwargs: the HTTP parameters (expects start_freq_mhz, stop_freq_mhz, num_stations, az, el, start_lst, end_lst)
:return: a tuple of the response body (list of {freq_mhz, sefd}) and HTTP status
"""
params = validate_and_set_defaults_for_sefd_lookup(kwargs)
result = lookup_stokes_i_sefd_service(**params)
return result, HTTPStatus.OK
[docs]
@error_handler
def a_over_t_lookup(**kwargs) -> ResponseTuple[AOverTLookupResponse]:
"""
Function which HTTP GET requests to /api/low/aovert/lookup are routed to.
:param kwargs: the HTTP parameters (expects start_freq_mhz, stop_freq_mhz, num_stations, az, el, start_lst, end_lst)
:return: a tuple of the response body (list of {freq_mhz, a_over_t}) and HTTP status
"""
params = validate_and_set_defaults_for_sefd_lookup(kwargs)
result = lookup_a_over_t_service(**params)
return result, HTTPStatus.OK