mth5.utils package

Submodules

mth5.utils.exceptions module

Exceptions raised by MTH5

Created on Wed May 13 19:07:21 2020

@author: jpeacock

exception mth5.utils.exceptions.MTH5Error

Bases: Exception

exception mth5.utils.exceptions.MTH5TableError

Bases: Exception

exception mth5.utils.exceptions.MTSchemaError

Bases: Exception

exception mth5.utils.exceptions.MTTSError

Bases: Exception

exception mth5.utils.exceptions.MTTimeError

Bases: Exception

mth5.utils.extract_subset_mth5 module

mth5.utils.extract_subset_mth5.extract_subset(source_file: Path, target_file: Path, subset_df: DataFrame, filters: str = 'all')

This function is a proof-of-concept of issue 219: exporting a subset

TODO: add check that subset_df is a subset of source_file TODO: add tests for source/target v0.1.0 TODO: add tests for source/target v0.2.0 TODO: Consider add tests for source v0.1.0/target v0.2.0 TODO: Consider add tests for source v0.2.0/target v0.1.0

Parameters:

• source_file – Where the data will be extracted from

• target_file – Where the data will be exported to

• subset_df – description of the data to extract

• filters – whether to bring all the filters or only those that are needed to describe the data.

Right now this is “all”, but TODO: support “required_only” filters, meaning that we only bring the filters from the selected channels.

Returns:

mth5.utils.fdsn_tools module

FDSN standards tools.

Tools for working with FDSN (Incorporated Research Institutions for Seismology) standards including SEED channel codes, period/measurement/orientation codes, and conversions between SEED and MT (magnetotelluric) channel formats.

Notes

Created on Wed Sep 30 11:47:01 2020

Author: Jared Peacock

License: MIT

References

FDSN Channel Codes: https://www.fdsn.org/seed_manual/SEEDManual_V2.4.pdf

mth5.utils.fdsn_tools.get_location_code(channel_obj: object) → str

Generate FDSN location code from channel metadata.

Creates a 2-character location code from the channel’s component and channel number.

Parameters:

channel_obj (object) – Channel metadata object with component and channel_number attributes. Expected to be of type ~mt_metadata.timeseries.Channel.

Returns:

2-character location code formatted as first letter of component and last digit of channel number (e.g., ‘E1’, ‘H0’).

Return type:

str

Examples

Generate location code for an electric component on channel 5:

>>> class MockChannel:
...     component = 'ex'
...     channel_number = 5
>>> get_location_code(MockChannel())
'E5'

Magnetic component on channel 12 (wraps to 2):

>>> class MockChannel:
...     component = 'hx'
...     channel_number = 12
>>> get_location_code(MockChannel())
'H2'

mth5.utils.fdsn_tools.get_measurement_code(measurement: str) → str

Get SEED sensor code from measurement type.

Maps measurement types to single-character SEED sensor codes. Performs case-insensitive substring matching.

Parameters:

measurement (str) – Measurement type (e.g., ‘electric’, ‘magnetics’, ‘temperature’, ‘tilt’, ‘pressure’, ‘humidity’, ‘gravity’, ‘calibration’, ‘rain_fall’, ‘water_current’, ‘wind’, ‘linear_strain’, ‘tide’, ‘creep’).

Returns:

Single character SEED sensor code. Returns ‘Y’ if measurement type not found in mapping dictionary.

Return type:

str

Notes

Measurement to code mapping: - ‘tilt’ → ‘A’ - ‘creep’ → ‘B’ - ‘calibration’ → ‘C’ - ‘pressure’ → ‘D’ - ‘magnetics’ → ‘F’ - ‘gravity’ → ‘G’ - ‘humidity’ → ‘I’ - ‘temperature’ → ‘K’ - ‘water_current’ → ‘O’ - ‘electric’ → ‘Q’ - ‘rain_fall’ → ‘R’ - ‘linear_strain’ → ‘S’ - ‘tide’ → ‘T’ - ‘wind’ → ‘W’ - unknown/auxiliary → ‘Y’

Examples

Get code for electric measurement:

>>> get_measurement_code('electric')
'Q'

Get code for magnetic field:

>>> get_measurement_code('magnetics')
'F'

Unknown measurement returns ‘Y’:

>>> get_measurement_code('unknown')
'Y'

mth5.utils.fdsn_tools.get_orientation_code(azimuth: float, orientation: str = 'horizontal') → str

Get SEED orientation code from azimuth and orientation type.

Maps azimuth angle to SEED orientation code based on whether the sensor is oriented horizontally or vertically.

Parameters:

• azimuth (float) – Azimuth angle in degrees where 0 is north, 90 is east, 180 is south, 270 is west. For vertical orientation, 0 = vertical down.

• orientation ({'horizontal', 'vertical'}, default 'horizontal') – Type of sensor orientation.

Returns:

Single character SEED orientation code.

Return type:

str

Raises:

ValueError – If orientation is not ‘horizontal’ or ‘vertical’.

Notes

Horizontal orientation codes (azimuths): - ‘N’: 0-15° (North) - ‘E’: 75-90° (East) - ‘1’: 15-45° (NE quadrant) - ‘2’: 45-75° (SE quadrant)

Vertical orientation codes: - ‘Z’: 0-15° (Primary vertical) - ‘3’: 15-75° (Alternate vertical)

Examples

Get code for northerly azimuth:

>>> get_orientation_code(10.0, orientation='horizontal')
'N'

Get code for easterly azimuth:

>>> get_orientation_code(85.0, orientation='horizontal')
'E'

Get code for vertical sensor:

>>> get_orientation_code(0.0, orientation='vertical')
'Z'

mth5.utils.fdsn_tools.get_period_code(sample_rate: float) → str

Get SEED sampling rate code from sample rate.

Determines the appropriate FDSN/SEED period code based on the sample rate in samples per second. Codes range from ‘Q’ (highest frequency, period < 1 μs) to ‘F’ (lowest frequency, period 1000-5000 s).

Parameters:

sample_rate (float) – Sample rate in samples per second.

Returns:

Single character SEED sampling rate code. Defaults to ‘A’ if no code matches the sample rate.

Return type:

str

Notes

Code mapping (frequency/period ranges): - ‘F’, ‘G’: 1-5 kHz - ‘D’, ‘C’: 250-1000 Hz - ‘E’, ‘H’: 80-250 Hz - ‘S’, ‘B’: 10-80 Hz - ‘M’: 1-10 Hz - ‘L’: 0.95-1.05 Hz - ‘V’: 0.095-0.105 Hz - ‘U’: 0.0095-0.0105 Hz - ‘R’: 0.0001-0.001 Hz - ‘P’: 0.00001-0.0001 Hz - ‘T’: 0.000001-0.00001 Hz - ‘Q’: < 0.000001 Hz

Examples

Get code for 100 Hz sample rate:

>>> get_period_code(100.0)
'B'

Get code for 1000 Hz:

>>> get_period_code(1000.0)
'D'

Get code for 10 Hz (default ‘A’):

>>> get_period_code(10.0)
'M'

mth5.utils.fdsn_tools.make_channel_code(channel_obj: object) → str

Generate 3-character SEED channel code from channel metadata.

Combines period code, measurement code, and orientation code into a standard 3-character FDSN channel code.

Parameters:

channel_obj (object) – Channel metadata object with attributes: sample_rate, component, type, measurement_azimuth, measurement_tilt. Expected to be of type ~mt_metadata.timeseries.Channel.

Returns:

3-character SEED channel code (e.g., ‘BHZ’, ‘HHE’).

Return type:

str

Notes

The channel code format is: [Period Code][Measurement Code][Orientation Code]

• Period code: based on sample_rate

• Measurement code: derived from component, with fallback to type

• Orientation code: depends on whether component is vertical (‘z’)

Examples

Create channel code for horizontal electric component:

>>> class MockChannel:
...     sample_rate = 100.0
...     component = 'ex'
...     type = 'electric'
...     measurement_azimuth = 0.0
...     measurement_tilt = 0.0
>>> make_channel_code(MockChannel())
'BQN'

Create channel code for vertical magnetic component:

>>> class MockChannel:
...     sample_rate = 100.0
...     component = 'hz'
...     type = 'magnetic'
...     measurement_azimuth = 0.0
...     measurement_tilt = 0.0
>>> make_channel_code(MockChannel())
'BFZ'

mth5.utils.fdsn_tools.make_mt_channel(code_dict: dict[str, dict[str, int] | str | bool], angle_tol: int = 15) → str

Convert FDSN code dictionary to magnetotelluric (MT) channel code.

Maps FDSN codes to MT channel naming convention (e.g., ‘ex’, ‘hy’, ‘hz’).

Parameters:

• code_dict (dict) – Dictionary with keys: - ‘component’ (str): Measurement type (e.g., ‘magnetics’, ‘electric’). - ‘vertical’ (bool): True for vertical, False for horizontal. - ‘orientation’ (dict): Orientation range with ‘min’ and ‘max’.

• angle_tol (int, default 15) – Angle tolerance in degrees for determining cardinal directions.

Returns:

2-character MT channel code (e.g., ‘ex’, ‘hy’, ‘hz’). Format: [component code][direction code] - Component: ‘e’ (electric) or ‘h’ (magnetic) - Direction: ‘x’, ‘y’, ‘z’ for cardinal or ‘1’, ‘2’, ‘3’ for intermediate

Return type:

str

Notes

Direction mapping for horizontal channels (0-90°): - ‘x’: North direction (0-15°) - ‘1’: NE quadrant (15-45°) - ‘y’: East direction (90-angle_tol to 90°) - ‘2’: SE quadrant (45-90-angle_tol°)

Vertical channels: - ‘z’: Primary vertical (0-15°) - ‘3’: Alternate vertical (15-90°)

Examples

Create north-oriented electric channel:

>>> code_dict = {
...     'component': 'electric',
...     'vertical': False,
...     'orientation': {'min': 0, 'max': 15}
... }
>>> make_mt_channel(code_dict)
'ex'

Create vertical magnetic channel:

>>> code_dict = {
...     'component': 'magnetics',
...     'vertical': True,
...     'orientation': {'min': 0, 'max': 15}
... }
>>> make_mt_channel(code_dict)
'hz'

mth5.utils.fdsn_tools.read_channel_code(channel_code: str) → dict[str, dict[str, int] | str | bool]

Parse FDSN channel code into components.

Decodes a 3-character SEED channel code into its constituent parts: period range, component type, orientation range, and vertical flag.

Parameters:

channel_code (str) – 3-character FDSN channel code (e.g., ‘BHZ’, ‘HHE’).

Returns:

Dictionary with keys: - ‘period’ (dict): Period range with ‘min’ and ‘max’ keys (Hz). - ‘component’ (str): Component type (e.g., ‘electric’, ‘magnetics’). - ‘orientation’ (dict): Angle range with ‘min’ and ‘max’ keys (degrees). - ‘vertical’ (bool): True if component is vertical, False otherwise.

Return type:

dict

Raises:

ValueError – If channel code is not 3 characters, contains invalid period code, or contains invalid orientation code.

Notes

Vertical components are identified by orientation codes ‘Z’ or ‘3’.

Examples

Decode a horizontal channel code:

>>> result = read_channel_code('BHE')
>>> result['component']
'magnetics'
>>> result['vertical']
False

Decode a vertical channel code:

>>> result = read_channel_code('BHZ')
>>> result['vertical']
True

mth5.utils.helpers module

mth5.utils.helpers.add_filters(m: str | Path | MTH5, filters_list: list[Any], survey_id: str = '') → None

Add filter objects to MTH5 file.

Adds a list of filter objects to the MTH5 file’s filter group. Automatically selects the appropriate filters group based on file version.

Parameters:

• m (str | pathlib.Path | MTH5) – Path to MTH5 file or MTH5 object.

• filters_list (list) – List of filter objects to add. Each filter should have a ‘name’ attribute and be compatible with the filters group.

• survey_id (str, default '') – Survey ID for file version 0.2.0. Required for version 0.2.0, ignored for version 0.1.0.

Raises:

• AttributeError – If filter objects lack required attributes.

• ValueError – If survey_id is not found in version 0.2.0 files.

Notes

File version 0.1.0 stores filters globally. File version 0.2.0 stores filters per survey.

Examples

Add filters to MTH5 file:

>>> from mth5.timeseries import Filter
>>> filters = [Filter(name='test_filter')]
>>> add_filters('/path/to/file.mth5', filters)

Add survey-specific filters (version 0.2.0):

>>> add_filters('/path/to/file.mth5', filters, survey_id='MT01')

mth5.utils.helpers.get_channel_summary(m: str | Path | MTH5, show: bool = True) → Any

Get channel summary from MTH5 file as pandas DataFrame.

Retrieves the channel summary table and converts to DataFrame. Automatically re-summarizes if the summary appears incomplete.

Parameters:

• m (str | pathlib.Path | MTH5) – Path to MTH5 file or MTH5 object.

• show (bool, default True) – Whether to log the summary DataFrame to console.

Returns:

Channel summary with station, run, and channel information.

Return type:

pandas.DataFrame

Warning

If the summary appears incomplete, the channel summary table is re-summarized which may take time for large files.

Examples

Get channel summary from file path:

>>> df = get_channel_summary('/path/to/file.mth5')
>>> print(df.shape)
(42, 8)

Get summary without logging:

>>> df = get_channel_summary('/path/to/file.mth5', show=False)

mth5.utils.helpers.get_compare_dict(input_dict: dict[str, Any]) → dict[str, Any]

Remove MTH5-specific metadata attributes for comparison.

Removes internal attributes added by MTH5 that may interfere with dictionary comparisons between metadata objects.

Parameters:

input_dict (dict) – Dictionary to clean, typically metadata dictionary.

Returns:

Dictionary with MTH5 internal attributes removed. Original dict is modified in-place.

Return type:

dict

Notes

Removed attributes: - hdf5_reference: HDF5 object reference (internal) - mth5_type: MTH5 data type marker (internal)

Examples

Clean metadata dictionary before comparison:

>>> metadata = {
...     'id': 'station_001',
...     'latitude': 45.5,
...     'hdf5_reference': <h5py reference>,
...     'mth5_type': 'Station'
... }
>>> clean = get_compare_dict(metadata)
>>> print(clean)
{'id': 'station_001', 'latitude': 45.5}

Safe to call with incomplete dicts:

>>> metadata = {'id': 'station_001'}
>>> clean = get_compare_dict(metadata)  # No error if keys absent

mth5.utils.helpers.get_version(m: str | Path | MTH5) → str

Get the file version from an MTH5 file.

Parameters:

m (str | pathlib.Path | MTH5) – Path to MTH5 file or MTH5 object.

Returns:

File version string (e.g., ‘0.1.0’, ‘0.2.0’).

Return type:

str

Examples

Get version from file path:

>>> version = get_version('/path/to/file.mth5')
>>> print(version)
'0.2.0'

Get version from MTH5 object:

>>> with MTH5() as m:
...     m.open_mth5('/path/to/file.mth5')
...     version = get_version(m)

mth5.utils.helpers.initialize_mth5(h5_path: str | Path, mode: str = 'a', file_version: str = '0.1.0') → MTH5

Initialize and open an MTH5 file for reading or writing.

Creates or opens an MTH5 file with specified file version. Optionally removes existing files before write operations.

Parameters:

• h5_path (str | pathlib.Path) – Path to MTH5 file. Created if it doesn’t exist.

• mode ({'r', 'w', 'a'}, default 'a') – File access mode: - ‘r’: read-only - ‘w’: write (overwrites existing file) - ‘a’: append/read-write

• file_version ({'0.1.0', '0.2.0'}, default '0.1.0') – MTH5 file format version.

Returns:

Initialized and opened MTH5 object.

Return type:

MTH5

Warning

When mode=’w’ and file exists, all open h5 files are closed before removal. This may affect other processes using HDF5 files.

Examples

Create a new MTH5 file:

>>> m = initialize_mth5('/path/to/file.mth5', mode='w')
>>> m.file_version
'0.1.0'
>>> m.close_mth5()

Open existing file for appending:

>>> m = initialize_mth5('/path/to/file.mth5', mode='a')
>>> m.add_station('MT001')
>>> m.close_mth5()

Open file with version 0.2.0 schema:

>>> m = initialize_mth5('/path/to/file.mth5', file_version='0.2.0')

mth5.utils.helpers.path_or_mth5_object(func: Callable[[...], T]) → Callable[[...], T]

Decorator allowing functions to accept MTH5 file paths or MTH5 objects.

Transparently converts file paths to MTH5 objects, opens the file, and passes the MTH5 object to the decorated function.

Parameters:

func (Callable) – A function that takes an MTH5 object as its first argument. Signature: func(mth5_obj: MTH5, *args, **kwargs) -> T

Returns:

Wrapped function accepting str/Path or MTH5 as first argument.

Return type:

Callable

Raises:

TypeError – If first argument is not a string, pathlib.Path, or MTH5 object.

Notes

The decorated function can be called with either: - A file path string or pathlib.Path - An MTH5 object

When given a file path, the decorator automatically opens the file in ‘append’ mode by default, unless overridden in kwargs.

TODO: add support for file_version in kwargs

Examples

Decorate a function to work with both paths and objects:

@path_or_mth5_object
def get_metadata(m: MTH5) -> dict:
    return m.survey_group.metadata.to_dict()

# Call with file path
metadata = get_metadata('/path/to/file.mth5')

# Call with MTH5 object
with MTH5() as m:
    m.open_mth5('/path/to/file.mth5', mode='r')
    metadata = get_metadata(m)

mth5.utils.helpers.read_back_data(mth5_path: str | Path, station_id: str, run_id: str, survey: str | None = None, close_mth5: bool = True, return_objects: list[str] | None = None) → dict[str, Any]

Read station/run data from MTH5 file for testing and validation.

Helper function to confirm MTH5 file accessibility and validate that data dimensions match expectations.

Parameters:

• mth5_path (str | pathlib.Path) – Full path to MTH5 file to read.

• station_id (str) – Station identifier (e.g., ‘PKD’, ‘MT001’).

• run_id (str) – Run identifier (e.g., ‘001’, ‘1’).

• survey (str, optional) – Survey identifier. Required for file version 0.2.0.

• close_mth5 (bool, default True) – Whether to close MTH5 object after reading. Set to False if you need to access the object later.

• return_objects (list of str, optional) – Specifies what objects to return. Options: - ‘run’: RunGroup object - ‘run_ts’: RunTS time series object If None, returns empty dict with only mth5_obj if close_mth5=False.

Returns:

Dictionary containing requested objects: - ‘run’: RunGroup (if ‘run’ in return_objects) - ‘run_ts’: RunTS (if ‘run_ts’ in return_objects) - ‘mth5_obj’: MTH5 (if close_mth5=False)

Return type:

dict

Warning

If close_mth5=False, the MTH5 object must be manually closed to avoid resource leaks.

Notes

This is primarily a testing utility. Data shape is logged to console.

Examples

Read run data and close immediately:

>>> result = read_back_data(
...     '/path/to/file.mth5',
...     'PKD',
...     '001',
...     return_objects=['run_ts']
... )
>>> ts = result['run_ts']
>>> print(ts.dataset.shape)

Read data and keep MTH5 object open:

>>> result = read_back_data(
...     '/path/to/file.mth5',
...     'MT001',
...     '1',
...     survey='survey_01',
...     close_mth5=False,
...     return_objects=['run', 'run_ts']
... )
>>> run = result['run']
>>> m = result['mth5_obj']
>>> # ... use objects ...
>>> m.close_mth5()

TODO: add path_or_mth5_decorator to this function

mth5.utils.helpers.station_in_mth5(m: str | Path | MTH5, station_id: str, survey_id: str | None = None) → bool

Check if a station exists in MTH5 file.

Determines whether a station with the given ID is present in the MTH5 file using the groups list.

Parameters:

• m (str | pathlib.Path | MTH5) – Path to MTH5 file or MTH5 object.

• station_id (str) – Station identifier (e.g., ‘PKD’, ‘MT001’).

• survey_id (str, optional) – Survey identifier. Required for file version 0.2.0, ignored for version 0.1.0.

Returns:

True if station exists, False otherwise.

Return type:

bool

Raises:

NotImplementedError – If file version is not 0.1.0 or 0.2.0.

Notes

File version 0.1.0 has global stations group. File version 0.2.0 has per-survey stations groups.

Alternative method: Use channel_summary DataFrame:

df = m.channel_summary.to_dataframe()
station_exists = station_id in df['Station'].unique()

Examples

Check if station exists (file version 0.1.0):

>>> exists = station_in_mth5('/path/to/file.mth5', 'PKD')
>>> print(exists)
True

Check in version 0.2.0 with survey ID:

>>> exists = station_in_mth5(
...     '/path/to/file.mth5',
...     'MT001',
...     survey_id='survey_01'
... )

mth5.utils.helpers.survey_in_mth5(m: str | Path | MTH5, survey_id: str | None = None) → bool

Check if a survey exists in MTH5 file.

Determines whether a survey with the given ID exists in the MTH5 file. Behavior varies by file version: 0.1.0 has a single survey, while 0.2.0 supports multiple surveys.

Parameters:

• m (str | pathlib.Path | MTH5) – Path to MTH5 file or MTH5 object.

• survey_id (str, optional) – Survey identifier. For file version 0.1.0, compared against the global survey ID. For version 0.2.0, checked in surveys group.

Returns:

True if survey exists, False otherwise.

Return type:

bool

Raises:

NotImplementedError – If file version is not 0.1.0 or 0.2.0.

Notes

File version 0.1.0 has a single survey with fixed ID. File version 0.2.0 supports multiple named surveys.

Alternative method: Use channel_summary DataFrame:

df = m.channel_summary.to_dataframe()
surveys = df['Survey'].unique()
survey_exists = survey_id in surveys

Examples

Check if survey exists (file version 0.1.0):

>>> exists = survey_in_mth5('/path/to/file.mth5', 'survey_01')
>>> print(exists)
True

Check in version 0.2.0:

>>> exists = survey_in_mth5('/path/to/file.mth5', survey_id='MT')
>>> if exists:
...     print(f"Survey MT found in file")

Module contents