From a681f2850d5b170a87d62dcd8eda39ea49fedff0 Mon Sep 17 00:00:00 2001 From: benjamink Date: Mon, 10 Feb 2025 14:50:43 -0800 Subject: [PATCH 01/10] Implement tem BaseData params classes --- simpeg_drivers/__init__.py | 10 +- .../frequency_domain/__init__.py | 5 +- .../frequency_domain/driver.py | 16 +- .../frequency_domain/params.py | 338 +++++++++----- .../electromagnetics/time_domain/__init__.py | 5 +- .../electromagnetics/time_domain/driver.py | 67 ++- .../electromagnetics/time_domain/params.py | 411 ++++++++++++------ simpeg_drivers/params.py | 4 +- tests/run_tests/driver_airborne_tem_test.py | 56 +-- tests/run_tests/driver_fem_test.py | 40 +- 10 files changed, 648 insertions(+), 304 deletions(-) diff --git a/simpeg_drivers/__init__.py b/simpeg_drivers/__init__.py index dfaeddd5..de7d9e6f 100644 --- a/simpeg_drivers/__init__.py +++ b/simpeg_drivers/__init__.py @@ -91,7 +91,10 @@ def assets_path() -> Path: ), "fem": ( "simpeg_drivers.electromagnetics.frequency_domain.driver", - {"inversion": "FrequencyDomainElectromagneticsDriver"}, + { + "forward": "FrequenceyDomainElectromagneticsForwardDriver", + "inversion": "FrequencyDomainElectromagneticsInversionDriver", + }, ), "joint cross gradient": ( "simpeg_drivers.joint.joint_cross_gradient.driver", @@ -99,7 +102,10 @@ def assets_path() -> Path: ), "tdem": ( "simpeg_drivers.electromagnetics.time_domain.driver", - {"inversion": "TimeDomainElectromagneticsDriver"}, + { + "forward": "TimeDomainElectromagneticsForwardDriver", + "inversion": "TimeDomainElectromagneticsInversionDriver", + }, ), "magnetotellurics": ( "simpeg_drivers.natural_sources.magnetotellurics.driver", diff --git a/simpeg_drivers/electromagnetics/frequency_domain/__init__.py b/simpeg_drivers/electromagnetics/frequency_domain/__init__.py index 9dc6d396..c55cf457 100644 --- a/simpeg_drivers/electromagnetics/frequency_domain/__init__.py +++ b/simpeg_drivers/electromagnetics/frequency_domain/__init__.py @@ -9,7 +9,10 @@ # ''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''' -from .params import FrequencyDomainElectromagneticsParams +from .params import ( + FrequencyDomainElectromagneticsForwardParams, + FrequencyDomainElectromagneticsInversionParams, +) # pylint: disable=unused-import # flake8: noqa diff --git a/simpeg_drivers/electromagnetics/frequency_domain/driver.py b/simpeg_drivers/electromagnetics/frequency_domain/driver.py index 78dd8cc3..a7fec593 100644 --- a/simpeg_drivers/electromagnetics/frequency_domain/driver.py +++ b/simpeg_drivers/electromagnetics/frequency_domain/driver.py @@ -14,12 +14,20 @@ from simpeg_drivers.driver import InversionDriver from .constants import validations -from .params import FrequencyDomainElectromagneticsParams +from .params import ( + FrequencyDomainElectromagneticsForwardParams, + FrequencyDomainElectromagneticsInversionParams, +) -class FrequencyDomainElectromagneticsDriver(InversionDriver): - _params_class = FrequencyDomainElectromagneticsParams +class FrequencyDomainElectromagneticsForwardDriver(InversionDriver): + _params_class = FrequencyDomainElectromagneticsForwardParams _validations = validations - def __init__(self, params: FrequencyDomainElectromagneticsParams): + def __init__(self, params: FrequencyDomainElectromagneticsForwardParams): super().__init__(params) + + +class FrequencyDomainElectromagneticsInversionDriver(InversionDriver): + _params_class = FrequencyDomainElectromagneticsInversionParams + _validations = validations diff --git a/simpeg_drivers/electromagnetics/frequency_domain/params.py b/simpeg_drivers/electromagnetics/frequency_domain/params.py index 2791f19a..32cbd194 100644 --- a/simpeg_drivers/electromagnetics/frequency_domain/params.py +++ b/simpeg_drivers/electromagnetics/frequency_domain/params.py @@ -11,68 +11,61 @@ from __future__ import annotations -from copy import deepcopy -from uuid import UUID +from pathlib import Path +from typing import ClassVar, TypeAlias + +from geoh5py.groups import PropertyGroup +from geoh5py.objects import ( + AirborneFEMReceivers, + LargeLoopGroundFEMReceivers, + MovingLoopGroundFEMReceivers, +) + +from simpeg_drivers import assets_path +from simpeg_drivers.params import BaseForwardData, BaseInversionData -from simpeg_drivers.params import InversionBaseParams -from .constants import ( - default_ui_json, - forward_defaults, - inversion_defaults, - validations, +Receivers: TypeAlias = ( + MovingLoopGroundFEMReceivers | LargeLoopGroundFEMReceivers | AirborneFEMReceivers ) -class FrequencyDomainElectromagneticsParams(InversionBaseParams): +class FrequencyDomainElectromagneticsForwardParams(BaseForwardData): """ - Parameter class for Frequency-domain Electromagnetic (FEM) -> conductivity inversion. + Parameter class for Frequency-domain Electromagnetic (FEM) simulation. + + :param z_real_channel_bool: Real impedance channel boolean. + :param z_imag_channel_bool: Imaginary impedance channel boolean. + :param model_type: Specify whether the models are provided in resistivity or conductivity. """ - _physical_property = "conductivity" - - def __init__(self, input_file=None, forward_only=False, **kwargs): - self._default_ui_json = deepcopy(default_ui_json) - self._forward_defaults = deepcopy(forward_defaults) - self._inversion_defaults = deepcopy(inversion_defaults) - self._inversion_type = "fem" - self._validations = validations - self._tx_offsets = None - self._z_real_channel_bool = None - self._z_real_channel = None - self._z_real_uncertainty = None - self._z_imag_channel_bool = None - self._z_imag_channel = None - self._z_imag_uncertainty = None - self._model_type = "Conductivity (S/m)" - - super().__init__(input_file=input_file, forward_only=forward_only, **kwargs) - - def data_channel(self, component: str): - """Return uuid of data channel.""" - return getattr(self, "_".join([component, "channel"]), None) - - def uncertainty_channel(self, component: str): - """Return uuid of uncertainty channel.""" - return getattr(self, "_".join([component, "uncertainty"]), None) - - def property_group_data(self, property_group: UUID): - """ - Return dictionary of channel/data. + name: ClassVar[str] = "Frequency Domain Electromagnetics Forward" + title: ClassVar[str] = "Frequency-domain EM (FEM) Forward" + default_ui_json: ClassVar[Path] = assets_path() / "uijson/fem_forward.ui.json" - :param property_group: Property group uid - """ - channels = self.data_object.channels - if self.forward_only: - out = {k: None for k in channels} - else: - group = self.data_object.find_or_create_property_group( - name=property_group.name - ) - properties = [self.geoh5.get_entity(p)[0].values for p in group.properties] - out = {f: properties[i] for i, f in enumerate(channels)} + inversion_type: str = "fem" + physical_property: str = "conductivity" - return out + data_object: Receivers + z_real_channel_bool: bool + z_imag_channel_bool: bool + model_type: str = "Conductivity (S/m)" + + @property + def tx_offsets(self): + """Return transmitter offsets from frequency metadata""" + + try: + offset_data = self.data_object.metadata["EM Dataset"][ + "Frequency configurations" + ] + tx_offsets = {k["Frequency"]: k["Offset"] for k in offset_data} + + except KeyError as exception: + msg = "Metadata must contain 'Frequency configurations' dictionary with 'Offset' data." + raise KeyError(msg) from exception + + return tx_offsets @property def unit_conversion(self): @@ -82,6 +75,20 @@ def unit_conversion(self): } return conversion[self.data_object.unit] + @property + def channels(self) -> list[str]: + return ["_".join(k.split("_")[:2]) for k in self.__dict__ if "channel" in k] + + def property_group_data(self, property_group: PropertyGroup): + """ + Return dictionary of channel/data. + + :param property_group: Property group uid + """ + _ = property_group + channels = self.data_object.channels + return {k: None for k in channels} + def data(self, component: str): """Returns array of data for chosen data component.""" property_group = self.data_channel(component) @@ -92,73 +99,194 @@ def uncertainty(self, component: str) -> float: uid = self.uncertainty_channel(component) return self.property_group_data(uid) - @property - def model_type(self): - """Model units.""" - return self._model_type - @model_type.setter - def model_type(self, val): - self.setter_validator("model_type", val) +class FrequencyDomainElectromagneticsInversionParams(BaseInversionData): + """ + Parameter class for Frequency-domain Electromagnetic (FEM) -> conductivity inversion. + + :param z_real_channel: Real impedance channel. + :param z_real_uncertainty: Real impedance uncertainty channel. + :param z_imag_channel: Imaginary impedance channel. + :param z_imag_uncertainty: Imaginary impedance uncertainty channel. + :param model_type: Specify whether the models are provided in resistivity or conductivity. + """ + + name: ClassVar[str] = "Frequency Domain Electromagnetics Inversion" + title: ClassVar[str] = "Frequency-domain EM (FEM) Inversion" + default_ui_json: ClassVar[Path] = assets_path() / "uijson/fem_inversion.ui.json" + + inversion_type: str = "fem" + physical_property: str = "conductivity" + + data_object: Receivers + z_real_channel: PropertyGroup | None = None + z_real_uncertainty: PropertyGroup | None = None + z_imag_channel: PropertyGroup | None = None + z_imag_uncertainty: PropertyGroup | None = None + model_type: str = "Conductivity (S/m)" @property def tx_offsets(self): - if self._tx_offsets is None and self.data_object is not None: - try: - offset_data = self.data_object.metadata["EM Dataset"][ - "Frequency configurations" - ] - self._tx_offsets = {k["Frequency"]: k["Offset"] for k in offset_data} - except KeyError as exception: - msg = "Metadata must contain 'Frequency configurations' dictionary with 'Offset' data." - raise KeyError(msg) from exception - - return self._tx_offsets + """Return transmitter offsets from frequency metadata""" - @property - def z_real_channel_bool(self): - return self._z_real_channel_bool + try: + offset_data = self.data_object.metadata["EM Dataset"][ + "Frequency configurations" + ] + tx_offsets = {k["Frequency"]: k["Offset"] for k in offset_data} - @z_real_channel_bool.setter - def z_real_channel_bool(self, val): - self.setter_validator("z_real_channel_bool", val) + except KeyError as exception: + msg = "Metadata must contain 'Frequency configurations' dictionary with 'Offset' data." + raise KeyError(msg) from exception - @property - def z_real_channel(self): - return self._z_real_channel + return tx_offsets - @z_real_channel.setter - def z_real_channel(self, val): - self.setter_validator("z_real_channel", val, fun=self._uuid_promoter) + @property + def unit_conversion(self): + """Return time unit conversion factor.""" + conversion = { + "Hertz (Hz)": 1.0, + } + return conversion[self.data_object.unit] @property - def z_real_uncertainty(self): - return self._z_real_uncertainty + def channels(self) -> list[str]: + return ["_".join(k.split("_")[:2]) for k in self.__dict__ if "channel" in k] - @z_real_uncertainty.setter - def z_real_uncertainty(self, val): - self.setter_validator("z_real_uncertainty", val, fun=self._uuid_promoter) + def property_group_data(self, property_group: PropertyGroup): + """ + Return dictionary of channel/data. - @property - def z_imag_channel_bool(self): - return self._z_imag_channel_bool + :param property_group: Property group containing TEM data. + """ + channels = self.data_object.channels + group = self.data_object.fetch_property_group(name=property_group.name) + properties = [self.geoh5.get_entity(p)[0].values for p in group.properties] + out = {f: properties[i] for i, f in enumerate(channels)} - @z_imag_channel_bool.setter - def z_imag_channel_bool(self, val): - self.setter_validator("z_imag_channel_bool", val) + return out - @property - def z_imag_channel(self): - return self._z_imag_channel + def data(self, component: str): + """Returns array of data for chosen data component.""" + property_group = self.data_channel(component) + return self.property_group_data(property_group) - @z_imag_channel.setter - def z_imag_channel(self, val): - self.setter_validator("z_imag_channel", val, fun=self._uuid_promoter) + def uncertainty(self, component: str) -> float: + """Returns uncertainty for chosen data component.""" + uid = self.uncertainty_channel(component) + return self.property_group_data(uid) - @property - def z_imag_uncertainty(self): - return self._z_imag_uncertainty - @z_imag_uncertainty.setter - def z_imag_uncertainty(self, val): - self.setter_validator("z_imag_uncertainty", val, fun=self._uuid_promoter) +# class FrequencyDomainElectromagneticsParams(InversionBaseParams): +# """ +# Parameter class for Frequency-domain Electromagnetic (FEM) -> conductivity inversion. +# """ +# +# _physical_property = "conductivity" +# +# def __init__(self, input_file=None, forward_only=False, **kwargs): +# self._default_ui_json = deepcopy(default_ui_json) +# self._forward_defaults = deepcopy(forward_defaults) +# self._inversion_defaults = deepcopy(inversion_defaults) +# self._inversion_type = "fem" +# self._validations = validations +# self._tx_offsets = None +# self._z_real_channel_bool = None +# self._z_real_channel = None +# self._z_real_uncertainty = None +# self._z_imag_channel_bool = None +# self._z_imag_channel = None +# self._z_imag_uncertainty = None +# self._model_type = "Conductivity (S/m)" +# +# super().__init__(input_file=input_file, forward_only=forward_only, **kwargs) +# +# +# @property +# def unit_conversion(self): +# """Return time unit conversion factor.""" +# conversion = { +# "Hertz (Hz)": 1.0, +# } +# return conversion[self.data_object.unit] +# +# def data(self, component: str): +# """Returns array of data for chosen data component.""" +# property_group = self.data_channel(component) +# return self.property_group_data(property_group) +# +# def uncertainty(self, component: str) -> float: +# """Returns uncertainty for chosen data component.""" +# uid = self.uncertainty_channel(component) +# return self.property_group_data(uid) +# +# @property +# def model_type(self): +# """Model units.""" +# return self._model_type +# +# @model_type.setter +# def model_type(self, val): +# self.setter_validator("model_type", val) +# +# @property +# def tx_offsets(self): +# if self._tx_offsets is None and self.data_object is not None: +# try: +# offset_data = self.data_object.metadata["EM Dataset"][ +# "Frequency configurations" +# ] +# self._tx_offsets = {k["Frequency"]: k["Offset"] for k in offset_data} +# except KeyError as exception: +# msg = "Metadata must contain 'Frequency configurations' dictionary with 'Offset' data." +# raise KeyError(msg) from exception +# +# return self._tx_offsets +# +# @property +# def z_real_channel_bool(self): +# return self._z_real_channel_bool +# +# @z_real_channel_bool.setter +# def z_real_channel_bool(self, val): +# self.setter_validator("z_real_channel_bool", val) +# +# @property +# def z_real_channel(self): +# return self._z_real_channel +# +# @z_real_channel.setter +# def z_real_channel(self, val): +# self.setter_validator("z_real_channel", val, fun=self._uuid_promoter) +# +# @property +# def z_real_uncertainty(self): +# return self._z_real_uncertainty +# +# @z_real_uncertainty.setter +# def z_real_uncertainty(self, val): +# self.setter_validator("z_real_uncertainty", val, fun=self._uuid_promoter) +# +# @property +# def z_imag_channel_bool(self): +# return self._z_imag_channel_bool +# +# @z_imag_channel_bool.setter +# def z_imag_channel_bool(self, val): +# self.setter_validator("z_imag_channel_bool", val) +# +# @property +# def z_imag_channel(self): +# return self._z_imag_channel +# +# @z_imag_channel.setter +# def z_imag_channel(self, val): +# self.setter_validator("z_imag_channel", val, fun=self._uuid_promoter) +# +# @property +# def z_imag_uncertainty(self): +# return self._z_imag_uncertainty +# +# @z_imag_uncertainty.setter +# def z_imag_uncertainty(self, val): +# self.setter_validator("z_imag_uncertainty", val, fun=self._uuid_promoter) diff --git a/simpeg_drivers/electromagnetics/time_domain/__init__.py b/simpeg_drivers/electromagnetics/time_domain/__init__.py index 0649217f..2aa3b346 100644 --- a/simpeg_drivers/electromagnetics/time_domain/__init__.py +++ b/simpeg_drivers/electromagnetics/time_domain/__init__.py @@ -9,7 +9,10 @@ # ''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''' -from .params import TimeDomainElectromagneticsParams +from .params import ( + TimeDomainElectromagneticsForwardParams, + TimeDomainElectromagneticsInversionParams, +) # pylint: disable=unused-import # flake8: noqa diff --git a/simpeg_drivers/electromagnetics/time_domain/driver.py b/simpeg_drivers/electromagnetics/time_domain/driver.py index 4e5ecb6b..ec8e00ef 100644 --- a/simpeg_drivers/electromagnetics/time_domain/driver.py +++ b/simpeg_drivers/electromagnetics/time_domain/driver.py @@ -20,15 +20,72 @@ from simpeg_drivers.utils.utils import tile_locations from .constants import validations -from .params import TimeDomainElectromagneticsParams +from .params import ( + TimeDomainElectromagneticsForwardParams, + TimeDomainElectromagneticsInversionParams, +) -class TimeDomainElectromagneticsDriver(InversionDriver): - _params_class = TimeDomainElectromagneticsParams +class TimeDomainElectromagneticsForwardDriver(InversionDriver): + _params_class = TimeDomainElectromagneticsForwardParams _validations = validations - def __init__(self, params: TimeDomainElectromagneticsParams): - super().__init__(params) + def get_tiles(self) -> list[np.ndarray]: + """ + Special method to tile the data based on the transmitters center locations. + + First the transmitter locations are grouped into groups using kmeans clustering. + Second, if the number of groups is less than the number of 'tile_spatial' value, the groups are + further divided into groups based on the clustering of receiver locations. + """ + if not isinstance(self.params.data_object, LargeLoopGroundTEMReceivers): + return super().get_tiles() + + tx_ids = self.params.data_object.transmitters.tx_id_property.values + unique_tile_ids = np.unique(tx_ids) + n_groups = np.min([len(unique_tile_ids), self.params.tile_spatial]) + locations = [] + for uid in unique_tile_ids: + locations.append( + np.mean( + self.params.data_object.transmitters.vertices[tx_ids == uid], + axis=0, + ) + ) + + # Tile transmitters spatially by loop center + tx_tiles = tile_locations( + np.vstack(locations), + n_groups, + method="kmeans", + ) + receivers_tx_ids = self.params.data_object.tx_id_property.values + tiles = [] + for _t_id, group in enumerate(tx_tiles): + sub_group = [] + for value in group: + receiver_ind = receivers_tx_ids == unique_tile_ids[value] + sub_group.append(np.where(receiver_ind)[0]) + + tiles.append(np.hstack(sub_group)) + + # If number of tiles remaining, brake up receivers spatially per transmitter + while len(tiles) < self.params.tile_spatial: + largest_group = np.argmax([len(tile) for tile in tiles]) + tile = tiles.pop(largest_group) + new_tiles = tile_locations( + self.params.data_object.vertices[tile], + 2, + method="kmeans", + ) + tiles += [tile[new_tiles[0]], tile[new_tiles[1]]] + + return tiles + + +class TimeDomainElectromagneticsInversionDriver(InversionDriver): + _params_class = TimeDomainElectromagneticsInversionParams + _validations = validations def get_tiles(self) -> list[np.ndarray]: """ diff --git a/simpeg_drivers/electromagnetics/time_domain/params.py b/simpeg_drivers/electromagnetics/time_domain/params.py index 6affbc8d..e40fa94b 100644 --- a/simpeg_drivers/electromagnetics/time_domain/params.py +++ b/simpeg_drivers/electromagnetics/time_domain/params.py @@ -11,95 +11,50 @@ from __future__ import annotations -from copy import deepcopy -from uuid import UUID +from pathlib import Path +from typing import ClassVar, TypeAlias + +from geoh5py.groups import PropertyGroup +from geoh5py.objects import ( + AirborneTEMReceivers, + LargeLoopGroundTEMReceivers, + MovingLoopGroundTEMReceivers, +) + +from simpeg_drivers import assets_path +from simpeg_drivers.params import BaseForwardData, BaseInversionData -from simpeg_drivers.params import InversionBaseParams -from .constants import ( - default_ui_json, - forward_defaults, - inversion_defaults, - validations, +Receivers: TypeAlias = ( + MovingLoopGroundTEMReceivers | LargeLoopGroundTEMReceivers | AirborneTEMReceivers ) -class TimeDomainElectromagneticsParams(InversionBaseParams): +class TimeDomainElectromagneticsForwardParams(BaseForwardData): """ - Parameter class for Time-domain Electromagnetic (TEM) -> conductivity inversion. + Parameter class for Time-domain Electromagnetic (TEM) -> conductivity forward simulation. + + :param z_channel_bool: Z-component data channel boolean. + :param z_channel_uncertainty: Z-component data channel uncertainty. + :param x_channel_bool: X-component data channel boolean. + :param x_channel_uncertainty: X-component data channel uncertainty. + :param y_channel_bool: Y-component data channel boolean. + :param y_channel_uncertainty: Y-component data channel uncertainty. + :param data_units: Units for the TEM data """ - _physical_property = "conductivity" - - def __init__(self, input_file=None, forward_only=False, **kwargs): - self._default_ui_json = deepcopy(default_ui_json) - self._forward_defaults = deepcopy(forward_defaults) - self._inversion_defaults = deepcopy(inversion_defaults) - self._inversion_type = "tdem" - self._validations = validations - self._data_units = "dB/dt (T/s)" - self._z_channel_bool = None - self._z_channel = None - self._z_uncertainty = None - self._x_channel_bool = None - self._x_channel = None - self._x_uncertainty = None - self._y_channel_bool = None - self._y_channel = None - self._y_uncertainty = None - - self._model_type = "Conductivity (S/m)" - - super().__init__(input_file=input_file, forward_only=forward_only, **kwargs) - - def data_channel(self, component: str): - """Return uuid of data channel.""" - return getattr(self, "_".join([component, "channel"]), None) - - @property - def data_units(self): - return self._data_units - - @data_units.setter - def data_units(self, val): - self.setter_validator("data_units", val) + name: ClassVar[str] = "Time Domain Electromagnetics Forward" + title: ClassVar[str] = "Time-domain EM (TEM) Forward" + default_ui_json: ClassVar[Path] = assets_path() / "uijson/tdem_forward.ui.json" - def uncertainty_channel(self, component: str): - """Return uuid of uncertainty channel.""" - return getattr(self, "_".join([component, "uncertainty"]), None) + inversion_type: str = "tdem" + physical_property: str = "conductivity" - def property_group_data(self, property_group: UUID): - data = {} - channels = self.data_object.channels - if self.forward_only: - return {k: None for k in channels} - else: - group = self.data_object.find_or_create_property_group( - name=property_group.name - ) - property_names = [ - self.geoh5.get_entity(p)[0].name for p in group.properties - ] - properties = [self.geoh5.get_entity(p)[0].values for p in group.properties] - for i, f in enumerate(channels): - try: - f_ind = property_names.index( - next(k for k in property_names if f"{f:.2e}" in k) - ) # Safer if data was saved with geoapps naming convention - data[f] = properties[f_ind] - except StopIteration: - data[f] = properties[i] # in case of other naming conventions - - return data - - @property - def model_type(self): - """Model units.""" - return self._model_type - - @model_type.setter - def model_type(self, val): - self.setter_validator("model_type", val) + data_object: Receivers + z_channel_bool: bool + x_channel_bool: bool + data_units: str = "dB/dt (T/s)" + model_type: str = "Conductivity (S/m)" @property def unit_conversion(self): @@ -111,6 +66,16 @@ def unit_conversion(self): } return conversion[self.data_object.unit] + def property_group_data(self, property_group: PropertyGroup): + """ + Return dictionary of channel/data. + + :param property_group: Property group containing em data. + """ + _ = property_group + channels = self.data_object.channels + return {k: None for k in channels} + def data(self, component: str): """Returns array of data for chosen data component.""" property_group = self.data_channel(component) @@ -121,74 +86,246 @@ def uncertainty(self, component: str) -> float: uid = self.uncertainty_channel(component) return self.property_group_data(uid) - @property - def z_channel_bool(self): - return self._z_channel_bool - - @z_channel_bool.setter - def z_channel_bool(self, val): - self.setter_validator("z_channel_bool", val) - - @property - def z_channel(self): - return self._z_channel - - @z_channel.setter - def z_channel(self, val): - self.setter_validator("z_channel", val, fun=self._uuid_promoter) - - @property - def z_uncertainty(self): - return self._z_uncertainty - @z_uncertainty.setter - def z_uncertainty(self, val): - self.setter_validator("z_uncertainty", val, fun=self._uuid_promoter) +class TimeDomainElectromagneticsInversionParams(BaseInversionData): + """ + Parameter class for Time-domain Electromagnetic (TEM) -> conductivity inversion. - @property - def x_channel_bool(self): - return self._x_channel_bool + :param z_channel: Z-component data channel. + :param z_uncertainty: Z-component data channel uncertainty. + :param x_channel: X-component data channel. + :param x_uncertainty: X-component data channel uncertainty. + :param y_channel: Y-component data channel. + :param y_uncertainty: Y-component data channel uncertainty. + :param model_type: Specify whether the models are provided in resistivity or conductivity. + :param data_units: Units for the TEM data + """ - @x_channel_bool.setter - def x_channel_bool(self, val): - self.setter_validator("x_channel_bool", val) + name: ClassVar[str] = "Time Domain Electromagnetics Inversion" + title: ClassVar[str] = "Time-domain EM (TEM) Inversion" + default_ui_json: ClassVar[Path] = assets_path() / "uijson/tdem_inversion.ui.json" - @property - def x_channel(self): - return self._x_channel + inversion_type: str = "tdem" + physical_property: str = "conductivity" - @x_channel.setter - def x_channel(self, val): - self.setter_validator("x_channel", val, fun=self._uuid_promoter) + data_object: Receivers + z_channel: PropertyGroup | None = None + z_uncertainty: PropertyGroup | None = None + x_channel: PropertyGroup | None = None + x_uncertainty: PropertyGroup | None = None + y_channel: PropertyGroup | None = None + y_uncertainty: PropertyGroup | None = None + data_units: str = "dB/dt (T/s)" + model_type: str = "Conductivity (S/m)" @property - def x_uncertainty(self): - return self._x_uncertainty - - @x_uncertainty.setter - def x_uncertainty(self, val): - self.setter_validator("x_uncertainty", val, fun=self._uuid_promoter) + def unit_conversion(self): + """Return time unit conversion factor.""" + conversion = { + "Seconds (s)": 1.0, + "Milliseconds (ms)": 1e-3, + "Microseconds (us)": 1e-6, + } + return conversion[self.data_object.unit] - @property - def y_channel_bool(self): - return self._y_channel_bool + def property_group_data(self, property_group: PropertyGroup): + """ + Return dictionary of channel/data. - @y_channel_bool.setter - def y_channel_bool(self, val): - self.setter_validator("y_channel_bool", val) + :param property_group: Property group containing TEM data. + """ + data = {} + channels = self.data_object.channels + group = self.data_object.fetch_property_group(name=property_group.name) + property_names = [self.geoh5.get_entity(p)[0].name for p in group.properties] + properties = [self.geoh5.get_entity(p)[0].values for p in group.properties] + for i, f in enumerate(channels): + try: + f_ind = property_names.index( + next(k for k in property_names if f"{f:.2e}" in k) + ) # Safer if data was saved with geoapps naming convention + data[f] = properties[f_ind] + except StopIteration: + data[f] = properties[i] # in case of other naming conventions + + return data - @property - def y_channel(self): - return self._y_channel + def data(self, component: str): + """Returns array of data for chosen data component.""" + property_group = self.data_channel(component) + return self.property_group_data(property_group) - @y_channel.setter - def y_channel(self, val): - self.setter_validator("y_channel", val, fun=self._uuid_promoter) + def uncertainty(self, component: str) -> float: + """Returns uncertainty for chosen data component.""" + uid = self.uncertainty_channel(component) + return self.property_group_data(uid) - @property - def y_uncertainty(self): - return self._y_uncertainty - @y_uncertainty.setter - def y_uncertainty(self, val): - self.setter_validator("y_uncertainty", val, fun=self._uuid_promoter) +# class TimeDomainElectromagneticsParams(InversionBaseParams): +# """ +# Parameter class for Time-domain Electromagnetic (TEM) -> conductivity inversion. +# """ +# +# _physical_property = "conductivity" +# +# def __init__(self, input_file=None, forward_only=False, **kwargs): +# self._default_ui_json = deepcopy(default_ui_json) +# self._forward_defaults = deepcopy(forward_defaults) +# self._inversion_defaults = deepcopy(inversion_defaults) +# self._inversion_type = "tdem" +# self._validations = validations +# self._data_units = "dB/dt (T/s)" +# self._z_channel_bool = None +# self._z_channel = None +# self._z_uncertainty = None +# self._x_channel_bool = None +# self._x_channel = None +# self._x_uncertainty = None +# self._y_channel_bool = None +# self._y_channel = None +# self._y_uncertainty = None +# +# self._model_type = "Conductivity (S/m)" +# +# super().__init__(input_file=input_file, forward_only=forward_only, **kwargs) +# +# def data_channel(self, component: str): +# """Return uuid of data channel.""" +# return getattr(self, "_".join([component, "channel"]), None) +# +# @property +# def data_units(self): +# return self._data_units +# +# @data_units.setter +# def data_units(self, val): +# self.setter_validator("data_units", val) +# +# def uncertainty_channel(self, component: str): +# """Return uuid of uncertainty channel.""" +# return getattr(self, "_".join([component, "uncertainty"]), None) +# +# def property_group_data(self, property_group: UUID): +# data = {} +# channels = self.data_object.channels +# if self.forward_only: +# return {k: None for k in channels} +# else: +# group = self.data_object.find_or_create_property_group( +# name=property_group.name +# ) +# property_names = [ +# self.geoh5.get_entity(p)[0].name for p in group.properties +# ] +# properties = [self.geoh5.get_entity(p)[0].values for p in group.properties] +# for i, f in enumerate(channels): +# try: +# f_ind = property_names.index( +# next(k for k in property_names if f"{f:.2e}" in k) +# ) # Safer if data was saved with geoapps naming convention +# data[f] = properties[f_ind] +# except StopIteration: +# data[f] = properties[i] # in case of other naming conventions +# +# return data +# +# @property +# def model_type(self): +# """Model units.""" +# return self._model_type +# +# @model_type.setter +# def model_type(self, val): +# self.setter_validator("model_type", val) +# +# @property +# def unit_conversion(self): +# """Return time unit conversion factor.""" +# conversion = { +# "Seconds (s)": 1.0, +# "Milliseconds (ms)": 1e-3, +# "Microseconds (us)": 1e-6, +# } +# return conversion[self.data_object.unit] +# +# def data(self, component: str): +# """Returns array of data for chosen data component.""" +# property_group = self.data_channel(component) +# return self.property_group_data(property_group) +# +# def uncertainty(self, component: str) -> float: +# """Returns uncertainty for chosen data component.""" +# uid = self.uncertainty_channel(component) +# return self.property_group_data(uid) +# +# @property +# def z_channel_bool(self): +# return self._z_channel_bool +# +# @z_channel_bool.setter +# def z_channel_bool(self, val): +# self.setter_validator("z_channel_bool", val) +# +# @property +# def z_channel(self): +# return self._z_channel +# +# @z_channel.setter +# def z_channel(self, val): +# self.setter_validator("z_channel", val, fun=self._uuid_promoter) +# +# @property +# def z_uncertainty(self): +# return self._z_uncertainty +# +# @z_uncertainty.setter +# def z_uncertainty(self, val): +# self.setter_validator("z_uncertainty", val, fun=self._uuid_promoter) +# +# @property +# def x_channel_bool(self): +# return self._x_channel_bool +# +# @x_channel_bool.setter +# def x_channel_bool(self, val): +# self.setter_validator("x_channel_bool", val) +# +# @property +# def x_channel(self): +# return self._x_channel +# +# @x_channel.setter +# def x_channel(self, val): +# self.setter_validator("x_channel", val, fun=self._uuid_promoter) +# +# @property +# def x_uncertainty(self): +# return self._x_uncertainty +# +# @x_uncertainty.setter +# def x_uncertainty(self, val): +# self.setter_validator("x_uncertainty", val, fun=self._uuid_promoter) +# +# @property +# def y_channel_bool(self): +# return self._y_channel_bool +# +# @y_channel_bool.setter +# def y_channel_bool(self, val): +# self.setter_validator("y_channel_bool", val) +# +# @property +# def y_channel(self): +# return self._y_channel +# +# @y_channel.setter +# def y_channel(self, val): +# self.setter_validator("y_channel", val, fun=self._uuid_promoter) +# +# @property +# def y_uncertainty(self): +# return self._y_uncertainty +# +# @y_uncertainty.setter +# def y_uncertainty(self, val): +# self.setter_validator("y_uncertainty", val, fun=self._uuid_promoter) diff --git a/simpeg_drivers/params.py b/simpeg_drivers/params.py index 7e5d9c99..9237ae77 100644 --- a/simpeg_drivers/params.py +++ b/simpeg_drivers/params.py @@ -298,12 +298,12 @@ class BaseInversionData(CoreData): lower_bound: float | FloatData | None = None upper_bound: float | FloatData | None = None - alpha_s: float | FloatData = 1.0 + alpha_s: float | FloatData | None = 1.0 length_scale_x: float | FloatData = 1.0 length_scale_y: float | FloatData = 1.0 length_scale_z: float | FloatData = 1.0 - s_norm: float | FloatData = 0.0 + s_norm: float | FloatData | None = 0.0 x_norm: float | FloatData = 2.0 y_norm: float | FloatData = 2.0 z_norm: float | FloatData = 2.0 diff --git a/tests/run_tests/driver_airborne_tem_test.py b/tests/run_tests/driver_airborne_tem_test.py index 2bfb0937..d51a8b5b 100644 --- a/tests/run_tests/driver_airborne_tem_test.py +++ b/tests/run_tests/driver_airborne_tem_test.py @@ -17,10 +17,15 @@ from geoh5py.workspace import Workspace from pytest import raises -from simpeg_drivers.electromagnetics.time_domain import TimeDomainElectromagneticsParams +from simpeg_drivers.electromagnetics.time_domain import ( + TimeDomainElectromagneticsForwardParams, + TimeDomainElectromagneticsInversionParams, +) from simpeg_drivers.electromagnetics.time_domain.driver import ( - TimeDomainElectromagneticsDriver, + TimeDomainElectromagneticsForwardDriver, + TimeDomainElectromagneticsInversionDriver, ) +from simpeg_drivers.params import ActiveCellsData from simpeg_drivers.utils.testing import check_target, setup_inversion_workspace from simpeg_drivers.utils.utils import get_inversion_output @@ -46,21 +51,19 @@ def test_bad_waveform(tmp_path: Path): padding_distance=400.0, flatten=False, ) - params = TimeDomainElectromagneticsParams( - forward_only=True, + params = TimeDomainElectromagneticsForwardParams( geoh5=geoh5, - mesh=model.parent.uid, - topography_object=topography.uid, - resolution=0.0, + mesh=model.parent, + active_cells=ActiveCellsData(topography_object=topography), z_from_topo=False, - data_object=survey.uid, - starting_model=model.uid, + data_object=survey, + starting_model=model, x_channel_bool=True, y_channel_bool=True, z_channel_bool=True, ) - params.workpath = tmp_path - fwr_driver = TimeDomainElectromagneticsDriver(params) + + fwr_driver = TimeDomainElectromagneticsForwardDriver(params) survey.channels[-1] = 1000.0 @@ -88,21 +91,19 @@ def test_airborne_tem_fwr_run( padding_distance=400.0, flatten=False, ) - params = TimeDomainElectromagneticsParams( - forward_only=True, + params = TimeDomainElectromagneticsForwardParams( geoh5=geoh5, - mesh=model.parent.uid, - topography_object=topography.uid, - resolution=0.0, + mesh=model.parent, + active_cells=ActiveCellsData(topography_object=topography), z_from_topo=False, - data_object=survey.uid, - starting_model=model.uid, + data_object=survey, + starting_model=model, x_channel_bool=True, y_channel_bool=True, z_channel_bool=True, ) - params.workpath = tmp_path - fwr_driver = TimeDomainElectromagneticsDriver(params) + + fwr_driver = TimeDomainElectromagneticsForwardDriver(params) fwr_driver.run() @@ -162,12 +163,11 @@ def test_airborne_tem_run(tmp_path: Path, max_iterations=1, pytest=True): orig_dBzdt = geoh5.get_entity("Iteration_0_z_[0]")[0].values # Run the inverse - params = TimeDomainElectromagneticsParams( + params = TimeDomainElectromagneticsInversionParams( geoh5=geoh5, - mesh=mesh.uid, - topography_object=topography.uid, - resolution=0.0, - data_object=survey.uid, + mesh=mesh, + active_cells=ActiveCellsData(topography_object=topography), + data_object=survey, starting_model=1e-3, reference_model=1e-3, chi_factor=1.0, @@ -188,9 +188,11 @@ def test_airborne_tem_run(tmp_path: Path, max_iterations=1, pytest=True): store_sensitivities="ram", **data_kwargs, ) - params.write_input_file(path=tmp_path, name="Inv_run") + params.write_ui_json(path=tmp_path / "Inv_run.ui.json") - driver = TimeDomainElectromagneticsDriver.start(str(tmp_path / "Inv_run.ui.json")) + driver = TimeDomainElectromagneticsInversionDriver.start( + str(tmp_path / "Inv_run.ui.json") + ) with geoh5.open() as run_ws: output = get_inversion_output( diff --git a/tests/run_tests/driver_fem_test.py b/tests/run_tests/driver_fem_test.py index c7a0a32e..9a32eb88 100644 --- a/tests/run_tests/driver_fem_test.py +++ b/tests/run_tests/driver_fem_test.py @@ -19,11 +19,14 @@ from geoh5py.groups import SimPEGGroup from simpeg_drivers.electromagnetics.frequency_domain.driver import ( - FrequencyDomainElectromagneticsDriver, + FrequencyDomainElectromagneticsForwardDriver, + FrequencyDomainElectromagneticsInversionDriver, ) from simpeg_drivers.electromagnetics.frequency_domain.params import ( - FrequencyDomainElectromagneticsParams, + FrequencyDomainElectromagneticsForwardParams, + FrequencyDomainElectromagneticsInversionParams, ) +from simpeg_drivers.params import ActiveCellsData from simpeg_drivers.utils.testing import check_target, setup_inversion_workspace from simpeg_drivers.utils.utils import get_inversion_output @@ -54,20 +57,18 @@ def test_fem_fwr_run( inversion_type="fem", flatten=True, ) - params = FrequencyDomainElectromagneticsParams( - forward_only=True, + params = FrequencyDomainElectromagneticsForwardParams( geoh5=geoh5, - mesh=model.parent.uid, - topography_object=topography.uid, - resolution=0.0, + mesh=model.parent, + active_cells=ActiveCellsData(topography_object=topography), z_from_topo=False, - data_object=survey.uid, - starting_model=model.uid, + data_object=survey, + starting_model=model, z_real_channel_bool=True, z_imag_channel_bool=True, ) - params.workpath = tmp_path - fwr_driver = FrequencyDomainElectromagneticsDriver(params) + + fwr_driver = FrequencyDomainElectromagneticsForwardDriver(params) fwr_driver.run() geoh5.close() @@ -121,18 +122,17 @@ def test_fem_run(tmp_path: Path, max_iterations=1, pytest=True): for comp, data_group, uncert_group in zip( components, data_groups, uncert_groups, strict=True ): - data_kwargs[f"{comp}_channel"] = data_group.uid - data_kwargs[f"{comp}_uncertainty"] = uncert_group.uid + data_kwargs[f"{comp}_channel"] = data_group + data_kwargs[f"{comp}_uncertainty"] = uncert_group orig_z_real_1 = geoh5.get_entity("Iteration_0_z_real_[0]")[0].values # Run the inverse - params = FrequencyDomainElectromagneticsParams( + params = FrequencyDomainElectromagneticsInversionParams( geoh5=geoh5, - mesh=mesh.uid, - topography_object=topography.uid, - resolution=0.0, - data_object=survey.uid, + mesh=mesh, + active_cells=ActiveCellsData(topography_object=topography), + data_object=survey, starting_model=1e-3, reference_model=1e-3, alpha_s=0.0, @@ -152,8 +152,8 @@ def test_fem_run(tmp_path: Path, max_iterations=1, pytest=True): sens_wts_threshold=1.0, **data_kwargs, ) - params.write_input_file(path=tmp_path, name="Inv_run") - driver = FrequencyDomainElectromagneticsDriver(params) + params.write_ui_json(path=tmp_path / "Inv_run.ui.json") + driver = FrequencyDomainElectromagneticsInversionDriver(params) driver.run() with geoh5.open() as run_ws: From 510b8ff63c319681593f5435b56f6b4126ce4842 Mon Sep 17 00:00:00 2001 From: benjamink Date: Mon, 10 Feb 2025 15:25:03 -0800 Subject: [PATCH 02/10] runtests running --- tests/run_tests/driver_airborne_tem_test.py | 1 + tests/run_tests/driver_ground_tem_test.py | 58 +++++++++++---------- 2 files changed, 31 insertions(+), 28 deletions(-) diff --git a/tests/run_tests/driver_airborne_tem_test.py b/tests/run_tests/driver_airborne_tem_test.py index d51a8b5b..13b47ff6 100644 --- a/tests/run_tests/driver_airborne_tem_test.py +++ b/tests/run_tests/driver_airborne_tem_test.py @@ -185,6 +185,7 @@ def test_airborne_tem_run(tmp_path: Path, max_iterations=1, pytest=True): coolingRate=4, max_cg_iterations=200, prctile=5, + sens_wts_threshold=1.0, store_sensitivities="ram", **data_kwargs, ) diff --git a/tests/run_tests/driver_ground_tem_test.py b/tests/run_tests/driver_ground_tem_test.py index 0a95d65b..f2d67193 100644 --- a/tests/run_tests/driver_ground_tem_test.py +++ b/tests/run_tests/driver_ground_tem_test.py @@ -16,10 +16,15 @@ import numpy as np from geoh5py.workspace import Workspace -from simpeg_drivers.electromagnetics.time_domain import TimeDomainElectromagneticsParams +from simpeg_drivers.electromagnetics.time_domain import ( + TimeDomainElectromagneticsForwardParams, + TimeDomainElectromagneticsInversionParams, +) from simpeg_drivers.electromagnetics.time_domain.driver import ( - TimeDomainElectromagneticsDriver, + TimeDomainElectromagneticsForwardDriver, + TimeDomainElectromagneticsInversionDriver, ) +from simpeg_drivers.params import ActiveCellsData from simpeg_drivers.utils.testing import check_target, setup_inversion_workspace from simpeg_drivers.utils.utils import get_inversion_output @@ -54,21 +59,19 @@ def test_tiling_ground_tem( flatten=True, ) - params = TimeDomainElectromagneticsParams( - forward_only=True, + params = TimeDomainElectromagneticsForwardParams( geoh5=geoh5, - mesh=model.parent.uid, - topography_object=topography.uid, - resolution=0.0, + mesh=model.parent, + active_cells=ActiveCellsData(topography_object=topography), z_from_topo=False, - data_object=survey.uid, - starting_model=model.uid, + data_object=survey, + starting_model=model, x_channel_bool=True, y_channel_bool=True, z_channel_bool=True, tile_spatial=4, ) - fwr_driver = TimeDomainElectromagneticsDriver(params) + fwr_driver = TimeDomainElectromagneticsForwardDriver(params) tiles = fwr_driver.get_tiles() @@ -104,21 +107,19 @@ def test_ground_tem_fwr_run( padding_distance=1000.0, flatten=True, ) - params = TimeDomainElectromagneticsParams( - forward_only=True, + params = TimeDomainElectromagneticsForwardParams( geoh5=geoh5, - mesh=model.parent.uid, - topography_object=topography.uid, - resolution=0.0, + mesh=model.parent, + active_cells=ActiveCellsData(topography_object=topography), z_from_topo=False, - data_object=survey.uid, - starting_model=model.uid, + data_object=survey, + starting_model=model, x_channel_bool=True, y_channel_bool=True, z_channel_bool=True, ) - params.workpath = tmp_path - fwr_driver = TimeDomainElectromagneticsDriver(params) + + fwr_driver = TimeDomainElectromagneticsForwardDriver(params) survey.transmitters.remove_cells([15]) survey.tx_id_property.name = "tx_id" @@ -143,7 +144,7 @@ def test_ground_tem_run(tmp_path: Path, max_iterations=1, pytest=True): ) with Workspace(workpath) as geoh5: - simpeg_group = geoh5.get_entity("Tdem Forward")[0] + simpeg_group = geoh5.get_entity("Time-domain EM (TEM) Forward")[0] survey = simpeg_group.get_entity("Ground TEM Rx")[0] mesh = geoh5.get_entity("mesh")[0] topography = geoh5.get_entity("topography")[0] @@ -189,12 +190,11 @@ def test_ground_tem_run(tmp_path: Path, max_iterations=1, pytest=True): orig_dBzdt = geoh5.get_entity("Iteration_0_z_[0]")[0].values # Run the inverse - params = TimeDomainElectromagneticsParams( + params = TimeDomainElectromagneticsInversionParams( geoh5=geoh5, - mesh=mesh.uid, - topography_object=topography.uid, - resolution=0.0, - data_object=survey.uid, + mesh=mesh, + active_cells=ActiveCellsData(topography_object=topography), + data_object=survey, starting_model=1e-3, reference_model=1e-3, chi_factor=0.1, @@ -212,13 +212,15 @@ def test_ground_tem_run(tmp_path: Path, max_iterations=1, pytest=True): coolingRate=2, max_cg_iterations=200, prctile=100, - # sens_wts_threshold=1., + sens_wts_threshold=1.0, store_sensitivities="ram", **data_kwargs, ) - params.write_input_file(path=tmp_path, name="Inv_run") + params.write_ui_json(path=tmp_path / "Inv_run.ui.json") - driver = TimeDomainElectromagneticsDriver.start(str(tmp_path / "Inv_run.ui.json")) + driver = TimeDomainElectromagneticsInversionDriver.start( + str(tmp_path / "Inv_run.ui.json") + ) with geoh5.open() as run_ws: output = get_inversion_output( From f7788a48fa18b3ec684bef929260ebdfb7b02af6 Mon Sep 17 00:00:00 2001 From: benjamink Date: Tue, 11 Feb 2025 08:16:24 -0800 Subject: [PATCH 03/10] remove em from write_default_uijson --- simpeg_drivers/utils/write_default_uijson.py | 36 -------------------- 1 file changed, 36 deletions(-) diff --git a/simpeg_drivers/utils/write_default_uijson.py b/simpeg_drivers/utils/write_default_uijson.py index c97e2aa1..fc02d7c7 100644 --- a/simpeg_drivers/utils/write_default_uijson.py +++ b/simpeg_drivers/utils/write_default_uijson.py @@ -20,10 +20,6 @@ from simpeg_drivers.electricals.induced_polarization.pseudo_three_dimensions.params import ( InducedPolarizationPseudo3DParams, ) -from simpeg_drivers.electromagnetics.frequency_domain import ( - FrequencyDomainElectromagneticsParams, -) -from simpeg_drivers.electromagnetics.time_domain import TimeDomainElectromagneticsParams from simpeg_drivers.joint.joint_cross_gradient import JointCrossGradientParams from simpeg_drivers.joint.joint_surveys import JointSurveysParams from simpeg_drivers.natural_sources import MagnetotelluricsParams, TipperParams @@ -52,38 +48,6 @@ def write_default_uijson(path: str | Path): filedict = { - "direct_current_inversion_pseudo3d.ui.json": DirectCurrentPseudo3DParams( - validate=False - ), - "direct_current_forward_pseudo3d.ui.json": DirectCurrentPseudo3DParams( - forward_only=True, validate=False - ), - "induced_polarization_inversion_pseudo3d.ui.json": InducedPolarizationPseudo3DParams( - validate=False - ), - "induced_polarization_forward_pseudo3d.ui.json": InducedPolarizationPseudo3DParams( - forward_only=True, validate=False - ), - "fem_inversion.ui.json": FrequencyDomainElectromagneticsParams( - forward_only=False, validate=False - ), - "fem_forward.ui.json": FrequencyDomainElectromagneticsParams( - forward_only=True, validate=False - ), - "tdem_inversion.ui.json": TimeDomainElectromagneticsParams( - forward_only=False, validate=False - ), - "tdem_forward.ui.json": TimeDomainElectromagneticsParams( - forward_only=True, validate=False - ), - "magnetotellurics_inversion.ui.json": MagnetotelluricsParams( - forward_only=False, validate=False - ), - "magnetotellurics_forward.ui.json": MagnetotelluricsParams( - forward_only=True, validate=False - ), - "tipper_inversion.ui.json": TipperParams(forward_only=False, validate=False), - "tipper_forward.ui.json": TipperParams(forward_only=True, validate=False), "joint_surveys_inversion.ui.json": JointSurveysParams( forward_only=False, validate=False ), From ed58960685922d0f12dc5f1bdd728f0258abcde5 Mon Sep 17 00:00:00 2001 From: benjamink Date: Tue, 11 Feb 2025 08:17:18 -0800 Subject: [PATCH 04/10] finish the job (remove imports) --- simpeg_drivers/utils/write_default_uijson.py | 7 ------- 1 file changed, 7 deletions(-) diff --git a/simpeg_drivers/utils/write_default_uijson.py b/simpeg_drivers/utils/write_default_uijson.py index fc02d7c7..b442fb83 100644 --- a/simpeg_drivers/utils/write_default_uijson.py +++ b/simpeg_drivers/utils/write_default_uijson.py @@ -14,15 +14,8 @@ import argparse from pathlib import Path -from simpeg_drivers.electricals.direct_current.pseudo_three_dimensions.params import ( - DirectCurrentPseudo3DParams, -) -from simpeg_drivers.electricals.induced_polarization.pseudo_three_dimensions.params import ( - InducedPolarizationPseudo3DParams, -) from simpeg_drivers.joint.joint_cross_gradient import JointCrossGradientParams from simpeg_drivers.joint.joint_surveys import JointSurveysParams -from simpeg_drivers.natural_sources import MagnetotelluricsParams, TipperParams active_data_channels = [ From a4760e8ad05b9ced43d0aefbdd50a68433d27672 Mon Sep 17 00:00:00 2001 From: benjamink Date: Tue, 11 Feb 2025 10:58:32 -0800 Subject: [PATCH 05/10] Update natural sources --- simpeg_drivers/__init__.py | 9 +- .../frequency_domain/params.py | 118 +---- .../electromagnetics/time_domain/params.py | 175 +------- simpeg_drivers/natural_sources/__init__.py | 7 +- .../magnetotellurics/__init__.py | 2 +- .../magnetotellurics/driver.py | 12 +- .../magnetotellurics/params.py | 418 ++++++------------ .../natural_sources/tipper/__init__.py | 2 +- .../natural_sources/tipper/driver.py | 12 +- .../natural_sources/tipper/params.py | 276 +++++------- tests/run_tests/driver_mt_test.py | 56 +-- tests/run_tests/driver_tipper_test.py | 44 +- 12 files changed, 335 insertions(+), 796 deletions(-) diff --git a/simpeg_drivers/__init__.py b/simpeg_drivers/__init__.py index de7d9e6f..0c5a9198 100644 --- a/simpeg_drivers/__init__.py +++ b/simpeg_drivers/__init__.py @@ -109,15 +109,18 @@ def assets_path() -> Path: ), "magnetotellurics": ( "simpeg_drivers.natural_sources.magnetotellurics.driver", - {"inversion": "MagnetotelluricsDriver"}, + { + "forward": "MagnetotelluricsForwardDriver", + "inversion": "MagnetotelluricsInversionDriver", + }, ), "tipper": ( "simpeg_drivers.natural_sources.tipper.driver", - {"inversion": "TipperDriver"}, + {"forward": "TipperForwardDriver", "inversion": "TipperInversionDriver"}, ), "gravity": ( "simpeg_drivers.potential_fields.gravity.driver", - {"inversion": "GravityInversionDriver", "forward": "GravityForwardDriver"}, + {"forward": "GravityForwardDriver", "inversion": "GravityInversionDriver"}, ), "magnetic scalar": ( "simpeg_drivers.potential_fields.magnetic_scalar.driver", diff --git a/simpeg_drivers/electromagnetics/frequency_domain/params.py b/simpeg_drivers/electromagnetics/frequency_domain/params.py index 32cbd194..0f5209bb 100644 --- a/simpeg_drivers/electromagnetics/frequency_domain/params.py +++ b/simpeg_drivers/electromagnetics/frequency_domain/params.py @@ -162,9 +162,8 @@ def property_group_data(self, property_group: PropertyGroup): channels = self.data_object.channels group = self.data_object.fetch_property_group(name=property_group.name) properties = [self.geoh5.get_entity(p)[0].values for p in group.properties] - out = {f: properties[i] for i, f in enumerate(channels)} - return out + return {f: properties[i] for i, f in enumerate(channels)} def data(self, component: str): """Returns array of data for chosen data component.""" @@ -175,118 +174,3 @@ def uncertainty(self, component: str) -> float: """Returns uncertainty for chosen data component.""" uid = self.uncertainty_channel(component) return self.property_group_data(uid) - - -# class FrequencyDomainElectromagneticsParams(InversionBaseParams): -# """ -# Parameter class for Frequency-domain Electromagnetic (FEM) -> conductivity inversion. -# """ -# -# _physical_property = "conductivity" -# -# def __init__(self, input_file=None, forward_only=False, **kwargs): -# self._default_ui_json = deepcopy(default_ui_json) -# self._forward_defaults = deepcopy(forward_defaults) -# self._inversion_defaults = deepcopy(inversion_defaults) -# self._inversion_type = "fem" -# self._validations = validations -# self._tx_offsets = None -# self._z_real_channel_bool = None -# self._z_real_channel = None -# self._z_real_uncertainty = None -# self._z_imag_channel_bool = None -# self._z_imag_channel = None -# self._z_imag_uncertainty = None -# self._model_type = "Conductivity (S/m)" -# -# super().__init__(input_file=input_file, forward_only=forward_only, **kwargs) -# -# -# @property -# def unit_conversion(self): -# """Return time unit conversion factor.""" -# conversion = { -# "Hertz (Hz)": 1.0, -# } -# return conversion[self.data_object.unit] -# -# def data(self, component: str): -# """Returns array of data for chosen data component.""" -# property_group = self.data_channel(component) -# return self.property_group_data(property_group) -# -# def uncertainty(self, component: str) -> float: -# """Returns uncertainty for chosen data component.""" -# uid = self.uncertainty_channel(component) -# return self.property_group_data(uid) -# -# @property -# def model_type(self): -# """Model units.""" -# return self._model_type -# -# @model_type.setter -# def model_type(self, val): -# self.setter_validator("model_type", val) -# -# @property -# def tx_offsets(self): -# if self._tx_offsets is None and self.data_object is not None: -# try: -# offset_data = self.data_object.metadata["EM Dataset"][ -# "Frequency configurations" -# ] -# self._tx_offsets = {k["Frequency"]: k["Offset"] for k in offset_data} -# except KeyError as exception: -# msg = "Metadata must contain 'Frequency configurations' dictionary with 'Offset' data." -# raise KeyError(msg) from exception -# -# return self._tx_offsets -# -# @property -# def z_real_channel_bool(self): -# return self._z_real_channel_bool -# -# @z_real_channel_bool.setter -# def z_real_channel_bool(self, val): -# self.setter_validator("z_real_channel_bool", val) -# -# @property -# def z_real_channel(self): -# return self._z_real_channel -# -# @z_real_channel.setter -# def z_real_channel(self, val): -# self.setter_validator("z_real_channel", val, fun=self._uuid_promoter) -# -# @property -# def z_real_uncertainty(self): -# return self._z_real_uncertainty -# -# @z_real_uncertainty.setter -# def z_real_uncertainty(self, val): -# self.setter_validator("z_real_uncertainty", val, fun=self._uuid_promoter) -# -# @property -# def z_imag_channel_bool(self): -# return self._z_imag_channel_bool -# -# @z_imag_channel_bool.setter -# def z_imag_channel_bool(self, val): -# self.setter_validator("z_imag_channel_bool", val) -# -# @property -# def z_imag_channel(self): -# return self._z_imag_channel -# -# @z_imag_channel.setter -# def z_imag_channel(self, val): -# self.setter_validator("z_imag_channel", val, fun=self._uuid_promoter) -# -# @property -# def z_imag_uncertainty(self): -# return self._z_imag_uncertainty -# -# @z_imag_uncertainty.setter -# def z_imag_uncertainty(self, val): -# self.setter_validator("z_imag_uncertainty", val, fun=self._uuid_promoter) diff --git a/simpeg_drivers/electromagnetics/time_domain/params.py b/simpeg_drivers/electromagnetics/time_domain/params.py index e40fa94b..a02a73ea 100644 --- a/simpeg_drivers/electromagnetics/time_domain/params.py +++ b/simpeg_drivers/electromagnetics/time_domain/params.py @@ -40,6 +40,7 @@ class TimeDomainElectromagneticsForwardParams(BaseForwardData): :param x_channel_uncertainty: X-component data channel uncertainty. :param y_channel_bool: Y-component data channel boolean. :param y_channel_uncertainty: Y-component data channel uncertainty. + :param model_type: Specify whether the models are provided in resistivity or conductivity. :param data_units: Units for the TEM data """ @@ -51,8 +52,8 @@ class TimeDomainElectromagneticsForwardParams(BaseForwardData): physical_property: str = "conductivity" data_object: Receivers - z_channel_bool: bool - x_channel_bool: bool + z_channel_bool: bool | None = None + x_channel_bool: bool | None = None data_units: str = "dB/dt (T/s)" model_type: str = "Conductivity (S/m)" @@ -159,173 +160,3 @@ def uncertainty(self, component: str) -> float: """Returns uncertainty for chosen data component.""" uid = self.uncertainty_channel(component) return self.property_group_data(uid) - - -# class TimeDomainElectromagneticsParams(InversionBaseParams): -# """ -# Parameter class for Time-domain Electromagnetic (TEM) -> conductivity inversion. -# """ -# -# _physical_property = "conductivity" -# -# def __init__(self, input_file=None, forward_only=False, **kwargs): -# self._default_ui_json = deepcopy(default_ui_json) -# self._forward_defaults = deepcopy(forward_defaults) -# self._inversion_defaults = deepcopy(inversion_defaults) -# self._inversion_type = "tdem" -# self._validations = validations -# self._data_units = "dB/dt (T/s)" -# self._z_channel_bool = None -# self._z_channel = None -# self._z_uncertainty = None -# self._x_channel_bool = None -# self._x_channel = None -# self._x_uncertainty = None -# self._y_channel_bool = None -# self._y_channel = None -# self._y_uncertainty = None -# -# self._model_type = "Conductivity (S/m)" -# -# super().__init__(input_file=input_file, forward_only=forward_only, **kwargs) -# -# def data_channel(self, component: str): -# """Return uuid of data channel.""" -# return getattr(self, "_".join([component, "channel"]), None) -# -# @property -# def data_units(self): -# return self._data_units -# -# @data_units.setter -# def data_units(self, val): -# self.setter_validator("data_units", val) -# -# def uncertainty_channel(self, component: str): -# """Return uuid of uncertainty channel.""" -# return getattr(self, "_".join([component, "uncertainty"]), None) -# -# def property_group_data(self, property_group: UUID): -# data = {} -# channels = self.data_object.channels -# if self.forward_only: -# return {k: None for k in channels} -# else: -# group = self.data_object.find_or_create_property_group( -# name=property_group.name -# ) -# property_names = [ -# self.geoh5.get_entity(p)[0].name for p in group.properties -# ] -# properties = [self.geoh5.get_entity(p)[0].values for p in group.properties] -# for i, f in enumerate(channels): -# try: -# f_ind = property_names.index( -# next(k for k in property_names if f"{f:.2e}" in k) -# ) # Safer if data was saved with geoapps naming convention -# data[f] = properties[f_ind] -# except StopIteration: -# data[f] = properties[i] # in case of other naming conventions -# -# return data -# -# @property -# def model_type(self): -# """Model units.""" -# return self._model_type -# -# @model_type.setter -# def model_type(self, val): -# self.setter_validator("model_type", val) -# -# @property -# def unit_conversion(self): -# """Return time unit conversion factor.""" -# conversion = { -# "Seconds (s)": 1.0, -# "Milliseconds (ms)": 1e-3, -# "Microseconds (us)": 1e-6, -# } -# return conversion[self.data_object.unit] -# -# def data(self, component: str): -# """Returns array of data for chosen data component.""" -# property_group = self.data_channel(component) -# return self.property_group_data(property_group) -# -# def uncertainty(self, component: str) -> float: -# """Returns uncertainty for chosen data component.""" -# uid = self.uncertainty_channel(component) -# return self.property_group_data(uid) -# -# @property -# def z_channel_bool(self): -# return self._z_channel_bool -# -# @z_channel_bool.setter -# def z_channel_bool(self, val): -# self.setter_validator("z_channel_bool", val) -# -# @property -# def z_channel(self): -# return self._z_channel -# -# @z_channel.setter -# def z_channel(self, val): -# self.setter_validator("z_channel", val, fun=self._uuid_promoter) -# -# @property -# def z_uncertainty(self): -# return self._z_uncertainty -# -# @z_uncertainty.setter -# def z_uncertainty(self, val): -# self.setter_validator("z_uncertainty", val, fun=self._uuid_promoter) -# -# @property -# def x_channel_bool(self): -# return self._x_channel_bool -# -# @x_channel_bool.setter -# def x_channel_bool(self, val): -# self.setter_validator("x_channel_bool", val) -# -# @property -# def x_channel(self): -# return self._x_channel -# -# @x_channel.setter -# def x_channel(self, val): -# self.setter_validator("x_channel", val, fun=self._uuid_promoter) -# -# @property -# def x_uncertainty(self): -# return self._x_uncertainty -# -# @x_uncertainty.setter -# def x_uncertainty(self, val): -# self.setter_validator("x_uncertainty", val, fun=self._uuid_promoter) -# -# @property -# def y_channel_bool(self): -# return self._y_channel_bool -# -# @y_channel_bool.setter -# def y_channel_bool(self, val): -# self.setter_validator("y_channel_bool", val) -# -# @property -# def y_channel(self): -# return self._y_channel -# -# @y_channel.setter -# def y_channel(self, val): -# self.setter_validator("y_channel", val, fun=self._uuid_promoter) -# -# @property -# def y_uncertainty(self): -# return self._y_uncertainty -# -# @y_uncertainty.setter -# def y_uncertainty(self, val): -# self.setter_validator("y_uncertainty", val, fun=self._uuid_promoter) diff --git a/simpeg_drivers/natural_sources/__init__.py b/simpeg_drivers/natural_sources/__init__.py index 4c01f5d1..4a7b37f4 100644 --- a/simpeg_drivers/natural_sources/__init__.py +++ b/simpeg_drivers/natural_sources/__init__.py @@ -9,8 +9,11 @@ # ''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''' -from .magnetotellurics import MagnetotelluricsParams -from .tipper import TipperParams +from .magnetotellurics import ( + MagnetotelluricsForwardParams, + MagnetotelluricsInversionParams, +) +from .tipper import TipperForwardParams, TipperInversionParams # pylint: disable=unused-import # flake8: noqa diff --git a/simpeg_drivers/natural_sources/magnetotellurics/__init__.py b/simpeg_drivers/natural_sources/magnetotellurics/__init__.py index bb61690f..345d9f88 100644 --- a/simpeg_drivers/natural_sources/magnetotellurics/__init__.py +++ b/simpeg_drivers/natural_sources/magnetotellurics/__init__.py @@ -9,7 +9,7 @@ # ''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''' -from .params import MagnetotelluricsParams +from .params import MagnetotelluricsForwardParams, MagnetotelluricsInversionParams # pylint: disable=unused-import # flake8: noqa diff --git a/simpeg_drivers/natural_sources/magnetotellurics/driver.py b/simpeg_drivers/natural_sources/magnetotellurics/driver.py index b293ad71..aa58f3c7 100644 --- a/simpeg_drivers/natural_sources/magnetotellurics/driver.py +++ b/simpeg_drivers/natural_sources/magnetotellurics/driver.py @@ -14,12 +14,14 @@ from simpeg_drivers.driver import InversionDriver from .constants import validations -from .params import MagnetotelluricsParams +from .params import MagnetotelluricsForwardParams, MagnetotelluricsInversionParams -class MagnetotelluricsDriver(InversionDriver): - _params_class = MagnetotelluricsParams +class MagnetotelluricsForwardDriver(InversionDriver): + _params_class = MagnetotelluricsForwardParams _validations = validations - def __init__(self, params: MagnetotelluricsParams): - super().__init__(params) + +class MagnetotelluricsInversionDriver(InversionDriver): + _params_class = MagnetotelluricsInversionParams + _validations = validations diff --git a/simpeg_drivers/natural_sources/magnetotellurics/params.py b/simpeg_drivers/natural_sources/magnetotellurics/params.py index 19df502f..999277c9 100644 --- a/simpeg_drivers/natural_sources/magnetotellurics/params.py +++ b/simpeg_drivers/natural_sources/magnetotellurics/params.py @@ -11,60 +11,135 @@ from __future__ import annotations -from copy import deepcopy -from uuid import UUID +from pathlib import Path +from typing import ClassVar -from simpeg_drivers.params import InversionBaseParams +from geoh5py.data import FloatData +from geoh5py.groups import PropertyGroup +from geoh5py.objects import MTReceivers -from .constants import ( - default_ui_json, - forward_defaults, - inversion_defaults, - validations, -) +from simpeg_drivers import assets_path +from simpeg_drivers.params import BaseForwardData, BaseInversionData -class MagnetotelluricsParams(InversionBaseParams): +class MagnetotelluricsForwardParams(BaseForwardData): """ - Parameter class for magnetotelluric->conductivity inversion. + Parameter class for magnetotelluric->conductivity simulation. + + :param zxx_real_channel_bool: Boolean for zxx real channel. + :param zxx_imag_channel_bool: Boolean for zxx imaginary channel. + :param zxy_real_channel_bool: Boolean for zxy real channel. + :param zxy_imag_channel_bool: Boolean for zxy imaginary channel. + :param zyx_real_channel_bool: Boolean for zyx real channel. + :param zyx_imag_channel_bool: Boolean for zyx imaginary channel. + :param zyy_real_channel_bool: Boolean for zyy real channel. + :param zyy_imag_channel_bool: Boolean for zyy imaginary channel. + :param background_conductivity: Background conductivity model. + :param model_type: Specify whether the models are provided in resistivity or conductivity. """ - _physical_property = "conductivity" + name: ClassVar[str] = "Magnetotellurics Forward" + title: ClassVar[str] = "Magnetotellurics Forward" + default_ui_json: ClassVar[Path] = ( + assets_path() / "uijson/magnetotellurics_forward.ui.json" + ) + + inversion_type: str = "magnetotellurics" + physical_property: str = "conductivity" + + data_object: MTReceivers + zxx_real_channel_bool: bool | None = None + zxx_imag_channel_bool: bool | None = None + zxy_real_channel_bool: bool | None = None + zxy_imag_channel_bool: bool | None = None + zyx_real_channel_bool: bool | None = None + zyx_imag_channel_bool: bool | None = None + zyy_real_channel_bool: bool | None = None + zyy_imag_channel_bool: bool | None = None + background_conductivity: float | FloatData + model_type: str = "Conductivity (S/m)" + + @property + def channels(self) -> list[str]: + return ["_".join(k.split("_")[:2]) for k in self.__dict__ if "channel" in k] + + def property_group_data(self, property_group: PropertyGroup): + """ + Return dictionary of channel/data. + + :param property_group: Property group uid + """ + _ = property_group + frequencies = self.data_object.channels + return {k: None for k in frequencies} - def __init__(self, input_file=None, forward_only=False, **kwargs): - self._default_ui_json = deepcopy(default_ui_json) - self._forward_defaults = deepcopy(forward_defaults) - self._inversion_defaults = deepcopy(inversion_defaults) - self._inversion_type = "magnetotellurics" - self._validations = validations - self._zxx_real_channel_bool = None - self._zxx_real_channel = None - self._zxx_real_uncertainty = None - self._zxx_imag_channel_bool = None - self._zxx_imag_channel = None - self._zxx_imag_uncertainty = None - self._zxy_real_channel_bool = None - self._zxy_real_channel = None - self._zxy_real_uncertainty = None - self._zxy_imag_channel_bool = None - self._zxy_imag_channel = None - self._zxy_imag_uncertainty = None - self._zyx_real_channel_bool = None - self._zyx_real_channel = None - self._zyx_real_uncertainty = None - self._zyx_imag_channel_bool = None - self._zyx_imag_channel = None - self._zyx_imag_uncertainty = None - self._zyy_real_channel_bool = None - self._zyy_real_channel = None - self._zyy_real_uncertainty = None - self._zyy_imag_channel_bool = None - self._zyy_imag_channel = None - self._zyy_imag_uncertainty = None - self._background_conductivity = None - self._model_type = "Conductivity (S/m)" + def data(self, component: str): + """Returns array of data for chosen data component.""" + property_group = self.data_channel(component) + return self.property_group_data(property_group) - super().__init__(input_file=input_file, forward_only=forward_only, **kwargs) + def uncertainty(self, component: str) -> float: + """Returns uncertainty for chosen data component.""" + uid = self.uncertainty_channel(component) + return self.property_group_data(uid) + + +class MagnetotelluricsInversionParams(BaseInversionData): + """ + Parameter class for magnetotelluric->conductivity inversion. + + :param zxx_real_channel: Real component of Zxx data. + :param zxx_real_uncertainty: Real component of Zxx uncertainty. + :param zxx_imag_channel: Imaginary component of Zxx data. + :param zxx_imag_uncertainty: Imaginary component of Zxx uncertainty. + :param zxy_real_channel: Real component of Zxy data. + :param zxy_real_uncertainty: Real component of Zxy uncertainty. + :param zxy_imag_channel: Imaginary component of Zxy data. + :param zxy_imag_uncertainty: Imaginary component of Zxy uncertainty. + :param zyx_real_channel: Real component of Zyx data. + :param zyx_real_uncertainty: Real component of Zyx uncertainty. + :param zyx_imag_channel: Imaginary component of Zyx data. + :param zyx_imag_uncertainty: Imaginary component of Zyx uncertainty. + :param zyy_real_channel: Real component of Zyy data. + :param zyy_real_uncertainty: Real component of Zyy uncertainty. + :param zyy_imag_channel: Imaginary component of Zyy data. + :param zyy_imag_uncertainty: Imaginary component of Zyy uncertainty. + :param background_conductivity: Background conductivity model. + :param model_type: Specify whether the models are provided in resistivity or conductivity. + """ + + name: ClassVar[str] = "Magnetotellurics Inversion" + title: ClassVar[str] = "Magnetotellurics Inversion" + default_ui_json: ClassVar[Path] = ( + assets_path() / "uijson/magnetotellurics_inversion.ui.json" + ) + + inversion_type: str = "magnetotellurics" + physical_property: str = "conductivity" + + data_object: MTReceivers + zxx_real_channel: PropertyGroup | None = None + zxx_real_uncertainty: PropertyGroup | None = None + zxx_imag_channel: PropertyGroup | None = None + zxx_imag_uncertainty: PropertyGroup | None = None + zxy_real_channel: PropertyGroup | None = None + zxy_real_uncertainty: PropertyGroup | None = None + zxy_imag_channel: PropertyGroup | None = None + zxy_imag_uncertainty: PropertyGroup | None = None + zyx_real_channel: PropertyGroup | None = None + zyx_real_uncertainty: PropertyGroup | None = None + zyx_imag_channel: PropertyGroup | None = None + zyx_imag_uncertainty: PropertyGroup | None = None + zyy_real_channel: PropertyGroup | None = None + zyy_real_uncertainty: PropertyGroup | None = None + zyy_imag_channel: PropertyGroup | None = None + zyy_imag_uncertainty: PropertyGroup | None = None + background_conductivity: float | FloatData + model_type: str = "Conductivity (S/m)" + + @property + def channels(self) -> list[str]: + return ["_".join(k.split("_")[:2]) for k in self.__dict__ if "channel" in k] def data_channel(self, component: str): """Return uuid of data channel.""" @@ -74,31 +149,29 @@ def uncertainty_channel(self, component: str): """Return uuid of uncertainty channel.""" return getattr(self, "_".join([component, "uncertainty"]), None) - def property_group_data(self, property_group: UUID): + def property_group_data(self, property_group: PropertyGroup): + """ + Return dictionary of channel/data. + + :param property_group: Property group uid + """ data = {} frequencies = self.data_object.channels - if self.forward_only: - return {k: None for k in frequencies} - else: - group = next( - k - for k in self.data_object.property_groups - if k.uid == property_group.uid - ) - property_names = [ - self.geoh5.get_entity(p)[0].name for p in group.properties - ] - properties = [self.geoh5.get_entity(p)[0].values for p in group.properties] - for i, f in enumerate(frequencies): - try: - f_ind = property_names.index( - next(k for k in property_names if f"{f:.2e}" in k) - ) # Safer if data was saved with geoapps naming convention - data[f] = properties[f_ind] - except StopIteration: - data[f] = properties[i] # in case of other naming conventions - - return data + group = next( + k for k in self.data_object.property_groups if k.uid == property_group.uid + ) + property_names = [self.geoh5.get_entity(p)[0].name for p in group.properties] + properties = [self.geoh5.get_entity(p)[0].values for p in group.properties] + for i, f in enumerate(frequencies): + try: + f_ind = property_names.index( + next(k for k in property_names if f"{f:.2e}" in k) + ) # Safer if data was saved with geoapps naming convention + data[f] = properties[f_ind] + except StopIteration: + data[f] = properties[i] # in case of other naming conventions + + return data def data(self, component: str): """Returns array of data for chosen data component.""" @@ -109,212 +182,3 @@ def uncertainty(self, component: str) -> float: """Returns uncertainty for chosen data component.""" uid = self.uncertainty_channel(component) return self.property_group_data(uid) - - @property - def model_type(self): - """Model units.""" - return self._model_type - - @model_type.setter - def model_type(self, val): - self.setter_validator("model_type", val) - - @property - def zxx_real_channel_bool(self): - return self._zxx_real_channel_bool - - @zxx_real_channel_bool.setter - def zxx_real_channel_bool(self, val): - self.setter_validator("zxx_real_channel_bool", val) - - @property - def zxx_real_channel(self): - return self._zxx_real_channel - - @zxx_real_channel.setter - def zxx_real_channel(self, val): - self.setter_validator("zxx_real_channel", val, fun=self._uuid_promoter) - - @property - def zxx_real_uncertainty(self): - return self._zxx_real_uncertainty - - @zxx_real_uncertainty.setter - def zxx_real_uncertainty(self, val): - self.setter_validator("zxx_real_uncertainty", val, fun=self._uuid_promoter) - - @property - def zxx_imag_channel_bool(self): - return self._zxx_imag_channel_bool - - @zxx_imag_channel_bool.setter - def zxx_imag_channel_bool(self, val): - self.setter_validator("zxx_imag_channel_bool", val) - - @property - def zxx_imag_channel(self): - return self._zxx_imag_channel - - @zxx_imag_channel.setter - def zxx_imag_channel(self, val): - self.setter_validator("zxx_imag_channel", val, fun=self._uuid_promoter) - - @property - def zxx_imag_uncertainty(self): - return self._zxx_imag_uncertainty - - @zxx_imag_uncertainty.setter - def zxx_imag_uncertainty(self, val): - self.setter_validator("zxx_imag_uncertainty", val, fun=self._uuid_promoter) - - @property - def zxy_real_channel_bool(self): - return self._zxy_real_channel_bool - - @zxy_real_channel_bool.setter - def zxy_real_channel_bool(self, val): - self.setter_validator("zxy_real_channel_bool", val) - - @property - def zxy_real_channel(self): - return self._zxy_real_channel - - @zxy_real_channel.setter - def zxy_real_channel(self, val): - self.setter_validator("zxy_real_channel", val, fun=self._uuid_promoter) - - @property - def zxy_real_uncertainty(self): - return self._zxy_real_uncertainty - - @zxy_real_uncertainty.setter - def zxy_real_uncertainty(self, val): - self.setter_validator("zxy_real_uncertainty", val, fun=self._uuid_promoter) - - @property - def zxy_imag_channel_bool(self): - return self._zxy_imag_channel_bool - - @zxy_imag_channel_bool.setter - def zxy_imag_channel_bool(self, val): - self.setter_validator("zxy_imag_channel_bool", val) - - @property - def zxy_imag_channel(self): - return self._zxy_imag_channel - - @zxy_imag_channel.setter - def zxy_imag_channel(self, val): - self.setter_validator("zxy_imag_channel", val, fun=self._uuid_promoter) - - @property - def zxy_imag_uncertainty(self): - return self._zxy_imag_uncertainty - - @zxy_imag_uncertainty.setter - def zxy_imag_uncertainty(self, val): - self.setter_validator("zxy_imag_uncertainty", val, fun=self._uuid_promoter) - - @property - def zyx_real_channel_bool(self): - return self._zyx_real_channel_bool - - @zyx_real_channel_bool.setter - def zyx_real_channel_bool(self, val): - self.setter_validator("zyx_real_channel_bool", val) - - @property - def zyx_real_channel(self): - return self._zyx_real_channel - - @zyx_real_channel.setter - def zyx_real_channel(self, val): - self.setter_validator("zyx_real_channel", val, fun=self._uuid_promoter) - - @property - def zyx_real_uncertainty(self): - return self._zyx_real_uncertainty - - @zyx_real_uncertainty.setter - def zyx_real_uncertainty(self, val): - self.setter_validator("zyx_real_uncertainty", val, fun=self._uuid_promoter) - - @property - def zyx_imag_channel_bool(self): - return self._zyx_imag_channel_bool - - @zyx_imag_channel_bool.setter - def zyx_imag_channel_bool(self, val): - self.setter_validator("zyx_imag_channel_bool", val) - - @property - def zyx_imag_channel(self): - return self._zyx_imag_channel - - @zyx_imag_channel.setter - def zyx_imag_channel(self, val): - self.setter_validator("zyx_imag_channel", val, fun=self._uuid_promoter) - - @property - def zyx_imag_uncertainty(self): - return self._zyx_imag_uncertainty - - @zyx_imag_uncertainty.setter - def zyx_imag_uncertainty(self, val): - self.setter_validator("zyx_imag_uncertainty", val, fun=self._uuid_promoter) - - @property - def zyy_real_channel_bool(self): - return self._zyy_real_channel_bool - - @zyy_real_channel_bool.setter - def zyy_real_channel_bool(self, val): - self.setter_validator("zyy_real_channel_bool", val) - - @property - def zyy_real_channel(self): - return self._zyy_real_channel - - @zyy_real_channel.setter - def zyy_real_channel(self, val): - self.setter_validator("zyy_real_channel", val, fun=self._uuid_promoter) - - @property - def zyy_real_uncertainty(self): - return self._zyy_real_uncertainty - - @zyy_real_uncertainty.setter - def zyy_real_uncertainty(self, val): - self.setter_validator("zyy_real_uncertainty", val, fun=self._uuid_promoter) - - @property - def zyy_imag_channel_bool(self): - return self._zyy_imag_channel_bool - - @zyy_imag_channel_bool.setter - def zyy_imag_channel_bool(self, val): - self.setter_validator("zyy_imag_channel_bool", val) - - @property - def zyy_imag_channel(self): - return self._zyy_imag_channel - - @zyy_imag_channel.setter - def zyy_imag_channel(self, val): - self.setter_validator("zyy_imag_channel", val, fun=self._uuid_promoter) - - @property - def zyy_imag_uncertainty(self): - return self._zyy_imag_uncertainty - - @zyy_imag_uncertainty.setter - def zyy_imag_uncertainty(self, val): - self.setter_validator("zyy_imag_uncertainty", val, fun=self._uuid_promoter) - - @property - def background_conductivity(self): - return self._background_conductivity - - @background_conductivity.setter - def background_conductivity(self, val): - self.setter_validator("background_conductivity", val, fun=self._uuid_promoter) diff --git a/simpeg_drivers/natural_sources/tipper/__init__.py b/simpeg_drivers/natural_sources/tipper/__init__.py index 2cb63156..72e7dc9c 100644 --- a/simpeg_drivers/natural_sources/tipper/__init__.py +++ b/simpeg_drivers/natural_sources/tipper/__init__.py @@ -9,7 +9,7 @@ # ''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''' -from .params import TipperParams +from .params import TipperForwardParams, TipperInversionParams # pylint: disable=unused-import # flake8: noqa diff --git a/simpeg_drivers/natural_sources/tipper/driver.py b/simpeg_drivers/natural_sources/tipper/driver.py index ec86f368..7ad64d6f 100644 --- a/simpeg_drivers/natural_sources/tipper/driver.py +++ b/simpeg_drivers/natural_sources/tipper/driver.py @@ -14,12 +14,14 @@ from simpeg_drivers.driver import InversionDriver from .constants import validations -from .params import TipperParams +from .params import TipperForwardParams, TipperInversionParams -class TipperDriver(InversionDriver): - _params_class = TipperParams +class TipperForwardDriver(InversionDriver): + _params_class = TipperForwardParams _validations = validations - def __init__(self, params: TipperParams): - super().__init__(params) + +class TipperInversionDriver(InversionDriver): + _params_class = TipperInversionParams + _validations = validations diff --git a/simpeg_drivers/natural_sources/tipper/params.py b/simpeg_drivers/natural_sources/tipper/params.py index d1ca825a..fb8bd31f 100644 --- a/simpeg_drivers/natural_sources/tipper/params.py +++ b/simpeg_drivers/natural_sources/tipper/params.py @@ -11,82 +11,57 @@ from __future__ import annotations -from copy import deepcopy -from uuid import UUID +from pathlib import Path +from typing import ClassVar -from simpeg_drivers.params import InversionBaseParams +from geoh5py.data import FloatData +from geoh5py.groups import PropertyGroup +from geoh5py.objects import TipperReceivers -from .constants import ( - default_ui_json, - forward_defaults, - inversion_defaults, - validations, -) +from simpeg_drivers import assets_path +from simpeg_drivers.params import BaseForwardData, BaseInversionData -class TipperParams(InversionBaseParams): +class TipperForwardParams(BaseForwardData): """ - Parameter class for magnetotelluric->conductivity inversion. + Parameter class for magnetotelluric->conductivity simulation. + + :param txz_real_channel_bool: Boolean for txz real channel. + :param txz_imag_channel_bool: Boolean for txz imaginary channel. + :param tyz_real_channel_bool: Boolean for tyz real channel. + :param tyz_imag_channel_bool: Boolean for tyz imaginary channel. + :param background_conductivity: Background conductivity model. + :param model_type: Specify whether the models are provided in resistivity or conductivity. """ - _physical_property = "conductivity" - - def __init__(self, input_file=None, forward_only=False, **kwargs): - self._default_ui_json = deepcopy(default_ui_json) - self._forward_defaults = deepcopy(forward_defaults) - self._inversion_defaults = deepcopy(inversion_defaults) - self._inversion_type = "tipper" - self._validations = validations - self._txz_real_channel_bool = None - self._txz_real_channel = None - self._txz_real_uncertainty = None - self._txz_imag_channel_bool = None - self._txz_imag_channel = None - self._txz_imag_uncertainty = None - self._tyz_real_channel_bool = None - self._tyz_real_channel = None - self._tyz_real_uncertainty = None - self._tyz_imag_channel_bool = None - self._tyz_imag_channel = None - self._tyz_imag_uncertainty = None - self._background_conductivity = None - self._model_type = "Conductivity (S/m)" - - super().__init__(input_file=input_file, forward_only=forward_only, **kwargs) + name: ClassVar[str] = "Tipper Forward" + title: ClassVar[str] = "Tipper Forward" + default_ui_json: ClassVar[Path] = assets_path() / "uijson/tipper_forward.ui.json" - def data_channel(self, component: str): - """Return uuid of data channel.""" - return getattr(self, "_".join([component, "channel"]), None) + inversion_type: str = "tipper" + physical_property: str = "conductivity" - def uncertainty_channel(self, component: str): - """Return uuid of uncertainty channel.""" - return getattr(self, "_".join([component, "uncertainty"]), None) + data_object: TipperReceivers + txz_real_channel_bool: bool | None = None + txz_imag_channel_bool: bool | None = None + tyz_real_channel_bool: bool | None = None + tyz_imag_channel_bool: bool | None = None + background_conductivity: float | FloatData + model_type: str = "Conductivity (S/m)" - def property_group_data(self, property_group: UUID): - data = {} + @property + def channels(self) -> list[str]: + return ["_".join(k.split("_")[:2]) for k in self.__dict__ if "channel" in k] + + def property_group_data(self, property_group: PropertyGroup): + """ + Return dictionary of channel/data. + + :param property_group: Property group uid + """ + _ = property_group frequencies = self.data_object.channels - if self.forward_only: - return {k: None for k in frequencies} - else: - group = next( - k - for k in self.data_object.property_groups - if k.uid == property_group.uid - ) - property_names = [ - self.geoh5.get_entity(p)[0].name for p in group.properties - ] - properties = [self.geoh5.get_entity(p)[0].values for p in group.properties] - for i, f in enumerate(frequencies): - try: - f_ind = property_names.index( - next(k for k in property_names if f"{f:.2e}" in k) - ) # Safer if data was saved with geoapps naming convention - data[f] = properties[f_ind] - except StopIteration: - data[f] = properties[i] # in case of other naming conventions - - return data + return {k: None for k in frequencies} def data(self, component: str): """Returns array of data for chosen data component.""" @@ -98,115 +73,84 @@ def uncertainty(self, component: str) -> float: uid = self.uncertainty_channel(component) return self.property_group_data(uid) - @property - def model_type(self): - """Model units.""" - return self._model_type - - @model_type.setter - def model_type(self, val): - self.setter_validator("model_type", val) - - @property - def txz_real_channel_bool(self): - return self._txz_real_channel_bool - - @txz_real_channel_bool.setter - def txz_real_channel_bool(self, val): - self.setter_validator("txz_real_channel_bool", val) - - @property - def txz_real_channel(self): - return self._txz_real_channel - - @txz_real_channel.setter - def txz_real_channel(self, val): - self.setter_validator("txz_real_channel", val, fun=self._uuid_promoter) - - @property - def txz_real_uncertainty(self): - return self._txz_real_uncertainty - - @txz_real_uncertainty.setter - def txz_real_uncertainty(self, val): - self.setter_validator("txz_real_uncertainty", val, fun=self._uuid_promoter) - - @property - def txz_imag_channel_bool(self): - return self._txz_imag_channel_bool - - @txz_imag_channel_bool.setter - def txz_imag_channel_bool(self, val): - self.setter_validator("txz_imag_channel_bool", val) - - @property - def txz_imag_channel(self): - return self._txz_imag_channel - - @txz_imag_channel.setter - def txz_imag_channel(self, val): - self.setter_validator("txz_imag_channel", val, fun=self._uuid_promoter) - - @property - def txz_imag_uncertainty(self): - return self._txz_imag_uncertainty - - @txz_imag_uncertainty.setter - def txz_imag_uncertainty(self, val): - self.setter_validator("txz_imag_uncertainty", val, fun=self._uuid_promoter) - - @property - def tyz_real_channel_bool(self): - return self._tyz_real_channel_bool - - @tyz_real_channel_bool.setter - def tyz_real_channel_bool(self, val): - self.setter_validator("tyz_real_channel_bool", val) - @property - def tyz_real_channel(self): - return self._tyz_real_channel - - @tyz_real_channel.setter - def tyz_real_channel(self, val): - self.setter_validator("tyz_real_channel", val, fun=self._uuid_promoter) +class TipperInversionParams(BaseInversionData): + """ + Parameter class for magnetotelluric->conductivity inversion. - @property - def tyz_real_uncertainty(self): - return self._tyz_real_uncertainty + :param txz_real_channel: Real component of Txz tipper data. + :param txz_real_uncertainty: Real component of Txz tipper uncertainty. + :param txz_imag_channel: Imaginary component of Txz tipper data. + :param txz_imag_uncertainty: Imaginary component of Txz tipper uncertainty. + :param tyz_real_channel: Real component of Tyz tipper data. + :param tyz_real_uncertainty: Real component of Tyz tipper uncertainty. + :param tyz_imag_channel: Imaginary component of Tyz tipper data. + :param tyz_imag_uncertainty: Imaginary component of Tyz tipper uncertainty. + :param background_conductivity: Background conductivity model. + :param model_type: Specify whether the models are provided in resistivity or conductivity. + """ - @tyz_real_uncertainty.setter - def tyz_real_uncertainty(self, val): - self.setter_validator("tyz_real_uncertainty", val, fun=self._uuid_promoter) + name: ClassVar[str] = "Tipper Inversion" + title: ClassVar[str] = "Tipper Inversion" + default_ui_json: ClassVar[Path] = assets_path() / "uijson/tipper_inversion.ui.json" + + inversion_type: str = "tipper" + physical_property: str = "conductivity" + + data_object: TipperReceivers + txz_real_channel: PropertyGroup | None = None + txz_real_uncertainty: PropertyGroup | None = None + txz_imag_channel: PropertyGroup | None = None + txz_imag_uncertainty: PropertyGroup | None = None + tyz_real_channel: PropertyGroup | None = None + tyz_real_uncertainty: PropertyGroup | None = None + tyz_imag_channel: PropertyGroup | None = None + tyz_imag_uncertainty: PropertyGroup | None = None + background_conductivity: float | FloatData + model_type: str = "Conductivity (S/m)" @property - def tyz_imag_channel_bool(self): - return self._tyz_imag_channel_bool + def channels(self) -> list[str]: + return ["_".join(k.split("_")[:2]) for k in self.__dict__ if "channel" in k] - @tyz_imag_channel_bool.setter - def tyz_imag_channel_bool(self, val): - self.setter_validator("tyz_imag_channel_bool", val) - - @property - def tyz_imag_channel(self): - return self._tyz_imag_channel + def data_channel(self, component: str): + """Return uuid of data channel.""" + return getattr(self, "_".join([component, "channel"]), None) - @tyz_imag_channel.setter - def tyz_imag_channel(self, val): - self.setter_validator("tyz_imag_channel", val, fun=self._uuid_promoter) + def uncertainty_channel(self, component: str): + """Return uuid of uncertainty channel.""" + return getattr(self, "_".join([component, "uncertainty"]), None) - @property - def tyz_imag_uncertainty(self): - return self._tyz_imag_uncertainty + def property_group_data(self, property_group: PropertyGroup): + """ + Return dictionary of channel/data. - @tyz_imag_uncertainty.setter - def tyz_imag_uncertainty(self, val): - self.setter_validator("tyz_imag_uncertainty", val, fun=self._uuid_promoter) + :param property_group: Property group uid + """ + data = {} + frequencies = self.data_object.channels + group = next( + k for k in self.data_object.property_groups if k.uid == property_group.uid + ) + property_names = [self.geoh5.get_entity(p)[0].name for p in group.properties] + properties = [self.geoh5.get_entity(p)[0].values for p in group.properties] + for i, f in enumerate(frequencies): + try: + f_ind = property_names.index( + next(k for k in property_names if f"{f:.2e}" in k) + ) # Safer if data was saved with geoapps naming convention + data[f] = properties[f_ind] + except StopIteration: + data[f] = properties[i] # in case of other naming conventions + + return data - @property - def background_conductivity(self): - return self._background_conductivity + def data(self, component: str): + """Returns array of data for chosen data component.""" + property_group = self.data_channel(component) + return self.property_group_data(property_group) - @background_conductivity.setter - def background_conductivity(self, val): - self.setter_validator("background_conductivity", val, fun=self._uuid_promoter) + def uncertainty(self, component: str) -> float: + """Returns uncertainty for chosen data component.""" + uid = self.uncertainty_channel(component) + return self.property_group_data(uid) diff --git a/tests/run_tests/driver_mt_test.py b/tests/run_tests/driver_mt_test.py index a99501be..7c3f3194 100644 --- a/tests/run_tests/driver_mt_test.py +++ b/tests/run_tests/driver_mt_test.py @@ -19,11 +19,14 @@ from geoh5py.workspace import Workspace from simpeg_drivers.natural_sources.magnetotellurics.driver import ( - MagnetotelluricsDriver, + MagnetotelluricsForwardDriver, + MagnetotelluricsInversionDriver, ) from simpeg_drivers.natural_sources.magnetotellurics.params import ( - MagnetotelluricsParams, + MagnetotelluricsForwardParams, + MagnetotelluricsInversionParams, ) +from simpeg_drivers.params import ActiveCellsData from simpeg_drivers.utils.testing import check_target, setup_inversion_workspace from simpeg_drivers.utils.utils import get_inversion_output @@ -53,15 +56,13 @@ def test_magnetotellurics_fwr_run( inversion_type="magnetotellurics", flatten=False, ) - params = MagnetotelluricsParams( - forward_only=True, + params = MagnetotelluricsForwardParams( geoh5=geoh5, - mesh=model.parent.uid, - topography_object=topography.uid, - resolution=0.0, + mesh=model.parent, + active_cells=ActiveCellsData(topography_object=topography), z_from_topo=False, - data_object=survey.uid, - starting_model=model.uid, + data_object=survey, + starting_model=model, background_conductivity=1e-2, zxx_real_channel_bool=True, zxx_imag_channel_bool=True, @@ -72,8 +73,8 @@ def test_magnetotellurics_fwr_run( zyy_real_channel_bool=True, zyy_imag_channel_bool=True, ) - params.workpath = tmp_path - fwr_driver = MagnetotelluricsDriver(params) + + fwr_driver = MagnetotelluricsForwardDriver(params) fwr_driver.run() @@ -138,18 +139,17 @@ def test_magnetotellurics_run(tmp_path: Path, max_iterations=1, pytest=True): for comp, data_group, uncert_group in zip( components, data_groups, uncert_groups, strict=True ): - data_kwargs[f"{comp}_channel"] = data_group.uid - data_kwargs[f"{comp}_uncertainty"] = uncert_group.uid + data_kwargs[f"{comp}_channel"] = data_group + data_kwargs[f"{comp}_uncertainty"] = uncert_group orig_zyy_real_1 = geoh5.get_entity("Iteration_0_zyy_real_[0]")[0].values # Run the inverse - params = MagnetotelluricsParams( + params = MagnetotelluricsInversionParams( geoh5=geoh5, - mesh=mesh.uid, - topography_object=topography.uid, - resolution=0.0, - data_object=survey.uid, + mesh=mesh, + active_cells=ActiveCellsData(topography_object=topography), + data_object=survey, starting_model=100.0, reference_model=100.0, alpha_s=1.0, @@ -169,8 +169,10 @@ def test_magnetotellurics_run(tmp_path: Path, max_iterations=1, pytest=True): store_sensitivities="ram", **data_kwargs, ) - params.write_input_file(path=tmp_path, name="Inv_run") - driver = MagnetotelluricsDriver.start(str(tmp_path / "Inv_run.ui.json")) + params.write_ui_json(path=tmp_path / "Inv_run.ui.json") + driver = MagnetotelluricsInversionDriver.start( + str(tmp_path / "Inv_run.ui.json") + ) with geoh5.open() as run_ws: output = get_inversion_output( @@ -186,18 +188,18 @@ def test_magnetotellurics_run(tmp_path: Path, max_iterations=1, pytest=True): # test that one channel works data_kwargs = {k: v for k, v in data_kwargs.items() if "zxx_real" in k} geoh5.open() - params = MagnetotelluricsParams( + params = MagnetotelluricsInversionParams( geoh5=geoh5, - mesh=geoh5.get_entity("mesh")[0].uid, - topography_object=topography.uid, - data_object=survey.uid, + mesh=geoh5.get_entity("mesh")[0], + active_cells=ActiveCellsData(topography_object=topography), + data_object=survey, starting_model=0.01, - conductivity_model=1e-2, + background_conductivity=1e-2, max_global_iterations=0, **data_kwargs, ) - params.write_input_file(path=tmp_path, name="Inv_run") - MagnetotelluricsDriver.start(str(tmp_path / "Inv_run.ui.json")) + params.write_ui_json(path=tmp_path / "Inv_run.ui.json") + MagnetotelluricsInversionDriver.start(str(tmp_path / "Inv_run.ui.json")) if __name__ == "__main__": diff --git a/tests/run_tests/driver_tipper_test.py b/tests/run_tests/driver_tipper_test.py index 7812ed57..51457ff6 100644 --- a/tests/run_tests/driver_tipper_test.py +++ b/tests/run_tests/driver_tipper_test.py @@ -16,8 +16,15 @@ from geoh5py.groups import SimPEGGroup from geoh5py.workspace import Workspace -from simpeg_drivers.natural_sources import TipperParams -from simpeg_drivers.natural_sources.tipper.driver import TipperDriver +from simpeg_drivers.natural_sources.tipper import ( + TipperForwardParams, + TipperInversionParams, +) +from simpeg_drivers.natural_sources.tipper.driver import ( + TipperForwardDriver, + TipperInversionDriver, +) +from simpeg_drivers.params import ActiveCellsData from simpeg_drivers.utils.testing import check_target, setup_inversion_workspace from simpeg_drivers.utils.utils import get_inversion_output @@ -48,15 +55,13 @@ def test_tipper_fwr_run( flatten=False, ) - params = TipperParams( - forward_only=True, + params = TipperForwardParams( geoh5=geoh5, - mesh=model.parent.uid, - topography_object=topography.uid, - resolution=0.0, + mesh=model.parent, + active_cells=ActiveCellsData(topography_object=topography), z_from_topo=False, - data_object=survey.uid, - starting_model=model.uid, + data_object=survey, + starting_model=model, model_type="Resistivity (Ohm-m)", background_conductivity=1e2, txz_real_channel_bool=True, @@ -64,8 +69,8 @@ def test_tipper_fwr_run( tyz_real_channel_bool=True, tyz_imag_channel_bool=True, ) - params.workpath = tmp_path - fwr_driver = TipperDriver(params) + + fwr_driver = TipperForwardDriver(params) # Should always be returning conductivity for simpeg simulations assert not np.any(np.exp(fwr_driver.models.starting) > 1.01) @@ -121,18 +126,17 @@ def test_tipper_run(tmp_path: Path, max_iterations=1, pytest=True): for comp, data_group, uncert_group in zip( components, data_groups, uncert_groups, strict=True ): - data_kwargs[f"{comp}_channel"] = data_group.uid - data_kwargs[f"{comp}_uncertainty"] = uncert_group.uid + data_kwargs[f"{comp}_channel"] = data_group + data_kwargs[f"{comp}_uncertainty"] = uncert_group orig_tyz_real_1 = geoh5.get_entity("Iteration_0_tyz_real_[0]")[0].values # Run the inverse - params = TipperParams( + params = TipperInversionParams( geoh5=geoh5, - mesh=mesh.uid, - topography_object=topography.uid, - resolution=0.0, - data_object=survey.uid, + mesh=mesh, + active_cells=ActiveCellsData(topography_object=topography), + data_object=survey, starting_model=1e2, reference_model=1e2, background_conductivity=1e2, @@ -155,8 +159,8 @@ def test_tipper_run(tmp_path: Path, max_iterations=1, pytest=True): store_sensitivities="ram", **data_kwargs, ) - params.write_input_file(path=tmp_path, name="Inv_run") - driver = TipperDriver.start(str(tmp_path / "Inv_run.ui.json")) + params.write_ui_json(path=tmp_path / "Inv_run.ui.json") + driver = TipperInversionDriver.start(str(tmp_path / "Inv_run.ui.json")) with geoh5.open() as run_ws: output = get_inversion_output( From f7a61d712d9f8b9e7c33f6fa3aaa2c6b338e65fd Mon Sep 17 00:00:00 2001 From: benjamink Date: Wed, 12 Feb 2025 10:25:41 -0800 Subject: [PATCH 06/10] Refactor with EMDataMixin and apply to MT params --- simpeg_drivers/components/data.py | 13 +- .../frequency_domain/params.py | 3 +- .../electromagnetics/time_domain/params.py | 25 +-- .../magnetotellurics/params.py | 134 ++++++++-------- .../natural_sources/tipper/params.py | 8 - simpeg_drivers/params.py | 148 ++++++++++++------ 6 files changed, 188 insertions(+), 143 deletions(-) diff --git a/simpeg_drivers/components/data.py b/simpeg_drivers/components/data.py index 7cfa5a8e..fe530506 100644 --- a/simpeg_drivers/components/data.py +++ b/simpeg_drivers/components/data.py @@ -89,7 +89,6 @@ def __init__(self, workspace: Workspace, params: InversionBaseParams): self.indices: np.ndarray | None = None self.vector: bool | None = None self.n_blocks: int | None = None - self.components: list[str] | None = None self.observed: dict[str, np.ndarray] = {} self.predicted: dict[str, np.ndarray] = {} self.uncertainties: dict[str, np.ndarray] = {} @@ -106,8 +105,10 @@ def _initialize(self) -> None: """Extract data from the workspace using params data.""" self.vector = True if self.params.inversion_type == "magnetic vector" else False self.n_blocks = 3 if self.params.inversion_type == "magnetic vector" else 1 - self.components, self.observed, self.uncertainties = self.get_data() - self.has_tensor = InversionData.check_tensor(self.components) + self.components = self.params.active_components + self.observed = self.params.data + self.uncertainties = self.params.uncertainties + self.has_tensor = InversionData.check_tensor(self.params.components) self.locations = super().get_locations(self.params.data_object) if "2d" in self.params.inversion_type: @@ -280,7 +281,7 @@ def normalize( """ d = deepcopy(data) for chan in getattr(self.params.data_object, "channels", [None]): - for comp in self.components: + for comp in self.params.active_components: if isinstance(d[comp], dict): if d[comp][chan] is not None: d[comp][chan] *= self.normalizations[chan][comp] @@ -298,7 +299,7 @@ def get_normalizations(self): normalizations = {} for chan in getattr(self.params.data_object, "channels", [None]): normalizations[chan] = {} - for comp in self.components: + for comp in self.params.components: normalizations[chan][comp] = np.ones(self.mask.sum()) if comp in ["potential", "chargeability"]: normalizations[chan][comp] = 1 @@ -488,7 +489,7 @@ def survey(self): @property def n_data(self): n_data = 0 - for comp in self.components: + for comp in self.params.active_components: if isinstance(self.observed[comp], dict): for channel in self.observed[comp]: n_data += len(self.observed[comp][channel]) diff --git a/simpeg_drivers/electromagnetics/frequency_domain/params.py b/simpeg_drivers/electromagnetics/frequency_domain/params.py index 0f5209bb..ae36ccf6 100644 --- a/simpeg_drivers/electromagnetics/frequency_domain/params.py +++ b/simpeg_drivers/electromagnetics/frequency_domain/params.py @@ -162,8 +162,9 @@ def property_group_data(self, property_group: PropertyGroup): channels = self.data_object.channels group = self.data_object.fetch_property_group(name=property_group.name) properties = [self.geoh5.get_entity(p)[0].values for p in group.properties] + data = {f: properties[i] for i, f in enumerate(channels)} - return {f: properties[i] for i, f in enumerate(channels)} + return data def data(self, component: str): """Returns array of data for chosen data component.""" diff --git a/simpeg_drivers/electromagnetics/time_domain/params.py b/simpeg_drivers/electromagnetics/time_domain/params.py index a02a73ea..e3dd269d 100644 --- a/simpeg_drivers/electromagnetics/time_domain/params.py +++ b/simpeg_drivers/electromagnetics/time_domain/params.py @@ -22,7 +22,7 @@ ) from simpeg_drivers import assets_path -from simpeg_drivers.params import BaseForwardData, BaseInversionData +from simpeg_drivers.params import BaseForwardData, BaseInversionData, EMDataMixin Receivers: TypeAlias = ( @@ -30,7 +30,7 @@ ) -class TimeDomainElectromagneticsForwardParams(BaseForwardData): +class TimeDomainElectromagneticsForwardParams(EMDataMixin, BaseForwardData): """ Parameter class for Time-domain Electromagnetic (TEM) -> conductivity forward simulation. @@ -77,18 +77,19 @@ def property_group_data(self, property_group: PropertyGroup): channels = self.data_object.channels return {k: None for k in channels} - def data(self, component: str): - """Returns array of data for chosen data component.""" - property_group = self.data_channel(component) - return self.property_group_data(property_group) - - def uncertainty(self, component: str) -> float: - """Returns uncertainty for chosen data component.""" - uid = self.uncertainty_channel(component) - return self.property_group_data(uid) + # + # def data(self, component: str): + # """Returns array of data for chosen data component.""" + # property_group = self.data_channel(component) + # return self.property_group_data(property_group) + # + # def uncertainty(self, component: str) -> float: + # """Returns uncertainty for chosen data component.""" + # uid = self.uncertainty_channel(component) + # return self.property_group_data(uid) -class TimeDomainElectromagneticsInversionParams(BaseInversionData): +class TimeDomainElectromagneticsInversionParams(EMDataMixin, BaseInversionData): """ Parameter class for Time-domain Electromagnetic (TEM) -> conductivity inversion. diff --git a/simpeg_drivers/natural_sources/magnetotellurics/params.py b/simpeg_drivers/natural_sources/magnetotellurics/params.py index 999277c9..b31e074a 100644 --- a/simpeg_drivers/natural_sources/magnetotellurics/params.py +++ b/simpeg_drivers/natural_sources/magnetotellurics/params.py @@ -19,10 +19,10 @@ from geoh5py.objects import MTReceivers from simpeg_drivers import assets_path -from simpeg_drivers.params import BaseForwardData, BaseInversionData +from simpeg_drivers.params import BaseForwardData, BaseInversionData, EMDataMixin -class MagnetotelluricsForwardParams(BaseForwardData): +class MagnetotelluricsForwardParams(EMDataMixin, BaseForwardData): """ Parameter class for magnetotelluric->conductivity simulation. @@ -63,28 +63,28 @@ class MagnetotelluricsForwardParams(BaseForwardData): def channels(self) -> list[str]: return ["_".join(k.split("_")[:2]) for k in self.__dict__ if "channel" in k] - def property_group_data(self, property_group: PropertyGroup): - """ - Return dictionary of channel/data. - - :param property_group: Property group uid - """ - _ = property_group - frequencies = self.data_object.channels - return {k: None for k in frequencies} - - def data(self, component: str): - """Returns array of data for chosen data component.""" - property_group = self.data_channel(component) - return self.property_group_data(property_group) - - def uncertainty(self, component: str) -> float: - """Returns uncertainty for chosen data component.""" - uid = self.uncertainty_channel(component) - return self.property_group_data(uid) - - -class MagnetotelluricsInversionParams(BaseInversionData): + # def property_group_data(self, property_group: PropertyGroup): + # """ + # Return dictionary of channel/data. + # + # :param property_group: Property group uid + # """ + # _ = property_group + # frequencies = self.data_object.channels + # return {k: None for k in frequencies} + # + # def data(self, component: str): + # """Returns array of data for chosen data component.""" + # property_group = self.data_channel(component) + # return self.property_group_data(property_group) + # + # def uncertainty(self, component: str) -> float: + # """Returns uncertainty for chosen data component.""" + # uid = self.uncertainty_channel(component) + # return self.property_group_data(uid) + + +class MagnetotelluricsInversionParams(EMDataMixin, BaseInversionData): """ Parameter class for magnetotelluric->conductivity inversion. @@ -137,48 +137,44 @@ class MagnetotelluricsInversionParams(BaseInversionData): background_conductivity: float | FloatData model_type: str = "Conductivity (S/m)" - @property - def channels(self) -> list[str]: - return ["_".join(k.split("_")[:2]) for k in self.__dict__ if "channel" in k] - - def data_channel(self, component: str): - """Return uuid of data channel.""" - return getattr(self, "_".join([component, "channel"]), None) - - def uncertainty_channel(self, component: str): - """Return uuid of uncertainty channel.""" - return getattr(self, "_".join([component, "uncertainty"]), None) - - def property_group_data(self, property_group: PropertyGroup): - """ - Return dictionary of channel/data. - - :param property_group: Property group uid - """ - data = {} - frequencies = self.data_object.channels - group = next( - k for k in self.data_object.property_groups if k.uid == property_group.uid - ) - property_names = [self.geoh5.get_entity(p)[0].name for p in group.properties] - properties = [self.geoh5.get_entity(p)[0].values for p in group.properties] - for i, f in enumerate(frequencies): - try: - f_ind = property_names.index( - next(k for k in property_names if f"{f:.2e}" in k) - ) # Safer if data was saved with geoapps naming convention - data[f] = properties[f_ind] - except StopIteration: - data[f] = properties[i] # in case of other naming conventions - - return data - - def data(self, component: str): - """Returns array of data for chosen data component.""" - property_group = self.data_channel(component) - return self.property_group_data(property_group) - - def uncertainty(self, component: str) -> float: - """Returns uncertainty for chosen data component.""" - uid = self.uncertainty_channel(component) - return self.property_group_data(uid) + # @property + # def channels(self) -> list[str]: + # return ["_".join(k.split("_")[:2]) for k in self.__dict__ if "channel" in k] + # + # def property_group_data(self, property_group: PropertyGroup): + # """ + # Return dictionary of channel/data. + # + # :param property_group: Property group uid + # """ + # data = {} + # frequencies = self.data_object.channels + # if property_group is None: + # return {k: None for k in frequencies} + # group = next( + # k for k in self.data_object.property_groups if k.uid == property_group.uid + # ) + # property_names = [self.geoh5.get_entity(p)[0].name for p in group.properties] + # properties = [self.geoh5.get_entity(p)[0].values for p in group.properties] + # for i, f in enumerate(frequencies): + # try: + # f_ind = property_names.index( + # next(k for k in property_names if f"{f:.2e}" in k) + # ) # Safer if data was saved with geoapps naming convention + # data[f] = properties[f_ind] + # except StopIteration: + # data[f] = properties[i] # in case of other naming conventions + # + # return data + + # + # @property + # def data(self, component: str): + # """Returns array of data for chosen data component.""" + # property_group = self.data_channel(component) + # return self.property_group_data(property_group) + # @property + # def uncertainty(self, component: str) -> float: + # """Returns uncertainty for chosen data component.""" + # uid = self.uncertainty_channel(component) + # return self.property_group_data(uid) diff --git a/simpeg_drivers/natural_sources/tipper/params.py b/simpeg_drivers/natural_sources/tipper/params.py index fb8bd31f..1aca23f0 100644 --- a/simpeg_drivers/natural_sources/tipper/params.py +++ b/simpeg_drivers/natural_sources/tipper/params.py @@ -113,14 +113,6 @@ class TipperInversionParams(BaseInversionData): def channels(self) -> list[str]: return ["_".join(k.split("_")[:2]) for k in self.__dict__ if "channel" in k] - def data_channel(self, component: str): - """Return uuid of data channel.""" - return getattr(self, "_".join([component, "channel"]), None) - - def uncertainty_channel(self, component: str): - """Return uuid of uncertainty channel.""" - return getattr(self, "_".join([component, "uncertainty"]), None) - def property_group_data(self, property_group: PropertyGroup): """ Return dictionary of channel/data. diff --git a/simpeg_drivers/params.py b/simpeg_drivers/params.py index 9237ae77..1a64d0a9 100644 --- a/simpeg_drivers/params.py +++ b/simpeg_drivers/params.py @@ -15,21 +15,23 @@ import warnings from copy import deepcopy from pathlib import Path -from typing import ClassVar +from typing import ClassVar, TypeAlias from uuid import UUID import numpy as np from geoapps_utils.driver.data import BaseData from geoapps_utils.driver.params import BaseParams from geoh5py.data import BooleanData, FloatData, NumericData -from geoh5py.groups import SimPEGGroup, UIJsonGroup -from geoh5py.objects import DrapeModel, Octree, Points +from geoh5py.groups import PropertyGroup, SimPEGGroup, UIJsonGroup +from geoh5py.objects import Octree, Points from geoh5py.shared.utils import fetch_active_workspace from geoh5py.ui_json import InputFile -from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator +from pydantic import BaseModel, ConfigDict, field_validator, model_validator -# from simpeg_drivers import assets_path +InversionData: TypeAlias = ( + dict[str, np.ndarray | None] | dict[str, dict[float, np.ndarray | None]] +) # pylint: disable=too-many-lines @@ -154,39 +156,60 @@ def workpath(self): return Path(self.geoh5.h5file).parent @property - def channels(self) -> list[str]: - return [k.split("_")[0] for k in self.__dict__ if "channel" in k] + def components(self) -> list[str]: + """Return list of component names.""" + return [self._component_name(k) for k in self.__dict__ if "channel" in k] - def data_channel(self, component: str) -> NumericData | None: - """Return the data object associated with the component.""" - return getattr(self, "_".join([component, "channel"]), None) + @property + def active_components(self) -> list[str]: + """Return list of active components.""" + return [k for k in self.components if self.component_data(k) is not None] - def data(self, component: str) -> np.ndarray | None: - """Returns array of data for chosen data component if it exists.""" - data_entity = self.data_channel(component) - if isinstance(data_entity, NumericData): - return data_entity.values.astype(float) - return None + @property + def data(self) -> InversionData: + """Return dictionary of data components and associated values.""" + out = {} + for k in self.active_components: + out[k] = self.component_data(k) + return out - def uncertainty_channel(self, component: str) -> NumericData | None: - """Return the uncertainty object associated with the component.""" - return getattr(self, "_".join([component, "uncertainty"]), None) + @property + def uncertainties(self) -> InversionData: + """Return dictionary of unceratinty components and associated values.""" + out = {} + for k in self.active_components: + out[k] = self.component_uncertainty(k) + return out - def uncertainty(self, component: str) -> np.ndarray | None: - """Returns uncertainty for chosen data component if it exists.""" + def component_data(self, component: str) -> np.ndarray | None: + """Return data values associated with the component.""" + data = getattr(self, "_".join([component, "channel"]), None) + if isinstance(data, NumericData): + data = data.values + return data - uncertainty_entity = self.uncertainty_channel(component) - if isinstance(uncertainty_entity, NumericData): - return uncertainty_entity.values.astype(float) + def component_uncertainty(self, component: str) -> np.ndarray | None: + """ + Return uncertainty values associated with the component. - data = self.data(component) - if data is not None: - if isinstance(uncertainty_entity, int | float): - return np.array([float(uncertainty_entity)] * len(data)) - else: - return data * 0.0 + 1.0 # Default + If the uncertainty is a float, it will be broadcasted to the same + shape as the data. - return None + :param component: Component name. + """ + data = getattr(self, "_".join([component, "uncertainty"]), None) + if isinstance(data, NumericData): + data = data.values + elif isinstance(data, float): + data *= np.ones_like(self.component_data(component)) + + return data + + def _component_name(self, component: str) -> str: + """Strip the '_channel' and '_channel_bool' suffixes from data name.""" + return "_".join( + [k for k in component.split("_") if k not in ["channel", "bool"]] + ) @property def padding_cells(self) -> int: @@ -209,14 +232,9 @@ class BaseForwardData(CoreData): forward_only: bool = True @property - def components(self) -> list[str]: - """Retrieve component names used to index channel and uncertainty data.""" - comps = [] - for c in self.channels: - if getattr(self, f"{c}_channel_bool", False): - comps.append(c) - - return comps + def active_components(self) -> list[str]: + """Return list of active components.""" + return [k for k in self.components if getattr(self, f"{k}_channel_bool")] class BaseInversionData(CoreData): @@ -345,15 +363,51 @@ class BaseInversionData(CoreData): distributed_workers: int | None = None no_data_value: float | None = None - @property - def components(self) -> list[str]: - """Retrieve component names used to index channel and uncertainty data.""" - comps = [] - for c in self.channels: - if getattr(self, f"{c}_channel", False): - comps.append(c) - return comps +class EMDataMixin: + """ + Mixin class to add data and uncertainty access from property groups. + """ + + def __init__(self, *args, **kwargs): + super().__init__(*args, **kwargs) + + def component_data(self, component: str): + """Return data values associated with the component.""" + property_group = getattr(self, "_".join([component, "channel"]), None) + return self.property_group_data(property_group) + + def component_uncertainty(self, component: str): + """Return uncertainty values associated with the component.""" + property_group = getattr(self, "_".join([component, "uncertainty"]), None) + return self.property_group_data(property_group) + + def property_group_data(self, property_group: PropertyGroup): + """ + Return dictionary of channel/data. + + :param property_group: Property group uid + """ + frequencies = self.data_object.channels + if property_group is None: + return {f: None for f in frequencies} + + data = {} + group = next( + k for k in self.data_object.property_groups if k.uid == property_group.uid + ) + property_names = [self.geoh5.get_entity(p)[0].name for p in group.properties] + properties = [self.geoh5.get_entity(p)[0].values for p in group.properties] + for i, f in enumerate(frequencies): + try: + f_ind = property_names.index( + next(k for k in property_names if f"{f:.2e}" in k) + ) # Safer if data was saved with geoapps naming convention + data[f] = properties[f_ind] + except StopIteration: + data[f] = properties[i] # in case of other naming conventions + + return data class InversionBaseParams(BaseParams): From 66fa0efd47dc59e31fee19173833b67f81277e56 Mon Sep 17 00:00:00 2001 From: benjamink Date: Wed, 12 Feb 2025 11:03:36 -0800 Subject: [PATCH 07/10] Tests passing using the new EMDataMixin --- simpeg_drivers/components/locations.py | 2 + .../frequency_domain/params.py | 108 +++++++-------- .../electromagnetics/time_domain/params.py | 80 +++++------ .../natural_sources/tipper/params.py | 130 +++++++++--------- simpeg_drivers/params.py | 6 +- tests/run_tests/driver_mt_test.py | 1 + 6 files changed, 167 insertions(+), 160 deletions(-) diff --git a/simpeg_drivers/components/locations.py b/simpeg_drivers/components/locations.py index cc4f21b5..5c9cdfbd 100644 --- a/simpeg_drivers/components/locations.py +++ b/simpeg_drivers/components/locations.py @@ -121,6 +121,8 @@ def get_locations(self, entity: ObjectBase) -> np.ndarray: return locations def _filter(self, a, mask): + if a is None: + return None for k, v in a.items(): if not isinstance(v, np.ndarray): a.update({k: self._filter(v, mask)}) diff --git a/simpeg_drivers/electromagnetics/frequency_domain/params.py b/simpeg_drivers/electromagnetics/frequency_domain/params.py index ae36ccf6..39870ef6 100644 --- a/simpeg_drivers/electromagnetics/frequency_domain/params.py +++ b/simpeg_drivers/electromagnetics/frequency_domain/params.py @@ -22,7 +22,7 @@ ) from simpeg_drivers import assets_path -from simpeg_drivers.params import BaseForwardData, BaseInversionData +from simpeg_drivers.params import BaseForwardData, BaseInversionData, EMDataMixin Receivers: TypeAlias = ( @@ -30,7 +30,7 @@ ) -class FrequencyDomainElectromagneticsForwardParams(BaseForwardData): +class FrequencyDomainElectromagneticsForwardParams(EMDataMixin, BaseForwardData): """ Parameter class for Frequency-domain Electromagnetic (FEM) simulation. @@ -75,32 +75,32 @@ def unit_conversion(self): } return conversion[self.data_object.unit] - @property - def channels(self) -> list[str]: - return ["_".join(k.split("_")[:2]) for k in self.__dict__ if "channel" in k] - - def property_group_data(self, property_group: PropertyGroup): - """ - Return dictionary of channel/data. - - :param property_group: Property group uid - """ - _ = property_group - channels = self.data_object.channels - return {k: None for k in channels} - - def data(self, component: str): - """Returns array of data for chosen data component.""" - property_group = self.data_channel(component) - return self.property_group_data(property_group) - - def uncertainty(self, component: str) -> float: - """Returns uncertainty for chosen data component.""" - uid = self.uncertainty_channel(component) - return self.property_group_data(uid) - - -class FrequencyDomainElectromagneticsInversionParams(BaseInversionData): + # @property + # def channels(self) -> list[str]: + # return ["_".join(k.split("_")[:2]) for k in self.__dict__ if "channel" in k] + # + # def property_group_data(self, property_group: PropertyGroup): + # """ + # Return dictionary of channel/data. + # + # :param property_group: Property group uid + # """ + # _ = property_group + # channels = self.data_object.channels + # return {k: None for k in channels} + # + # def data(self, component: str): + # """Returns array of data for chosen data component.""" + # property_group = self.data_channel(component) + # return self.property_group_data(property_group) + # + # def uncertainty(self, component: str) -> float: + # """Returns uncertainty for chosen data component.""" + # uid = self.uncertainty_channel(component) + # return self.property_group_data(uid) + + +class FrequencyDomainElectromagneticsInversionParams(EMDataMixin, BaseInversionData): """ Parameter class for Frequency-domain Electromagnetic (FEM) -> conductivity inversion. @@ -149,29 +149,29 @@ def unit_conversion(self): } return conversion[self.data_object.unit] - @property - def channels(self) -> list[str]: - return ["_".join(k.split("_")[:2]) for k in self.__dict__ if "channel" in k] - - def property_group_data(self, property_group: PropertyGroup): - """ - Return dictionary of channel/data. - - :param property_group: Property group containing TEM data. - """ - channels = self.data_object.channels - group = self.data_object.fetch_property_group(name=property_group.name) - properties = [self.geoh5.get_entity(p)[0].values for p in group.properties] - data = {f: properties[i] for i, f in enumerate(channels)} - - return data - - def data(self, component: str): - """Returns array of data for chosen data component.""" - property_group = self.data_channel(component) - return self.property_group_data(property_group) - - def uncertainty(self, component: str) -> float: - """Returns uncertainty for chosen data component.""" - uid = self.uncertainty_channel(component) - return self.property_group_data(uid) + # @property + # def channels(self) -> list[str]: + # return ["_".join(k.split("_")[:2]) for k in self.__dict__ if "channel" in k] + # + # def property_group_data(self, property_group: PropertyGroup): + # """ + # Return dictionary of channel/data. + # + # :param property_group: Property group containing TEM data. + # """ + # channels = self.data_object.channels + # group = self.data_object.fetch_property_group(name=property_group.name) + # properties = [self.geoh5.get_entity(p)[0].values for p in group.properties] + # data = {f: properties[i] for i, f in enumerate(channels)} + # + # return data + # + # def data(self, component: str): + # """Returns array of data for chosen data component.""" + # property_group = self.data_channel(component) + # return self.property_group_data(property_group) + # + # def uncertainty(self, component: str) -> float: + # """Returns uncertainty for chosen data component.""" + # uid = self.uncertainty_channel(component) + # return self.property_group_data(uid) diff --git a/simpeg_drivers/electromagnetics/time_domain/params.py b/simpeg_drivers/electromagnetics/time_domain/params.py index e3dd269d..e903f34b 100644 --- a/simpeg_drivers/electromagnetics/time_domain/params.py +++ b/simpeg_drivers/electromagnetics/time_domain/params.py @@ -67,15 +67,15 @@ def unit_conversion(self): } return conversion[self.data_object.unit] - def property_group_data(self, property_group: PropertyGroup): - """ - Return dictionary of channel/data. - - :param property_group: Property group containing em data. - """ - _ = property_group - channels = self.data_object.channels - return {k: None for k in channels} + # def property_group_data(self, property_group: PropertyGroup): + # """ + # Return dictionary of channel/data. + # + # :param property_group: Property group containing em data. + # """ + # _ = property_group + # channels = self.data_object.channels + # return {k: None for k in channels} # # def data(self, component: str): @@ -130,34 +130,34 @@ def unit_conversion(self): } return conversion[self.data_object.unit] - def property_group_data(self, property_group: PropertyGroup): - """ - Return dictionary of channel/data. - - :param property_group: Property group containing TEM data. - """ - data = {} - channels = self.data_object.channels - group = self.data_object.fetch_property_group(name=property_group.name) - property_names = [self.geoh5.get_entity(p)[0].name for p in group.properties] - properties = [self.geoh5.get_entity(p)[0].values for p in group.properties] - for i, f in enumerate(channels): - try: - f_ind = property_names.index( - next(k for k in property_names if f"{f:.2e}" in k) - ) # Safer if data was saved with geoapps naming convention - data[f] = properties[f_ind] - except StopIteration: - data[f] = properties[i] # in case of other naming conventions - - return data - - def data(self, component: str): - """Returns array of data for chosen data component.""" - property_group = self.data_channel(component) - return self.property_group_data(property_group) - - def uncertainty(self, component: str) -> float: - """Returns uncertainty for chosen data component.""" - uid = self.uncertainty_channel(component) - return self.property_group_data(uid) + # def property_group_data(self, property_group: PropertyGroup): + # """ + # Return dictionary of channel/data. + # + # :param property_group: Property group containing TEM data. + # """ + # data = {} + # channels = self.data_object.channels + # group = self.data_object.fetch_property_group(name=property_group.name) + # property_names = [self.geoh5.get_entity(p)[0].name for p in group.properties] + # properties = [self.geoh5.get_entity(p)[0].values for p in group.properties] + # for i, f in enumerate(channels): + # try: + # f_ind = property_names.index( + # next(k for k in property_names if f"{f:.2e}" in k) + # ) # Safer if data was saved with geoapps naming convention + # data[f] = properties[f_ind] + # except StopIteration: + # data[f] = properties[i] # in case of other naming conventions + # + # return data + # + # def data(self, component: str): + # """Returns array of data for chosen data component.""" + # property_group = self.data_channel(component) + # return self.property_group_data(property_group) + # + # def uncertainty(self, component: str) -> float: + # """Returns uncertainty for chosen data component.""" + # uid = self.uncertainty_channel(component) + # return self.property_group_data(uid) diff --git a/simpeg_drivers/natural_sources/tipper/params.py b/simpeg_drivers/natural_sources/tipper/params.py index 1aca23f0..f7b65603 100644 --- a/simpeg_drivers/natural_sources/tipper/params.py +++ b/simpeg_drivers/natural_sources/tipper/params.py @@ -19,10 +19,10 @@ from geoh5py.objects import TipperReceivers from simpeg_drivers import assets_path -from simpeg_drivers.params import BaseForwardData, BaseInversionData +from simpeg_drivers.params import BaseForwardData, BaseInversionData, EMDataMixin -class TipperForwardParams(BaseForwardData): +class TipperForwardParams(EMDataMixin, BaseForwardData): """ Parameter class for magnetotelluric->conductivity simulation. @@ -49,32 +49,32 @@ class TipperForwardParams(BaseForwardData): background_conductivity: float | FloatData model_type: str = "Conductivity (S/m)" - @property - def channels(self) -> list[str]: - return ["_".join(k.split("_")[:2]) for k in self.__dict__ if "channel" in k] - - def property_group_data(self, property_group: PropertyGroup): - """ - Return dictionary of channel/data. - - :param property_group: Property group uid - """ - _ = property_group - frequencies = self.data_object.channels - return {k: None for k in frequencies} - - def data(self, component: str): - """Returns array of data for chosen data component.""" - property_group = self.data_channel(component) - return self.property_group_data(property_group) - - def uncertainty(self, component: str) -> float: - """Returns uncertainty for chosen data component.""" - uid = self.uncertainty_channel(component) - return self.property_group_data(uid) - - -class TipperInversionParams(BaseInversionData): + # @property + # def channels(self) -> list[str]: + # return ["_".join(k.split("_")[:2]) for k in self.__dict__ if "channel" in k] + # + # def property_group_data(self, property_group: PropertyGroup): + # """ + # Return dictionary of channel/data. + # + # :param property_group: Property group uid + # """ + # _ = property_group + # frequencies = self.data_object.channels + # return {k: None for k in frequencies} + # + # def data(self, component: str): + # """Returns array of data for chosen data component.""" + # property_group = self.data_channel(component) + # return self.property_group_data(property_group) + # + # def uncertainty(self, component: str) -> float: + # """Returns uncertainty for chosen data component.""" + # uid = self.uncertainty_channel(component) + # return self.property_group_data(uid) + + +class TipperInversionParams(EMDataMixin, BaseInversionData): """ Parameter class for magnetotelluric->conductivity inversion. @@ -109,40 +109,40 @@ class TipperInversionParams(BaseInversionData): background_conductivity: float | FloatData model_type: str = "Conductivity (S/m)" - @property - def channels(self) -> list[str]: - return ["_".join(k.split("_")[:2]) for k in self.__dict__ if "channel" in k] - - def property_group_data(self, property_group: PropertyGroup): - """ - Return dictionary of channel/data. - - :param property_group: Property group uid - """ - data = {} - frequencies = self.data_object.channels - group = next( - k for k in self.data_object.property_groups if k.uid == property_group.uid - ) - property_names = [self.geoh5.get_entity(p)[0].name for p in group.properties] - properties = [self.geoh5.get_entity(p)[0].values for p in group.properties] - for i, f in enumerate(frequencies): - try: - f_ind = property_names.index( - next(k for k in property_names if f"{f:.2e}" in k) - ) # Safer if data was saved with geoapps naming convention - data[f] = properties[f_ind] - except StopIteration: - data[f] = properties[i] # in case of other naming conventions - - return data - - def data(self, component: str): - """Returns array of data for chosen data component.""" - property_group = self.data_channel(component) - return self.property_group_data(property_group) - - def uncertainty(self, component: str) -> float: - """Returns uncertainty for chosen data component.""" - uid = self.uncertainty_channel(component) - return self.property_group_data(uid) + # @property + # def channels(self) -> list[str]: + # return ["_".join(k.split("_")[:2]) for k in self.__dict__ if "channel" in k] + # + # def property_group_data(self, property_group: PropertyGroup): + # """ + # Return dictionary of channel/data. + # + # :param property_group: Property group uid + # """ + # data = {} + # frequencies = self.data_object.channels + # group = next( + # k for k in self.data_object.property_groups if k.uid == property_group.uid + # ) + # property_names = [self.geoh5.get_entity(p)[0].name for p in group.properties] + # properties = [self.geoh5.get_entity(p)[0].values for p in group.properties] + # for i, f in enumerate(frequencies): + # try: + # f_ind = property_names.index( + # next(k for k in property_names if f"{f:.2e}" in k) + # ) # Safer if data was saved with geoapps naming convention + # data[f] = properties[f_ind] + # except StopIteration: + # data[f] = properties[i] # in case of other naming conventions + # + # return data + # + # def data(self, component: str): + # """Returns array of data for chosen data component.""" + # property_group = self.data_channel(component) + # return self.property_group_data(property_group) + # + # def uncertainty(self, component: str) -> float: + # """Returns uncertainty for chosen data component.""" + # uid = self.uncertainty_channel(component) + # return self.property_group_data(uid) diff --git a/simpeg_drivers/params.py b/simpeg_drivers/params.py index 1a64d0a9..49feb246 100644 --- a/simpeg_drivers/params.py +++ b/simpeg_drivers/params.py @@ -163,7 +163,11 @@ def components(self) -> list[str]: @property def active_components(self) -> list[str]: """Return list of active components.""" - return [k for k in self.components if self.component_data(k) is not None] + return [ + k + for k in self.components + if getattr(self, "_".join([k, "channel"])) is not None + ] @property def data(self) -> InversionData: diff --git a/tests/run_tests/driver_mt_test.py b/tests/run_tests/driver_mt_test.py index 7c3f3194..9c76ccb4 100644 --- a/tests/run_tests/driver_mt_test.py +++ b/tests/run_tests/driver_mt_test.py @@ -165,6 +165,7 @@ def test_magnetotellurics_run(tmp_path: Path, max_iterations=1, pytest=True): background_conductivity=100.0, max_global_iterations=max_iterations, initial_beta_ratio=1e3, + sens_wts_threshold=1.0, prctile=100, store_sensitivities="ram", **data_kwargs, From fd69ac1df3931a5feb18e899e4a0540cba55ace2 Mon Sep 17 00:00:00 2001 From: benjamink Date: Wed, 12 Feb 2025 12:09:47 -0800 Subject: [PATCH 08/10] fix tests --- simpeg_drivers/components/data.py | 2 +- tests/data_test.py | 17 ++++------------- 2 files changed, 5 insertions(+), 14 deletions(-) diff --git a/simpeg_drivers/components/data.py b/simpeg_drivers/components/data.py index fe530506..95e9b46d 100644 --- a/simpeg_drivers/components/data.py +++ b/simpeg_drivers/components/data.py @@ -299,7 +299,7 @@ def get_normalizations(self): normalizations = {} for chan in getattr(self.params.data_object, "channels", [None]): normalizations[chan] = {} - for comp in self.params.components: + for comp in self.params.active_components: normalizations[chan][comp] = np.ones(self.mask.sum()) if comp in ["potential", "chargeability"]: normalizations[chan][comp] = 1 diff --git a/tests/data_test.py b/tests/data_test.py index 953711bd..065ecffd 100644 --- a/tests/data_test.py +++ b/tests/data_test.py @@ -204,7 +204,7 @@ def test_get_uncertainty_component(tmp_path: Path): params = get_mvi_params(tmp_path, tmi_uncertainty=1.0) geoh5 = params.geoh5 data = InversionData(geoh5, params) - unc = data.get_data()[2]["tmi"] + unc = params.uncertainties["tmi"] assert len(np.unique(unc)) == 1 assert np.unique(unc)[0] == 1 assert len(unc) == data.entity.n_vertices @@ -214,23 +214,14 @@ def test_normalize(tmp_path: Path): params = get_mvi_params(tmp_path) geoh5 = params.geoh5 data = InversionData(geoh5, params) - len_data = len(data.observed["tmi"]) - data.observed = { - "tmi": np.arange(len_data, dtype=float), - "gz": np.arange(len_data, dtype=float), - } - data.components = list(data.observed.keys()) data.normalizations = data.get_normalizations() test_data = data.normalize(data.observed) - assert np.all( - np.hstack(list(data.normalizations[None].values())).tolist() - == np.repeat([1, -1], len_data) - ) - assert all(test_data["gz"] == (-1 * data.observed["gz"])) + assert all(test_data["tmi"] == params.data["tmi"]) + assert len(test_data) == 1 def test_get_survey(tmp_path: Path): - params = get_mvi_params(tmp_path) + params = get_mvi_params(tmp_path, tmi_uncertainty=1.0) geoh5 = params.geoh5 data = InversionData(geoh5, params) survey = data.create_survey() From ad883aa3cda18f101bfad4c36236a3d169b6c3e8 Mon Sep 17 00:00:00 2001 From: benjamink Date: Wed, 12 Feb 2025 12:40:55 -0800 Subject: [PATCH 09/10] cleanup --- .../frequency_domain/params.py | 51 --------------- .../electromagnetics/time_domain/params.py | 53 ---------------- .../magnetotellurics/params.py | 62 ------------------- .../natural_sources/tipper/params.py | 62 ------------------- 4 files changed, 228 deletions(-) diff --git a/simpeg_drivers/electromagnetics/frequency_domain/params.py b/simpeg_drivers/electromagnetics/frequency_domain/params.py index 39870ef6..f1ead435 100644 --- a/simpeg_drivers/electromagnetics/frequency_domain/params.py +++ b/simpeg_drivers/electromagnetics/frequency_domain/params.py @@ -75,30 +75,6 @@ def unit_conversion(self): } return conversion[self.data_object.unit] - # @property - # def channels(self) -> list[str]: - # return ["_".join(k.split("_")[:2]) for k in self.__dict__ if "channel" in k] - # - # def property_group_data(self, property_group: PropertyGroup): - # """ - # Return dictionary of channel/data. - # - # :param property_group: Property group uid - # """ - # _ = property_group - # channels = self.data_object.channels - # return {k: None for k in channels} - # - # def data(self, component: str): - # """Returns array of data for chosen data component.""" - # property_group = self.data_channel(component) - # return self.property_group_data(property_group) - # - # def uncertainty(self, component: str) -> float: - # """Returns uncertainty for chosen data component.""" - # uid = self.uncertainty_channel(component) - # return self.property_group_data(uid) - class FrequencyDomainElectromagneticsInversionParams(EMDataMixin, BaseInversionData): """ @@ -148,30 +124,3 @@ def unit_conversion(self): "Hertz (Hz)": 1.0, } return conversion[self.data_object.unit] - - # @property - # def channels(self) -> list[str]: - # return ["_".join(k.split("_")[:2]) for k in self.__dict__ if "channel" in k] - # - # def property_group_data(self, property_group: PropertyGroup): - # """ - # Return dictionary of channel/data. - # - # :param property_group: Property group containing TEM data. - # """ - # channels = self.data_object.channels - # group = self.data_object.fetch_property_group(name=property_group.name) - # properties = [self.geoh5.get_entity(p)[0].values for p in group.properties] - # data = {f: properties[i] for i, f in enumerate(channels)} - # - # return data - # - # def data(self, component: str): - # """Returns array of data for chosen data component.""" - # property_group = self.data_channel(component) - # return self.property_group_data(property_group) - # - # def uncertainty(self, component: str) -> float: - # """Returns uncertainty for chosen data component.""" - # uid = self.uncertainty_channel(component) - # return self.property_group_data(uid) diff --git a/simpeg_drivers/electromagnetics/time_domain/params.py b/simpeg_drivers/electromagnetics/time_domain/params.py index e903f34b..6d9795c7 100644 --- a/simpeg_drivers/electromagnetics/time_domain/params.py +++ b/simpeg_drivers/electromagnetics/time_domain/params.py @@ -67,27 +67,6 @@ def unit_conversion(self): } return conversion[self.data_object.unit] - # def property_group_data(self, property_group: PropertyGroup): - # """ - # Return dictionary of channel/data. - # - # :param property_group: Property group containing em data. - # """ - # _ = property_group - # channels = self.data_object.channels - # return {k: None for k in channels} - - # - # def data(self, component: str): - # """Returns array of data for chosen data component.""" - # property_group = self.data_channel(component) - # return self.property_group_data(property_group) - # - # def uncertainty(self, component: str) -> float: - # """Returns uncertainty for chosen data component.""" - # uid = self.uncertainty_channel(component) - # return self.property_group_data(uid) - class TimeDomainElectromagneticsInversionParams(EMDataMixin, BaseInversionData): """ @@ -129,35 +108,3 @@ def unit_conversion(self): "Microseconds (us)": 1e-6, } return conversion[self.data_object.unit] - - # def property_group_data(self, property_group: PropertyGroup): - # """ - # Return dictionary of channel/data. - # - # :param property_group: Property group containing TEM data. - # """ - # data = {} - # channels = self.data_object.channels - # group = self.data_object.fetch_property_group(name=property_group.name) - # property_names = [self.geoh5.get_entity(p)[0].name for p in group.properties] - # properties = [self.geoh5.get_entity(p)[0].values for p in group.properties] - # for i, f in enumerate(channels): - # try: - # f_ind = property_names.index( - # next(k for k in property_names if f"{f:.2e}" in k) - # ) # Safer if data was saved with geoapps naming convention - # data[f] = properties[f_ind] - # except StopIteration: - # data[f] = properties[i] # in case of other naming conventions - # - # return data - # - # def data(self, component: str): - # """Returns array of data for chosen data component.""" - # property_group = self.data_channel(component) - # return self.property_group_data(property_group) - # - # def uncertainty(self, component: str) -> float: - # """Returns uncertainty for chosen data component.""" - # uid = self.uncertainty_channel(component) - # return self.property_group_data(uid) diff --git a/simpeg_drivers/natural_sources/magnetotellurics/params.py b/simpeg_drivers/natural_sources/magnetotellurics/params.py index b31e074a..3fbcb8af 100644 --- a/simpeg_drivers/natural_sources/magnetotellurics/params.py +++ b/simpeg_drivers/natural_sources/magnetotellurics/params.py @@ -63,26 +63,6 @@ class MagnetotelluricsForwardParams(EMDataMixin, BaseForwardData): def channels(self) -> list[str]: return ["_".join(k.split("_")[:2]) for k in self.__dict__ if "channel" in k] - # def property_group_data(self, property_group: PropertyGroup): - # """ - # Return dictionary of channel/data. - # - # :param property_group: Property group uid - # """ - # _ = property_group - # frequencies = self.data_object.channels - # return {k: None for k in frequencies} - # - # def data(self, component: str): - # """Returns array of data for chosen data component.""" - # property_group = self.data_channel(component) - # return self.property_group_data(property_group) - # - # def uncertainty(self, component: str) -> float: - # """Returns uncertainty for chosen data component.""" - # uid = self.uncertainty_channel(component) - # return self.property_group_data(uid) - class MagnetotelluricsInversionParams(EMDataMixin, BaseInversionData): """ @@ -136,45 +116,3 @@ class MagnetotelluricsInversionParams(EMDataMixin, BaseInversionData): zyy_imag_uncertainty: PropertyGroup | None = None background_conductivity: float | FloatData model_type: str = "Conductivity (S/m)" - - # @property - # def channels(self) -> list[str]: - # return ["_".join(k.split("_")[:2]) for k in self.__dict__ if "channel" in k] - # - # def property_group_data(self, property_group: PropertyGroup): - # """ - # Return dictionary of channel/data. - # - # :param property_group: Property group uid - # """ - # data = {} - # frequencies = self.data_object.channels - # if property_group is None: - # return {k: None for k in frequencies} - # group = next( - # k for k in self.data_object.property_groups if k.uid == property_group.uid - # ) - # property_names = [self.geoh5.get_entity(p)[0].name for p in group.properties] - # properties = [self.geoh5.get_entity(p)[0].values for p in group.properties] - # for i, f in enumerate(frequencies): - # try: - # f_ind = property_names.index( - # next(k for k in property_names if f"{f:.2e}" in k) - # ) # Safer if data was saved with geoapps naming convention - # data[f] = properties[f_ind] - # except StopIteration: - # data[f] = properties[i] # in case of other naming conventions - # - # return data - - # - # @property - # def data(self, component: str): - # """Returns array of data for chosen data component.""" - # property_group = self.data_channel(component) - # return self.property_group_data(property_group) - # @property - # def uncertainty(self, component: str) -> float: - # """Returns uncertainty for chosen data component.""" - # uid = self.uncertainty_channel(component) - # return self.property_group_data(uid) diff --git a/simpeg_drivers/natural_sources/tipper/params.py b/simpeg_drivers/natural_sources/tipper/params.py index f7b65603..cfd25f34 100644 --- a/simpeg_drivers/natural_sources/tipper/params.py +++ b/simpeg_drivers/natural_sources/tipper/params.py @@ -49,30 +49,6 @@ class TipperForwardParams(EMDataMixin, BaseForwardData): background_conductivity: float | FloatData model_type: str = "Conductivity (S/m)" - # @property - # def channels(self) -> list[str]: - # return ["_".join(k.split("_")[:2]) for k in self.__dict__ if "channel" in k] - # - # def property_group_data(self, property_group: PropertyGroup): - # """ - # Return dictionary of channel/data. - # - # :param property_group: Property group uid - # """ - # _ = property_group - # frequencies = self.data_object.channels - # return {k: None for k in frequencies} - # - # def data(self, component: str): - # """Returns array of data for chosen data component.""" - # property_group = self.data_channel(component) - # return self.property_group_data(property_group) - # - # def uncertainty(self, component: str) -> float: - # """Returns uncertainty for chosen data component.""" - # uid = self.uncertainty_channel(component) - # return self.property_group_data(uid) - class TipperInversionParams(EMDataMixin, BaseInversionData): """ @@ -108,41 +84,3 @@ class TipperInversionParams(EMDataMixin, BaseInversionData): tyz_imag_uncertainty: PropertyGroup | None = None background_conductivity: float | FloatData model_type: str = "Conductivity (S/m)" - - # @property - # def channels(self) -> list[str]: - # return ["_".join(k.split("_")[:2]) for k in self.__dict__ if "channel" in k] - # - # def property_group_data(self, property_group: PropertyGroup): - # """ - # Return dictionary of channel/data. - # - # :param property_group: Property group uid - # """ - # data = {} - # frequencies = self.data_object.channels - # group = next( - # k for k in self.data_object.property_groups if k.uid == property_group.uid - # ) - # property_names = [self.geoh5.get_entity(p)[0].name for p in group.properties] - # properties = [self.geoh5.get_entity(p)[0].values for p in group.properties] - # for i, f in enumerate(frequencies): - # try: - # f_ind = property_names.index( - # next(k for k in property_names if f"{f:.2e}" in k) - # ) # Safer if data was saved with geoapps naming convention - # data[f] = properties[f_ind] - # except StopIteration: - # data[f] = properties[i] # in case of other naming conventions - # - # return data - # - # def data(self, component: str): - # """Returns array of data for chosen data component.""" - # property_group = self.data_channel(component) - # return self.property_group_data(property_group) - # - # def uncertainty(self, component: str) -> float: - # """Returns uncertainty for chosen data component.""" - # uid = self.uncertainty_channel(component) - # return self.property_group_data(uid) From 6620740df60e1d7278863d8bcf88d3c5e528ca44 Mon Sep 17 00:00:00 2001 From: benjamink Date: Wed, 12 Feb 2025 12:58:57 -0800 Subject: [PATCH 10/10] clean out unused functions for coverage boost --- simpeg_drivers/utils/surveys.py | 51 --------------------------------- 1 file changed, 51 deletions(-) diff --git a/simpeg_drivers/utils/surveys.py b/simpeg_drivers/utils/surveys.py index 7dce4bd6..992de52d 100644 --- a/simpeg_drivers/utils/surveys.py +++ b/simpeg_drivers/utils/surveys.py @@ -143,54 +143,3 @@ def get_unique_locations(survey: BaseSurvey) -> np.ndarray: locations = survey.receiver_locations return np.unique(locations, axis=0) - - -def is_outlier(population: list[float | int], value: float, n_std: int | float = 3): - """ - use a standard deviation threshold to determine if value is an outlier for the population. - - :param population: list of values. - :param value: single value to detect outlier status - :param n_std (optional): - - :return True if the deviation of value from the mean exceeds the standard deviation threshold. - """ - mean = np.mean(population) - std = np.std(population) - deviation = np.abs(mean - value) - return deviation > n_std * std - - -def next_neighbor(tree: cKDTree, point: list[float], nodes: list[int], n: int = 3): - """ - Returns smallest distance neighbor that has not yet been traversed. - - :param: tree: kd-tree computed for the point cloud of possible neighbors. - :param: point: Current point being traversed. - :param: nodes: Traversed point ids. - """ - distances, neighbors = tree.query(point, n) - new_ids = new_neighbors(distances, neighbors, nodes) - if any(new_ids): - distances = distances[new_ids] - neighbors = neighbors[new_ids] - next_id = np.argmin(distances) - return distances[next_id], neighbors[next_id] - - else: - return next_neighbor(tree, point, nodes, n + 3) - - -def new_neighbors(distances: np.ndarray, neighbors: np.ndarray, nodes: list[int]): - """ - Index into neighbor arrays excluding zero distance and past neighbors. - - :param: distances: previously computed distances - :param: neighbors: Possible neighbors - :param: nodes: Traversed point ids. - """ - ind = [ - i in nodes if distances[neighbors.tolist().index(i)] != 0 else False - for i in neighbors - ] - return np.where(ind)[0].tolist()