import copy
from typing import Any, Callable, Dict
from .._common import split_interface_version
from .version import (
PST_CONFIG_VER2_4,
PST_CONFIG_VER2_5,
PST_CONFIG_VER3_0,
check_interface_version,
)
PST_CONFIGURE_SCAN_DF = {
"dispersion_measure": 100.0,
"output_frequency_channels": 1,
"stokes_parameters": "Q",
"num_bits_out": 16,
"time_decimation_factor": 10,
"frequency_decimation_factor": 4,
"requantisation_scale": 1.0,
"requantisation_length": 1.0,
}
PST_CONFIGURE_SCAN_PT = {
"dispersion_measure": 100.0,
"rotation_measure": 0.0,
"ephemeris": "",
"pulsar_phase_predictor": "",
"output_frequency_channels": 1,
"output_phase_bins": 64,
"num_sk_config": 1,
"sk_config": [
{
"sk_range": [0.8, 0.9],
"sk_integration_limit": 100,
"sk_excision_limit": 25.0,
}
],
"target_snr": 0.0,
}
PST_CONFIGURE_SCAN_FT = {
"num_bits_out": 4,
"num_channels": 1,
"channels": [100],
"requantisation_scale": 1.0,
"requantisation_length": 1.0,
}
PST_CONFIGURE_SCAN_BASE = {
"interface": "https://schema.skao.int/ska-pst-configure/2.4",
"common": {
"config_id": "sbi-mvp01-20240111",
"subarray_id": 1,
"eb_id": "eb-x321-20240111-10012",
},
"pst": {"scan": {}},
}
LOW_PST_CONFIGURE_SCAN_COMMON = {
"activation_time": "2024-01-11T23:11:17Z",
"num_of_polarizations": 2,
"udp_nsamp": 32,
"wt_nsamp": 32,
"udp_nchan": 24,
"num_frequency_channels": 432,
"centre_frequency": 100000000.0,
"total_bandwidth": 1562500.0,
"bits_per_sample": 32,
"timing_beam_id": "1",
"observation_mode": "VOLTAGE_RECORDER",
"observer_id": "jdoe",
"project_id": "project1",
"pointing_id": "pointing1",
"source": "J1921+2153",
"itrf": [5109360.133, 2006852.586, -3238948.127],
"receiver_id": "receiver3",
"feed_polarization": "LIN",
"feed_handedness": 1,
"feed_angle": 1.234,
"feed_tracking_mode": "FA",
"feed_position_angle": 10.0,
"oversampling_ratio": [4, 3],
"coordinates": {
"equinox": 2000.0,
"ra": "19:21:44.815",
"dec": "21:53:02.400",
},
"max_scan_length": 20000.0,
"subint_duration": 30.0,
"receptors": ["SKA001", "SKA036"],
"receptor_weights": [0.4, 0.6],
"num_channelization_stages": 2,
"channelization_stages": [
{
"num_filter_taps": 1,
"filter_coefficients": [1.0],
"num_frequency_channels": 1024,
"oversampling_ratio": [32, 27],
},
{
"num_filter_taps": 1,
"filter_coefficients": [1.0],
"num_frequency_channels": 256,
"oversampling_ratio": [4, 3],
},
],
}
MID_PST_CONFIGURE_SCAN_COMMON = {
"activation_time": "2024-01-11T23:11:17Z",
"bits_per_sample": 16,
"timing_beam_id": "1",
"num_of_polarizations": 2,
"udp_nsamp": 4,
"wt_nsamp": 4,
"udp_nchan": 185,
"num_frequency_channels": 46435,
"centre_frequency": 10550000000.0,
"total_bandwidth": 2496345600.0,
"observation_mode": "VOLTAGE_RECORDER",
"observer_id": "jdoe",
"project_id": "project1",
"pointing_id": "pointing1",
"source": "J1921+2153",
"itrf": [5109360.133, 2006852.586, -3238948.127],
"receiver_id": "receiver3",
"feed_polarization": "LIN",
"feed_handedness": 1,
"feed_angle": 1.234,
"feed_tracking_mode": "FA",
"feed_position_angle": 10.0,
"oversampling_ratio": [8, 7],
"coordinates": {
"equinox": 2000.0,
"ra": "19:21:44.815",
"dec": "21:53:02.400",
},
"max_scan_length": 20000.0,
"subint_duration": 30.0,
"receptors": ["SKA001", "SKA036"],
"receptor_weights": [0.4, 0.6],
"num_channelization_stages": 2,
"channelization_stages": [
{
"num_filter_taps": 1,
"filter_coefficients": [1.0],
"num_frequency_channels": 13,
"oversampling_ratio": [10, 9],
},
{
"num_filter_taps": 1,
"filter_coefficients": [1.0],
"num_frequency_channels": 4096,
"oversampling_ratio": [8, 7],
},
],
}
def _get_base_example(
version: str, pst_processing_mode: str, frequency_band: str
) -> dict:
(major, minor) = split_interface_version(version)
example = copy.deepcopy(PST_CONFIGURE_SCAN_BASE)
config_id_suffix = pst_processing_mode.lower().replace("_", "-")
example["common"]["config_id"] += f"-{config_id_suffix}"
example["common"]["frequency_band"] = frequency_band
if (major, minor) == (2, 4):
example["interface"] = PST_CONFIG_VER2_4
elif (major, minor) == (2, 5):
example["interface"] = PST_CONFIG_VER2_5
else:
example["interface"] = PST_CONFIG_VER3_0
return example
def _get_scan_common_example(
version: str,
*args: Any,
example: dict,
pst_processing_mode: str,
**kwargs: Any,
) -> dict:
(major, minor) = split_interface_version(version)
if (major, minor) >= (3, 0):
# ensure PST processing mode is set
example["pst_processing_mode"] = pst_processing_mode
# itrf was renamed to delay_centre
example["delay_centre"] = example.pop("itrf")
coordinates = example.pop("coordinates")
example["target"] = {
"target_name": example.pop("source"),
"reference_frame": "icrs",
"attrs": {
"c1": 290.43672917,
"c2": 21.8840, # coordinates.pop("dec"),
"epoch": coordinates.pop("equinox"),
},
}
# remove deprecated fields
del example["observation_mode"]
del example["activation_time"]
del example["bits_per_sample"]
del example["num_of_polarizations"]
del example["udp_nsamp"]
del example["wt_nsamp"]
del example["udp_nchan"]
del example["num_frequency_channels"]
del example["feed_polarization"]
del example["feed_handedness"]
del example["feed_angle"]
del example["feed_tracking_mode"]
del example["feed_position_angle"]
del example["oversampling_ratio"]
del example["num_channelization_stages"]
del example["channelization_stages"]
del example["pointing_id"]
else:
# ensure PST processing mode is set using depreciated
# field of observation_mode
example["observation_mode"] = pst_processing_mode
return example
def _get_pt_example(
version: str, *args: Any, common_example: dict, **kwargs: Any
) -> dict:
example = _get_scan_common_example(
version,
*args,
example=common_example,
pst_processing_mode="PULSAR_TIMING",
**kwargs,
)
example["pt"] = copy.deepcopy(PST_CONFIGURE_SCAN_PT)
return example
def _get_vr_example(
version: str, *args: Any, common_example: dict, **kwargs: Any
) -> dict:
return _get_scan_common_example(
version,
*args,
example=common_example,
pst_processing_mode="VOLTAGE_RECORDER",
**kwargs,
)
def _get_ft_example(
version: str, *args: Any, common_example: dict, **kwargs: Any
) -> dict:
(major, minor) = split_interface_version(version)
example = _get_scan_common_example(
version,
*args,
example=common_example,
pst_processing_mode="FLOW_THROUGH",
**kwargs,
)
ft_example = copy.deepcopy(PST_CONFIGURE_SCAN_FT)
if (major, minor) >= (2, 5):
del ft_example["num_channels"]
del ft_example["requantisation_length"]
ft_example["channels"] = [0, 100]
ft_example["polarizations"] = "Both"
ft_example["requantisation_init_time"] = 1.0
if (major, minor) >= (3, 0):
# restructure to Flow through mode config from version 3.0 onwards
ft_example["channel_polarisation_selection"] = {
"channels": ft_example.pop("channels"),
"polarisations": ft_example.pop("polarizations"),
}
ft_example["rescale"] = {
"algorithm": "MedianMAD",
"periodic_update": True,
"timescale": ft_example.pop("requantisation_init_time"),
}
ft_example["requantisation"] = {
"num_bits_out": ft_example.pop("num_bits_out"),
"scale": ft_example.pop("requantisation_scale"),
}
example["ft"] = ft_example
return example
def _get_ds_example(
version: str, *args: Any, common_example: dict, **kwargs: Any
) -> dict:
(major, minor) = split_interface_version(version)
assert (major, minor) < (
3,
0,
), "Dynamic spectrum has been renamed to detected filter bank"
example = _get_scan_common_example(
version,
*args,
example=common_example,
pst_processing_mode="DYNAMIC_SPECTRUM",
**kwargs,
)
example["ds"] = copy.deepcopy(PST_CONFIGURE_SCAN_DF)
return example
def _get_df_example(
version: str, *args: Any, common_example: dict, **kwargs: Any
) -> dict:
(major, minor) = split_interface_version(version)
assert (major, minor) >= (
3,
0,
), "Detected filterbank schema only valid from version 3.0 onwards."
example = _get_scan_common_example(
version,
*args,
example=common_example,
pst_processing_mode="DETECTED_FILTERBANK",
**kwargs,
)
example["df"] = copy.deepcopy(PST_CONFIGURE_SCAN_DF)
return example
SCAN_TYPE_MAP: Dict[str, Callable[..., dict]] = {
"pst_scan_vr": _get_vr_example,
"pst_scan_pt": _get_pt_example,
"pst_scan_ft": _get_ft_example,
"pst_scan_ds": _get_ds_example,
"pst_scan_df": _get_df_example,
}
[docs]
def get_pst_config_example(version: str, scan_type: str = None) -> dict:
"""Generate examples for PST configuration strings
This will delegate to the appropriate telescope example (i.e. Mid or Low)
depending on the prefix of the ``scan_type`` parameter. If the
``scan_type`` is prefixed with ``mid_`` then a SKAMid PST scan config
example is returned else a SKALow PST scan config example is returned.
Valid values of ``scan_type`` are:
* pst_scan_vr
* pst_scan_pt
* pst_scan_ft
* pst_scan_ds
* pst_scan_df
* low_pst_scan_vr
* low_pst_scan_pt
* low_pst_scan_ft
* low_pst_scan_ds
* low_pst_scan_df
* mid_pst_scan_vr
* mid_pst_scan_pt
* mid_pst_scan_ft
* mid_pst_scan_ds
* mid_pst_scan_df
:param version: Version URI of configuration format
:param scan: Includes SDP receive addresses for a scan? `None`
means that this is "template" configuration as passed to
TMC.
"""
check_interface_version(version)
if scan_type.startswith("mid_"):
common_example = copy.deepcopy(MID_PST_CONFIGURE_SCAN_COMMON)
frequency_band = "5b"
effective_scan_type = scan_type[4:]
else:
common_example = copy.deepcopy(LOW_PST_CONFIGURE_SCAN_COMMON)
frequency_band = "low"
effective_scan_type = scan_type
if scan_type.startswith("low_"):
effective_scan_type = effective_scan_type[4:]
(major, minor) = split_interface_version(version)
if (major, minor) >= (2, 4):
scan_config = SCAN_TYPE_MAP[effective_scan_type](
version, common_example=common_example
)
try:
pst_processing_mode = scan_config["pst_processing_mode"]
except KeyError:
pst_processing_mode = scan_config["observation_mode"]
config = _get_base_example(
version,
pst_processing_mode=pst_processing_mode,
frequency_band=frequency_band,
)
config["pst"]["scan"] = scan_config
return config
raise ValueError(
f"Could not generate example for schema {version} and {scan_type=}!"
)