ENH: Add inplace volumes result and schema

pyproject.toml

@@ -51,6 +51,7 @@ dev = [
     "coverage>=4.1",
     "ert",
     "hypothesis",
+    "jsonschema",
     "mypy",
     "pandas-stubs",
     "pyarrow-stubs",

src/fmu/dataio/_products/inplace_volumes.py

@@ -0,0 +1,79 @@
+from __future__ import annotations
+
+from typing import TYPE_CHECKING, Final, List, Literal, Optional, Union
+
+from pydantic import BaseModel, Field, RootModel
+from pydantic.json_schema import GenerateJsonSchema
+
+if TYPE_CHECKING:
+    from typing import Any, Mapping
+
+# These are to be used when creating the 'product' key in metadata.
+VERSION: Final[str] = "0.1.0"
+SCHEMA: Final[str] = (
+    "https://main-fmu-schemas-prod.radix.equinor.com/schemas"
+    f"/products/volumes/{VERSION}/inplace_volumes.json"
+)  # TODO: This URL is as-yet undecided.
+
+
+class InplaceVolumesResultRow(BaseModel):
+    """Represents a row in a static inplace volumes export.
+
+    These fields are the current agreed upon standard result. Changes to this model
+    should increase the version number in a way that corresponds to the schema
+    versioning specification (i.e. they are a patch, minor, or major change)."""
+
+    ZONE: Union[str]
+    REGION: Union[str]
+    FACIES: Optional[Union[str]] = Field(default=None)
+    LICENSE: Optional[Union[str, int]] = Field(default=None)
+
+    BULK_OIL: Optional[float] = Field(default=None, ge=0.0)
+    NET_OIL: Optional[float] = Field(default=None, ge=0.0)
+    PORV_OIL: Optional[float] = Field(default=None, ge=0.0)
+    HCPV_OIL: Optional[float] = Field(default=None, ge=0.0)
+    STOIIP_OIL: Optional[float] = Field(default=None, ge=0.0)
+    ASSOCIATEDGAS_OIL: Optional[float] = Field(default=None, ge=0.0)
+
+    BULK_GAS: Optional[float] = Field(default=None, ge=0.0)
+    NET_GAS: Optional[float] = Field(default=None, ge=0.0)
+    PORV_GAS: Optional[float] = Field(default=None, ge=0.0)
+    HCPV_GAS: Optional[float] = Field(default=None, ge=0.0)
+    GIIP_GAS: Optional[float] = Field(default=None, ge=0.0)
+    ASSOCIATEDOIL_GAS: Optional[float] = Field(default=None, ge=0.0)
+
+    BULK_TOTAL: float = Field(ge=0.0)
+    NET_TOTAL: Optional[float] = Field(default=None, ge=0.0)
+    PORV_TOTAL: float = Field(ge=0.0)
+
+
+class InplaceVolumesResult(RootModel):
+    """Represents the resultant static inplace volumes csv file, which is naturally a
+    list of rows.
+
+    Consumers who retrieve this csv file must reading it into a json-dictionary
+    equivalent format to validate it against the schema."""
+
+    root: List[InplaceVolumesResultRow]
+
+
+class InplaceVolumesJsonSchema(GenerateJsonSchema):
+    """Implements a schema generator so that some additional fields may be added."""
+
+    def generate(
+        self,
+        schema: Mapping[str, Any],
+        mode: Literal["validation", "serialization"] = "validation",
+    ) -> dict[str, Any]:
+        json_schema = super().generate(schema, mode=mode)
+        json_schema["$schema"] = self.schema_dialect
+        json_schema["$id"] = SCHEMA
+        json_schema["version"] = VERSION
+
+        return json_schema
+
+
+def dump() -> dict[str, Any]:
+    return InplaceVolumesResult.model_json_schema(
+        schema_generator=InplaceVolumesJsonSchema
+    )

tests/test_export_rms/test_export_rms_volumetrics.py

@@ -1,12 +1,19 @@
 """Test the dataio running RMS spesici utility function for volumetrics"""
 
+from copy import deepcopy
 from pathlib import Path
 
+import jsonschema
 import pandas as pd
 import pytest
 
 import fmu.dataio as dataio
 from fmu.dataio._logging import null_logger
+from fmu.dataio._products.inplace_volumes import (
+    InplaceVolumesResult,
+    InplaceVolumesResultRow,
+    dump,
+)
 from tests.utils import inside_rms
 
 logger = null_logger(__name__)
@@ -103,3 +110,81 @@ def test_rms_volumetrics_export_function(
 
     assert "volumes" in metadata["data"]["content"]
     assert metadata["access"]["classification"] == "restricted"
+
+
+@inside_rms
+def test_inplace_volumes_payload_validates_against_model(
+    mock_project_variable, voltable_as_dataframe, rmssetup_with_fmuconfig, monkeypatch
+):
+    """Tests that the volume table exported is validated against the payload result
+    model."""
+
+    import rmsapi
+    import rmsapi.jobs as jobs
+
+    from fmu.dataio.export.rms.inplace_volumes import _ExportVolumetricsRMS
+
+    monkeypatch.chdir(rmssetup_with_fmuconfig)
+
+    assert rmsapi.__version__ == "1.7"
+    assert "Report" in jobs.Job.get_job("whatever").get_arguments.return_value
+
+    instance = _ExportVolumetricsRMS(
+        mock_project_variable,
+        "Geogrid",
+        "geogrid_vol",
+    )
+    assert instance._volume_table_name == "geogrid_volumes"
+
+    # patch the dataframe which originally shall be retrieved from RMS
+    monkeypatch.setattr(instance, "_dataframe", voltable_as_dataframe)
+
+    out = instance._export_volume_table()
+    with open(out.items[0].absolute_path) as f:
+        df = pd.read_csv(f).to_dict(orient="records")
+    InplaceVolumesResult.model_validate(df)
+
+
+@inside_rms
+def test_inplace_volumes_payload_validates_against_schema(
+    mock_project_variable, voltable_as_dataframe, rmssetup_with_fmuconfig, monkeypatch
+):
+    """Tests that the volume table exported is validated against the payload result
+    schema."""
+
+    import rmsapi
+    import rmsapi.jobs as jobs
+
+    from fmu.dataio.export.rms.inplace_volumes import _ExportVolumetricsRMS
+
+    monkeypatch.chdir(rmssetup_with_fmuconfig)
+
+    assert rmsapi.__version__ == "1.7"
+    assert "Report" in jobs.Job.get_job("whatever").get_arguments.return_value
+
+    instance = _ExportVolumetricsRMS(
+        mock_project_variable,
+        "Geogrid",
+        "geogrid_vol",
+    )
+    assert instance._volume_table_name == "geogrid_volumes"
+
+    # patch the dataframe which originally shall be retrieved from RMS
+    monkeypatch.setattr(instance, "_dataframe", voltable_as_dataframe)
+
+    out = instance._export_volume_table()
+    with open(out.items[0].absolute_path) as f:
+        df = pd.read_csv(f).to_dict(orient="records")
+
+    jsonschema.validate(instance=df, schema=dump())
+
+
+@inside_rms
+def test_inplace_volumes_export_and_result_columns_are_the_same() -> None:
+    from fmu.dataio.export.rms.inplace_volumes import _RENAME_COLUMNS_FROM_RMS
+
+    rename_columns = deepcopy(_RENAME_COLUMNS_FROM_RMS)
+    del rename_columns["Proj. real."]
+    export_columns = rename_columns.values()
+    result_columns = InplaceVolumesResultRow.model_fields.keys()
+    assert set(export_columns) == set(result_columns)