/
test_client.py
118 lines (105 loc) · 4.49 KB
/
test_client.py
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
# -*- coding: utf-8 -*-
"""
The obspy.clients.neic.client test suite.
"""
import unittest
import pytest
from obspy.core.utcdatetime import UTCDateTime
from obspy.clients.neic import Client
pytestmark = pytest.mark.network
class ClientTestCase(unittest.TestCase):
"""
Test cases for obspy.clients.neic.client.Client.
"""
@classmethod
def setUpClass(cls):
cls.client = Client(host="137.227.224.97", port=2061, timeout=8)
cls.starttime = UTCDateTime() - 5 * 60 * 60
def test_get_waveform(self):
"""
Tests get_waveforms method. Tests against get_waveforms_nscl method.
"""
client = self.client
t = self.starttime
duration = 1.0
st = client.get_waveforms_nscl("IUANMO BH.00", t, duration)
# try a series of requests, compare against get_waveforms_nscl
args = [["IU", "ANMO", "00", "BH."],
["??", "ANMO", "0?", "BH[Z21]"],
["IU", "ANM.*", "00", "B??"],
["IU", "ANMO", "0*", "BH."],
]
for args_ in args:
st2 = client.get_waveforms(*args_, starttime=t,
endtime=t + duration)
self.assertEqual(st, st2)
def test_get_waveform_nscl(self):
"""
Tests get_waveforms_nscl method.
"""
client = self.client
t = self.starttime
duration_long = 3600.0
duration = 1.0
components = ["1", "2", "Z"]
# try one longer request to see if fetching multiple blocks works
st = client.get_waveforms_nscl("IUANMO BH.00", t, duration_long)
# merge to avoid failing tests simply due to gaps
st.merge()
st.sort()
self.assertEqual(len(st), 3)
for tr, component in zip(st, components):
stats = tr.stats
self.assertEqual(stats.station, "ANMO")
self.assertEqual(stats.network, "IU")
self.assertEqual(stats.location, "00")
self.assertEqual(stats.channel, "BH" + component)
# requested data duration has some minor fluctuations sometimes but
# should be pretty close to the expected duration.
# it should not be over a delta longer than expected (it should be
# trimmed correctly if more data is returned) but sometimes it's
# one delta shorter
self.assertTrue(
abs(duration_long - (stats.endtime - stats.starttime)) <=
tr.stats.delta)
# if the following fails this is likely due to a change at the
# requested station and simply has to be adapted
self.assertEqual(stats.sampling_rate, 40.0)
self.assertEqual(len(tr), 144001)
# now use shorter piece, this is faster and less error prone (gaps etc)
st = client.get_waveforms_nscl("IUANMO BH.00", t, duration)
st.sort()
# test returned stream
self.assertEqual(len(st), 3)
for tr, component in zip(st, components):
stats = tr.stats
self.assertEqual(stats.station, "ANMO")
self.assertEqual(stats.network, "IU")
self.assertEqual(stats.location, "00")
self.assertEqual(stats.channel, "BH" + component)
# requested data duration has some minor fluctuations sometimes but
# should be pretty close to the expected duration.
# it should not be over a delta longer than expected (it should be
# trimmed correctly if more data is returned) but sometimes it's
# one delta shorter
self.assertTrue(
abs(duration - (stats.endtime - stats.starttime)) <=
tr.stats.delta)
# if the following fails this is likely due to a change at the
# requested station and simply has to be adapted
self.assertEqual(stats.sampling_rate, 40.0)
self.assertEqual(len(tr), 41)
# try a series of regex patterns that should return the same data
st = client.get_waveforms_nscl("IUANMO BH", t, duration)
patterns = ["IUANMO BH...",
"IUANMO BH.*",
"IUANMO BH[Z12].*",
"IUANMO BH[Z12]..",
"..ANMO BH.*"]
for pattern in patterns:
st2 = client.get_waveforms_nscl(pattern, t, duration)
self.assertEqual(st, st2)
def suite():
return unittest.makeSuite(ClientTestCase, 'test')
if __name__ == '__main__':
unittest.main(defaultTest='suite')