diff --git a/LoopStructural/modelling/core/_model_exporter.py b/LoopStructural/modelling/core/_model_exporter.py new file mode 100644 index 00000000..7bdfc679 --- /dev/null +++ b/LoopStructural/modelling/core/_model_exporter.py @@ -0,0 +1,129 @@ +"""Surface/block-model export logic for GeologicalModel (see API.md). + +Extracted from GeologicalModel to separate export/visualization data prep +from feature-container orchestration. GeologicalModel's @public_api methods +(``get_fault_surfaces``, ``get_stratigraphic_surfaces``, ``get_block_model``, +``save``) stay defined directly on the class -- their __qualname__ is part +of the CI-checked stable API surface -- and delegate to the staticmethods +here. +""" + +import pathlib + +from ...geometry import StructuredGrid +from ...utils import getLogger + +logger = getLogger(__name__) + + +class ModelExporter: + @staticmethod + def get_fault_surfaces(model, faults=None): + if faults is None: + faults = [] + surfaces = [] + if len(faults) == 0: + faults = model.fault_names() + + for f in faults: + surfaces.extend(model.get_feature_by_name(f).surfaces([0], model.bounding_box)) + return surfaces + + @staticmethod + def get_stratigraphic_surfaces(model, units=None, bottoms=True): + if units is None: + units = [] + ## TODO change the stratigraphic column to its own class and have methods to get the relevant surfaces + surfaces = [] + units = [] + if model.stratigraphic_column is None: + return [] + units = model.stratigraphic_column.get_isovalues() + units_for_group = {} + for name, u in units.items(): + if u['group'] not in model: + logger.warning(f"Group {u['group']} not found in model") + continue + if u['group'] not in units_for_group: + units_for_group[u['group']] = [] + u['name'] = name + units_for_group[u['group']].append(u) + for group, us in units_for_group.items(): + feature = model.get_feature_by_name(group) + values = [u['value'] for u in us] + colours = [u['colour'] for u in us] + names = [u['name'] for u in us] + surfaces.extend( + feature.surfaces(values, model.bounding_box, name=names, colours=colours) + ) + + return surfaces + + @staticmethod + def get_block_model(model, name='block model'): + # NOTE: bounding_box.structured_grid() returns loop_common's + # interpolation-support StructuredGrid (no properties dict); use + # LoopStructural's own geometry StructuredGrid for storing values. + grid = StructuredGrid( + origin=model.bounding_box.origin, + step_vector=model.bounding_box.step_vector, + nsteps=model.bounding_box.nsteps, + name=name, + ) + + grid.cell_properties['stratigraphy'] = model.evaluate_model( + model.rescale(model.bounding_box.cell_centres()) + ) + return grid, model.stratigraphic_ids() + + @staticmethod + def save( + model, + filename: str, + block_model: bool = True, + stratigraphic_surfaces=True, + fault_surfaces=True, + stratigraphic_data=True, + fault_data=True, + ): + path = pathlib.Path(filename) + extension = path.suffix + parent = path.parent + name = path.stem + stratigraphic_surfaces = model.get_stratigraphic_surfaces() + if fault_surfaces: + for s in model.get_fault_surfaces(): + ## geoh5 can save everything into the same file + if extension == ".geoh5" or extension == '.omf': + s.save(filename) + else: + s.save(f'{parent}/{name}_{s.name}{extension}') + if stratigraphic_surfaces: + for s in model.get_stratigraphic_surfaces(): + if extension == ".geoh5" or extension == '.omf': + s.save(filename) + else: + s.save(f'{parent}/{name}_{s.name}{extension}') + if block_model: + grid, _ids = model.get_block_model() + if extension == ".geoh5" or extension == '.omf': + grid.save(filename) + else: + grid.save(f'{parent}/{name}_block_model{extension}') + if stratigraphic_data and model.stratigraphic_column is not None: + for group in model.stratigraphic_column: + if group == "faults": + continue + for data in model.__getitem__(group).get_data(): + if extension == ".geoh5" or extension == '.omf': + data.save(filename) + else: + data.save(f'{parent}/{name}_{group}_data{extension}') + if fault_data: + for f in model.fault_names(): + for d in model.__getitem__(f).get_data(): + if extension == ".geoh5" or extension == '.omf': + + d.save(filename) + else: + d.save(f'{parent}/{name}_{group}{extension}') diff --git a/LoopStructural/modelling/core/geological_model.py b/LoopStructural/modelling/core/geological_model.py index 8511e76a..5b050cb8 100644 --- a/LoopStructural/modelling/core/geological_model.py +++ b/LoopStructural/modelling/core/geological_model.py @@ -3,7 +3,6 @@ """ from __future__ import annotations -import pathlib import warnings import numpy as np @@ -11,7 +10,7 @@ from LoopStructural import LoopStructuralConfig -from ...geometry import BoundingBox, StructuredGrid +from ...geometry import BoundingBox from ...modelling.features import ( BaseFeature, FeatureType, @@ -43,6 +42,7 @@ convert_feature_to_structural_frame as _convert_feature_to_structural_frame, ) from ._feature_registry import FeatureBuilderRegistry +from ._model_exporter import ModelExporter from ._model_serializer import ModelSerializer from .stratigraphic_column import StratigraphicColumn @@ -2259,62 +2259,15 @@ def stratigraphic_ids(self): @public_api(tier="stable") def get_fault_surfaces(self, faults: list[str] | None = None): - if faults is None: - faults = [] - surfaces = [] - if len(faults) == 0: - faults = self.fault_names() - - for f in faults: - surfaces.extend(self.get_feature_by_name(f).surfaces([0], self.bounding_box)) - return surfaces + return ModelExporter.get_fault_surfaces(self, faults=faults) @public_api(tier="stable") def get_stratigraphic_surfaces(self, units: list[str] | None = None, bottoms: bool = True): - if units is None: - units = [] - ## TODO change the stratigraphic column to its own class and have methods to get the relevant surfaces - surfaces = [] - units = [] - if self.stratigraphic_column is None: - return [] - units = self.stratigraphic_column.get_isovalues() - units_for_group = {} - for name, u in units.items(): - if u['group'] not in self: - logger.warning(f"Group {u['group']} not found in model") - continue - if u['group'] not in units_for_group: - units_for_group[u['group']] = [] - u['name'] = name - units_for_group[u['group']].append(u) - for group, us in units_for_group.items(): - feature = self.get_feature_by_name(group) - values = [u['value'] for u in us] - colours = [u['colour'] for u in us] - names = [u['name'] for u in us] - surfaces.extend( - feature.surfaces(values, self.bounding_box, name=names, colours=colours) - ) - - return surfaces + return ModelExporter.get_stratigraphic_surfaces(self, units=units, bottoms=bottoms) @public_api(tier="stable") def get_block_model(self, name='block model'): - # NOTE: bounding_box.structured_grid() returns loop_common's - # interpolation-support StructuredGrid (no properties dict); use - # LoopStructural's own geometry StructuredGrid for storing values. - grid = StructuredGrid( - origin=self.bounding_box.origin, - step_vector=self.bounding_box.step_vector, - nsteps=self.bounding_box.nsteps, - name=name, - ) - - grid.cell_properties['stratigraphy'] = self.evaluate_model( - self.rescale(self.bounding_box.cell_centres()) - ) - return grid, self.stratigraphic_ids() + return ModelExporter.get_block_model(self, name=name) @public_api(tier="stable") def save( @@ -2326,47 +2279,15 @@ def save( stratigraphic_data=True, fault_data=True, ): - path = pathlib.Path(filename) - extension = path.suffix - parent = path.parent - name = path.stem - stratigraphic_surfaces = self.get_stratigraphic_surfaces() - if fault_surfaces: - for s in self.get_fault_surfaces(): - ## geoh5 can save everything into the same file - if extension == ".geoh5" or extension == '.omf': - s.save(filename) - else: - s.save(f'{parent}/{name}_{s.name}{extension}') - if stratigraphic_surfaces: - for s in self.get_stratigraphic_surfaces(): - if extension == ".geoh5" or extension == '.omf': - s.save(filename) - else: - s.save(f'{parent}/{name}_{s.name}{extension}') - if block_model: - grid, _ids = self.get_block_model() - if extension == ".geoh5" or extension == '.omf': - grid.save(filename) - else: - grid.save(f'{parent}/{name}_block_model{extension}') - if stratigraphic_data and self.stratigraphic_column is not None: - for group in self.stratigraphic_column: - if group == "faults": - continue - for data in self.__getitem__(group).get_data(): - if extension == ".geoh5" or extension == '.omf': - data.save(filename) - else: - data.save(f'{parent}/{name}_{group}_data{extension}') - if fault_data: - for f in self.fault_names(): - for d in self.__getitem__(f).get_data(): - if extension == ".geoh5" or extension == '.omf': - - d.save(filename) - else: - d.save(f'{parent}/{name}_{group}{extension}') + ModelExporter.save( + self, + filename, + block_model=block_model, + stratigraphic_surfaces=stratigraphic_surfaces, + fault_surfaces=fault_surfaces, + stratigraphic_data=stratigraphic_data, + fault_data=fault_data, + ) # Wire the built-in feature types up to GeologicalModel.create_and_add_feature