Esercitazione: Uso delle API Microsoft Planetary Computer Pro per inserire e visualizzare i dati

Le raccolte STAC (SpatioTemporal Asset Catalog) vengono usate all'interno di un GeoCatalog per indicizzare e archiviare asset spatiotemporal correlati. Nella presente esercitazione end-to-end, creerai una nuova raccolta STAC, inserirai immagini Sentinel-2 nella raccolta ed eseguirai query su tali immagini tramite le API di GeoCatalog.

In questa esercitazione, farai:

• Creerà la tua collezione STAC personalizzata all'interno di un GeoCatalog di Planetary Computer Pro
• Inserire immagini satellitari in tale raccolta dall'Agenzia spaziale europea
• Configurare la raccolta in modo che le immagini nella raccolta possano essere visualizzate nell'interfaccia Web di Planetary Computer Pro
• Eseguire query sui dati dall'interno della raccolta STAC usando l'API STAC di Planetary Computer Pro

Questa esercitazione mostra e spiega le funzionalità tramite frammenti di codice, per un'esperienza in stile notebook interattivo, scarica questa esercitazione come notebook di Jupyter.

Prerequisiti

Prima di eseguire questa esercitazione, è necessario un GeoCatalog di Planetary Computer Pro distribuito nella sottoscrizione di Azure. È anche necessario un ambiente per eseguire questo notebook e installare i pacchetti necessari. È consigliabile eseguire questa esercitazione tramite una macchina virtuale di Azure Machine Learning o l'integrazione del notebook di Visual Studio Code in un ambiente virtuale Python. Tuttavia, questo notebook deve essere eseguito ovunque sia possibile eseguire notebook jupyter, purché siano soddisfatti i requisiti seguenti:

• Python 3.10 o versione successiva
• L'interfaccia della riga di comando di Azure è installata ed è stato eseguito az login per accedere all'account Azure
• I requisiti richiesti elencati nella sezione Opzioni del Tutorial sono installati.

Aprire un notebook di Jupyter in Azure Machine Learning o VS Code

Accedere ad Azure con l'interfaccia della riga di comando di Azure

Il comando seguente consente di accedere ad Azure usando l'interfaccia della riga di comando di Azure. Eseguire il comando e seguire le istruzioni per accedere.

!az login

Selezionare le opzioni del tutorial

Prima di eseguire questa esercitazione, è necessario accedere come collaboratore a un'istanza GeoCatalog esistente. Immettere l'URL dell'istanza GeoCatalog nella variabile geocatalog_url. In questa esercitazione si creerà una raccolta per le immagini sentinel-2 fornite dall'Agenzia spaziale europea (ESA) attualmente archiviata nel Catalogo dati dei computer planetari di 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

Importare i pacchetti necessari

Prima di poter creare una raccolta STAC, è necessario importare alcuni pacchetti Python e definire le funzioni helper per recuperare il token di accesso necessario.

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

Creare una raccolta STAC

Definire un file JSON di collezione STAC

Successivamente, si definisce una raccolta STAC come elemento JSON. Per questa esercitazione, utilizzare un JSON di una collezione STAC esistente per la raccolta Sentinel-2-l2a nella piattaforma Planetary Computer di Microsoft. Alla raccolta viene assegnato un ID casuale e un titolo in modo da non entrare in conflitto con altre raccolte esistenti.

# 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

Quando si crea una raccolta all'interno di GeoCatalog un file JSON di raccolta non può avere asset a livello di raccolta (ad esempio un'anteprima della raccolta) associati alla raccolta, quindi rimuovere prima di tutto gli asset esistenti (non preoccuparsi di aggiungere l'anteprima in un secondo momento).

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

Aprire l'interfaccia Web GeoCatalog e visualizzare la nuova raccolta elencata nella scheda "Raccolte".

Anteprima della raccolta Access

Successivamente si desidera aggiungere un'anteprima alla nostra raccolta per visualizzarla insieme alla raccolta. Ai fini di questa demo, usare la miniatura della raccolta Sentinel-2 esistente all'interno del Planetary Computer di 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

Aggiungi la miniatura al GeoCatalog di Planetary Computer Pro

Dopo aver letto l'anteprima, è possibile aggiungerla alla raccolta pubblicandola nell'endpoint dell'API degli asset di raccolta geoCatalogs insieme al codice JSON dell'asset richiesto.

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

Leggere la nuova raccolta dall'interno di Planetary Computer Pro GeoCatalog

Aggiornare il browser e dovrebbe essere possibile visualizzare l'anteprima. È anche possibile recuperare la raccolta JSON a livello di codice eseguendo la chiamata seguente all'endpoint delle raccolte:

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

Inserire elementi e asset STAC

Dopo aver creato questa raccolta, è possibile inserire nuovi elementi STAC nella raccolta STAC usando l'API Items di GeoCatalog. Eseguire questo processo in base a:

1. Ottenere un token SAS dal Planetary Computer di Microsoft
2. Registrare il token come origine di inserimento all'interno di GeoCatalog
3. Invia gli elementi STAC da quella collezione all'API degli elementi di GeoCatalog
4. Verificare che gli elementi siano stati acquisiti correttamente

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)

Interrogare il computer planetario

Prima è necessario eseguire una query sul Planetary Computer per cercare immagini Sentinel-2 che soddisfano le nostre specifiche esigenze. In questo caso, si cercano immagini Sentinel-2 nel Computer Planetario che soddisfano i criteri seguenti.

• Raccolta - Immagini della raccolta Sentinel-2-l2a
• Intervallo di tempo: raccolto tra il 4 febbraio e il 11 febbraio 2024
• Area di interesse - Immagini raccolte in Islanda meridionale (rappresentato come un riquadro di delimitazione)

Eseguendo questa ricerca, è possibile vedere che gli elementi STAC corrispondenti sono disponibili all'interno 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]

Registrare un'origine di inserimento

Prima di poter inserire questi elementi STAC e i relativi asset (immagini) in una raccolta GeoCatalog, è necessario determinare se è necessario registrare una nuova origine di inserimento. Le fonti di ingestione vengono utilizzate da GeoCatalog per tenere traccia delle posizioni di archiviazione (contenitori BLOB di Azure) a cui ha accesso.

La registrazione di un'origine di inserimento viene eseguita fornendo al GeoCatalogo la posizione del contenitore di archiviazione e un token SAS con autorizzazioni di lettura per accedere al contenitore. Se gli elementi STAC o gli asset correlati si trovano in un contenitore di archiviazione a cui il GeoCatalog non ha accesso, l'inserimento avrà esito negativo.

Per avviare questo processo, occorre innanzitutto richiedere un token SAS (Shared Access Signature) dal Planetary Computer, che ci concede l'accesso in lettura al contenitore in cui risiedono le immagini del 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']}")

Successivamente, tenta di registrare questo contenitore Microsoft Azure Blob Storage e il SAS token associato come fonte di inserimento con GeoCatalog. C'è la possibilità che esista già un'origine di inserimento per questo contenitore di archiviazione. In tal caso, trova l'ID dell'origine di inserimento esistente.

Avvertimento

Se viene trovata un'origine di inserimento duplicata con un token che scade nei prossimi 15 minuti, viene eliminato e sostituito. L'eliminazione di un'origine di inserimento attualmente utilizzata da inserimenti in corso potrebbe interrompere questi ultimi.

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

Inserire elementi STAC usando l'API Items di GeoCatalog

Dopo aver registrato una sorgente di inserimento o averne convalidato l'esistenza, gli elementi STAC trovati all'interno del Planetary Computer verranno inseriti utilizzando l'API Items di GeoCatalog. A tale scopo, pubblicare ogni elemento nell'API Items che crea una nuova operazione di inserimento all'interno di 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']}")

Dato che l'inserimento di elementi Sentinel-2 può richiedere poco tempo, è possibile eseguire questo codice per controllare lo stato delle operazioni di inserimento usando l'API Operazioni di 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)

Dovrebbe essere possibile aggiornare il Web browser e fare clic sulla scheda Elementi per visualizzare questi elementi appena caricati.

Gestione delle raccolte

Ora che questi elementi STAC sono stati inseriti e gli asset associati (immagini) nella raccolta STAC, è necessario fornire a GeoCatalog altri file di configurazione prima di poter visualizzare questi elementi nell'interfaccia Web GeoCatalog.

Configurazione del rendering della raccolta

Prima di tutto, scaricare un file di configurazione di render per questa raccolta dal Planetary Computer. Questo file di configurazione può essere letto da GeoCatalog per eseguire il rendering delle immagini in modi diversi all'interno di Explorer. Questo perché gli elementi STAC possono contenere molti asset (immagini) diversi che possono essere combinati per creare immagini completamente nuove di una determinata area che evidenziano funzionalità visibili o non visibili. Ad esempio, gli elementi STAC sentinel-2 hanno oltre 12 immagini diverse da diverse parti dello spettro elettromagnetico. Questa configurazione di rendering indica a GeoCatalog come combinare queste immagini in modo che possa visualizzare immagini in Colore naturale o Colore falso (colore infrarosso).

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

Dopo aver letto questa configurazione delle opzioni di rendering da Planetary Computer, è possibile abilitare queste opzioni di rendering per la raccolta pubblicando questa configurazione nell'endpoint delle opzioni di rendering.

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

Definizioni di mosaico

Analogamente alla configurazione di rendering illustrata in precedenza, Explorer di GeoCatalog consente di specificare una o più definizioni di mosaico per la raccolta. Queste definizioni di mosaico consentono di indicare a Explorer di GeoCatalog come filtrare gli elementi visualizzati al suo interno. Ad esempio, una configurazione di rendering di base (illustrata nella cella successiva) indica a GeoCatalog di visualizzare l'immagine più recente per qualsiasi area specificata. Le configurazioni di rendering più avanzate consentono di creare visualizzazioni diverse, come l'immagine meno nuvolosa per una determinata posizione acquisita nell'ottobre 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}
)

Aprire l'interfaccia Web GeoCatalog

L'esercizio è terminato. È stata creata una raccolta, sono stati aggiunti elementi e asset STAC ed è stata aggiornata la raccolta in modo da includere i file di configurazione necessari in modo che possa essere visualizzata tramite Esplora risorse all'interno dell'interfaccia Web GeoCatalog.

Tornare a GeoCatalog Explorer nell'interfaccia Web per visualizzare la raccolta.

Raccolta di query tramite l'API STAC

Dopo aver visualizzato la raccolta in GeoCatalog Explorer, verrà illustrato come usare le API STAC di GeoCatalog per cercare e recuperare gli elementi e gli asset STAC di GEOCatalog per un'ulteriore analisi.

Questo processo inizia pubblicando una ricerca nell'API STAC di GeoCatalog. In particolare, cercherai le immagini nella tua raccolta che rientrano nel rilevamento iniziale originale usato per estrarre le immagini dal Computer Planetario.

Non sorprendentemente, questa query restituisce tutti gli elementi STAC che hai inserito precedentemente nella tua raccolta.

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

Nella query precedente è stato fornito anche un altro parametro: sign:true. Ciò istruisce GeoCatalog a restituire un href firmato (href dell'elemento + token SAS) che consente di leggere gli asset specificati dall'Archiviazione BLOB di 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

Pulire le risorse

Elimina elementi

A questo punto, è stata creata una raccolta GeoCatalog, sono stati aggiunti elementi e asset alla raccolta e sono stati recuperati gli elementi e gli asset usando l'API STAC di GeoCatalog. Per la fase finale di questa esercitazione, tu rimuoverai questi elementi ed eliminerai la tua raccolta.

# Delete all items

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

È possibile verificare che tutti gli elementi siano stati eliminati eseguendo il comando successivo. Si noti che potrebbero essere necessari un minuto o due per eliminare completamente gli elementi e gli asset associati.

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

Elimina raccolta

A questo punto, come passaggio finale, è possibile eliminare completamente la raccolta dall'istanza 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}")

Passaggi successivi

Creare una raccolta STAC tramite l'interfaccia Web

Contenuti correlati

In questa esercitazione end-to-end è stato illustrato il processo di creazione di una nuova raccolta STAC, l'inserimento di immagini Sentinel-2 nella raccolta e l'esecuzione di query su tali immagini tramite le API di GeoCatalog. Per altre informazioni su ognuno di questi argomenti, esplorare questi altri materiali:

• Creare un GeoCatalog
• Creare una raccolta
• Inserire elementi STAC
• Creare una configurazione di rendering
• Configurare le impostazioni dei riquadri della raccolta
• Configurazione mosaico
• Configurazione degli oggetti interrogabili