Biodiversity API

The helpers shared by the four biodiversity backends (gbif, obis, wdpa, iucn). For how to use them, see Biodiversity usage.

from earthlens.biodiversity import wkt_from_bbox, occurrences_to_fc, warn_license, LicenseWarning

Geometry

earthlens.biodiversity.wkt_from_bbox(space)

Build a counter-clockwise WKT polygon from a spatial extent's bbox.

SpatialExtent exposes the bbox edges as .west/.south/.east/.north but has no .wkt(), so the cluster builds the geometry= filter the GBIF / OBIS / WDPA APIs accept here. shapely.geometry.box emits a counter-clockwise ring, which GBIF requires.

Parameters:

Name	Type	Description	Default
space	SpatialExtent	A spatial extent exposing .west, .south, .east, and .north float properties (typically self.space on a backend).	required

Returns:

Type	Description
str	A POLYGON((...)) WKT string spanning the bounding box.

Examples:

• Build the WKT for a small box:

  >>> from earthlens.base import SpatialExtent
  >>> from earthlens.biodiversity import wkt_from_bbox
  >>> extent = SpatialExtent.from_pairs(lat_lim=(10.0, 20.0), lon_lim=(0.0, 5.0))
  >>> wkt_from_bbox(extent)
  'POLYGON ((5 10, 5 20, 0 20, 0 10, 5 10))'
  

Source code in libs/core/src/earthlens/biodiversity/_helpers.py

def wkt_from_bbox(space: SpatialExtent) -> str:
    """Build a counter-clockwise WKT polygon from a spatial extent's bbox.

    `SpatialExtent` exposes the bbox edges as `.west/.south/.east/.north` but
    has no `.wkt()`, so the cluster builds the `geometry=` filter the GBIF /
    OBIS / WDPA APIs accept here. `shapely.geometry.box` emits a
    counter-clockwise ring, which GBIF requires.

    Args:
        space: A spatial extent exposing `.west`, `.south`, `.east`, and
            `.north` float properties (typically `self.space` on a backend).

    Returns:
        A `POLYGON((...))` WKT string spanning the bounding box.

    Examples:
        - Build the WKT for a small box:
            ```python
            >>> from earthlens.base import SpatialExtent
            >>> from earthlens.biodiversity import wkt_from_bbox
            >>> extent = SpatialExtent.from_pairs(lat_lim=(10.0, 20.0), lon_lim=(0.0, 5.0))
            >>> wkt_from_bbox(extent)
            'POLYGON ((5 10, 5 20, 0 20, 0 10, 5 10))'

            ```
    """
    return cast("str", box(space.west, space.south, space.east, space.north).wkt)

Occurrence conversion

earthlens.biodiversity.occurrences_to_fc(records, *, lat_field, lon_field, columns)

Map occurrence rows to a points FeatureCollection (EPSG:4326).

Accepts both shapes the cluster produces: a list[dict] of records (GBIF's occ.search()["results"]) or a pandas.DataFrame (the value pyobis's .execute() returns). The output is one feature per row, restricted and ordered to columns with their declared dtypes, plus a geometry column of shapely.Point(lon, lat). A row whose latitude or longitude is missing gets a null geometry rather than an invalid POINT (nan nan) that would corrupt a written file. An empty input yields an empty FeatureCollection carrying exactly columns, so the result type is identical whether or not the query matched anything.

Parameters:

Name	Type	Description	Default
records	Iterable[Mapping] | DataFrame	Occurrence rows as a list[dict] / iterable of mappings, or a pandas.DataFrame already shaped one row per occurrence.	required
lat_field	str	Name of the latitude column (e.g. "decimalLatitude").	required
lon_field	str	Name of the longitude column (e.g. "decimalLongitude").	required
columns	Mapping[str, str]	Ordered mapping of output column name to pandas dtype; the result carries exactly these attribute columns.	required

Returns:

Name	Type	Description
FeatureCollection	FeatureCollection	One feature per row, CRS EPSG:4326; rows with a missing coordinate carry a null geometry.

Source code in libs/core/src/earthlens/biodiversity/_helpers.py

def occurrences_to_fc(
    records: Iterable[Mapping] | pd.DataFrame,
    *,
    lat_field: str,
    lon_field: str,
    columns: Mapping[str, str],
) -> FeatureCollection:
    """Map occurrence rows to a points `FeatureCollection` (EPSG:4326).

    Accepts both shapes the cluster produces: a `list[dict]` of records (GBIF's
    `occ.search()["results"]`) or a `pandas.DataFrame` (the value `pyobis`'s
    `.execute()` returns). The output is one feature per row, restricted and
    ordered to `columns` with their declared dtypes, plus a `geometry` column
    of `shapely.Point(lon, lat)`. A row whose latitude or longitude is missing
    gets a null geometry rather than an invalid `POINT (nan nan)` that would
    corrupt a written file. An empty input yields an empty FeatureCollection
    carrying exactly `columns`, so the result type is identical whether or not
    the query matched anything.

    Args:
        records: Occurrence rows as a `list[dict]` / iterable of mappings, or a
            `pandas.DataFrame` already shaped one row per occurrence.
        lat_field: Name of the latitude column (e.g. `"decimalLatitude"`).
        lon_field: Name of the longitude column (e.g. `"decimalLongitude"`).
        columns: Ordered mapping of output column name to pandas dtype; the
            result carries exactly these attribute columns.

    Returns:
        FeatureCollection: One feature per row, CRS `EPSG:4326`; rows with a
            missing coordinate carry a null geometry.
    """
    frame = (
        records.copy()
        if isinstance(records, pd.DataFrame)
        else pd.DataFrame(list(records), columns=list(columns))
    )
    if frame.empty:
        return _empty_fc(columns)

    frame = frame.reindex(columns=list(columns))
    for column, dtype in columns.items():
        frame[column] = frame[column].astype(dtype)

    points = [
        Point(lon, lat) if pd.notna(lon) and pd.notna(lat) else None
        for lon, lat in zip(frame[lon_field], frame[lat_field])
    ]
    # Tag the GeoSeries with the frame's own index so geopandas does not align
    # a default RangeIndex against a non-default frame index (which would null
    # every geometry) — OBIS frames can carry a non-default index.
    geometry = gpd.GeoSeries(points, index=frame.index, crs=CRS)
    gdf = gpd.GeoDataFrame(frame, geometry=geometry, crs=CRS)
    return FeatureCollection(gdf)

Licensing

earthlens.biodiversity.LicenseWarning

Bases: UserWarning

Warns that a downloaded result carries license obligations.

Emitted by warn_license when a result's license is non-commercial (CC-BY-NC), share-alike, or otherwise restricts redistribution (the custom Protected Planet / IUCN Red List terms). A downstream commercial user must be told the obligation rides along with the data rather than discovering it silently.

Promoted here from the Overture backend so every biodiversity source — and Overture — raises the same warning class; earthlens.overture._helpers re-exports it for backward compatibility.

Source code in libs/core/src/earthlens/biodiversity/_helpers.py

class LicenseWarning(UserWarning):
    """Warns that a downloaded result carries license obligations.

    Emitted by `warn_license` when a result's license is non-commercial
    (`CC-BY-NC`), share-alike, or otherwise restricts redistribution (the
    custom Protected Planet / IUCN Red List terms). A downstream commercial
    user must be told the obligation rides along with the data rather than
    discovering it silently.

    Promoted here from the Overture backend so every biodiversity source — and
    Overture — raises the same warning class; `earthlens.overture._helpers`
    re-exports it for backward compatibility.
    """

earthlens.biodiversity.warn_license(license_id, label, *, detail=None)

Emit a LicenseWarning when a result's license is restrictive.

No-ops for permissive licenses (CC0, CC-BY) so a caller can pass every record's license unconditionally. detail appends a source-specific obligation to the message.

Parameters:

Name	Type	Description	Default
license_id	str	The license id/label on the result (e.g. "CC_BY_NC_4_0", or one of WDPA_LICENSE / IUCN_LICENSE).	required
label	str	A short source/dataset label for the message (e.g. "gbif").	required
detail	str | None	Optional source-specific obligation appended to the message.	None

Returns:

Type	Description
bool	True if a warning was emitted, False otherwise.

Source code in libs/core/src/earthlens/biodiversity/_helpers.py

def warn_license(license_id: str, label: str, *, detail: str | None = None) -> bool:
    """Emit a `LicenseWarning` when a result's license is restrictive.

    No-ops for permissive licenses (`CC0`, `CC-BY`) so a caller can pass every
    record's license unconditionally. `detail` appends a source-specific
    obligation to the message.

    Args:
        license_id: The license id/label on the result (e.g. `"CC_BY_NC_4_0"`,
            or one of `WDPA_LICENSE` / `IUCN_LICENSE`).
        label: A short source/dataset label for the message (e.g. `"gbif"`).
        detail: Optional source-specific obligation appended to the message.

    Returns:
        `True` if a warning was emitted, `False` otherwise.
    """
    if license_id not in RESTRICTIVE_LICENSES:
        return False
    message = (
        f"{label}: '{license_id}' carries non-commercial / restricted-redistribution "
        f"obligations"
    )
    if detail:
        # ASCII hyphen (not an em-dash) on purpose: this string is printed by
        # `warnings.warn` to stderr, which on a default Windows cp1252 console
        # would `UnicodeEncodeError` on `—`. Keep this ASCII.
        message += f" - {detail}"
    message += ". Honour attribution and do not redistribute without permission."
    warnings.warn(message, LicenseWarning, stacklevel=2)
    return True