Tutoriel : Utilisation des API Microsoft Planetary Computer Pro pour ingérer et visualiser des données

Les collections STAC (catalogue de ressources SpatioTemporal) sont utilisées dans un GeoCatalog pour indexer et stocker les ressources spatiotemporales associées. Dans ce tutoriel de bout en bout, vous allez créer une collection STAC, ingérer des images Sentinel-2 dans la collection et interroger ces images via les API de GeoCatalog.

Dans ce tutoriel, vous allez :

• Créera votre propre collection STAC au sein d’un Ordinateur Planetary Pro GeoCatalog
• Ingérer des images satellites dans cette collection de l’Agence spatiale européenne
• Configurez la collection afin que l’imagerie de la collection puisse être visualisée dans l’interface web de Planetary Computer Pro
• Interroger des données à partir de la collection STAC à l’aide de l’API STAC de Planetary Computer Pro

Ce tutoriel présente et explique les fonctionnalités par le biais d’extraits de code, pour une expérience de style de bloc-notes interactive, téléchargez ce didacticiel en tant que notebook Jupyter.

Conditions préalables

Avant d’exécuter ce tutoriel, vous avez besoin d’un GeoCatalog Planetary Computer Pro déployé dans votre abonnement Azure. Vous avez également besoin d’un environnement pour exécuter ce notebook et installer les packages nécessaires. Nous vous suggérons d’exécuter ce tutoriel via une machine virtuelle Azure Machine Learning ou l’intégration du notebook de Visual Studio Code dans un environnement virtuel Python. Toutefois, ce notebook doit s’exécuter partout où vous pouvez exécuter des notebooks Jupyter, à condition que les conditions suivantes soient remplies :

• Python 3.10 ou version ultérieure
• Azure CLI est installé et vous avez exécuté az login pour vous connecter à votre compte Azure
• Les conditions requises répertoriées dans la section Options du didacticiel sont installées

Ouvrir un notebook Jupyter dans Azure Machine Learning ou VS Code

Connectez-vous à Azure avec Azure CLI

La commande suivante vous connecte à Azure à l'aide de l'interface de ligne de commande Azure (Azure CLI). Exécutez la commande et suivez les instructions pour vous connecter.

!az login

Sélectionner des options de didacticiel

Avant d’exécuter ce didacticiel, vous avez besoin d’un accès contributeur à une instance GeoCatalog existante. Entrez l’URL de votre instance GeoCatalog dans la variable geocatalog_url. Dans ce tutoriel, vous allez créer une collection pour les images Sentinel-2 fournies par l’Agence spatiale européenne (ESA) qui est actuellement stockée dans le catalogue de données d’ordinateurs planétaires 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

Importer les packages requis

Avant de pouvoir créer une collection STAC, vous devez importer quelques packages Python et définir des fonctions d’assistance pour récupérer le jeton d’accès requis.

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

Créer une collection STAC

Définir un json de collection STAC

Ensuite, vous définissez une collection STAC en tant qu’élément JSON. Pour ce tutoriel, utilisez un fichier JSON de collection STAC existant pour la collection Sentinel-2-l2a dans l’ordinateur planétaire de Microsoft. Votre collection reçoit un ID et un titre aléatoires afin de ne pas entrer en conflit avec d’autres collections existantes.

# 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

Lors de la création d’une collection dans GeoCatalog, un json de collection ne peut pas avoir de ressources au niveau de la collection (par exemple, une miniature de collection) associées à la collection. Supprimez d’abord ces ressources existantes (ne vous inquiétez pas d’ajouter la miniature ultérieurement).

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

Ouvrez votre interface web GeoCatalog et vous devez voir votre nouvelle collection répertoriée sous l’onglet « Collections ».

Miniature de collection Access

Ensuite, vous souhaitez ajouter une miniature à notre collection à afficher avec notre collection. Pour les besoins de cette démonstration, utilisez la miniature de la collection Sentinel-2 existante dans l’ordinateur planetaire 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

Ajouter une miniature à votre géocatalogue pro de l’ordinateur planétaire

Après avoir lu la miniature, vous pouvez l’ajouter à notre collection en la publiant sur le point de terminaison de l’API des ressources de collection de GeoCatalogs, ainsi que le json de ressource requis.

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

Lire une nouvelle collection à partir de votre Ordinateur Planetary Pro GeoCatalog

Actualisez votre navigateur et vous devriez être en mesure de voir la miniature. Vous pouvez également récupérer la collection JSON par programmation en effectuant l’appel suivant au point de terminaison de regroupement :

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

Ingérer des éléments STAC et des ressources

Après avoir créé cette collection, vous êtes prêt à ingérer de nouveaux éléments STAC dans votre collection STAC à l’aide de l’API Items de GeoCatalog ! Effectuez ce processus en procédant comme suit :

1. Obtention d’un jeton SAS à partir de l’ordinateur planetaire de Microsoft
2. Inscrire ce jeton en tant que source d’ingestion dans GeoCatalog
3. Publier des éléments STAC de cette collection vers l’API Item de GeoCatalog
4. Vérifier que les éléments ont été ingérés correctement

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)

Effectuer une requête à l’ordinateur de la Planète

Tout d’abord, vous devez interroger l’ordinateur planetaire pour rechercher des images Sentinel-2 qui correspondent à nos exigences spécifiques. Dans ce cas, vous recherchez des images Sentinel-2 dans l’ordinateur planetaire qui correspondent aux critères suivants :

• Collection - Images de la collection Sentinel-2-l2a
• Intervalle de temps - Collecté entre le 4 février et le 11 février 2024
• Zone d’intérêt - Imagerie collectée sur le sud de l’Islande (définie comme un cadre englobant)

En effectuant cette recherche, vous pouvez voir les éléments STAC correspondants se trouvent dans l’ordinateur planetaire.

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

Inscrire une source d’ingestion

Avant de pouvoir ingérer ces éléments STAC et leurs ressources associées (images) dans une collection GeoCatalog, vous devez déterminer si vous devez inscrire une nouvelle source d’ingestion. Les sources d’ingestion sont utilisées par GeoCatalog pour suivre les emplacements de stockage (conteneurs stockage Blob Azure) auxquels il a accès.

L’inscription d’une source d’ingestion est effectuée en fournissant à GeoCatalog l’emplacement du conteneur de stockage et un jeton SAP avec des autorisations de lecture pour accéder au conteneur. Si les éléments STAC ou leurs ressources associées se trouvent dans un conteneur de stockage et que votre GeoCatalog n'a pas accès, l'ingestion échouera.

Pour démarrer ce processus, vous demandez d’abord un jeton SAS à partir de l’ordinateur planetaire qui nous accorde l’accès en lecture au conteneur où résident les images 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']}")

Ensuite, essayez d’inscrire ce conteneur Stockage Blob Azure et le jeton SAP associé en tant que source d’ingestion avec GeoCatalog. Il est possible qu’une source d’ingestion existe déjà pour ce conteneur de stockage. Si c’est le cas, recherchez l’ID de la source d’ingestion existante.

Avertissement

Si une source d’ingestion en double est trouvée avec un jeton qui expire dans les 15 prochaines minutes, elle est supprimée et remplacée. La suppression d’une source d’ingestion en cours d’exécution peut interrompre ces ingestions.

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

Ingérer des éléments STAC à l’aide de l’API Items de GeoCatalog

Maintenant que vous avez inscrit une source d’ingestion ou validé qu’une source existe, vous allez ingérer les éléments STAC que vous avez trouvés dans l’ordinateur planetaire à l’aide de l’API Éléments de GeoCatalog. Pour ce faire, publiez chaque élément dans l’API Items qui crée une opération d’ingestion dans 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']}")

Étant donné que l’ingestion d’éléments Sentinel-2 peut prendre un peu de temps, vous pouvez exécuter ce code pour vérifier l’état de vos opérations d’ingestion à l’aide de l’API Operations 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)

Vous devez être en mesure d’actualiser votre navigateur web et de cliquer sur l’onglet Éléments pour afficher ces éléments nouvellement chargés.

Gestion des collections

Maintenant que vous avez ingéré ces éléments STAC et leurs ressources associées (images) dans la collection STAC, vous devez fournir à votre GeoCatalog d'autres fichiers de configuration avant de pouvoir visualiser ces éléments dans l'interface web de GeoCatalog.

Configuration de rendu pour la collection

Tout d’abord, téléchargez un fichier de configuration de rendu pour cette collection à partir de l’ordinateur planetaire. Ce fichier de configuration peut être lu par GeoCatalog pour afficher des images de différentes manières dans l’Explorateur. Cela est dû au fait que les éléments STAC peuvent contenir de nombreuses ressources différentes (images) qui peuvent être combinées pour créer des images entièrement nouvelles d’une zone donnée qui mettent en évidence des fonctionnalités visibles ou non visibles. Par exemple, les éléments STAC Sentinel-2 ont plus de 12 images différentes de différentes parties du spectre électromagnétique. Cette configuration de rendu indique à GeoCatalog comment combiner ces images afin d’afficher des images en couleur naturelle ou en fausse couleur (infrarouge de couleur).

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

Après avoir lu cette configuration des options de rendu à partir du Planetary Computer, vous pouvez activer ces options de rendu pour la collection en publiant cette configuration sur l'endpoint des options de rendu.

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

Définitions de mosaïque

À l’instar de la configuration de rendu décrite ci-dessus, l’Explorateur de GeoCatalog nous permet de spécifier une ou plusieurs définitions de mosaïque pour la collection. Ces définitions de mosaïque nous permettent d’indiquer à l’Explorateur de GeoCatalog comment filtrer les éléments qui sont affichés dans l’Explorateur. Par exemple, une configuration de rendu de base (illustrée dans la cellule suivante) indique à GeoCatalog d’afficher l’image la plus récente pour une zone donnée. Les configurations de rendu plus avancées nous permettent d’afficher différentes vues telles que l’image la moins nuageux pour un emplacement donné capturé en octobre 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}
)

Ouvrir l’interface web GeoCatalog

Félicitations ! Vous avez créé une collection, ajouté des éléments et des ressources STAC et mis à jour votre collection pour inclure les fichiers de configuration requis afin qu’elle puisse être consultée via l’Explorateur dans l’interface web GeoCatalog.

Revenez à l’Explorateur GeoCatalog dans l’interface web pour afficher votre collection !

Collection de requêtes via l’API STAC

Maintenant que vous avez consulté votre collection dans l’Explorateur GeoCatalog, vous allez découvrir comment utiliser les API STAC de GeoCatalog pour rechercher et récupérer des éléments et des ressources STAC pour une analyse plus approfondie.

Ce processus commence par publier une recherche dans l’API STAC de GeoCatalog. Plus précisément, vous allez rechercher des images dans votre collection qui se trouvent dans la zone englobante d’origine que vous avez utilisée pour extraire l’imagerie de l’ordinateur planetaire.

Sans surprise, cette requête retourne tous les éléments STAC que vous avez précédemment placés dans votre collection.

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

Dans votre requête précédente, vous avez également fourni un autre paramètre : sign :true. Cela indique à GeoCatalog de retourner un href signé (élément href + jeton SAS) qui vous permet de lire les ressources fournies à partir d'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

Nettoyer les ressources

Supprimer des éléments

À ce stade, vous avez créé une collection GeoCatalog, ajouté des éléments et des ressources à la collection, puis récupéré ces éléments et ressources à l’aide de l’API STAC de GeoCatalog. Pour la dernière phase de ce didacticiel, vous allez supprimer ces éléments et supprimer votre collection.

# Delete all items

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

Vous pouvez confirmer que tous vos éléments ont été supprimés en exécutant la commande suivante. Notez qu’il peut prendre une minute ou deux pour supprimer entièrement les éléments et leurs ressources associées.

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

Supprimer une collection

À présent, vous souhaiterez peut-être supprimer entièrement votre collection de votre instance 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}")

Étapes suivantes

Créer une collection STAC via l’interface web

Contenu connexe

Dans ce didacticiel de bout en bout, vous avez parcouru le processus de création d’une collection STAC, de l’ingestion d’images Sentinel-2 dans la collection et de l’interrogation de ces images via les API de GeoCatalog. Si vous souhaitez en savoir plus sur chacune de ces rubriques, explorez ces autres documents :

• Créer un GeoCatalog
• Créer un regroupement
• Ingérer des éléments STAC
• Créer une configuration de rendu
• Configurer les paramètres des vignettes de collection
• Configuration de mosaïque
• Configuration des éléments interrogeables