"""
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, Tuple, TypedDict, TypeVar, Union
from ska_ost_senscalc.common.service import WeightingResponse
from ska_ost_senscalc.low.service import (
SensitivityResponse,
convert_continuum_input_and_calculate,
convert_pss_input_and_calculate,
convert_zoom_input_and_calculate,
get_subarray_response,
get_weighting_response,
)
from ska_ost_senscalc.low.validation import (
validate_and_set_defaults_for_continuum,
validate_and_set_defaults_for_pss,
validate_and_set_defaults_for_zoom,
validate_weighting_params,
)
LOGGER = logging.getLogger("senscalc")
[docs]
class ErrorModelResponse(TypedDict):
"""
ErrorModelResponse is a typed dictionary representing the JSON object returned
by the API when validation of the request fails or there is another error.
"""
title: str
detail: str
T = TypeVar("T")
ResponseTuple = Tuple[Union[T, ErrorModelResponse], HTTPStatus]
[docs]
def calculate_continuum_sensitivity(**kwargs) -> ResponseTuple[SensitivityResponse]:
"""
Function which HTTP GET requests to /api/low/continuum/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:`ErrorModelResponse`) and HTTP status, which Connexion will wrap into a Response
"""
LOGGER.debug(
"Request received for LOW continuum sensitivity for input parameters %s",
kwargs,
)
try:
validated_params = validate_and_set_defaults_for_continuum(kwargs)
return (
convert_continuum_input_and_calculate(validated_params),
HTTPStatus.OK,
)
except ValueError as err:
return (
_validation_error_body(err),
HTTPStatus.BAD_REQUEST,
)
except Exception:
LOGGER.exception("Exception occurred with LOW calculate_sensitivity")
return (
ErrorModelResponse(detail="Internal Server Error"),
HTTPStatus.INTERNAL_SERVER_ERROR,
)
[docs]
def subarrays():
"""
Function that GET requests to the /api/low/subarrays are routed to.
Returns a response containing a list of available subarrays
"""
try:
LOGGER.debug("Request received for LOW subarrays")
return (
get_subarray_response(),
HTTPStatus.OK,
)
except Exception as err:
LOGGER.exception("Exception occurred with LOW api.subarrays")
return (
{
"title": "Internal Server Error",
"detail": repr(err),
},
HTTPStatus.INTERNAL_SERVER_ERROR,
)
[docs]
def calculate_zoom_sensitivity(**kwargs) -> ResponseTuple[SensitivityResponse]:
"""
Function which HTTP GET requests to /api/low/zoom/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:`ErrorModelResponse`) and HTTP status, which Connexion will wrap into a Response
"""
LOGGER.debug(
"Request received for LOW zoom sensitivity for input parameters %s",
kwargs,
)
try:
validated_params = validate_and_set_defaults_for_zoom(kwargs)
return (
convert_zoom_input_and_calculate(validated_params),
HTTPStatus.OK,
)
except ValueError as err:
return (
_validation_error_body(err),
HTTPStatus.BAD_REQUEST,
)
except Exception:
LOGGER.exception("Exception occurred with LOW calculate_sensitivity")
return (
ErrorModelResponse(detail="Internal Server Error"),
HTTPStatus.INTERNAL_SERVER_ERROR,
)
def _validation_error_body(err: ValueError) -> ErrorModelResponse:
"""
:param err: a ValueError with an arg for each separate validation error
:return: the error defined in the OpenAPI spec, in particular with details as a string
"""
return {
"title": "Validation Error",
"detail": ";".join(str(a) for a in err.args),
}
[docs]
def calculate_pss_sensitivity(**kwargs) -> ResponseTuple[SensitivityResponse]:
"""
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:`ErrorModelResponse`) and HTTP status, which Connexion will wrap into a Response
"""
LOGGER.debug(
"Request received for LOW PSS sensitivity for input parameters %s",
kwargs,
)
try:
validated_params = validate_and_set_defaults_for_pss(kwargs)
return (
convert_pss_input_and_calculate(validated_params),
HTTPStatus.OK,
)
except ValueError as err:
return (
_validation_error_body(err),
HTTPStatus.BAD_REQUEST,
)
except Exception:
LOGGER.exception("Exception occurred with LOW calculate_sensitivity")
return (
ErrorModelResponse(detail="Internal Server Error"),
HTTPStatus.INTERNAL_SERVER_ERROR,
)
[docs]
class WeightingCalculator:
"""
WeightingCalculator defines the entry point for image weighting
calculations. It serves to link the OpenAPI endpoints with the image
weighting algorithms.
"""
@staticmethod
def weighting(
*, # force kw-only args
spectral_mode: str,
freq_centre: float | int,
pointing_centre: str,
subarray_configuration: str,
weighting_mode: str,
robustness: int | None = None,
subband_frequencies_mhz: List[int] | None = None,
**other,
) -> ResponseTuple[WeightingResponse]:
if other:
LOGGER.warning(f"Unexpected query argument(s): {other}")
try:
try:
validated_params = validate_weighting_params(
spectral_mode=spectral_mode,
freq_centre=freq_centre,
pointing_centre=pointing_centre,
subarray_configuration=subarray_configuration,
weighting_mode=weighting_mode,
robustness=robustness,
)
except ValueError as e:
return (
dict(title="Bad Request", detail=f"{e.args}"),
HTTPStatus.BAD_REQUEST,
)
return (
get_weighting_response(validated_params, subband_frequencies_mhz),
HTTPStatus.OK,
)
except Exception as e:
LOGGER.exception("Exception occurred with LOW weighting", exc_info=e)
return (
dict(title="Internal Server Error", detail=str(e)),
HTTPStatus.INTERNAL_SERVER_ERROR,
)