Add single node CRG template

src/mdio/builder/template_registry.py

@@ -30,6 +30,9 @@
 from mdio.builder.templates.seismic_3d_poststack import Seismic3DPostStackTemplate
 from mdio.builder.templates.seismic_3d_receiver_gathers import Seismic3DReceiverGathersTemplate
 from mdio.builder.templates.seismic_3d_shot_receiver_line import Seismic3DShotReceiverLineGathersTemplate
+from mdio.builder.templates.seismic_3d_single_node_continuous_receiver_gathers import (
+    Seismic3DSingleNodeContinuousReceiverGathersTemplate,
+)
 from mdio.builder.templates.seismic_3d_streamer_field import Seismic3DStreamerFieldRecordsTemplate
 from mdio.builder.templates.seismic_3d_streamer_shot import Seismic3DStreamerShotGathersTemplate
 
@@ -152,6 +155,9 @@ def _register_default_templates(self) -> None:
         # OBN (Ocean Bottom Node) data
         self.register(Seismic3DObnReceiverGathersTemplate())
 
+        # Single-node continuously recording receivers
+        self.register(Seismic3DSingleNodeContinuousReceiverGathersTemplate())
+
         # Land/OBC shot-receiver data
         self.register(Seismic3DShotReceiverLineGathersTemplate())
 

src/mdio/builder/templates/seismic_3d_single_node_continuous_receiver_gathers.py

@@ -0,0 +1,75 @@
+"""Seismic3DSingleNodeContinuousReceiverGathersTemplate MDIO v1 dataset template."""
+
+from typing import Any
+
+from mdio.builder.schemas.dtype import ScalarType
+from mdio.builder.schemas.v1.units import TimeUnitEnum
+from mdio.builder.schemas.v1.units import TimeUnitModel
+from mdio.builder.schemas.v1.variable import CoordinateMetadata
+from mdio.builder.templates.base import AbstractDatasetTemplate
+from mdio.builder.templates.types import CoordinateSpec
+from mdio.builder.templates.types import DimCoordinateTypes
+from mdio.builder.templates.types import SeismicDataDomain
+
+EPOCH_UNIT = TimeUnitModel(time=TimeUnitEnum.MICROSECOND)
+
+
+class Seismic3DSingleNodeContinuousReceiverGathersTemplate(AbstractDatasetTemplate):
+    """Single-node continuous receiver gathers template.
+
+    Dimensions are ``component``, ``epoch`` (int64 microseconds), and ``time``.
+    Physical coordinates ``group_coord_x`` / ``group_coord_y`` are indexed by ``component``.
+
+    Special handling for the component dimension:
+        If the SEG-Y spec does not contain a ``component`` field, ingestion synthesizes the
+        dimension with constant value 1 for all traces and logs a warning. This is driven by
+        ``synthesize_missing_dims`` and handled by ``ComponentSynthesisStrategy``.
+    """
+
+    def __init__(self, data_domain: SeismicDataDomain = "time"):
+        if data_domain != "time":
+            msg = "SingleNodeContRecvrGathers only supports the time domain, got {data_domain!r}"
+            raise ValueError(msg.format(data_domain=data_domain))
+
+        super().__init__(data_domain=data_domain)
+
+        self._dim_names = ("component", "epoch", self._data_domain)
+        self.synthesize_missing_dims = ("component",)
+        self._physical_coord_names = ("group_coord_x", "group_coord_y")
+        self._logical_coord_names = ()
+        self._var_chunk_shape = (1, 512, 4096)
+        self.add_units({"epoch": EPOCH_UNIT})
+
+    @property
+    def _name(self) -> str:
+        return "SingleNodeContRecvrGathers"
+
+    def _load_dataset_attributes(self) -> dict[str, Any]:
+        return {"surveyType": "3D", "gatherType": "continuous_receiver"}
+
+    def declare_coordinate_specs(self) -> tuple[CoordinateSpec, ...]:
+        """Declare node-position coordinates for single-node continuous gathers."""
+        return (
+            CoordinateSpec(name="group_coord_x", dimensions=("component",), dtype=ScalarType.FLOAT64),
+            CoordinateSpec(name="group_coord_y", dimensions=("component",), dtype=ScalarType.FLOAT64),
+        )
+
+    def declare_dim_coordinate_types(self) -> DimCoordinateTypes:
+        """Declare the data types for each dimension coordinate in this template."""
+        return {
+            "component": ScalarType.UINT8,
+            "epoch": ScalarType.INT64,
+            self._data_domain: ScalarType.INT32,
+        }
+
+    def _add_coordinates(self) -> None:
+        for name in ("component", "epoch", self._data_domain):
+            self._add_dimension_coordinate(name)
+
+        for name in ("group_coord_x", "group_coord_y"):
+            self._builder.add_coordinate(
+                name,
+                dimensions=("component",),
+                data_type=ScalarType.FLOAT64,
+                metadata=CoordinateMetadata(units_v1=self.get_unit_by_key(name)),
+            )

tests/unit/v1/templates/test_seismic_3d_single_node_continuous_receiver_gathers.py

@@ -0,0 +1,143 @@
+"""Unit tests for Seismic3DSingleNodeContinuousReceiverGathersTemplate."""
+
+import pytest
+from tests.unit.v1.helpers import validate_variable
+
+from mdio.builder.schemas.chunk_grid import RegularChunkGrid
+from mdio.builder.schemas.compressors import Blosc
+from mdio.builder.schemas.compressors import BloscCname
+from mdio.builder.schemas.dtype import ScalarType
+from mdio.builder.schemas.dtype import StructuredType
+from mdio.builder.schemas.v1.dataset import Dataset
+from mdio.builder.schemas.v1.units import LengthUnitEnum
+from mdio.builder.schemas.v1.units import LengthUnitModel
+from mdio.builder.schemas.v1.units import TimeUnitEnum
+from mdio.builder.schemas.v1.units import TimeUnitModel
+from mdio.builder.templates.seismic_3d_single_node_continuous_receiver_gathers import (
+    Seismic3DSingleNodeContinuousReceiverGathersTemplate,
+)
+
+UNITS_METER = LengthUnitModel(length=LengthUnitEnum.METER)
+UNITS_SECOND = TimeUnitModel(time=TimeUnitEnum.SECOND)
+
+DATASET_SIZE_MAP = {
+    "component": 1,
+    "epoch": 2048,
+    "time": 4096,
+}
+DATASET_DTYPE_MAP = {
+    "component": "uint8",
+    "epoch": "int64",
+    "time": "int32",
+}
+EXPECTED_COORDINATES = ["group_coord_x", "group_coord_y"]
+EXPECTED_CHUNK_SHAPE = (1, 512, 4096)
+
+
+def _validate_coordinates_headers_trace_mask(dataset: Dataset, headers: StructuredType, domain: str) -> None:
+    """Validate the coordinate, headers, trace_mask variables in the dataset."""
+    assert len(dataset.variables) == 8
+
+    validate_variable(
+        dataset,
+        name="headers",
+        dims=[(k, v) for k, v in DATASET_SIZE_MAP.items() if k != domain],
+        coords=EXPECTED_COORDINATES,
+        dtype=headers,
+    )
+
+    validate_variable(
+        dataset,
+        name="trace_mask",
+        dims=[(k, v) for k, v in DATASET_SIZE_MAP.items() if k != domain],
+        coords=EXPECTED_COORDINATES,
+        dtype=ScalarType.BOOL,
+    )
+
+    for dim_name, dim_size in DATASET_SIZE_MAP.items():
+        validate_variable(
+            dataset,
+            name=dim_name,
+            dims=[(dim_name, dim_size)],
+            coords=[dim_name],
+            dtype=ScalarType(DATASET_DTYPE_MAP[dim_name]),
+        )
+
+    for coord_name in EXPECTED_COORDINATES:
+        coord = validate_variable(
+            dataset,
+            name=coord_name,
+            dims=[("component", DATASET_SIZE_MAP["component"])],
+            coords=[coord_name],
+            dtype=ScalarType.FLOAT64,
+        )
+        assert coord.metadata.units_v1.length == LengthUnitEnum.METER
+
+    epoch = next(v for v in dataset.variables if v.name == "epoch")
+    assert epoch.metadata.units_v1.time == TimeUnitEnum.MICROSECOND
+
+
+class TestSeismic3DSingleNodeContinuousReceiverGathersTemplate:
+    """Unit tests for Seismic3DSingleNodeContinuousReceiverGathersTemplate."""
+
+    def test_configuration(self) -> None:
+        """Test template configuration and attributes."""
+        t = Seismic3DSingleNodeContinuousReceiverGathersTemplate()
+
+        assert t.name == "SingleNodeContRecvrGathers"
+        assert t._dim_names == ("component", "epoch", "time")
+        assert t.calculated_dimension_names == ()
+        assert t.synthesize_missing_dims == ("component",)
+        assert t._physical_coord_names == ("group_coord_x", "group_coord_y")
+        assert t._logical_coord_names == ()
+        assert t._var_chunk_shape == EXPECTED_CHUNK_SHAPE
+        assert t.declare_dim_coordinate_types()["epoch"] == ScalarType.INT64
+
+        assert t._builder is None
+        assert t._dim_sizes == ()
+
+        attrs = t._load_dataset_attributes()
+        assert attrs == {"surveyType": "3D", "gatherType": "continuous_receiver"}
+        assert t.default_variable_name == "amplitude"
+
+    def test_chunk_shape(self) -> None:
+        """Default chunk shape is (1, 512, 4096) over (component, epoch, time)."""
+        t = Seismic3DSingleNodeContinuousReceiverGathersTemplate()
+
+        assert t.full_chunk_shape == EXPECTED_CHUNK_SHAPE
+        assert 1 * 512 * 4096 * 4 == 8 * 1024 * 1024
+
+    def test_build_dataset(self, structured_headers: StructuredType) -> None:
+        """Test building a complete dataset with the template."""
+        t = Seismic3DSingleNodeContinuousReceiverGathersTemplate()
+
+        t.add_units({"group_coord_x": UNITS_METER, "group_coord_y": UNITS_METER})
+        t.add_units({"time": UNITS_SECOND})
+
+        sizes = tuple(DATASET_SIZE_MAP.values())
+        dataset = t.build_dataset("SingleNodeSurvey", sizes=sizes, header_dtype=structured_headers)
+
+        assert dataset.metadata.name == "SingleNodeSurvey"
+        assert dataset.metadata.attributes["surveyType"] == "3D"
+        assert dataset.metadata.attributes["gatherType"] == "continuous_receiver"
+        assert dataset.metadata.attributes["defaultVariableName"] == "amplitude"
+
+        _validate_coordinates_headers_trace_mask(dataset, structured_headers, "time")
+
+        seismic = validate_variable(
+            dataset,
+            name="amplitude",
+            dims=list(DATASET_SIZE_MAP.items()),
+            coords=EXPECTED_COORDINATES,
+            dtype=ScalarType.FLOAT32,
+        )
+        assert isinstance(seismic.compressor, Blosc)
+        assert seismic.compressor.cname == BloscCname.zstd
+        assert isinstance(seismic.metadata.chunk_grid, RegularChunkGrid)
+        assert seismic.metadata.chunk_grid.configuration.chunk_shape == EXPECTED_CHUNK_SHAPE
+        assert seismic.metadata.stats_v1 is None
+
+    def test_depth_domain_rejected(self) -> None:
+        """Depth domain is rejected."""
+        with pytest.raises(ValueError, match="only supports the time domain"):
+            Seismic3DSingleNodeContinuousReceiverGathersTemplate(data_domain="depth")

tests/unit/v1/templates/test_template_registry.py

@@ -38,6 +38,7 @@
     "StreamerShotGathers3D",
     "StreamerFieldRecords3D",
     "ObnReceiverGathers3D",
+    "SingleNodeContRecvrGathers",
     "ShotReceiverLineGathers3D",
 ]
 
@@ -245,7 +246,7 @@ def test_list_all_templates(self) -> None:
         registry.register(template2)
 
         templates = registry.list_all_templates()
-        assert len(templates) == 22 + 2  # 22 default + 2 custom
+        assert len(templates) == 23 + 2  # 23 default + 2 custom
         assert "Template_One" in templates
         assert "Template_Two" in templates
 
@@ -255,7 +256,7 @@ def test_clear_templates(self) -> None:
 
         # Default templates are always installed
         templates = list_templates()
-        assert len(templates) == 22
+        assert len(templates) == 23
 
         # Add some templates
         template1 = MockDatasetTemplate("Template1")
@@ -264,7 +265,7 @@ def test_clear_templates(self) -> None:
         registry.register(template1)
         registry.register(template2)
 
-        assert len(registry.list_all_templates()) == 22 + 2  # 22 default + 2 custom
+        assert len(registry.list_all_templates()) == 23 + 2  # 23 default + 2 custom
 
         # Clear all
         registry.clear()
@@ -397,7 +398,7 @@ def test_list_templates_global(self) -> None:
         register_template(template2)
 
         templates = list_templates()
-        assert len(templates) == 24  # 22 default + 2 custom
+        assert len(templates) == 25  # 23 default + 2 custom
         assert "template1" in templates
         assert "template2" in templates
 
@@ -440,7 +441,7 @@ def register_template_worker(template_id: int) -> None:
         assert len(errors) == 0
         assert len(results) == 10
         # Including default templates
-        assert len(registry.list_all_templates()) == 32  # 22 default + 10 registered
+        assert len(registry.list_all_templates()) == 33  # 23 default + 10 registered
 
         # Check all templates are registered
         for i in range(10):