Skip to content
Snippets Groups Projects

Compare revisions

Changes are shown as if the source revision was being merged into the target revision. Learn more about comparing revisions.

Source

Select target project
No results found

Target

Select target project
  • software_public/passoft/sohstationviewer
1 result
Show changes
Hide whitespace changes
Inline Side-by-side
tests/controller/test_processing.py
View file @ b794b813
...@@ -12,7 +12,7 @@ from sohstationviewer.controller.processing import ( ...@@ -12,7 +12,7 @@ from sohstationviewer.controller.processing import (
get_data_type_from_file get_data_type_from_file
) )
from sohstationviewer.database.extract_data import get_signature_channels from sohstationviewer.database.extract_data import get_signature_channels
from PySide2 import QtWidgets from PySide6 import QtWidgets
TEST_DATA_DIR = Path(__file__).resolve().parent.parent.joinpath('test_data') TEST_DATA_DIR = Path(__file__).resolve().parent.parent.joinpath('test_data')
...@@ -288,6 +288,22 @@ class TestDetectDataType(TestCase): ...@@ -288,6 +288,22 @@ class TestDetectDataType(TestCase):
f"{self.dir2.name}: Q330\n\n" f"{self.dir2.name}: Q330\n\n"
f"Please have only data that related to each other.") f"Please have only data that related to each other.")
def test_same_data_types_different_multiplex(self):
"""
Test basic functionality of detect_data_type - the given directories
contain same data types but different multiplex.
"""
returned_data_types = [('Q330', True), ('Q330', False)]
self.mock_get_data_type_from_file.side_effect = returned_data_types
with self.assertRaises(Exception) as context:
detect_data_type([self.dir1.name, self.dir2.name])
self.assertEqual(
str(context.exception),
"There are both multiplexed and non-multiplexed data "
"detected.\n\nPlease have only data that related to"
" each other.")
def test_unknown_data_type(self): def test_unknown_data_type(self):
""" """
Test basic functionality of detect_data_type - can't detect any data Test basic functionality of detect_data_type - can't detect any data
......
tests/controller/test_util.py
View file @ b794b813
...@@ -18,7 +18,8 @@ from sohstationviewer.controller.util import ( ...@@ -18,7 +18,8 @@ from sohstationviewer.controller.util import (
get_time_4, get_time_4,
get_val, get_val,
rtn_pattern, rtn_pattern,
add_thousand_separator add_thousand_separator,
get_valid_file_count
) )
TEST_DATA_DIR = os.path.realpath(os.path.join( TEST_DATA_DIR = os.path.realpath(os.path.join(
...@@ -367,3 +368,11 @@ class TestFmti(TestCase): ...@@ -367,3 +368,11 @@ class TestFmti(TestCase):
val = -62362.32523 val = -62362.32523
expected = '-62,362' expected = '-62,362'
self.assertEqual(add_thousand_separator(val), expected) self.assertEqual(add_thousand_separator(val), expected)
class TestGetTotalFiles(TestCase):
def test_get_valid_file_count(self):
list_of_dir = [os.path.join(TEST_DATA_DIR, 'Centaur-sample'),
os.path.join(TEST_DATA_DIR, 'Q330-sample')]
total_files = get_valid_file_count(list_of_dir)
self.assertEqual(total_files, 6)
tests/model/mseed_data/test_mseed.py
View file @ b794b813
...@@ -21,7 +21,7 @@ class TestMSeed(TestCase): ...@@ -21,7 +21,7 @@ class TestMSeed(TestCase):
args = { args = {
'data_type': 'Q330', 'data_type': 'Q330',
'is_multiplex': False, 'is_multiplex': False,
'folder': '_', 'list_of_dir': ['_'],
'on_unittest': True 'on_unittest': True
} }
with self.assertRaises(ProcessingDataError) as context: with self.assertRaises(ProcessingDataError) as context:
...@@ -36,7 +36,7 @@ class TestMSeed(TestCase): ...@@ -36,7 +36,7 @@ class TestMSeed(TestCase):
args = { args = {
'data_type': 'Pegasus', 'data_type': 'Pegasus',
'is_multiplex': False, 'is_multiplex': False,
'folder': pegasus_data, 'list_of_dir': [pegasus_data],
'req_soh_chans': ['_'], 'req_soh_chans': ['_'],
'on_unittest': True 'on_unittest': True
} }
...@@ -59,7 +59,7 @@ class TestMSeed(TestCase): ...@@ -59,7 +59,7 @@ class TestMSeed(TestCase):
args = { args = {
'data_type': 'Pegasus', 'data_type': 'Pegasus',
'is_multiplex': False, 'is_multiplex': False,
'folder': pegasus_data, 'list_of_dir': [pegasus_data],
'req_soh_chans': ['VE1'], 'req_soh_chans': ['VE1'],
'on_unittest': True 'on_unittest': True
} }
...@@ -85,7 +85,7 @@ class TestMSeed(TestCase): ...@@ -85,7 +85,7 @@ class TestMSeed(TestCase):
args = { args = {
'data_type': 'Pegasus', 'data_type': 'Pegasus',
'is_multiplex': False, 'is_multiplex': False,
'folder': pegasus_data, 'list_of_dir': [pegasus_data],
'req_wf_chans': ['HH1'], 'req_wf_chans': ['HH1'],
'req_soh_chans': ['_'], 'req_soh_chans': ['_'],
'on_unittest': True 'on_unittest': True
...@@ -111,7 +111,7 @@ class TestMSeed(TestCase): ...@@ -111,7 +111,7 @@ class TestMSeed(TestCase):
args = { args = {
'data_type': 'Q330', 'data_type': 'Q330',
'is_multiplex': False, 'is_multiplex': False,
'folder': q330_data, 'list_of_dir': [q330_data],
'req_soh_chans': ['LOG'], 'req_soh_chans': ['LOG'],
} }
obj = MSeed(**args) obj = MSeed(**args)
...@@ -135,7 +135,7 @@ class TestMSeed(TestCase): ...@@ -135,7 +135,7 @@ class TestMSeed(TestCase):
args = { args = {
'data_type': 'Q330', 'data_type': 'Q330',
'is_multiplex': True, 'is_multiplex': True,
'folder': blockettes_data, 'list_of_dir': [blockettes_data],
'req_soh_chans': ['ACE'], 'req_soh_chans': ['ACE'],
} }
obj = MSeed(**args) obj = MSeed(**args)
...@@ -155,7 +155,7 @@ class TestMSeed(TestCase): ...@@ -155,7 +155,7 @@ class TestMSeed(TestCase):
args = { args = {
'data_type': 'Q330', 'data_type': 'Q330',
'is_multiplex': False, 'is_multiplex': False,
'folder': multiplex_data, 'list_of_dir': [multiplex_data],
'req_soh_chans': [], 'req_soh_chans': [],
'req_wf_chans': ['EL1'] 'req_wf_chans': ['EL1']
} }
...@@ -177,7 +177,7 @@ class TestMSeed(TestCase): ...@@ -177,7 +177,7 @@ class TestMSeed(TestCase):
args = { args = {
'data_type': 'Q330', 'data_type': 'Q330',
'is_multiplex': True, 'is_multiplex': True,
'folder': multiplex_data, 'list_of_dir': [multiplex_data],
'req_soh_chans': [], 'req_soh_chans': [],
'req_wf_chans': ['EL1'] 'req_wf_chans': ['EL1']
} }
...@@ -200,7 +200,7 @@ class TestMSeed(TestCase): ...@@ -200,7 +200,7 @@ class TestMSeed(TestCase):
args = { args = {
'data_type': 'Q330', 'data_type': 'Q330',
'is_multiplex': False, 'is_multiplex': False,
'folder': multiplex_data, 'list_of_dir': [multiplex_data],
'req_soh_chans': [], 'req_soh_chans': [],
'req_wf_chans': ['EL2'] 'req_wf_chans': ['EL2']
} }
...@@ -212,7 +212,7 @@ class TestMSeed(TestCase): ...@@ -212,7 +212,7 @@ class TestMSeed(TestCase):
args = { args = {
'data_type': 'Q330', 'data_type': 'Q330',
'is_multiplex': True, 'is_multiplex': True,
'folder': multiplex_data, 'list_of_dir': [multiplex_data],
'req_soh_chans': [], 'req_soh_chans': [],
'req_wf_chans': ['EL2'] 'req_wf_chans': ['EL2']
} }
...@@ -229,13 +229,39 @@ class TestMSeed(TestCase): ...@@ -229,13 +229,39 @@ class TestMSeed(TestCase):
self.assertEqual(len(obj.waveform_data['3203']['EL2']['tracesInfo']), self.assertEqual(len(obj.waveform_data['3203']['EL2']['tracesInfo']),
1) 1)
def test_select_2_folders(self):
# is_multiplex = True => the selected channel will be read
args = {
'data_type': 'Q330',
'is_multiplex': True,
'list_of_dir': [multiplex_data, blockettes_data],
'req_soh_chans': [],
'req_wf_chans': ['EL2'],
'gap_minimum': 60
}
obj = MSeed(**args)
self.assertEqual(list(obj.waveform_data.keys()), ['3203'])
self.assertEqual(list(obj.waveform_data['3203'].keys()), ['EL2'])
self.assertEqual(obj.waveform_data['3203']['EL2']['samplerate'], 200)
self.assertEqual(obj.waveform_data['3203']['EL2']['startTmEpoch'],
1671730004.3100293)
self.assertEqual(obj.waveform_data['3203']['EL2']['endTmEpoch'],
1671735657.9148998)
self.assertEqual(obj.waveform_data['3203']['EL2']['size'], 268576)
self.assertEqual(obj.waveform_data['3203']['EL2']['gaps'],
[[1671730720.5549, 1671735031.2799978]])
self.assertEqual(len(obj.waveform_data['3203']['EL2']['tracesInfo']),
1)
self.assertEqual(obj.gaps['3203'],
[[1671730720.5549, 1671735031.2799978]])
def test_existing_time_range(self): def test_existing_time_range(self):
# check if data_time is from the given range, end time may get # check if data_time is from the given range, end time may get
# a little greater than read_end according to record's end time # a little greater than read_end according to record's end time
args = { args = {
'data_type': 'Q330', 'data_type': 'Q330',
'is_multiplex': False, 'is_multiplex': False,
'folder': q330_data, 'list_of_dir': [q330_data],
'req_soh_chans': [], 'req_soh_chans': [],
'read_start': 1625456018.0, 'read_start': 1625456018.0,
'read_end': 1625505627.9998999 'read_end': 1625505627.9998999
...@@ -252,7 +278,7 @@ class TestMSeed(TestCase): ...@@ -252,7 +278,7 @@ class TestMSeed(TestCase):
args = { args = {
'data_type': 'Q330', 'data_type': 'Q330',
'is_multiplex': False, 'is_multiplex': False,
'folder': q330_data, 'list_of_dir': [q330_data],
'req_soh_chans': [], 'req_soh_chans': [],
'read_start': 1625356018.0, 'read_start': 1625356018.0,
'read_end': 1625405627.9998999 'read_end': 1625405627.9998999
...@@ -269,7 +295,7 @@ class TestMSeed(TestCase): ...@@ -269,7 +295,7 @@ class TestMSeed(TestCase):
args = { args = {
'data_type': 'Q330', 'data_type': 'Q330',
'is_multiplex': False, 'is_multiplex': False,
'folder': q330_data, 'list_of_dir': [q330_data],
'req_soh_chans': [], 'req_soh_chans': [],
'req_wf_chans': ['LHE'] 'req_wf_chans': ['LHE']
} }
...@@ -292,7 +318,7 @@ class TestMSeed(TestCase): ...@@ -292,7 +318,7 @@ class TestMSeed(TestCase):
args = { args = {
'data_type': 'Q330', 'data_type': 'Q330',
'is_multiplex': True, 'is_multiplex': True,
'folder': q330_data, 'list_of_dir': [q330_data],
'req_soh_chans': [], 'req_soh_chans': [],
'req_wf_chans': [], 'req_wf_chans': [],
'include_mp123zne': True 'include_mp123zne': True
...@@ -315,7 +341,7 @@ class TestMSeed(TestCase): ...@@ -315,7 +341,7 @@ class TestMSeed(TestCase):
args = { args = {
'data_type': 'Centaur', 'data_type': 'Centaur',
'is_multiplex': True, 'is_multiplex': True,
'folder': centaur_data, 'list_of_dir': [centaur_data],
'req_soh_chans': [], 'req_soh_chans': [],
'gap_minimum': 60 'gap_minimum': 60
} }
...@@ -340,7 +366,7 @@ class TestMSeed(TestCase): ...@@ -340,7 +366,7 @@ class TestMSeed(TestCase):
args = { args = {
'data_type': 'Centaur', 'data_type': 'Centaur',
'is_multiplex': True, 'is_multiplex': True,
'folder': centaur_data, 'list_of_dir': [centaur_data],
'req_soh_chans': [], 'req_soh_chans': [],
'gap_minimum': None 'gap_minimum': None
} }
......
tests/model/reftek_data/test_reftek.py
View file @ b794b813
...@@ -17,7 +17,7 @@ class TestReftek(TestCase): ...@@ -17,7 +17,7 @@ class TestReftek(TestCase):
# raise exception when path not exist # raise exception when path not exist
args = { args = {
'data_type': 'RT130', 'data_type': 'RT130',
'folder': '_', 'list_of_dir': ['_'],
'rt130_waveform_data_req': False, 'rt130_waveform_data_req': False,
'on_unittest': True 'on_unittest': True
} }
...@@ -31,7 +31,7 @@ class TestReftek(TestCase): ...@@ -31,7 +31,7 @@ class TestReftek(TestCase):
def test_read_soh(self): def test_read_soh(self):
args = { args = {
'data_type': 'RT130', 'data_type': 'RT130',
'folder': reftek_data, 'list_of_dir': [reftek_data],
'req_soh_chans': [], 'req_soh_chans': [],
'rt130_waveform_data_req': False, 'rt130_waveform_data_req': False,
'on_unittest': True 'on_unittest': True
...@@ -56,13 +56,13 @@ class TestReftek(TestCase): ...@@ -56,13 +56,13 @@ class TestReftek(TestCase):
'\n150:00:00:00 REF TEK 130' '\n150:00:00:00 REF TEK 130'
'\r\n150:00:00:00 CPU SOFTWARE') '\r\n150:00:00:00 CPU SOFTWARE')
self.assertEqual(list(obj.soh_data.keys()), [('92EB', '25')]) self.assertEqual(list(obj.soh_data.keys()), [('92EB', '25')])
self.assertEqual(list(obj.soh_data[('92EB', '25')].keys()), self.assertEqual(sorted(list(obj.soh_data[('92EB', '25')].keys())),
expected_soh) sorted(expected_soh))
def test_read_waveform(self): def test_read_waveform(self):
args = { args = {
'data_type': 'RT130', 'data_type': 'RT130',
'folder': reftek_data, 'list_of_dir': [reftek_data],
'req_soh_chans': [], 'req_soh_chans': [],
'req_wf_chans': [1], 'req_wf_chans': [1],
'rt130_waveform_data_req': True, 'rt130_waveform_data_req': True,
...@@ -85,7 +85,7 @@ class TestReftek(TestCase): ...@@ -85,7 +85,7 @@ class TestReftek(TestCase):
def test_read_mass_pos(self): def test_read_mass_pos(self):
args = { args = {
'data_type': 'RT130', 'data_type': 'RT130',
'folder': reftek_data, 'list_of_dir': [reftek_data],
'req_soh_chans': ['_'], 'req_soh_chans': ['_'],
'include_mp123zne': True, 'include_mp123zne': True,
'rt130_waveform_data_req': False, 'rt130_waveform_data_req': False,
...@@ -107,7 +107,7 @@ class TestReftek(TestCase): ...@@ -107,7 +107,7 @@ class TestReftek(TestCase):
with self.subTest("no gap_minimum set"): with self.subTest("no gap_minimum set"):
args = { args = {
'data_type': 'RT130', 'data_type': 'RT130',
'folder': reftek_gap_data, 'list_of_dir': [reftek_gap_data],
'req_soh_chans': [], 'req_soh_chans': [],
'req_wf_chans': ['*'], 'req_wf_chans': ['*'],
'rt130_waveform_data_req': True, 'rt130_waveform_data_req': True,
...@@ -129,7 +129,7 @@ class TestReftek(TestCase): ...@@ -129,7 +129,7 @@ class TestReftek(TestCase):
with self.subTest("has gap_minimum set"): with self.subTest("has gap_minimum set"):
args = { args = {
'data_type': 'RT130', 'data_type': 'RT130',
'folder': reftek_gap_data, 'list_of_dir': [reftek_gap_data],
'req_soh_chans': [], 'req_soh_chans': [],
'req_wf_chans': ['*'], 'req_wf_chans': ['*'],
'rt130_waveform_data_req': True, 'rt130_waveform_data_req': True,
...@@ -148,3 +148,45 @@ class TestReftek(TestCase): ...@@ -148,3 +148,45 @@ class TestReftek(TestCase):
['TEXT', ('98AD', '0')]) ['TEXT', ('98AD', '0')])
self.assertEqual(obj.gaps[('98AD', '0')], self.assertEqual(obj.gaps[('98AD', '0')],
[[1648493999.64, 1648508400.64]]) [[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]])