create_empty_like

Author: dmitriyrepin

pyproject.toml

@@ -55,6 +55,7 @@ dev = [
     "pre-commit-hooks>=6.0.0",
     "pytest>=8.4.2",
     "pytest-dependency>=0.6.0",
+    "pytest-order>=1.3.0",
     "typeguard>=4.4.4",
     "xdoctest[colors]>=1.3.0",
     "Pygments>=2.19.2"

src/mdio/api/__init__.py

@@ -1,5 +1,6 @@
 """Public API."""
 
 from mdio.api.create import create_empty
+from mdio.api.create import create_empty_like
 
-__all__ = ["create_empty"]
+__all__ = ["create_empty", "create_empty_like"]

src/mdio/api/create.py

@@ -2,9 +2,12 @@
 
 from __future__ import annotations
 
+from datetime import UTC
+from datetime import datetime
 from typing import TYPE_CHECKING
 
 from mdio.api.io import _normalize_path
+from mdio.api.io import open_mdio
 from mdio.api.io import to_mdio
 from mdio.builder.template_registry import TemplateRegistry
 from mdio.builder.xarray_builder import to_xarray_dataset
@@ -84,3 +87,81 @@ def create_empty(  # noqa PLR0913
     meta_ds = dataset[drop_vars_delayed + ["trace_mask"]]
     to_mdio(meta_ds, output_path=output_path, mode="r+", compute=True)
 
+
+def create_empty_like(  # noqa PLR0913
+    input_path: UPath | Path | str,
+    output_path: UPath | Path | str,
+    keep_coordinates: bool = False,
+    overwrite: bool = False,
+) -> xr_Dataset:
+    """A function that creates an empty MDIO v1 file with the same structure as an existing one.
+
+    Args:
+        input_path: The path of the input MDIO file.
+        output_path: The path of the output MDIO file.
+                     If None, the output will not be written to disk.
+        keep_coordinates: Whether to keep the coordinates in the output file.
+        overwrite: Whether to overwrite the output file if it exists.
+
+    Returns:
+        The output MDIO dataset.
+
+    Raises:
+        FileExistsError: If the output location already exists and overwrite is False.
+    """
+    input_path = _normalize_path(input_path)
+    output_path = _normalize_path(output_path) if output_path is not None else None
+
+    if not overwrite and output_path is not None and output_path.exists():
+        err = f"Output location '{output_path.as_posix()}' exists. Set `overwrite=True` if intended."
+        raise FileExistsError(err)
+
+    ds = open_mdio(input_path)
+
+    # Create a copy with the same structure but no data or,
+    # optionally, coordinates
+    ds_output = ds.copy(data=None).reset_coords(drop=not keep_coordinates)
+
+    # Dataset
+    # Keep the name (which is the same as the used template name) and the original API version
+    # ds_output.attrs["name"]
+    # ds_output.attrs["apiVersion"]
+    ds_output.attrs["createdOn"] = datetime.now(UTC)
+
+    # Coordinates
+    if not keep_coordinates:
+        for coord_name in ds_output.coords:
+            ds_output[coord_name].attrs["unitsV1"] = None
+
+    # MDIO attributes
+    attr = ds_output.attrs["attributes"]
+    if attr is not None:
+        attr.pop("gridOverrides", None)  # Empty dataset should not have gridOverrides
+        # Keep the original values for the following attributes
+        # attr["defaultVariableName"]
+        # attr["surveyType"]
+        # attr["gatherType"]
+
+    # "All traces should be marked as dead in empty dataset"
+    if "trace_mask" in ds_output.variables:
+        ds_output["trace_mask"][:] = False
+
+    # Data variable
+    var_name = attr["defaultVariableName"]
+    var = ds_output[var_name]
+    var.attrs["statsV1"] = None
+    if not keep_coordinates:
+        var.attrs["unitsV1"] = None
+
+    # SEG-Y file header
+    if "segy_file_header" in ds_output.variables:
+        segy_file_header = ds_output["segy_file_header"]
+        if segy_file_header is not None:
+            segy_file_header.attrs["textHeader"] = None
+            segy_file_header.attrs["binaryHeader"] = None
+            segy_file_header.attrs["rawBinaryHeader"] = None
+
+    if output_path is not None:
+        to_mdio(ds_output, output_path=output_path, mode="w", compute=True)
+
+    return ds_output

src/mdio/builder/dataset_builder.py

@@ -6,7 +6,6 @@
 from enum import auto
 from typing import Any
 
-from mdio import __version__
 from mdio.builder.formatting_html import dataset_builder_repr_html
 from mdio.builder.schemas.compressors import ZFP
 from mdio.builder.schemas.compressors import Blosc
@@ -59,6 +58,8 @@ class MDIODatasetBuilder:
     """
 
     def __init__(self, name: str, attributes: dict[str, Any] | None = None):
+        from mdio import __version__  # noqa: PLC0415 -  fixed circular import in mdio package and dataset_builder.py
+
         self._metadata = DatasetMetadata(
             name=name,
             api_version=__version__,

src/mdio/converters/__init__.py

@@ -1,6 +1,30 @@
 """MDIO Data conversion API."""
 
-from mdio.converters.mdio import mdio_to_segy
-from mdio.converters.segy import segy_to_mdio
+from typing import TYPE_CHECKING
+from typing import Any
+
+if TYPE_CHECKING:
+    from mdio.converters.mdio import mdio_to_segy
+    from mdio.converters.segy import segy_to_mdio
 
 __all__ = ["mdio_to_segy", "segy_to_mdio"]
+
+
+def __getattr__(name: str) -> Any:  # noqa: ANN401 - required for dynamic attribute access
+    """Lazy import for converters to avoid circular imports."""
+    if name == "mdio_to_segy":
+        from mdio.converters.mdio import (  # noqa: PLC0415 - intentionally inside the function to avoid circular imports
+            mdio_to_segy,
+        )
+
+        return mdio_to_segy
+
+    if name == "segy_to_mdio":
+        from mdio.converters.segy import (  # noqa: PLC0415 - intentionally inside the function to avoid circular imports
+            segy_to_mdio,
+        )
+
+        return segy_to_mdio
+
+    err = f"module {__name__!r} has no attribute {name!r}"
+    raise AttributeError(err)

tests/conftest.py

@@ -47,6 +47,18 @@ def zarr_tmp(tmp_path_factory: pytest.TempPathFactory) -> Path:
     return tmp_path_factory.mktemp(r"mdio")
 
 
+@pytest.fixture(scope="session")
+def teapot_mdio_tmp(tmp_path_factory: pytest.TempPathFactory) -> Path:
+    """Make a temp file for the output MDIO."""
+    return tmp_path_factory.mktemp(r"teapot.mdio")
+
+
+@pytest.fixture(scope="module")
+def mdio_4d_tmp(tmp_path_factory: pytest.TempPathFactory) -> Path:
+    """Make a temp file for the output MDIO."""
+    return tmp_path_factory.mktemp(r"tmp_4d.mdio")
+
+
 @pytest.fixture(scope="module")
 def zarr_tmp2(tmp_path_factory: pytest.TempPathFactory) -> Path:  # pragma: no cover - used by disabled test
     """Make a temp file for the output MDIO."""
@@ -63,4 +75,28 @@ def segy_export_tmp(tmp_path_factory: pytest.TempPathFactory) -> Path:
 @pytest.fixture(scope="class")
 def empty_mdio_dir(tmp_path_factory: pytest.TempPathFactory) -> Path:
     """Make a temp file for empty MDIO testing."""
-    return tmp_path_factory.mktemp(r"empty_mdio")
+    return tmp_path_factory.mktemp(r"empty_mdio_dir")
+
+
+#
+# Uncomment the function below for local debugging
+#
+# @pytest.fixture(scope="session")
+# def tmp_path_factory() -> pytest.TempPathFactory:
+#     """Custom tmp_path_factory implementation for local debugging."""
+#     from pathlib import Path  # noqa: PLC0415
+
+#     class DebugTempPathFactory:
+#         def __init__(self) -> None:
+#             self._retention_policy = "all"
+
+#         def mktemp(self, basename: str, numbered: bool = True) -> Path:
+#             _ = numbered
+#             path = self.getbasetemp() / basename
+#             path.mkdir(parents=True, exist_ok=True)
+#             return path
+
+#         def getbasetemp(self) -> Path:
+#             return Path("tmp")
+
+#     return DebugTempPathFactory()