From 8fd1b60a110340695f92ba1ef2476e099d578fce Mon Sep 17 00:00:00 2001 From: rhaegar325 Date: Mon, 7 Sep 2026 22:10:15 +1000 Subject: [PATCH 1/3] add engine defination in xr.open_dataset --- src/access_moppy/base.py | 4 +- src/access_moppy/derivations/calc_utils.py | 2 +- src/access_moppy/executors/dask_config.py | 4 +- src/access_moppy/ocean_supergrid.py | 4 +- src/access_moppy/qc/cmip7.py | 2 +- src/access_moppy/qc/plots.py | 3 +- src/access_moppy/utilities.py | 13 +++- tests/unit/test_xarray_engine_pinned.py | 88 ++++++++++++++++++++++ 8 files changed, 111 insertions(+), 9 deletions(-) create mode 100644 tests/unit/test_xarray_engine_pinned.py diff --git a/src/access_moppy/base.py b/src/access_moppy/base.py index 07a31fd1..413b62fe 100644 --- a/src/access_moppy/base.py +++ b/src/access_moppy/base.py @@ -758,7 +758,9 @@ def load_dataset(self, required_vars: Optional[List[str]] = None): # is reused for both the frequency-validation time-independence check # and the _has_time check below, avoiding a duplicate open and the # file-handle leak that an unguarded open_dataset would cause. - with xr.open_dataset(self.input_paths[0], decode_cf=False) as _probe: + with xr.open_dataset( + self.input_paths[0], engine="netcdf4", decode_cf=False + ) as _probe: _probe_dims = set(_probe.dims) _probe_target_vars = ( [v for v in required_vars if v in _probe.data_vars] diff --git a/src/access_moppy/derivations/calc_utils.py b/src/access_moppy/derivations/calc_utils.py index 568783a7..162b437d 100644 --- a/src/access_moppy/derivations/calc_utils.py +++ b/src/access_moppy/derivations/calc_utils.py @@ -470,7 +470,7 @@ def load_ressource_data(ressource_file: str, var_name: str) -> xr.DataArray: """ resource_path = get_bundled_resource_path(ressource_file) with as_file(resource_path) as resolved: - ds = xr.open_dataset(str(resolved)) + ds = xr.open_dataset(str(resolved), engine="netcdf4") if var_name not in ds: raise ValueError( f"Variable '{var_name}' not found in resource file '{ressource_file}'. " diff --git a/src/access_moppy/executors/dask_config.py b/src/access_moppy/executors/dask_config.py index 63d8f0ca..a4e6e61d 100644 --- a/src/access_moppy/executors/dask_config.py +++ b/src/access_moppy/executors/dask_config.py @@ -280,7 +280,9 @@ def _estimate_worker_memory_gb(variable, input_files, model_id): mapping = load_model_mappings(variable, model_id) model_vars = (mapping.get(cmor_name, {}) or {}).get("model_variables", []) or [] - with xr.open_dataset(input_files[0], decode_cf=False) as ds0: + with xr.open_dataset( + input_files[0], engine="netcdf4", decode_cf=False + ) as ds0: steps_per_file = int(ds0.sizes.get("time", 1)) present = [v for v in model_vars if v in ds0.variables] per_file_mb = ( diff --git a/src/access_moppy/ocean_supergrid.py b/src/access_moppy/ocean_supergrid.py index 5b0f3563..f97f220f 100644 --- a/src/access_moppy/ocean_supergrid.py +++ b/src/access_moppy/ocean_supergrid.py @@ -90,7 +90,9 @@ def load_supergrid(self, supergrid_file: str): for the 10 km grid), so actual computation is deferred until the first call to :meth:`extract_grid` via :meth:`_compute_grid`. """ - self.supergrid = xr.open_dataset(supergrid_file, chunks={}) + self.supergrid = xr.open_dataset( + supergrid_file, engine="netcdf4", chunks={} + ) self._grid_computed = False def _compute_grid(self): diff --git a/src/access_moppy/qc/cmip7.py b/src/access_moppy/qc/cmip7.py index 88d4a841..64a4a01f 100644 --- a/src/access_moppy/qc/cmip7.py +++ b/src/access_moppy/qc/cmip7.py @@ -362,7 +362,7 @@ def _validate_cmip7_output(output_path: str | Path) -> ValidationResult: """ path = Path(output_path) - with xr.open_dataset(path, chunks="auto") as ds: + with xr.open_dataset(path, engine="netcdf4", chunks="auto") as ds: attrs = dict(ds.attrs) variable_id = attrs.get("variable_id") experiment_id = attrs.get("experiment_id") diff --git a/src/access_moppy/qc/plots.py b/src/access_moppy/qc/plots.py index d1bfb4a2..7e92071a 100644 --- a/src/access_moppy/qc/plots.py +++ b/src/access_moppy/qc/plots.py @@ -441,7 +441,7 @@ def generate_qc_plots( stem = path.stem try: - with xr.open_dataset(path, use_cftime=True) as ds: + with xr.open_dataset(path, engine="netcdf4", use_cftime=True) as ds: var_name = _find_primary_variable(ds) da = ds[var_name] units = str(da.attrs.get("units", "")) @@ -542,6 +542,7 @@ def generate_qc_plots_for_split_files( try: with xr.open_mfdataset( sorted(str(p) for p in paths), + engine="netcdf4", combine="by_coords", use_cftime=True, ) as ds: diff --git a/src/access_moppy/utilities.py b/src/access_moppy/utilities.py index 27bca754..1072e060 100644 --- a/src/access_moppy/utilities.py +++ b/src/access_moppy/utilities.py @@ -998,6 +998,7 @@ def _detect_frequency_from_concatenated_files( with ( xr.open_mfdataset( sampled_files, + engine="netcdf4", decode_cf=False, chunks={}, concat_dim=time_coord, @@ -1046,7 +1047,9 @@ def _detect_frequency_from_individual_files( # Detect frequency from each file for file_path in file_paths: try: - with xr.open_dataset(file_path, decode_cf=False, chunks={}) as ds: + with xr.open_dataset( + file_path, engine="netcdf4", decode_cf=False, chunks={} + ) as ds: freq = detect_time_frequency_lazy(ds, time_coord) if freq is not None: frequencies.append(freq) @@ -1148,7 +1151,9 @@ def _validate_monthly_files_individually( # Detect frequency from each file for file_path in file_paths: try: - with xr.open_dataset(file_path, decode_cf=False, chunks={}) as ds: + with xr.open_dataset( + file_path, engine="netcdf4", decode_cf=False, chunks={} + ) as ds: freq = detect_time_frequency_lazy(ds, time_coord) if freq is not None: frequencies.append(freq) @@ -1890,7 +1895,9 @@ def _validate_frequency_consistency_detailed( for file_path in file_paths: try: # Open file lazily - no data is loaded into memory here - with xr.open_dataset(file_path, decode_cf=False, chunks={}) as ds: + with xr.open_dataset( + file_path, engine="netcdf4", decode_cf=False, chunks={} + ) as ds: freq = detect_time_frequency_lazy(ds, time_coord) if freq is not None: frequencies.append(freq) diff --git a/tests/unit/test_xarray_engine_pinned.py b/tests/unit/test_xarray_engine_pinned.py new file mode 100644 index 00000000..472c7ae9 --- /dev/null +++ b/tests/unit/test_xarray_engine_pinned.py @@ -0,0 +1,88 @@ +"""Every xarray open in the package must name its engine. + +When ``engine`` is omitted, xarray auto-detects it by importing *every* +registered ``xarray.backends`` entry point and asking each one +``guess_can_open()``. The analysis3 environment registers 37 of them, +including argopy's ``argo`` and ``erddapy`` backends, which reach out to +raw.githubusercontent.com on import. A Gadi compute node has no route +there, so those imports block until they time out. + +The cost is paid once per process, at the *first* open that does not name +an engine -- measured at 38-46s per batch job. It is not attached to any +particular call site: pinning some opens and not others simply moves the +tax to the first one still unpinned (measured: pinning only the input-file +opens moved 40s out of the Dask sizing probe and into the CMIP7 range gate +in ``qc/cmip7.py``). So this has to hold for the whole package, which is +why the check is structural rather than a test of one function. + +Every file MOPPy opens is netCDF -- model output (NETCDF4 for the UM +atmosphere, NETCDF4_CLASSIC for MOM ocean and CICE ice), its own +intermediate output, and the bundled ``resources/*.nc`` -- all of which the +``netcdf4`` engine reads. +""" + +import ast +from pathlib import Path + +import pytest + +import access_moppy + +_OPENERS = {"open_dataset", "open_mfdataset", "open_zarr"} + + +def _unpinned_opens(): + """Yield ``path:lineno`` for every xarray open that omits ``engine``. + + A call that forwards ``**kwargs`` is accepted: ``CMORiser.load_dataset`` + builds one ``common_kwargs`` dict (which sets ``engine``) and expands it + into three ``open_mfdataset`` calls, and the expansion is opaque here. + """ + root = Path(access_moppy.__file__).parent + for path in sorted(root.rglob("*.py")): + if ".ipynb_checkpoints" in str(path): + continue + try: + tree = ast.parse(path.read_text()) + except SyntaxError: # pragma: no cover - not our source to fix + continue + for node in ast.walk(tree): + if ( + isinstance(node, ast.Call) + and isinstance(node.func, ast.Attribute) + and node.func.attr in _OPENERS + ): + keywords = {kw.arg for kw in node.keywords} + if "engine" not in keywords and None not in keywords: + yield f"{path.relative_to(root)}:{node.lineno}" + + +@pytest.mark.unit +def test_every_xarray_open_pins_the_engine(): + unpinned = list(_unpinned_opens()) + assert not unpinned, ( + "these xarray opens omit engine= and would pay the backend " + f"auto-detection cost: {unpinned}" + ) + + +@pytest.mark.unit +def test_load_dataset_common_kwargs_pins_the_engine(): + """Close the ``**kwargs`` hole the structural check has to allow.""" + source = Path(access_moppy.__file__).parent / "base.py" + tree = ast.parse(source.read_text()) + pinned = [ + value.value + for node in ast.walk(tree) + if isinstance(node, ast.Assign) + and any( + isinstance(t, ast.Name) and t.id == "common_kwargs" for t in node.targets + ) + and isinstance(node.value, ast.Dict) + for key, value in zip(node.value.keys, node.value.values) + if isinstance(key, ast.Constant) and key.value == "engine" + ] + assert pinned == ["netcdf4"], ( + "load_dataset's common_kwargs must pin engine='netcdf4'; it is " + "expanded into the open_mfdataset calls the structural check skips" + ) From 1a3feee307cc7ff43b248655b4934e88062de6b5 Mon Sep 17 00:00:00 2001 From: rhaegar325 Date: Mon, 7 Sep 2026 22:34:17 +1000 Subject: [PATCH 2/3] pre-commit and code coverage fix --- src/access_moppy/executors/dask_config.py | 4 +- src/access_moppy/ocean_supergrid.py | 4 +- tests/unit/test_frequency_detection.py | 63 +++++++++++++++++++++++ 3 files changed, 65 insertions(+), 6 deletions(-) diff --git a/src/access_moppy/executors/dask_config.py b/src/access_moppy/executors/dask_config.py index a4e6e61d..8f84fcfe 100644 --- a/src/access_moppy/executors/dask_config.py +++ b/src/access_moppy/executors/dask_config.py @@ -280,9 +280,7 @@ def _estimate_worker_memory_gb(variable, input_files, model_id): mapping = load_model_mappings(variable, model_id) model_vars = (mapping.get(cmor_name, {}) or {}).get("model_variables", []) or [] - with xr.open_dataset( - input_files[0], engine="netcdf4", decode_cf=False - ) as ds0: + with xr.open_dataset(input_files[0], engine="netcdf4", decode_cf=False) as ds0: steps_per_file = int(ds0.sizes.get("time", 1)) present = [v for v in model_vars if v in ds0.variables] per_file_mb = ( diff --git a/src/access_moppy/ocean_supergrid.py b/src/access_moppy/ocean_supergrid.py index f97f220f..8d644e3b 100644 --- a/src/access_moppy/ocean_supergrid.py +++ b/src/access_moppy/ocean_supergrid.py @@ -90,9 +90,7 @@ def load_supergrid(self, supergrid_file: str): for the 10 km grid), so actual computation is deferred until the first call to :meth:`extract_grid` via :meth:`_compute_grid`. """ - self.supergrid = xr.open_dataset( - supergrid_file, engine="netcdf4", chunks={} - ) + self.supergrid = xr.open_dataset(supergrid_file, engine="netcdf4", chunks={}) self._grid_computed = False def _compute_grid(self): diff --git a/tests/unit/test_frequency_detection.py b/tests/unit/test_frequency_detection.py index bdc9c9cf..ba4fba0b 100644 --- a/tests/unit/test_frequency_detection.py +++ b/tests/unit/test_frequency_detection.py @@ -16,6 +16,7 @@ IncompatibleFrequencyError, _detect_frequency_from_access_metadata, _detect_frequency_from_concatenated_files, + _detect_frequency_from_individual_files, _parse_access_frequency_metadata, detect_time_frequency_lazy, is_frequency_compatible, @@ -829,3 +830,65 @@ def test_frequency_validation_in_driver(self): print(f"Detected daily frequency: {freq}") print("Basic tests completed successfully!") + + +class TestIndividualFileFrequencyDetection: + """Cover the per-file fallback used when concatenation fails. + + ``_detect_frequency_from_individual_files`` opens each input file on its + own. It swallows per-file errors and only raises once *every* file has + failed, so a broken open here degrades to + ``ValueError: Could not detect frequency from any input files`` rather + than surfacing the real cause -- which is why this needs a test that + actually reads files rather than one that inspects the call. + """ + + def _write(self, path, day_offsets): + xr.Dataset( + {"tas": (["time", "lat", "lon"], np.zeros((len(day_offsets), 1, 1)))}, + coords={ + "time": ( + ["time"], + np.array(day_offsets, dtype=float), + {"units": "days since 2000-01-01"}, + ), + "lat": (["lat"], np.array([0.0])), + "lon": (["lon"], np.array([0.0])), + }, + ).to_netcdf(path) + + def test_detects_daily_frequency_across_separate_files(self): + with tempfile.TemporaryDirectory() as tmpdir: + files = [] + for i, offsets in enumerate(([0.0, 1.0, 2.0], [3.0, 4.0, 5.0])): + path = Path(tmpdir) / f"day_{i}.nc" + self._write(path, offsets) + files.append(str(path)) + + with warnings.catch_warnings(record=True) as caught: + warnings.simplefilter("always") + detected = _detect_frequency_from_individual_files(files) + + assert detected == pd.Timedelta(days=1) + assert not [w for w in caught if "Error processing file" in str(w.message)] + + def test_accepts_a_single_path_as_a_string(self): + with tempfile.TemporaryDirectory() as tmpdir: + path = Path(tmpdir) / "monthly.nc" + self._write(path, [0.0, 31.0, 62.0]) + + detected = _detect_frequency_from_individual_files(str(path)) + + assert detected == pd.Timedelta(days=31) + + def test_raises_when_no_file_yields_a_frequency(self): + with tempfile.TemporaryDirectory() as tmpdir: + path = Path(tmpdir) / "not_netcdf.nc" + path.write_text("this is not a netCDF file") + + with warnings.catch_warnings(record=True) as caught: + warnings.simplefilter("always") + with pytest.raises(ValueError, match="Could not detect frequency"): + _detect_frequency_from_individual_files([str(path)]) + + assert [w for w in caught if "Error processing file" in str(w.message)] From b80e749f136ecbb33d167f1f08d8312744932613 Mon Sep 17 00:00:00 2001 From: rhaegar325 Date: Tue, 8 Sep 2026 12:34:17 +1000 Subject: [PATCH 3/3] add tests for code coverage --- tests/unit/test_frequency_detection.py | 24 ++++++++++++++++++++ tests/unit/test_qc_plots.py | 31 ++++++++++++++++++++++++++ 2 files changed, 55 insertions(+) diff --git a/tests/unit/test_frequency_detection.py b/tests/unit/test_frequency_detection.py index ba4fba0b..672c7d3c 100644 --- a/tests/unit/test_frequency_detection.py +++ b/tests/unit/test_frequency_detection.py @@ -881,6 +881,30 @@ def test_accepts_a_single_path_as_a_string(self): assert detected == pd.Timedelta(days=31) + def test_warns_when_a_readable_file_has_no_detectable_frequency(self): + """A file that opens fine but carries a single timestep. + + This is the other arm of ``if freq is not None`` -- distinct from a + file that fails to open, and the one that produces the "Could not + detect frequency for file" warning rather than "Error processing". + """ + with tempfile.TemporaryDirectory() as tmpdir: + good = Path(tmpdir) / "daily.nc" + self._write(good, [0.0, 1.0, 2.0]) + single = Path(tmpdir) / "one_step.nc" + self._write(single, [0.0]) + + with warnings.catch_warnings(record=True) as caught: + warnings.simplefilter("always") + detected = _detect_frequency_from_individual_files( + [str(good), str(single)] + ) + + assert detected == pd.Timedelta(days=1) + messages = [str(w.message) for w in caught] + assert any("Could not detect frequency for file" in m for m in messages) + assert not any("Error processing file" in m for m in messages) + def test_raises_when_no_file_yields_a_frequency(self): with tempfile.TemporaryDirectory() as tmpdir: path = Path(tmpdir) / "not_netcdf.nc" diff --git a/tests/unit/test_qc_plots.py b/tests/unit/test_qc_plots.py index 1e64c2ec..71593db7 100644 --- a/tests/unit/test_qc_plots.py +++ b/tests/unit/test_qc_plots.py @@ -281,6 +281,37 @@ def _fake(path, *, qc_dir=None, comparison_store=None, preferred_member=None): assert called_with == [nc_path] + def test_combined_overlay_loaded_when_comparison_store_given(self, temp_dir): + """The split-files path forwards its own open dataset to _load_overlay. + + `generate_qc_plots` has an equivalent branch that is covered; this one + was not, so the `comparison_store` arm of + `generate_qc_plots_for_split_files` had never been executed. + """ + paths = _write_split_cmip_files(temp_dir, "tas_Amon", n_chunks=2) + qc_dir = temp_dir / "qc" + from access_moppy.qc import plots as plots_module + + with patch.object( + plots_module, "_load_overlay", return_value=None + ) as mock_overlay: + result = plots_module.generate_qc_plots_for_split_files( + paths, + qc_dir=qc_dir, + comparison_store=temp_dir / "comparison", + preferred_member="r1i1p1f1", + ) + + assert result == qc_dir + assert (qc_dir / "tas_Amon_timeseries.png").exists() + mock_overlay.assert_called_once() + kwargs = mock_overlay.call_args.kwargs + assert kwargs["var_name"] == "tas" + assert kwargs["store_path"] == temp_dir / "comparison" + assert kwargs["preferred_member"] == "r1i1p1f1" + # the dataset handed to the overlay is the combined one, not a single split + assert kwargs["ds"].sizes["time"] == 8 + def test_combined_timeseries_written(self, temp_dir): paths = _write_split_cmip_files(temp_dir, "tas_Amon", n_chunks=2) qc_dir = temp_dir / "qc"