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
......@@ -3,12 +3,14 @@ import configparser
from pathlib import Path
from typing import Union, List, Optional
from PySide2 import QtCore, QtGui, QtWidgets
from PySide2.QtWidgets import (
from PySide6 import QtCore, QtGui, QtWidgets
from PySide6.QtWidgets import (
QMainWindow, QWidget, QTextBrowser, QPushButton, QLineEdit, QDateEdit,
QListWidget, QCheckBox, QRadioButton, QMenu, QAction, QLabel, QFrame,
QVBoxLayout, QHBoxLayout, QGridLayout, QAbstractItemView, QShortcut,
QActionGroup, QButtonGroup,
QListWidget, QCheckBox, QRadioButton, QMenu, QLabel, QFrame,
QVBoxLayout, QHBoxLayout, QGridLayout, QAbstractItemView, QButtonGroup,
)
from PySide6.QtGui import (
QAction, QActionGroup, QShortcut
)
from sohstationviewer.view.calendar.calendar_widget import CalendarWidget
......
from __future__ import annotations
from PySide2.QtGui import QCloseEvent
from PySide2.QtWidgets import QWidget
from PySide6.QtGui import QCloseEvent
from PySide6.QtWidgets import QWidget
class DialogAlreadyOpenedError(Exception):
......
......@@ -12,7 +12,7 @@ from sohstationviewer.controller.processing import (
get_data_type_from_file
)
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')
......@@ -288,6 +288,22 @@ class TestDetectDataType(TestCase):
f"{self.dir2.name}: Q330\n\n"
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):
"""
Test basic functionality of detect_data_type - can't detect any data
......
......@@ -18,7 +18,8 @@ from sohstationviewer.controller.util import (
get_time_4,
get_val,
rtn_pattern,
add_thousand_separator
add_thousand_separator,
get_valid_file_count
)
TEST_DATA_DIR = os.path.realpath(os.path.join(
......@@ -367,3 +368,11 @@ class TestFmti(TestCase):
val = -62362.32523
expected = '-62,362'
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)
......@@ -21,7 +21,7 @@ class TestMSeed(TestCase):
args = {
'data_type': 'Q330',
'is_multiplex': False,
'folder': '_',
'list_of_dir': ['_'],
'on_unittest': True
}
with self.assertRaises(ProcessingDataError) as context:
......@@ -36,7 +36,7 @@ class TestMSeed(TestCase):
args = {
'data_type': 'Pegasus',
'is_multiplex': False,
'folder': pegasus_data,
'list_of_dir': [pegasus_data],
'req_soh_chans': ['_'],
'on_unittest': True
}
......@@ -59,7 +59,7 @@ class TestMSeed(TestCase):
args = {
'data_type': 'Pegasus',
'is_multiplex': False,
'folder': pegasus_data,
'list_of_dir': [pegasus_data],
'req_soh_chans': ['VE1'],
'on_unittest': True
}
......@@ -85,7 +85,7 @@ class TestMSeed(TestCase):
args = {
'data_type': 'Pegasus',
'is_multiplex': False,
'folder': pegasus_data,
'list_of_dir': [pegasus_data],
'req_wf_chans': ['HH1'],
'req_soh_chans': ['_'],
'on_unittest': True
......@@ -111,7 +111,7 @@ class TestMSeed(TestCase):
args = {
'data_type': 'Q330',
'is_multiplex': False,
'folder': q330_data,
'list_of_dir': [q330_data],
'req_soh_chans': ['LOG'],
}
obj = MSeed(**args)
......@@ -135,7 +135,7 @@ class TestMSeed(TestCase):
args = {
'data_type': 'Q330',
'is_multiplex': True,
'folder': blockettes_data,
'list_of_dir': [blockettes_data],
'req_soh_chans': ['ACE'],
}
obj = MSeed(**args)
......@@ -155,7 +155,7 @@ class TestMSeed(TestCase):
args = {
'data_type': 'Q330',
'is_multiplex': False,
'folder': multiplex_data,
'list_of_dir': [multiplex_data],
'req_soh_chans': [],
'req_wf_chans': ['EL1']
}
......@@ -177,7 +177,7 @@ class TestMSeed(TestCase):
args = {
'data_type': 'Q330',
'is_multiplex': True,
'folder': multiplex_data,
'list_of_dir': [multiplex_data],
'req_soh_chans': [],
'req_wf_chans': ['EL1']
}
......@@ -200,7 +200,7 @@ class TestMSeed(TestCase):
args = {
'data_type': 'Q330',
'is_multiplex': False,
'folder': multiplex_data,
'list_of_dir': [multiplex_data],
'req_soh_chans': [],
'req_wf_chans': ['EL2']
}
......@@ -212,7 +212,7 @@ class TestMSeed(TestCase):
args = {
'data_type': 'Q330',
'is_multiplex': True,
'folder': multiplex_data,
'list_of_dir': [multiplex_data],
'req_soh_chans': [],
'req_wf_chans': ['EL2']
}
......@@ -229,13 +229,39 @@ class TestMSeed(TestCase):
self.assertEqual(len(obj.waveform_data['3203']['EL2']['tracesInfo']),
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):
# 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
args = {
'data_type': 'Q330',
'is_multiplex': False,
'folder': q330_data,
'list_of_dir': [q330_data],
'req_soh_chans': [],
'read_start': 1625456018.0,
'read_end': 1625505627.9998999
......@@ -252,7 +278,7 @@ class TestMSeed(TestCase):
args = {
'data_type': 'Q330',
'is_multiplex': False,
'folder': q330_data,
'list_of_dir': [q330_data],
'req_soh_chans': [],
'read_start': 1625356018.0,
'read_end': 1625405627.9998999
......@@ -269,7 +295,7 @@ class TestMSeed(TestCase):
args = {
'data_type': 'Q330',
'is_multiplex': False,
'folder': q330_data,
'list_of_dir': [q330_data],
'req_soh_chans': [],
'req_wf_chans': ['LHE']
}
......@@ -292,7 +318,7 @@ class TestMSeed(TestCase):
args = {
'data_type': 'Q330',
'is_multiplex': True,
'folder': q330_data,
'list_of_dir': [q330_data],
'req_soh_chans': [],
'req_wf_chans': [],
'include_mp123zne': True
......@@ -315,7 +341,7 @@ class TestMSeed(TestCase):
args = {
'data_type': 'Centaur',
'is_multiplex': True,
'folder': centaur_data,
'list_of_dir': [centaur_data],
'req_soh_chans': [],
'gap_minimum': 60
}
......@@ -340,7 +366,7 @@ class TestMSeed(TestCase):
args = {
'data_type': 'Centaur',
'is_multiplex': True,
'folder': centaur_data,
'list_of_dir': [centaur_data],
'req_soh_chans': [],
'gap_minimum': None
}
......
......@@ -17,7 +17,7 @@ class TestReftek(TestCase):
# raise exception when path not exist
args = {
'data_type': 'RT130',
'folder': '_',
'list_of_dir': ['_'],
'rt130_waveform_data_req': False,
'on_unittest': True
}
......@@ -31,7 +31,7 @@ class TestReftek(TestCase):
def test_read_soh(self):
args = {
'data_type': 'RT130',
'folder': reftek_data,
'list_of_dir': [reftek_data],
'req_soh_chans': [],
'rt130_waveform_data_req': False,
'on_unittest': True
......@@ -56,13 +56,13 @@ class TestReftek(TestCase):
'\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(list(obj.soh_data[('92EB', '25')].keys()),
expected_soh)
self.assertEqual(sorted(list(obj.soh_data[('92EB', '25')].keys())),
sorted(expected_soh))
def test_read_waveform(self):
args = {
'data_type': 'RT130',
'folder': reftek_data,
'list_of_dir': [reftek_data],
'req_soh_chans': [],
'req_wf_chans': [1],
'rt130_waveform_data_req': True,
......@@ -85,7 +85,7 @@ class TestReftek(TestCase):
def test_read_mass_pos(self):
args = {
'data_type': 'RT130',
'folder': reftek_data,
'list_of_dir': [reftek_data],
'req_soh_chans': ['_'],
'include_mp123zne': True,
'rt130_waveform_data_req': False,
......@@ -107,7 +107,7 @@ class TestReftek(TestCase):
with self.subTest("no gap_minimum set"):
args = {
'data_type': 'RT130',
'folder': reftek_gap_data,
'list_of_dir': [reftek_gap_data],
'req_soh_chans': [],
'req_wf_chans': ['*'],
'rt130_waveform_data_req': True,
......@@ -129,7 +129,7 @@ class TestReftek(TestCase):
with self.subTest("has gap_minimum set"):
args = {
'data_type': 'RT130',
'folder': reftek_gap_data,
'list_of_dir': [reftek_gap_data],
'req_soh_chans': [],
'req_wf_chans': ['*'],
'rt130_waveform_data_req': True,
......@@ -148,3 +148,45 @@ class TestReftek(TestCase):
['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]])