from unittest import TestCase
from pathlib import Path
from sohstationviewer.model.reftek_data.reftek import RT130
from sohstationviewer.model.general_data.general_data import \
ProcessingDataError
TEST_DATA_DIR = Path(__file__).resolve().parent.parent.parent.joinpath(
'test_data')
reftek_data = TEST_DATA_DIR.joinpath("RT130-sample")
reftek_gap_data = TEST_DATA_DIR.joinpath("RT130-gap")
class TestReftek(TestCase):
def test_path_not_exist(self):
# raise exception when path not exist
args = {
'data_type': 'RT130',
'list_of_dir': ['_'],
'rt130_waveform_data_req': False,
'on_unittest': True
}
with self.assertRaises(ProcessingDataError) as context:
RT130(**args)
self.assertEqual(
str(context.exception),
"Path '_' not exist"
)
def test_read_soh(self):
args = {
'data_type': 'RT130',
'list_of_dir': [reftek_data],
'req_soh_chans': [],
'rt130_waveform_data_req': False,
'on_unittest': True
}
expected_soh = [
'SOH/Data Def', 'Battery Volt', 'DAS Temp', 'Backup Volt',
'Disk Usage1', 'Disk Usage2', 'Dump Called/Comp', 'GPS On/Off/Err',
'GPS Lk/Unlk', 'Clk Phase Err']
obj = RT130(**args)
self.assertEqual(obj.found_data_streams, [9])
self.assertEqual(obj.keys, [('92EB', '25')])
self.assertEqual(
list(obj.stream_header_by_key_chan[('92EB', '25')].keys()),
[])
self.assertEqual(list(obj.log_data.keys()), ['TEXT', ('92EB', '25')])
self.assertEqual(len(obj.log_data['TEXT']), 0)
self.assertEqual(list(obj.log_data[('92EB', '25')].keys()), ['SOH'])
self.assertEqual(len(obj.log_data[('92EB', '25')]['SOH']), 1)
self.assertEqual(
obj.log_data[('92EB', '25')]['SOH'][0][:100],
'\nState of Health 17:150:00:00:00:000 ST: 92EB'
'\n150:00:00:00 REF TEK 130'
'\r\n150:00:00:00 CPU SOFTWARE')
self.assertEqual(list(obj.soh_data.keys()), [('92EB', '25')])
self.assertEqual(sorted(list(obj.soh_data[('92EB', '25')].keys())),
sorted(expected_soh))
def test_read_waveform(self):
args = {
'data_type': 'RT130',
'list_of_dir': [reftek_data],
'req_soh_chans': [],
'req_wf_chans': [1],
'rt130_waveform_data_req': True,
'on_unittest': True
}
expected_waveform = ['DS1-1', 'DS1-2', 'DS1-3']
obj = RT130(**args)
self.assertEqual(obj.found_data_streams, [9, 1, 1])
self.assertEqual(obj.keys, [('92EB', '25')])
self.assertEqual(
list(obj.stream_header_by_key_chan[('92EB', '25')].keys()),
expected_waveform)
self.assertEqual(list(obj.waveform_data[('92EB', '25')].keys()),
expected_waveform)
self.assertEqual(list(obj.log_data.keys()), ['TEXT', ('92EB', '25')])
self.assertIn('Event DS1',
list(obj.soh_data[('92EB', '25')].keys()))
def test_read_mass_pos(self):
args = {
'data_type': 'RT130',
'list_of_dir': [reftek_data],
'req_soh_chans': ['_'],
'include_mp123zne': True,
'rt130_waveform_data_req': False,
'on_unittest': True
}
expected_mass_pos = ['MassPos1', 'MassPos2', 'MassPos3']
obj = RT130(**args)
self.assertEqual(obj.found_data_streams, [9])
self.assertEqual(obj.keys, [('92EB', '25')])
self.assertEqual(
list(obj.stream_header_by_key_chan[('92EB', '25')].keys()),
expected_mass_pos)
self.assertEqual(list(obj.mass_pos_data[('92EB', '25')].keys()),
expected_mass_pos)
self.assertEqual(list(obj.log_data.keys()), ['TEXT', ('92EB', '25')])
def test_gap(self):
expected_waveform = ['DS2-1', 'DS2-2', 'DS2-3']
with self.subTest("no gap_minimum set"):
args = {
'data_type': 'RT130',
'list_of_dir': [reftek_gap_data],
'req_soh_chans': [],
'req_wf_chans': ['*'],
'rt130_waveform_data_req': True,
'on_unittest': True
}
obj = RT130(**args)
self.assertEqual(obj.found_data_streams, [2, 2])
self.assertEqual(obj.keys, [('98AD', '0')])
self.assertEqual(
list(obj.stream_header_by_key_chan[('98AD', '0')].keys()),
expected_waveform)
self.assertEqual(list(obj.waveform_data[('98AD', '0')].keys()),
expected_waveform)
self.assertEqual(list(obj.log_data.keys()),
['TEXT', ('98AD', '0')])
self.assertEqual(obj.gaps[('98AD', '0')], [])
with self.subTest("has gap_minimum set"):
args = {
'data_type': 'RT130',
'list_of_dir': [reftek_gap_data],
'req_soh_chans': [],
'req_wf_chans': ['*'],
'rt130_waveform_data_req': True,
'gap_minimum': 60,
'on_unittest': True
}
obj = RT130(**args)
self.assertEqual(obj.found_data_streams, [2, 2])
self.assertEqual(obj.keys, [('98AD', '0')])
self.assertEqual(
list(obj.stream_header_by_key_chan[('98AD', '0')].keys()),
expected_waveform)
self.assertEqual(list(obj.waveform_data[('98AD', '0')].keys()),
expected_waveform)
self.assertEqual(list(obj.log_data.keys()),
['TEXT', ('98AD', '0')])
self.assertEqual(obj.gaps[('98AD', '0')],
[[1648493999.64, 1648508400.64]])
def test_select_2_folders(self):
args = {
'data_type': 'RT130',
'list_of_dir': [reftek_data, reftek_gap_data],
'req_soh_chans': [],
'req_wf_chans': ['*'],
'rt130_waveform_data_req': True,
'gap_minimum': 60,
'on_unittest': True
}
expected_soh = [
'SOH/Data Def', 'Battery Volt', 'DAS Temp', 'Backup Volt',
'Disk Usage1', 'Disk Usage2', 'Dump Called/Comp', 'GPS On/Off/Err',
'GPS Lk/Unlk', 'Clk Phase Err', 'Event DS1', 'Event DS9']
expected_waveform = ['DS1-1', 'DS1-2', 'DS1-3']
obj = RT130(**args)
self.assertEqual(obj.found_data_streams, [9, 1, 1, 2, 2])
self.assertEqual(obj.keys, [('92EB', '25'), ('98AD', '0')])
self.assertEqual(obj.selected_key, ('92EB', '25'))
self.assertEqual(
list(obj.stream_header_by_key_chan[('92EB', '25')].keys()),
expected_waveform)
self.assertEqual(list(obj.log_data.keys()),
['TEXT', ('92EB', '25'), ('98AD', '0')])
self.assertEqual(len(obj.log_data['TEXT']), 0)
self.assertEqual(list(obj.log_data[('92EB', '25')].keys()), ['SOH'])
self.assertEqual(len(obj.log_data[('92EB', '25')]['SOH']), 1)
self.assertEqual(
obj.log_data[('92EB', '25')]['SOH'][0][:100],
'\nState of Health 17:150:00:00:00:000 ST: 92EB'
'\n150:00:00:00 REF TEK 130'
'\r\n150:00:00:00 CPU SOFTWARE')
self.assertEqual(list(obj.soh_data.keys()),
[('92EB', '25'), ('98AD', '0')])
self.assertEqual(sorted(list(obj.soh_data[('92EB', '25')].keys())),
sorted(expected_soh))
self.assertEqual(list(obj.gaps.keys()),
[('92EB', '25'), ('98AD', '0')])
self.assertEqual(obj.gaps[('92EB', '25')], [])
self.assertEqual(obj.gaps[('98AD', '0')],
[[1648493999.64, 1648508400.64]])