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,71 @@
+from __future__ import annotations
+
+from typing import TYPE_CHECKING, Final, List, Literal, Optional
+
+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"/file_formats/volumes/{VERSION}/inplace_volumes.json"
+)
+
+
+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)."""
+
+    FLUID: Literal["oil", "gas", "water"]
+    ZONE: str
+    REGION: Optional[str] = Field(default=None)
+    FACIES: Optional[str] = Field(default=None)
+    LICENSE: Optional[str] = Field(default=None)
+
+    BULK: float = Field(ge=0.0)
+    NET: Optional[float] = Field(default=None, ge=0.0)
+    PORV: float = Field(ge=0.0)
+    HCPV: Optional[float] = Field(default=None, ge=0.0)
+    STOIIP: Optional[float] = Field(default=None, ge=0.0)
+    GIIP: Optional[float] = Field(default=None, ge=0.0)
+    ASSOCIATEDGAS: Optional[float] = Field(default=None, ge=0.0)
+    ASSOCIATEDOIL: Optional[float] = Field(default=None, 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

@@ -3,12 +3,18 @@
 import unittest.mock as mock
 from pathlib import Path
 
+import jsonschema
 import numpy as np
 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__)
@@ -322,3 +328,46 @@ def test_rms_volumetrics_export_function(
     assert "volumes" in metadata["data"]["content"]
     assert metadata["access"]["classification"] == "restricted"
     assert metadata["data"]["table_index"] == _TABLE_INDEX_COLUMNS
+
+
+@inside_rms
+def test_inplace_volumes_payload_validates_against_model(
+    exportvolumetrics,
+    monkeypatch,
+):
+    """Tests that the volume table exported is validated against the payload result
+    model."""
+
+    out = exportvolumetrics._export_volume_table()
+    with open(out.items[0].absolute_path) as f:
+        df = pd.read_csv(f).replace(np.nan, None).to_dict(orient="records")
+    InplaceVolumesResult.model_validate(df)  # Throws if invalid
+
+
+@inside_rms
+def test_inplace_volumes_payload_validates_against_schema(
+    exportvolumetrics,
+    monkeypatch,
+):
+    """Tests that the volume table exported is validated against the payload result
+    schema."""
+
+    out = exportvolumetrics._export_volume_table()
+    with open(out.items[0].absolute_path) as f:
+        df = pd.read_csv(f).replace(np.nan, None).to_dict(orient="records")
+    jsonschema.validate(instance=df, schema=dump())  # Throws if invalid
+
+
+@inside_rms
+def test_inplace_volumes_export_and_result_columns_are_the_same(
+    mock_project_variable,
+    mocked_rmsapi_modules,
+) -> None:
+    from fmu.dataio.export.rms.inplace_volumes import (
+        _TABLE_INDEX_COLUMNS,
+        _VOLUMETRIC_COLUMNS,
+    )
+
+    export_columns = _TABLE_INDEX_COLUMNS + _VOLUMETRIC_COLUMNS
+    result_columns = InplaceVolumesResultRow.model_fields.keys()
+    assert set(export_columns) == set(result_columns)