Planetary Computer

Fused loads data from the Microsoft Planetary Computer data catalog with the Python pystac_client library.

@fused.udf
def udf(bbox: fused.types.TileGDF=None, time_of_interest="2023-11-01/2023-12-30"):

    import odc.stac
    import planetary_computer
    import pystac_client

    utils = fused.load(
        "https://github.com/fusedio/udfs/tree/f928ee1/public/common/"
    ).utils

    red_band = "B04"
    nir_band = "B08"
    catalog = pystac_client.Client.open(
        "https://planetarycomputer.microsoft.com/api/stac/v1",
        modifier=planetary_computer.sign_inplace,
    )

    items = catalog.search(
        collections=["sentinel-2-l2a"],
        bbox=bbox.total_bounds,
        datetime=time_of_interest,
        query={"eo:cloud_cover": {"lt": 10}},
    ).item_collection()

    resolution = int(5 * 2 ** (15 - bbox.z[0]))
    ds = odc.stac.load(
        items,
        crs="EPSG:3857",
        bands=[nir_band, red_band],
        resolution=resolution,
        bbox=bbox.total_bounds,
    ).astype(float)

    # Calculate NDVI
    ndvi = (ds[nir_band] - ds[red_band]) / (ds[nir_band] + ds[red_band])

    arr = ndvi.max(dim="time")
    return utils.arr_to_plasma(arr.values, min_max=(0, 0.8), reverse=False)