From 68a0c5b4992a58e9cfadb4e1ea4377e13a927a6f Mon Sep 17 00:00:00 2001 From: Richard van Schaik Date: Sun, 25 Oct 2026 03:04:06 +0100 Subject: [PATCH 1/5] fix(fci): bounded FDHSI entry selection Cache measured provider coordinate layouts across repeat cycles with version and chunk-set guards. Fetch demanded metadata pages concurrently outside HDF5 locks and record selection lineage. Co-Authored-By: Claude Opus 5.5 --- docs/fci-eumetsat.md | 37 +++- .../connectors/fci_eumetsat/connector.py | 164 ++++++++++++++++-- tests/test_fci_eumetsat.py | 93 ++++++++++ 3 files changed, 269 insertions(+), 25 deletions(-) diff --git a/docs/fci-eumetsat.md b/docs/fci-eumetsat.md index 4974cdc..efefaa9 100644 --- a/docs/fci-eumetsat.md +++ b/docs/fci-eumetsat.md @@ -85,13 +85,36 @@ No rectangular query bbox is presented as the provider footprint. ## Chunk downloads and native grids EUMDAC supplies `AccessToken`, `DataStore.get_collection().search()`, -`Product.entries` and `Product.open(entry=...)`. L1c fetch inspects HDF5 -metadata and coordinate vectors with EUMDAC byte-range requests across body -entries, then downloads the complete entries containing requested-channel -pixel centres inside the bbox, plus the trailer. Chunk heights can vary; -selection uses their published scan coordinates. Pixel arrays in other body -entries are not decoded for selection. A server that does not honor byte ranges -raises an error. This transport behavior still needs a credentialed live run. +`Product.entries` and `Product.open(entry=...)`. On a cold L1c layout, fetch +reads the provider's HDF5 projection and requested-channel x/y coordinate +vectors across BODY entries. It downloads complete entries containing +requested-channel pixel centres inside the bbox, plus the trailer. Chunk +heights can vary; selection uses measured coordinates rather than a table +of inferred chunk rows. Pixel arrays are not decoded for selection. + +Reuse the same `FCIConnector` for successive repeat cycles. It keeps up to eight +measured layouts in memory, partitioned by collection, published product and +processor versions, platform/product/encoding/processing modes, exact BODY +chunk-number set (including count), and requested channel set. A matching later +cycle evaluates its bbox against those grids with **zero metadata range calls**; +whole-entry downloads and catalogue requests still occur. Different bboxes +reuse the coordinates. Changing channels measures a separate layout. Missing +version/identity metadata or a changed version/chunk set triggers measured +selection for that product. No cache survives a connector/process restart. + +`FCIResult.entry_selection`, Dataset attrs and canonical `entry_selection` +record `mode` (`measured` or `cached`), fallback `reason`, BODY count, worker +limit and coordinate basis. Reuse assumes provider grids remain stable within +that versioned identity. The synthetic fixtures prove the cache mechanism; +real-header stability and the 25-cycle/30-minute cloud budget still require +RefCal's cloud-side proof. No hardcoded PUG chunk layout is used. + +Cold measurements fetch only demanded 4 KiB metadata pages (plus 64-byte +superblocks), using a bounded pool: `FCIConnector(metadata_workers=4)` by +default. Local HDF5 inspection pauses on a missing page; network reads run +outside HDF5's global lock so separate BODY entries can progress concurrently. +A server that ignores byte ranges raises an error. This transport behavior +still needs a credentialed cloud run. L2 products are single NetCDF entries and are downloaded before local clipping. Short cache filenames avoid Windows NetCDF path limits; result metadata retains the original provider entry names. diff --git a/src/spectraccess/connectors/fci_eumetsat/connector.py b/src/spectraccess/connectors/fci_eumetsat/connector.py index 036bc6c..30e1899 100644 --- a/src/spectraccess/connectors/fci_eumetsat/connector.py +++ b/src/spectraccess/connectors/fci_eumetsat/connector.py @@ -6,6 +6,8 @@ import os import re import shutil +from collections import OrderedDict +from concurrent.futures import ThreadPoolExecutor from copy import deepcopy from contextlib import contextmanager from dataclasses import dataclass, field @@ -117,6 +119,7 @@ class FCIResult: bbox: tuple[float, float, float, float] entries: tuple[str, ...] local_files: Mapping[str, str] = field(default_factory=dict) + entry_selection: Mapping[str, Any] = field(default_factory=dict) def _path(result, entry): @@ -137,6 +140,13 @@ def _provider_failure(exc, stage, collection=None, *, values=()): provider_error=FCIProviderError, authorization_error=FCIAuthorizationError) +class _RangeNeeded(Exception): + """Suspend HDF5 inspection until a missing metadata page is fetched.""" + + def __init__(self, stream, block): + self.stream, self.block = stream, block + + class _EntryRange(io.RawIOBase): """Seekable HDF5 metadata access using EUMDAC's maintained byte-range API. @@ -144,9 +154,10 @@ class _EntryRange(io.RawIOBase): pixel arrays are never accessed while choosing entries. A server that ignores Range is rejected rather than downloading a complete repeat cycle. """ - block_size = 65536 + block_size = 4096 - def __init__(self, product, entry): + def __init__(self, product, entry, *, deferred=False): + self.deferred = deferred self.product, self.entry, self.position = product, entry, 0 self.blocks = {} header = self._range(0, 64) @@ -193,12 +204,18 @@ def read(self, size=-1): block = self.position // self.block_size start = block * self.block_size if block not in self.blocks: - self.blocks[block] = self._range(start, min(start + self.block_size, self.length)) + if self.deferred: + raise _RangeNeeded(self, block) + self.load_block(block) stop = min(end, start + len(self.blocks[block])) parts.append(self.blocks[block][self.position - start:stop - start]) self.position = stop return b"".join(parts) + def load_block(self, block): + start = block * self.block_size + self.blocks[block] = self._range(start, min(start + self.block_size, self.length)) + def readinto(self, buffer): data = self.read(len(buffer)) buffer[:len(data)] = data @@ -240,20 +257,94 @@ def _window(x, y, projection, bbox): return slice(rows[0], rows[-1] + 1), slice(min(cols), max(cols) + 1) -def _entry_intersects(product, entry, bbox, channels): - with _EntryRange(product, entry) as stream, h5netcdf.File(stream, "r") as nc: - data = nc.groups["data"] - projection = dict(data.variables["mtg_geos_projection"].attrs) +def _entry_geometry(stream, channels): + # Retain measured vectors, including variable chunk heights and channel grids. + # Never infer chunk rows from their ordinal number. + stream.seek(0) + with h5py.File(stream, "r") as nc: + data = nc["data"] + projection = {k: np.asarray(v).item() if np.asarray(v).size == 1 else v + for k, v in data["mtg_geos_projection"].attrs.items()} + grids = {} for channel in channels: - if channel not in data.groups: - continue - group = data.groups[channel].groups["measured"] - rows, cols = _window(_angles(group.variables["x"]), _angles(group.variables["y"]), projection, bbox) - if rows.stop and cols.stop: - return True + if channel in data: + group = data[channel]["measured"] + grids[channel] = (_angles(group["x"]), _angles(group["y"])) + return projection, grids + + +def _geometry_intersects(geometry, bbox): + projection, grids = geometry + for x, y in grids.values(): + rows, cols = _window(x, y, projection, bbox) + if rows.stop and cols.stop: + return True return False +def _entry_intersects(product, entry, bbox, channels): + with _EntryRange(product, entry) as stream: + return _geometry_intersects(_entry_geometry(stream, channels), bbox) + + +def _measure_entries(product, entries, channels, token, workers): + """Read metadata pages concurrently, without holding HDF5's global lock. + + HDF5 inspection is local and serial. A cache miss unwinds the file handle; + the pool fetches one missing page per entry, then inspection resumes from + cached pages. This avoids serial network waits inside HDF5's Python lock. + Auth context must be propagated explicitly to the transport threads. + """ + def create(entry): + with _eumdac_context(token=token): + return _EntryRange(product, entry, deferred=True) + + def load(needed): + with _eumdac_context(token=token): + needed.stream.load_block(needed.block) + + geometry = {} + with ThreadPoolExecutor(max_workers=workers) as pool: + streams = list(pool.map(create, entries)) + try: + pending = list(zip(entries, streams)) + while pending: + missing, remaining = [], [] + for entry, stream in pending: + try: + geometry[entry] = _entry_geometry(stream, channels) + except _RangeNeeded as needed: + missing.append(needed) + remaining.append((entry, stream)) + list(pool.map(load, missing)) + pending = remaining + finally: + for stream in streams: + stream.close() + return geometry + + +def _layout_identity(target, entries, channels): + info = target.raw.get("download_properties", {}).get("productInformation", {}) + if not info: + info = target.raw.get("properties", {}).get("productInformation", {}) + version = info.get("productVersion", info.get("version")) + processing = info.get("processingInformation", {}).get("processorVersion", info.get("processingVersion")) + identities = [] + for entry in entries: + match = _SUFFIX.search(entry) + if match is None or "-FDHSI-FD-" not in entry: + return None + # Platform, product kind, distribution format and processing modes also + # partition layouts. Timestamps and repeat-cycle number do not. + identities.append((entry.split("-CHK-")[0], entry.split("-BODY-")[1].split("_")[0], + tuple(entry[:match.start()].split("_")[-2:]), + match.groups()[2:5], int(match[7]))) + if not version or not processing or not identities or len(set(identities)) != len(identities): + return None + return (target.collection, str(version), str(processing), tuple(sorted(identities)), tuple(sorted(channels))) + + def _file_info(path): match = _SUFFIX.search(path.name) if match is None or "-FDHSI-FD-" not in path.name: @@ -316,8 +407,39 @@ class FCIConnector(Connector): """EUMDAC discovery and chunk fetch, Satpy values, native-window Dataset.""" credential_provider = "eumetsat" - def __init__(self, *, credentials: CredentialSource | None = None): + def __init__(self, *, credentials: CredentialSource | None = None, metadata_workers=4): + if not isinstance(metadata_workers, int) or metadata_workers < 1: + raise ValueError("metadata_workers must be a positive integer") self.credentials = credentials + self.metadata_workers = metadata_workers + self._entry_layouts = OrderedDict() + + def _select_bodies(self, product, target, entries, channels, bbox, token): + key = _layout_identity(target, entries, channels) + cached = self._entry_layouts.get(key) if key is not None else None + reason = None + if cached is not None: + self._entry_layouts.move_to_end(key) + mode = "cached" + else: + reason = ("missing_version_or_chunk_identity" if key is None else + "version_or_chunk_set_mismatch" if self._entry_layouts else "cold_cache") + geometry = _measure_entries(product, entries, channels, token, self.metadata_workers) + mode = "measured" + if key is not None: + # Store by chunk identity rather than provider entry order. + ordered = sorted(entries, key=lambda e: int(_SUFFIX.search(e)[7])) + self._entry_layouts[key] = tuple(geometry[e] for e in ordered) + while len(self._entry_layouts) > 8: + self._entry_layouts.popitem(last=False) + if cached is not None: + ordered = sorted(entries, key=lambda e: int(_SUFFIX.search(e)[7])) + geometry = dict(zip(ordered, cached)) + selected = {e for e in entries if _geometry_intersects(geometry[e], bbox)} + return selected, {"mode": mode, "reason": reason, "body_count": len(entries), + "metadata_workers": self.metadata_workers, + "basis": "measured_provider_coordinates"} + def _store(self, credentials=None): credential = resolve("eumetsat", credentials if credentials is not None else self.credentials) @@ -366,19 +488,24 @@ def fetch(self, target, *, dest=None, bbox=None, channels=CHANNELS, credentials= import hashlib path = path / hashlib.sha256((target.collection + target.product_id).encode()).hexdigest()[:20] path.mkdir(parents=True, exist_ok=True) - downloaded, local_files = [], {} + downloaded, local_files, selection = [], {}, {} with _eumdac_context(token=store.token): try: product = store.get_product(target.collection, target.product_id) available = tuple(product.entries) l1c = target.collection == COLLECTIONS["l1c"] selected = [] + bodies = tuple(e for e in available if e.endswith(".nc") and "-BODY-" in e) + selected_bodies = set() + if l1c: + selected_bodies, selection = self._select_bodies( + product, target, bodies, channels, bbox, store.token) for entry in available: if not entry.endswith(".nc"): continue if not l1c or "-TRAIL-" in entry: selected.append(entry) - elif "-BODY-" in entry and _entry_intersects(product, entry, bbox, channels): + elif entry in selected_bodies: selected.append(entry) if l1c and any("-BODY-" in e for e in selected) and not any("-TRAIL-" in e for e in selected): raise ValueError("FCI repeat cycle has no trailer entry") @@ -399,7 +526,7 @@ def fetch(self, target, *, dest=None, bbox=None, channels=CHANNELS, credentials= local_files[filename] = local_name except Exception as exc: raise _provider_failure(exc, "entry fetch", target.collection, values=_auth_values(token=store.token)) from exc - return FCIResult(path, target, datetime.now(timezone.utc), bbox, tuple(downloaded), local_files) + return FCIResult(path, target, datetime.now(timezone.utc), bbox, tuple(downloaded), local_files, selection) def parse(self, result, **kwargs): return self.parse_canonical(result, **kwargs) @@ -411,6 +538,7 @@ def parse_canonical(self, result, *, bbox=None, channels=CHANNELS): frame = target_to_canonical(result.target, retrieved_at=result.retrieved_at, provider_attrs=ds.attrs.get("provider_attrs", {}), qa=ds.attrs.get("provider_product_qa")) + frame["entry_selection"] = [dict(result.entry_selection)] if any("radiometric_noise_lut_noise" in name for name in ds): frame["unc_definition"] = NOISE_DEFINITION return frame @@ -427,7 +555,7 @@ def read(self, result, *, bbox=None, channels=CHANNELS): integration_start=result.target.start.isoformat(), integration_end=result.target.end.isoformat(), fetched_bbox=result.bbox, requested_bbox=bbox, source_entries=result.entries, - provider_coverage=result.target.coverage) + provider_coverage=result.target.coverage, entry_selection=dict(result.entry_selection)) if result.target.footprint: dataset.attrs["footprint_geometry"] = result.target.footprint return dataset diff --git a/tests/test_fci_eumetsat.py b/tests/test_fci_eumetsat.py index 33a3341..6fedcaa 100644 --- a/tests/test_fci_eumetsat.py +++ b/tests/test_fci_eumetsat.py @@ -493,3 +493,96 @@ def test_invalid_discovery_inputs(kwargs): params.update(kwargs) with pytest.raises(ValueError): FCIConnector().discover(**params) + + +def test_fetch_reuses_measured_layout_across_cycles(fixture, tmp_path, monkeypatch): + http = DataStoreHTTP(fixture, monkeypatch) + connector = FCIConnector(credentials=CREDENTIAL) + targets = connector.discover(bbox=BBOX, start=START, end=START + timedelta(minutes=20), products=("l1c",)) + bbox = (-.1, -.05, .1, -.005) + first = connector.fetch(targets[0], dest=tmp_path, bbox=bbox, channels=("ir_105", "nir_13")) + first_calls = sum(bool(r) for _, r in http.entries_requested) + assert first_calls > 0 and first.entry_selection["mode"] == "measured" + http.calls.clear() + # Reversing provider order must not associate one chunk's grid with another. + links = http.search["features"][1]["properties"]["links"]["sip-entries"] + links.reverse() + second = connector.fetch(targets[1], dest=tmp_path, bbox=bbox, channels=("nir_13", "ir_105")) + second_calls = sum(bool(r) for _, r in http.entries_requested) + assert second_calls == 0 + assert set(first.entries) == set(second.entries) + assert second.entry_selection["mode"] == "cached" + # A changed bbox is evaluated from cached coordinates, not cached selection. + http.calls.clear() + outside = connector.fetch(targets[1], dest=tmp_path, bbox=(150., -10., 160., 10.), channels=("ir_105", "nir_13")) + assert not any("-BODY-" in e for e in outside.entries) + assert not any(r for _, r in http.entries_requested) + # Independent cold measurement selects the same entries. + measured = FCIConnector(credentials=CREDENTIAL).fetch(targets[1], dest=tmp_path, bbox=bbox, channels=("ir_105", "nir_13")) + assert set(measured.entries) == set(second.entries) + assert connector.read(second, channels=("ir_105",)).attrs["entry_selection"]["mode"] == "cached" + assert connector.parse_canonical(second, channels=("ir_105",)).iloc[0].entry_selection["mode"] == "cached" + print(f"metadata ranges: first={first_calls}, second={second_calls}") + + +@pytest.mark.parametrize("mismatch", ["version", "processing", "count", "missing_version"]) +def test_fetch_layout_mismatch_is_measured(fixture, tmp_path, monkeypatch, mismatch): + from dataclasses import replace + from copy import deepcopy + http = DataStoreHTTP(fixture, monkeypatch) + connector = FCIConnector(credentials=CREDENTIAL) + targets = connector.discover(bbox=BBOX, start=START, end=START + timedelta(minutes=20), products=("l1c",)) + connector.fetch(targets[0], dest=tmp_path, channels=("ir_105",)) + changed = targets[1] + if mismatch == "count": + links = http.search["features"][1]["properties"]["links"]["sip-entries"] + links[:] = [link for link in links if "0003.nc" not in link["title"]] + else: + raw = deepcopy(changed.raw) + info = raw["download_properties"]["productInformation"] + if mismatch == "version": + info["productVersion"] = "changed" + elif mismatch == "processing": + info["processingInformation"]["processorVersion"] = "changed" + else: + info.pop("productVersion") + changed = replace(changed, raw=raw) + http.calls.clear() + fetched = connector.fetch(changed, dest=tmp_path, channels=("ir_105",)) + assert any(r for _, r in http.entries_requested) + assert fetched.entry_selection["mode"] == "measured" + assert fetched.entry_selection["reason"] == ("missing_version_or_chunk_identity" if mismatch == "missing_version" else "version_or_chunk_set_mismatch") + measured = FCIConnector(credentials=CREDENTIAL).fetch(changed, dest=tmp_path, channels=("ir_105",)) + assert set(fetched.entries) == set(measured.entries) + + +def test_fetch_metadata_ranges_are_bounded_and_concurrent(fixture, tmp_path, monkeypatch): + import threading + import time + http = DataStoreHTTP(fixture, monkeypatch) + original = http.send + lock = threading.Lock() + active = peak = 0 + + def delayed(request): + nonlocal active, peak + if not request.headers.get("Range"): + return original(request) + with lock: + active += 1 + peak = max(peak, active) + try: + time.sleep(.002) + return original(request) + finally: + with lock: + active -= 1 + + monkeypatch.setattr(http, "send", delayed) + FCIConnector(credentials=CREDENTIAL, metadata_workers=2).fetch(target(), dest=tmp_path, channels=("ir_105",)) + assert peak == 2 + for _, byte_range in http.entries_requested: + if byte_range: + begin, end = map(int, byte_range.removeprefix("bytes=").split("-")) + assert end - begin + 1 <= 4096 + assert all(stream.closed for stream in http.streams) From dedc906bc712de3a56a5499ea1dc672bc652a616 Mon Sep 17 00:00:00 2001 From: Richard van Schaik Date: Sun, 25 Oct 2026 05:38:52 +0100 Subject: [PATCH 2/5] fix(fci): persist measured entry layouts across processes Co-Authored-By: Claude Opus 5.5 --- .../connectors/fci_eumetsat/connector.py | 24 +++++++++++++++++++ tests/test_fci_eumetsat.py | 23 ++++++++++++++++++ 2 files changed, 47 insertions(+) diff --git a/src/spectraccess/connectors/fci_eumetsat/connector.py b/src/spectraccess/connectors/fci_eumetsat/connector.py index 30e1899..85df1f9 100644 --- a/src/spectraccess/connectors/fci_eumetsat/connector.py +++ b/src/spectraccess/connectors/fci_eumetsat/connector.py @@ -2,6 +2,8 @@ from __future__ import annotations import io +import gzip +import json import itertools import os import re @@ -414,6 +416,28 @@ def __init__(self, *, credentials: CredentialSource | None = None, metadata_work self.metadata_workers = metadata_workers self._entry_layouts = OrderedDict() + def export_layout_cache(self): + """Portable measured coordinates; contains no credentials or pixels.""" + rows = [[key, [[projection, {channel: [x.tolist(), y.tolist()] + for channel, (x, y) in grids.items()}] for projection, grids in layout]] + for key, layout in self._entry_layouts.items()] + return gzip.compress(json.dumps({"schema": "fci-entry-layout/1", "layouts": rows}).encode()) + + def import_layout_cache(self, data): + """Restore layouts; normal version/chunk identity checks still apply.""" + def frozen(value): + return tuple(frozen(v) for v in value) if isinstance(value, list) else value + body = json.loads(gzip.decompress(data)) + if body.get("schema") != "fci-entry-layout/1": + raise ValueError("unsupported FCI entry layout cache") + layouts = OrderedDict() + for key, layout in body["layouts"][-8:]: + layouts[frozen(key)] = tuple((projection, {channel: (np.asarray(x), np.asarray(y)) + for channel, (x, y) in grids.items()}) for projection, grids in layout) + self._entry_layouts.update(layouts) + while len(self._entry_layouts) > 8: + self._entry_layouts.popitem(last=False) + def _select_bodies(self, product, target, entries, channels, bbox, token): key = _layout_identity(target, entries, channels) cached = self._entry_layouts.get(key) if key is not None else None diff --git a/tests/test_fci_eumetsat.py b/tests/test_fci_eumetsat.py index 6fedcaa..531ff72 100644 --- a/tests/test_fci_eumetsat.py +++ b/tests/test_fci_eumetsat.py @@ -525,6 +525,29 @@ def test_fetch_reuses_measured_layout_across_cycles(fixture, tmp_path, monkeypat print(f"metadata ranges: first={first_calls}, second={second_calls}") +def test_fetch_restores_layout_in_fresh_connector_and_measures_mismatch(fixture, tmp_path, monkeypatch): + from copy import deepcopy + from dataclasses import replace + http = DataStoreHTTP(fixture, monkeypatch) + connector = FCIConnector(credentials=CREDENTIAL) + targets = connector.discover(bbox=BBOX, start=START, end=START + timedelta(minutes=20), products=("l1c",)) + first = connector.fetch(targets[0], dest=tmp_path, bbox=BBOX, channels=("ir_105",)) + data = connector.export_layout_cache() + restored = FCIConnector(credentials=CREDENTIAL) + restored.import_layout_cache(data) + http.calls.clear() + warm = restored.fetch(targets[1], dest=tmp_path, bbox=BBOX, channels=("ir_105",)) + assert warm.entry_selection["mode"] == "cached" + assert not any(r for _, r in http.entries_requested) + assert set(first.entries) == set(warm.entries) + raw = deepcopy(targets[1].raw) + raw["download_properties"]["productInformation"]["productVersion"] = "changed" + http.calls.clear() + mismatch = restored.fetch(replace(targets[1], raw=raw), dest=tmp_path, bbox=BBOX, channels=("ir_105",)) + assert mismatch.entry_selection["mode"] == "measured" + assert any(r for _, r in http.entries_requested) + + @pytest.mark.parametrize("mismatch", ["version", "processing", "count", "missing_version"]) def test_fetch_layout_mismatch_is_measured(fixture, tmp_path, monkeypatch, mismatch): from dataclasses import replace From b71d48cd4318b303996a0c2e233ee537d2b81158 Mon Sep 17 00:00:00 2001 From: Richard van Schaik Date: Sun, 25 Oct 2026 06:29:50 +0100 Subject: [PATCH 3/5] fix(fci): bind incomplete-version layouts to product checksum Add test_fetch_incomplete_version_reuses_only_checksum_bound_product: fresh-process exact-product reuse and checksum mismatch remeasurement. Keep versioned cross-cycle cache guards. Co-Authored-By: Claude Opus 5.5 --- .../connectors/fci_eumetsat/connector.py | 8 +++++- tests/test_fci_eumetsat.py | 28 +++++++++++++++++++ 2 files changed, 35 insertions(+), 1 deletion(-) diff --git a/src/spectraccess/connectors/fci_eumetsat/connector.py b/src/spectraccess/connectors/fci_eumetsat/connector.py index 85df1f9..50239ea 100644 --- a/src/spectraccess/connectors/fci_eumetsat/connector.py +++ b/src/spectraccess/connectors/fci_eumetsat/connector.py @@ -342,8 +342,14 @@ def _layout_identity(target, entries, channels): identities.append((entry.split("-CHK-")[0], entry.split("-BODY-")[1].split("_")[0], tuple(entry[:match.start()].split("_")[-2:]), match.groups()[2:5], int(match[7]))) - if not version or not processing or not identities or len(set(identities)) != len(identities): + if not identities or len(set(identities)) != len(identities): return None + if not version or not processing: + checksum = target.raw.get("properties", {}).get("md5") + # Incomplete processing metadata cannot establish cross-cycle identity. + # A published archive checksum can still bind this exact product. + return ((target.collection, "exact-product-md5", target.product_id, str(checksum), + tuple(sorted(identities)), tuple(sorted(channels))) if checksum else None) return (target.collection, str(version), str(processing), tuple(sorted(identities)), tuple(sorted(channels))) diff --git a/tests/test_fci_eumetsat.py b/tests/test_fci_eumetsat.py index 531ff72..9188d68 100644 --- a/tests/test_fci_eumetsat.py +++ b/tests/test_fci_eumetsat.py @@ -548,6 +548,34 @@ def test_fetch_restores_layout_in_fresh_connector_and_measures_mismatch(fixture, assert any(r for _, r in http.entries_requested) +def test_fetch_incomplete_version_reuses_only_checksum_bound_product(fixture, tmp_path, monkeypatch): + from copy import deepcopy + from dataclasses import replace + http = DataStoreHTTP(fixture, monkeypatch) + connector = FCIConnector(credentials=CREDENTIAL) + target = connector.discover(bbox=BBOX, start=START, end=START, products=("l1c",))[0] + raw = deepcopy(target.raw) + info = raw["download_properties"]["productInformation"] + info["productVersion"] = "Original" + info["processingInformation"].pop("processorVersion") + raw["properties"]["md5"] = "synthetic-archive-checksum-a" + target = replace(target, raw=raw) + first = connector.fetch(target, dest=tmp_path, channels=("ir_105",)) + restored = FCIConnector(credentials=CREDENTIAL) + restored.import_layout_cache(connector.export_layout_cache()) + http.calls.clear() + same = restored.fetch(target, dest=tmp_path, channels=("ir_105",)) + assert same.entry_selection["mode"] == "cached" + assert set(first.entries) == set(same.entries) + assert not any(r for _, r in http.entries_requested) + raw = deepcopy(raw) + raw["properties"]["md5"] = "synthetic-archive-checksum-b" + http.calls.clear() + changed = restored.fetch(replace(target, raw=raw), dest=tmp_path, channels=("ir_105",)) + assert changed.entry_selection["mode"] == "measured" + assert any(r for _, r in http.entries_requested) + + @pytest.mark.parametrize("mismatch", ["version", "processing", "count", "missing_version"]) def test_fetch_layout_mismatch_is_measured(fixture, tmp_path, monkeypatch, mismatch): from dataclasses import replace From de577f52cc36fb610199153de8fb24c382d62f2f Mon Sep 17 00:00:00 2001 From: Richard van Schaik Date: Sun, 25 Oct 2026 06:31:01 +0100 Subject: [PATCH 4/5] test(fci): exercise missing version without an archive checksum Co-Authored-By: Claude Opus 5.5 --- tests/test_fci_eumetsat.py | 1 + 1 file changed, 1 insertion(+) diff --git a/tests/test_fci_eumetsat.py b/tests/test_fci_eumetsat.py index 9188d68..641821c 100644 --- a/tests/test_fci_eumetsat.py +++ b/tests/test_fci_eumetsat.py @@ -597,6 +597,7 @@ def test_fetch_layout_mismatch_is_measured(fixture, tmp_path, monkeypatch, misma info["processingInformation"]["processorVersion"] = "changed" else: info.pop("productVersion") + raw["properties"].pop("md5", None) changed = replace(changed, raw=raw) http.calls.clear() fetched = connector.fetch(changed, dest=tmp_path, channels=("ir_105",)) From d98dcebf4255cc63f62163d9d8e825ba95b75500 Mon Sep 17 00:00:00 2001 From: Richard van Schaik Date: Sun, 25 Oct 2026 08:04:44 +0100 Subject: [PATCH 5/5] fix(fci): order native rows by measured coordinates Provider body enumeration is unordered. Sort all row-aligned values, flags and pixel times together. Keep native-coordinate gaps. Parameterize the native parity test over reversed entry order. Co-Authored-By: Claude Opus 5.5 --- src/spectraccess/connectors/fci_eumetsat/connector.py | 2 +- tests/test_fci_eumetsat.py | 9 +++++++-- 2 files changed, 8 insertions(+), 3 deletions(-) diff --git a/src/spectraccess/connectors/fci_eumetsat/connector.py b/src/spectraccess/connectors/fci_eumetsat/connector.py index 50239ea..beb7ce2 100644 --- a/src/spectraccess/connectors/fci_eumetsat/connector.py +++ b/src/spectraccess/connectors/fci_eumetsat/connector.py @@ -702,7 +702,7 @@ def _read_l1(result, bbox, channels): output = xr.Dataset() for channel, chunks in pieces.items(): if chunks: - merged = xr.concat(chunks, dim=f"y_{channel}", join="exact", combine_attrs="drop_conflicts") + merged = xr.concat(chunks, dim=f"y_{channel}", join="exact", combine_attrs="drop_conflicts").sortby(f"y_{channel}") # Concatenation contains only existing segment rows; gaps are never padded. if len(np.unique(merged[f"y_{channel}"])) != merged.sizes[f"y_{channel}"]: raise ValueError("overlapping FCI body rows in repeat cycle") diff --git a/tests/test_fci_eumetsat.py b/tests/test_fci_eumetsat.py index 641821c..dddafe7 100644 --- a/tests/test_fci_eumetsat.py +++ b/tests/test_fci_eumetsat.py @@ -156,10 +156,15 @@ def test_spike_real_satpy(fixture): mod._close_l1(clipped) -def test_native_values_time_flags_noise_and_dimensions(fixture): +@pytest.mark.parametrize('reverse_entries', [False, True]) +def test_native_values_time_flags_noise_and_dimensions(fixture, reverse_entries): + native = result(fixture) + if reverse_entries: + native = result(fixture, names=native.entries[::-1]) with mod.satpy.config.set({"readers.clip_negative_radiances": True}): - ds = FCIConnector().read(result(fixture), channels=CHANNELS) + ds = FCIConnector().read(native, channels=CHANNELS) assert ds.ir_105.shape == (8, 4) + assert np.all(np.diff(ds.y_ir_105) > 0) assert ds.nir_13.shape == (16, 8) assert ds.ir_105.values[0, 0] == -5 assert np.isnan(ds.ir_105.values[3, 0])