# -*- coding: utf-8 -*-
#
# This file is part of the SKA PST project.
#
# Distributed under the terms of the BSD 3-clause new license.
# See LICENSE for more info.
"""This module provides a utility method to convert CSP configuration request into a PST internal format."""
from __future__ import annotations
import copy
import dataclasses
import logging
from typing import Any
from ska_control_model import PstProcessingMode
from ska_pst.common.constants import MEGA_HERTZ
from .telescope_configuration import TelescopeConfig
_logger = logging.getLogger(__name__)
def _handle_detected_filterbank_config(detected_filterbank_params: dict) -> dict:
# Migrate legacy DF parameters
if "stokes_parameters" in detected_filterbank_params:
_logger.warning(
"Deprecated field 'stokes_parameters' has been removed in version 4.0 PST "
"schema. Please use 'polarisation_state' instead."
)
# Schema was version 3.x
stokes_parameters: str = detected_filterbank_params.pop("stokes_parameters")
if stokes_parameters == "I":
polarisation_state = "Intensity"
else:
if stokes_parameters != "IQUV":
_logger.warning(f"Unsupported value of 'stokes_parameters' with value {stokes_parameters}")
polarisation_state = "Stokes"
detected_filterbank_params["polarisation_state"] = polarisation_state
return detected_filterbank_params
[docs]def convert_csp_config_to_pst_config(
telescope_config: TelescopeConfig, csp_configure_scan_request: dict
) -> dict:
"""
Convert a CSP configure scan request into a PST configure scan request.
This method flattens the configuration by selection the common and pst/scan
paths of the CSP request and putting them all into one dictionary.
It takes the ``frequency_band`` value in the ``common`` section of the request
and gets the frequency band configuration from the ``telescope_config`` to
augment the request. For the version 3.0+ of the PST schema a lot of these
parameters have been removed from the CSP/PST request but the values are needed
within the PST LMC to send down to CORE applications.
:param telescope_config: the telescope configuration that PST BEAM is running in.
:type telescope_config: TelescopeConfig
:param csp_configure_scan_request: the CSP configure scan request as Python dict.
:type csp_configure_scan_request: dict
:return: a dictionary used internally by PST to handling configuring of the PST
BEAM.
:rtype: dict
"""
common_configure = csp_configure_scan_request["common"]
if telescope_config.name == "low":
# force using a low Frequency Band if the facility is SKALow
common_configure["frequency_band"] = "low"
pst_configuration: dict = copy.deepcopy(csp_configure_scan_request["pst"]["scan"])
_update_pst_processing_mode(pst_configuration)
if "itrf" in pst_configuration:
pst_configuration["delay_centre"] = pst_configuration.pop("itrf")
_handle_source_and_coords(pst_configuration)
if "max_scan_length" in pst_configuration:
del pst_configuration["max_scan_length"]
# rename processing mode sections
if "ft" in pst_configuration:
pst_configuration["flow_through_params"] = pst_configuration.pop("ft")
if "df" in pst_configuration:
pst_configuration["detected_filterbank_params"] = _handle_detected_filterbank_config(
pst_configuration.pop("df")
)
frequency_band = common_configure["frequency_band"]
frequency_band_config = telescope_config.frequency_bands[frequency_band]
receiver_config = frequency_band_config.receiver_config
cbf_pst_config = frequency_band_config.cbf_pst_config
pst_configuration["bandwidth_mhz"] = bandwidth_mhz = pst_configuration["total_bandwidth"] / MEGA_HERTZ
pst_configuration["centre_freq_mhz"] = pst_configuration["centre_frequency"] / MEGA_HERTZ
pst_configuration["nchan"] = nchan = cbf_pst_config.nchan_for_bandwidth(bandwidth_mhz=bandwidth_mhz)
def _check_deprecated_field(name: str, expected: Any) -> None:
actual: Any | None = pst_configuration.get(name, None)
if actual is not None and actual != expected:
_logger.warning(
f"Deprecated field {name}'s value of {actual} is not the expected value {expected} "
f"for frequency band {frequency_band}. Field will be removed in version 3.0 of PST schema"
)
_check_deprecated_field("bits_per_sample", cbf_pst_config.nbit * cbf_pst_config.ndim)
_check_deprecated_field("num_of_polarizations", cbf_pst_config.npol)
_check_deprecated_field("udp_nsamp", cbf_pst_config.udp_nsamp)
_check_deprecated_field("wt_nsamp", cbf_pst_config.wt_nsamp)
_check_deprecated_field("udp_nchan", cbf_pst_config.udp_nchan)
_check_deprecated_field("num_frequency_channels", nchan)
_check_deprecated_field("oversampling_ratio", [*cbf_pst_config.oversampling_ratio])
_check_deprecated_field("num_channelization_stages", cbf_pst_config.num_channelisation_stages)
if "channelization_stages" in pst_configuration:
_logger.warning(
"Deprecated field 'channelization_stages' will be removed in version 3.0 of PST schema"
)
receiver_id = pst_configuration["receiver_id"]
if receiver_id != receiver_config.receiver_id:
_logger.warning(
f"The value of {receiver_id=} is different to expected value "
f"'{receiver_config.receiver_id}' for current frequency band '{frequency_band}'"
)
# the following dictionary uses the order of
# 1 frequency band config
# 2 receiver config
# 3 the common config
# 4 the PST part of the configuration
#
# by receiver config before the pst configuration it allows this to be future compatible when the
# receiver feed configuration will not be in the schema but determined by the receiver id. If
# the values in a version of the schema that supports the values, then those values will overwrite
# is coming from the receiver config.
return {
"telescope_config": telescope_config,
"cbf_pst_config": cbf_pst_config,
**dataclasses.asdict(receiver_config),
**common_configure,
**pst_configuration,
}
def _update_pst_processing_mode(pst_configuration: dict) -> None:
"""
Update the configuration with the PST processing mode.
For older schema versions had a field ``observation_mode`` which
is the same as the ``pst_processing_mode`` in later versions.
This method will also turn the value into a ``PstProcessingMode``
enum value.
:param pst_configuration: the PST scan configuration.
:type pst_configuration: dict
"""
if "observation_mode" in pst_configuration:
_logger.warning(
"Deprecated field 'observation_mode' will be renamed to 'pst_processing_mode' "
"in version 3.0 of PST schema"
)
pst_configuration["pst_processing_mode"] = PstProcessingMode[
pst_configuration.pop("observation_mode")
]
else:
pst_configuration["pst_processing_mode"] = PstProcessingMode[
pst_configuration.pop("pst_processing_mode")
]
def _handle_source_and_coords(pst_configuration: dict) -> None:
"""
Handle the source and coordinates configuration.
For PST schema versions < 3.0 this will map the ``coordinates``
su-dictionary to the ``stt_crd1``, ``stt_crd2` and ``equinox``
keys.
For PST schema versions >= 3.0 this will map the SKA SkyCoords
dictionary to the appropriate internal representation.
:param pst_configuration: the PST scan configuration.
:type pst_configuration: dict
"""
if "source" in pst_configuration:
# This branch is for deprecated PST schema versions (i.e. < 3.0)
coords = pst_configuration.pop("coordinates")
pst_configuration["stt_crd1"] = coords["ra"]
pst_configuration["stt_crd2"] = coords["dec"]
if "equinox" in coords:
pst_configuration["equinox"] = coords["equinox"]
else:
# This branch supports versions of 3.0+ of the PST Schema which uses
# the SKA sky_direction format
target = pst_configuration.pop("target")
target_attrs = target["attrs"]
pst_configuration["source"] = target["target_name"]
pst_configuration["stt_crd1"] = target_attrs["c1"]
pst_configuration["stt_crd2"] = target_attrs["c2"]
if "epoch" in target_attrs:
pst_configuration["equinox"] = target_attrs["epoch"]