Skip to content
Permalink
Browse files

flake8

  • Loading branch information...
megies committed May 15, 2019
1 parent 042d1f6 commit d05f2bfdaf36672a5fd61e0a2117adcc7f72f968
Showing with 21 additions and 18 deletions.
  1. +11 −6 obspy/io/mseed/core.py
  2. +10 −12 obspy/io/mseed/tests/test_mseed_reading_and_writing.py
@@ -498,9 +498,12 @@ def allocate_data(samplecount, sampletype):

def _np_copy_astype(data, dtype):
"""
Defines a helper function to copy data, replaces use of: trace.data.copy().astype(dtype)
This is done to avoid copying data in memory twice.
This helper workaround function can be dropped once bumping minimum numpy version to >=1.7.0
Helper function to copy data, replacing `trace.data.copy().astype(dtype)`
This is done to avoid copying data in memory twice that could happen due to
an API change in numpy's `astype` method (`copy` kwarg).
This helper workaround function can be dropped once bumping minimum numpy
version to >=1.7.0
"""
try:
return data.astype(dtype, copy=True)
@@ -797,16 +800,18 @@ def _write_mseed(stream, filename, encoding=None, reclen=None, byteorder=None,
trace_attr['encoding'] = 1
elif trace.data.dtype.type == np.dtype(native_str('|S1')).type:
trace_attr['encoding'] = 0
# int64 data not supported; if possible downcast to int32, else create error message
# After bumping up to numpy 1.9.0 this check can be replaced by numpy.can_cast()
# int64 data not supported; if possible downcast to int32, else
# create error message. After bumping up to numpy 1.9.0 this check
# can be replaced by numpy.can_cast()
elif trace.data.dtype.type == np.int64:
# check if data can be safely downcast to int32
ii32 = np.iinfo(np.int32)
if abs(trace.max()) <= ii32.max:
trace_data.append(_np_copy_astype(trace.data, np.int32))
trace_attr['encoding'] = 11
else:
msg = "int64 data only supported when writing MSEED if it can be downcast to int32 type data."
msg = ("int64 data only supported when writing MSEED if "
"it can be downcast to int32 type data.")
raise ObsPyMSEEDError(msg)
else:
msg = "Unsupported data type %s in Stream[%i].data" % \
@@ -155,14 +155,13 @@ def test_read_and_write_traces(self):
np.testing.assert_array_equal(this_stream[0].data,
new_stream[0].data)


def test_downcast_int64_to_int32(self):
"""
Checks that sample stream of dtype int64 can be downcast to int 32 and written to mseed
by downcasting the data to int32 type data.
Checks that sample stream of dtype int64 can be downcast to int 32 and
written to mseed by downcasting the data to int32 type data.
"""
# create a dummy stream with int64 data
x = np.array([1,2,-3,4], dtype = np.int64)
# create a dummy stream with int64 data
x = np.array([1, 2, -3, 4], dtype=np.int64)
tr = Trace(x)
st = Stream()
st.append(tr)
@@ -171,14 +170,15 @@ def test_downcast_int64_to_int32(self):
with io.BytesIO() as buf:
st.write(buf, format="mseed")
st2 = read(buf)
res = np.array_equal(st,st2)
self.assertEqual(st[0].data.dtype.type, np.int64)
self.assertTrue(res)
self.assertEqual(len(st), len(st2))
for tr, tr2 in zip(st, st2):
self.assertTrue(np.array_equal(tr.data, tr2.data))
self.assertEqual(st[0].data.dtype.type, np.int64)

# Test that error message is indeed raised when data cannot be downcast
# Create dummy stream that cannot be properly downcast to int64
x_int64 = np.array([1, 2, 3, 2**55], dtype=np.int64)
x_int64_negative = np.array([1, -2, 3, -2**55], dtype=np.int64)
x_int64 = np.array([1, 2, 3, 2 ** 55], dtype=np.int64)
x_int64_negative = np.array([1, -2, 3, -2 ** 55], dtype=np.int64)
tr_int64 = Trace(x_int64)
tr_int64_negative = Trace(x_int64_negative)
st_int64 = Stream()
@@ -189,10 +189,8 @@ def test_downcast_int64_to_int32(self):
with io.BytesIO() as buf:
with self.assertRaises(ObsPyMSEEDError):
st_int64.write(buf, format="mseed")
st_int64_2 = read(buf)
with self.assertRaises(ObsPyMSEEDError):
st_int64_negative.write(buf, format="mseed")
st_int64_negative_2 = read(buf)

def test_get_record_information(self):
"""

0 comments on commit d05f2bf

Please sign in to comment.
You can’t perform that action at this time.