Tutorial: Uso de las API de Microsoft Planetary Computer Pro para ingerir y visualizar datos

Las colecciones de STAC (Catálogo de activos spatiotemporales) se usan dentro de geocatalog para indexar y almacenar activos spatiotemporales relacionados. En este tutorial completo, crearás una nueva colección de STAC, introducirás imágenes de Sentinel-2 en la colección e interrogarás esas imágenes a través de las API de GeoCatalog.

En este tutorial, usted hará lo siguiente:

• Creará su propia colección de STAC dentro de un geocatálogo de Planetary Computer Pro.
• Ingesta de imágenes satélite en esa colección de la Agencia Espacial Europea
• Configure la colección para que las imágenes de la colección se puedan visualizar en la interfaz web de Planetary Computer Pro.
• Consulta de datos desde la colección STAC mediante la API de STAC de Planetary Computer Pro

En este tutorial se muestran y explican las funcionalidades a través de fragmentos de código, para obtener una experiencia interactiva de estilo de cuaderno, descargue este tutorial como un cuaderno de Jupyter Notebook.

Prerrequisitos

Antes de ejecutar este tutorial, necesita una instancia de Planetary Computer Pro GeoCatalog implementada en su suscripción de Azure. También necesita un entorno para ejecutar este cuaderno e instalar los paquetes necesarios. Se recomienda ejecutar este tutorial a través de una máquina virtual de Azure Machine Learning o la integración de cuadernos de Visual Studio Code en un entorno virtual de Python. Sin embargo, este cuaderno debe ejecutarse siempre que pueda ejecutar cuadernos de Jupyter Notebook, siempre que se cumplan los siguientes requisitos:

• Python 3.10 o posterior
• La CLI de Azure está instalada y ha ejecutado az login para iniciar sesión en su cuenta de Azure.
• Se instalan los requisitos necesarios enumerados en la sección Opciones del tutorial.

Apertura de un cuaderno de Jupyter Notebook en Azure Machine Learning o VS Code

Inicio de sesión en Azure con la CLI de Azure

El siguiente comando le registra en Azure mediante la CLI de Azure. Ejecute el comando y siga las instrucciones para iniciar sesión.

!az login

Selección de las opciones del tutorial

Antes de ejecutar este tutorial, necesita acceso de colaborador a una instancia de GeoCatalog existente. Escriba la dirección URL de la instancia de GeoCatalog en la variable geocatalog_url. En este tutorial, crearás una colección para las imágenes del Sentinel-2 proporcionadas por la Agencia Espacial Europea (ESA) que se almacenan actualmente en el Catálogo de Datos del Ordenador Planetario de Microsoft.

# URL for your given GeoCatalog
geocatalog_url = (
    "<GEOCATALOG_URL>"
)
geocatalog_url = geocatalog_url.rstrip("/")  # Remove trailing slash if present

api_version = "2025-04-30-preview"

# User selections for demo

# Collection within the Planetary Computer
pc_collection = "sentinel-2-l2a"

# Bounding box for AOI
bbox_aoi = [-22.455626, 63.834083, -22.395201, 63.880750]

# Date range to search for imagery
param_date_range = "2024-02-04/2024-02-11"

# Maximum number of items to ingest
param_max_items = 6

Importación de los paquetes necesarios

Para poder crear una colección de STAC, debe importar algunos paquetes de Python y definir funciones auxiliares para recuperar el token de acceso necesario.

pip install pystac-client azure-identity requests pillow

# Import the required packages
import json
import random
import string
import time
from datetime import datetime, timedelta, timezone
from io import BytesIO
from typing import Any, Optional, Dict

import requests
from azure.identity import AzureCliCredential
from IPython.display import Markdown as md
from IPython.display import clear_output
from PIL import Image
from pystac_client import Client

# Function to get a bearer token for the  Planetary Computer Pro API
MPC_APP_ID = "https://geocatalog.spatio.azure.com"

_access_token = None
def getBearerToken():
    global _access_token
    if not _access_token or datetime.fromtimestamp(_access_token.expires_on) < datetime.now() + timedelta(minutes=5):
        credential = AzureCliCredential()
        _access_token = credential.get_token(f"{MPC_APP_ID}/.default")

    return {"Authorization": f"Bearer {_access_token.token}"}

# Method to print error messages when checking response status
def raise_for_status(r: requests.Response) -> None:
    try:
        r.raise_for_status()
    except requests.exceptions.HTTPError as e:
        try:
            print(json.dumps(r.json(), indent=2))
        except:
            print(r.content)
        finally:
            raise

Creación de una colección STAC

Definición de un JSON de colección STAC

A continuación, defina una colección STAC como un elemento JSON. Para este tutorial, use un JSON de colección STAC existente para la colección Sentinel-2-l2a dentro del Equipo planetario de Microsoft. A la colección se le asigna un identificador aleatorio y un título para no entrar en conflicto con otras colecciones existentes.

# Load example STAC collection JSON

response = requests.get(
    f"https://planetarycomputer.microsoft.com/api/stac/v1/collections/{pc_collection}"
)
raise_for_status(response)
stac_collection = response.json()

collection_id = pc_collection + "-tutorial-" + str(random.randint(0, 1000))

# Genereate a unique name for the test collection
stac_collection["id"] = collection_id
stac_collection["title"] = collection_id

# Determine the storage account and container for the assets
pc_storage_account = stac_collection.pop("msft:storage_account")
pc_storage_container = stac_collection.pop("msft:container")
pc_collection_asset_container = (
    f"https://{pc_storage_account}.blob.core.windows.net/{pc_storage_container}"
)

# View your STAC collection JSON
stac_collection

Al crear una colección dentro de GeoCatalog, un JSON de colección no puede tener ningún recurso de nivel de colección (como una miniatura de colección) asociado a la colección, por lo que primero quite esos recursos existentes (no se preocupe de agregar la miniatura más adelante).

# Save the thumbnail url
thumbnail_url = stac_collection['assets']['thumbnail']['href']

# Remove the assets field from the JSON (you'll see how to add this back later)
print("Removed the following items from the STAC Collection JSON:")
stac_collection.pop('assets')

# Create a STAC collection by posting to the STAC collections API

collections_endpoint = f"{geocatalog_url}/stac/collections"

response = requests.post(
    collections_endpoint,
    json=stac_collection,
    headers=getBearerToken(),
    params={"api-version": api_version}
)

if response.status_code==201:
    print("STAC Collection created named:",stac_collection['title'])
else:
    raise_for_status(response)

Abra la interfaz web de GeoCatalog y debería ver la nueva colección en la pestaña "Colecciones".

Acceder a la miniatura de la colección

A continuación, quiere agregar una miniatura a nuestra colección para que se muestre junto con nuestra colección. Para los fines de esta demostración, use la miniatura de la colección Sentinel-2 existente dentro del Planetary Computer de Microsoft.

# Read thumbnail for your collection

thumbnail_response = requests.get(thumbnail_url)
raise_for_status(thumbnail_response)
img = Image.open(BytesIO(thumbnail_response.content))
img

Añade miniatura a tu Planetary Computer Pro GeoCatalog

Después de leer la miniatura, puede agregarla a nuestra colección publicándola en el punto de conexión de la API de recursos de recopilación de GeoCatalogs junto con el JSON del recurso necesario.

# Define the GeoCatalog collections API endpoint
collection_assets_endpoint = f"{geocatalog_url}/stac/collections/{collection_id}/assets"

# Read the example thumbnail from this collection from the Planetary Computer
thumbnail = {"file": ("lulc.png", thumbnail_response.content)}

# Define the STAC collection asset type - thumbnail in this case
asset = {
    "data": '{"key": "thumbnail", "href":"", "type": "image/png", '
    '"roles":  ["test_asset"], "title": "test_asset"}'
}

# Post the thumbnail to the GeoCatalog collections asset endpoint
response = requests.post(
    collection_assets_endpoint,
    data=asset,
    files=thumbnail,
    headers=getBearerToken(),
    params={"api-version": api_version}
)

if response.status_code==201:
    print("STAC Collection thumbnail updated for:",stac_collection['title'])
else:
    raise_for_status(response)

Lee la nueva colección desde tu geocatalog de Planetary Computer Pro

Actualice el explorador y debería poder ver la miniatura. También puede recuperar el JSON de la colección de forma programática realizando la siguiente llamada al endpoint de colecciones:

# Request the collection JSON from your GeoCatalog
collection_endpoint = f"{geocatalog_url}/stac/collections/{stac_collection['id']}"

response = requests.get(
    collection_endpoint,
    json={'collection_id':stac_collection['id']},
    headers=getBearerToken(),
    params={"api-version": api_version}
)

if response.status_code==200:
    print("STAC Collection successfully read:",stac_collection['title'])
else:
    raise_for_status(response)

response.json()

print(f"""
You successfully created a new STAC Collection in GeoCatalog named {collection_id}.
You can view your collection by visiting the GeoCatalog Explorer: {geocatalog_url}/collections
""")

Ingesta de elementos y activos de STAC

Después de crear esta colección, estará listo para ingerir nuevos elementos de STAC en la colección de STAC mediante la API de elementos de GeoCatalog. Realice este proceso mediante:

1. Obtención de un token de SAS del equipo planetario de Microsoft
2. Registro de ese token como origen de ingesta en GeoCatalog
3. Publicar elementos de STAC de esa colección en la API item de GeoCatalog
4. Comprobación de que los elementos se han ingerido correctamente

ingestion_sources_endpoint = f"{geocatalog_url}/inma/ingestion-sources"
ingestion_source_endpoint = lambda id: f"{geocatalog_url}/inma/ingestion-sources/{id}"


def find_ingestion_source(container_url: str) -> Optional[Dict[str, Any]]:

    response = requests.get(
        ingestion_sources_endpoint,
        headers=getBearerToken(),
        params={"api-version": api_version},
    )

    for source in response.json()["value"]:
        ingestion_source_id = source["id"]

        response = requests.get(
            ingestion_source_endpoint(ingestion_source_id),
            headers=getBearerToken(),
            params={"api-version": api_version},
        )
        raise_for_status(response)

        response = response.json()

        if response["connectionInfo"]["containerUrl"] == container_url:
            return response


def create_ingestion_source(container_url: str, sas_token: str):
    response = requests.post(
        ingestion_sources_endpoint,
        json={
            "kind": "SasToken",
            "connectionInfo": {
                "containerUrl": container_url,
                "sasToken": sas_token,
            },
        },
        headers=getBearerToken(),
        params={"api-version": api_version},
    )
    raise_for_status(response)


def remove_ingestion_source(ingestion_source_id: str):
    response = requests.delete(
        ingestion_source_endpoint(ingestion_source_id),
        headers=getBearerToken(),
        params={"api-version": api_version},
    )
    raise_for_status(response)

Consulta el ordenador planetario

En primer lugar, debe consultar la Computadora Planetaria para buscar imágenes de Sentinel-2 que coincidan con nuestros requisitos específicos. En este caso, busca imágenes de Sentinel-2 en la Planetary Computer que coincidan con los criterios siguientes.

• Colección: imágenes de la colección Sentinel-2-l2a
• Intervalo de tiempo: recopilado entre el 4 de febrero y el 11 de febrero de 2024
• Área de interés: Imágenes recopiladas sobre el sur de Islandia (definida como un rectángulo delimitador)

Al realizar esta búsqueda, puede ver que los elementos de STAC coincidentes se encuentran dentro del Planetary Computer.

# Search criteria
print("Using the below parameters to search the Planetary Computer:\n")
print("Collection:", pc_collection)
print("Bounding box for area of interest:",bbox_aoi)
print("Date range:",param_date_range)
print("Max number of items:",param_max_items)

# Query the Planetary Computer

# Connect to the Planetary Computer
catalog = Client.open("https://planetarycomputer.microsoft.com/api/stac/v1")

search = catalog.search(collections=[pc_collection], bbox=bbox_aoi, datetime=param_date_range)
total_items = search.item_collection()

items = total_items[:param_max_items]
print("Total number of matching items:",len(total_items))
print("Total number of items for ingest base on user selected parameter:",len(items))

if total_items==0:
    print("No items matched your user specified parameters used at the top of this demo. Update these parameters")

# Print an example STAC item returned by the Planetary Computer
items[0]

Registro de un origen de ingesta

Para poder ingerir estos elementos de STAC y sus recursos relacionados (imágenes) en una colección GeoCatalog, debe determinar si necesita registrar un nuevo origen de ingesta. GeoCatalog usa orígenes de ingesta para realizar un seguimiento de las ubicaciones de almacenamiento (contenedores de Azure Blob Storage) a las que tiene acceso.

Para registrar un origen de ingesta, se proporciona GeoCatalog la ubicación del contenedor de almacenamiento y un token de SAS con permisos de lectura para acceder al contenedor. Si los elementos de STAC o sus recursos relacionados se encuentran en un contenedor de almacenamiento, el GeoCatalog no ha recibido acceso a la ingesta producirá un error.

Para iniciar este proceso, primero solicita un token de SAS desde el equipo planetario que nos concede acceso de lectura al contenedor donde residen las imágenes de Sentinel-2.

# Request API token from the Planetary Computer

pc_token = requests.get("https://planetarycomputer.microsoft.com/api/sas/v1/token/{}".format(pc_collection)).json()
print(f"Planetary Computer API Token will expire {pc_token['msft:expiry']}")

A continuación, intente registrar este contenedor de Azure Blob Storage y el token de SAS asociado como origen de ingesta con GeoCatalog. Existe el potencial de que ya exista un origen de ingesta para este contenedor de almacenamiento. Si es así, busque el identificador del origen de ingesta existente.

Advertencia

Si se encuentra un origen de ingesta duplicado con un token que vaya a expirar en los próximos 15 minutos, se elimina y se reemplaza por uno nuevo. La eliminación de un origen de ingesta que está en uso mediante la ejecución de ingestas actualmente puede interrumpir esas ingestas.

existing_ingestion_source: Optional[Dict[str, Any]] = find_ingestion_source(pc_collection_asset_container)

if existing_ingestion_source:
    connection_info = existing_ingestion_source["connectionInfo"]
    expiration = datetime.fromisoformat(connection_info["expiration"].split('.')[0]) # works in all Python 3.X versions
    expiration = expiration.replace(tzinfo=timezone.utc) # set timezone to UTC
    if expiration < datetime.now(tz=timezone.utc) + timedelta(minutes=15):
        print(f"Recreating existing ingestion source for {pc_collection_asset_container}")
        remove_ingestion_source(existing_ingestion_source["id"])
        create_ingestion_source(pc_collection_asset_container, pc_token["token"])
    else:
        print(f"Using existing ingestion source for {pc_collection_asset_container} with expiration {expiration}")
else:
    print(f"Creating ingestion source for {pc_collection_asset_container}")
    create_ingestion_source(pc_collection_asset_container, pc_token["token"])

Ingesta de elementos de STAC mediante la API de elementos de GeoCatalog

Ahora que registró un origen de ingesta o validó que existe un origen, ingerirá los elementos de STAC que encontró en el equipo planetario mediante la API de elementos de GeoCatalog. Para ello, publique cada elemento en la API Items, que crea una nueva operación de ingesta en GeoCatalog.

# Ingest items

items_endpoint = f"{geocatalog_url}/stac/collections/{collection_id}/items"

operation_ids = []

for item in items:

    item_json = item.to_dict()
    item_json['collection'] = collection_id

    # Remove non-static assets
    del(item_json['assets']['rendered_preview'])
    del(item_json['assets']['preview'])
    del(item_json['assets']['tilejson'])

    response = requests.post(
        items_endpoint,
        json=item_json,
        headers=getBearerToken(),
        params={"api-version": api_version}
    )

    operation_ids.append(response.json()['id'])
    print(f"Ingesting item {item_json['id']} with operation id {response.json()['id']}")

Dado que la ingesta de elementos de Sentinel-2 puede tardar un poco, puede ejecutar este código para comprobar el estado de las operaciones de ingesta mediante la API de operaciones de GeoCatalog.

# Check the status of the operations
operations_endpoint = f"{geocatalog_url}/inma/operations"
# Loop through all the operations ids until the status of each operation ids is "Finished"
pending=True

start = time.time()

while pending:
    # Count the number of operation ids that are finished vs unfinished
    num_running = 0
    num_finished = 0
    num_failed = 0
    clear_output(wait=True)
    for operation_id in operation_ids:
        response = requests.get(
            f"{operations_endpoint}/{operation_id}",
            headers=getBearerToken(),
            params={"api-version": api_version},
        )
        raise_for_status(response)
        status = response.json()["status"]
        print(f"Operation id {operation_id} status: {status}")
        if status == "Running":
            num_running+=1
        elif status == "Failed":
            num_failed+=1
        elif status == "Succeeded":
            num_finished+=1
    
    num_running
    stop=time.time()
    # Print the sumary of num finished, num running and num failed
    
    print("Ingesting Imagery:")
    print(f"\tFinished: {num_finished}\n\tRunning: {num_running}\n\tFailed: {num_failed}")
    print("Time Elapsed (seconds):",str(stop-start))
    
    if num_running == 0:
        pending=False
        print(f"Ingestion Complete!\n\t{num_finished} items ingested.\n\t{num_failed} items failed.")

    else:
        print(f"Waiting for {num_running} operations to finish")
        time.sleep(5)

Debería poder actualizar el explorador web y hacer clic en la pestaña Elementos para ver estos elementos recién cargados.

Administración de colecciones

Ahora que ha ingerido estos elementos de STAC y sus recursos asociados (imágenes) en la colección STAC, debe proporcionar a GeoCatalog algunos archivos de configuración adicionales antes de poder visualizar estos elementos en la interfaz web de GeoCatalog.

Configuración de representación de recopilación

Primero, descargue un archivo de configuración de renderizado para esta colección desde la Computadora Planetaria. GeoCatalog puede leer este archivo de configuración para representar imágenes de diferentes maneras dentro del Explorador. Esto se debe a que los elementos de STAC pueden contener muchos recursos diferentes (imágenes) que se pueden combinar para crear imágenes completamente nuevas de un área determinada que resalte características visibles o no visibles. Por ejemplo, los elementos STAC de Sentinel-2 tienen más de 12 imágenes diferentes de diferentes partes del espectro electromagnética. Esta configuración de representación indica a GeoCatalog cómo combinar estas imágenes para que pueda mostrar imágenes en color natural o color falso (infrarrojo de color).

# Read render JSON from Planetary Computer

render_json = requests.get("https://planetarycomputer.microsoft.com/api/data/v1/mosaic/info?collection={}".format(pc_collection)).json()
render_json['renderOptions']

Después de leer esta configuración de opciones de representación desde la Computadora Planetaria, puede habilitar estas opciones de representación para la colección publicando esta configuración en el endpoint de opciones de representación.

# Post render options config to GeoCatalog render-options API

render_config_endpoint = f"{geocatalog_url}/stac/collections/{collection_id}/configurations/render-options"

for render_option in render_json['renderOptions']:

    # Rename render configs such that they can be stored by GeoCatalog
    render_option['id'] = render_option['name'].translate(str.maketrans('', '', string.punctuation)).lower().replace(" ","-")[:30]

    # Post render definition
    response = requests.post(
        render_config_endpoint,
        json=render_option,
        headers=getBearerToken(),
        params={"api-version": api_version}
    )

Definiciones de mosaico

De forma similar a la configuración de representación descrita anteriormente, el Explorador de GeoCatalog nos permite especificar una o varias definiciones de mosaico para la colección. Estas definiciones de mosaico nos permiten indicar al Explorador de GeoCatalog cómo filtrar qué elementos se muestran en el Explorador. Por ejemplo, una configuración básica de representación (que se muestra en la celda siguiente) indica a GeoCatalog que muestre la imagen más reciente de cualquier área determinada. Las configuraciones de representación más avanzadas nos permiten representar diferentes vistas, como la imagen menos nublada para una ubicación determinada capturada en octubre de 2023.

# Post mosaic definition

mosiacs_config_endpoint = f"{geocatalog_url}/stac/collections/{collection_id}/configurations/mosaics"

response = requests.post(
    mosiacs_config_endpoint,
    json={"id": "mos1",
          "name": "Most recent available",
          "description": "Most recent available imagery in this collection",
          "cql": []
    },
    headers=getBearerToken(),
    params={"api-version": api_version}
)

Abrir la interfaz web del GeoCatálogo

¡Enhorabuena! Ha creado una colección, ha agregado elementos y recursos de STAC y ha actualizado la colección para incluir los archivos de configuración necesarios para que se pueda ver a través del Explorador dentro de la interfaz web geocatalog.

Vuelva al Explorador GeoCatalog en la interfaz web para ver la colección.

Recopilación de consultas a través de la API de STAC

Ahora que ha visto su colección en el Explorador de GeoCatalog, vamos a guiarle a través de cómo usar las APIs de STAC de GeoCatalog para buscar y recuperar elementos y recursos de STAC para un análisis más detallado.

Este proceso comienza publicando una búsqueda en la API de STAC de GeoCatalog. En concreto, buscará imágenes dentro de la colección que se encuentra dentro del rectángulo de selección original que usó para extraer imágenes del equipo planetario.

De forma inesperada, esta consulta devuelve todos los elementos de STAC que ha colocado anteriormente dentro de la colección.

stac_search_endpoint = f"{geocatalog_url}/stac/search"

response = requests.post(
    stac_search_endpoint,
    json={"collections":[collection_id],
          "bbox":bbox_aoi
    },
    headers=getBearerToken(),
    params={"api-version": api_version, "sign": "true"}
)

matching_items = response.json()['features']
print(len(matching_items))

En la consulta anterior, también proporcionó otro parámetro: sign:true. Esto indica a GeoCatalog que devuelva un href firmado (elemento href + token de SAS) que le permite leer los recursos especificados de Azure Blob Storage.

# Download one of the assets bands, band 09
asset_href = matching_items[0]['assets']['B09']['href']
print(asset_href)

response = requests.get(asset_href)
img = Image.open(BytesIO(response.content))
img

Limpieza de recursos

Eliminar elementos

En este momento, ha creado una colección GeoCatalog, ha agregado elementos y recursos a la colección y ha recuperado esos elementos y recursos mediante la API de STAC de GeoCatalog. Para la fase final de este tutorial, va a quitar estos elementos y eliminar la colección.

# Delete all items

for item in matching_items:
    response = requests.delete(
        f"{items_endpoint}/{item['id']}",
        headers=getBearerToken(),
        params={"api-version": api_version}
    )

Puede confirmar que todos los elementos se eliminaron ejecutando el comando siguiente. Tenga en cuenta que puede tardar un minuto o dos en eliminar completamente los elementos y sus recursos asociados.

# Confirm that all the items have been deleted
response = requests.post(
    stac_search_endpoint,
    json={"collections":[stac_collection['id']],
          "bbox": bbox_aoi
    },
    headers=getBearerToken(),
    params={"api-version": api_version, "sign": "true"}
)

matching_items = response.json()['features']
print(len(matching_items))

Eliminar colección

Ahora como paso final, es posible que quiera eliminar completamente la colección de la instancia de GeoCatalog.

# Delete the collection
response = requests.delete(
    f"{collections_endpoint}/{collection_id}",
    headers=getBearerToken(),
    params={"api-version": api_version}
)

raise_for_status(response)
print(f"STAC Collection deleted: {collection_id}")

Pasos siguientes

Creación de una colección de STAC a través de la interfaz web

Contenido relacionado

En este tutorial de un extremo a otro, ha recorrido el proceso de creación de una nueva colección de STAC, la ingesta de imágenes de Sentinel-2 en la colección y la consulta de esas imágenes a través de las API de GeoCatalog. Si desea obtener más información sobre cada uno de estos temas, explore estos otros materiales:

• Creación de un geocatalog
• Creación de una colección
• Ingesta de elementos de STAC
• Crear una configuración de representación
• Configurar la configuración de azulejos de colección
• Configuración de mosaico
• Configuración de Elementos Consultables