|
16 | 16 | from structlog.testing import capture_logs |
17 | 17 |
|
18 | 18 | from eopf_geozarr.s2_optimization.s2_multiscale import ( |
| 19 | + _coarsen_variable, |
19 | 20 | calculate_aligned_chunk_size, |
20 | 21 | calculate_simple_shard_dimensions, |
21 | 22 | create_downsampled_resolution_group, |
@@ -219,6 +220,31 @@ def test_create_multiscale_from_datatree( |
219 | 220 | assert [k for k in o_keys if expected_structure_flat[k] != observed_structure_flat[k]] == [] |
220 | 221 |
|
221 | 222 |
|
| 223 | +# --------------------------------------------------------------------------- |
| 224 | +# _coarsen_variable |
| 225 | +# --------------------------------------------------------------------------- |
| 226 | + |
| 227 | + |
| 228 | +def test_coarsen_variable_classification() -> None: |
| 229 | + """Classification variables should be downsampled via subsample.""" |
| 230 | + data = np.arange(16, dtype="uint8").reshape(4, 4) |
| 231 | + var = xr.DataArray(data, dims=["y", "x"], coords={"y": np.arange(4.0), "x": np.arange(4.0)}) |
| 232 | + result = _coarsen_variable("scl", var, factor=2) |
| 233 | + assert result.shape == (2, 2) |
| 234 | + assert result.dtype == np.uint8 |
| 235 | + # subsample picks top-left of each 2x2 block |
| 236 | + np.testing.assert_array_equal(result.values, data[::2, ::2]) |
| 237 | + |
| 238 | + |
| 239 | +def test_coarsen_variable_quality_mask() -> None: |
| 240 | + """Quality mask variables should be downsampled via max.""" |
| 241 | + data = np.array([[0, 1], [2, 3]], dtype="uint8") |
| 242 | + var = xr.DataArray(data, dims=["y", "x"], coords={"y": np.arange(2.0), "x": np.arange(2.0)}) |
| 243 | + result = _coarsen_variable("quality_cirrus", var, factor=2) |
| 244 | + assert result.shape == (1, 1) |
| 245 | + assert result.values.item() == 3 |
| 246 | + |
| 247 | + |
222 | 248 | # --------------------------------------------------------------------------- |
223 | 249 | # inject_missing_bands |
224 | 250 | # --------------------------------------------------------------------------- |
@@ -274,12 +300,16 @@ def test_inject_missing_bands_respects_bands_filter() -> None: |
274 | 300 | def test_inject_missing_bands_skips_existing() -> None: |
275 | 301 | """Bands already present in the dataset should not be overwritten.""" |
276 | 302 | dt = _make_reflectance_datatree() |
277 | | - r10m_ds = dt["measurements/reflectance/r10m"].to_dataset() |
| 303 | + r20m_ds = dt["measurements/reflectance/r20m"].to_dataset() |
278 | 304 |
|
279 | | - result = inject_missing_bands(r10m_ds, dt, target_resolution=10) |
| 305 | + # Pre-populate b08 with a sentinel value so we can verify it is NOT replaced. |
| 306 | + sentinel = np.full((60, 60), 999, dtype="uint16") |
| 307 | + r20m_ds["b08"] = (["y", "x"], sentinel) |
| 308 | + |
| 309 | + result = inject_missing_bands(r20m_ds, dt, target_resolution=20, bands={"b08"}) |
280 | 310 |
|
281 | | - # No bands have a native resolution finer than 10m, so nothing changes |
282 | | - assert set(result.data_vars) == set(r10m_ds.data_vars) |
| 311 | + # b08 was already present — inject_missing_bands must leave it untouched. |
| 312 | + np.testing.assert_array_equal(result["b08"].values, sentinel) |
283 | 313 |
|
284 | 314 |
|
285 | 315 | def test_inject_missing_bands_noop_when_no_source() -> None: |
|
0 commit comments