ENH: Simplify and rename the export_volumetrics function

src/fmu/dataio/_utils.py

@@ -382,3 +382,16 @@ def get_geometry_ref(
     relpath = gmeta["file"]["relative_path"]
 
     return geom_name, relpath
+
+
+def _load_config_from_path(config_path: Path) -> dict[str, Any]:
+    """Retrieve the global config data by reading the global config file."""
+    logger.debug("Set global config...")
+
+    if not isinstance(config_path, Path):
+        raise ValueError("The config_path argument needs to be a path")
+
+    if not config_path.is_file():
+        raise FileNotFoundError(f"Cannot find file for global config: {config_path}")
+
+    return ut.yaml_load(config_path)

src/fmu/dataio/export/rms/__init__.py

@@ -1,3 +1,3 @@
-from .volumetrics import export_rms_volumetrics, export_volumetrics
+from .inplace_volumes import export_inplace_volumes, export_rms_volumetrics
 
-__all__ = ["export_volumetrics", "export_rms_volumetrics"]
+__all__ = ["export_inplace_volumes", "export_rms_volumetrics"]

src/fmu/dataio/export/rms/_conditional_rms_imports.py

@@ -10,7 +10,7 @@
 _logger = null_logger(__name__)
 
 
-def import_rms_package() -> dict[str, Any] | None:
+def import_rms_package() -> tuple[Any, Any]:
     """
     Attempts to import the 'rmsapi' package first. If 'rmsapi' is not available,
     it attempts to import the 'roxar' package while suppressing deprecation warnings.
@@ -21,7 +21,6 @@ def import_rms_package() -> dict[str, Any] | None:
         import rmsapi
         import rmsapi.jobs as jobs
 
-        return {"rmsapi": rmsapi, "jobs": jobs}
     except ImportError:
         try:
             with warnings.catch_warnings():
@@ -31,13 +30,14 @@ def import_rms_package() -> dict[str, Any] | None:
                 import roxar as rmsapi
                 import roxar.jobs as jobs
 
-                return {"rmsapi": rmsapi, "jobs": jobs}
         except ImportError:
             raise ImportError(
                 "Neither 'roxar' nor 'rmsapi' are available. You have to be inside "
                 "RMS to use this function."
             )
 
+    return (rmsapi, jobs)
+
 
 if TYPE_CHECKING:
     import rmsapi

src/fmu/dataio/export/rms/_utils.py

@@ -0,0 +1,53 @@
+from __future__ import annotations
+
+from typing import TYPE_CHECKING, Any, Final
+
+from packaging.version import parse as versionparse
+
+from fmu.dataio._logging import null_logger
+
+from ._conditional_rms_imports import import_rms_package
+
+if TYPE_CHECKING:
+    from packaging.version import Version
+
+
+logger: Final = null_logger(__name__)
+
+
+rmsapi, _ = import_rms_package()
+
+RMS_API_PROJECT_MAPPING = {
+    "1.10": "14.2",
+    "1.9": "14.1",
+    "1.7": "13.1",
+}
+
+
+def _get_rmsapi_version() -> Version:
+    """Get the rmsapi version"""
+    return versionparse(rmsapi.__version__)
+
+
+def _check_rmsapi_version(minimum_version: str) -> None:
+    """Check if we are working in a RMS API, and also check RMS versions"""
+    logger.debug("Check API version...")
+    if minimum_version not in RMS_API_PROJECT_MAPPING:
+        raise ValueError(
+            "The minimum version has not been mapped to a RMS project "
+            "version, it should be added to the 'RMS_API_PROJECT_MAPPING'"
+        )
+    if _get_rmsapi_version() < versionparse(minimum_version):
+        raise RuntimeError(
+            f"You need at least API version {minimum_version} "
+            f"(RMS {RMS_API_PROJECT_MAPPING[minimum_version]}) to use this function."
+        )
+    logger.debug("Check API version... DONE")
+
+
+def _get_rms_project_units(project: Any) -> str:
+    """See if the RMS project is defined in metric or feet."""
+
+    units = project.project_units
+    logger.debug("Units are %s", units)
+    return str(units)

src/fmu/dataio/export/rms/inplace_volumes.py

@@ -0,0 +1,180 @@
+from __future__ import annotations
+
+import warnings
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Any, Final
+
+import pandas as pd
+
+import fmu.dataio as dio
+from fmu.dataio._logging import null_logger
+from fmu.dataio._model.enums import Classification
+from fmu.dataio._utils import _load_config_from_path
+from fmu.dataio.export._decorators import experimental
+from fmu.dataio.export._export_result import ExportResult, ExportResultItem
+from fmu.dataio.export.rms._conditional_rms_imports import import_rms_package
+from fmu.dataio.export.rms._utils import _check_rmsapi_version, _get_rms_project_units
+
+rmsapi, rmsjobs = import_rms_package()
+
+_logger: Final = null_logger(__name__)
+
+# rename columns to FMU standard
+_RENAME_COLUMNS_FROM_RMS: Final = {
+    "Proj. real.": "REAL",
+    "Zone": "ZONE",
+    "Segment": "REGION",
+    "Boundary": "LICENSE",
+    "Facies": "FACIES",
+    "BulkOil": "BULK_OIL",
+    "NetOil": "NET_OIL",
+    "PoreOil": "PORV_OIL",
+    "HCPVOil": "HCPV_OIL",
+    "STOIIP": "STOIIP_OIL",
+    "AssociatedGas": "ASSOCIATEDGAS_OIL",
+    "BulkGas": "BULK_GAS",
+    "NetGas": "NET_GAS",
+    "PoreGas": "PORV_GAS",
+    "HCPVGas": "HCPV_GAS",
+    "GIIP": "GIIP_GAS",
+    "AssociatedLiquid": "ASSOCIATEDOIL_GAS",
+    "Bulk": "BULK_TOTAL",
+    "Net": "NET_TOTAL",
+    "Pore": "PORV_TOTAL",
+}
+
+
+@dataclass
+class _ExportVolumetricsRMS:
+    project: Any
+    grid_name: str
+    volume_job_name: str
+    config_path: str | Path = "../../fmuconfig/output/global_variables.yml"
+
+    def __post_init__(self) -> None:
+        _logger.debug("Process data, estiblish state prior to export.")
+        self._config = _load_config_from_path(Path(self.config_path))
+        self._volume_job = self._get_rms_volume_job_settings()
+        self._volume_table_name = self._read_volume_table_name_from_job()
+        self._dataframe = self._voltable_as_dataframe()
+        _logger.debug("Process data... DONE")
+
+    @property
+    def _classification(self) -> Classification:
+        """Get default classification."""
+        return Classification.restricted
+
+    def _get_rms_volume_job_settings(self) -> dict:
+        """Get information out from the RMS job API."""
+        _logger.debug("RMS VOLJOB settings...")
+        return rmsjobs.Job.get_job(
+            owner=["Grid models", self.grid_name, "Grid"],
+            type="Volumetrics",
+            name=self.volume_job_name,
+        ).get_arguments()
+
+    def _read_volume_table_name_from_job(self) -> str:
+        """Read the volume table name from RMS."""
+        _logger.debug("Read volume table name from RMS...")
+        voltable = self._volume_job.get("Report")
+        if isinstance(voltable, list):
+            voltable = voltable[0]
+
+        volume_table_name = voltable.get("ReportTableName")
+        if not volume_table_name:
+            raise RuntimeError(
+                "You need to configure output to Report file: Report table "
+                "in the volumetric job. Provide a table name and rerun the job."
+            )
+
+        _logger.debug("The volume table name is %s", volume_table_name)
+        return volume_table_name
+
+    def _voltable_as_dataframe(self) -> pd.DataFrame:
+        """Convert table to pandas dataframe"""
+        _logger.debug("Read values and convert to pandas dataframe...")
+
+        dict_values = (
+            self.project.volumetric_tables[self._volume_table_name]
+            .get_data_table()
+            .to_dict()
+        )
+        _logger.debug("Dict values are: %s", dict_values)
+        return (
+            pd.DataFrame.from_dict(dict_values)
+            .rename(columns=_RENAME_COLUMNS_FROM_RMS)
+            .drop("REAL", axis=1, errors="ignore")
+        )
+
+    def _export_volume_table(self) -> ExportResult:
+        """Do the actual volume table export using dataio setup."""
+
+        edata = dio.ExportData(
+            config=self._config,
+            content="volumes",
+            unit="m3" if _get_rms_project_units(self.project) == "metric" else "ft3",
+            vertical_domain="depth",
+            domain_reference="msl",
+            subfolder="volumes",
+            classification=self._classification,
+            name=self.grid_name,
+            rep_include=False,
+        )
+        absolute_export_path = edata.export(self._dataframe)
+        _logger.debug("Volume result to: %s", absolute_export_path)
+        return ExportResult(
+            items=[
+                ExportResultItem(
+                    absolute_path=Path(absolute_export_path),
+                )
+            ],
+        )
+
+    def export(self) -> ExportResult:
+        """Export the volume table."""
+        return self._export_volume_table()
+
+
+@experimental
+def export_inplace_volumes(
+    project: Any,
+    grid_name: str,
+    volume_job_name: str,
+    config_path: str | Path = "../../fmuconfig/output/global_variables.yml",
+) -> ExportResult:
+    """Simplified interface when exporting volume tables (and assosiated data) from RMS.
+
+    Args:
+        project: The 'magic' project variable in RMS.
+        grid_name: Name of 3D grid model in RMS.
+        volume_job_name: Name of the volume job.
+        config_path: Optional. Path to the global_variables file. As default, it assumes
+            the current standard in FMU:
+            ``'../../fmuconfig/output/global_variables.yml'``
+
+    Note:
+        This function is experimental and may change in future versions.
+    """
+
+    _check_rmsapi_version(minimum_version="1.7")
+
+    return _ExportVolumetricsRMS(
+        project,
+        grid_name,
+        volume_job_name,
+        config_path=config_path,
+    ).export()
+
+
+# keep the old name for now but not log (will be removed soon as we expect close to
+# zero usage so far)
+def export_rms_volumetrics(*args, **kwargs) -> ExportResult:  # type: ignore
+    """Deprecated function. Use export_inplace_volumes instead."""
+    warnings.warn(
+        "export_rms_volumetrics is deprecated and will be removed in a future release. "
+        "Use export_inplace_volumes instead.",
+        FutureWarning,
+        stacklevel=2,
+    )
+    return export_inplace_volumes(*args, **kwargs)