Add repack_hdf5 utility and tests

Introduce repack_hdf5 in mth5.utils.h5_tools to copy HDF5 file-level attributes, groups, datasets, and their attributes into a new file (mirroring h5repack behavior). The function validates input existence, prevents accidental overwrite unless overwrite=True, and returns the output Path. Add tests to verify structure/data/attributes are copied, missing source raises FileNotFoundError, and overwrite behavior raises FileExistsError unless allowed. Uses h5py and pathlib.

Author: kujaku11

mth5/utils/h5_tools.py

@@ -0,0 +1,85 @@
+"""Utilities for HDF5 maintenance tasks."""
+
+from __future__ import annotations
+
+from pathlib import Path
+from typing import Union
+
+import h5py
+
+
+PathLike = Union[str, Path]
+
+
+def repack_hdf5(
+    input_file: PathLike,
+    output_file: PathLike,
+    *,
+    overwrite: bool = False,
+) -> Path:
+    """Repack an HDF5 file by copying all objects into a new file.
+
+    Repacking can reduce file size when metadata or datasets were repeatedly
+    modified and the source file contains unused internal space.
+
+    This mirrors the core idea of ``h5repack``: rewrite all objects into a new
+    HDF5 container so only live content remains. It does *not* expose the full
+    ``h5repack`` feature set (for example: filter/layout transformation,
+    chunk-size retuning, and advanced option flags).
+
+    Parameters
+    ----------
+    input_file : str | pathlib.Path
+        Source HDF5 file to repack.
+    output_file : str | pathlib.Path
+        Destination path for the repacked HDF5 file.
+    overwrite : bool, default=False
+        If ``True``, overwrite ``output_file`` when it exists.
+
+    Returns
+    -------
+    pathlib.Path
+        The output path.
+
+    Raises
+    ------
+    FileNotFoundError
+        If ``input_file`` does not exist.
+    FileExistsError
+        If ``output_file`` exists and ``overwrite`` is ``False``.
+
+    Examples
+    --------
+    >>> from mth5.utils.h5_tools import repack_hdf5
+    >>> repacked = repack_hdf5("survey_original.h5", "survey_repacked.h5", overwrite=True)
+    >>> print(repacked)
+    survey_repacked.h5
+    """
+
+    source_path = Path(input_file)
+    target_path = Path(output_file)
+
+    if not source_path.exists():
+        raise FileNotFoundError(f"Input file does not exist: {source_path}")
+
+    if target_path.exists() and not overwrite:
+        raise FileExistsError(
+            f"Output file already exists: {target_path}. "
+            "Set overwrite=True to replace it."
+        )
+
+    if target_path.exists() and overwrite:
+        target_path.unlink()
+
+    with h5py.File(source_path, "r") as source_h5, h5py.File(
+        target_path, "w"
+    ) as target_h5:
+        # Copy file-level attributes.
+        for key, value in source_h5.attrs.items():
+            target_h5.attrs[key] = value
+
+        # Copy root objects recursively (groups, datasets, attributes).
+        for name in source_h5.keys():
+            source_h5.copy(name, target_h5, name=name)
+
+    return target_path

tests/utils/test_h5_tools.py

@@ -0,0 +1,65 @@
+import h5py
+import numpy as np
+import pytest
+
+from mth5.utils.h5_tools import repack_hdf5
+
+
+def test_repack_hdf5_copies_structure_data_and_attributes(tmp_path):
+    """Repacking should copy groups, datasets, and attrs into a new file."""
+    source = tmp_path / "source.h5"
+    target = tmp_path / "repacked.h5"
+
+    with h5py.File(source, "w") as h5:
+        h5.attrs["file_attr"] = "root"
+
+        group = h5.create_group("level1")
+        group.attrs["group_attr"] = 7
+
+        data = np.arange(10, dtype=np.float64)
+        dataset = group.create_dataset("values", data=data)
+        dataset.attrs["unit"] = "counts"
+
+    returned_path = repack_hdf5(source, target)
+    assert returned_path == target
+    assert target.exists()
+
+    with h5py.File(target, "r") as h5:
+        assert h5.attrs["file_attr"] == "root"
+        assert "level1" in h5
+
+        group = h5["level1"]
+        assert group.attrs["group_attr"] == 7
+
+        dataset = group["values"]
+        np.testing.assert_array_equal(dataset[()], np.arange(10, dtype=np.float64))
+        assert dataset.attrs["unit"] == "counts"
+
+
+def test_repack_hdf5_raises_if_source_is_missing(tmp_path):
+    """Missing source file should raise FileNotFoundError."""
+    missing_source = tmp_path / "does_not_exist.h5"
+    target = tmp_path / "output.h5"
+
+    with pytest.raises(FileNotFoundError):
+        repack_hdf5(missing_source, target)
+
+
+def test_repack_hdf5_respects_overwrite_flag(tmp_path):
+    """Output file should not be overwritten unless explicitly requested."""
+    source = tmp_path / "source.h5"
+    target = tmp_path / "output.h5"
+
+    with h5py.File(source, "w") as h5:
+        h5.create_dataset("x", data=np.array([1, 2, 3], dtype=np.int64))
+
+    # First write creates target.
+    repack_hdf5(source, target)
+
+    # Second write should fail without overwrite.
+    with pytest.raises(FileExistsError):
+        repack_hdf5(source, target)
+
+    # Overwrite should succeed.
+    returned_path = repack_hdf5(source, target, overwrite=True)
+    assert returned_path == target