Partilhar via

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

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

Neste tutorial, você:

  • Criará a sua própria coleção STAC dentro de um GeoCatálogo do Planetary Computer Pro
  • Integrar imagens de satélite da Agência Espacial Europeia nessa coleção
  • Configure a coleção para que as imagens da coleção possam ser visualizadas na interface web do Planetary Computer Pro
  • Consultar dados de dentro da coleção STAC usando a API STAC do Planetary Computer Pro

Este tutorial mostra e explica os recursos por meio de trechos de código, para uma experiência interativa no estilo de notebook, baixe este tutorial como um bloco de anotações Jupyter.

Pré-requisitos

Antes de executar este tutorial, você precisa de um GeoCatalog do Planetary Computer Pro 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 do bloco de anotações do Visual Studio Code em um ambiente virtual Python. No entanto, este notebook deve ser executado onde quer que se possam executar notebooks Jupyter, desde que os seguintes requisitos sejam atendidos.

  • Python 3.10 ou posterior
  • A CLI do Azure está instalada e você executou az login para fazer logon em sua conta do Azure
  • Os requisitos necessários listados na seção Opções do tutorial estão instalados

Abrir um bloco de anotações Jupyter no Azure Machine Learning ou VS Code

Faça logon no Azure com a CLI do Azure

O comando a seguir faz logon no Azure usando a CLI do Azure. Execute o comando e siga as instruções para iniciar sessão.

!az login

Selecionar opções do tutorial

Antes de executar este tutorial, você precisa de acesso de colaborador a uma instância existente do GeoCatalog. Insira a url da sua instância GeoCatalog na variável geocatalog_url. Neste tutorial, você criará uma coleção de imagens do Sentinel-2 fornecidas pela Agência Espacial Europeia (ESA) que está atualmente armazenada no Planetary Computer Data Catalog 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 uma coleção STAC JSON

Em seguida, você define uma coleção STAC como um item JSON. Para este tutorial, use uma coleção STAC JSON existente para a coleção Sentinel-2-l2a no Planetary Computer da Microsoft. A sua coleção recebe um ID e um título aleatórios 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 em adicionar a miniatura de volta 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 web do GeoCatalog e deverá ver a sua nova coleção listada no separador "Coleções".

Miniatura da coleção do Access

Em seguida, você deseja adicionar uma miniatura à nossa coleção para ser exibida junto com a nossa coleção. Para os fins desta demonstração, use a miniatura da coleção Sentinel-2 existente no Planetary Computer 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

Adicione uma miniatura ao seu Planetary Computer Pro GeoCatálogo

Depois de ler a miniatura, você pode adicioná-la à nossa coleção publicando-a no endpoint da API de ativos de coleção do GeoCatalogs junto com o json de ativos 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 a partir do seu GeoCatálogo Planetary Computer Pro

Atualize seu navegador e você deve ser capaz de ver a miniatura. Você também pode recuperar o JSON da coleção programaticamente fazendo a seguinte chamada ao 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 STAC e recursos

Depois de criar esta coleção, você está pronto para ingerir novos itens STAC em sua coleção STAC usando a API de itens do seu GeoCatalog! Realize este processo ao:

  1. Obtendo um token SAS do Planetary Computer da Microsoft
  2. Registre esse token como uma fonte de ingestão no GeoCatalog
  3. Postar itens STAC dessa coleção na API de itens do GeoCatalog
  4. Verifique 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 procurar imagens do Sentinel-2 que correspondam aos nossos requisitos específicos. Neste caso, está à procura de imagens do Sentinel-2 no Computador Planetário que correspondam aos seguintes critérios:

  • Coleção - Imagens da coleção Sentinel-2-l2a
  • Intervalo de tempo - Recolhido entre 4 e 11 de fevereiro de 2024
  • Área de interesse - Imagens recolhidas sobre o sul da Islândia (definida como uma caixa delimitadora)

Ao realizar esta pesquisa, pode ver que os itens STAC correspondentes são encontrados dentro do 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]

Registar uma fonte de ingestão

Antes de poder 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 rastrear quais locais de armazenamento (contêineres de Azure Blob Storage) a que tem acesso.

O registro de uma fonte de ingestão é realizado 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 os seus ativos relacionados estiverem localizados num contentor de armazenamento ao qual o seu GeoCatalog não teve acesso, a ingestão irá falhar.

Para iniciar esse processo, você primeiro solicita um token SAS do Computador Planetário que nos concede acesso de leitura ao contêiner onde as imagens do Sentinel-2 residem.

# 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 de Armazenamento de Blob do Azure e o token SAS associado como uma fonte de ingestão com o GeoCatalog. Existe o potencial de que já exista uma fonte de ingestão para este recipiente de armazenamento. Em caso afirmativo, localize o ID da fonte de ingestão existente.

Advertência

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 a ser utilizada por ingestões atualmente em execução 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 uma fonte existe, você ingerirá os itens STAC que encontrou 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']}")

Como a ingestão de itens do Sentinel-2 pode levar um pouco de tempo, você pode executar esse código para verificar o status de suas 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 clique na guia Itens para ver esses itens recém-carregados.

Gestão de cobranças

Agora que ingeriu estes itens STAC e os seus ativos associados (imagens) na coleção STAC, precisa de fornecer ao seu GeoCatálogo alguns outros ficheiros de configuração antes de poder visualizar estes itens na interface web do GeoCatálogo.

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

Primeiro, baixe um arquivo de configuração de renderização para esta coleção do Planetary Computer. Este 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 destacam recursos visíveis ou não visíveis. Por exemplo, os itens STAC do Sentinel-2 têm mais de 12 imagens diferentes de diferentes porções do espectro eletromagnético. Esta configuração de renderização instrui o GeoCatalog sobre como combinar essas imagens para que ele 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 Computador Planetário, pode ativar estas opções de renderização para a coleção ao enviar esta configuração para o ponto final 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 ao Render Config discutido acima, o Explorer 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 GeoCatalog's Explorer 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. Configurações de renderização mais avançadas nos permitem renderizar diferentes visualizações, como a imagem menos nublada para um determinado local capturada 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 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 possa ser visualizada através do Explorer na interface da Web GeoCatalog.

Navegue de volta ao GeoCatalog Explorer na interface web para ver a sua coleção!

Coleta de consultas via API STAC

Agora que visualizou sua coleção no GeoCatalog Explorer, você explicará como usar as APIs STAC do GeoCatalog para pesquisar e recuperar itens e ativos STAC para análise posterior.

Este processo começa por publicar uma pesquisa na API STAC do seu GeoCatalog. Especificamente, você procurará imagens dentro da sua coleção que estejam dentro do limite inicial usado para extrair imagens do Computador Planetário.

Sem surpresa, esta consulta devolve todos os itens STAC que colocou anteriormente na 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 ler 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

Limpeza de 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 do 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 seus 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))

Eliminar coleção

Agora, como etapa final, convém 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óximos passos

Criar uma coleção STAC através da interface web

Conteúdo relacionado

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

  • Last updated on