# -*- 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__)
[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"])
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")
]
# rename processing mode sections
if "ft" in pst_configuration:
pst_configuration["flow_through_params"] = pst_configuration.pop("ft")
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.pop("total_bandwidth") / MEGA_HERTZ
pst_configuration["centre_freq_mhz"] = pst_configuration.pop("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 {
"cbf_pst_config": cbf_pst_config,
**dataclasses.asdict(receiver_config),
**common_configure,
**pst_configuration,
}