"""
Unit tests for the sensitivity_calculator.frontend_adapter module
"""
import numpy as np
import astropy.units as u
from astropy.coordinates import SkyCoord
import pytest
from sensitivity_calculator.frontend_adapter import (
calculate_continuum_int_time,
calculate_continuum_sensitivity,
calculate_line_int_time,
calculate_line_sensitivity,
get_chunks,
translate_backend_state,
construct_frontend_result,
map_calculator_to_fields,
)
from sensitivity_calculator.mid import MidCalculator
target_gc = SkyCoord(l=0.0 * u.deg, b=0.0 * u.deg, frame="galactic")
target_cena = SkyCoord(frame="icrs", ra="13h25m27.6s", dec="−43d01m09s")
[docs]def test_default_continuum_sensitivity():
"""This is an end to end testing of the SC ContinuumSensitivity
component. A range of possible inputs are tried and the results
compared with those calculated directly.
'Expert' values are not set, so the SC will calculate the
standard defaults where necessary.
"""
# simulate some variation in input parameters across bands
obsfreq = {"Band 1": 0.7e9, "Band 2": 1.35e9, "Band 5a": 6.5e9, "Band 5b": 11.9e9}
bandwidth = {"Band 1": 0.7e9, "Band 2": 0.4e9, "Band 5a": 0.8e9, "Band 5b": 1.2e9}
chunk_centres = {
"Band 1": [0.7e9],
"Band 2": [1.2167e9, 1.35e9, 1.4833e9],
"Band 5a": [6.5e9],
"Band 5b": [11.6e9, 12.2e9],
}
chunk_width = {
"Band 1": 0.7e9,
"Band 2": 0.133333e9,
"Band 5a": 0.8e9,
"Band 5b": 0.6e9,
}
continuum_resolution = {
"Band 1": 0.21e3,
"Band 2": 0.21e3,
"Band 5a": 6.72e3,
"Band 5b": 1.68e3,
}
array_config = "full"
weathers = ["Good", "Average", "Bad"]
elevation = 90.0
band = "Band 1"
int_time = 10.0
# get result via frontend. Setting 'expert' parameters to None will make the SC
# generate default values
result = calculate_continuum_sensitivity(
target_lon=target_gc.icrs.ra.hourangle,
target_lat=target_gc.icrs.dec.degree,
target_frame="icrs",
obs_band=band,
continuum_obs_freq_scaled=obsfreq[band],
continuum_bandwidth_scaled=bandwidth[band],
weathers=weathers,
elevation=elevation,
array_config=array_config,
continuum_int_time_scaled=int_time,
eta_system=None,
eta_point=None,
eta_coherence=None,
eta_digitisation=None,
eta_correlation=None,
eta_bandpass=None,
n_ska=None,
eta_dish_ska=None,
Tsys_ska=None,
Tspl_ska=None,
Trcv_ska=None,
n_meer=None,
eta_dish_meer=None,
Tsys_meer=None,
Tspl_meer=None,
Trcv_meer=None,
Tsky=None,
Tgal=None,
alpha=None,
chunk_centres=chunk_centres[band],
chunk_width=chunk_width[band],
continuum_resolution=continuum_resolution[band],
)
# now duplicate results by running the SC directly
direct_result = {}
for weather in weathers:
mc = MidCalculator(
band,
obsfreq[band] * u.Hz,
bandwidth[band] * u.Hz,
array_config,
target_gc,
weather,
elevation * u.deg,
)
cont_sens = mc.calculate_sensitivity(int_time * u.s)
backend_state = mc.state()
chunk_mc = MidCalculator(
band,
chunk_centres[band] * u.Hz,
chunk_width[band] * u.Hz,
array_config,
target_gc,
weather,
elevation * u.deg,
)
chunk_sens = chunk_mc.calculate_sensitivity(int_time * u.s)
line_mc = MidCalculator(
band,
obsfreq[band] * u.Hz,
continuum_resolution[band] * u.Hz,
array_config,
target_gc,
weather,
elevation=elevation * u.deg,
)
line_sens = line_mc.calculate_sensitivity(int_time * u.s)
frontend_state = translate_backend_state(backend_state)
weather_result = construct_frontend_result(
frontend_state,
cont_sens=cont_sens,
chunk_sens=chunk_sens,
line_sens=line_sens,
chunk_centres=chunk_centres[band] * u.Hz,
chunk_width=chunk_width[band] * u.Hz,
)
direct_result[weather] = weather_result
# print("result 1", result)
# print("direct_result 1", direct_result)
assert result == direct_result
# now try with 'commissioning' Tsys set
Tsys = 25.0
result = calculate_continuum_sensitivity(
target_lon=target_gc.icrs.ra.hourangle,
target_lat=target_gc.icrs.dec.degree,
target_frame="icrs",
obs_band=band,
continuum_obs_freq_scaled=obsfreq[band],
continuum_bandwidth_scaled=bandwidth[band],
weathers=weathers,
elevation=elevation,
array_config=array_config,
continuum_int_time_scaled=int_time,
eta_system=None,
eta_point=None,
eta_coherence=None,
eta_digitisation=None,
eta_correlation=None,
eta_bandpass=None,
n_ska=None,
eta_dish_ska=None,
Tsys_ska=Tsys,
Tspl_ska=None,
Trcv_ska=None,
n_meer=None,
eta_dish_meer=None,
Tsys_meer=Tsys,
Tspl_meer=None,
Trcv_meer=None,
Tsky=None,
Tgal=None,
alpha=None,
chunk_centres=chunk_centres[band],
chunk_width=chunk_width[band],
continuum_resolution=continuum_resolution[band],
)
# now duplicate results by running the SC directly
direct_result = {}
for weather in weathers:
mc = MidCalculator(
band,
obsfreq[band] * u.Hz,
bandwidth[band] * u.Hz,
array_config,
target_gc,
weather,
elevation * u.deg,
Tsys_ska=Tsys * u.K,
Tsys_meer=Tsys * u.K,
)
cont_sens = mc.calculate_sensitivity(int_time * u.s)
backend_state = mc.state()
chunk_mc = MidCalculator(
band,
chunk_centres[band] * u.Hz,
chunk_width[band] * u.Hz,
array_config,
target_gc,
weather,
elevation * u.deg,
Tsys_ska=Tsys * u.K,
Tsys_meer=Tsys * u.K,
)
chunk_sens = chunk_mc.calculate_sensitivity(int_time * u.s)
line_mc = MidCalculator(
band,
obsfreq[band] * u.Hz,
continuum_resolution[band] * u.Hz,
array_config,
target_gc,
weather,
elevation * u.deg,
Tsys_ska=Tsys * u.K,
Tsys_meer=Tsys * u.K,
)
line_sens = line_mc.calculate_sensitivity(int_time * u.s)
frontend_state = translate_backend_state(backend_state)
weather_result = construct_frontend_result(
frontend_state,
cont_sens=cont_sens,
chunk_sens=chunk_sens,
line_sens=line_sens,
chunk_centres=chunk_centres[band] * u.Hz,
chunk_width=chunk_width[band] * u.Hz,
)
direct_result[weather] = weather_result
# print("result 2", result)
# print("direct result 2", direct_result)
assert result == direct_result
# now try with some other expert values set
Trcv = 10.0
Tspl = 20.0
Tsky = 15.0
Tgal = 50.0
alpha = 2.75
eta_dish_ska = 0.9
eta_dish_meer = 0.7
n_ska = 50
n_meer = 20
result = calculate_continuum_sensitivity(
target_lon=target_gc.icrs.ra.hourangle,
target_lat=target_gc.icrs.dec.degree,
target_frame="icrs",
obs_band=band,
continuum_obs_freq_scaled=obsfreq[band],
continuum_bandwidth_scaled=bandwidth[band],
weathers=weathers,
elevation=elevation,
array_config=array_config,
continuum_int_time_scaled=int_time,
eta_system=None,
eta_point=None,
eta_coherence=None,
eta_digitisation=None,
eta_correlation=None,
eta_bandpass=None,
n_ska=n_ska,
eta_dish_ska=eta_dish_ska,
Tsys_ska=None,
Tspl_ska=Tspl,
Trcv_ska=Trcv,
n_meer=n_meer,
eta_dish_meer=eta_dish_meer,
Tsys_meer=None,
Tspl_meer=Tspl,
Trcv_meer=Trcv,
Tsky=Tsky,
Tgal=Tgal,
alpha=alpha,
chunk_centres=chunk_centres[band],
chunk_width=chunk_width[band],
continuum_resolution=continuum_resolution[band],
)
# now duplicate results by running the SC directly
direct_result = {}
for weather in weathers:
mc = MidCalculator(
band,
obsfreq[band] * u.Hz,
bandwidth[band] * u.Hz,
array_config,
target_gc,
weather,
elevation * u.deg,
Trcv_ska=10.0 * u.K,
Trcv_meer=10.0 * u.K,
Tspl_ska=20.0 * u.K,
Tspl_meer=20.0 * u.K,
Tsky=15.0 * u.K,
Tgal=50.0 * u.K,
alpha=2.75,
eta_dish_ska=0.9,
eta_dish_meer=0.7,
n_ska=50,
n_meer=20,
)
cont_sens = mc.calculate_sensitivity(int_time * u.s)
backend_state = mc.state()
chunk_mc = MidCalculator(
band,
chunk_centres[band] * u.Hz,
chunk_width[band] * u.Hz,
array_config,
target_gc,
weather,
elevation * u.deg,
Trcv_ska=10.0 * u.K,
Trcv_meer=10.0 * u.K,
Tspl_ska=20.0 * u.K,
Tspl_meer=20.0 * u.K,
Tsky=15.0 * u.K,
Tgal=50.0 * u.K,
alpha=2.75,
eta_dish_ska=0.9,
eta_dish_meer=0.7,
n_ska=50,
n_meer=20,
)
chunk_sens = chunk_mc.calculate_sensitivity(int_time * u.s)
line_mc = MidCalculator(
band,
obsfreq[band] * u.Hz,
continuum_resolution[band] * u.Hz,
array_config,
target_gc,
weather,
elevation * u.deg,
Trcv_ska=10.0 * u.K,
Trcv_meer=10.0 * u.K,
Tspl_ska=20.0 * u.K,
Tspl_meer=20.0 * u.K,
Tsky=15.0 * u.K,
Tgal=50.0 * u.K,
alpha=2.75,
eta_dish_ska=0.9,
eta_dish_meer=0.7,
n_ska=50,
n_meer=20,
)
line_sens = line_mc.calculate_sensitivity(int_time * u.s)
frontend_state = translate_backend_state(backend_state)
weather_result = construct_frontend_result(
frontend_state,
cont_sens=cont_sens,
chunk_sens=chunk_sens,
line_sens=line_sens,
chunk_centres=chunk_centres[band] * u.Hz,
chunk_width=chunk_width[band] * u.Hz,
)
direct_result[weather] = weather_result
# print("result 3", result)
# print("direct_result 3", direct_result)
assert result == direct_result
[docs]def test_default_continuum_int_time():
"""This is an end to end testing of the SC ContinuumIntTime
component. A range of possible inputs are tried and the results
compared with those calculated directly.
'Expert' values are not set, so the SC will calculate the
standard defaults where necessary.
"""
# simulate some variation in input parameters across bands
obsfreq = {"Band 1": 0.7e9, "Band 2": 1.35e9, "Band 5a": 6.5e9, "Band 5b": 11.9e9}
bandwidth = {"Band 1": 0.7e9, "Band 2": 0.4e9, "Band 5a": 0.8e9, "Band 5b": 1.2e9}
chunk_centres = {
"Band 1": [0.7e9],
"Band 2": [1.2167e9, 1.35e9, 1.4833e9],
"Band 5a": [6.5e9],
"Band 5b": [11.6e9, 12.2e9],
}
chunk_width = {
"Band 1": 0.7e9,
"Band 2": 0.133333e9,
"Band 5a": 0.8e9,
"Band 5b": 0.6e9,
}
continuum_resolution = {
"Band 1": 0.21e3,
"Band 2": 0.21e3,
"Band 5a": 6.72e3,
"Band 5b": 1.68e3,
}
array_config = "full"
weathers = ["Average"]
elevation = 90.0
band = "Band 2"
sensitivity = 5.0e-3
result = calculate_continuum_int_time(
target_lon=target_cena.ra.hourangle,
target_lat=target_cena.dec.degree,
target_frame="icrs",
obs_band=band,
continuum_obs_freq_scaled=obsfreq[band],
continuum_bandwidth_scaled=bandwidth[band],
weathers=weathers,
elevation=elevation,
array_config=array_config,
continuum_sensitivity_scaled=sensitivity,
eta_system=None,
eta_point=None,
eta_coherence=None,
eta_digitisation=None,
eta_correlation=None,
eta_bandpass=None,
n_ska=None,
eta_dish_ska=None,
Tsys_ska=None,
Tspl_ska=None,
Trcv_ska=None,
n_meer=None,
eta_dish_meer=None,
Tsys_meer=None,
Tspl_meer=None,
Trcv_meer=None,
Tsky=None,
Tgal=None,
alpha=None,
chunk_centres=chunk_centres[band],
chunk_width=chunk_width[band],
continuum_resolution=continuum_resolution[band],
)
# get duplicate results by running the SC directly
direct_result = {}
for weather in weathers:
mc = MidCalculator(
band,
obsfreq[band] * u.Hz,
bandwidth[band] * u.Hz,
array_config,
target_cena,
weather,
elevation * u.deg,
)
cont_int_time = mc.calculate_integration_time(sensitivity * u.Jy)
backend_state = mc.state()
chunk_mc = MidCalculator(
band,
chunk_centres[band] * u.Hz,
chunk_width[band] * u.Hz,
array_config,
target_cena,
weather,
elevation * u.deg,
)
chunk_int_times = chunk_mc.calculate_integration_time(sensitivity * u.Jy)
line_mc = MidCalculator(
band,
obsfreq[band] * u.Hz,
continuum_resolution[band] * u.Hz,
array_config,
target_cena,
weather,
elevation * u.deg,
)
line_int_times = line_mc.calculate_integration_time(sensitivity * u.Jy)
frontend_state = translate_backend_state(backend_state)
weather_result = construct_frontend_result(
frontend_state,
cont_int_time=cont_int_time,
chunk_int_times=chunk_int_times,
line_int_times=line_int_times,
chunk_centres=chunk_centres[band] * u.Hz,
chunk_width=chunk_width[band] * u.Hz,
)
direct_result[weather] = weather_result
# pp = pprint.PrettyPrinter()
# print("result 4")
# pp.pprint(result)
# print("direct_result 4")
# pp.pprint(direct_result)
assert result == direct_result
[docs]def test_default_line_sensitivity():
"""This is an end to end testing of the SC LineSensitivity
component. A range of possible inputs are tried and the results
compared with those calculated directly.
'Expert' values are not set, so the SC will calculate the
standard defaults where necessary.
"""
array_configs = ["full"]
# simulate some variation in input parameters across bands
zoomfreqs = {
"Band 1": [0.7e9],
"Band 2": [1.35e9],
"Band 5a": [6.5e9, 7.2e9, 6.1e9],
"Band 5b": [11.9e9],
}
zoomres = {
"Band 1": [0.21e3, 3.36e3, 13.44e3],
"Band 2": [0.21e3, 3.36e3, 13.44e3],
"Band 5a": [0.21e3, 3.36e3, 13.44e3],
"Band 5b": [0.21e3, 3.36e3, 13.44e3],
}
int_times = [10.0]
array_config = "full"
weathers = ["Bad"]
elevation = 90.0
band = "Band 5a"
int_time = 10.0
result = calculate_line_sensitivity(
target_lon=target_cena.ra.hourangle,
target_lat=target_cena.dec.degree,
target_frame="icrs",
obs_band=band,
zoom_freqs_scaled=zoomfreqs[band],
zoom_resolutions=zoomres[band],
weathers=weathers,
elevation=elevation,
array_config=array_config,
line_int_time_scaled=int_time,
eta_system=None,
eta_point=None,
eta_coherence=None,
eta_digitisation=None,
eta_correlation=None,
eta_bandpass=None,
n_ska=None,
eta_dish_ska=None,
Tsys_ska=None,
Tspl_ska=None,
Trcv_ska=None,
n_meer=None,
eta_dish_meer=None,
Tsys_meer=None,
Tspl_meer=None,
Trcv_meer=None,
Tsky=None,
Tgal=None,
alpha=None,
)
# get duplicate results by running the SC directly
direct_result = {}
for weather in weathers:
zoom_sensitivities = []
for i, zoom_freq in enumerate(zoomfreqs[band] * u.Hz):
if not np.isnan(zoom_freq.value):
mc = MidCalculator(
band,
zoom_freq,
zoomres[band][i] * u.Hz,
array_config,
target_cena,
weather,
elevation * u.deg,
)
zoom_sensitivity = mc.calculate_sensitivity(int_time * u.s)
zoom_sensitivities.append(zoom_sensitivity)
else:
zoom_sensitivities.append(None)
backend_state = mc.state()
frontend_state = translate_backend_state(backend_state)
weather_result = construct_frontend_result(
frontend_state, zoom_sens=zoom_sensitivities
)
direct_result[weather] = weather_result
# print("result 5", result)
# print("direct_result 5", direct_result)
assert result == direct_result
[docs]def test_default_line_int_time():
"""This is an end to end testing of the SC LineIntTime
component. A range of possible inputs are tried and the results
compared with those calculated directly.
'Expert' values are not set, so the SC will calculate the
standard defaults where necessary.
"""
array_configs = ["full"]
sensitivities = [5.0e-3]
# simulate some variation in input parameters across bands
zoomfreqs = {
"Band 1": [0.7e9],
"Band 2": [1.35e9],
"Band 5a": [6.5e9, 7.2e9, 6.1e9],
"Band 5b": [11.9e9],
}
zoomres = {
"Band 1": [0.21e3, 3.36e3, 13.44e3],
"Band 2": [0.21e3, 3.36e3, 13.44e3],
"Band 5a": [0.21e3, 3.36e3, 13.44e3],
"Band 5b": [0.21e3, 3.36e3, 13.44e3],
}
array_config = "full"
weathers = ["Good"]
elevation = 90.0
band = "Band 5b"
sensitivity = 5.0e-3
result = calculate_line_int_time(
target_lon=target_cena.ra.hourangle,
target_lat=target_cena.dec.degree,
target_frame="icrs",
obs_band=band,
zoom_freqs_scaled=zoomfreqs[band],
zoom_resolutions=zoomres[band],
weathers=weathers,
elevation=elevation,
array_config=array_config,
line_sensitivity_scaled=sensitivity,
eta_system=None,
eta_point=None,
eta_coherence=None,
eta_digitisation=None,
eta_correlation=None,
eta_bandpass=None,
n_ska=None,
eta_dish_ska=None,
Tsys_ska=None,
Tspl_ska=None,
Trcv_ska=None,
n_meer=None,
eta_dish_meer=None,
Tsys_meer=None,
Tspl_meer=None,
Trcv_meer=None,
Tsky=None,
Tgal=None,
alpha=None,
)
# get duplicate results by running the SC directly
direct_result = {}
for weather in weathers:
zoom_int_times = []
for i, zoom_freq in enumerate(zoomfreqs[band] * u.Hz):
if not np.isnan(zoom_freq.value):
mc = MidCalculator(
band,
zoom_freq,
zoomres[band][i] * u.Hz,
array_config,
target_cena,
weather,
elevation * u.deg,
)
zoom_int_time = mc.calculate_integration_time(sensitivity * u.Jy)
zoom_int_times.append(zoom_int_time)
else:
zoom_int_times.append(None)
backend_state = mc.state()
frontend_state = translate_backend_state(backend_state)
weather_result = construct_frontend_result(
frontend_state, zoom_int_times=zoom_int_times
)
direct_result[weather] = weather_result
# print("result 6", result)
# print("direct_result 6", direct_result)
assert result == direct_result
[docs]def test_get_chunks():
"""This tests the get_chunks method for one set of inputs.
"""
n_chunks = 3
obs_freq = 0.7e9
bandwidth = 0.4e9
result = get_chunks(n_chunks, obs_freq, bandwidth)
assert result == (
[566666666.6666666, 700000000.0, 833333333.3333333],
133333333.33333333,
)
from numpy import array
[docs]def test_translate_backend_state():
"""This tests the translate_backend_state method for one set of inputs.
"""
test_state = {
"weather": 5,
"eta_point": array([0.99998795]),
"eta_coherence": array([0.99995826]),
"eta_digitisation": 0.999,
"eta_correlation": 0.98,
"eta_bandpass": 1.0,
"n_ska": 133,
"eta_dish_ska": array([0.82556592]),
"n_meer": 64,
"eta_dish_meer": array([0.70841861]),
"alpha": 2.75,
"Tsys_ska": array([220.78438594]),
"Tspl_ska": 3.0,
"Trcv_ska": array([15.075]),
"Tsys_meer": array([207.96011905]),
"Tspl_meer": 4.0,
"Trcv_meer": array([10.46]),
"Tsky": array([198.12154941]),
"Tgal": array([195.03757696]),
"altitude": 88.21504776179147,
}
result = translate_backend_state(test_state)
# print("result 7", result)
assert result == {
"pwv": "5",
"etaPointing": "1",
"etaCoherence": "1",
"etaDigitisation": "0.999",
"etaCorrelation": "0.98",
"etaBandpass": "1",
"nSKA": "133",
"etaSKA": "0.826",
"nMeer": "64",
"etaMeer": "0.708",
"alpha": "2.75",
"Tsys_SKA": "221",
"Tspl_SKA": "3",
"Trcv_SKA": "15.1",
"Tsys_Meer": "208",
"Tspl_Meer": "4",
"Trcv_Meer": "10.5",
"Tsky": "198",
"Tgal": "195",
"elevation": "88.2",
}
[docs]def test_construct_frontend_result():
"""This tests the translate_frontend_result method for one set of inputs.
"""
with pytest.raises(RuntimeError) as err:
assert construct_frontend_result("blah")
assert str(err.value) == "no results input"
result = construct_frontend_result(
frontend_state="testing1,2,3",
cont_sens=[1.11111] * u.Jy,
chunk_sens=[1.0, 2.22222, 3.3333333] * u.Jy,
line_sens=[15.667788] * u.Jy,
chunk_centres=[1.0, 1.5, 2.98765] * u.GHz,
chunk_width=0.33 * u.GHz,
)
# print("result sens", result)
assert result == {
"calculator_state": "testing1,2,3",
"cont_sens": "1.1111 Jy",
"chunk_centres": ["1 GHz", "1.5 GHz", "2.9876 GHz"],
"chunk_width": "0.33 GHz",
"chunk_sensitivities": ["1 Jy", "2.2222 Jy", "3.3333 Jy"],
"cont_line_sens": "15.668 Jy",
}
result = construct_frontend_result(
frontend_state="testing1,2,3",
cont_int_time=[10.0123456] * u.s,
chunk_int_times=[10.0, 20.01111, 30.0] * u.s,
line_int_times=[100.0] * u.s,
chunk_centres=[1.0, 1.5, 2.98765] * u.GHz,
chunk_width=0.33 * u.GHz,
)
# print("result times", result)
assert result == {
"calculator_state": "testing1,2,3",
"cont_int_time": "10.012 s",
"chunk_centres": ["1 GHz", "1.5 GHz", "2.9876 GHz"],
"chunk_width": "0.33 GHz",
"chunk_int_times": ["10 s", "20.011 s", "30 s"],
"cont_line_int_time": "100 s",
}
result = construct_frontend_result(
frontend_state="testing1,2,3",
# set list of lists because MidCalculator returns an array when it is
# called by frontend_adapter for each zoom
zoom_sens=[[15.0], [30.0], [45.666], [60.0]] * u.mJy,
)
#print("result zoom_sens", result)
assert result == {
"calculator_state": "testing1,2,3",
"zoom_sensitivities": ["0.015 Jy", "0.03 Jy", "0.045666 Jy", "0.06 Jy"],
}
result = construct_frontend_result(
frontend_state="testing1,2,3",
zoom_int_times=[[60], [70], [80], [90.6666]] * u.s,
)
#print("result zoom_int_times", result)
assert result == {
"calculator_state": "testing1,2,3",
"zoom_int_times": ["60 s", "70 s", "80 s", "90.667 s"],
}