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
Commits on Source (2)
No preview for this file type
No preview for this file type
......@@ -12,7 +12,7 @@ from sohstationviewer.model.general_data.general_data import GeneralData
from sohstationviewer.model.general_data.general_data_helper import read_text
from sohstationviewer.model.mseed_data.mseed_helper import \
retrieve_nets_from_data_dict
retrieve_nets_from_data_dict, split_vst_channel
from sohstationviewer.model.mseed_data.record_reader_helper import \
MSeedReadError
from sohstationviewer.model.mseed_data.mseed_reader import MSeedReader
......@@ -41,6 +41,8 @@ class MSeed(GeneralData):
self.distribute_log_text_to_station()
self.retrieve_nets_from_data_dicts()
super().finalize_data()
if self.data_type == 'Pegasus':
self.split_vst_channel()
def read_folders(self) -> None:
"""
......@@ -153,3 +155,7 @@ class MSeed(GeneralData):
if 'TXT' not in self.log_data[sta]:
self.log_data[sta]['TXT'] = []
self.log_data[sta]['TXT'].append(self.log_texts[path2file])
def split_vst_channel(self):
for data_set_id in self.data_set_ids:
split_vst_channel(data_set_id, self.soh_data)
from typing import List, Dict
from copy import deepcopy
import numpy as np
def retrieve_nets_from_data_dict(data_dict: Dict,
......@@ -15,3 +17,35 @@ def retrieve_nets_from_data_dict(data_dict: Dict,
for c in data_dict[sta_id]:
nets_by_sta[sta_id].update(
data_dict[sta_id][c]['nets'])
def split_vst_channel(selected_data_set_id: str, data_dict: Dict):
"""
Each value consists of 4 bit.
This function will split each bit into one array that will be the data
of the channel VST<bit position> which has other keys' info from the
original channel VST.
The original VST channel is removed at the end of the function.
:param selected_data_set_id: the id of the selected data set
:param data_dict: data of the selected key
"""
try:
selected_data_dict = data_dict[selected_data_set_id]
except KeyError:
return
vst_channels_data = [[], [], [], []]
if 'VST' not in selected_data_dict:
return
data = selected_data_dict['VST']['tracesInfo'][0]['data']
for d in data:
b = '{0:04b}'.format(d)
for i in range(4):
vst_channels_data[i].append(int(b[i]))
for i in range(4):
name = f'VST{3 - i}'
selected_data_dict[name] = deepcopy(
selected_data_dict['VST'])
selected_data_dict[name]['tracesInfo'][0]['data'] = np.array(
vst_channels_data[i])
del selected_data_dict['VST']
......@@ -774,6 +774,12 @@ class ChannelPreferDialog(OneWindowAtATimeDialog):
f" WHERE dataType='{data_type}'"
f" AND param NOT IN ('Seismic data', 'Mass position')"
f" ORDER BY dataType ASC")
# VST is divided into 4 channel in DB for plotting,
# But to read from file, we need original name VST
if ('VST0',) in channel_rows:
for i in range(4):
channel_rows.remove((f'VST{i}',))
channel_rows.append(('VST',))
return sorted([c[0] for c in channel_rows])
@staticmethod
......
......@@ -256,6 +256,14 @@ class MultiThreadedPlottingWidget(PlottingWidget):
self.start_tm = start_tm
self.end_tm = end_tm
self.time_ticks_total = time_ticks_total
if 'VST' in pref_order:
# pref_order use original name VST to read from file.
# For plotting, we need the name that will show in the figure.
vst_index = pref_order.index('VST')
pref_order.remove('VST')
for i in range(3, -1, -1):
pref_order.insert(vst_index, f'VST{i}')
self.pref_order = pref_order
if not self.is_working:
self.reset_widget()
......
......@@ -236,7 +236,7 @@ class Plotting:
ax,
sample_no_list=[len(points_list[0]), None, len(points_list[1])],
sample_no_colors=[colors[0], None, colors[1]],
sample_no_pos=[0.25, None, 0.75],
sample_no_pos=[0.15, None, 0.85],
chan_db_info=chan_db_info)
# x_bottom, x_top are the times of data points to be displayed at
......
import numpy as np
from sohstationviewer.model.mseed_data.mseed_helper import (
retrieve_nets_from_data_dict
retrieve_nets_from_data_dict, split_vst_channel
)
from tests.base_test_case import BaseTestCase
......@@ -22,3 +24,78 @@ class TestRetrieveNetsFromDataDict(BaseTestCase):
self.assertEqual(sorted(list(self.nets_by_sta['STA1'])),
['NET1', 'NET2', 'NET3'])
self.assertEqual(sorted(list(self.nets_by_sta['STA2'])), ['NET1'])
class TestSpitVSTChannel(BaseTestCase):
def test_has_vst(self):
data_dict = {
'STA': {
'VST': {
'tracesInfo': [{
'time': np.array([1, 2, 3, 4, 5]),
'data': np.array([0, 1, 5, 10, 11])}]
}}}
split_vst_channel('STA', data_dict)
self.assertNotIn('VST', data_dict['STA'].keys())
self.assertIn('VST0', data_dict['STA'].keys())
self.assertIn('VST1', data_dict['STA'].keys())
self.assertIn('VST2', data_dict['STA'].keys())
self.assertIn('VST3', data_dict['STA'].keys())
self.assertListEqual(
data_dict['STA']['VST0']['tracesInfo'][0]['time'].tolist(),
[1, 2, 3, 4, 5])
self.assertListEqual(
data_dict['STA']['VST0']['tracesInfo'][0]['data'].tolist(),
[0, 1, 1, 0, 1])
self.assertListEqual(
data_dict['STA']['VST1']['tracesInfo'][0]['data'].tolist(),
[0, 0, 0, 1, 1])
self.assertListEqual(
data_dict['STA']['VST2']['tracesInfo'][0]['data'].tolist(),
[0, 0, 1, 0, 0])
self.assertListEqual(
data_dict['STA']['VST3']['tracesInfo'][0]['data'].tolist(),
[0, 0, 0, 1, 1])
def test_has_vst_empty(self):
data_dict = {
'STA': {
'VST': {
'tracesInfo': [{
'time': np.array([]),
'data': np.array([])}]
}}}
split_vst_channel('STA', data_dict)
self.assertNotIn('VST', data_dict['STA'].keys())
self.assertIn('VST0', data_dict['STA'].keys())
self.assertIn('VST1', data_dict['STA'].keys())
self.assertIn('VST2', data_dict['STA'].keys())
self.assertIn('VST3', data_dict['STA'].keys())
self.assertListEqual(
data_dict['STA']['VST0']['tracesInfo'][0]['time'].tolist(),
[])
self.assertListEqual(
data_dict['STA']['VST0']['tracesInfo'][0]['data'].tolist(),
[])
self.assertListEqual(
data_dict['STA']['VST1']['tracesInfo'][0]['data'].tolist(),
[])
self.assertListEqual(
data_dict['STA']['VST2']['tracesInfo'][0]['data'].tolist(),
[])
self.assertListEqual(
data_dict['STA']['VST3']['tracesInfo'][0]['data'].tolist(),
[])
def test_has_no_vst(self):
data_dict = {
'STA': {
'VDP': {
'tracesInfo': [{
'time': np.array([1, 2]),
'data': np.array([1, 2])}]
}}}
split_vst_channel('STA', data_dict)
self.assertNotIn('VST0', data_dict['STA'].keys())