Implement new header analysis utilities and enhance ingestion pipeline

- Introduced `header_analysis.py` for analyzing SEG-Y headers to determine geometry and create indices for various acquisition types.
- Added functions to analyze streamer and gun geometries, improving the handling of multi-gun acquisitions.
- Refactored the ingestion pipeline to utilize the new header analysis utilities, enhancing the schema resolution and index strategy phases.
- Updated the `run_segy_ingestion` function to apply resolved dimensions and chunk shapes more effectively, ensuring compatibility with grid overrides.

Author: BrianMichell

src/mdio/builder/templates/base.py

@@ -17,7 +17,6 @@
 from mdio.builder.schemas.dtype import StructuredType
 from mdio.builder.schemas.v1.units import AllUnitModel
 from mdio.builder.schemas.v1.variable import VariableMetadata
-from mdio.builder.schemas.v1.variable import VariableMetadata
 from mdio.core.utils_write import MAX_COORDINATES_BYTES
 from mdio.core.utils_write import MAX_SIZE_LIVE_MASK
 from mdio.core.utils_write import get_constrained_chunksize
@@ -144,6 +143,30 @@ def add_units(self, units: dict[str, AllUnitModel]) -> None:
                 raise ValueError(msg)
         self._units |= units
 
+    def apply_resolved_dimensions(
+        self,
+        dim_names: tuple[str, ...],
+        chunk_shape: tuple[int, ...],
+    ) -> None:
+        """Update the template's dimension layout from a resolved schema.
+
+        Supported entry point for the ingestion pipeline to push back dimension names
+        and chunk shape after the SchemaResolver has applied grid overrides
+        (e.g. NonBinned, HasDuplicates), instead of mutating private attributes.
+
+        Args:
+            dim_names: Final ordered dimension names.
+            chunk_shape: Chunk shape matching ``dim_names`` length.
+
+        Raises:
+            ValueError: If ``len(chunk_shape) != len(dim_names)``.
+        """
+        if len(chunk_shape) != len(dim_names):
+            msg = f"chunk_shape length {len(chunk_shape)} does not match dim_names length {len(dim_names)}"
+            raise ValueError(msg)
+        self._dim_names = tuple(dim_names)
+        self._var_chunk_shape = tuple(chunk_shape)
+
     @property
     def name(self) -> str:
         """Returns the name of the template."""
@@ -191,25 +214,22 @@ def coordinate_names(self) -> tuple[str, ...]:
 
     @property
     def full_chunk_shape(self) -> tuple[int, ...]:
-        """Returns the chunk shape for the variables."""
-        # If dimension sizes are not set yet, return the stored shape as-is
+        """Returns the chunk shape for the variables, expanding -1 to the full dim size."""
         if len(self._dim_sizes) != len(self._dim_names):
             return self._var_chunk_shape
 
-        # Expand -1 values to full dimension sizes
         return tuple(
             dim_size if chunk_size == -1 else chunk_size
             for chunk_size, dim_size in zip(self._var_chunk_shape, self._dim_sizes, strict=False)
         )
 
     @full_chunk_shape.setter
     def full_chunk_shape(self, shape: tuple[int, ...]) -> None:
-        """Sets the chunk shape for the variables."""
+        """Sets the chunk shape; values must be positive ints or -1 (means full dim size)."""
         if len(shape) != len(self._dim_names):
             msg = f"Chunk shape {shape} has {len(shape)} dimensions, expected {len(self._dim_names)}"
             raise ValueError(msg)
 
-        # Validate that all values are positive integers or -1
         for chunk_size in shape:
             if chunk_size != -1 and chunk_size <= 0:
                 msg = f"Chunk size must be positive integer or -1, got {chunk_size}"

src/mdio/converters/segy.py

@@ -2,80 +2,24 @@
 
 from __future__ import annotations
 
-import base64
-import logging
+import warnings
 from typing import TYPE_CHECKING
 
-import numpy as np
-import zarr
-from segy.config import SegyFileSettings
-from segy.config import SegyHeaderOverrides
-from segy.standards.codes import MeasurementSystem as SegyMeasurementSystem
-from segy.standards.fields import binary as binary_header_fields
-
-from mdio.api.io import _normalize_path
-from mdio.api.io import to_mdio
-from mdio.builder.schemas.chunk_grid import RegularChunkGrid
-from mdio.builder.schemas.chunk_grid import RegularChunkShape
-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.v1.units import AngleUnitEnum
-from mdio.builder.schemas.v1.units import AngleUnitModel
-from mdio.builder.schemas.v1.units import LengthUnitEnum
-from mdio.builder.schemas.v1.units import LengthUnitModel
-from mdio.builder.schemas.v1.variable import VariableMetadata
-from mdio.builder.xarray_builder import to_xarray_dataset
-from mdio.constants import ZarrFormat
-from mdio.converters.exceptions import GridTraceCountError
-from mdio.converters.exceptions import GridTraceSparsityError
-from mdio.converters.type_converter import to_structured_type
-from mdio.core.config import MDIOSettings
-from mdio.core.grid import Grid
-from mdio.core.utils_write import MAX_COORDINATES_BYTES
-from mdio.core.utils_write import MAX_SIZE_LIVE_MASK
-from mdio.core.utils_write import get_constrained_chunksize
-from mdio.segy import blocked_io
-from mdio.segy.file import get_segy_file_info
+from mdio.ingestion.pipeline import run_segy_ingestion
 from mdio.segy.geometry import GridOverrides
-from mdio.segy.scalar import SCALE_COORDINATE_KEYS
-from mdio.segy.scalar import _apply_coordinate_scalar
-from mdio.segy.utilities import get_grid_plan
 
 if TYPE_CHECKING:
     from pathlib import Path
     from typing import Any
 
-    from segy.arrays import HeaderArray as SegyHeaderArray
+    from segy.config import SegyHeaderOverrides
     from segy.schema import SegySpec
     from upath import UPath
-    from xarray import Dataset as xr_Dataset
 
-    from mdio.builder.schemas import Dataset
     from mdio.builder.templates.base import AbstractDatasetTemplate
-    from mdio.core.dimension import Dimension
-    from mdio.segy.file import SegyFileArguments
-    from mdio.segy.file import SegyFileInfo
-
-logger = logging.getLogger(__name__)
-
-
-MEASUREMENT_SYSTEM_KEY = binary_header_fields.Rev0.MEASUREMENT_SYSTEM_CODE.model.name
-ANGLE_UNIT_KEYS = ["angle", "azimuth"]
-SPATIAL_UNIT_KEYS = [
-    "cdp_x",
-    "cdp_y",
-    "source_coord_x",
-    "source_coord_y",
-    "group_coord_x",
-    "group_coord_y",
-    "offset",
-]
 
 
-
-
-def segy_to_mdio(  # noqa PLR0913
+def segy_to_mdio(  # noqa: PLR0913
     segy_spec: SegySpec,
     mdio_template: AbstractDatasetTemplate,
     input_path: UPath | Path | str,
@@ -84,30 +28,33 @@ def segy_to_mdio(  # noqa PLR0913
     grid_overrides: GridOverrides | dict[str, Any] | None = None,
     segy_header_overrides: SegyHeaderOverrides | None = None,
 ) -> None:
-    """A function that converts a SEG-Y file to an MDIO v1 file.
+    """Convert a SEG-Y file to an MDIO v1 file.
 
-    Ingest a SEG-Y file according to the segy_spec. This could be a spec from registry or custom.
+    This is the v1.x public entry point. Internally it dispatches to the v1.2
+    refactored ingestion pipeline (:func:`mdio.ingestion.pipeline.run_segy_ingestion`).
 
     Args:
         segy_spec: The SEG-Y specification to use for the conversion.
         mdio_template: The MDIO template to use for the conversion.
         input_path: The universal path of the input SEG-Y file.
         output_path: The universal path for the output MDIO v1 file.
         overwrite: Whether to overwrite the output file if it already exists. Defaults to False.
-        grid_overrides: Grid override configuration. Can be a GridOverrides instance for type
-            safety, or a dict for backward compatibility. See GridOverrides class for available
-            options.
+        grid_overrides: Grid override configuration. Prefer a typed
+            :class:`mdio.GridOverrides` instance. A ``dict`` is still accepted for backward
+            compatibility but will emit a :class:`DeprecationWarning` and be coerced.
         segy_header_overrides: Option to override specific SEG-Y headers during ingestion.
 
     Raises:
-        FileExistsError: If the output location already exists and overwrite is False.
+        FileExistsError: If the output location already exists and ``overwrite`` is False.
     """
-    # Convert dict to GridOverrides if needed for type safety
     if isinstance(grid_overrides, dict):
-        grid_overrides = GridOverrides.model_validate(grid_overrides) if grid_overrides else None
-
-    # Use ingestion pipeline
-    from mdio.ingestion.pipeline import run_segy_ingestion
+        warnings.warn(
+            "Passing 'grid_overrides' as a dict is deprecated and will be removed in a future "
+            "release. Use 'mdio.GridOverrides(...)' (snake_case fields or legacy aliases both work).",
+            DeprecationWarning,
+            stacklevel=2,
+        )
+        grid_overrides = GridOverrides.from_legacy_dict(grid_overrides) if grid_overrides else None
 
     return run_segy_ingestion(
         segy_spec=segy_spec,

src/mdio/ingestion/coordinate_utils.py

@@ -1,8 +1,4 @@
-"""Coordinate handling utilities for MDIO ingestion.
-
-This module contains functions for extracting, populating, and managing
-coordinates during SEG-Y to MDIO conversion.
-"""
+"""Coordinate handling utilities for MDIO ingestion."""
 
 from __future__ import annotations
 
@@ -32,7 +28,6 @@
 logger = logging.getLogger(__name__)
 
 
-# Constants for unit handling
 MEASUREMENT_SYSTEM_KEY = binary_header_fields.Rev0.MEASUREMENT_SYSTEM_CODE.model.name
 ANGLE_UNIT_KEYS = ["angle", "azimuth"]
 SPATIAL_UNIT_KEYS = [
@@ -53,7 +48,6 @@ def get_coordinates(
 ) -> tuple[list[Dimension], dict[str, SegyHeaderArray]]:
     """Get the data dim and non-dim coordinates from the SEG-Y headers and MDIO template.
 
-    Select a subset of the segy_dimensions that corresponds to the MDIO dimensions.
     Uses the grid's actual dimensions (which may have been transformed by grid overrides).
     The last dimension is always the vertical domain dimension.
 
@@ -67,20 +61,16 @@ def get_coordinates(
                     the SEG-Y headers.
 
     Returns:
-        A tuple containing:
-            - A list of dimension coordinates (1-D arrays).
-            - A dict of non-dimension coordinates (str: N-D arrays).
+        A tuple of (dimension coordinates as 1-D arrays, non-dimension coordinates as N-D arrays).
     """
-    # Use grid's actual dimensions (may differ from template after grid overrides)
     dimensions_coords = list(grid.dims)
 
-    # Extract non-dimension coordinates from headers
     non_dim_coords: dict[str, SegyHeaderArray] = {}
     for coord_name in mdio_template.coordinate_names:
         if coord_name not in segy_headers.dtype.names:
             err = f"Coordinate '{coord_name}' not found in SEG-Y headers."
             raise ValueError(err)
-        # Copy the data to allow segy_headers to be garbage collected
+        # Copy so segy_headers can be garbage collected.
         non_dim_coords[coord_name] = np.array(segy_headers[coord_name])
 
     return dimensions_coords, non_dim_coords
@@ -103,52 +93,37 @@ def populate_non_dim_coordinates(
     drop_vars_delayed: list[str],
     spatial_coordinate_scalar: int,
 ) -> tuple[xr_Dataset, list[str]]:
-    """Populate the xarray dataset with coordinate variables.
+    """Populate the xarray dataset with non-dimension coordinate variables.
 
-    Memory optimization: Processes coordinates one at a time and explicitly
-    releases intermediate arrays to reduce peak memory usage.
+    Coordinates are processed one at a time and intermediate arrays are released
+    explicitly to keep peak memory low for large surveys.
     """
-    non_data_domain_dims = grid.dim_names[:-1]  # minus the data domain dimension
+    non_data_domain_dims = grid.dim_names[:-1]
 
-    # Process coordinates one at a time to minimize peak memory
-    coord_names = list(coordinates.keys())
-    for coord_name in coord_names:
-        coord_values = coordinates.pop(coord_name)  # Remove from dict to free memory
+    for coord_name in list(coordinates.keys()):
+        coord_values = coordinates.pop(coord_name)
         da_coord = dataset[coord_name]
 
-        # Get coordinate shape from dataset (uses dask shape, no memory allocation)
-        coord_shape = da_coord.shape
-
-        # Create output array with fill value
         fill_value = da_coord.encoding.get("_FillValue") or da_coord.encoding.get("fill_value")
         if fill_value is None:
             fill_value = np.nan
-        tmp_coord_values = np.full(coord_shape, fill_value, dtype=da_coord.dtype)
+        tmp_coord_values = np.full(da_coord.shape, fill_value, dtype=da_coord.dtype)
 
-        # Compute slices for this coordinate's dimensions
         coord_axes = tuple(non_data_domain_dims.index(coord_dim) for coord_dim in da_coord.dims)
         coord_slices = tuple(slice(None) if idx in coord_axes else 0 for idx in range(len(non_data_domain_dims)))
 
-        # Read only the required slice from grid map
         coord_trace_indices = np.asarray(grid.map[coord_slices])
-
-        # Find valid (non-null) indices
         not_null = coord_trace_indices != grid.map.fill_value
 
-        # Populate values efficiently
         if not_null.any():
-            valid_indices = coord_trace_indices[not_null]
-            tmp_coord_values[not_null] = coord_values[valid_indices]
+            tmp_coord_values[not_null] = coord_values[coord_trace_indices[not_null]]
 
-        # Apply scalar if needed
         if coord_name in SCALE_COORDINATE_KEYS:
             tmp_coord_values = _apply_coordinate_scalar(tmp_coord_values, spatial_coordinate_scalar)
 
-        # Assign to dataset
         dataset[coord_name][:] = tmp_coord_values
         drop_vars_delayed.append(coord_name)
 
-        # Explicitly release intermediate arrays
         del tmp_coord_values, coord_trace_indices, not_null, coord_values
 
     return dataset, drop_vars_delayed
@@ -178,22 +153,17 @@ def get_spatial_coordinate_unit(segy_file_info: SegyFileInfo) -> LengthUnitModel
 def apply_runtime_units(template: AbstractDatasetTemplate, segy_file_info: SegyFileInfo) -> AbstractDatasetTemplate:
     """Update the template with dynamic and some pre-defined units."""
     unit = get_spatial_coordinate_unit(segy_file_info)
-
-    # Add units for pre-defined: angle and azimuth etc.
     new_units = {key: AngleUnitModel(angle=AngleUnitEnum.DEGREES) for key in ANGLE_UNIT_KEYS}
 
-    # If a spatial unit is not provided, we return as is
     if unit is None:
         template.add_units(new_units)
         return template
 
-    # Dynamically add units based on the spatial coordinate unit
     for key in SPATIAL_UNIT_KEYS:
         current_value = template.get_unit_by_key(key)
         if current_value is not None:
             logger.warning("Unit for %s already in template. Will keep the original unit: %s", key, current_value)
             continue
-
         new_units[key] = unit
 
     template.add_units(new_units)