Source code for ska_ost_senscalc.low.service
"""
The service layer is responsible for turning validated inputs into the relevant calculation inputs,
calling any calculation functions and collating the results.
"""
from copy import deepcopy
from typing import List, Optional, Tuple, TypedDict
from astropy.units import Hz, Quantity
from ska_ost_senscalc.common.model import (
CalculatorInputPSS,
LowSpectralMode,
WeightingInput,
)
from ska_ost_senscalc.common.pss import convert_continuum_to_bf_sensitivity
from ska_ost_senscalc.common.service import (
SubbandWeightingResponse,
WeightingResponse,
get_single_weighting_response,
get_subbands,
)
from ska_ost_senscalc.common.spectropolarimetry import (
SpectropolarimetryInput,
SpectropolarimetryResults,
get_spectropolarimetry_results,
)
from ska_ost_senscalc.low.bright_source_lookup import BrightSourceCatalog
from ska_ost_senscalc.low.calculator import calculate_sensitivity
from ska_ost_senscalc.low.model import CalculatorInput
from ska_ost_senscalc.low.validation import LOW_CONTINUUM_CHANNEL_WIDTH_KHZ
from ska_ost_senscalc.subarray import SubarrayStorage
from ska_ost_senscalc.utilities import Telescope
subarray_storage = SubarrayStorage(Telescope.LOW)
FLUX_DENSITY_THRESHOLD_JY = 10.0
[docs]
class SensitivityResponse(TypedDict):
"""
SensitivityResponse is a typed dictionary constrained to match the
schema of a single sensitivity calculation.
"""
continuum_sensitivity: Optional[Quantity]
continuum_subband_sensitivities: Optional[List[SubbandResponse]]
spectral_sensitivity: Optional[Quantity]
# For pulsar search:
folded_pulse_sensitivity: Optional[Quantity]
spectropolarimetry_results: SpectropolarimetryResults
warning: Optional[str]
[docs]
def convert_continuum_input_and_calculate(user_input: dict) -> SensitivityResponse:
"""
:param user_input: A kwarg dict of the HTTP parameters sent by the user
:return: a SensitivityResponse containing the calculated sensitivity and its units
"""
num_stations = _num_stations_from_input(user_input)
continuum_calculator_input = CalculatorInput(
freq_centre=user_input["freq_centre"],
bandwidth=user_input["bandwidth_mhz"],
num_stations=num_stations,
pointing_centre=user_input["pointing_centre"],
duration=user_input["duration"],
elevation_limit=user_input["elevation_limit"],
)
line_calculator_input = CalculatorInput(
freq_centre=user_input["freq_centre"],
bandwidth=(
(LOW_CONTINUUM_CHANNEL_WIDTH_KHZ / 1e3)
* user_input["spectral_averaging_factor"]
),
num_stations=num_stations,
pointing_centre=user_input["pointing_centre"],
duration=user_input["duration"],
elevation_limit=user_input["elevation_limit"],
)
spectropolarimetry_input = SpectropolarimetryInput(
bandwidth=Quantity(user_input["bandwidth_mhz"], "MHz"),
frequency=Quantity(user_input["freq_centre"], "MHz"),
effective_channel_width=Quantity(
LOW_CONTINUUM_CHANNEL_WIDTH_KHZ * user_input["spectral_averaging_factor"],
"kHz",
),
)
continuum_result, warning = _get_calculation_value(continuum_calculator_input)
# Warnings will be identical, so we don't need this one:
line_result, _ = _get_calculation_value(line_calculator_input)
subband_sensitivities = _get_subband_sensitivities(user_input, num_stations)
spectropolarimetry_results = get_spectropolarimetry_results(
spectropolarimetry_input
)
return SensitivityResponse(
continuum_sensitivity=continuum_result,
continuum_subband_sensitivities=subband_sensitivities,
spectral_sensitivity=line_result,
warning=warning,
spectropolarimetry_results=spectropolarimetry_results,
)
[docs]
def convert_zoom_input_and_calculate(user_input: dict) -> SensitivityResponse:
"""
:param user_input: A kwarg dict of the HTTP parameters sent by the user
:return: a dict containing the calculated sensitivity and its units
"""
num_stations = _num_stations_from_input(user_input)
calculator_input = CalculatorInput(
freq_centre=user_input["freq_centre"],
bandwidth=user_input["spectral_resolution_hz"] * 1e-6, # Convert to MHz
num_stations=num_stations,
pointing_centre=user_input["pointing_centre"],
duration=user_input["duration"],
elevation_limit=user_input["elevation_limit"],
)
spectropolarimetry_input = SpectropolarimetryInput(
bandwidth=Quantity(user_input["total_bandwidth_khz"], "kHz"),
frequency=Quantity(user_input["freq_centre"], "MHz"),
effective_channel_width=Quantity(user_input["spectral_resolution_hz"], "Hz"),
)
sensitivity, warning = _get_calculation_value(calculator_input)
spectropolarimetry_results = get_spectropolarimetry_results(
spectropolarimetry_input
)
return SensitivityResponse(
spectral_sensitivity=sensitivity,
warning=warning,
spectropolarimetry_results=spectropolarimetry_results,
)
[docs]
def convert_pss_input_and_calculate(user_input: dict) -> SensitivityResponse:
"""
:param user_input: A kwarg dict of the HTTP parameters sent by the user
:return: a dict containing the calculated sensitivity and its units
"""
num_stations = _num_stations_from_input(user_input)
calculator_input_pss = CalculatorInputPSS(
freq_centre=user_input["freq_centre"],
bandwidth=user_input["bandwidth_mhz"],
chan_width=user_input["spectral_resolution_hz"],
num_stations=num_stations,
pointing_centre=user_input["pointing_centre"],
duration=user_input["duration"],
elevation_limit=user_input["elevation_limit"],
dm=user_input["dm"],
pulse_period=user_input["pulse_period"],
intrinsic_pulse_width=user_input["intrinsic_pulse_width"],
)
# First, estimate the corresponding continuum sensitivity
continuum_input = CalculatorInput(
freq_centre=calculator_input_pss.freq_centre,
bandwidth=calculator_input_pss.chan_width * 1e-6, # Convert to MHz
num_stations=calculator_input_pss.num_stations,
pointing_centre=calculator_input_pss.pointing_centre,
duration=calculator_input_pss.duration,
elevation_limit=calculator_input_pss.elevation_limit,
)
continuum_sensitivity, warning = _get_calculation_value(continuum_input)
# Convert the continuum sensitivity to folded-pulse sensitivity
folded_pulse_sensitivity = convert_continuum_to_bf_sensitivity(
continuum_sensitivity,
calculator_input_pss,
)
return SensitivityResponse(
folded_pulse_sensitivity=folded_pulse_sensitivity, warning=warning
)
def get_weighting_response(
weighting_input: WeightingInput, subband_frequencies_mhz: List[int] | None = None
) -> WeightingResponse:
result = dict(**get_single_weighting_response(weighting_input))
if subband_frequencies_mhz is not None and len(subband_frequencies_mhz) > 1:
result["subbands"] = [
SubbandWeightingResponse(
subband_frequency=subband_frequency_mhz * 1e6 * Hz,
weighting=_get_weighting_for_subband(
subband_frequency_mhz, weighting_input
),
)
for subband_frequency_mhz in subband_frequencies_mhz
]
return result
[docs]
def get_subarray_response():
"""
return the appropriate subarray objects
"""
return [
{
"name": subarray.name,
"label": subarray.label,
"n_stations": subarray.n_stations,
}
for subarray in subarray_storage.list()
]
def _get_weighting_for_subband(
subband_frequency_mhz: list[int], weighting_input: WeightingInput
):
subband_weighting_input = deepcopy(weighting_input)
subband_weighting_input.freq_centre = [subband_frequency_mhz * 1e6 * Hz]
# continuum mode should always be assumed in
# the lookup table regardless of subband bandwidth
subband_weighting_input.calc_mode = LowSpectralMode.CONTINUUM
return get_single_weighting_response(subband_weighting_input)
def _get_calculation_value(
calculator_input: CalculatorInput,
) -> Tuple[Quantity, Optional[str]]:
result = calculate_sensitivity(calculator_input)
sensitivity = Quantity(result.sensitivity, result.units)
warning = _check_for_warning(calculator_input)
return sensitivity, warning
def _check_for_warning(calculator_input: CalculatorInput) -> Optional[str]:
mwa_cat = BrightSourceCatalog(threshold_jy=FLUX_DENSITY_THRESHOLD_JY)
if mwa_cat.check_for_bright_sources(
calculator_input.pointing_centre, calculator_input.freq_centre
):
return (
"The specified pointing contains at least one source brighter "
+ f"than {FLUX_DENSITY_THRESHOLD_JY} Jy. Your observation may be "
+ "dynamic range limited."
)
return None
def _num_stations_from_input(user_input: dict) -> int:
"""
If the user has given a subarray_configuration, extract the num_stations from that.
Otherwise, use the value given by the user.
Validation has checked that one and only on of these fields is present in the input.
:param user_input: a dict of the parameters given by the user
:return: the num_stations to use in the calculation
"""
if "subarray_configuration" in user_input:
subarray = subarray_storage.load_by_name(user_input["subarray_configuration"])
return subarray.n_stations
return user_input["num_stations"]
def _get_subband_sensitivities(
user_input: dict, num_stations: int
) -> List[SubbandResponse]:
if user_input.get("n_subbands", 1) == 1:
return []
subband_frequencies_mhz, subband_bandwidth = get_subbands(
user_input["n_subbands"], user_input["freq_centre"], user_input["bandwidth_mhz"]
)
return [
SubbandResponse(
subband_frequency=Quantity(subband_frequency, "MHz"),
sensitivity=_get_calculation_value(
CalculatorInput(
freq_centre=subband_frequency,
bandwidth=subband_bandwidth,
num_stations=num_stations,
pointing_centre=user_input["pointing_centre"],
duration=user_input["duration"],
elevation_limit=user_input["elevation_limit"],
)
)[0],
)
for subband_frequency in subband_frequencies_mhz
]