Skip to content

eii.compute

Full EII computation utilities (on-the-fly).

Aggregation

Combined Ecosystem Integrity Index calculation.

calculate_eii(aoi, method=DEFAULT_AGGREGATION_METHOD, year_range=None, include_seasonality=True)

Calculate the full Ecosystem Integrity Index for an area.

Parameters:

Name Type Description Default
aoi Geometry

Earth Engine geometry defining the area of interest.

 required
method Literal['minimum', 'product', 'min_fuzzy_logic', 'geometric_mean']

Method to combine components. Options: - "minimum": Simple minimum of components (most conservative) - "product": Product of components - "min_fuzzy_logic": Minimum with fuzzy compensation (default) - "geometric_mean": Geometric mean of components

 DEFAULT_AGGREGATION_METHOD
year_range list[str] | None

Date range for actual NPP [start_date, end_date]. Defaults to OBSERVED_NPP_YEAR_RANGE.

 None
include_seasonality bool

Include seasonality in functional integrity.

 True

Returns:

Type Description
dict[str, Image]

Dictionary containing functional_integrity, structural_integrity,
dict[str, Image]

compositional_integrity, and combined eii images.
Source code in src/eii/compute/integrity.py 
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
def calculate_eii(
    aoi: ee.Geometry,
    method: Literal[
        "minimum", "product", "min_fuzzy_logic", "geometric_mean"
    ] = DEFAULT_AGGREGATION_METHOD,
    year_range: list[str] | None = None,
    include_seasonality: bool = True,
) -> dict[str, ee.Image]:
    """
    Calculate the full Ecosystem Integrity Index for an area.

    Args:
        aoi: Earth Engine geometry defining the area of interest.
        method: Method to combine components. Options:
            - "minimum": Simple minimum of components (most conservative)
            - "product": Product of components
            - "min_fuzzy_logic": Minimum with fuzzy compensation (default)
            - "geometric_mean": Geometric mean of components
        year_range: Date range for actual NPP [start_date, end_date].
            Defaults to OBSERVED_NPP_YEAR_RANGE.
        include_seasonality: Include seasonality in functional integrity.

    Returns:
        Dictionary containing functional_integrity, structural_integrity,
        compositional_integrity, and combined eii images.
    """
    from .compositional import calculate_compositional_integrity
    from .npp import calculate_functional_integrity
    from .structural import calculate_structural_integrity

    if year_range is None:
        year_range = OBSERVED_NPP_YEAR_RANGE

    # Buffer AOI for edge calculations
    buffered_aoi = aoi.buffer(20000).bounds()

    # Calculate components
    functional_result = calculate_functional_integrity(
        aoi=buffered_aoi,
        year_range=year_range,
        include_seasonality=include_seasonality,
    )
    functional = functional_result["functional_integrity"]

    structural = calculate_structural_integrity(aoi=buffered_aoi)

    compositional = calculate_compositional_integrity(aoi=buffered_aoi)

    # Combine components
    eii = combine_components(
        functional=functional,
        structural=structural,
        compositional=compositional,
        method=method,
    )

    # Clip to original AOI
    return {
        "functional_integrity": functional.clip(aoi),
        "structural_integrity": structural.clip(aoi),
        "compositional_integrity": compositional.clip(aoi),
        "eii": eii.clip(aoi),
    }

combine_components(functional, structural, compositional, method=DEFAULT_AGGREGATION_METHOD)

Combine integrity components into a single EII score.

Parameters:

Name Type Description Default
functional Image

Functional integrity image (0-1).

 required
structural Image

Structural integrity image (0-1).

 required
compositional Image

Compositional integrity image (0-1).

 required
method Literal['minimum', 'product', 'min_fuzzy_logic', 'geometric_mean']

Aggregation method.

 DEFAULT_AGGREGATION_METHOD

Returns:

Type Description
Image

Combined EII image (0-1).
Source code in src/eii/compute/integrity.py 
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
def combine_components(
    functional: ee.Image,
    structural: ee.Image,
    compositional: ee.Image,
    method: Literal[
        "minimum", "product", "min_fuzzy_logic", "geometric_mean"
    ] = DEFAULT_AGGREGATION_METHOD,
) -> ee.Image:
    """
    Combine integrity components into a single EII score.

    Args:
        functional: Functional integrity image (0-1).
        structural: Structural integrity image (0-1).
        compositional: Compositional integrity image (0-1).
        method: Aggregation method.

    Returns:
        Combined EII image (0-1).
    """
    components = ee.Image.cat([functional, structural, compositional])

    if method == "minimum":
        eii = components.reduce(ee.Reducer.min()).rename("eii")

    elif method == "product":
        eii = components.reduce(ee.Reducer.product()).rename("eii")

    elif method == "min_fuzzy_logic":
        # Fuzzy logic combination:
        # EII = M * F, where M = min, F = fuzzy sum of other two
        M = components.reduce(ee.Reducer.min())
        Med = components.reduce(ee.Reducer.median())
        Max = components.reduce(ee.Reducer.max())

        # Fuzzy sum: F = y + z - y*z
        F = Med.add(Max).subtract(Med.multiply(Max))

        eii = M.multiply(F).rename("eii")

    elif method == "geometric_mean":
        # Geometric mean = (a * b * c)^(1/n) where n=3 for three components
        product = components.reduce(ee.Reducer.product())
        eii = product.pow(1.0 / 3.0).rename("eii")

    else:
        raise ValueError(f"Unknown method '{method}'")

    return eii

Functional integrity (NPP)

NPP-based functional integrity calculation.

This module handles: - Predictor stack setup for the NPP model - NPP model inference - Functional integrity score calculation

calculate_functional_integrity(aoi=None, year_range=OBSERVED_NPP_YEAR_RANGE, include_seasonality=True, use_precomputed=True, model_asset_path=DEFAULT_NPP_MODEL_PATH, natural_npp_asset_path=NATURAL_NPP_ASSET_PATH, observed_npp_asset_path=OBSERVED_NPP_ASSET_PATH, natural_npp_use_tiled_collection=False, absolute_diff_percentile='p95')

Calculate NPP-based functional integrity with magnitude and seasonality dimensions.

Magnitude integrity: symmetric deviation score + absolute diff score (weight: 2/3) Seasonality integrity: comparison of observed vs natural intra-annual std (weight: 1/3)

Parameters:

Name Type Description Default
aoi Geometry | None

Area of interest. If None, returns unclipped global image.

 None
year_range list[str]

Date range for observed NPP (3-year rolling window).

 OBSERVED_NPP_YEAR_RANGE
include_seasonality bool

Include seasonality dimension in scoring.

 True
use_precomputed bool

Use pre-computed natural NPP tiles (recommended).

 True
model_asset_path str

Path to NPP model (only used if use_precomputed=False).

 DEFAULT_NPP_MODEL_PATH
natural_npp_asset_path str

Asset path for pre-computed natural NPP.

 NATURAL_NPP_ASSET_PATH
observed_npp_asset_path str

Asset path for observed NPP annual tiles.

 OBSERVED_NPP_ASSET_PATH
natural_npp_use_tiled_collection bool

Use tiled collection for natural NPP.

 False
absolute_diff_percentile str

Percentile key for absolute NPP penalty (e.g. "p80").

 'p95'

Returns:

Type Description
dict[str, Image]

Dictionary with intermediate layers and final functional_integrity score.
Source code in src/eii/compute/npp.py 
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
def calculate_functional_integrity(
    aoi: ee.Geometry | None = None,
    year_range: list[str] = OBSERVED_NPP_YEAR_RANGE,
    include_seasonality: bool = True,
    use_precomputed: bool = True,
    model_asset_path: str = DEFAULT_NPP_MODEL_PATH,
    natural_npp_asset_path: str = NATURAL_NPP_ASSET_PATH,
    observed_npp_asset_path: str = OBSERVED_NPP_ASSET_PATH,
    natural_npp_use_tiled_collection: bool = False,
    absolute_diff_percentile: str = "p95",
) -> dict[str, ee.Image]:
    """
    Calculate NPP-based functional integrity with magnitude and seasonality dimensions.

    Magnitude integrity: symmetric deviation score + absolute diff score (weight: 2/3)
    Seasonality integrity: comparison of observed vs natural intra-annual std (weight: 1/3)

    Args:
        aoi: Area of interest. If None, returns unclipped global image.
        year_range: Date range for observed NPP (3-year rolling window).
        include_seasonality: Include seasonality dimension in scoring.
        use_precomputed: Use pre-computed natural NPP tiles (recommended).
        model_asset_path: Path to NPP model (only used if use_precomputed=False).
        natural_npp_asset_path: Asset path for pre-computed natural NPP.
        observed_npp_asset_path: Asset path for observed NPP annual tiles.
        natural_npp_use_tiled_collection: Use tiled collection for natural NPP.
        absolute_diff_percentile: Percentile key for absolute NPP penalty (e.g. "p80").

    Returns:
        Dictionary with intermediate layers and final functional_integrity score.
    """
    if use_precomputed:
        natural_npp = load_natural_npp(
            aoi=aoi,
            asset_path=natural_npp_asset_path,
            use_tiled_collection=natural_npp_use_tiled_collection,
        )
        potential_npp = natural_npp["natural_npp_mean"].rename("potential_npp")
        natural_npp_std = natural_npp["natural_npp_std"]
    else:
        if include_seasonality:
            raise ValueError("include_seasonality=True requires use_precomputed=True")
        npp_model = ee.Classifier.load(model_asset_path)
        predictors = setup_predictor_stack()
        if aoi is not None:
            predictors = predictors.clip(aoi)
        potential_npp = predictors.classify(npp_model).rename("potential_npp")
        natural_npp_std = None

    observed = _load_observed_npp(
        aoi=aoi,
        year_range=year_range,
        asset_path=observed_npp_asset_path,
    )
    actual_npp = observed["observed_npp_mean"].rename("actual_npp")

    relative_npp = actual_npp.divide(potential_npp).rename("relative_npp")
    npp_difference = actual_npp.subtract(potential_npp).abs().rename("npp_difference")

    smooth_kernel = ee.Kernel.square(1, "pixels")
    relative_npp = relative_npp.focal_mean(kernel=smooth_kernel, iterations=1)
    npp_difference = npp_difference.focal_mean(kernel=smooth_kernel, iterations=1)
    proportional_score = _calculate_symmetric_deviation_score(relative_npp)
    absolute_score = _apply_npp_absolute_diff_scaling(
        npp_difference,
        percentile_key=absolute_diff_percentile,
    )
    magnitude_integrity = (
        proportional_score.add(absolute_score).divide(2.0).rename("magnitude_integrity")
    )

    if include_seasonality and natural_npp_std is not None:
        observed_std = observed["observed_npp_std"]
        std_ratio = observed_std.divide(natural_npp_std).rename("std_ratio")
        std_deviation = std_ratio.subtract(1).abs()
        seasonality_integrity = (
            ee.Image(1).divide(ee.Image(1).add(std_deviation)).rename("seasonality_integrity")
        )
        functional_integrity = (
            magnitude_integrity.multiply(MAGNITUDE_WEIGHT)
            .add(seasonality_integrity.multiply(SEASONALITY_WEIGHT))
            .rename("functional_integrity")
        )
    else:
        seasonality_integrity = None
        functional_integrity = magnitude_integrity.rename("functional_integrity")

    return {
        "potential_npp": potential_npp,
        "natural_npp_std": natural_npp_std,
        "actual_npp": actual_npp,
        "relative_npp": relative_npp,
        "npp_difference": npp_difference,
        "proportional_score": proportional_score,
        "absolute_score": absolute_score,
        "magnitude_integrity": magnitude_integrity,
        "seasonality_integrity": seasonality_integrity,
        "functional_integrity": functional_integrity,
    }

load_natural_npp(aoi=None, asset_path=NATURAL_NPP_ASSET_PATH, use_tiled_collection=False)

Load pre-computed natural NPP mean/std from an image or tiled collection.

Source code in src/eii/compute/npp.py 
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
def load_natural_npp(
    aoi: ee.Geometry | None = None,
    asset_path: str = NATURAL_NPP_ASSET_PATH,
    use_tiled_collection: bool = False,
) -> dict[str, ee.Image]:
    """Load pre-computed natural NPP mean/std from an image or tiled collection."""
    if use_tiled_collection:
        from .._utils.gee import load_tiled_collection

        collection = load_tiled_collection(asset_path)
        natural_npp_image = collection.mosaic()
    else:
        natural_npp_image = ee.Image(asset_path)

    natural_npp_mean = natural_npp_image.select("natural_npp_mean")
    natural_npp_std = natural_npp_image.select("natural_npp_std")

    if aoi is not None:
        natural_npp_mean = natural_npp_mean.clip(aoi)
        natural_npp_std = natural_npp_std.clip(aoi)

    return {"natural_npp_mean": natural_npp_mean, "natural_npp_std": natural_npp_std}

load_natural_npp_tiles(aoi=None, asset_path=NATURAL_NPP_ASSET_PATH, use_tiled_collection=False)

Backward-compatible alias for loading natural NPP.

Source code in src/eii/compute/npp.py 
300
301
302
303
304
305
306
307
308
309
310
def load_natural_npp_tiles(
    aoi: ee.Geometry | None = None,
    asset_path: str = NATURAL_NPP_ASSET_PATH,
    use_tiled_collection: bool = False,
) -> dict[str, ee.Image]:
    """Backward-compatible alias for loading natural NPP."""
    return load_natural_npp(
        aoi=aoi,
        asset_path=asset_path,
        use_tiled_collection=use_tiled_collection,
    )

load_npp_diff_percentiles(asset_path=NPP_DIFF_PERCENTILES_ASSET_PATH)

Load NPP difference percentiles from GEE asset.

The asset is a FeatureCollection with a single feature containing percentile breaks as properties (p05, p10, ..., p95).

Parameters:

Name Type Description Default
asset_path str

Path to the percentiles asset.

 NPP_DIFF_PERCENTILES_ASSET_PATH

Returns:

Type Description
dict[str, float]

Dictionary with percentile values.

Raises:

Type Description
RuntimeError

If the percentiles asset is not available. Run compute_npp_decile_breaks.ipynb to create it.
Source code in src/eii/compute/npp.py 
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
def load_npp_diff_percentiles(
    asset_path: str = NPP_DIFF_PERCENTILES_ASSET_PATH,
) -> dict[str, float]:
    """
    Load NPP difference percentiles from GEE asset.

    The asset is a FeatureCollection with a single feature containing
    percentile breaks as properties (p05, p10, ..., p95).

    Args:
        asset_path: Path to the percentiles asset.

    Returns:
        Dictionary with percentile values.

    Raises:
        RuntimeError: If the percentiles asset is not available.
            Run compute_npp_decile_breaks.ipynb to create it.
    """
    global _NPP_DIFF_PERCENTILES_CACHE

    if _NPP_DIFF_PERCENTILES_CACHE is not None:
        return _NPP_DIFF_PERCENTILES_CACHE

    try:
        fc = ee.FeatureCollection(asset_path)
        props = fc.first().getInfo()["properties"]

        percentiles = {f"p{p}": props[f"p{p}"] for p in range(5, 100, 5)}
        _NPP_DIFF_PERCENTILES_CACHE = percentiles
        return percentiles

    except Exception as e:
        raise RuntimeError(
            f"NPP diff percentiles asset not found at: {asset_path}\n"
            "Run pipelines/modeling/eii_calculation/compute_npp_decile_breaks.ipynb "
            "to create it."
        ) from e

setup_predictor_stack(resolution=SPATIAL_RESOLUTION, include_lat_lon=None, include_regional_tpi=True)

Set up the predictor stack for NPP model inference.

Parameters:

Name Type Description Default
resolution int

Output resolution in meters.

 SPATIAL_RESOLUTION
include_lat_lon bool | None

Whether to include lat/lon as predictors. If None, uses INCLUDE_LAT_LON_PREDICTORS from settings.

 None
include_regional_tpi bool

Whether to calculate and include a regional TPI (Topographic Position Index) based on a coarser DEM. Defaults to True.

 True

Returns:

Type Description
Image

Multi-band image containing all predictor variables.
Source code in src/eii/compute/npp.py 
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
def setup_predictor_stack(
    resolution: int = SPATIAL_RESOLUTION,
    include_lat_lon: bool | None = None,
    include_regional_tpi: bool = True,
) -> ee.Image:
    """
    Set up the predictor stack for NPP model inference.

    Args:
        resolution: Output resolution in meters.
        include_lat_lon: Whether to include lat/lon as predictors.
            If None, uses INCLUDE_LAT_LON_PREDICTORS from settings.
        include_regional_tpi: Whether to calculate and include a regional TPI
            (Topographic Position Index) based on a coarser DEM.
            Defaults to True.

    Returns:
        Multi-band image containing all predictor variables.
    """
    if include_lat_lon is None:
        include_lat_lon = INCLUDE_LAT_LON_PREDICTORS

    # WorldClim bioclimatic variables
    worldclim_bio = ee.Image("WORLDCLIM/V1/BIO")
    bio01 = worldclim_bio.select("bio01").rename("mean_annual_temp")
    bio04 = worldclim_bio.select("bio04").rename("temp_seasonality")
    bio12 = worldclim_bio.select("bio12").rename("annual_precip")
    bio15 = worldclim_bio.select("bio15").rename("precip_seasonality")

    # Aridity index
    aridity = (
        ee.Image("projects/sat-io/open-datasets/global_ai/global_ai_yearly")
        .select("b1")
        .rename("aridity")
    )

    # Topography
    dem = ee.Image("MERIT/DEM/v1_0_3")
    elevation = dem.select("dem").rename("elevation")
    slope = (
        ee.ImageCollection("projects/sat-io/open-datasets/Geomorpho90m/slope")
        .mosaic()
        .rename("slope")
    )
    tpi = (
        ee.ImageCollection("projects/sat-io/open-datasets/Geomorpho90m/tpi").mosaic().rename("tpi")
    )
    tri = (
        ee.ImageCollection("projects/sat-io/open-datasets/Geomorpho90m/tri").mosaic().rename("tri")
    )
    cti = (
        ee.ImageCollection("projects/sat-io/open-datasets/Geomorpho90m/cti").mosaic().rename("cti")
    )
    northness = (
        ee.ImageCollection("projects/sat-io/open-datasets/Geomorpho90m/northness")
        .mosaic()
        .rename("northness")
    )
    eastness = (
        ee.ImageCollection("projects/sat-io/open-datasets/Geomorpho90m/eastness")
        .mosaic()
        .rename("eastness")
    )
    sand = ee.Image(
        "projects/landler-open-data/assets/datasets/soilgrids/sand/sand_15-30cm_mean_gapfilled"
    ).rename("sand")
    clay = ee.Image(
        "projects/landler-open-data/assets/datasets/soilgrids/clay/clay_15-30cm_mean_gapfilled"
    ).rename("clay")
    ph = ee.Image(
        "projects/landler-open-data/assets/datasets/soilgrids/phh2o/phh2o_15-30cm_mean_gapfilled"
    ).rename("ph")

    # CHELSA NPP (climatological reference - summarizes climatic NPP potential)
    chelsa_npp = ee.Image(
        "projects/landler-open-data/assets/datasets/chelsa/npp/chelsa_npp_1981_2010_v2-1"
    ).rename("chelsa_npp")

    # Base predictors
    predictor_bands = [
        bio01,
        bio04,
        bio12,
        bio15,
        aridity,
        elevation,
        slope,
        tpi,
        tri,
        cti,
        northness,
        eastness,
        chelsa_npp,
        sand,
        clay,
        ph,
    ]

    if include_regional_tpi:
        dem_coarse = dem.reproject(crs="EPSG:4326", scale=500)
        regional_mean = dem_coarse.focal_mean(2000, "circle", "meters")
        tpi_regional = elevation.subtract(regional_mean).rename("tpi_regional")
        predictor_bands.append(tpi_regional)

    if include_lat_lon:
        lat_lon = ee.Image.pixelLonLat().rename(["longitude", "latitude"])
        predictor_bands.append(lat_lon)

    continuous_predictors = ee.Image.cat(predictor_bands)
    continuous_predictors = continuous_predictors.setDefaultProjection(crs="EPSG:4326", scale=90)
    continuous_predictors = continuous_predictors.reduceResolution(
        reducer=ee.Reducer.mean(), maxPixels=1024
    ).reproject(crs="EPSG:4326", scale=resolution)

    return continuous_predictors

setup_response(product='clms', year_range=NPP_YEAR_RANGE, include_std=False)

Set up the response variable(s) (observed NPP).

We model the "Average Annual NPP Sum". This is calculated by averaging the 'yearly NPP sum' assets over the provided range.

Parameters:

Name Type Description Default
product str

NPP data source ("clms" or "modis").

 'clms'
year_range list[str]

Date range [start, end]. Defaults to ["2014-01-01", "2025-01-01"].

 NPP_YEAR_RANGE
include_std bool

If True, also compute inter-annual standard deviation.

 False

Returns:

Type Description
Image

Image with "longterm_avg_npp_sum" band (representing the multi-year average of annual sums),
Image

and optionally "longterm_avg_npp_sd" band.
Source code in src/eii/compute/npp.py 
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
def setup_response(
    product: str = "clms",
    year_range: list[str] = NPP_YEAR_RANGE,
    include_std: bool = False,
) -> ee.Image:
    """
    Set up the response variable(s) (observed NPP).

    We model the "Average Annual NPP Sum".
    This is calculated by averaging the 'yearly NPP sum' assets over the provided range.

    Args:
        product: NPP data source ("clms" or "modis").
        year_range: Date range [start, end]. Defaults to ["2014-01-01", "2025-01-01"].
        include_std: If True, also compute inter-annual standard deviation.

    Returns:
        Image with "longterm_avg_npp_sum" band (representing the multi-year average of annual sums),
        and optionally "longterm_avg_npp_sd" band.
    """

    if product == "modis":
        # MOD17A3HGF is already Annual Sum
        npp_collection = (
            ee.ImageCollection("MODIS/061/MOD17A3HGF")
            .filter(ee.Filter.date(year_range[0], year_range[1]))
            .select("Npp")
        )
    else:
        from .._utils.gee import load_tiled_collection

        raw_collection = load_tiled_collection(
            "projects/landler-open-data/assets/datasets/clms/npp/annual"
        ).filter(ee.Filter.date(year_range[0], year_range[1]))

        npp_collection = raw_collection.select([0])

    npp_mean = npp_collection.mean().rename("longterm_avg_npp_sum")

    if include_std:
        if product == "modis":
            npp_std = npp_collection.reduce(ee.Reducer.stdDev()).rename("longterm_avg_npp_sd")
        else:
            npp_std = raw_collection.select([1]).mean().rename("longterm_avg_npp_sd")

        return npp_mean.addBands(npp_std)

    return npp_mean

Structural integrity

Structural integrity calculation using quality-weighted core area metrics.

Structural integrity measures landscape fragmentation AND habitat quality by: 1. Identifying core habitat (interior areas unaffected by edge effects) 2. Weighting core pixels by habitat quality class based on HMI

Quality Classes

HMI < 0.1: Pristine (weight 4) HMI 0.1-0.2: Low-impact (weight 3) HMI 0.2-0.3: Moderate (weight 2) HMI 0.3-0.4: Semi-natural (weight 1) HMI >= 0.4: Modified (weight 0)

Score interpretation

1.0 = All pristine core habitat 0.25 = All semi-natural core habitat 0.0 = No core habitat (fragmented or modified)

calculate_structural_integrity(aoi=None, edge_depth_m=DEFAULT_EDGE_DEPTH_M, neighborhood_m=DEFAULT_NEIGHBORHOOD_M, scale_m=DEFAULT_SCALE_M)

Calculate structural integrity using quality-weighted core area.

Core area is habitat that survives erosion by the edge depth. Each core pixel is weighted by its habitat quality class (pristine=4, semi-natural=1). The final score reflects both fragmentation AND habitat quality.

Parameters:

Name Type Description Default
aoi Geometry | None

Area of interest geometry. If None, returns global unclipped image suitable for tiled export pipelines.

 None
edge_depth_m int

Edge effect penetration depth in meters. Areas within this distance of habitat boundaries are considered edge-affected. Default 300m based on literature (typical range 100-500m).

 DEFAULT_EDGE_DEPTH_M
neighborhood_m int

Landscape analysis radius in meters. Default 5km balances local actionability with landscape-scale processes.

 DEFAULT_NEIGHBORHOOD_M
scale_m int

Output scale in meters for reprojection. Default 300m matches the native HMI resolution.

 DEFAULT_SCALE_M

Returns:

Type Description
Image

Structural integrity score (0-1) where:
Image
  • 1.0 = all pristine core habitat
Image
  • 0.25 = all semi-natural core habitat
Image
  • 0.0 = no core habitat (highly fragmented or modified)
Source code in src/eii/compute/structural.py 
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
def calculate_structural_integrity(
    aoi: ee.Geometry | None = None,
    edge_depth_m: int = DEFAULT_EDGE_DEPTH_M,
    neighborhood_m: int = DEFAULT_NEIGHBORHOOD_M,
    scale_m: int = DEFAULT_SCALE_M,
) -> ee.Image:
    """
    Calculate structural integrity using quality-weighted core area.

    Core area is habitat that survives erosion by the edge depth. Each core
    pixel is weighted by its habitat quality class (pristine=4, semi-natural=1).
    The final score reflects both fragmentation AND habitat quality.

    Args:
        aoi: Area of interest geometry. If None, returns global unclipped image
            suitable for tiled export pipelines.
        edge_depth_m: Edge effect penetration depth in meters. Areas within
            this distance of habitat boundaries are considered edge-affected.
            Default 300m based on literature (typical range 100-500m).
        neighborhood_m: Landscape analysis radius in meters. Default 5km
            balances local actionability with landscape-scale processes.
        scale_m: Output scale in meters for reprojection. Default 300m matches
            the native HMI resolution.

    Returns:
        Structural integrity score (0-1) where:
        - 1.0 = all pristine core habitat
        - 0.25 = all semi-natural core habitat
        - 0.0 = no core habitat (highly fragmented or modified)
    """
    hmi = _load_hmi()

    habitat_binary = hmi.lt(HMI_THRESHOLD).rename("habitat")

    core_habitat = habitat_binary.focal_min(
        radius=edge_depth_m,
        kernelType="circle",
        units="meters",
    ).rename("core")

    quality_class = _create_quality_class(hmi)

    # Weighted core is 0 for non-habitat and edge pixels, 1-4 for core based on quality
    # Do NOT mask - we want 0s to contribute to the neighborhood mean
    weighted_core = core_habitat.multiply(quality_class)

    structural_integrity = (
        weighted_core.reduceNeighborhood(
            reducer=ee.Reducer.mean(),
            kernel=ee.Kernel.circle(radius=neighborhood_m, units="meters"),
        )
        .divide(MAX_QUALITY_WEIGHT)
        .unmask(0)
        .reproject(crs="EPSG:4326", scale=scale_m)
        .rename("structural_integrity")
    )

    if aoi is not None:
        structural_integrity = structural_integrity.clip(aoi)

    return structural_integrity

Compositional integrity

Compositional Integrity Module

This module handles functionality related to calculating compositional integrity based on biodiversity metrics like the Biodiversity Intactness Index (BII).

calculate_compositional_integrity(aoi=None, year=2020, asset_path=DEFAULT_BII_ASSET_PATH)

Calculate compositional ecosystem integrity based on Biodiversity Intactness Index.

Parameters:

Name Type Description Default
aoi Geometry | None

Area of interest (optional). If None, returns unclipped global image.

 None
year int

Year to use for BII data.

 2020
asset_path str

Earth Engine ImageCollection path for BII data.

 DEFAULT_BII_ASSET_PATH

Returns:

Type Description
Image

Compositional integrity score (0-1 scale).
Source code in src/eii/compute/compositional.py 
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
def calculate_compositional_integrity(
    aoi: ee.Geometry | None = None,
    year: int = 2020,
    asset_path: str = DEFAULT_BII_ASSET_PATH,
) -> ee.Image:
    """
    Calculate compositional ecosystem integrity based on Biodiversity Intactness Index.

    Args:
        aoi: Area of interest (optional). If None, returns unclipped global image.
        year: Year to use for BII data.
        asset_path: Earth Engine ImageCollection path for BII data.

    Returns:
        Compositional integrity score (0-1 scale).
    """
    bii_collection = ee.ImageCollection(asset_path)
    start_date = f"{year}-01-01"
    end_date = f"{year}-12-31"

    year_filtered = bii_collection.filterDate(start_date, end_date)
    most_recent = bii_collection.sort("system:time_start", False).first()

    bii_image = ee.Image(
        ee.Algorithms.If(
            year_filtered.size().gt(0),
            year_filtered.mosaic(),
            most_recent,
        )
    )

    compositional_integrity = bii_image.rename("compositional_integrity")

    if aoi is not None:
        compositional_integrity = compositional_integrity.clip(aoi)

    return compositional_integrity