Source code for dataio.export.rms.fluid_contact_outlines
from __future__ import annotations
from pathlib import Path
from typing import TYPE_CHECKING, Any, Final
from fmu.dataio._export import ExportConfig, export_with_metadata
from fmu.dataio._logging import null_logger
from fmu.dataio.exceptions import ValidationError
from fmu.dataio.export._base import SimpleExportBase
from fmu.dataio.export._export_result import ExportResult, ExportResultItem
from fmu.dataio.export.rms._utils import (
get_polygons_in_general2d_folder,
get_rms_project_units,
list_folder_names_in_general2d_folder,
validate_name_in_stratigraphy,
)
from fmu.datamodels.common.enums import Classification
from fmu.datamodels.fmu_results.data import FluidContact
from fmu.datamodels.fmu_results.enums import (
Content,
DomainReference,
FluidContactType,
VerticalDomain,
)
from fmu.datamodels.standard_results.enums import StandardResultName
if TYPE_CHECKING:
import xtgeo
_logger: Final = null_logger(__name__)
GENERAL2D_FOLDER = "fluid_contact_outlines"
class _ExportFluidContactOutlines(SimpleExportBase):
def __init__(self, project: Any) -> None:
super().__init__()
_logger.debug("Process data, establish state prior to export.")
self.project = project
self._contact_outlines = self._get_contact_outlines()
self._unit = "m" if get_rms_project_units(project) == "metric" else "ft"
_logger.debug("Process data... DONE")
def _get_contacts(self) -> list[FluidContactType]:
"""
Get FluidContactTypes from available subfolder names in the main folder.
Folders with invalid contact names will be skipped. If no valid contact
names are found an error is raised.
"""
contact_folders = list_folder_names_in_general2d_folder(
self.project, folder_path=[GENERAL2D_FOLDER]
)
valid_contact_folders = []
for contact in contact_folders:
try:
valid_contact_folders.append(FluidContactType(contact))
except ValueError:
_logger.info(f"{contact} is not a valid contact name, skipping folder.")
continue
_logger.debug(f"Found valid contact folders {valid_contact_folders}.")
return valid_contact_folders
def _contact_folder_present(self) -> bool:
"""Check if the main contact folder is present in General 2D data"""
return GENERAL2D_FOLDER in self.project.general2d_data.folders
def _get_contact_outlines(
self,
) -> dict[FluidContactType, list[xtgeo.Polygons]]:
"""
Get a dictionary with fluid contact outline polygons per contact folder
found in the main folder.
"""
if self._contact_folder_present() and (contacts := self._get_contacts()):
return {
contact: get_polygons_in_general2d_folder(
self.project, folder_path=[GENERAL2D_FOLDER, contact.value]
)
for contact in contacts
}
raise ValueError(
"Could not detect any fluid contact outlines from RMS. "
f"Ensure the folder '{GENERAL2D_FOLDER}' exists in the "
"'General 2D data' folder, and that it contains minimum one subfolder "
f"with a valid contact name: {list(FluidContactType.__members__)}. The "
"contact outlines should be contained inside these subfolders."
)
def _get_export_config(self, contact: FluidContactType, name: str) -> ExportConfig:
"""Export config for the standard result."""
return (
ExportConfig.builder()
.content(Content.fluid_contact, FluidContact(contact=contact))
.domain(VerticalDomain.depth, DomainReference.msl)
.file_config(
name=name,
subfolder=f"{StandardResultName.fluid_contact_outline.value}/{contact.value}",
)
.access(Classification.internal, rep_include=True)
.global_config(self._config)
.standard_result(StandardResultName.fluid_contact_outline)
.build()
)
def _export_contact_outline(
self, contact: FluidContactType, pol: xtgeo.Polygons
) -> ExportResultItem:
"""Export a fluid contact outline as a standard result"""
export_config = self._get_export_config(contact, pol.name)
absolute_export_path = export_with_metadata(export_config, pol)
_logger.debug("Surface exported to: %s", absolute_export_path)
return ExportResultItem(
absolute_path=Path(absolute_export_path),
)
def _export_data_as_standard_result(self) -> ExportResult:
"""Do the actual export using dataio setup."""
result_items = []
for contact, polygons in self._contact_outlines.items():
for pol in polygons:
result_items.append(self._export_contact_outline(contact, pol))
return ExportResult(items=result_items)
def _validate_data_pre_export(self) -> None:
"""Data validations."""
# TODO: Check that all contacts have positive values
for contact, polygons in self._contact_outlines.items():
for pol in polygons:
try:
validate_name_in_stratigraphy(pol.name, self._config)
except ValidationError as err:
raise ValidationError(
f"Error detected for polygon '{pol.name}' in the contact "
f"folder '{contact.value}'. Detailed information:\n{str(err)}"
) from err