fix_cmip7_repack_compression

src/access_moppy/base.py

@@ -2707,13 +2707,83 @@ def _record_range_gate(self, result: object) -> None:
             record["units"] = result.units
         self._record_gate("range", str(record.pop("result")), **record)
 
+    def _cmip7repack_chunk_target(self, path: Path) -> int:
+        """Chunk-size target (bytes) for ``cmip7repack -d`` on this file.
+
+        ``cmip7repack`` only lengthens a chunk along the leading (time)
+        dimension, and only applies shuffle/zlib/Fletcher32 to variables it
+        actually rechunks.  A 3-D field whose single-timestep slice exceeds
+        half the 4 MiB default therefore keeps netCDF4's default
+        one-timestep chunk and is never compressed at all -- p19 atmosphere
+        misses by 0.2%.  Ask for two timesteps' worth so the rechunk, and
+        with it the compression, always happens.
+        """
+        default = 4194304  # cmip7repack's own default, and its minimum
+
+        try:
+            with nc.Dataset(path, "r") as ds:
+                # cmip7repack picks the data variable the same way.
+                var_id = (
+                    ds.getncattr("variable_id")
+                    if "variable_id" in ds.ncattrs()
+                    else self.output_name
+                )
+                var = ds.variables.get(var_id)
+                if var is None:
+                    return default
+                slice_bytes = var.dtype.itemsize
+                for size in var.shape[1:]:
+                    slice_bytes *= int(size)
+        except OSError as exc:
+            logger.warning(
+                "Could not size the cmip7repack chunk target for %s (%s); "
+                "falling back to the %d byte default",
+                path,
+                exc,
+                default,
+            )
+            return default
+
+        return max(default, 2 * slice_bytes)
+
+    def _verify_repack_compression(self, path: Path) -> Optional[str]:
+        """Return a reason string if the repacked data variable is uncompressed.
+
+        ``cmip7repack`` exits 0 whether or not it rechunked anything, so the
+        exit status alone cannot tell a real repack from a silent no-op.
+        Read the filters back off the file instead.
+        """
+        try:
+            with nc.Dataset(path, "r") as ds:
+                var_id = (
+                    ds.getncattr("variable_id")
+                    if "variable_id" in ds.ncattrs()
+                    else self.output_name
+                )
+                var = ds.variables.get(var_id)
+                if var is None:
+                    return f"variable {var_id!r} not found in the repacked file"
+                filters = var.filters() or {}
+                if not filters.get("zlib"):
+                    return (
+                        f"cmip7repack left {var_id} uncompressed "
+                        f"(chunking {var.chunking()}, filters {filters})"
+                    )
+        except OSError as exc:
+            return f"could not reopen the repacked file: {exc}"
+
+        return None
+
     def _repack_cmip7_output(self, path: Path):
         """Repack a CMIP7 netCDF file in place after writing it."""
         if getattr(self.vocab, "mip_era", None) != "CMIP7":
             return
 
-        cmd = ["cmip7repack", "-o", str(path)]
-        logger.info("Repacking CMIP7 output with cmip7repack: %s", path)
+        chunk_target = self._cmip7repack_chunk_target(path)
+        cmd = ["cmip7repack", "-o", "-d", str(chunk_target), str(path)]
+        logger.info(
+            "Repacking CMIP7 output with cmip7repack (-d %d): %s", chunk_target, path
+        )
 
         try:
             subprocess.run(  # noqa: S603  # nosec B603
@@ -2735,10 +2805,15 @@ def _repack_cmip7_output(self, path: Path):
             )
             raise
 
-        # A repack failure aborts the variable, so reaching here means the file
-        # was repacked. Record it: without this, a downstream reader can only
-        # infer the repack from the task not having failed.
-        self._record_gate("repack", "pass", tool="cmip7repack")
+        # cmip7repack exits 0 even when it rechunks nothing, so the exit
+        # status alone would record a silent no-op as a pass. Read the
+        # filters back off the file to see what actually happened.
+        reason = self._verify_repack_compression(path)
+        if reason:
+            logger.warning("cmip7repack did not compress %s: %s", path, reason)
+            self._record_gate("repack", "warn", tool="cmip7repack", message=reason)
+        else:
+            self._record_gate("repack", "pass", tool="cmip7repack")
 
     def _prepare_string_coordinates(self):
         """

tests/unit/test_base.py

@@ -2209,7 +2209,7 @@ def test_write_repacks_cmip7_output(self, cmoriser_with_dataset, temp_dir):
         intermediate = {}
 
         def inspect_intermediate(*args, **kwargs):
-            path = Path(args[0][2])
+            path = Path(args[0][-1])
             with nc.Dataset(path) as dataset:
                 intermediate["data_model"] = dataset.data_model
                 intermediate["filters"] = dataset.variables["tas"].filters()
@@ -2231,8 +2231,11 @@ def inspect_intermediate(*args, **kwargs):
 
         output_files = list(Path(temp_dir).glob("*.nc"))
         assert len(output_files) == 1
+        # This tas is far under 4 MiB per timestep, so the chunk target stays
+        # at cmip7repack's default; a p19 field asks for more (see
+        # tests/unit/test_qc_gates.py).
         mock_run.assert_called_once_with(
-            ["cmip7repack", "-o", str(output_files[0])],
+            ["cmip7repack", "-o", "-d", "4194304", str(output_files[0])],
             check=True,
             capture_output=True,
             text=True,

tests/unit/test_qc_gates.py

@@ -188,9 +188,10 @@ def test_write_records_the_range_gate_the_validator_returned(tmp_path):
 @pytest.mark.unit
 def test_repack_gate_is_recorded_when_cmip7repack_succeeds(tmp_path):
     cmoriser = _cmoriser(tmp_path)
+    path = _repacked_file(tmp_path, (3, 19, 144, 192), zlib=True)
 
     with patch("access_moppy.base.subprocess.run") as run_mock:
-        cmoriser._repack_cmip7_output(tmp_path / "tas.nc")
+        cmoriser._repack_cmip7_output(path)
 
     run_mock.assert_called_once()
     assert cmoriser.qc_gates["repack"] == {"result": "pass", "tool": "cmip7repack"}
@@ -434,3 +435,113 @@ def test_enforce_compliance_still_works_without_a_callback(
 
     assert report_path.exists()
     assert output_file.exists()
+
+
+# ── Sizing the repack chunk target ──────────────────────────────────────────
+
+
+def _repacked_file(
+    tmp_path, shape, *, zlib: bool, name: str = "tas.nc", declared_id: str = "tas"
+) -> Path:
+    """A netCDF4 file holding one data variable, as cmip7repack would see it.
+
+    ``declared_id`` is the ``variable_id`` global attribute; point it at a name
+    the file does not hold to exercise the "no such variable" paths.
+    """
+    import netCDF4 as nc
+
+    path = tmp_path / name
+    dims = ["time", "plev", "lat", "lon"][-len(shape) :]
+    with nc.Dataset(path, "w") as ds:
+        ds.setncattr("variable_id", declared_id)
+        for dim, size in zip(dims, shape):
+            ds.createDimension(dim, size)
+        ds.createVariable("tas", "f4", dims, zlib=zlib, shuffle=zlib)
+    return path
+
+
+@pytest.mark.unit
+def test_chunk_target_covers_two_timesteps_of_a_p19_field(tmp_path):
+    """A p19 slice is 2101248 B: the 4 MiB default fits one timestep, not two,
+    so cmip7repack would leave the variable unchunked and uncompressed."""
+    cmoriser = _cmoriser(tmp_path)
+    path = _repacked_file(tmp_path, (3, 19, 144, 192), zlib=False)
+
+    assert cmoriser._cmip7repack_chunk_target(path) == 2 * 19 * 144 * 192 * 4
+
+
+@pytest.mark.unit
+def test_chunk_target_stays_at_the_default_for_a_surface_field(tmp_path):
+    """Two timesteps of a surface field are far under 4 MiB; asking for less
+    than the default is rejected by cmip7repack, so we must not."""
+    cmoriser = _cmoriser(tmp_path)
+    path = _repacked_file(tmp_path, (3, 145, 192), zlib=False)
+
+    assert cmoriser._cmip7repack_chunk_target(path) == 4194304
+
+
+@pytest.mark.unit
+def test_chunk_target_is_passed_to_cmip7repack(tmp_path):
+    cmoriser = _cmoriser(tmp_path)
+    path = _repacked_file(tmp_path, (3, 19, 144, 192), zlib=True)
+
+    with patch("access_moppy.base.subprocess.run") as run_mock:
+        cmoriser._repack_cmip7_output(path)
+
+    cmd = run_mock.call_args.args[0]
+    assert cmd[cmd.index("-d") + 1] == str(2 * 19 * 144 * 192 * 4)
+
+
+# ── Verifying the repack actually compressed ────────────────────────────────
+
+
+@pytest.mark.unit
+def test_repack_gate_warns_when_the_data_variable_is_left_uncompressed(tmp_path):
+    """cmip7repack exits 0 when it rechunks nothing, so exit status alone would
+    record a silent no-op as a pass."""
+    cmoriser = _cmoriser(tmp_path)
+    path = _repacked_file(tmp_path, (3, 19, 144, 192), zlib=False)
+
+    with patch("access_moppy.base.subprocess.run"):
+        cmoriser._repack_cmip7_output(path)
+
+    gate = cmoriser.qc_gates["repack"]
+    assert gate["result"] == "warn"
+    assert "uncompressed" in gate["message"]
+
+
+@pytest.mark.unit
+def test_chunk_target_falls_back_when_the_data_variable_is_absent(tmp_path):
+    """variable_id can name a variable the file does not hold — see
+    ISSUE-cmip7-out-name-cmip6-fallback.md, where exactly that happened."""
+    cmoriser = _cmoriser(tmp_path)
+    path = _repacked_file(tmp_path, (3, 19, 144, 192), zlib=False, declared_id="ta")
+
+    assert cmoriser._cmip7repack_chunk_target(path) == 4194304
+
+
+@pytest.mark.unit
+def test_repack_gate_warns_when_the_data_variable_is_missing(tmp_path):
+    cmoriser = _cmoriser(tmp_path)
+    path = _repacked_file(tmp_path, (3, 19, 144, 192), zlib=True, declared_id="ta")
+
+    with patch("access_moppy.base.subprocess.run"):
+        cmoriser._repack_cmip7_output(path)
+
+    gate = cmoriser.qc_gates["repack"]
+    assert gate["result"] == "warn"
+    assert "not found" in gate["message"]
+
+
+@pytest.mark.unit
+def test_repack_gate_warns_when_the_repacked_file_cannot_be_reopened(tmp_path):
+    """cmip7repack overwrites in place; if that left nothing readable behind,
+    the gate must say so rather than record a pass."""
+    cmoriser = _cmoriser(tmp_path)
+
+    with patch("access_moppy.base.subprocess.run"):
+        cmoriser._repack_cmip7_output(tmp_path / "gone.nc")
+
+    gate = cmoriser.qc_gates["repack"]
+    assert gate["result"] == "warn"
+    assert "could not reopen" in gate["message"]