test_extract_data.py

from sohstationviewer.database.extract_data import (
    get_chan_plot_info,
    get_seismic_chan_label,
    get_signature_channels,
    get_color_def,
    get_color_ranges,
    convert_actual_channel_to_db_channel_w_question_mark,
    get_convert_factor
)
from tests.base_test_case import BaseTestCase


class TestGetChanPlotInfo(BaseTestCase):
    def test_good_soh_channel_and_data_type(self):
        """
        Test basic functionality of get_chan_plot_info - channel and data type
        combination exists in database table `Channels`
        """
        expected_result = {'param': 'SOH data definitions',
                           'dbChannel': 'SOH/Data Def',
                           'channel': 'SOH/Data Def',
                           'plotType': 'upDownDots',
                           'height': 2,
                           'unit': '',
                           'convertFactor': 1,
                           'dbLabel': None,
                           'label': 'SOH/Data Def',
                           'fixPoint': 0,
                           'valueColors': 'Up:#00FFFF|Down:#FFFFFF'}
        self.assertDictEqual(get_chan_plot_info('SOH/Data Def', 'RT130'),
                             expected_result)

    def test_masspos_channel(self):
        with self.subTest("Mass position 'VM'"):
            expected_result = {'param': 'Mass position',
                               'dbChannel': 'VM?',
                               'channel': 'VM1',
                               'plotType': 'linesMasspos',
                               'height': 4,
                               'unit': 'V',
                               'convertFactor': 0.1,
                               'dbLabel': 'MassPos',
                               'label': 'VM1-MassPos',
                               'fixPoint': 1,
                               'valueColors': ''}
            self.assertDictEqual(get_chan_plot_info('VM1', 'Q330'),
                                 expected_result)

        with self.subTest("Mass position 'MassPos'"):
            expected_result = {'param': 'Mass position',
                               'dbChannel': 'MassPos?',
                               'channel': 'MassPos1',
                               'plotType': 'linesMasspos',
                               'height': 4,
                               'unit': 'V',
                               'convertFactor': 1,
                               'dbLabel': None,
                               'label': 'MassPos1',
                               'fixPoint': 1,
                               'valueColors': ''}
            self.assertDictEqual(get_chan_plot_info('MassPos1', 'RT130'),
                                 expected_result)

    def test_seismic_channel(self):
        with self.subTest("RT130 Seismic"):
            expected_result = {'param': 'Seismic data',
                               'dbChannel': 'SEISMIC',
                               'channel': 'DS2-1',
                               'plotType': 'linesSRate',
                               'height': 8,
                               'unit': '',
                               'convertFactor': 1,
                               'dbLabel': None,
                               'fixPoint': 0,
                               'valueColors': '',
                               'label': 'DS2-1'}
            self.assertDictEqual(get_chan_plot_info('DS2-1', 'RT130'),
                                 expected_result)

        with self.subTest("MSeed Seismic"):
            expected_result = {'param': 'Seismic data',
                               'dbChannel': 'SEISMIC',
                               'channel': 'LHE',
                               'plotType': 'linesSRate',
                               'height': 8,
                               'unit': '',
                               'convertFactor': 1,
                               'dbLabel': 'SeismicData',
                               'fixPoint': 0,
                               'valueColors': '',
                               'label': 'LHE-EW'}
            self.assertDictEqual(get_chan_plot_info('LHE', 'Q330'),
                                 expected_result)

    def test_data_type_is_unknown(self):
        """
        Test basic functionality of get_chan_plot_info - data type is the
        string 'Unknown'.
        """
        # Channel does not exist in database
        expected_result = {'param': 'Default',
                           'dbChannel': 'DEFAULT',
                           'channel': 'DEFAULT',
                           'plotType': 'linesDots',
                           'height': 2,
                           'unit': '',
                           'convertFactor': 1,
                           'dbLabel': '',
                           'label': 'DEFAULT-Bad Channel ID',
                           'fixPoint': 0,
                           'valueColors': 'Line:#00FF00'}
        self.assertDictEqual(get_chan_plot_info('Bad Channel ID', 'Unknown'),
                             expected_result)

        # Channel exist in database
        expected_result = {'param': 'Clock phase error',
                           'dbChannel': 'LCE',
                           'channel': 'LCE',
                           'plotType': 'linesDots',
                           'height': 3,
                           'unit': 'us',
                           'convertFactor': 1,
                           'dbLabel': 'PhaseError',
                           'label': 'LCE-PhaseError',
                           'fixPoint': 0,
                           'valueColors': 'Line:#FFFFFF|Dot:#FFFF00'}
        self.assertDictEqual(
            get_chan_plot_info('LCE', 'Unknown'),
            expected_result)

    def test_bad_channel_or_data_type(self):
        """
        Test basic functionality of get_chan_plot_info - channel and data type
        combination does not exist in database table Channels and data type is
        not the string 'Unknown'.
        """
        # noinspection PyDictCreation
        expected_result = {'param': 'Default',
                           'dbChannel': 'DEFAULT',
                           'channel': 'DEFAULT',
                           'plotType': 'linesDots',
                           'height': 2,
                           'unit': '',
                           'convertFactor': 1,
                           'dbLabel': '',
                           'label': 'DEFAULT-SOH/Data Def',
                           'fixPoint': 0,
                           'valueColors': 'Line:#00FF00'}

        # Data type has None value. None value comes from
        # controller.processing.detect_data_type.
        expected_result['label'] = 'DEFAULT-SOH/Data Def'
        self.assertDictEqual(get_chan_plot_info('SOH/Data Def', None),
                             expected_result)

        # Channel and data type are empty strings
        expected_result['label'] = 'DEFAULT-'
        self.assertDictEqual(get_chan_plot_info('', ''),
                             expected_result)

        # Channel exists in database but data type does not
        expected_result['label'] = 'DEFAULT-SOH/Data Def'
        self.assertDictEqual(
            get_chan_plot_info('SOH/Data Def', 'Bad Data Type'),
            expected_result
        )

        # Data type exists in database but channel does not
        expected_result['label'] = 'DEFAULT-Bad Channel ID'
        self.assertDictEqual(get_chan_plot_info('Bad Channel ID', 'RT130'),
                             expected_result)

        # Both channel and data type exists in database but not their
        # combination
        expected_result['label'] = 'DEFAULT-SOH/Data Def'
        self.assertDictEqual(get_chan_plot_info('SOH/Data Def', 'Q330'),
                             expected_result)


class TestGetSeismicChanLabel(BaseTestCase):
    def test_good_channel_id(self):
        """
        Test basic functionality of get_seismic_chan_label - channel ID ends
        in one of the keys in conf.dbSettings.dbConf['seisLabel'] or
        starts with 'DS'
        """
        # Channel ID does not start with 'DS'
        self.assertEqual(get_seismic_chan_label('CH1'), 'CH1-NS')
        self.assertEqual(get_seismic_chan_label('CH2'), 'CH2-EW')
        self.assertEqual(get_seismic_chan_label('CHG'), 'CHG')

        # Channel ID starts with 'DS'
        self.assertEqual(get_seismic_chan_label('DS-TEST-CHANNEL'),
                         'DS-TEST-CHANNEL')

    def test_bad_channel_id(self):
        """
        Test basic functionality of get_seismic_chan_label - channel ID does
            not end in one of the keys in conf.dbSettings.dbConf['seisLabel']
            or is the empty string.
        """
        self.assertRaises(IndexError, get_seismic_chan_label, '')


class TestGetSignatureChannels(BaseTestCase):
    def test_get_signature_channels(self):
        """Test basic functionality of get_signature_channels"""
        self.assertIsInstance(get_signature_channels(), dict)


class TestGetColorDef(BaseTestCase):
    def test_get_color_def(self):
        """Test basic functionality of get_color_def"""
        colors = get_color_def()
        expected_colors = ['K', 'U', 'C', 'G', 'Y', 'R', 'M', 'E']
        self.assertListEqual(colors, expected_colors)


class TestGetColorRanges(BaseTestCase):
    def test_get_color_ranges(self):
        """Test basic functionality of get_color_ranges"""
        names, all_counts, all_display_strings = get_color_ranges()
        num_color_def = 7

        expected_names = ['Antarctica', 'Low', 'Medium', 'High']
        self.assertEqual(names, expected_names)

        # Check that each name correspond to a list of counts and list of
        # of strings to display
        self.assertEqual(len(names), len(all_counts))
        self.assertEqual(len(names), len(all_display_strings))

        # Check that each list of counts have enough items
        for counts in all_counts:
            self.assertEqual(len(counts), num_color_def)

        # Check that each list of strings to display have enough items
        for display_strings in all_display_strings:
            self.assertEqual(len(display_strings), num_color_def + 1)


class TestConvertActualChannelToDBChannelWQuestionMark(BaseTestCase):
    def test_question_channel(self):
        ret = convert_actual_channel_to_db_channel_w_question_mark('VM1', '')
        self.assertEqual(ret, 'VM?')

    def test_non_question_channel(self):
        ret = convert_actual_channel_to_db_channel_w_question_mark('VCO', '')
        self.assertEqual(ret, 'VCO')

    def test_q330_letter_masspos_channel(self):
        for suffix in ['Z', 'N', 'E', 'U', 'V', 'W']:
            channel = 'VM' + suffix
            with self.subTest(f'Test {channel}'):
                ret = convert_actual_channel_to_db_channel_w_question_mark(
                    channel, 'Q330'
                )
                self.assertEqual(ret, 'VM?')


class TestGetConvertFactor(BaseTestCase):
    def test_question_channel(self):
        ret = get_convert_factor('VM1', 'Centaur')
        self.assertEqual(ret, 10**(-6))

    def test_non_question_channel(self):
        ret = get_convert_factor('VEP', 'Q330')
        self.assertEqual(ret, 0.15)