from datetime import datetime
from pathlib import Path
from unittest.mock import patch
from sohstationviewer.conf.constants import SOFTWARE_VERSION
from sohstationviewer.model.reftek_data.reftek import RT130
from sohstationviewer.model.general_data.general_data import \
ProcessingDataError
from tests.base_test_case import BaseTestCase
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")
# noinspection PyPep8Naming
class Mock_datetime(datetime): # noqa: N801
"""
Class used to mock datetime.now() so that it returns a static time.
"""
@classmethod
def now(cls, tz=None) -> datetime:
"""
Mock datetime.now to return a static datetime object.
:return: a static datetime object
"""
# The formatting method we used does not take millisecond into
# account, so we can omit it.
return datetime(2023, 10, 5, 15, 8, 5)
class TestReftek(BaseTestCase):
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"
)
@patch('datetime.datetime', Mock_datetime)
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)
formatted_time = 'Thu Oct 5 15:08:05 2023'
self.assertEqual(
obj.log_data[('92EB', '25')]['SOH'][0][:100],
f'SOHStationViewer: v{SOFTWARE_VERSION} '
f'Run time (UTC): {formatted_time}'
'\n\nState of Health 17:150:00:0')
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]])
@patch('datetime.datetime', Mock_datetime)
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']
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)
formatted_time = 'Thu Oct 5 15:08:05 2023'
self.assertEqual(
obj.log_data[('92EB', '25')]['SOH'][0][:100],
f'SOHStationViewer: v{SOFTWARE_VERSION} '
f'Run time (UTC): {formatted_time}'
'\n\nState of Health 17:150:00:0')
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]])