Tutorial: Usando as APIs do Microsoft Planetry Computer Pro para ingerir e visualizar dados

Coleções STAC (SpatioTemporal Asset Catalog) são usadas em um GeoCatalog para indexar e armazenar ativos espaciais relacionados. Neste tutorial de ponta a ponta, você criará uma nova coleção STAC, ingerirá imagens Sentinel-2 na coleção e consultará essas imagens por meio das APIs do GeoCatalog.

Neste tutorial, você:

• Crie sua própria coleção STAC dentro de um Planetary Computer Pro GeoCatalog
• Importar imagens de satélite para essa coleção da Agência Espacial Europeia
• Configure a coleção para que as imagens na coleção possam ser visualizadas na interface da Web do Planetry Computer Pro
• Consultar dados de dentro da coleção STAC usando a API STAC do Planetry Computer Pro

Este tutorial demonstra e explica as capacidades por meio de trechos de código, proporcionando uma experiência interativa no estilo de notebook. Baixe este tutorial como um notebook Jupyter.

Pré-requisitos

Antes de executar este tutorial, você precisa de um Planetary Computer Pro GeoCatalog implantado em sua assinatura do Azure. Você também precisa de um ambiente para executar este notebook e instalar os pacotes necessários. Sugerimos executar este tutorial por meio de uma Máquina Virtual do Azure Machine Learning ou da integração de blocos de anotações do Visual Studio Code em um ambiente virtual do Python. No entanto, este notebook pode ser executado em qualquer lugar que você use notebooks Jupyter, desde que sejam atendidos os seguintes requisitos.

• Python 3.10 ou posterior
• A CLI do Azure foi instalada, e você executou o comando az login para acessar sua conta do Azure
• Os requisitos necessários listados na seção Opções do Tutorial são instalados

Abra um notebook Jupyter no Azure Machine Learning ou VS Code

Faça logon no Azure com a CLI do Azure

O comando a seguir registra você no Azure usando a CLI do Azure. Execute o comando e siga as instruções para fazer logon.

!az login

Selecionar opções do tutorial

Antes de executar este tutorial, você precisa de acesso de colaborador a uma instância do GeoCatalog existente. Insira a URL da instância do GeoCatalog na variável geocatalog_url. Neste tutorial, você criará uma coleção para imagens do Sentinel-2 fornecidas pela Agência Espacial Europeia (ESA) que está atualmente armazenada no Catálogo de Dados do Computador Planetário da 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

Importar os pacotes necessários

Antes de criar uma coleção STAC, você precisa importar alguns pacotes python e definir funções auxiliares para recuperar o token de acesso necessário.

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

Criar uma coleção STAC

Definir um JSON da coleção STAC

Em seguida, você define uma coleção STAC como um item JSON. Para esse tutorial, use um JSON de coleção STAC existente para a coleção Sentinel-2-l2a no Computador Planetário da Microsoft. Sua coleção recebe uma ID aleatória e um título para não entrar em conflito com outras coleções 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

Ao criar uma coleção no GeoCatalog, um JSON de coleção não pode ter nenhum ativo de nível de coleção (como uma miniatura de coleção) associado à coleção, portanto, primeiro remova esses ativos existentes (não se preocupe, adicione a miniatura novamente mais tarde).

# 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 a interface da Web do GeoCatalog e você deverá ver sua nova coleção listada na guia "Coleções".

Miniatura da coleção de acesso

Em seguida, você deseja adicionar uma miniatura à nossa coleção a ser exibida junto com nossa coleção. Para fins dessa demonstração, use a miniatura da coleção Sentinel-2 existente no Computador Planetário da 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

Adicionar miniatura ao GeoCatalog Pro do Computador Planetário

Depois de ler a miniatura, você pode adicioná-la à nossa coleção publicando-a no ponto de extremidade da API de ativos da coleção do seu GeoCatalogs, junto com o ativo json necessário.

# 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)

Leia a nova coleção de dentro do seu GeoCatalog do Planetary Computer Pro

Atualize o navegador e você poderá ver a miniatura. Você também pode recuperar a coleção JSON programaticamente fazendo a seguinte chamada para o endpoint de coleções:

# 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
""")

Ingerir itens e ativos do STAC

Depois de criar essa coleção, você estará pronto para ingerir novos itens STAC em sua coleção STAC usando a API de Itens do GeoCatalog! Realize esse processo:

1. Obtendo um token SAS do Computador Planetário da Microsoft
2. Registrar esse token como uma fonte de ingestão no GeoCatalog
3. Enviar itens STAC dessa coleção na API de Itens do GeoCatalog
4. Verificar se os itens foram ingeridos com êxito

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)

Consultar o Computador Planetário

Primeiro, você precisa consultar o Computador Planetário para pesquisar imagens sentinelas-2 que correspondam aos nossos requisitos específicos. Nesse caso, você está procurando imagens Sentinel-2 no Computador Planetário que correspondam aos seguintes critérios:

• Coleção - Imagens da coleção Sentinel-2-l2a
• Intervalo de tempo – Coletado entre 4 de fevereiro e 11 de fevereiro de 2024
• Área de interesse - Imagens coletadas no sul da Islândia (definidas como uma caixa delimitadora)

Ao executar essa pesquisa, você pode ver que os itens STAC correspondentes são encontrados no Computador Planetário.

# 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]

Registrar uma fonte de ingestão

Antes de ingerir esses itens STAC e seus ativos relacionados (imagens) em uma coleção GeoCatalog, você precisa determinar se precisa registrar uma nova fonte de ingestão. As fontes de ingestão são usadas pelo GeoCatalog para monitorar os locais de armazenamento (contêineres do Armazenamento de Blobs do Azure) aos quais ele tem acesso.

O registro de uma fonte de ingestão é feito fornecendo ao GeoCatalog o local do contêiner de armazenamento e um token SAS com permissões de leitura para acessar o contêiner. Se os itens STAC ou seus ativos relacionados estiverem localizados em um contêiner de armazenamento e seu GeoCatalog não tiver recebido acesso, a ingestão falhará.

Para iniciar esse processo, primeiro você solicita um token SAS do Computador Planetário que nos concede acesso de leitura ao contêiner em que residem as imagens do 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']}")

Em seguida, tente registrar esse contêiner do Armazenamento de Blobs do Azure e o token SAS associado como uma fonte de ingestão com o GeoCatalog. Há o potencial de que uma fonte de ingestão já exista para esse contêiner de armazenamento. Nesse caso, localize a ID da fonte de ingestão existente.

Aviso

Se uma fonte de ingestão duplicada for encontrada com um token que expira nos próximos 15 minutos, ela será excluída e substituída. Excluir uma fonte de ingestão que esteja em uso pelas ingestões em execução no momento pode interromper essas ingestões.

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"])

Ingerir itens STAC usando a API de itens do GeoCatalog

Agora que você registrou uma fonte de ingestão ou validou que existe uma fonte, você ingerirá os itens STAC encontrados no Computador Planetário usando a API de Itens do GeoCatalog. Faça isso postando cada item na API de Itens que cria uma nova operação de ingestão no 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']}")

Considerando que a ingestão de itens do Sentinel-2 pode levar um pouco de tempo, você pode executar esse código para verificar o status das operações de ingestão usando a API de Operações do 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)

Você deve ser capaz de atualizar seu navegador da Web e clicar na guia Itens para ver esses itens recém-carregados.

Gestão de coleções

Agora que você ingeriu esses itens STAC e seus ativos associados (imagens) na coleção STAC, é necessário fornecer ao GeoCatalog alguns outros arquivos de configuração antes de poder visualizar esses itens na interface da Web do GeoCatalog.

Configuração de renderização da coleção

Primeiro baixe um arquivo de configuração de renderização para esta coleção do Computador Planetário. Esse arquivo de configuração pode ser lido pelo GeoCatalog para renderizar imagens de diferentes maneiras dentro do Explorer. Isso ocorre porque os itens STAC podem conter muitos ativos diferentes (imagens) que podem ser combinados para criar imagens totalmente novas de uma determinada área que realçam recursos visíveis ou não visíveis. Por exemplo, os itens STAC sentinel-2 têm mais de 12 imagens diferentes de diferentes partes do espectro eletromagnético. Essa configuração de renderização instrui o GeoCatalog sobre como combinar essas imagens para que ela possa exibir imagens em Cor Natural ou Cor Falsa (Cor Infravermelha).

# 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']

Depois de ler esta configuração de opções de renderização do Planetary Computer, você pode habilitar essas opções de renderização para a coleção enviando esta configuração para o endpoint de opções de renderização.

# 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}
    )

Definições de mosaico

Semelhante à Configuração de Renderização discutida acima, o Explorador do GeoCatalog nos permite especificar uma ou mais definições de mosaico para a coleção. Essas definições de mosaico nos permitem instruir o Explorador do GeoCatalog sobre como filtrar quais itens são exibidos no Explorer. Por exemplo, uma configuração de renderização básica (mostrada na próxima célula) instrui o GeoCatalog a exibir a imagem mais recente para qualquer área específica. Configurações de renderização mais avançadas nos permitem renderizar exibições diferentes, como a imagem menos nublada para um determinado local capturado em outubro 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 a interface da Web do GeoCatalog

Parabéns! Você criou uma coleção, adicionou itens e ativos stac e atualizou sua coleção para incluir os arquivos de configuração necessários para que ela possa ser exibida por meio do Explorer dentro da interface da Web GeoCatalog.

Navegue de volta para o GeoCatalog Explorer na interface da Web para exibir sua coleção!

Coleção de consultas por meio da API STAC

Agora que viu sua coleção no Explorador do GeoCatalog, você aprenderá a usar as APIs STAC do GeoCatalog para pesquisar e recuperar itens e ativos STAC para análise posterior.

Esse processo começa postando uma pesquisa na API STAC do GeoCatalog. Mais especificamente, você pesquisará imagens em sua coleção que estejam dentro da caixa delimitadora original usada para extrair imagens do Computador Planetário.

Sem surpresa, essa consulta retorna todos os itens STAC que você colocou anteriormente em sua coleção.

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))

Na consulta anterior, você também forneceu outro parâmetro: sign:true. Isso instrui o GeoCatalog a retornar um href assinado (item href + token SAS) que permite que você leia os ativos fornecidos do Armazenamento de Blobs do Azure.

# 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

Limpar os recursos

Excluir itens

Neste ponto, você criou uma Coleção GeoCatalog, adicionou itens e ativos à coleção e recuperou esses itens e ativos usando a API STAC da GeoCatalog. Para a fase final deste tutorial, você removerá esses itens e excluirá sua coleção.

# Delete all items

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

Você pode confirmar que todos os itens foram excluídos executando o próximo comando. Observe que pode levar um ou dois minutos para excluir totalmente os itens e seus ativos associados.

# 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))

Excluir coleção

Agora, como uma etapa final, talvez você queira excluir totalmente sua coleção da instância do 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}")

Próximas etapas

Criar uma coleção STAC por meio da interface da Web

Conteúdo relacionado

Neste tutorial de ponta a ponta, você percorreu o processo de criação de uma nova coleção STAC, ingestão de imagens do Sentinel-2 na coleção e fazendo consultas dessas imagens por meio das APIs do GeoCatalog. Se você quiser saber mais sobre cada um desses tópicos, explore estes outros materiais:

• Criar um GeoCatalog
• Criar um conjunto
• Ingerir itens STAC
• Criar uma configuração de renderização
• Configurar as configurações do bloco de coleta
• Configuração do Mosaico
• Configuração de consultas