Source code for ska_tmc_cdm.schemas.subarray_node.configure.csp

"""
This module defines Marshmallow schemas that map the CDM classes for
SubArrayNode CSP configuration to/from JSON.
"""
import copy
import json
from marshmallow import Schema, fields, post_load, pre_dump, post_dump
from marshmallow.validate import OneOf
from ska_tmc_cdm.messages.subarray_node.configure.csp import (
    FSPFunctionMode,
    FSPConfiguration,
    SubarrayConfiguration,
    CommonConfiguration,
    CBFConfiguration,
    CSPConfiguration
)
from ska_tmc_cdm.messages.subarray_node.configure.core import ReceiverBand
from ska_tmc_cdm.schemas import CODEC
from ska_tmc_cdm.schemas.shared import ValidatingSchema

__all__ = ["CSPConfigurationSchema", "FSPConfigurationSchema",
           "SubarrayConfigurationSchema", "CommonConfigurationSchema",
           "CBFConfigurationSchema"]


[docs]@CODEC.register_mapping(SubarrayConfiguration) class SubarrayConfigurationSchema(Schema): subarray_name = fields.String(data_key="subarray_name", required=True)
[docs] @post_load def create(self, data, **_): """ Convert parsed JSON back into a SubarrayConfiguration object. :param data: dict containing parsed JSON values :param _: kwargs passed by Marshmallow :return: SubarrayConfiguration instance populated to match JSON :rtype: SubarrayConfiguration """ subarray_name = data["subarray_name"] return SubarrayConfiguration(subarray_name)
[docs]@CODEC.register_mapping(CommonConfiguration) class CommonConfigurationSchema(Schema): config_id = fields.String(data_key="config_id", required=True) frequency_band = fields.String(data_key="frequency_band", required=True) subarray_id = fields.Integer(data_key="subarray_id", required=True) band_5_tuning = fields.List(fields.Float, data_key="band_5_tuning")
[docs] @pre_dump def convert( self, common_configuration: CommonConfiguration, **_ ): # pylint: disable=no-self-use """ Process CommonConfiguration instance so that it is ready for conversion to JSON. :param CommonConfiguration: Common configuration to process :param _: kwargs passed by Marshmallow :return: CommonConfiguration instance populated to match JSON """ # Convert Python Enum to its string value copied = copy.deepcopy(common_configuration) if hasattr(copied, 'frequency_band'): copied.frequency_band = common_configuration.frequency_band.value return copied
[docs] @post_dump def filter_nulls(self, data, **_): # pylint: disable=no-self-use """ Filter out null values from JSON. :param data: Marshmallow-provided dict containing parsed object values :param _: kwargs passed by Marshmallow :return: dict suitable for FSP configuration """ result = {k: v for k, v in data.items() if v is not None} return result
[docs] @post_load def create(self, data, **_): # pylint: disable=no-self-use """ Convert parsed JSON back into a CSPConfiguration object. :param data: dict containing parsed JSON values :param _: kwargs passed by Marshmallow :return: CommonConfiguration instance populated to match JSON """ config_id = data["config_id"] frequency_band = data["frequency_band"] frequency_band_enum = ReceiverBand(frequency_band) subarray_id = data["subarray_id"] band_5_tuning = data.get("band_5_tuning", None) return CommonConfiguration(config_id, frequency_band_enum, subarray_id, band_5_tuning)
[docs]@CODEC.register_mapping(FSPConfiguration) class FSPConfigurationSchema(Schema): """ Marshmallow schema for the subarray_node.FSPConfiguration class """ fsp_id = fields.Integer(data_key="fsp_id", required=True) function_mode = fields.String( data_key="function_mode", validate=OneOf(["CORR", "PSS-BF", "PST-BF", "VLBI"]), required=True, ) frequency_slice_id = fields.Integer(data_key="frequency_slice_id", required=True) zoom_factor = fields.Integer(data_key="zoom_factor", required=True) integration_factor = fields.Integer(data_key="integration_factor", required=True) channel_averaging_map = fields.List( fields.Tuple((fields.Integer, fields.Integer)), data_key="channel_averaging_map" ) output_link_map = fields.List( fields.Tuple((fields.Integer, fields.Integer)), data_key="output_link_map" ) channel_offset = fields.Integer(data_key="channel_offset") zoom_window_tuning = fields.Integer(data_key="zoom_window_tuning")
[docs] @pre_dump def convert( self, fsp_configuration: FSPConfiguration, **_ ): # pylint: disable=no-self-use """ Process FSPConfiguration instance so that it is ready for conversion to JSON. :param fsp_configuration: FSP configuration to process :param _: kwargs passed by Marshmallow :return: FspConfiguration instance populated to match JSON """ # Convert Python Enum to its string value copied = copy.deepcopy(fsp_configuration) copied.function_mode = fsp_configuration.function_mode.value return copied
[docs] @post_dump def filter_nulls(self, data, **_): # pylint: disable=no-self-use """ Filter out null values from JSON. :param data: Marshmallow-provided dict containing parsed object values :param _: kwargs passed by Marshmallow :return: dict suitable for FSP configuration """ result = {k: v for k, v in data.items() if v is not None} return result
[docs] @post_load def create(self, data, **_): # pylint: disable=no-self-use """ Convert parsed JSON back into a FSPConfiguration object. :param data: dict containing parsed JSON values :param _: kwargs passed by Marshmallow :return: FSPConfiguration instance populated to match JSON """ fsp_id = data["fsp_id"] function_mode = data["function_mode"] function_mode_enum = FSPFunctionMode(function_mode) frequency_slice_id = int(data["frequency_slice_id"]) zoom_factor = data["zoom_factor"] integration_factor = data["integration_factor"] # optional arguments channel_averaging_map = data.get("channel_averaging_map", None) output_link_map = data.get("output_link_map", None) channel_offset = data.get("channel_offset", None) zoom_window_tuning = data.get("zoom_window_tuning", None) return FSPConfiguration( fsp_id, function_mode_enum, frequency_slice_id, integration_factor, zoom_factor, channel_averaging_map=channel_averaging_map, output_link_map=output_link_map, channel_offset=channel_offset, zoom_window_tuning=zoom_window_tuning, )
[docs]@CODEC.register_mapping(CBFConfiguration) class CBFConfigurationSchema(Schema): fsp_configs = fields.Nested(FSPConfigurationSchema, many=True, data_key="fsp")
[docs] @post_load def create(self, data, **_): """ Convert parsed JSON back into a CBFConfiguration object. :param data: dict containing parsed JSON values :param _: kwargs passed by Marshmallow :return: CBFConfiguration instance populated to match JSON :rtype: CBFConfiguration """ fsp_configs = data["fsp_configs"] return CBFConfiguration(fsp_configs)
[docs]@CODEC.register_mapping(CSPConfiguration) class CSPConfigurationSchema(ValidatingSchema): """ Marshmallow schema for the subarray_node.CSPConfiguration class """ interface = fields.String() subarray_config = fields.Nested(SubarrayConfigurationSchema, data_key="subarray") common_config = fields.Nested(CommonConfigurationSchema, data_key="common") cbf_config = fields.Nested(CBFConfigurationSchema, data_key="cbf")
[docs] @post_load def create(self, data, **_): # pylint: disable=no-self-use """ Convert parsed JSON back into a CSPConfiguration object. :param data: dict containing parsed JSON values :param _: kwargs passed by Marshmallow :return: CSPConfiguration instance populated to match JSON """ interface = data.get("interface", None) subarray_config = data.get("subarray_config", None) common_config = data.get("common_config", None) cbf_config = data.get("cbf_config", None) return CSPConfiguration( interface=interface, subarray_config=subarray_config, common_config=common_config, cbf_config=cbf_config )
[docs] @post_dump def validate_on_dump(self, data, **_): """ Validating the structure of JSON against schemas and Filter out null values from JSON. :param data: Marshmallow-provided dict containing parsed object values :param _: kwargs passed by Marshmallow :return: dict suitable for SubArrayNode configuration """ # filter out null values from JSON data = {k: v for k, v in data.items() if v is not None} # convert tuples to lists data = json.loads(json.dumps(data)) data = super().validate_on_dump(data) return data