Newer
Older
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""Tests for `lemi_data` module."""
import logging
import pickle
import unittest
from obspy import UTCDateTime
from pathlib import Path
from lemi2seed.lemi_data import LemiData, log_file_path
OUTPUT_MSEED = Path(__file__).resolve().parent.joinpath('MSEED')
TEST_DIR = Path(__file__).resolve().parent.joinpath('test_data')
SCR_DIR = "lemi2seed.lemi_data"
logging.config.fileConfig(log_file_path)
logger = logging.getLogger(SCR_DIR)
class TestLemiData(unittest.TestCase):
"""Test suite for lemi_data.LemiData class"""
def setUp(self):
"""Set up test fixtures"""
self.path2data = TEST_DIR.joinpath("DATA0110")
self.path2data_corrupted_1 = TEST_DIR.joinpath("DATA0110_Corrupted_Level1")
self.path2data_corrupted_2 = TEST_DIR.joinpath("DATA0110_Corrupted_Level2")
files = ['202009302105.TXT', '202009302112.TXT', '202009302114.TXT', '202010010000.TXT']
self.data_files = [self.path2data.joinpath(x) for x in files]
def test_scan_path2data_valid_filenaming(self):
"""Test basic functionality of scan_path2data."""
lemi_data = LemiData(self.path2data, OUTPUT_MSEED)
self.assertListEqual(lemi_data.data_files, self.data_files)
def test_scan_path2data_erroneous_filenaming(self):
"""Test basic functionality of scan_path2data."""
path2data = TEST_DIR.joinpath('DATA0110_Erroneous_Filenaming')
with self.assertRaises(SystemExit) as cmd:
LemiData(path2data, OUTPUT_MSEED)
self.assertEqual(cmd.exception.code, 1)
def test_parse_file_valid_file(self):
"""Test basic functionality of parse_file."""
lemi_data = LemiData(self.path2data, OUTPUT_MSEED)
data = lemi_data.parse_file(lemi_data.data_files[0])
input_file = self.path2data.joinpath('test_data_file.pkl')
with open(input_file, 'rb') as fin:
expected_data = pickle.load(fin)
self.assertDictEqual(data, expected_data)
def test_parse_file_corrupted_no_CR(self):
"""Test basic functionality of parse_file."""
lemi_data = LemiData(self.path2data_corrupted_1, OUTPUT_MSEED)
data_file = lemi_data.data_files[0]
with self.assertLogs(logger, level='WARNING') as cmd:
data = lemi_data.parse_file(data_file)
msg = ("The data file - {} - may have been corrupted. Skipping file!"
.format(data_file))
self.assertEqual(cmd.output, [":".join(['WARNING', SCR_DIR, msg])])
self.assertIsNone(data)
def test_parse_file_corrupted_missing_column(self):
"""Test basic functionality of parse_file."""
lemi_data = LemiData(self.path2data_corrupted_1, OUTPUT_MSEED)
data_file = lemi_data.data_files[1]
with self.assertLogs(logger, level='WARNING') as cmd:
data = lemi_data.parse_file(data_file)
msg = ("The data file - {} - may have been corrupted. Skipping file!"
.format(data_file))
self.assertEqual(cmd.output, [":".join(['WARNING', SCR_DIR, msg])])
self.assertIsNone(data)
def test_parse_file_corrupted_erroneous_time_formatting(self):
"""Test basic functionality of parse_file."""
lemi_data = LemiData(self.path2data_corrupted_2, OUTPUT_MSEED)
data_file = lemi_data.data_files[0]
with self.assertLogs(logger) as cmd:
data = lemi_data.parse_file(data_file)
msg = ("The data file - {} - may have been corrupted. Skipping file!"
.format(data_file))
self.assertEqual(cmd.output, [":".join(['WARNING', SCR_DIR, msg])])
self.assertIsNone(data)
def test_parse_file_corrupted_erroneous_coordinate_formatting(self):
"""Test basic functionality of parse_file."""
lemi_data = LemiData(self.path2data_corrupted_2, OUTPUT_MSEED)
data_file = lemi_data.data_files[1]
with self.assertLogs(logger) as cmd:
data = lemi_data.parse_file(data_file)
msg = ("The data file - {} - may have been corrupted. Skipping file!"
.format(data_file))
self.assertEqual(cmd.output, [":".join(['WARNING', SCR_DIR, msg])])
self.assertIsNone(data)
def test_parse_all_files(self):
"""Test basic functionality of parse_all_files."""
lemi_data = LemiData(self.path2data, OUTPUT_MSEED)
lemi_data.parse_all_files()
input_file = self.path2data.joinpath('test_data_files.pkl')
with open(input_file, 'rb') as fin:
expected_data = pickle.load(fin)
self.assertDictEqual(lemi_data.data, expected_data)
def test_order_files(self):
"""Test basic functionality of order_files."""
files = self.path2data.glob('*.TXT')
ordered_data_files = LemiData.order_files(files)
self.assertListEqual(ordered_data_files, self.data_files)
def test_reformat_data_valid_formatting(self):
"""Test basic functionality of reformat_data."""
output = {'Time_stamp': UTCDateTime('2020-09-30T21:05:00.000000Z'),
'Lat': 34.0483955, 'Lon': -107.12843720000001,
'Elev': 2202.2, 'Hx': 23777.803, 'Hy': 210.731,
'Hz': 41834.585, 'E1': 136.521, 'E2': -110.769,
'E3': 174.046, 'E4': 16.178, 'Ui': 13.01, 'Te': 46.75,
'Tf': 37.09, 'Sn': 12.0, 'Fq': 2.0, 'Ce': 0.0}
line = ('2020 09 30 21 05 00 23777.803 210.731 41834.585 46.75 '
'37.09 136.521 -110.769 174.046 16.178 13.01 2202.2 '
'3404.83955 N 10712.84372 W 12 2 0')
columns = line.strip().split()
self.assertDictEqual(LemiData.reformat_data(columns), output)
def test_reformat_data_erroneous_time_formatting(self):
"""Test basic functionality of reformat_data."""
line = ('2020 09 30 21 05 23777.803 23777.803 210.731 41834.585 '
'46.75 37.09 136.521 -110.769 174.046 16.178 13.01 '
'2202.2 3404.83955 N 10712.84372 W 12 2 0')
columns = line.strip().split()
self.assertIsNone(LemiData.reformat_data(columns))
def test_reformat_data_erroneous_coordinate_formatting(self):
"""Test basic functionality of reformat_data."""
line = ('2020 09 30 21 12 00 23775.684 212.121 41834.749 46.68 '
'36.46 133.427 -111.936 171.563 15.734 13.00 2200.8 '
'3404.83994 Z 10712.84418 W 12 2 0')
columns = line.strip().split()
self.assertIsNone(LemiData.reformat_data(columns))
def test_detect_gaps(self):
"""Test basic functionality of detect_gaps."""
lemi_data = LemiData(self.path2data, OUTPUT_MSEED)
lemi_data.parse_all_files()
time_stamp = lemi_data.data["Time_stamp"]
ind_gaps = LemiData.detect_gaps(time_stamp)
self.assertEqual(len(ind_gaps), 3)
self.assertListEqual([time_stamp[ind] for ind in ind_gaps],
[UTCDateTime('2020-09-30T21:12:00.000000Z'),
UTCDateTime('2020-09-30T21:14:00.000000Z'),
UTCDateTime('2020-10-01T00:00:00.000000Z')])
def test_detect_new_day(self):
"""Test basic functionality of detect_new_day."""
lemi_data = LemiData(self.path2data, OUTPUT_MSEED)
lemi_data.parse_all_files()
time_stamp = lemi_data.data["Time_stamp"]
ind_day = LemiData.detect_new_day(time_stamp)
self.assertEqual(len(ind_day), 1)
self.assertEqual(time_stamp[ind_day[0]],
UTCDateTime('2020-10-01T00:00:00.000000Z'))
def test_create_data_array(self):
"""Test basic functionality of create_data_array."""
lemi_data = LemiData(self.path2data, OUTPUT_MSEED)
lemi_data.parse_all_files()
lemi_data.create_data_array()
input_file = self.path2data.joinpath('test_data_np_files.pkl')
with open(input_file, 'rb') as fin:
expected_data_np = pickle.load(fin)
self.assertEqual(lemi_data.stats['nbr_runs'], 4)
self.assertSequenceEqual(lemi_data.data_np[0], expected_data_np)
def test_update_stats_1(self):
"""Test basic functionality of update_stats_1."""
lemi_data = LemiData(self.path2data, OUTPUT_MSEED)
lemi_data.parse_all_files()
lemi_data.update_stats_1()
self.assertAlmostEqual(lemi_data.stats['latitude'], 34.04839, 5)
self.assertAlmostEqual(lemi_data.stats['longitude'], -107.12844, 5)
self.assertAlmostEqual(lemi_data.stats['elevation'], 2201.77122, 5)
self.assertEqual(lemi_data.stats['time_period_start'],
UTCDateTime('2020-09-30T21:05:00.000000Z'))
self.assertEqual(lemi_data.stats['time_period_end'],
UTCDateTime('2020-10-01T00:05:59.000000Z'))
def test_update_array_list_of_channels(self):
"""Test basic functionality of update_array."""
lemi_data = LemiData(self.path2data, OUTPUT_MSEED)
lemi_data.parse_all_files()
lemi_data.create_data_array()
data_channels = ['E1', 'E2', 'Hx', 'Hy', 'Hz']
e_info = {'E1': 'Ex', 'E2': 'Ey'}
lemi_data.update_array(data_channels, e_info)
self.assertNotIn('E3', lemi_data.data_np['channel_number'])
self.assertNotIn('E4', lemi_data.data_np['channel_number'])
def test_update_array_electrode_info(self):
"""Test basic functionality of update_array."""
lemi_data = LemiData(self.path2data, OUTPUT_MSEED)
lemi_data.parse_all_files()
lemi_data.create_data_array()
data_channels = ['E1', 'E2', 'E3', 'E4', 'Hx', 'Hy', 'Hz']
e_info = {'E1': 'Ex', 'E2': 'Ey', 'E3': 'Ex', 'E4': 'Ey'}
lemi_data.update_array(data_channels, e_info)
data_np_E1 = lemi_data.data_np[lemi_data.data_np['channel_number'] == 'E1']
self.assertSetEqual(set(data_np_E1['component']), {'Ex'})
self.assertSetEqual(set(data_np_E1['channel_name']), {'LQN'})
self.assertSetEqual(set(data_np_E1['location']), {'00'})
data_np_E2 = lemi_data.data_np[lemi_data.data_np['channel_number'] == 'E2']
self.assertSetEqual(set(data_np_E2['component']), {'Ey'})
self.assertSetEqual(set(data_np_E2['channel_name']), {'LQE'})
self.assertSetEqual(set(data_np_E2['location']), {'00'})
data_np_E3 = lemi_data.data_np[lemi_data.data_np['channel_number'] == 'E3']
self.assertSetEqual(set(data_np_E3['component']), {'Ex'})
self.assertSetEqual(set(data_np_E3['channel_name']), {'LQN'})
self.assertSetEqual(set(data_np_E3['location']), {'01'})
data_np_E4 = lemi_data.data_np[lemi_data.data_np['channel_number'] == 'E4']
self.assertSetEqual(set(data_np_E4['component']), {'Ey'})
self.assertSetEqual(set(data_np_E4['channel_name']), {'LQE'})
self.assertSetEqual(set(data_np_E4['location']), {'01'})
def test_prep_data(self):
"""Test basic functionality if prep_data."""
lemi_data = LemiData(self.path2data_corrupted_1, OUTPUT_MSEED)
with self.assertRaises(SystemExit) as cmd:
lemi_data.prep_data()
self.assertEqual(cmd.exception.code, 1)