Pipeline¶

agribound.pipeline runs the end-to-end workflow: seed → output-reuse check → composite (+ LULC raster prefetch) → fine-tuning → engine → SAM refinement → study-area selection → post-processing → LULC crop filter → metadata columns → evaluation → export → provenance record. build_composite runs only the first stage. See the Quickstart and Reproducibility.

pipeline ¶

Main pipeline orchestrator.

:func:delineate runs the end-to-end workflow::

seed -> reuse check -> composite (+ LULC raster prefetch) -> fine-tuning
     -> engine -> SAM refinement -> study-area selection -> post-processing
     -> LULC crop filter -> metadata columns -> evaluation -> export
     -> provenance record

Composites cover the study area's bounding box in the export CRS (they are not masked to the study-area polygons), so the study-area selection (config.aoi_selection, default "representative_point") removes predictions outside an irregular study area before post-processing.

:func:build_composite runs only the first stage (composite or embedding download), so data preparation can happen on a node with network access and delineation later on a GPU node.

Every stage runs inside :meth:agribound.provenance.RunRecorder.step; the record is written to <output_path>.provenance.json and decides whether an existing output can be reused (:func:agribound.provenance.reuse_mismatch: configuration hash, study-area fingerprint and results versions).

delineate ¶

delineate(study_area: str | None = None, source: str = 'sentinel2', year: int = 2024, engine: str = 'delineate-anything', output_path: str | None = None, gee_project: str | None = None, local_tif_path: str | None = None, reference_boundaries: str | None = None, fine_tune: bool = False, config: AgriboundConfig | None = None, **kwargs: Any) -> gpd.GeoDataFrame

Run the complete field boundary delineation pipeline.

Parameters:

Name	Type	Description	Default
`study_area`	`str or None`	Vector file, GEE asset ID, `"bbox:minx,miny,maxx,maxy"` or WKT (EPSG:4326). Optional for `source="local"`.	`None`
`source`	`str`	Satellite source (see :func:`agribound.list_sources`).	`'sentinel2'`
`year`	`int`	Target year (default 2024).	`2024`
`engine`	`str`	Delineation engine (see :func:`agribound.list_engines`).	`'delineate-anything'`
`output_path`	`str or None`	Output vector file. Default `"fields_{source}_{year}.<ext>"` in the current directory, with the extension of `output_format`.	`None`
`gee_project`	`str or None`	GEE project ID (required for GEE-based sources).	`None`
`local_tif_path`	`str or None`	Local GeoTIFF (required when `source="local"`).	`None`
`reference_boundaries`	`str or None`	Reference field boundaries for fine-tuning or evaluation.	`None`
`fine_tune`	`bool`	Fine-tune the engine on reference_boundaries before inference.	`False`
`config`	`AgriboundConfig or None`	Pre-built configuration. Keyword arguments, and named arguments whose value differs from both their default and the configuration, are applied on top of it with :meth:`AgriboundConfig.merged` (logged). The caller's object is never modified.	`None`
`**kwargs`	`Any`	Any other :class:`AgriboundConfig` field (`min_area` and `simplify` are accepted as aliases of `min_field_area_m2` and `simplify_tolerance`).	`{}`

Returns:

Type Description

GeoDataFrame

Field boundary polygons with metadata columns id, metrics:area (m², EPSG:6933), metrics:perimeter (m, geodesic), agribound:compactness (Polsby-Popper 4πA/P²), determination:method, determination:datetime (last day of the imagery the engine read, 23:59:59 UTC: the end of date_range or year; when the engine composited its own windows (FTW's two seasonal windows, also as an ensemble member), the latest end of those windows replaces it, and can fall in year + 1), agribound:engine, agribound:source, agribound:year, agribound:version and agribound:run_id. gdf.attrs holds run_id, engine_meta, lulc_stats, sam_stats and evaluation_metrics when available.

Raises:

Type Description

FileExistsError

If output_path exists, overwrite is False, and its provenance record is missing, failed, has a different configuration hash, was made from a study-area file whose geometry has changed at the same path (without a study area: from a local raster whose path, size or modification time has changed), or has different results versions (:mod:agribound._results: a component's output for the same configuration has changed since the release that made it).

RuntimeError

If the LULC filter fails and lulc_on_error="raise" (the default).

Notes

An existing output is returned without recomputation when its provenance record reports a successful run with the same :func:agribound.provenance.config_hash, study-area fingerprint (facts["aoi_fingerprint"]) and results versions (facts["results_versions"]); see :func:agribound.provenance.reuse_mismatch. A record of agribound <= 1.0.0 has neither fact: it counts as results version 1, and a study-area file (or local raster) cannot be verified (logged at WARNING; the output is reused). With provenance=False no record is written, so a later run on the same output_path needs overwrite=True.

Zero detections are not an early exit: an empty output file, the evaluation (if requested) and the provenance record are still written.

After delineation (and SAM refinement) the polygons are restricted to the study-area geometry according to aoi_selection (default "representative_point"; see :class:~agribound.config.AgriboundConfig and :func:select_in_study_area); the provenance fact aoi_selection records the rule and the counts before and after.

Examples:

>>> import agribound
>>> gdf = agribound.delineate(
...     study_area="area.geojson",
...     source="sentinel2",
...     year=2024,
...     engine="delineate-anything",
...     gee_project="my-project",
... )

Source code in agribound/pipeline.py

def delineate(
    study_area: str | None = None,
    source: str = "sentinel2",
    year: int = 2024,
    engine: str = "delineate-anything",
    output_path: str | None = None,
    gee_project: str | None = None,
    local_tif_path: str | None = None,
    reference_boundaries: str | None = None,
    fine_tune: bool = False,
    config: AgriboundConfig | None = None,
    **kwargs: Any,
) -> gpd.GeoDataFrame:
    """Run the complete field boundary delineation pipeline.

    Parameters
    ----------
    study_area : str or None
        Vector file, GEE asset ID, ``"bbox:minx,miny,maxx,maxy"`` or WKT
        (EPSG:4326). Optional for ``source="local"``.
    source : str
        Satellite source (see :func:`agribound.list_sources`).
    year : int
        Target year (default 2024).
    engine : str
        Delineation engine (see :func:`agribound.list_engines`).
    output_path : str or None
        Output vector file. Default ``"fields_{source}_{year}.<ext>"`` in the
        current directory, with the extension of ``output_format``.
    gee_project : str or None
        GEE project ID (required for GEE-based sources).
    local_tif_path : str or None
        Local GeoTIFF (required when ``source="local"``).
    reference_boundaries : str or None
        Reference field boundaries for fine-tuning or evaluation.
    fine_tune : bool
        Fine-tune the engine on *reference_boundaries* before inference.
    config : AgriboundConfig or None
        Pre-built configuration. Keyword arguments, and named arguments whose
        value differs from both their default and the configuration, are
        applied on top of it with :meth:`AgriboundConfig.merged` (logged).
        The caller's object is never modified.
    **kwargs
        Any other :class:`AgriboundConfig` field (``min_area`` and
        ``simplify`` are accepted as aliases of ``min_field_area_m2`` and
        ``simplify_tolerance``).

    Returns
    -------
    geopandas.GeoDataFrame
        Field boundary polygons with metadata columns ``id``,
        ``metrics:area`` (m², EPSG:6933), ``metrics:perimeter`` (m,
        geodesic), ``agribound:compactness`` (Polsby-Popper 4πA/P²),
        ``determination:method``, ``determination:datetime`` (last day of
        the imagery the engine read, 23:59:59 UTC: the end of ``date_range``
        or ``year``; when the engine composited its own windows (FTW's two
        seasonal windows, also as an ensemble member), the latest end of
        those windows replaces it, and can fall in ``year + 1``),
        ``agribound:engine``, ``agribound:source``,
        ``agribound:year``, ``agribound:version`` and ``agribound:run_id``.
        ``gdf.attrs`` holds ``run_id``, ``engine_meta``, ``lulc_stats``,
        ``sam_stats`` and ``evaluation_metrics`` when available.

    Raises
    ------
    FileExistsError
        If *output_path* exists, ``overwrite`` is *False*, and its provenance
        record is missing, failed, has a different configuration hash, was
        made from a study-area file whose geometry has changed at the same
        path (without a study area: from a local raster whose path, size or
        modification time has changed), or has different results versions
        (:mod:`agribound._results`: a component's output for the same
        configuration has changed since the release that made it).
    RuntimeError
        If the LULC filter fails and ``lulc_on_error="raise"`` (the default).

    Notes
    -----
    An existing output is returned without recomputation when its
    provenance record reports a successful run with the same
    :func:`agribound.provenance.config_hash`, study-area fingerprint
    (``facts["aoi_fingerprint"]``) and results versions
    (``facts["results_versions"]``); see
    :func:`agribound.provenance.reuse_mismatch`. A record of agribound <=
    1.0.0 has neither fact: it counts as results version 1, and a
    study-area file (or local raster) cannot be verified (logged at WARNING;
    the output is reused). With ``provenance=False`` no record is
    written, so a later run on the same *output_path* needs
    ``overwrite=True``.

    Zero detections are not an early exit: an empty output file, the
    evaluation (if requested) and the provenance record are still written.

    After delineation (and SAM refinement) the polygons are restricted to the
    study-area geometry according to ``aoi_selection`` (default
    ``"representative_point"``; see :class:`~agribound.config.AgriboundConfig`
    and :func:`select_in_study_area`); the provenance fact ``aoi_selection``
    records the rule and the counts before and after.

    Examples
    --------
    >>> import agribound
    >>> gdf = agribound.delineate(
    ...     study_area="area.geojson",
    ...     source="sentinel2",
    ...     year=2024,
    ...     engine="delineate-anything",
    ...     gee_project="my-project",
    ... )
    """
    named = {
        "study_area": study_area,
        "source": source,
        "year": year,
        "engine": engine,
        "output_path": output_path,
        "gee_project": gee_project,
        "local_tif_path": local_tif_path,
        "reference_boundaries": reference_boundaries,
        "fine_tune": fine_tune,
    }
    kwargs = _resolve_aliases(kwargs)
    if config is None:
        config = _config_from_arguments(named, kwargs)
    else:
        config = _apply_overrides(config, named, kwargs)

    from agribound._repro import seed_everything

    # Work on a private copy: stages (e.g. fine-tuning) add engine parameters.
    config = copy.deepcopy(config)
    seed_everything(config.seed)

    logger.info(
        "Agribound pipeline: source=%s, engine=%s, year=%d, output=%s",
        config.source,
        config.engine,
        config.year,
        config.output_path,
    )

    existing = _load_existing_output(config)
    if existing is not None:
        return existing

    from agribound.provenance import RunRecorder, reuse_facts, write_provenance

    recorder = RunRecorder(config)
    # Checked by the reuse test of later runs, with the configuration hash.
    for key, value in reuse_facts(config).items():
        recorder.set(key, value)
    output_file = Path(config.output_path)
    start = time.perf_counter()
    try:
        with recorder:
            gdf = _run_stages(config, recorder)
    except BaseException:
        if config.provenance and not output_file.exists():
            # Record the failure; never overwrite the record of an older output.
            try:
                write_provenance(config.output_path, recorder.to_dict())
            except Exception as exc:  # pragma: no cover - best effort
                logger.warning("Could not write provenance for the failed run: %s", exc)
        raise

    if config.provenance:
        path = write_provenance(config.output_path, recorder.to_dict())
        gdf.attrs["provenance_path"] = str(path)
        logger.info("Provenance written to %s", path)

    logger.info(
        "Pipeline complete: %d field boundaries in %.1f s -> %s",
        len(gdf),
        time.perf_counter() - start,
        config.output_path,
    )
    return gdf

build_composite ¶

build_composite(config: AgriboundConfig) -> str

Build (or load from cache) the input raster for config.

Runs the composite builder of config.source (GEE composite, USGS export, local file or embedding download). When config.lulc_filter is enabled and config.lulc_mode == "raster", the LULC raster for the study area is also downloaded into the cache, and a GEE-asset study area is saved to the working directory (:func:agribound.io.vector.read_config_study_area), so that the delineation stage can run without network access. Builders cache their outputs with :func:agribound._cache.cache_path, so repeated calls reuse the files.

Parameters:

Name	Type	Description	Default
`config`	`AgriboundConfig`	Pipeline configuration.	required

Returns:

Type	Description
`str`	Path to the composite (or embedding) GeoTIFF.

Raises:

Type	Description
`ValueError`	If no study area is configured for a non-local source.
`RuntimeError`	If the LULC raster prefetch fails and `config.lulc_on_error="raise"`.

Source code in agribound/pipeline.py

def build_composite(config: AgriboundConfig) -> str:
    """Build (or load from cache) the input raster for *config*.

    Runs the composite builder of ``config.source`` (GEE composite, USGS
    export, local file or embedding download). When ``config.lulc_filter`` is
    enabled and ``config.lulc_mode == "raster"``, the LULC raster for the
    study area is also downloaded into the cache, and a GEE-asset study area
    is saved to the working directory
    (:func:`agribound.io.vector.read_config_study_area`), so that the
    delineation stage can run without network access. Builders cache their
    outputs with :func:`agribound._cache.cache_path`, so repeated calls reuse
    the files.

    Parameters
    ----------
    config : AgriboundConfig
        Pipeline configuration.

    Returns
    -------
    str
        Path to the composite (or embedding) GeoTIFF.

    Raises
    ------
    ValueError
        If no study area is configured for a non-local source.
    RuntimeError
        If the LULC raster prefetch fails and ``config.lulc_on_error="raise"``.
    """
    return _build_composite(config, recorder=None)

study_area_in_crs ¶

study_area_in_crs(config: AgriboundConfig, crs: Any) -> Any

Return the study area of config as one valid polygonal geometry in crs.

The features are unioned, repaired and reduced to their polygonal parts in the study area's own CRS (EPSG:4326 when it has none, with a WARNING), densified (:data:_AOI_MAX_SEGMENT_DEG / :data:_AOI_MAX_SEGMENT_M) and reprojected to crs.

Parameters:

Name	Type	Description	Default
`config`	`AgriboundConfig`	Configuration with a `study_area`, read with :func:`agribound.io.vector.read_config_study_area`: a GEE asset ID is read from its local copy in the working directory when one exists, else from Earth Engine with the configured credentials.	required
`crs`	`CRS or str`	Target CRS.	required

Returns:

Type	Description
`Polygon or MultiPolygon`

Raises:

Type	Description
`ValueError`	If the study area has no polygonal area.

Source code in agribound/pipeline.py

def study_area_in_crs(config: AgriboundConfig, crs: Any) -> Any:
    """Return the study area of *config* as one valid polygonal geometry in *crs*.

    The features are unioned, repaired and reduced to their polygonal parts
    in the study area's own CRS (EPSG:4326 when it has none, with a WARNING),
    densified (:data:`_AOI_MAX_SEGMENT_DEG` / :data:`_AOI_MAX_SEGMENT_M`) and
    reprojected to *crs*.

    Parameters
    ----------
    config : AgriboundConfig
        Configuration with a ``study_area``, read with
        :func:`agribound.io.vector.read_config_study_area`: a GEE asset ID is
        read from its local copy in the working directory when one exists,
        else from Earth Engine with the configured credentials.
    crs : pyproj.CRS or str
        Target CRS.

    Returns
    -------
    shapely.geometry.Polygon or shapely.geometry.MultiPolygon

    Raises
    ------
    ValueError
        If the study area has no polygonal area.
    """
    import pyproj
    import shapely

    from agribound.io.vector import read_config_study_area
    from agribound.postprocess.simplify import make_polygonal

    aoi = read_config_study_area(config)
    if aoi.crs is None:
        logger.warning("Study area %s has no CRS; assuming EPSG:4326", config.study_area)
        aoi = aoi.set_crs("EPSG:4326")
    geom = make_polygonal(shapely.union_all(aoi.geometry.values))
    if geom is None or geom.is_empty or geom.area <= 0:
        raise ValueError(f"Study area {config.study_area!r} has no polygonal area")
    step = _AOI_MAX_SEGMENT_DEG if pyproj.CRS(aoi.crs).is_geographic else _AOI_MAX_SEGMENT_M
    dense = shapely.segmentize(geom, max_segment_length=step)
    projected = gpd.GeoSeries([dense], crs=aoi.crs).to_crs(crs).iloc[0]
    result = make_polygonal(projected)
    if result is None or result.is_empty:
        raise ValueError(f"Study area {config.study_area!r} is empty in {crs}")
    return result

select_in_study_area ¶

select_in_study_area(gdf: GeoDataFrame, aoi: Any, rule: str) -> tuple[gpd.GeoDataFrame, dict[str, Any]]

Restrict polygons to a study-area geometry.

Parameters:

Name	Type	Description	Default
`gdf`	`GeoDataFrame`	Polygons (any CRS; aoi must be in the same CRS).	required
`aoi`	`shapely geometry`	Polygonal study area, e.g. from :func:`study_area_in_crs`.	required
`rule`	`str`	`"representative_point"`: keep rows whose :meth:`~shapely.Geometry.representative_point` (a point guaranteed to lie in the polygon) intersects aoi (inside it or on its boundary); geometries are unchanged. `"intersects"`: keep rows whose geometry intersects aoi; geometries are unchanged. `"clip"`: replace each geometry by its intersection with aoi (after repairing invalid geometries) and keep rows with a non-empty polygonal result; lines and points of the intersection are discarded. `"none"`: keep all. Null and empty geometries are dropped by every rule except `"none"`.	required

Returns:

Type	Description
`tuple`	`(selected, stats)` with `stats = {"rule", "n_before", "n_after"}` and, for `"clip"`, `"n_clipped"` (rows extending outside aoi: cut, or dropped when nothing polygonal remained). Attributes and `gdf.attrs` are kept; the index is reset.

Raises:

Type	Description
`ValueError`	For an unknown rule.

Source code in agribound/pipeline.py

def select_in_study_area(
    gdf: gpd.GeoDataFrame, aoi: Any, rule: str
) -> tuple[gpd.GeoDataFrame, dict[str, Any]]:
    """Restrict polygons to a study-area geometry.

    Parameters
    ----------
    gdf : geopandas.GeoDataFrame
        Polygons (any CRS; *aoi* must be in the same CRS).
    aoi : shapely geometry
        Polygonal study area, e.g. from :func:`study_area_in_crs`.
    rule : str
        ``"representative_point"``: keep rows whose
        :meth:`~shapely.Geometry.representative_point` (a point guaranteed to
        lie in the polygon) intersects *aoi* (inside it or on its boundary);
        geometries are unchanged. ``"intersects"``: keep rows whose geometry
        intersects *aoi*; geometries are unchanged. ``"clip"``: replace each
        geometry by its intersection with *aoi* (after repairing invalid
        geometries) and keep rows with a non-empty polygonal result; lines
        and points of the intersection are discarded. ``"none"``: keep all.
        Null and empty geometries are dropped by every rule except ``"none"``.

    Returns
    -------
    tuple
        ``(selected, stats)`` with ``stats = {"rule", "n_before", "n_after"}``
        and, for ``"clip"``, ``"n_clipped"`` (rows extending outside *aoi*:
        cut, or dropped when nothing polygonal remained). Attributes and
        ``gdf.attrs`` are kept; the index is reset.

    Raises
    ------
    ValueError
        For an unknown *rule*.
    """
    import shapely

    from agribound.postprocess.simplify import make_polygonal

    n_before = len(gdf)
    stats: dict[str, Any] = {"rule": rule, "n_before": n_before}
    if rule == "none" or n_before == 0:
        stats["n_after"] = n_before
        return gdf, stats
    if rule not in ("representative_point", "intersects", "clip"):
        raise ValueError(f"Unknown aoi_selection rule {rule!r}")

    geoms = np.asarray(gdf.geometry.array, dtype=object)
    # Invalid geometries are tested (and clipped) in repaired form; the
    # selection rules keep the original geometry.
    work = np.array(
        [g if g is None or g.is_empty or g.is_valid else make_polygonal(g) for g in geoms],
        dtype=object,
    )
    present = np.array([g is not None and not g.is_empty for g in work], dtype=bool)
    shapely.prepare(aoi)
    if rule == "representative_point":
        keep = present.copy()
        keep[present] = shapely.intersects(aoi, shapely.point_on_surface(work[present]))
        selected = gdf[keep]
    elif rule == "intersects":
        keep = present.copy()
        keep[present] = shapely.intersects(aoi, work[present])
        selected = gdf[keep]
    else:
        clipped = np.full(n_before, None, dtype=object)
        n_clipped = 0
        for i in np.flatnonzero(present):
            if shapely.contains(aoi, work[i]):
                clipped[i] = geoms[i]
                continue
            n_clipped += 1
            part = make_polygonal(shapely.intersection(work[i], aoi))
            if part is not None and not part.is_empty:
                clipped[i] = part
        keep = np.array([g is not None for g in clipped], dtype=bool)
        selected = gdf[keep].copy()
        selected[gdf.geometry.name] = gpd.GeoSeries(
            list(clipped[keep]), index=selected.index, crs=gdf.crs
        )
        stats["n_clipped"] = int(n_clipped)
    selected = selected.reset_index(drop=True)
    selected.attrs = dict(gdf.attrs)
    stats["n_after"] = len(selected)
    return selected, stats