Provenance and Reproducibility

See Reproducibility for how these fit together. agribound._cache and agribound._repro are internal modules; their functions are documented because the cache keys and seeding define what a run reuses.

Provenance records

provenance

Run provenance: what was run, with which inputs, versions and resources.

:class:RunRecorder collects a JSON-serialisable record of one pipeline run (configuration and its hash, seed, package versions, platform, device, step timings, peak memory, engine metadata, facts and warnings). The pipeline writes it next to the output as <output_path>.provenance.json (:func:provenance_path) and uses :func:reuse_mismatch (the :func:config_hash, the study-area fingerprint and the results versions of :mod:agribound._results) to decide whether an existing output can be reused.

RunRecorder

Collect provenance for one pipeline run.

Use as a context manager; the block's wall time and outcome ("success" or "failed" with the error) are recorded on exit. While the block runs, every WARNING (or higher) logged by an agribound logger (agribound.*, in any thread of the process) is added to the record's warnings as well, e.g. an engine's note that the input resolution is outside its training range. Identical messages are kept once, and at most :data:MAX_RECORDED_WARNINGS are stored (warnings_not_recorded counts the rest).

Parameters:

Name	Type	Description	Default
config	AgriboundConfig	Configuration of the run. Its hash and dictionary are captured when the recorder is created, before any stage can modify it.	required
run_id	str or None	Run identifier; a new one from :func:agribound._repro.new_run_id is generated when None.	None

Examples:

>>> with RunRecorder(config) as rec:
...     with rec.step("composite"):
...         raster = build_composite(config)
...     rec.set("raster_path", raster)
>>> write_provenance(config.output_path, rec.to_dict())

Source code in agribound/provenance.py

class RunRecorder:
    """Collect provenance for one pipeline run.

    Use as a context manager; the block's wall time and outcome
    (``"success"`` or ``"failed"`` with the error) are recorded on exit.
    While the block runs, every WARNING (or higher) logged by an
    ``agribound`` logger (``agribound.*``, in any thread of the process) is
    added to the record's ``warnings`` as well, e.g. an engine's note that
    the input resolution is outside its training range. Identical messages
    are kept once, and at most :data:`MAX_RECORDED_WARNINGS` are stored
    (``warnings_not_recorded`` counts the rest).

    Parameters
    ----------
    config : AgriboundConfig
        Configuration of the run. Its hash and dictionary are captured when
        the recorder is created, before any stage can modify it.
    run_id : str or None
        Run identifier; a new one from :func:`agribound._repro.new_run_id` is
        generated when *None*.

    Examples
    --------
    >>> with RunRecorder(config) as rec:
    ...     with rec.step("composite"):
    ...         raster = build_composite(config)
    ...     rec.set("raster_path", raster)
    >>> write_provenance(config.output_path, rec.to_dict())
    """

    def __init__(self, config: Any, run_id: str | None = None) -> None:
        from agribound._repro import new_run_id

        self.config = config
        self.run_id = run_id or new_run_id()
        self.config_dict = to_jsonable(config.to_dict())
        self.config_hash = config_hash(config)
        self.steps: list[dict[str, Any]] = []
        self.facts: dict[str, Any] = {}
        self.warnings: list[str] = []
        self.warnings_not_recorded = 0
        self._warning_handler: _WarningCollector | None = None
        self.engine_meta: dict[str, Any] = {}
        self.status = "created"
        self.error: str | None = None
        self.started_utc: str | None = None
        self.finished_utc: str | None = None
        self._t0: float | None = None
        self._wall_s: float | None = None
        # Resource figures frozen when the run finishes (live values before that).
        self._peak_rss_mb: float | None = None
        self._torch_mem_mb: float | None = None
        self._versions: dict[str, str] | None = None

    # Context manager -----------------------------------------------------

    def __enter__(self) -> RunRecorder:
        self.started_utc = _utc_now()
        self._t0 = time.perf_counter()
        self.status = "running"
        if self._warning_handler is None:
            self._warning_handler = _WarningCollector(self)
            logging.getLogger("agribound").addHandler(self._warning_handler)
        return self

    def __exit__(self, exc_type, exc, tb) -> bool:
        if self._warning_handler is not None:
            logging.getLogger("agribound").removeHandler(self._warning_handler)
            self._warning_handler = None
        self.finished_utc = _utc_now()
        if self._t0 is not None:
            self._wall_s = round(time.perf_counter() - self._t0, 3)
        self._peak_rss_mb = _peak_rss_mb()
        self._torch_mem_mb = _torch_max_memory_mb()
        if exc is None:
            self.status = "success"
        else:
            self.status = "failed"
            self.error = f"{exc_type.__name__}: {exc}"
        return False

    # Recording -----------------------------------------------------------

    @contextlib.contextmanager
    def step(self, name: str) -> Iterator[dict[str, Any]]:
        """Time a pipeline step.

        Yields the step's record (a dict) so callers can attach details.
        A failing step is recorded with ``status="failed"`` and the error is
        re-raised.
        """
        record: dict[str, Any] = {"name": name, "started_utc": _utc_now(), "status": "running"}
        self.steps.append(record)
        t0 = time.perf_counter()
        try:
            yield record
        except BaseException as exc:
            record["status"] = "failed"
            record["error"] = f"{type(exc).__name__}: {exc}"
            raise
        else:
            record["status"] = "success"
        finally:
            record["wall_s"] = round(time.perf_counter() - t0, 3)

    def set(self, key: str, value: Any) -> None:
        """Record a fact (counts, backend, weights, dataset, ...)."""
        self.facts[str(key)] = to_jsonable(value)

    def add_warning(self, msg: str) -> None:
        """Record a warning (a message already recorded is not added again).

        Inside the ``with`` block, WARNING records of the ``agribound``
        loggers are recorded automatically, so a caller that also logs the
        message does not create a duplicate.
        """
        text = str(msg)
        if text in self.warnings:
            return
        if len(self.warnings) >= MAX_RECORDED_WARNINGS:
            self.warnings_not_recorded += 1
            return
        self.warnings.append(text)

    def record_engine_meta(self, meta: dict) -> None:
        """Merge engine metadata (``gdf.attrs["engine_meta"]``) into the record."""
        if not meta:
            return
        if not isinstance(meta, dict):
            meta = {"value": meta}
        self.engine_meta.update(to_jsonable(meta))

    # Output --------------------------------------------------------------

    def _device(self) -> str | None:
        try:
            return self.config.resolve_device()
        except Exception:  # pragma: no cover
            return None

    def to_dict(self) -> dict[str, Any]:
        """Return the provenance record as a JSON-serialisable dictionary."""
        from agribound._repro import collect_versions
        from agribound._version import __version__

        wall_s = self._wall_s
        if wall_s is None and self._t0 is not None:
            wall_s = round(time.perf_counter() - self._t0, 3)
        finished = self.finished_utc is not None
        if self._versions is None:
            self._versions = collect_versions()
        env = {k: os.environ[k] for k in _ENV_KEYS if k in os.environ}
        record = {
            "schema_version": PROVENANCE_SCHEMA_VERSION,
            "agribound_version": __version__,
            "run_id": self.run_id,
            "status": self.status,
            "error": self.error,
            "config_hash": self.config_hash,
            "seed": getattr(self.config, "seed", None),
            "config": self.config_dict,
            "versions": dict(self._versions),
            "platform": platform.platform(),
            "machine": platform.machine(),
            "hostname": socket.gethostname(),
            "python": sys.version,
            "device": self._device(),
            "started_utc": self.started_utc,
            "finished_utc": self.finished_utc,
            "wall_s": wall_s,
            "peak_rss_mb": self._peak_rss_mb if finished else _peak_rss_mb(),
            "torch_max_memory_mb": self._torch_mem_mb if finished else _torch_max_memory_mb(),
            "steps": list(self.steps),
            "facts": dict(self.facts),
            "warnings": list(self.warnings),
            "warnings_not_recorded": self.warnings_not_recorded,
            "engine_meta": dict(self.engine_meta),
            "gee_workload_tag": getattr(self.config, "gee_workload_tag", None),
            "git": _git_info(),
            "environment": env,
        }
        return to_jsonable(record)

step

step(name: str) -> Iterator[dict[str, Any]]

Time a pipeline step.

Yields the step's record (a dict) so callers can attach details. A failing step is recorded with status="failed" and the error is re-raised.

Source code in agribound/provenance.py

@contextlib.contextmanager
def step(self, name: str) -> Iterator[dict[str, Any]]:
    """Time a pipeline step.

    Yields the step's record (a dict) so callers can attach details.
    A failing step is recorded with ``status="failed"`` and the error is
    re-raised.
    """
    record: dict[str, Any] = {"name": name, "started_utc": _utc_now(), "status": "running"}
    self.steps.append(record)
    t0 = time.perf_counter()
    try:
        yield record
    except BaseException as exc:
        record["status"] = "failed"
        record["error"] = f"{type(exc).__name__}: {exc}"
        raise
    else:
        record["status"] = "success"
    finally:
        record["wall_s"] = round(time.perf_counter() - t0, 3)

set

set(key: str, value: Any) -> None

Record a fact (counts, backend, weights, dataset, ...).

Source code in agribound/provenance.py

def set(self, key: str, value: Any) -> None:
    """Record a fact (counts, backend, weights, dataset, ...)."""
    self.facts[str(key)] = to_jsonable(value)

add_warning

add_warning(msg: str) -> None

Record a warning (a message already recorded is not added again).

Inside the with block, WARNING records of the agribound loggers are recorded automatically, so a caller that also logs the message does not create a duplicate.

Source code in agribound/provenance.py

def add_warning(self, msg: str) -> None:
    """Record a warning (a message already recorded is not added again).

    Inside the ``with`` block, WARNING records of the ``agribound``
    loggers are recorded automatically, so a caller that also logs the
    message does not create a duplicate.
    """
    text = str(msg)
    if text in self.warnings:
        return
    if len(self.warnings) >= MAX_RECORDED_WARNINGS:
        self.warnings_not_recorded += 1
        return
    self.warnings.append(text)

record_engine_meta

record_engine_meta(meta: dict) -> None

Merge engine metadata (gdf.attrs["engine_meta"]) into the record.

Source code in agribound/provenance.py

def record_engine_meta(self, meta: dict) -> None:
    """Merge engine metadata (``gdf.attrs["engine_meta"]``) into the record."""
    if not meta:
        return
    if not isinstance(meta, dict):
        meta = {"value": meta}
    self.engine_meta.update(to_jsonable(meta))

to_dict

to_dict() -> dict[str, Any]

Return the provenance record as a JSON-serialisable dictionary.

Source code in agribound/provenance.py

def to_dict(self) -> dict[str, Any]:
    """Return the provenance record as a JSON-serialisable dictionary."""
    from agribound._repro import collect_versions
    from agribound._version import __version__

    wall_s = self._wall_s
    if wall_s is None and self._t0 is not None:
        wall_s = round(time.perf_counter() - self._t0, 3)
    finished = self.finished_utc is not None
    if self._versions is None:
        self._versions = collect_versions()
    env = {k: os.environ[k] for k in _ENV_KEYS if k in os.environ}
    record = {
        "schema_version": PROVENANCE_SCHEMA_VERSION,
        "agribound_version": __version__,
        "run_id": self.run_id,
        "status": self.status,
        "error": self.error,
        "config_hash": self.config_hash,
        "seed": getattr(self.config, "seed", None),
        "config": self.config_dict,
        "versions": dict(self._versions),
        "platform": platform.platform(),
        "machine": platform.machine(),
        "hostname": socket.gethostname(),
        "python": sys.version,
        "device": self._device(),
        "started_utc": self.started_utc,
        "finished_utc": self.finished_utc,
        "wall_s": wall_s,
        "peak_rss_mb": self._peak_rss_mb if finished else _peak_rss_mb(),
        "torch_max_memory_mb": self._torch_mem_mb if finished else _torch_max_memory_mb(),
        "steps": list(self.steps),
        "facts": dict(self.facts),
        "warnings": list(self.warnings),
        "warnings_not_recorded": self.warnings_not_recorded,
        "engine_meta": dict(self.engine_meta),
        "gee_workload_tag": getattr(self.config, "gee_workload_tag", None),
        "git": _git_info(),
        "environment": env,
    }
    return to_jsonable(record)

to_jsonable

to_jsonable(value: Any) -> Any

Recursively convert value to JSON-serialisable builtins.

NumPy scalars/arrays, tuples, sets, paths, datetimes and objects with to_dict/isoformat are converted; non-finite floats become None; anything else becomes str(value).

Source code in agribound/provenance.py

def to_jsonable(value: Any) -> Any:
    """Recursively convert *value* to JSON-serialisable builtins.

    NumPy scalars/arrays, tuples, sets, paths, datetimes and objects with
    ``to_dict``/``isoformat`` are converted; non-finite floats become
    *None*; anything else becomes ``str(value)``.
    """
    if value is None or isinstance(value, bool | int | str):
        return value
    if isinstance(value, float):
        return value if math.isfinite(value) else None
    if isinstance(value, dict):
        return {str(k): to_jsonable(v) for k, v in value.items()}
    if isinstance(value, list | tuple):
        return [to_jsonable(v) for v in value]
    if isinstance(value, set | frozenset):
        return sorted((to_jsonable(v) for v in value), key=str)
    if isinstance(value, Path):
        return str(value)
    if isinstance(value, _dt.datetime | _dt.date):
        return value.isoformat()
    try:
        import numpy as np

        if isinstance(value, np.generic):
            return to_jsonable(value.item())
        if isinstance(value, np.ndarray):
            return to_jsonable(value.tolist())
    except ImportError:  # pragma: no cover - numpy is a core dependency
        pass
    if hasattr(value, "to_dict") and callable(value.to_dict):
        try:
            return to_jsonable(value.to_dict())
        except Exception:
            pass
    return str(value)

canonical_config

canonical_config(config: Any) -> dict[str, Any]

Return the configuration fields that define the result (see :data:HASH_EXCLUDED_FIELDS).

Source code in agribound/provenance.py

def canonical_config(config: Any) -> dict[str, Any]:
    """Return the configuration fields that define the result (see :data:`HASH_EXCLUDED_FIELDS`)."""
    data = config.to_dict() if hasattr(config, "to_dict") else dict(config)
    return {k: to_jsonable(v) for k, v in sorted(data.items()) if k not in HASH_EXCLUDED_FIELDS}

config_hash

config_hash(config: Any) -> str

Return the SHA-1 hex digest of the canonical YAML of config.

Fields listed in :data:HASH_EXCLUDED_FIELDS (output location and format, caching/provenance switches, credentials, request tuning and execution resources) are excluded, so configurations that differ only in these fields hash identically. This includes device, which can change the polygons slightly (see :data:HASH_EXCLUDED_FIELDS).

Parameters:

Name	Type	Description	Default
config	AgriboundConfig or dict	Configuration.	required

Returns:

Type	Description
str	40-character hexadecimal digest.

Source code in agribound/provenance.py

def config_hash(config: Any) -> str:
    """Return the SHA-1 hex digest of the canonical YAML of *config*.

    Fields listed in :data:`HASH_EXCLUDED_FIELDS` (output location and format,
    caching/provenance switches, credentials, request tuning and execution
    resources) are excluded, so configurations that differ only in these
    fields hash identically. This includes ``device``, which can change the
    polygons slightly (see :data:`HASH_EXCLUDED_FIELDS`).

    Parameters
    ----------
    config : AgriboundConfig or dict
        Configuration.

    Returns
    -------
    str
        40-character hexadecimal digest.
    """
    text = yaml.safe_dump(canonical_config(config), sort_keys=True, default_flow_style=False)
    return hashlib.sha1(text.encode("utf-8")).hexdigest()

provenance_path

provenance_path(output_path: str | Path) -> Path

Return the provenance sidecar path f"{output_path}.provenance.json".

Source code in agribound/provenance.py

def provenance_path(output_path: str | Path) -> Path:
    """Return the provenance sidecar path ``f"{output_path}.provenance.json"``."""
    return Path(f"{output_path}.provenance.json")

write_provenance

write_provenance(output_path: str | Path, record: dict) -> Path

Write record as JSON next to output_path (atomic replace).

Parameters:

Name	Type	Description	Default
output_path	str or Path	Output vector path the record describes.	required
record	dict	Provenance record (converted with :func:to_jsonable).	required

Returns:

Type	Description
Path	Path of the written sidecar.

Source code in agribound/provenance.py

def write_provenance(output_path: str | Path, record: dict) -> Path:
    """Write *record* as JSON next to *output_path* (atomic replace).

    Parameters
    ----------
    output_path : str or Path
        Output vector path the record describes.
    record : dict
        Provenance record (converted with :func:`to_jsonable`).

    Returns
    -------
    pathlib.Path
        Path of the written sidecar.
    """
    path = provenance_path(output_path)
    path.parent.mkdir(parents=True, exist_ok=True)
    tmp = path.with_name(f".{path.name}.{os.getpid()}.tmp")
    with open(tmp, "w") as f:
        json.dump(to_jsonable(record), f, indent=2, sort_keys=False)
        f.write("\n")
    os.replace(tmp, path)
    return path

read_provenance

read_provenance(output_path: str | Path) -> dict | None

Read the provenance sidecar of output_path.

Returns:

Type	Description
dict or None	The record, or None if the sidecar is missing or not valid JSON.

Source code in agribound/provenance.py

def read_provenance(output_path: str | Path) -> dict | None:
    """Read the provenance sidecar of *output_path*.

    Returns
    -------
    dict or None
        The record, or *None* if the sidecar is missing or not valid JSON.
    """
    path = provenance_path(output_path)
    if not path.exists():
        return None
    try:
        with open(path) as f:
            data = json.load(f)
    except (OSError, json.JSONDecodeError) as exc:
        logger.warning("Could not read provenance file %s: %s", path, exc)
        return None
    return data if isinstance(data, dict) else None

reuse_facts

reuse_facts(config: Any) -> dict[str, Any]

Return the facts that :func:reuse_mismatch checks besides :func:config_hash.

aoi_fingerprint is :func:agribound._cache.aoi_fingerprint: the study-area geometry (read from the file, bbox: string or WKT), the ID string of a GEE asset (so no Earth Engine access is needed; the asset's features are not covered), or the local raster's resolved path, size and modification time when there is no study area. It is left out when the study area cannot be read (the composite stage then reports the error). results_versions is :func:agribound._results.results_versions.

Parameters:

Name	Type	Description	Default
config	AgriboundConfig	Configuration of the run.	required

Returns:

Type	Description
dict	{"aoi_fingerprint": str, "results_versions": dict}.

Source code in agribound/provenance.py

def reuse_facts(config: Any) -> dict[str, Any]:
    """Return the facts that :func:`reuse_mismatch` checks besides :func:`config_hash`.

    ``aoi_fingerprint`` is :func:`agribound._cache.aoi_fingerprint`: the
    study-area geometry (read from the file, ``bbox:`` string or WKT), the ID
    string of a GEE asset (so no Earth Engine access is needed; the asset's
    features are not covered), or the local raster's resolved path, size and
    modification time when there is no study area. It is left out when the
    study area cannot be read (the composite stage then reports the error).
    ``results_versions`` is :func:`agribound._results.results_versions`.

    Parameters
    ----------
    config : AgriboundConfig
        Configuration of the run.

    Returns
    -------
    dict
        ``{"aoi_fingerprint": str, "results_versions": dict}``.
    """
    from agribound._cache import aoi_fingerprint
    from agribound._results import results_versions

    facts: dict[str, Any] = {}
    try:
        facts["aoi_fingerprint"] = aoi_fingerprint(config)
    except Exception as exc:
        logger.debug("No study-area fingerprint for %r: %s", config.study_area, exc)
    facts["results_versions"] = results_versions(config)
    return facts

reuse_mismatch

reuse_mismatch(record: dict, config: Any, output: str | Path | None = None) -> str | None

Return why the output that record describes cannot be reused for config.

The output of a successful run is reused only when:

• its config_hash equals :func:config_hash of config;
• its facts["results_versions"] equal :func:agribound._results.results_versions of config (a missing fact or component counts as version 1, i.e. agribound <= 1.0.0), so an output of a component whose results have changed since is not reused;
• its facts["aoi_fingerprint"] equals :func:agribound._cache.aoi_fingerprint of config when the study area is a file (a file whose geometry changed at the same path is not reused) or, without a study area, for the local raster (compared by resolved path, size and modification time). A bbox:, WKT or GEE asset study area is covered by the configuration hash already.

A record without aoi_fingerprint (agribound <= 1.0.0) is accepted, with a WARNING that the study-area file (or local raster) could not be verified; so is a record whose fingerprint cannot be compared because the file cannot be read now.

Parameters:

Name	Type	Description	Default
record	dict	Provenance record of a successful run (:func:read_provenance).	required
config	AgriboundConfig	Configuration of the new run.	required
output	(str, Path or None)	The existing output, named in the warnings.	None

Returns:

Type	Description
str or None	None when the output can be reused, else the reason as a clause ("it was produced ...") for an error message.

Source code in agribound/provenance.py

def reuse_mismatch(record: dict, config: Any, output: str | Path | None = None) -> str | None:
    """Return why the output that *record* describes cannot be reused for *config*.

    The output of a successful run is reused only when:

    - its ``config_hash`` equals :func:`config_hash` of *config*;
    - its ``facts["results_versions"]`` equal
      :func:`agribound._results.results_versions` of *config* (a missing fact
      or component counts as version 1, i.e. agribound <= 1.0.0), so an
      output of a component whose results have changed since is not reused;
    - its ``facts["aoi_fingerprint"]`` equals
      :func:`agribound._cache.aoi_fingerprint` of *config* when the study
      area is a file (a file whose geometry changed at the same path is not
      reused) or, without a study area, for the local raster (compared by
      resolved path, size and modification time). A ``bbox:``, WKT or GEE
      asset study area is covered by the configuration hash already.

    A record without ``aoi_fingerprint`` (agribound <= 1.0.0) is accepted,
    with a WARNING that the study-area file (or local raster) could not be
    verified; so is a record whose fingerprint cannot be compared because
    the file cannot be read now.

    Parameters
    ----------
    record : dict
        Provenance record of a successful run (:func:`read_provenance`).
    config : AgriboundConfig
        Configuration of the new run.
    output : str, Path or None
        The existing output, named in the warnings.

    Returns
    -------
    str or None
        *None* when the output can be reused, else the reason as a clause
        (``"it was produced ..."``) for an error message.
    """
    from agribound._cache import aoi_fingerprint
    from agribound._results import results_versions
    from agribound._version import __version__

    current_hash = config_hash(config)
    if record.get("config_hash") != current_hash:
        return (
            f"it was produced with a different configuration (config_hash "
            f"{str(record.get('config_hash'))[:12]} != {current_hash[:12]})"
        )

    facts = record.get("facts") or {}
    version = record.get("agribound_version")
    made_by = f"agribound {version}" if version else "an unknown agribound version"
    target = "the existing output" + (f" {str(output)!r}" if output is not None else "")
    reasons: list[str] = []

    current_versions = results_versions(config)
    recorded_versions = facts.get("results_versions")
    if not isinstance(recorded_versions, dict):
        recorded_versions = {}
    changed = {
        name: (recorded_versions.get(name, 1), current)
        for name, current in current_versions.items()
        if recorded_versions.get(name, 1) != current
    }
    if changed:
        changes = ", ".join(f"{name} {old} -> {new}" for name, (old, new) in changed.items())
        reasons.append(
            f"it was produced by {made_by}, whose results for this configuration differ from "
            f"those of agribound {__version__} (results versions of agribound._results: "
            f"{changes})"
        )

    checked = _fingerprinted_input(config)
    recorded_aoi = facts.get("aoi_fingerprint")
    if checked is not None and recorded_aoi is None:
        if not reasons:
            logger.warning(
                "The provenance record of %s (written by %s) has no study-area fingerprint, so "
                "it cannot be verified that the output was made from the current %s; reusing "
                "it. Pass overwrite=True (CLI: --overwrite) if the file has changed since that "
                "run.",
                target,
                made_by,
                checked,
            )
    elif checked is not None:
        study_area = str(getattr(config, "study_area", "") or "").strip()
        try:
            if not study_area and not Path(config.local_tif_path).expanduser().exists():
                raise FileNotFoundError(f"{config.local_tif_path} does not exist")
            current_aoi = aoi_fingerprint(config)
        except Exception as exc:
            current_aoi = None
            if not reasons:
                logger.warning(
                    "Could not read the %s to check that %s was made from it (%s: %s); "
                    "reusing the output without that check",
                    checked,
                    target,
                    type(exc).__name__,
                    exc,
                )
        if current_aoi is not None and current_aoi != recorded_aoi:
            if study_area:
                what = "for a different study area: the geometry in the"
                how = "study-area fingerprint"
            else:
                what = "from a different local raster: the path, size or modification time of the"
                how = "fingerprint"
            reasons.append(
                f"it was produced {what} {checked} differs from the one recorded for that run "
                f"({how} {recorded_aoi} -> {current_aoi})"
            )
    return " and ".join(reasons) or None

Cache keys

_cache

Content-addressed cache keys for intermediate files.

Every module that writes an intermediate artefact (composites, window composites, embeddings, engine inputs/outputs, fine-tuning data, LULC rasters) names it with :func:cache_path, so that runs over different study areas, years, date ranges or compositing settings never reuse each other's files, even when they share one cache directory.

The key is a 12-character SHA-1 prefix over :data:CACHE_SCHEMA_VERSION, a fingerprint of the study area (:func:aoi_fingerprint), the source, year and date range, the compositing and export settings, source-specific options and any extra parts supplied by the caller (for example a model name).

CACHE_SCHEMA_VERSION module-attribute

CACHE_SCHEMA_VERSION: str = '2'

Bump when radiometry or export semantics change, to invalidate old caches.

gee_asset_fingerprint

gee_asset_fingerprint(asset_id: str) -> str

Return the 12-hex fingerprint of a GEE asset ID, as used by :func:aoi_fingerprint.

The asset is not read: the fingerprint depends on the ID string only.

Source code in agribound/_cache.py

def gee_asset_fingerprint(asset_id: str) -> str:
    """Return the 12-hex fingerprint of a GEE asset ID, as used by :func:`aoi_fingerprint`.

    The asset is not read: the fingerprint depends on the ID string only.
    """
    return _sha1(f"gee-asset:{asset_id}")[:_KEY_LENGTH]

aoi_fingerprint

aoi_fingerprint(config: Any) -> str

Return a 12-hex SHA-1 fingerprint of the configured study area.

• GEE asset IDs are fingerprinted by the asset ID string (the asset is not downloaded; :func:gee_asset_fingerprint). Cached files therefore do not change when the asset's features change under the same ID; the local copy of the asset kept by :func:agribound.io.vector.read_config_study_area has the same key.
• Files, "bbox:..." strings and WKT are read, reprojected to EPSG:4326, unioned, snapped to a 1e-7 degree grid, normalised, and the 2-D WKB is hashed. Results are memoised per path and modification time.
• An empty study area (source="local" without clipping) is fingerprinted by the local raster's resolved path, size and modification time.

Parameters:

Name	Type	Description	Default
config	AgriboundConfig	Configuration providing study_area (and local_tif_path).	required

Returns:

Type	Description
str	12 hexadecimal characters.

Source code in agribound/_cache.py

def aoi_fingerprint(config: Any) -> str:
    """Return a 12-hex SHA-1 fingerprint of the configured study area.

    - GEE asset IDs are fingerprinted by the asset ID string (the asset is not
      downloaded; :func:`gee_asset_fingerprint`). Cached files therefore do
      not change when the asset's features change under the same ID; the
      local copy of the asset kept by
      :func:`agribound.io.vector.read_config_study_area` has the same key.
    - Files, ``"bbox:..."`` strings and WKT are read, reprojected to
      EPSG:4326, unioned, snapped to a 1e-7 degree grid, normalised, and the
      2-D WKB is hashed. Results are memoised per path and modification time.
    - An empty study area (``source="local"`` without clipping) is
      fingerprinted by the local raster's resolved path, size and
      modification time.

    Parameters
    ----------
    config : AgriboundConfig
        Configuration providing ``study_area`` (and ``local_tif_path``).

    Returns
    -------
    str
        12 hexadecimal characters.
    """
    study_area = str(getattr(config, "study_area", "") or "").strip()

    if not study_area:
        tif = getattr(config, "local_tif_path", None)
        if tif:
            resolved = str(Path(tif).expanduser().resolve())
            mtime, size = _file_signature(resolved)
            return _sha1(f"local-raster:{resolved}:{size}:{mtime}")[:_KEY_LENGTH]
        return _sha1("no-study-area")[:_KEY_LENGTH]

    if _is_gee_asset(study_area):
        return gee_asset_fingerprint(study_area)

    is_path = not study_area.lower().startswith("bbox:") and _path_exists(study_area)
    if is_path:
        resolved = str(Path(study_area).resolve())
        mtime, size = _file_signature(resolved)
        memo_key = (resolved, mtime, size)
    else:
        memo_key = (study_area, None, None)
    cached = _AOI_FINGERPRINT_CACHE.get(memo_key)
    if cached is not None:
        return cached

    import shapely

    from agribound.io.vector import read_study_area

    gdf = read_study_area(study_area)
    if gdf.crs is None:
        logger.warning("Study area %s has no CRS; assuming EPSG:4326", study_area)
        gdf = gdf.set_crs("EPSG:4326")
    elif not gdf.crs.equals("EPSG:4326"):
        gdf = gdf.to_crs("EPSG:4326")
    geom = gdf.geometry.union_all()
    geom = shapely.set_precision(geom, _COORD_PRECISION)
    geom = shapely.normalize(geom)
    wkb = shapely.to_wkb(geom, output_dimension=2, byte_order=1, include_srid=False)
    fingerprint = hashlib.sha1(wkb).hexdigest()[:_KEY_LENGTH]
    _AOI_FINGERPRINT_CACHE[memo_key] = fingerprint
    return fingerprint

cache_key

cache_key(config: Any, *parts: object, include_temporal: bool = True) -> str

Return a 12-hex cache key for config and optional extra parts.

Parameters:

Name	Type	Description	Default
config	AgriboundConfig	Pipeline configuration.	required
*parts	object	Additional values (model names, window labels, parameters) that distinguish the artefact. They are hashed as str(part) in order.	()
include_temporal	bool	When False, year and date_range are left out, for artefacts that do not depend on time.	True

Returns:

Type	Description
str	12 hexadecimal characters.

Notes

Hashed fields: :data:CACHE_SCHEMA_VERSION, :func:aoi_fingerprint, source, year and date_range (if include_temporal), composite_method, cloud_cover_max, export_crs, s2_cloud_mask, naip_resolution_m; cloud_score_threshold when Cloud Score+ masking is selected; tessera_version/tessera_variant for embedding sources; google_embedding_backend for Google embeddings; the USGS service URL, state and year-fallback flag for USGS NAIP Plus; and the local raster's path, size and modification time for local sources.

Source code in agribound/_cache.py

def cache_key(config: Any, *parts: object, include_temporal: bool = True) -> str:
    """Return a 12-hex cache key for *config* and optional extra *parts*.

    Parameters
    ----------
    config : AgriboundConfig
        Pipeline configuration.
    *parts : object
        Additional values (model names, window labels, parameters) that
        distinguish the artefact. They are hashed as ``str(part)`` in order.
    include_temporal : bool
        When *False*, ``year`` and ``date_range`` are left out, for artefacts
        that do not depend on time.

    Returns
    -------
    str
        12 hexadecimal characters.

    Notes
    -----
    Hashed fields: :data:`CACHE_SCHEMA_VERSION`, :func:`aoi_fingerprint`,
    ``source``, ``year`` and ``date_range`` (if *include_temporal*),
    ``composite_method``, ``cloud_cover_max``, ``export_crs``,
    ``s2_cloud_mask``, ``naip_resolution_m``; ``cloud_score_threshold`` when
    Cloud Score+ masking is selected; ``tessera_version``/``tessera_variant``
    for embedding sources; ``google_embedding_backend`` for Google embeddings;
    the USGS service URL, state and year-fallback flag for USGS NAIP Plus; and
    the local raster's path, size and modification time for local sources.
    """
    payload = {
        "fields": [
            [name, _canonical(value)] for name, value in _key_fields(config, include_temporal)
        ],
        "parts": [str(p) for p in parts],
    }
    text = json.dumps(payload, sort_keys=True, separators=(",", ":"))
    return _sha1(text)[:_KEY_LENGTH]

cache_path

cache_path(config: Any, stem: str, suffix: str, *parts: object, include_temporal: bool = True) -> Path

Return config.get_working_dir() / f"{stem}_{key}{suffix}".

Parameters:

Name	Type	Description	Default
config	AgriboundConfig	Pipeline configuration.	required
stem	str	Human-readable file name prefix (may contain sub-directories).	required
suffix	str	File suffix including the dot (e.g. ".tif"), or "" for a directory name.	required
*parts	object	Extra key parts (see :func:cache_key).	()
include_temporal	bool	See :func:cache_key.	True

Returns:

Type	Description
Path	Path inside the working directory; its parent directory exists.

Source code in agribound/_cache.py

def cache_path(
    config: Any,
    stem: str,
    suffix: str,
    *parts: object,
    include_temporal: bool = True,
) -> Path:
    """Return ``config.get_working_dir() / f"{stem}_{key}{suffix}"``.

    Parameters
    ----------
    config : AgriboundConfig
        Pipeline configuration.
    stem : str
        Human-readable file name prefix (may contain sub-directories).
    suffix : str
        File suffix including the dot (e.g. ``".tif"``), or ``""`` for a
        directory name.
    *parts : object
        Extra key parts (see :func:`cache_key`).
    include_temporal : bool
        See :func:`cache_key`.

    Returns
    -------
    pathlib.Path
        Path inside the working directory; its parent directory exists.
    """
    key = cache_key(config, *parts, include_temporal=include_temporal)
    path = Path(config.get_working_dir()) / f"{stem}_{key}{suffix}"
    path.parent.mkdir(parents=True, exist_ok=True)
    return path

clear_fingerprint_cache

clear_fingerprint_cache() -> None

Forget memoised study-area fingerprints (mainly for tests).

Source code in agribound/_cache.py

def clear_fingerprint_cache() -> None:
    """Forget memoised study-area fingerprints (mainly for tests)."""
    _AOI_FINGERPRINT_CACHE.clear()

Seeding and versions

_repro

Reproducibility helpers: seeding, seeded generators, version capture, run IDs.

seed_everything

seed_everything(seed: int, deterministic: bool = False) -> None

Seed Python, NumPy, torch (CPU, CUDA, MPS) and Lightning.

Also sets PYTHONHASHSEED (inherited by subprocesses; it cannot change string hashing of the running interpreter). torch and Lightning are seeded only if they are installed.

Parameters:

Name	Type	Description	Default
seed	int	Seed in [0, 2**32 - 1].	required
deterministic	bool	Also request deterministic torch kernels (torch.use_deterministic_algorithms(True, warn_only=True), cuDNN deterministic mode, CUBLAS_WORKSPACE_CONFIG). This can slow down training and inference.	False

Raises:

Type	Description
ValueError	If seed is out of range.

Source code in agribound/_repro.py

def seed_everything(seed: int, deterministic: bool = False) -> None:
    """Seed Python, NumPy, torch (CPU, CUDA, MPS) and Lightning.

    Also sets ``PYTHONHASHSEED`` (inherited by subprocesses; it cannot change
    string hashing of the running interpreter). torch and Lightning are
    seeded only if they are installed.

    Parameters
    ----------
    seed : int
        Seed in ``[0, 2**32 - 1]``.
    deterministic : bool
        Also request deterministic torch kernels
        (``torch.use_deterministic_algorithms(True, warn_only=True)``,
        cuDNN deterministic mode, ``CUBLAS_WORKSPACE_CONFIG``). This can slow
        down training and inference.

    Raises
    ------
    ValueError
        If *seed* is out of range.
    """
    seed = int(seed)
    if not 0 <= seed <= _MAX_SEED:
        raise ValueError(f"seed must be in [0, {_MAX_SEED}], got {seed}")

    os.environ["PYTHONHASHSEED"] = str(seed)
    random.seed(seed)
    np.random.seed(seed)

    if importlib.util.find_spec("torch") is not None:
        try:
            import torch

            # Seeds the CPU generator and every CUDA, MPS and XPU device.
            torch.manual_seed(seed)
            if deterministic:
                os.environ.setdefault("CUBLAS_WORKSPACE_CONFIG", ":4096:8")
                torch.use_deterministic_algorithms(True, warn_only=True)
                torch.backends.cudnn.deterministic = True
                torch.backends.cudnn.benchmark = False
        except Exception as exc:  # pragma: no cover - broken torch install
            logger.warning("Could not seed torch: %s", exc)

    _seed_lightning(seed)

get_rng

get_rng(config: Any, *salt: object) -> np.random.Generator

Return a NumPy generator seeded from config.seed and salt.

The same seed and salt give the same stream in every process (the salt is hashed with SHA-256, not Python's randomised hash).

Parameters:

Name	Type	Description	Default
config	AgriboundConfig or int	Configuration (its seed is used) or an integer seed.	required
*salt	object	Values that separate independent streams (e.g. "split", a tile id).	()

Returns:

Type	Description
Generator

Source code in agribound/_repro.py

def get_rng(config: Any, *salt: object) -> np.random.Generator:
    """Return a NumPy generator seeded from ``config.seed`` and *salt*.

    The same seed and salt give the same stream in every process (the salt is
    hashed with SHA-256, not Python's randomised ``hash``).

    Parameters
    ----------
    config : AgriboundConfig or int
        Configuration (its ``seed`` is used) or an integer seed.
    *salt : object
        Values that separate independent streams (e.g. ``"split"``, a tile id).

    Returns
    -------
    numpy.random.Generator
    """
    seed = int(config) if isinstance(config, int | np.integer) else int(config.seed)
    entropy = [seed] + [_stable_int(s) for s in salt]
    return np.random.default_rng(np.random.SeedSequence(entropy))

collect_versions

collect_versions(extra: Iterable[str] = ()) -> dict[str, str]

Return versions of Python, agribound, GDAL and known installed packages.

Package versions are read from distribution metadata, so nothing heavy is imported. Packages that are not installed are omitted.

Parameters:

Name	Type	Description	Default
extra	iterable of str	Additional distribution names to include.	()

Returns:

Type	Description
dict[str, str]	Name -> version.

Source code in agribound/_repro.py

def collect_versions(extra: Iterable[str] = ()) -> dict[str, str]:
    """Return versions of Python, agribound, GDAL and known installed packages.

    Package versions are read from distribution metadata, so nothing heavy is
    imported. Packages that are not installed are omitted.

    Parameters
    ----------
    extra : iterable of str
        Additional distribution names to include.

    Returns
    -------
    dict[str, str]
        Name -> version.
    """
    from agribound._version import __version__

    versions: dict[str, str] = {
        "python": sys.version.split()[0],
        "agribound": __version__,
    }
    for dist in (*KNOWN_PACKAGES, *extra):
        try:
            versions[dist] = importlib.metadata.version(dist)
        except importlib.metadata.PackageNotFoundError:
            continue
        except Exception:  # pragma: no cover - malformed metadata
            continue
    try:
        import rasterio

        versions["gdal"] = str(rasterio.__gdal_version__)
    except Exception:  # pragma: no cover - rasterio is a core dependency
        pass
    return versions

new_run_id

new_run_id() -> str

Return a new run ID such as "20260926T170102Z-3f2a9c" (UTC time + 6 hex).

Source code in agribound/_repro.py

def new_run_id() -> str:
    """Return a new run ID such as ``"20260926T170102Z-3f2a9c"`` (UTC time + 6 hex)."""
    stamp = _dt.datetime.now(_dt.UTC).strftime("%Y%m%dT%H%M%SZ")
    return f"{stamp}-{secrets.token_hex(3)}"