# -*- coding: utf-8 -*-
"""
GSE2/GSE1 bindings to ObsPy core module.
"""
import numpy as np
from obspy import Stream, Trace
from . import libgse1, libgse2
from obspy.core.util import open_bytes_stream
[docs]
def _is_gse2(file):
"""
Checks whether a file is GSE2 or not.
:type file: str or file-like object
:param file: GSE2 file to be checked.
:rtype: bool
:return: ``True`` if a GSE2 file.
"""
with open_bytes_stream(file) as io_stream:
try:
libgse2.is_gse2(io_stream)
except Exception:
return False
return True
[docs]
def _read_gse2(file, headonly=False, verify_chksum=True,
**kwargs): # @UnusedVariable
"""
Reads a GSE2 file and returns a Stream object.
GSE2 files containing multiple WID2 entries/traces are supported.
.. warning::
This function should NOT be called directly, it registers via the
ObsPy :func:`~obspy.core.stream.read` function, call this instead.
:type file: str or file-like object
:param file: GSE2 file to be read.
:type headonly: bool, optional
:param headonly: If True read only head of GSE2 file.
:type verify_chksum: bool, optional
:param verify_chksum: If True verify Checksum and raise Exception if
it is not correct.
:rtype: :class:`~obspy.core.stream.Stream`
:returns: Stream object containing header and data.
.. rubric:: Example
>>> from obspy import read
>>> st = read("/path/to/loc_RJOB20050831023349.z")
"""
traces = []
with open_bytes_stream(file) as f:
# reading multiple gse2 parts
while True:
try:
if headonly:
header = libgse2.read_header(f)
traces.append(Trace(header=header))
else:
header, data = libgse2.read(f, verify_chksum=verify_chksum)
traces.append(Trace(header=header, data=data))
except EOFError:
break
return Stream(traces=traces)
[docs]
def _write_gse2(stream, filename, inplace=False, **kwargs): # @UnusedVariable
"""
Write GSE2 file from a Stream object.
.. warning::
This function should NOT be called directly, it registers via the
the :meth:`~obspy.core.stream.Stream.write` method of an
ObsPy :class:`~obspy.core.stream.Stream` object, call this instead.
:type stream: :class:`~obspy.core.stream.Stream`
:param stream: The ObsPy Stream object to write.
:type filename: str
:param filename: Name of file to write.
:type inplace: bool, optional
:param inplace: If True, do compression not on a copy of the data but
on the data itself - note this will change the data values and make
them therefore unusable!
.. rubric:: Example
>>> from obspy import read
>>> st = read()
>>> st.write('filename.gse', format='GSE2') #doctest: +SKIP
"""
#
# Translate the common (renamed) entries
with open(filename, 'wb') as f:
# write multiple gse2 parts
for trace in stream:
dt = np.dtype(np.int32)
if trace.data.dtype.name == dt.name:
trace.data = np.ascontiguousarray(trace.data, dt)
else:
msg = "GSE2 data must be of type %s, but are of type %s" % \
(dt.name, trace.data.dtype)
raise Exception(msg)
libgse2.write(trace.stats, trace.data, f, inplace)
[docs]
def _is_gse1(file):
"""
Checks whether a file is GSE1 or not.
:type file: str or file-like object
:param file: GSE1 file to be checked.
:rtype: bool
:return: ``True`` if a GSE1 file.
"""
with open_bytes_stream(file) as io_stream:
try:
data = io_stream.readline()
except Exception:
return False
return data.startswith(b'WID1') or data.startswith(b'XW01')
[docs]
def _read_gse1(file, headonly=False, verify_chksum=True,
**kwargs): # @UnusedVariable
"""
Reads a GSE1 file and returns a Stream object.
GSE1 files containing multiple WID1 entries/traces are supported.
.. warning::
This function should NOT be called directly, it registers via the
ObsPy :func:`~obspy.core.stream.read` function, call this instead.
:type file: str or file-like object
:param file: GSE2 file to be read.
:type headonly: bool, optional
:param headonly: If True read only header of GSE1 file.
:type verify_chksum: bool, optional
:param verify_chksum: If True verify Checksum and raise Exception if
it is not correct.
:rtype: :class:`~obspy.core.stream.Stream`
:returns: Stream object containing header and data.
.. rubric:: Example
>>> from obspy import read
>>> st = read("/path/to/y2000.gse")
"""
traces = []
# read GSE1 file
with open_bytes_stream(file) as fh:
while True:
try:
if headonly:
header = libgse1.read_header(fh)
traces.append(Trace(header=header))
else:
header, data = \
libgse1.read(fh, verify_chksum=verify_chksum)
traces.append(Trace(header=header, data=data))
except EOFError:
break
return Stream(traces=traces)