Loading csp-lmc-common/csp_lmc_common/csp_manage_json.py +26 −22 Original line number Diff line number Diff line import json from ska_telmodel.csp.schema import validate_csp_config from ska_telmodel.csp.version import validate_csp_config_version class JsonConfiguration: """ Loading @@ -18,50 +19,53 @@ class JsonConfiguration: #ADR-22 if 'interface' in self.config.keys(): interface = self.config['interface'] if validate_csp_config_version(interface): version_str = interface.split('/')[-1] self.major_version = int(version_str.split('.')[0]) self.minor_version = int(version_str.split('.')[1]) self.major_version , self.minor_version = map(int, version_str.split('.')) else: message = 'Wrong json schema URI!' self.logger.error(message) raise ValueError(message) #ADR-18 elif 'common' in self.config.keys(): self.major_version = 0 self.minor_version = 2 self.major_version, self.minor_version = (1,0) #pre ADR-18 else: self.major_version = 0 self.minor_version = 1 self.version = [self.major_version, self.minor_version] self.major_version, self.minor_version = (0,1) self.version = (self.major_version, self.minor_version) self.logger.info(f'Version is{self.major_version}{self.minor_version}') def get_section(self, name:str): """ Returns the correpondant section of json configuration Returns the correspondant section of json configuration. """ try: return self.config[name] except KeyError as e: self.logger.warning(e) self.logger.error(e) raise ValueError(e) from e def conversion_02_01(self): def conversion_10_01(self): """ Converts the json from version 0.2 to version 0.1 Converts the json from version 1.0 (ADR-18) to version 0.1. Note: Current Mid.CBF supports 0.1 Json schema. """ common_dict = self.get_section('common') cbf_dict = self.get_section('cbf') config_converted = {**common_dict, **cbf_dict} validate_csp_config(version=0, template=True, config=config_converted) #check template parameter validate_csp_config(version=0, config=config_converted) self.config = config_converted def build_json_cbf(self) -> dict: " def build_json_cbf(self) -> str: """ Returns the json configuration dictionary to be passed to cbf. If a version 0.2 is given, it converts to 0.1 " """ self.detect_version() if self.version == [0, 1]: if self.version == (0, 1): pass elif self.version == [0,2]: self.conversion_02_01() elif self.version == (1, 0): self.conversion_10_01() else: raise ValueError(f'version{self.version}is not supported') return json.dumps(self.config) csp-lmc-common/tests/unit/json_config_test.py 0 → 100644 +175 −0 Original line number Diff line number Diff line from csp_lmc_common.csp_manage_json import JsonConfiguration import logging import pytest LOGGER = logging.getLogger(__name__) json_string_test_ADR_22 = """ { "interface": "https://schema.skatelescope.org/ska-csp-configure/1.0", "common": { "id": "sbi-mvp01-20200325-00001-science_A", "frequencyBand": "1" }, "cbf": { "fsp": [ { "fspID": 1, "functionMode": "CORR", "frequencySliceID": 1, "integrationTime": 1400, "corrBandwidth": 0, "channelAveragingMap": [[0,2], [744,0]], "fspChannelOffset": 0, "outputLinkMap": [[0,0], [200,1]] }, { "fspID": 2, "functionMode": "CORR", "frequencySliceID": 2, "integrationTime": 1400, "corrBandwidth": 0, "channelAveragingMap": [[0,2], [744,0]], "fspChannelOffset": 744, "outputLinkMap": [[0,4], [200,5]] } ] } } """ json_string_test_ADR_18 = """ { "common": { "id": "sbi-mvp01-20200325-00001-science_A", "frequencyBand": "1" }, "cbf": { "fsp": [ { "fspID": 1, "functionMode": "CORR", "frequencySliceID": 1, "integrationTime": 1400, "corrBandwidth": 0, "channelAveragingMap": [[0,2], [744,0]], "fspChannelOffset": 0, "outputLinkMap": [[0,0], [200,1]] }, { "fspID": 2, "functionMode": "CORR", "frequencySliceID": 2, "integrationTime": 1400, "corrBandwidth": 0, "channelAveragingMap": [[0,2], [744,0]], "fspChannelOffset": 744, "outputLinkMap": [[0,4], [200,5]] } ] } } """ #function mode missing json_string_test_ADR_18_invalid = """ { "common": { "id": "sbi-mvp01-20200325-00001-science_A", "frequencyBand": "1", }, "cbf": { "fsp": [ { "fspID": 1, "frequencySliceID": 1, "integrationTime": 1400, "corrBandwidth": 0, "channelAveragingMap": [[0,2], [744,0]], "fspChannelOffset": 0, "outputLinkMap": [[0,0], [200,1]] }, { "fspID": 2, "functionMode": "CORR", "frequencySliceID": 2, "integrationTime": 1400, "corrBandwidth": 0, "channelAveragingMap": [[0,2], [744,0]], "fspChannelOffset": 744, "outputLinkMap": [[0,4], [200,5]] } ] } } """ json_string_test_pre_ADR_18 = """ { "id": "sbi-mvp01-20200325-00001-science_A", "frequencyBand": "1", "fsp": [ { "fspID": 1, "functionMode": "CORR", "frequencySliceID": 1, "integrationTime": 1400, "corrBandwidth": 0, "channelAveragingMap": [[0,2], [744,0]], "fspChannelOffset": 0, "outputLinkMap": [[0,0], [200,1]] }, { "fspID": 2, "functionMode": "CORR", "frequencySliceID": 2, "integrationTime": 1400, "corrBandwidth": 0, "channelAveragingMap": [[0,2], [744,0]], "fspChannelOffset": 744, "outputLinkMap": [[0,4], [200,5]] } ] } """ json_string_invalid_URI = '{"interface": "invaliduri.com/0.1"}' class TestJsonConfiguration: def init_config(self): self.json_config_ADR22 = JsonConfiguration(json_string_test_ADR_22, LOGGER) self.json_config_ADR18 = JsonConfiguration(json_string_test_ADR_18, LOGGER) self.json_config_pre_ADR18 = JsonConfiguration(json_string_test_pre_ADR_18, LOGGER) def test_detect_version(self): self.init_config() self.json_config_ADR22.detect_version() assert self.json_config_ADR22.major_version == 1 assert self.json_config_ADR22.minor_version == 0 self.json_config_ADR18.detect_version() assert self.json_config_ADR18.major_version == 1 assert self.json_config_ADR18.minor_version == 0 self.json_config_pre_ADR18.detect_version() assert self.json_config_pre_ADR18.major_version == 0 assert self.json_config_pre_ADR18.minor_version == 1 def test_detect_version_with_invalid_uri(self): with pytest.raises(ValueError, match= r'Wrong json schema URI!'): JsonConfiguration(json_string_invalid_URI, LOGGER).detect_version() def test_get_section_when_section_is_present(self): self.init_config() assert isinstance(self.json_config_ADR18.get_section('cbf'), dict) assert 'fsp' in self.json_config_ADR18.get_section('cbf').keys() assert isinstance(self.json_config_pre_ADR18.get_section('fsp'), list) def test_get_section_when_section_is_not_present(self): self.init_config() with pytest.raises(ValueError): self.json_config_pre_ADR18.get_section('cbf') with pytest.raises(ValueError): self.json_config_ADR18.get_section('fsp') def test_conversion_10_01(self): self.init_config() self.json_config_ADR18.conversion_10_01() assert self.json_config_ADR18.config == self.json_config_pre_ADR18.config def test_conversion_10_01_invalid_schema(self): with pytest.raises(ValueError): JsonConfiguration(json_string_test_ADR_18_invalid, LOGGER).conversion_10_01() def test_build_json_cbf_version(self): self.init_config() assert self.json_config_ADR18.build_json_cbf() == self.json_config_pre_ADR18.build_json_cbf() assert self.json_config_ADR22.build_json_cbf() == self.json_config_pre_ADR18.build_json_cbf() def test_build_json_cbf_version_not_supported(self): self.init_config() # version in interface is changed to 2.0 to force the failing of the test self.json_config_ADR22.config['interface'] = "https://schema.skatelescope.org/ska-csp-configure/2.0" with pytest.raises(ValueError, match=r'version.*is not supported'): self.json_config_ADR22.build_json_cbf() Loading
csp-lmc-common/csp_lmc_common/csp_manage_json.py +26 −22 Original line number Diff line number Diff line import json from ska_telmodel.csp.schema import validate_csp_config from ska_telmodel.csp.version import validate_csp_config_version class JsonConfiguration: """ Loading @@ -18,50 +19,53 @@ class JsonConfiguration: #ADR-22 if 'interface' in self.config.keys(): interface = self.config['interface'] if validate_csp_config_version(interface): version_str = interface.split('/')[-1] self.major_version = int(version_str.split('.')[0]) self.minor_version = int(version_str.split('.')[1]) self.major_version , self.minor_version = map(int, version_str.split('.')) else: message = 'Wrong json schema URI!' self.logger.error(message) raise ValueError(message) #ADR-18 elif 'common' in self.config.keys(): self.major_version = 0 self.minor_version = 2 self.major_version, self.minor_version = (1,0) #pre ADR-18 else: self.major_version = 0 self.minor_version = 1 self.version = [self.major_version, self.minor_version] self.major_version, self.minor_version = (0,1) self.version = (self.major_version, self.minor_version) self.logger.info(f'Version is{self.major_version}{self.minor_version}') def get_section(self, name:str): """ Returns the correpondant section of json configuration Returns the correspondant section of json configuration. """ try: return self.config[name] except KeyError as e: self.logger.warning(e) self.logger.error(e) raise ValueError(e) from e def conversion_02_01(self): def conversion_10_01(self): """ Converts the json from version 0.2 to version 0.1 Converts the json from version 1.0 (ADR-18) to version 0.1. Note: Current Mid.CBF supports 0.1 Json schema. """ common_dict = self.get_section('common') cbf_dict = self.get_section('cbf') config_converted = {**common_dict, **cbf_dict} validate_csp_config(version=0, template=True, config=config_converted) #check template parameter validate_csp_config(version=0, config=config_converted) self.config = config_converted def build_json_cbf(self) -> dict: " def build_json_cbf(self) -> str: """ Returns the json configuration dictionary to be passed to cbf. If a version 0.2 is given, it converts to 0.1 " """ self.detect_version() if self.version == [0, 1]: if self.version == (0, 1): pass elif self.version == [0,2]: self.conversion_02_01() elif self.version == (1, 0): self.conversion_10_01() else: raise ValueError(f'version{self.version}is not supported') return json.dumps(self.config)
csp-lmc-common/tests/unit/json_config_test.py 0 → 100644 +175 −0 Original line number Diff line number Diff line from csp_lmc_common.csp_manage_json import JsonConfiguration import logging import pytest LOGGER = logging.getLogger(__name__) json_string_test_ADR_22 = """ { "interface": "https://schema.skatelescope.org/ska-csp-configure/1.0", "common": { "id": "sbi-mvp01-20200325-00001-science_A", "frequencyBand": "1" }, "cbf": { "fsp": [ { "fspID": 1, "functionMode": "CORR", "frequencySliceID": 1, "integrationTime": 1400, "corrBandwidth": 0, "channelAveragingMap": [[0,2], [744,0]], "fspChannelOffset": 0, "outputLinkMap": [[0,0], [200,1]] }, { "fspID": 2, "functionMode": "CORR", "frequencySliceID": 2, "integrationTime": 1400, "corrBandwidth": 0, "channelAveragingMap": [[0,2], [744,0]], "fspChannelOffset": 744, "outputLinkMap": [[0,4], [200,5]] } ] } } """ json_string_test_ADR_18 = """ { "common": { "id": "sbi-mvp01-20200325-00001-science_A", "frequencyBand": "1" }, "cbf": { "fsp": [ { "fspID": 1, "functionMode": "CORR", "frequencySliceID": 1, "integrationTime": 1400, "corrBandwidth": 0, "channelAveragingMap": [[0,2], [744,0]], "fspChannelOffset": 0, "outputLinkMap": [[0,0], [200,1]] }, { "fspID": 2, "functionMode": "CORR", "frequencySliceID": 2, "integrationTime": 1400, "corrBandwidth": 0, "channelAveragingMap": [[0,2], [744,0]], "fspChannelOffset": 744, "outputLinkMap": [[0,4], [200,5]] } ] } } """ #function mode missing json_string_test_ADR_18_invalid = """ { "common": { "id": "sbi-mvp01-20200325-00001-science_A", "frequencyBand": "1", }, "cbf": { "fsp": [ { "fspID": 1, "frequencySliceID": 1, "integrationTime": 1400, "corrBandwidth": 0, "channelAveragingMap": [[0,2], [744,0]], "fspChannelOffset": 0, "outputLinkMap": [[0,0], [200,1]] }, { "fspID": 2, "functionMode": "CORR", "frequencySliceID": 2, "integrationTime": 1400, "corrBandwidth": 0, "channelAveragingMap": [[0,2], [744,0]], "fspChannelOffset": 744, "outputLinkMap": [[0,4], [200,5]] } ] } } """ json_string_test_pre_ADR_18 = """ { "id": "sbi-mvp01-20200325-00001-science_A", "frequencyBand": "1", "fsp": [ { "fspID": 1, "functionMode": "CORR", "frequencySliceID": 1, "integrationTime": 1400, "corrBandwidth": 0, "channelAveragingMap": [[0,2], [744,0]], "fspChannelOffset": 0, "outputLinkMap": [[0,0], [200,1]] }, { "fspID": 2, "functionMode": "CORR", "frequencySliceID": 2, "integrationTime": 1400, "corrBandwidth": 0, "channelAveragingMap": [[0,2], [744,0]], "fspChannelOffset": 744, "outputLinkMap": [[0,4], [200,5]] } ] } """ json_string_invalid_URI = '{"interface": "invaliduri.com/0.1"}' class TestJsonConfiguration: def init_config(self): self.json_config_ADR22 = JsonConfiguration(json_string_test_ADR_22, LOGGER) self.json_config_ADR18 = JsonConfiguration(json_string_test_ADR_18, LOGGER) self.json_config_pre_ADR18 = JsonConfiguration(json_string_test_pre_ADR_18, LOGGER) def test_detect_version(self): self.init_config() self.json_config_ADR22.detect_version() assert self.json_config_ADR22.major_version == 1 assert self.json_config_ADR22.minor_version == 0 self.json_config_ADR18.detect_version() assert self.json_config_ADR18.major_version == 1 assert self.json_config_ADR18.minor_version == 0 self.json_config_pre_ADR18.detect_version() assert self.json_config_pre_ADR18.major_version == 0 assert self.json_config_pre_ADR18.minor_version == 1 def test_detect_version_with_invalid_uri(self): with pytest.raises(ValueError, match= r'Wrong json schema URI!'): JsonConfiguration(json_string_invalid_URI, LOGGER).detect_version() def test_get_section_when_section_is_present(self): self.init_config() assert isinstance(self.json_config_ADR18.get_section('cbf'), dict) assert 'fsp' in self.json_config_ADR18.get_section('cbf').keys() assert isinstance(self.json_config_pre_ADR18.get_section('fsp'), list) def test_get_section_when_section_is_not_present(self): self.init_config() with pytest.raises(ValueError): self.json_config_pre_ADR18.get_section('cbf') with pytest.raises(ValueError): self.json_config_ADR18.get_section('fsp') def test_conversion_10_01(self): self.init_config() self.json_config_ADR18.conversion_10_01() assert self.json_config_ADR18.config == self.json_config_pre_ADR18.config def test_conversion_10_01_invalid_schema(self): with pytest.raises(ValueError): JsonConfiguration(json_string_test_ADR_18_invalid, LOGGER).conversion_10_01() def test_build_json_cbf_version(self): self.init_config() assert self.json_config_ADR18.build_json_cbf() == self.json_config_pre_ADR18.build_json_cbf() assert self.json_config_ADR22.build_json_cbf() == self.json_config_pre_ADR18.build_json_cbf() def test_build_json_cbf_version_not_supported(self): self.init_config() # version in interface is changed to 2.0 to force the failing of the test self.json_config_ADR22.config['interface'] = "https://schema.skatelescope.org/ska-csp-configure/2.0" with pytest.raises(ValueError, match=r'version.*is not supported'): self.json_config_ADR22.build_json_cbf()
