Coverage for C: \ Users \ peaco \ OneDrive \ Documents \ GitHub \ mth5 \ mth5 \ io \ reader.py: 79%
38 statements
« prev ^ index » next coverage.py v7.13.1, created at 2026-01-10 00:01 -0800
1# -*- coding: utf-8 -*-
2"""
3Universal reader for magnetotelluric time series data files.
5This module provides a plugin-like system for reading various MT data formats
6and returning appropriate :class:`mth5.timeseries` objects. The reader
7automatically detects file types and dispatches to the correct parser.
9Plugin Structure
10----------------
11If you are writing your own reader, implement the following structure:
13 * Class object that reads the given file format
14 * A reader function named read_{file_type} (e.g., read_nims)
15 * Return value should be a :class:`mth5.timeseries.MTTS` or
16 :class:`mth5.timeseries.RunTS` object plus extra metadata as a
17 dictionary with keys formatted as {level.attribute}
19Example Implementation
20----------------------
21.. code-block:: python
23 class NewFile:
24 def __init__(self, fn):
25 self.fn = fn
27 def read_header(self):
28 return header_information
30 def read_newfile(self):
31 ex, ey, hx, hy, hz = read_in_channels_as_MTTS
32 return RunTS([ex, ey, hx, hy, hz])
34 def read_newfile(fn):
35 new_file_obj = NewFile(fn)
36 run_obj = new_file_obj.read_newfile()
37 return run_obj, extra_metadata
39Then add your reader to the readers dictionary for automatic detection.
41See Also
42--------
43Existing readers in `mth5.io` for implementation guidance.
45Created on Wed Aug 26 10:32:45 2020
47:author: Jared Peacock
48:license: MIT
49"""
51# =============================================================================
52# Imports
53# =============================================================================
54from __future__ import annotations
56from pathlib import Path
57from typing import Any, Callable
59from loguru import logger
61from mth5.io import lemi, metronix, miniseed, nims, phoenix, usgs_ascii, zen
64# =============================================================================
65# Reader registry for MT data formats
66# =============================================================================
67readers: dict[str, dict[str, Any]] = {
68 "zen": {"file_types": ["z3d"], "reader": zen.read_z3d},
69 "nims": {"file_types": ["bin", "bnn"], "reader": nims.read_nims},
70 "usgs_ascii": {
71 "file_types": ["asc", "zip"],
72 "reader": usgs_ascii.read_ascii,
73 },
74 "miniseed": {
75 "file_types": ["miniseed", "ms", "mseed"],
76 "reader": miniseed.read_miniseed,
77 },
78 "lemi424": {
79 "file_types": ["txt"],
80 "reader": lemi.read_lemi424,
81 },
82 "phoenix": {
83 "file_types": ["bin", "td_30", "td_150", "td_24k"],
84 "reader": phoenix.read_phoenix,
85 },
86 "metronix": {
87 "file_types": ["atss"],
88 "reader": metronix.read_atss,
89 },
90}
93def get_reader(extension: str) -> tuple[str, Callable]:
94 """
95 Get the appropriate reader function for a file extension.
97 Searches the reader registry to find the correct parser function
98 for the given file extension. Handles ambiguous extensions by
99 issuing warnings when multiple readers might apply.
101 Parameters
102 ----------
103 extension : str
104 File extension (without the dot) to find a reader for
106 Returns
107 -------
108 tuple[str, Callable]
109 Tuple containing:
110 - Reader name (str): Identifier for the reader type
111 - Reader function (Callable): Function to parse files of this type
113 Raises
114 ------
115 ValueError
116 If no reader is found for the given file extension
118 Examples
119 --------
120 >>> reader_name, reader_func = get_reader("z3d")
121 >>> print(reader_name) # "zen"
122 >>> data = reader_func("/path/to/file.z3d")
124 Notes
125 -----
126 Some extensions like "bin" are ambiguous and could match multiple
127 readers (NIMS or Phoenix). A warning is issued in such cases.
128 """
129 if extension in ["bin"]:
130 logger.warning("Suggest inputing file type, bin could be nims or phoenix")
131 for key, vdict in readers.items():
132 if extension.lower() in vdict["file_types"]:
133 return key, vdict["reader"]
134 msg = f"Could not find a reader for file type {extension}"
135 logger.error(msg)
136 raise ValueError(msg)
139def read_file(
140 fn: str | Path | list[str | Path], file_type: str | None = None, **kwargs: Any
141) -> Any:
142 """
143 Universal reader for magnetotelluric time series data files.
145 Automatically detects the file type based on extension and dispatches
146 to the appropriate reader function. Supports both single files and
147 lists of files for multi-file formats.
149 Parameters
150 ----------
151 fn : str, Path, or list of str/Path
152 Full path(s) to data file(s) to be read. For multi-file formats,
153 pass a list of file paths.
154 file_type : str, optional
155 Specific reader type to use if file extension is ambiguous.
156 Must be one of the keys in the readers registry, by default None
157 **kwargs : dict
158 Additional keyword arguments passed to the specific reader function.
159 Supported arguments depend on the file format and reader.
161 Returns
162 -------
163 MTTS or RunTS
164 Time series object containing the data:
165 - :class:`mth5.timeseries.MTTS` for single channel data
166 - :class:`mth5.timeseries.RunTS` for multi-channel run data
168 Raises
169 ------
170 IOError
171 If any specified file does not exist
172 KeyError
173 If the specified file_type is not supported
174 ValueError
175 If no reader can be found for the file extension
177 Examples
178 --------
179 Read a single Z3D file (auto-detected)
181 >>> data = read_file("/path/to/station_001.z3d")
182 >>> print(type(data)) # <class 'mth5.timeseries.ChannelTS'>
184 Read with explicit file type for ambiguous extensions
186 >>> data = read_file("/path/to/data.bin", file_type="nims")
187 >>> print(data.n_channels)
189 Read multiple files for a multi-file format
191 >>> files = ["/path/to/file1.asc", "/path/to/file2.asc"]
192 >>> run_data = read_file(files, sample_rate=1.0)
194 Notes
195 -----
196 Supported file types and extensions:
197 - zen: .z3d (Zonge Z3D files)
198 - nims: .bin, .bnn (USGS NIMS files)
199 - usgs_ascii: .asc, .zip (USGS ASCII format)
200 - miniseed: .miniseed, .ms, .mseed (miniSEED format)
201 - lemi424: .txt (LEMI-424 format)
202 - phoenix: .bin, .td_30, .td_150, .td_24k (Phoenix formats)
203 - metronix: .atss (Metronix ADU format)
205 For ambiguous extensions like .bin, specify file_type explicitly.
206 """
208 if isinstance(fn, (list, tuple)):
209 fn = [Path(ff) for ff in fn]
210 if not fn[0].exists():
211 msg = f"Could not find file {fn}. Check path."
212 logger.error(msg)
213 raise IOError(msg)
214 file_ext = fn[0].suffix[1:]
216 else:
217 fn = Path(fn)
218 if not fn.exists():
219 msg = f"Could not find file {fn}. Check path."
220 logger.error(msg)
221 raise IOError(msg)
222 file_ext = fn.suffix[1:]
224 if file_type is not None:
225 try:
226 file_reader = readers[file_type]["reader"]
227 except KeyError:
228 msg = (
229 f"Reader for the file type {file_type} does not currently exist. "
230 f"Current readers {list(readers.keys())}"
231 )
232 logger.error(msg)
233 raise KeyError(msg)
234 else:
235 file_type, file_reader = get_reader(file_ext)
236 return file_reader(fn, **kwargs)