import sys import os import warnings import obspy import argparse import unit import sacutil import numpy as np from obspy.io.xseed import Parser from obspy.io.xseed.utils import SEEDParserException from obspy.core.util import AttribDict from obspy.core import Stream def convert_seed(seed_file, inv_file=None, input_unit=None, output_filetype=None, remove_response=False, zero_mean=False, pre_filtering=None, output_unit=None, single_output_file=False, output_dir="."): """ Function to convert provided file to SAC, ASCII or MSEED :type seed_file: str :param seed_file: File name or path to the seed (seed or msd) file. :type inv_file: str :param inv_file: File name or path to inventory file in any inventory format supported by ObsPy. :type input_unit: str :param input_unit: Unit of the input file: 'ACC', 'DISP' or 'VEL'. Specified only for miniSEED files (full SEED files are always in 'COUNTS'). If the value is not set, no unit ('COUNTS') is assumed. :type output_filetype: str :param output_filetype: Output filetype. Choose either SAC, MSEED, SLIST or INTERNAL_ASCII. When used SLIST, sample values are saved with `%+.10e` formatting. INTERNAL_ASCII is an internal format used for some apps, that uses only one column of values This will be default formatting since ObsPy 1.1.0 :type remove_response: bool :param remove_response: Response removal from the waveform Defaults to False. Removal of the instrument's response from the waveform By default it will save waveform without response removal procedure This function uses obspy.core.stream.Stream.remove_response. When set to True, user has to specify parameters zero_mean, pre_filtering and output_unit. Otherwise the script will raise an exception. :type zero_mean: bool, optional :param zero_mean: If true the mean of the data is subtracted For details check the ObsPy's documentation for obspy.core.stream.Stream.remove_response :type pre_filtering: (float, float, float, float), optional :param pre_filtering: Apply a bandpass filter to the data trace before deconvolution. The list or tuple defines the four corner frequencies (f1,f2,f3,f4) of a cosine taper which is one between f2 and f3 and tapers to zero for f1 < f < f2 and f3 < f < f4. For details check the ObsPy's documentation for obspy.core.stream.Stream.remove_response :type output_unit: str, optional :param output_unit: Unit of the output waveform. This parameter converts counts to real values during response removal, or, if response_removal is false is just written to the converted file. When used ``auto`` the script will determine the unit automatically with use of second letter of a channel code which determines instrument type. For details see Appendix A to SEED Manual v2.4. In case user need to convert all channels to common unit, there can be chosen ``DISP`` for displacement, ``VEL`` for velocity and ``ACC`` for acceleration. When this parameter is None and remove_response is set to True script will throw an Exception. :type single_output_file: bool :param single_output_file: if True, a single file with all traces will be created (available only for conversion to MSEED), otherwise, separate files with names in format ``network.station.location.channel.starttime_microsecond.output_filetype`` will be created :type output_dir: str, optional :param output_dir: Directory to which output files are written. Defaults to current directory. :return: Filenames of created files. Filenames follow pattern: ``network.station.location.channel.starttime_microsecond.SAC`` :raises .. note:: For more information on parameters used in this function see ObsPy's documentation in ``obspy.core.stream.Stream.remove_response`` """ st = obspy.read(seed_file) if inv_file: inv = obspy.read_inventory(inv_file) elif remove_response: raise ValueError("Cannot remove response if inventory is not specified") else: inv = None if output_unit: if inv: # assuming all the traces have the same unit input_type = unit.determine_instrument_type_from_inventory(inv, st[0].id, st[0].stats.starttime) else: try: # if we don't have the inventory and we have a full SEED file, we can attempt to read unit from it resp = Parser(seed_file) # assuming all the traces have the same unit input_type = unit.determine_instrument_type_from_blockette(resp, st[0].id) except (OSError, SEEDParserException): warnings.warn("The provided file is not a SEED volume - cannot determine unit automatically") if output_unit == "auto": output_unit = input_type if remove_response and output_unit is None: raise ValueError("You have to provide output_unit parameter" " for response removal procedure.\n" "Available output_unit values are DISP, " "VEL, ACC. Provide one or skip removing " "response by passing " "remove_response=False parameter") if remove_response: if pre_filtering is None: tr_sampling_rate = st[0].stats.sampling_rate pre_filt = [0.005, 0.006, 0.9 * 0.5 * tr_sampling_rate, 0.5 * tr_sampling_rate] else: pre_filt = pre_filtering try: _remove_response(st, inv, input_type, output_unit, pre_filt, zero_mean) except (ValueError, SEEDParserException) as exc: raise Exception("There was a problem with removing response from the file: " + str(exc)) elif output_unit and input_unit and input_unit != output_unit: _convert_unit(st, input_unit, output_unit) # we are not converting anything, but the input file already had a unit (might happen only with mseed files), # so it should be written to the output file elif input_unit: output_unit = input_unit if single_output_file: if output_filetype != "MSEED": raise ValueError("Writing all traces to one file is available only for conversion to MSEED format") return _convert_to_single_mseed(seed_file, st, output_dir) else: return _convert_to_separate_files(st, inv, remove_response, output_unit, output_filetype, output_dir) # remove response and correct numerical errors (depending on the input and output type) def _remove_response(stream, inv, input_type, output_unit, pre_filt, zero_mean): if input_type == "ACC" and output_unit == "DISP": stream.remove_response(inventory=inv, pre_filt=pre_filt, zero_mean=zero_mean, output="VEL") stream.filter("highpass", freq=1.0) stream.integrate(method="cumtrapz") stream.filter("highpass", freq=0.5) else: stream.remove_response(inventory=inv, pre_filt=pre_filt, zero_mean=zero_mean, output=output_unit) if ((input_type == "ACC" and output_unit == "VEL") or (input_type == "VEL" and (output_unit == "DISP" or output_unit == "VEL"))): stream.filter("highpass", freq=1.0) # unit conversion - applied when the input data already has response removed and is converted to one of the units def _convert_unit(stream, input_unit, output_unit): if input_unit == "DISP": if output_unit == "ACC": # DIFF x2 stream.differentiate(method="gradient") stream.differentiate(method="gradient") elif output_unit == "VEL": # DIFF stream.differentiate(method="gradient") elif input_unit == "VEL": if output_unit == "ACC": # DIFF stream.differentiate(method="gradient") elif output_unit == "DISP": # INTEG + FILTER stream.integrate(method="cumtrapz") stream.filter("highpass", freq=1.0) elif input_unit == "ACC": if output_unit == "VEL": # INTEG + FILTER stream.integrate(method="cumtrapz") stream.filter("highpass", freq=1.0) elif output_unit == "DISP": # (INTEG + FILTER) x2 stream.integrate(method="cumtrapz") stream.filter("highpass", freq=1.0) stream.integrate(method="cumtrapz") stream.filter("highpass", freq=0.5) else: raise TypeError("Cannot convert from ", input_unit) def _convert_to_single_mseed(seed_filename, stream, output_dir): result_filename = os.path.splitext(os.path.basename(seed_filename))[0] + ".msd" stream.write(output_dir + "/" + result_filename, format="MSEED") return [result_filename] def _convert_to_separate_files(stream, inv, remove_response, output_unit, output_filetype, output_dir): output_stream = Stream() output_file_extension = output_filetype for tr in stream: if output_filetype == "SAC": output_file_extension = "sac" tr = sacutil.to_sac_trace(tr, output_unit, remove_response, inv) elif output_filetype in ["SLIST", "TSPAIR", "SH_ASC"]: output_file_extension = output_filetype.lower() + ".ascii" slist_unit = unit.translate_instrument_type_to_unit(output_unit) if output_unit else "COUNTS" tr.stats.ascii = AttribDict({"unit": slist_unit}) elif output_filetype == "MSEED": output_file_extension = "msd" elif output_filetype == "INTERNAL_ASCII": output_file_extension = "ascii" output_stream.append(tr) return _write_separate_output_files(output_stream, output_filetype, output_file_extension, output_dir) def _write_separate_output_files(stream, output_filetype, file_extension, output_dir): """ Writes all Trace objects present in Stream to separate files. Filetype is set according to User's will. :type stream: obspy.core.stream.Stream :param stream: Obspy stream with traces to save separately. :type output_filetype: str :param output_filetype: Contains filetype to save. Currently supported ``SAC``, ``SLIST`` and ``MSEED``. """ result_filenames = list() for tr in stream: result_filename = ".".join([tr.id, str(tr.stats.starttime.microsecond), file_extension]) result_filepath = output_dir + "/" + result_filename if output_filetype == "INTERNAL_ASCII": _write_ascii_one_column(tr, result_filepath) else: tr.write(result_filepath, format=output_filetype) result_filenames.append(result_filename) return result_filenames def _write_ascii_one_column(trace, filename): """ Writes trace data into an ASCII file in one-column format (i.e. no header, single column of values). """ with open(filename, 'wb') as fh: data = trace.data.reshape((-1, 1)) dtype = data.dtype.name fmt = '%f' if dtype.startswith('int'): fmt = '%d' elif dtype.startswith('float64'): fmt = '%+.16e' np.savetxt(fh, data, delimiter=b'\t', fmt=fmt.encode('ascii', 'strict')) def main(argv): def str2bool(v): if v.lower() in ("True", "TRUE", "yes", "true", "t", "y", "1"): return True elif v.lower() in ("False", "FALSE", "no", "false", "f", "n", "0"): return False else: raise argparse.ArgumentTypeError("Boolean value expected.") parser = argparse.ArgumentParser(description="Convert provided file" " to SAC or SLIST (ASCII).") parser.add_argument("seed_file", help="Provide SEED or mSEED file to convert") parser.add_argument("--inv_file", help="Provide inventory file") parser.add_argument("--input_unit", help="Provide input unit. " "ACC, VEL or DISP are available.", type=str, default=None, required=False) parser.add_argument("--output_filetype", help="Provide output filetype. " "MSEED, SAC, SLIST and INTERNAL_ASCII are available.", type=str, default=None, required=True) parser.add_argument("--remove_response", help="Remove instrument's response from the waveform", type=str2bool, default=False) parser.add_argument("--zero_mean", help="If true the mean of the data is subtracted", type=str2bool, default=False, required=False) parser.add_argument("--pre_filtering", help="Apply a bandpass filter to the data trace " "before deconvolution", type=float, nargs="+", default=None, required=False) parser.add_argument("--output_unit", help="Unit to which waveform should be converted " "during response removal. When set to auto, " "the unit will be automatically determined with " "use of the second letter of channel code which " "determines the instrument type." "For details see Appendix A, SEED Manual v2.4.", type=str, default=None, required=False) parser.add_argument("--single_output_file", help="Flag deciding whether all the traces will be written to " "a single file, if set to False, each trace will be " "written to a separate file. True is available only " "for conversion to MSEED", type=str2bool, default=False, required=False) parser.add_argument("--output_dir", help="Directory to which output files are written. " "Defaults to current directory.", type=str, default=".", required=False) args = parser.parse_args() filenames = convert_seed(**vars(args)) print('Created files:') print(', '.join(filenames)) return if __name__ == "__main__": main(sys.argv)