diff --git a/src/access_moppy/base.py b/src/access_moppy/base.py index 8e77fd43..0b1e72b8 100644 --- a/src/access_moppy/base.py +++ b/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): """ diff --git a/tests/unit/test_base.py b/tests/unit/test_base.py index 9ccaacb6..7fa5bd2d 100644 --- a/tests/unit/test_base.py +++ b/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, diff --git a/tests/unit/test_qc_gates.py b/tests/unit/test_qc_gates.py index 7df3f28b..7ce26131 100644 --- a/tests/unit/test_qc_gates.py +++ b/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"]