# -*- coding: utf-8 -*-
"""
Created on Wed Dec 23 17:08:40 2020
Need to make a group for FAP and FIR filters.
:copyright:
Jared Peacock (jpeacock@usgs.gov)
:license: MIT
"""
# =============================================================================
# Imports
# =============================================================================
from mth5.groups.base import BaseGroup
from mth5.groups.filter_groups import (
ZPKGroup,
CoefficientGroup,
TimeDelayGroup,
FAPGroup,
FIRGroup,
)
# =============================================================================
# Filters Group
# =============================================================================
[docs]class FiltersGroup(BaseGroup):
"""
Not implemented yet
"""
def __init__(self, group, **kwargs):
super().__init__(group, **kwargs)
try:
self.zpk_group = ZPKGroup(self.hdf5_group.create_group("zpk"))
except ValueError:
self.zpk_group = ZPKGroup(self.hdf5_group["zpk"])
try:
self.coefficient_group = CoefficientGroup(
self.hdf5_group.create_group("coefficient")
)
except ValueError:
self.coefficient_group = CoefficientGroup(self.hdf5_group["coefficient"])
try:
self.time_delay_group = TimeDelayGroup(
self.hdf5_group.create_group("time_delay")
)
except ValueError:
self.time_delay_group = TimeDelayGroup(self.hdf5_group["time_delay"])
try:
self.fap_group = FAPGroup(self.hdf5_group.create_group("fap"))
except ValueError:
self.fap_group = FAPGroup(self.hdf5_group["fap"])
try:
self.fir_group = FIRGroup(self.hdf5_group.create_group("fir"))
except ValueError:
self.fir_group = FIRGroup(self.hdf5_group["fir"])
@property
def filter_dict(self):
filter_dict = {}
filter_dict.update(self.zpk_group.filter_dict)
filter_dict.update(self.coefficient_group.filter_dict)
filter_dict.update(self.time_delay_group.filter_dict)
filter_dict.update(self.fap_group.filter_dict)
filter_dict.update(self.fir_group.filter_dict)
return filter_dict
[docs] def add_filter(self, filter_object):
"""
Add a filter dataset based on type
current types are:
* zpk --> zeros, poles, gain
* fap --> frequency look up table
* time_delay --> time delay filter
* coefficient --> coefficient filter
:param filter_object: An MT metadata filter object
:type filter_object: :class:`mt_metadata.timeseries.filters`
"""
filter_object.name = filter_object.name.replace("/", " per ")
if filter_object.type in ["zpk", "poles_zeros"]:
try:
return self.zpk_group.from_object(filter_object)
except ValueError:
self.logger.debug(f"group {filter_object.name} already exists")
return self.zpk_group.get_filter(filter_object.name)
elif filter_object.type in ["coefficient"]:
try:
return self.coefficient_group.from_object(filter_object)
except ValueError:
self.logger.debug(f"group {filter_object.name} already exists")
return self.coefficient_group.get_filter(filter_object.name)
elif filter_object.type in ["time_delay", "time delay"]:
try:
return self.time_delay_group.from_object(filter_object)
except ValueError:
self.logger.debug(f"group {filter_object.name} already exists")
return self.time_delay_group.get_filter(filter_object.name)
elif filter_object.type in ["fap", "frequency response table"]:
try:
return self.fap_group.from_object(filter_object)
except ValueError:
self.logger.debug(f"group {filter_object.name} already exists")
return self.fap_group.get_filter(filter_object.name)
elif filter_object.type in ["fir"]:
try:
return self.fir_group.from_object(filter_object)
except ValueError:
self.logger.debug(f"group {filter_object.name} already exists")
return self.fir_group.get_filter(filter_object.name)
[docs] def get_filter(self, name):
"""
Get a filter by name
"""
try:
hdf5_ref = self.filter_dict[name]["hdf5_ref"]
except KeyError:
msg = f"Could not find {name} in the filter dictionary"
self.logger.error(msg)
raise KeyError(msg)
return self.hdf5_group[hdf5_ref]
[docs] def to_filter_object(self, name):
"""
return the MT metadata representation of the filter
"""
try:
f_type = self.filter_dict[name]["type"]
except KeyError:
msg = "Could not find %s in the filter dictionary"
self.logger.error(msg, name)
raise KeyError(msg, name)
if f_type in ["zpk"]:
return self.zpk_group.to_object(name)
elif f_type in ["coefficient"]:
return self.coefficient_group.to_object(name)
elif f_type in ["time_delay", "time delay"]:
return self.time_delay_group.to_object(name)
elif f_type in ["fap", "frequency response table"]:
return self.fap_group.to_object(name)
elif f_type in ["fir"]:
return self.fir_group.to_object(name)