ENH: Apply ERT env variables in FMU runs

examples/run_examples.sh

@@ -2,8 +2,6 @@
 set -e
 # The examples scripts are ran and included in the documentation!
 
-export RUN_DATAIO_EXAMPLES=1  # this is important when running examples!
-
 current=$PWD
 
 # empty current results
@@ -14,6 +12,14 @@ rm -rf examples/s/d/nn/xcase/iter-0/*
 # Note! run from RUNPATH, NOT being inside RMS but need RUN_DATAIO_EXAMPLES env!
 cd $current/examples/s/d/nn/xcase/realization-0/iter-0/rms/bin
 
+# fake an ERT FMU run
+export _ERT_EXPERIMENT_ID=6a8e1e0f-9315-46bb-9648-8de87151f4c7
+export _ERT_ENSEMBLE_ID=b027f225-c45d-477d-8f33-73695217ba14
+export _ERT_SIMULATION_MODE=test_run
+export _ERT_ITERATION_NUMBER=0
+export _ERT_REALIZATION_NUMBER=0
+export _ERT_RUNPATH=$current/examples/s/d/nn/xcase/realization-0/iter-0
+
 python export_faultpolygons.py
 python export_propmaps.py
 
@@ -25,6 +31,10 @@ python export_volumetables.py
 # Emulate FMU run with 3 realizations and export data to disk
 for num in 0 1 9; do
     cd $current/examples/s/d/nn/xcase/realization-${num}/iter-0/rms/bin
+
+    export _ERT_REALIZATION_NUMBER=$num
+    export _ERT_RUNPATH=${current}/examples/s/d/nn/xcase/realization-${num}/iter-0
+
     python export_a_surface.py
 done
 

examples/s/d/nn/xcase/realization-0/iter-0/any/bin/export_grid3d.py

@@ -6,8 +6,8 @@
 import xtgeo
 from fmu.config import utilities as ut
 
+logging.basicConfig(level=logging.DEBUG)
 logger = logging.getLogger(__name__)
-logger.setLevel(logging.WARNING)
 
 CFG = ut.yaml_load("../../fmuconfig/output/global_variables.yml")
 

pyproject.toml

@@ -89,10 +89,9 @@ match = '(?!(test_|_)).*\.py'
 
 [tool.ruff]
 line-length = 88
+exclude = ["version.py"]
 [tool.ruff.lint]
-ignore = [
-    "C901",
-]
+ignore = ["C901"]
 select = [
     "C",
     "E",

src/fmu/dataio/_definitions.py

@@ -2,6 +2,7 @@
 from __future__ import annotations
 
 from dataclasses import dataclass, field
+from enum import Enum, unique
 from typing import Final
 
 SCHEMA: Final = (
@@ -11,6 +12,10 @@
 SOURCE: Final = "fmu"
 
 
+class ValidationError(ValueError, KeyError):
+    """Raise error while validating."""
+
+
 @dataclass
 class _ValidFormats:
     surface: dict = field(
@@ -123,3 +128,39 @@ class _ValidFormats:
     "case_symlink_realization": "To case/share, with symlinks on realizations level",
     "preprocessed": "To share/preprocessed; from interactive runs but re-used later",
 }
+
+
+@unique
+class FmuContext(Enum):
+    """Use a Enum class for fmu_context entries."""
+
+    REALIZATION = "To realization-N/iter_M/share"
+    CASE = "To casename/share, but will also work on project disk"
+    CASE_SYMLINK_REALIZATION = "To case/share, with symlinks on realizations level"
+    PREPROCESSED = "To share/preprocessed; from interactive runs but re-used later"
+    NON_FMU = "Not ran in a FMU setting, e.g. interactive RMS"
+
+    @classmethod
+    def has_key(cls, key: str) -> bool:
+        return key.upper() in cls._member_names_
+
+    @classmethod
+    def list_valid(cls) -> dict:
+        return {member.name: member.value for member in cls}
+
+    @classmethod
+    def get(cls, key: FmuContext | str) -> FmuContext:
+        """Get the enum member with a case-insensitive key."""
+        if isinstance(key, cls):
+            key_upper = key.name
+        elif isinstance(key, str):
+            key_upper = key.upper()
+        else:
+            raise ValidationError("The input must be a str or FmuContext instance")
+
+        if not cls.has_key(key_upper):
+            raise ValidationError(
+                f"Invalid key <{key_upper}>. Valid keys: {cls.list_valid().keys()}"
+            )
+
+        return cls[key_upper]

src/fmu/dataio/_filedata_provider.py

@@ -8,9 +8,10 @@
 from copy import deepcopy
 from dataclasses import dataclass, field
 from pathlib import Path
-from typing import Any, Final, Optional
+from typing import Any, Final, Literal, Optional
 from warnings import warn
 
+from ._definitions import FmuContext
 from ._logging import null_logger
 
 logger: Final = null_logger(__name__)
@@ -60,8 +61,6 @@ def __post_init__(self) -> None:
         self.forcefolder_is_absolute = False
         self.subfolder = self.dataio.subfolder
 
-        self.fmu_context = self.dataio._usecontext  # may be None!
-
         logger.info("Initialize %s", self.__class__)
 
     def derive_filedata(self) -> None:
@@ -161,25 +160,32 @@ def _get_path(self) -> tuple[Path, Path | None]:
         """Construct and get the folder path(s)."""
         linkdest = None
 
-        dest = self._get_path_generic(mode=self.fmu_context, allow_forcefolder=True)
+        dest = self._get_path_generic(
+            mode=self.dataio.fmu_context, allow_forcefolder=True
+        )
 
-        if self.fmu_context == "case_symlink_realization":
+        if self.dataio.fmu_context == FmuContext.CASE_SYMLINK_REALIZATION:
             linkdest = self._get_path_generic(
-                mode="realization", allow_forcefolder=False, info=self.fmu_context
+                mode=FmuContext.REALIZATION,
+                allow_forcefolder=False,
+                info=self.dataio.fmu_context.name,
             )
 
         return dest, linkdest
 
     def _get_path_generic(
-        self, mode: str = "realization", allow_forcefolder: bool = True, info: str = ""
+        self,
+        mode: Literal[FmuContext.REALIZATION, FmuContext.PREPROCESSED],
+        allow_forcefolder: bool = True,
+        info: str = "",
     ) -> Path:
         """Generically construct and get the folder path and verify."""
         dest = None
 
         outroot = deepcopy(self.rootpath)
 
         logger.info("FMU context is %s", mode)
-        if mode == "realization":
+        if mode == FmuContext.REALIZATION:
             if self.realname:
                 outroot = outroot / self.realname  # TODO: if missing self.realname?
 
@@ -188,14 +194,14 @@ def _get_path_generic(
 
         outroot = outroot / "share"
 
-        if mode == "preprocessed":
+        if mode == FmuContext.PREPROCESSED:
             outroot = outroot / "preprocessed"
             if self.dataio.forcefolder and self.dataio.forcefolder.startswith("/"):
                 raise ValueError(
                     "Cannot use absolute path to 'forcefolder' with preprocessed data"
                 )
 
-        if mode != "preprocessed":
+        if mode != FmuContext.PREPROCESSED:
             if self.dataio.is_observation:
                 outroot = outroot / "observations"
             else: