diff --git a/src/openbench/core/evaluation.py b/src/openbench/core/evaluation.py index 943773d8..aaf14720 100644 --- a/src/openbench/core/evaluation.py +++ b/src/openbench/core/evaluation.py @@ -231,6 +231,8 @@ def process_metric(self, metric, s, o, vkey=""): pb_da = pb.rename(metric) else: pb_da = xr.DataArray(pb, coords=[o.lat, o.lon], dims=["lat", "lon"], name=metric) + # ponytail: compute before HDF5 write; lazy source reads inside to_netcdf can hang on Windows/netCDF4. + pb_da = pb_da.load() # Use output manager if available, otherwise fallback to original method if self.output_manager: @@ -261,6 +263,8 @@ def process_score(self, score, s, o, vkey=""): pb = getattr(self, score)(s, o) pb = pb.squeeze() pb_da = xr.DataArray(pb, coords=[o.lat, o.lon], dims=["lat", "lon"], name=score) + # ponytail: compute before HDF5 write; lazy source reads inside to_netcdf can hang on Windows/netCDF4. + pb_da = pb_da.load() # Use output manager if available, otherwise fallback to original method if self.output_manager: diff --git a/tests/test_netcdf_lifecycle_atomic.py b/tests/test_netcdf_lifecycle_atomic.py index 03ad6ff1..1abe7849 100644 --- a/tests/test_netcdf_lifecycle_atomic.py +++ b/tests/test_netcdf_lifecycle_atomic.py @@ -1,6 +1,7 @@ import os from pathlib import Path +import dask.array as da import numpy as np import pytest import xarray as xr @@ -125,6 +126,48 @@ def failing_to_netcdf(self, path, *args, **kwargs): assert not list(target_dir.glob(".Runoff_ref_TestRef_sim_SimA_Overall_Score.nc.*.tmp.nc")) +@pytest.mark.parametrize( + ("method_name", "result_name", "output_name", "result"), + [ + ( + "process_metric", + "bias", + "Runoff_ref_TestRef_sim_SimA_bias.nc", + xr.DataArray( + da.from_array(np.array([[2.0, 3.0], [4.0, 5.0]]), chunks=(1, 2)), + coords={"lat": [0.0, 1.0], "lon": [10.0, 11.0]}, + dims=("lat", "lon"), + ), + ), + ( + "process_score", + "Overall_Score", + "Runoff_ref_TestRef_sim_SimA_Overall_Score.nc", + da.from_array(np.array([[0.5, 0.6], [0.7, 0.8]]), chunks=(1, 2)), + ), + ], +) +def test_evaluation_netcdf_outputs_materialize_lazy_results_before_write( + tmp_path, monkeypatch, method_name, result_name, output_name, result +): + import openbench.core.evaluation as evaluation_module + + processor = _evaluation_processor(tmp_path) + setattr(processor, result_name, lambda _s, _o: result) + writes = [] + + def fake_write(data, output_path, **_kwargs): + writes.append((hasattr(data.data, "compute"), data.values.copy(), Path(output_path).name)) + + monkeypatch.setattr(evaluation_module, "_write_netcdf_atomic", fake_write) + + getattr(processor, method_name)(result_name, _lat_lon_array(), _lat_lon_array()) + + assert writes[0][0] is False + assert writes[0][2] == output_name + np.testing.assert_allclose(writes[0][1], result.compute() if hasattr(result, "compute") else result.values) + + def test_core_comparison_uses_atomic_netcdf_writes_only(): source = Path("src/openbench/core/comparison.py").read_text(encoding="utf-8")