Coverage for C: \ Users \ peaco \ OneDrive \ Documents \ GitHub \ mth5 \ mth5 \ groups \ filters.py: 81%
95 statements
« prev ^ index » next coverage.py v7.13.1, created at 2026-01-27 20:09 -0800
1# -*- coding: utf-8 -*-
2"""
3Filter groups manager for handling multiple filter types in MTH5.
5This module provides a unified interface for managing different types of filters
6including zeros-poles-gain (ZPK), coefficients, time delays, frequency-amplitude-phase (FAP),
7and finite impulse response (FIR) filters.
9:copyright:
10 Jared Peacock (jpeacock@usgs.gov)
12:license: MIT
14"""
16# =============================================================================
17# Imports
18# =============================================================================
19from __future__ import annotations
21from typing import Any
23import h5py
25from mth5.groups.base import BaseGroup
26from mth5.groups.filter_groups import (
27 CoefficientGroup,
28 FAPGroup,
29 FIRGroup,
30 TimeDelayGroup,
31 ZPKGroup,
32)
35# =============================================================================
36# Filters Group
37# =============================================================================
40class FiltersGroup(BaseGroup):
41 """
42 Container for managing all filter types in MTH5 format.
44 This class provides a unified interface for organizing and accessing filters
45 of different types. It automatically creates and manages subgroups for each
46 filter type (ZPK, Coefficient, Time Delay, FAP, and FIR) within the HDF5
47 file structure.
49 Filter Types
50 -----------
51 - **zpk**: Zeros, Poles, and Gain representation
52 - **coefficient**: FIR coefficient filter
53 - **time_delay**: Time delay filter
54 - **fap**: Frequency-Amplitude-Phase (FAP) lookup table
55 - **fir**: Finite Impulse Response filter
57 Parameters
58 ----------
59 group : h5py.Group
60 HDF5 group object for the filters container.
61 **kwargs
62 Additional keyword arguments passed to BaseGroup.
64 Attributes
65 ----------
66 zpk_group : ZPKGroup
67 Subgroup for zeros-poles-gain filters.
68 coefficient_group : CoefficientGroup
69 Subgroup for coefficient filters.
70 time_delay_group : TimeDelayGroup
71 Subgroup for time delay filters.
72 fap_group : FAPGroup
73 Subgroup for frequency-amplitude-phase filters.
74 fir_group : FIRGroup
75 Subgroup for FIR filters.
77 Examples
78 --------
79 >>> import h5py
80 >>> from mth5.groups.filters import FiltersGroup
81 >>> with h5py.File('data.h5', 'r') as f:
82 ... filters = FiltersGroup(f['Filters'])
83 ... all_filters = filters.filter_dict
84 ... zpk_filter = filters.to_filter_object('my_zpk_filter')
85 """
87 def __init__(self, group: h5py.Group, **kwargs) -> None:
88 super().__init__(group, **kwargs)
90 try:
91 self.zpk_group = ZPKGroup(self.hdf5_group.create_group("zpk"))
92 except ValueError:
93 self.zpk_group = ZPKGroup(self.hdf5_group["zpk"])
94 try:
95 self.coefficient_group = CoefficientGroup(
96 self.hdf5_group.create_group("coefficient")
97 )
98 except ValueError:
99 self.coefficient_group = CoefficientGroup(self.hdf5_group["coefficient"])
100 try:
101 self.time_delay_group = TimeDelayGroup(
102 self.hdf5_group.create_group("time_delay")
103 )
104 except ValueError:
105 self.time_delay_group = TimeDelayGroup(self.hdf5_group["time_delay"])
106 try:
107 self.fap_group = FAPGroup(self.hdf5_group.create_group("fap"))
108 except ValueError:
109 self.fap_group = FAPGroup(self.hdf5_group["fap"])
110 try:
111 self.fir_group = FIRGroup(self.hdf5_group.create_group("fir"))
112 except ValueError:
113 self.fir_group = FIRGroup(self.hdf5_group["fir"])
115 @property
116 def filter_dict(self) -> dict[str, Any]:
117 """
118 Get a dictionary of all filters across all filter type groups.
120 Aggregates filters from all subgroups (ZPK, Coefficient, Time Delay, FAP, FIR)
121 into a single dictionary for convenient access and querying.
123 Returns
124 -------
125 dict[str, Any]
126 Dictionary mapping filter names to filter metadata dictionaries.
127 Each entry contains filter information including type and HDF5 reference.
129 Examples
130 --------
131 >>> filters = FiltersGroup(h5_group)
132 >>> all_filters = filters.filter_dict
133 >>> print(list(all_filters.keys()))
134 ['my_zpk_filter', 'lowpass_coefficient', 'time_delay_1', ...]
135 >>> print(all_filters['my_zpk_filter']['type'])
136 'zpk'
137 """
138 filter_dict = {}
139 filter_dict.update(self.zpk_group.filter_dict)
140 filter_dict.update(self.coefficient_group.filter_dict)
141 filter_dict.update(self.time_delay_group.filter_dict)
142 filter_dict.update(self.fap_group.filter_dict)
143 filter_dict.update(self.fir_group.filter_dict)
145 return filter_dict
147 def add_filter(self, filter_object: object) -> object:
148 """
149 Add a filter dataset based on its type.
151 Automatically detects the filter type and routes the filter to the
152 appropriate subgroup. Filter names are normalized to lowercase and
153 forward slashes are replaced with " per " for consistency.
155 Parameters
156 ----------
157 filter_object : mt_metadata.timeseries.filters
158 An MT metadata filter object with a 'type' attribute.
159 Supported types:
161 - 'zpk', 'poles_zeros': Zeros-Poles-Gain filter
162 - 'coefficient': Coefficient filter
163 - 'time_delay', 'time delay': Time delay filter
164 - 'fap', 'frequency response table': Frequency-Amplitude-Phase filter
165 - 'fir': Finite Impulse Response filter
167 Returns
168 -------
169 object
170 Filter group object from the appropriate subgroup.
172 Notes
173 -----
174 If a filter with the same name already exists, the existing filter
175 is returned instead of creating a duplicate.
177 Examples
178 --------
179 >>> from mt_metadata.timeseries.filters import ZPK
180 >>> filters = FiltersGroup(h5_group)
181 >>> zpk_filter = ZPK(name='my_filter')
182 >>> added_filter = filters.add_filter(zpk_filter)
184 Add coefficient filter:
186 >>> from mt_metadata.timeseries.filters import Coefficient
187 >>> coeff_filter = Coefficient(name='lowpass')
188 >>> filters.add_filter(coeff_filter)
189 """
190 self.logger.debug(f"Type of filter {type(filter_object)}")
191 # make everything lower case for consistency
192 filter_object.name = filter_object.name.replace("/", " per ").lower()
194 if filter_object.type in ["zpk", "poles_zeros"]:
195 try:
196 return self.zpk_group.from_object(filter_object)
197 except ValueError:
198 self.logger.debug(f"group {filter_object.name} already exists")
199 return self.zpk_group.get_filter(filter_object.name)
200 elif filter_object.type in ["coefficient"]:
201 try:
202 return self.coefficient_group.from_object(filter_object)
203 except ValueError:
204 self.logger.debug(f"group {filter_object.name} already exists")
205 return self.coefficient_group.get_filter(filter_object.name)
206 elif filter_object.type in ["time_delay", "time delay"]:
207 try:
208 return self.time_delay_group.from_object(filter_object)
209 except ValueError:
210 self.logger.debug(f"group {filter_object.name} already exists")
211 return self.time_delay_group.get_filter(filter_object.name)
212 elif filter_object.type in ["fap", "frequency response table"]:
213 try:
214 return self.fap_group.from_object(filter_object)
215 except ValueError:
216 self.logger.debug(f"group {filter_object.name} already exists")
217 return self.fap_group.get_filter(filter_object.name)
218 elif filter_object.type in ["fir"]:
219 try:
220 return self.fir_group.from_object(filter_object)
221 except ValueError:
222 self.logger.debug(f"group {filter_object.name} already exists")
223 return self.fir_group.get_filter(filter_object.name)
225 def get_filter(self, name: str) -> h5py.Dataset | h5py.Group:
226 """
227 Retrieve a filter dataset by name.
229 Looks up the filter by name in the aggregated filter dictionary and
230 returns the HDF5 dataset or group object.
232 Parameters
233 ----------
234 name : str
235 Name of the filter to retrieve.
237 Returns
238 -------
239 h5py.Dataset or h5py.Group
240 HDF5 dataset or group object for the requested filter.
242 Raises
243 ------
244 KeyError
245 If the filter name is not found in the filter dictionary.
247 Examples
248 --------
249 >>> filters = FiltersGroup(h5_group)
250 >>> filter_dataset = filters.get_filter('my_zpk_filter')
251 >>> print(filter_dataset.attrs)
252 """
254 try:
255 hdf5_ref = self.filter_dict[name]["hdf5_ref"]
256 except KeyError:
257 msg = f"Could not find {name} in the filter dictionary"
258 self.logger.error(msg)
259 raise KeyError(msg)
260 return self.hdf5_group[hdf5_ref]
262 def to_filter_object(self, name: str) -> object:
263 """
264 Convert a filter HDF5 dataset to an MT metadata filter object.
266 Retrieves the filter metadata from the HDF5 file and converts it to
267 the appropriate MT metadata filter class based on filter type.
269 Parameters
270 ----------
271 name : str
272 Name of the filter to convert.
274 Returns
275 -------
276 object
277 MT metadata filter object (ZPK, Coefficient, TimeDelay, FAP, or FIR).
279 Raises
280 ------
281 KeyError
282 If the filter name is not found in the filter dictionary.
284 Examples
285 --------
286 >>> filters = FiltersGroup(h5_group)
287 >>> zpk_filter = filters.to_filter_object('my_zpk_filter')
288 >>> print(zpk_filter.name)
289 'my_zpk_filter'
290 >>> print(type(zpk_filter))
291 <class 'mt_metadata.timeseries.filters.ZPK'>
293 Get different filter types:
295 >>> coeff_filter = filters.to_filter_object('lowpass_coefficient')
296 >>> fap_filter = filters.to_filter_object('frequency_response_1')
297 """
299 try:
300 f_type = self.filter_dict[name]["type"]
301 except KeyError:
302 msg = f"Could not find {name} in the filter dictionary"
303 self.logger.error(msg)
304 raise KeyError(msg, name)
305 if f_type in ["zpk"]:
306 return self.zpk_group.to_object(name)
307 elif f_type in ["coefficient"]:
308 return self.coefficient_group.to_object(name)
309 elif f_type in ["time_delay", "time delay"]:
310 return self.time_delay_group.to_object(name)
311 elif f_type in ["fap", "frequency response table"]:
312 return self.fap_group.to_object(name)
313 elif f_type in ["fir"]:
314 return self.fir_group.to_object(name)