mth5.clients

Submodules

Classes

FDSN

USGSGeomag

PhoenixClient

ZenClient

LEMI424Client

MetronixClient

NIMSClient

Package Contents

class mth5.clients.FDSN(client: str = 'IRIS', **kwargs)[source]

Bases: mth5.clients.base.ClientBase

logger
request_columns = ['network', 'station', 'location', 'channel', 'start', 'end']
client = 'IRIS'
property run_list_ne_stream_intervals_message: str

note about not equal stream intervals

get_run_list_from_station_id(m: mth5.mth5.MTH5, station_id: str, survey_id: str | None = None) → list[str][source]

ignored_groups created to address issue #153. This might be better placed closer to the core of mth5.

Parameters:

  • m

  • station_id

Returns:

run_list

Return type:

list of strings

stream_boundaries(streams: obspy.Stream) → tuple[list[obspy.UTCDateTime], list[obspy.UTCDateTime]][source]

Identify start and end times of streams

Parameters:

streams (obspy.core.stream.Stream)

get_station_streams(station_id: str) → obspy.Stream[source]

Get streams for a certain station

get_run_group(mth5_obj_or_survey, station_id: str, run_id: str)[source]

This method is key to merging wrangle_runs_into_containers_v1 and wrangle_runs_into_containers_v2. Because a v1 mth5 object can get a survey group with the same method as can a v2 survey_group

Thus we can replace run_group = m.stations_group.get_station(station_id).add_run(run_id) & run_group = survey_group.stations_group.get_station(station_id).add_run(run_id) with run_group = mth5_obj_or_survey.stations_group.get_station(station_id).add_run(run_id) :param mth5_obj_or_survey: :type mth5_obj_or_survey: mth5.mth5.MTH5 or mth5.groups.survey.SurveyGroup

pack_stream_into_run_group(run_group, run_stream: obspy.Stream)[source]
run_timings_match_stream_timing(run_group, stream_start: obspy.UTCDateTime, stream_end: obspy.UTCDateTime) → bool[source]

Checks start and end times in the run. Compares start and end times of runs to start and end times of traces. If True, will packs runs based on time spans.

Parameters:
Return type:

bool

wrangle_runs_into_containers(m: mth5.mth5.MTH5, station_id: str, survey_group=None) → None[source]

Note 1: There used to be two separate functions for this, but now there is one run_group_source is defined as either m or survey_group depending on v0.1.0 or 0.2.0

Note 2: If/elif/elif/else Logic: The strategy is to add the group first. This will get the already filled in metadata to update the run_ts_obj. Then get streams an add existing metadata.

Parameters:

  • m

  • streams

  • station_id

  • survey_group

make_mth5_from_fdsn_client(df: pandas.DataFrame | str | pathlib.Path, path: str | pathlib.Path | None = None, client: str | None = None, interact: bool = False) → pathlib.Path[source]

Create an MTH5 file from an FDSN data center request.

Parameters:

  • df (pandas.DataFrame or str or Path) –

    DataFrame or path to CSV with columns:

    • ’network’ : FDSN Network code

    • ’station’ : FDSN Station code

    • ’location’ : FDSN Location code

    • ’channel’ : FDSN Channel code

    • ’start’ : Start time YYYY-MM-DDThh:mm:ss

    • ’end’ : End time YYYY-MM-DDThh:mm:ss

  • path (str or Path, optional) – Path to save MTH5 file (default: current directory).

  • client (str, optional) – FDSN client name (default: “IRIS”).

  • interact (bool, optional) – Deprecated. If True, logs a warning (default: False).

Returns:

file_name – Path to the created MTH5 file.

Return type:

Path

Raises:

  • AttributeError – If the input DataFrame is not properly formatted.

  • ValueError – If the values of the DataFrame are not correct.

Examples

>>> from mth5.clients.fdsn import FDSN
>>> import pandas as pd
>>> df = pd.DataFrame({
...     'network': ['XX'],
...     'station': ['1234'],
...     'location': [''],
...     'channel': ['LHZ'],
...     'start': ['2022-01-01T00:00:00'],
...     'end': ['2022-01-02T00:00:00']
... })
>>> client = FDSN()
>>> file_path = client.make_mth5_from_fdsn_client(df)
property streams

obspy.Stream object

make_mth5_from_inventory_and_streams(inventory: obspy.Inventory | str | pathlib.Path, streams: obspy.Stream | list[str | pathlib.Path], save_path: str | pathlib.Path | None = None) → pathlib.Path[source]

Create an MTH5 file from an ObsPy Inventory and waveform streams.

Parameters:

  • inventory (obspy.Inventory or str or Path) – ObsPy Inventory object or path to StationXML file.

  • streams (obspy.Stream or list of str or Path) – ObsPy Stream object or list of file paths to waveform data.

  • save_path (str or Path, optional) – Path to save MTH5 file (default: current directory).

Returns:

file_name – Path to the created MTH5 file.

Return type:

Path

Examples

>>> from mth5.clients.fdsn import FDSN
>>> inv = ... # ObsPy Inventory
>>> streams = ... # ObsPy Stream
>>> client = FDSN()
>>> file_path = client.make_mth5_from_inventory_and_streams(inv, streams)
build_network_dict(df: pandas.DataFrame, client: obspy.clients.fdsn.Client) → dict[source]

Build a dictionary of networks keyed by network_id and start_time.

Parameters:

  • df (pandas.DataFrame) – Request DataFrame.

  • client (obspy.clients.fdsn.Client) – FDSN client instance.

Returns:

networks – Dictionary of networks.

Return type:

dict

Examples

>>> networks = client.build_network_dict(df, client)
build_station_dict(df: pandas.DataFrame, client: obspy.clients.fdsn.Client, networks_dict: dict) → dict[source]

Build a dictionary of stations keyed by network_id and start_time.

Parameters:

  • df (pandas.DataFrame) – Request DataFrame.

  • client (obspy.clients.fdsn.Client) – FDSN client instance.

  • networks_dict (dict) – Dictionary of networks.

Returns:

stations – Dictionary of stations.

Return type:

dict

Examples

>>> stations = client.build_station_dict(df, client, networks_dict)
get_waveforms_from_request_row(client: obspy.clients.fdsn.Client, row) → obspy.Stream[source]

Retrieve waveform data for a request row.

Parameters:

  • client (obspy.clients.fdsn.Client) – FDSN client instance.

  • row (pandas.Series) – Row of request DataFrame.

Returns:

streams – ObsPy Stream object with waveform data.

Return type:

obspy.Stream

Examples

>>> streams = client.get_waveforms_from_request_row(client, row)
get_inventory_from_df(df: pandas.DataFrame | str | pathlib.Path, client: str | None = None, data: bool = True, max_tries: int = 10) → tuple[obspy.Inventory, obspy.Stream][source]

Get an ObsPy Inventory and Stream from a DataFrame request.

Parameters:

  • df (pandas.DataFrame or str or Path) –

    DataFrame or path to CSV with columns:

    • ’network’ : FDSN Network code

    • ’station’ : FDSN Station code

    • ’location’ : FDSN Location code

    • ’channel’ : FDSN Channel code

    • ’start’ : Start time YYYY-MM-DDThh:mm:ss

    • ’end’ : End time YYYY-MM-DDThh:mm:ss

  • client (str, optional) – FDSN client name (default: self.client).

  • data (bool, optional) – If True, retrieves waveform data (default: True).

  • max_tries (int, optional) – Maximum number of retry attempts (default: 10).

Returns:

  • inventory (obspy.Inventory) – Inventory of metadata requested.

  • streams (obspy.Stream) – Stream of waveform data.

Examples

>>> from mth5.clients.fdsn import FDSN
>>> import pandas as pd
>>> df = pd.DataFrame({
...     'network': ['XX'],
...     'station': ['1234'],
...     'location': [''],
...     'channel': ['LHZ'],
...     'start': ['2022-01-01T00:00:00'],
...     'end': ['2022-01-02T00:00:00']
... })
>>> client = FDSN()
>>> inv, streams = client.get_inventory_from_df(df)
get_df_from_inventory(inventory: obspy.Inventory) → pandas.DataFrame[source]

Create a DataFrame from an ObsPy Inventory object.

Parameters:

inventory (obspy.Inventory) – ObsPy Inventory object.

Returns:

df – DataFrame in request format.

Return type:

pandas.DataFrame

Examples

>>> df = client.get_df_from_inventory(inventory)
get_unique_networks_and_stations(df: pandas.DataFrame) → list[dict][source]

Get unique networks and stations from a request DataFrame.

Parameters:

df (pandas.DataFrame) – Request DataFrame.

Returns:

unique_list – List of network dictionaries with stations.

Return type:

list of dict

Examples

>>> unique_list = client.get_unique_networks_and_stations(df)
make_filename(df: pandas.DataFrame) → str[source]

Make a filename from a request DataFrame of networks and stations.

Parameters:

df (pandas.DataFrame) – Request DataFrame.

Returns:

filename – Filename in the format network_01+stations_network_02+stations.h5

Return type:

str

Examples

>>> filename = client.make_filename(df)
get_fdsn_channel_map() → dict[str, str][source]

Get mapping of FDSN channel codes to internal codes.

Returns:

FDSN_CHANNEL_MAP – Dictionary mapping FDSN channel codes.

Return type:

dict

Examples

>>> channel_map = client.get_fdsn_channel_map()
class mth5.clients.USGSGeomag(**kwargs)[source]
save_path
mth5_filename = None
interact = False
request_columns = ['observatory', 'type', 'elements', 'sampling_period', 'start', 'end']
h5_compression = 'gzip'
h5_compression_opts = 4
h5_shuffle = True
h5_fletcher32 = True
h5_data_level = 1
mth5_file_mode = 'w'
mth5_version = '0.2.0'
property h5_kwargs
validate_request_df(request_df)[source]

Make sure the input request dataframe has the appropriate columns

Parameters:

request_df (pandas.DataFrame) – request dataframe

Returns:

valid request dataframe

Return type:

pandas.DataFrame

add_run_id(request_df)[source]

Add run id to request df

Parameters:

request_df (pandas.DataFrame) – request dataframe

Returns:

add a run number to unique time windows for each observatory at each unique sampling period.

Return type:

pandas.DataFrame

make_mth5_from_geomag(request_df)[source]

Download geomagnetic observatory data from USGS webservices into an MTH5 using a request dataframe or csv file.

Parameters:

request_df (pandas.DataFrame, str or Path if csv file) –

DataFrame with columns

  • ’observatory’ –> Observatory code

  • ’type’ –> data type [ ‘variation’ | ‘adjusted’ | ‘quasi-definitive’ | ‘definitive’ ]

  • ’elements’ –> Elements to get [D, DIST, DST, E, E-E, E-N, F, G, H, SQ, SV, UK1, UK2, UK3, UK4, X, Y, Z]

  • ’sampling_period’ –> sample period [ 1 | 60 | 3600 ]

  • ’start’ –> Start time YYYY-MM-DDThh:mm:ss

  • ’end’ –> End time YYYY-MM-DDThh:mm:ss

Returns:

if interact is True an MTH5 object is returned otherwise the path to the file is returned

Return type:

Path or mth5.mth5.MTH5

See also

https://www.usgs.gov/tools/web-service-geomagnetism-data

class mth5.clients.PhoenixClient(data_path: str | pathlib.Path, sample_rates: list[int] = [150, 24000], save_path: str | pathlib.Path | None = None, receiver_calibration_dict: dict | str | pathlib.Path = {}, sensor_calibration_dict: dict | str | pathlib.Path = {}, mth5_filename: str = 'from_phoenix.h5', **kwargs: dict)[source]

Bases: mth5.clients.base.ClientBase

property receiver_calibration_dict: dict

Receiver calibrations.

Returns:

Dictionary mapping receiver IDs to calibration file paths.

Return type:

dict

Examples

>>> client = PhoenixClient('data/path')
>>> client.receiver_calibration_dict = {'RX001': Path('RX001_rxcal.json')}
>>> client.receiver_calibration_dict
{'RX001': Path('RX001_rxcal.json')}
property sensor_calibration_dict: dict

Sensor calibration dictionary.

Returns:

Dictionary mapping sensor IDs to PhoenixCalibration objects.

Return type:

dict

Examples

>>> client = PhoenixClient('data/path')
>>> client.sensor_calibration_dict = {'H001': PhoenixCalibration('H001_scal.json')}
>>> client.sensor_calibration_dict['H001']
<PhoenixCalibration object>
collection
make_mth5_from_phoenix(**kwargs: dict) → str | pathlib.Path | None[source]

Make an MTH5 from Phoenix files.

Split into runs, account for filters. Updates the MTH5 file with Phoenix data.

Parameters:

**kwargs (dict) – Optional keyword arguments to override instance attributes.

Returns:

Path to the saved MTH5 file.

Return type:

str, Path, or None

Examples

>>> client = PhoenixClient('data/path', save_path='output.h5')
>>> client.make_mth5_from_phoenix()
'output.h5'
class mth5.clients.ZenClient(data_path, sample_rates=[4096, 1024, 256], save_path=None, calibration_path=None, mth5_filename='from_zen.h5', **kwargs)[source]

Bases: mth5.clients.base.ClientBase

property calibration_path

Path to calibration data

collection
station_stem = None
get_run_dict()[source]

Get Run information

Returns:

DESCRIPTION

Return type:

TYPE

get_survey(station_dict)[source]

get survey name from a dictionary of a single station of runs :param station_dict: DESCRIPTION :type station_dict: TYPE :return: DESCRIPTION :rtype: TYPE

make_mth5_from_zen(survey_id=None, combine=True, **kwargs)[source]

Make an MTH5 from Phoenix files. Split into runs, account for filters

Parameters:

  • data_path (TYPE, optional) – DESCRIPTION, defaults to None

  • sample_rates (TYPE, optional) – DESCRIPTION, defaults to None

  • save_path (TYPE, optional) – DESCRIPTION, defaults to None

Returns:

DESCRIPTION

Return type:

TYPE

class mth5.clients.LEMI424Client(data_path: str | pathlib.Path, save_path: str | pathlib.Path | None = None, mth5_filename: str = 'from_lemi424.h5', **kwargs: Any)[source]

Bases: mth5.clients.base.ClientBase

collection
make_mth5_from_lemi424(survey_id: str, station_id: str, **kwargs: Any) → pathlib.Path[source]

Create an MTH5 file from LEMI 424 long period data.

Parameters:

  • survey_id (str) – Survey identifier.

  • station_id (str) – Station identifier.

  • **kwargs (Any) – Additional keyword arguments to set as attributes.

Returns:

Path to the created mth5 file.

Return type:

Path

Examples

>>> client = LEMI424Client(data_path="./data")
>>> client.make_mth5_from_lemi424("SURVEY1", "ST01")
PosixPath('data/from_lemi424.h5')
class mth5.clients.MetronixClient(data_path, sample_rates=[128], save_path=None, calibration_path=None, mth5_filename='from_metronix.h5', **kwargs)[source]

Bases: mth5.clients.base.ClientBase

calibration_path = None
collection
get_run_dict(run_name_zeros=0)[source]

get run information

Returns:

DESCRIPTION

Return type:

TYPE

get_survey_id(station_dict)[source]

get survey name from a dictionary of a single station of runs :param station_dict: DESCRIPTION :type station_dict: TYPE :return: DESCRIPTION :rtype: TYPE

set_station_metadata(station_dict, station_group)[source]

set station group metadata from information in the station dict

Parameters:

  • station_dict (TYPE) – DESCRIPTION

  • station_group (TYPE) – DESCRIPTION

Returns:

DESCRIPTION

Return type:

TYPE

make_mth5_from_metronix(run_name_zeros=0, **kwargs)[source]

Create an MTH5 from new ATSS + JSON style Metronix data.

Parameters:

**kwargs –

DESCRIPTION

Returns:

DESCRIPTION

Return type:

TYPE

class mth5.clients.NIMSClient(data_path: str | pathlib.Path, sample_rates: list[int] = [1, 8], save_path: str | pathlib.Path | None = None, calibration_path: str | pathlib.Path | None = None, mth5_filename: str = 'from_nims.h5', **kwargs)[source]

Bases: mth5.clients.base.ClientBase

property calibration_path: pathlib.Path | None

Path to calibration data.

Returns:

Path to calibration file, or None if not set.

Return type:

Path or None

Examples

>>> client = NIMSClient('data_dir')
>>> client.calibration_path = 'calib.dat'
>>> print(client.calibration_path)
PosixPath('calib.dat')
collection
get_run_dict() → dict[source]

Get run information from the NIMS collection.

Returns:

Dictionary of run information.

Return type:

dict

Examples

>>> client = NIMSClient('data_dir')
>>> runs = client.get_run_dict()
>>> print(list(runs.keys()))
['station1', 'station2']
get_survey(station_dict: dict) → str[source]

Get survey name from a dictionary of a single station of runs.

Parameters:

station_dict (dict) – Dictionary of runs for a station.

Returns:

Survey name.

Return type:

str

Examples

>>> client = NIMSClient('data_dir')
>>> runs = client.get_run_dict()
>>> survey = client.get_survey(runs['station1'])
>>> print(survey)
'survey_name'
make_mth5_from_nims(survey_id: str = 'default_survey', combine: bool = True, **kwargs) → str | pathlib.Path[source]

Make an MTH5 file from Phoenix NIMS files. Splits into runs, accounts for filters.

Parameters:

  • survey_id (str, optional) – Survey identifier. Default is “default_survey”.

  • combine (bool, optional) – Whether to combine runs. Default is True.

  • **kwargs – Additional keyword arguments to set as attributes.

Returns:

Path to the saved MTH5 file.

Return type:

str or Path

Examples

>>> client = NIMSClient('data_dir')
>>> mth5_path = client.make_mth5_from_nims(survey_id='survey1')
>>> print(mth5_path)
'output_dir/from_nims.h5'