Kurz: Použití rozhraní MICROSOFT Planety Computer Pro API k ingestování a vizualizaci dat

Kolekce STAC (SpatioTemporal Asset Catalog) se používají v rámci GeoCatalogu k indexování a ukládání souvisejících patiotemporálních prostředků. V tomto uceleném kurzu vytvoříte novou kolekci STAC, ingestujete image Sentinel-2 do kolekce a dotazujete se na tyto obrázky prostřednictvím rozhraní API GeoCatalogu.

V tomto kurzu se naučíte:

Vytvoří vaši vlastní kolekci STAC v prostředí Planetary Computer Pro GeoCatalog.
Ingestování satelitních snímků do této kolekce od Evropské vesmírné agentury
Nakonfigurujte kolekci tak, aby bylo možné obrazy v kolekci vizualizovat ve webovém rozhraní planetárního počítače Pro.
Dotazování dat z kolekce STAC pomocí rozhraní STAC API planetárního počítače Pro

Tento kurz ukazuje a vysvětluje možnosti prostřednictvím fragmentů kódu pro interaktivní prostředí stylu poznámkového bloku a stáhněte si tento kurz jako poznámkový blok Jupyter.

Požadavky

Před spuštěním tohoto kurzu potřebujete planetární počítač Pro GeoCatalog nasazený ve svém předplatném Azure. K provedení tohoto poznámkového bloku a instalaci potřebných balíčků potřebujete také prostředí. Doporučujeme spustit tento kurz prostřednictvím integrace poznámkového bloku služby Azure Machine Learning nebo editoru Visual Studio Code ve virtuálním prostředí Pythonu. Tento poznámkový blok by ale měl běžet všude, kde můžete spouštět poznámkové bloky Jupyter, pokud jsou splněné následující požadavky:

Python 3.10 nebo novější
Nainstaluje se Azure CLI a spuštěním příkazu az login se přihlaste ke svému účtu Azure.
Nainstalují se nezbytné požadavky uvedené v části Možnosti kurzu.

Otevření poznámkového bloku Jupyter ve službě Azure Machine Learning nebo VS Code

Přihlášení k Azure pomocí Azure CLI

Následující příkaz vás do Azure přihlásí pomocí Azure CLI. Spusťte příkaz a postupujte podle pokynů pro přihlášení.

!az login

Výběr možností kurzu

Před spuštěním tohoto kurzu potřebujete přístup přispěvatele k existující instanci GeoCatalog. Do proměnné geocatalog_url zadejte adresu URL instance GeoCatalogu. V tomto kurzu vytvoříte kolekci obrázků Sentinel-2, kterou poskytuje Evropská vesmírná agentura (ESA), která je aktuálně uložena v katalogu Planety Computer Data Catalog společnosti 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

Import požadovaných balíčků

Než budete moct vytvořit kolekci STAC, musíte importovat několik balíčků Pythonu a definovat pomocné funkce pro načtení požadovaného přístupového tokenu.

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

Vytvoření kolekce STAC

Definujte kolekci STAC ve formátu JSON

Dále definujete kolekci STAC jako položku JSON. Pro účely tohoto kurzu použijte existující soubor JSON kolekce STAC pro kolekci Sentinel-2-l2a v rámci Planetary Computer společnosti Microsoft. Kolekce má přiřazené náhodné ID a název, takže není v konfliktu s jinými existujícími kolekcemi.

# 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

Při vytváření kolekce v rámci GeoCatalogu nemůže JSON kolekce obsahovat žádné prostředky na úrovni kolekce, jako je miniatura kolekce. Nejprve odeberte všechny existující prostředky, nebojte se - miniaturu můžete přidat později.

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

Otevřete webové rozhraní GeoCatalog a měli byste vidět novou kolekci uvedenou na kartě Kolekce.

Zobrazit miniaturu kolekce

Dále chcete do naší kolekce přidat miniaturu, která se zobrazí společně s naší kolekcí. Pro účely této ukázky použijte miniaturu z existující kolekce Sentinel-2 v rámci planetárního počítače Microsoftu.

# Read thumbnail for your collection

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

Přidání miniatury do planetárního počítače Pro GeoCatalog

Po prohlédnutí náhledu obrázku ji můžete přidat do vaší kolekce odesláním do koncového bodu rozhraní API prostředků kolekce GeoCatalogs spolu s požadovaným JSON souborem s daty.

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

Přečtěte si novou kolekci z planetárního počítače Pro GeoCatalog

Aktualizujte prohlížeč a měli byste vidět miniaturu. Kód JSON kolekce můžete také načíst programově provedením následujícího volání koncového bodu kolekcí:

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

Ingestování položek a prostředků STAC

Po vytvoření této kolekce jste připraveni ingestovat nové položky STAC do vaší kolekce STAC pomocí rozhraní API položek geoCatalogu! Tento proces provedete takto:

Získání tokenu SAS z planetárního počítače Microsoftu
Registrace tokenu jako zdroje příjmu dat v rámci GeoCatalogu
Pošlete položky STAC z této kolekce do API pro položky GeoCatalogu.
Ověřte, zda byly položky úspěšně ingesovány

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)

Dotazování planetárního počítače

Nejprve je potřeba zadat dotaz na planetární počítač a vyhledat obrázky Sentinel-2, které odpovídají našim konkrétním požadavkům. V tomto případě hledáte image Sentinel-2 v planetárním počítači, který odpovídá následujícím kritériím:

Kolekce – obrázky z kolekce Sentinel-2-l2a
Časový rozsah – shromažďováno od 4. února do 11. února 2024
Oblast zájmu – snímky shromážděné nad jižním Islandem (definované jako ohraničené území)

Provedením tohoto hledání můžete vidět, že odpovídající položky STAC jsou nalezeny v planetárním počítači.

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

Registrace zdroje příjmu dat

Než budete moct tyto položky STAC a související prostředky (image) ingestovat do kolekce GeoCatalog, musíte zjistit, jestli potřebujete zaregistrovat nový zdroj příjmu dat. Zdroje příjmu dat používají GeoCatalog ke sledování umístění úložiště (kontejnerů Azure Blob Storage), ke kterým má přístup.

Registrace zdroje příjmu dat se provádí tak, že službě GeoCatalog poskytnete umístění kontejneru úložiště a token SAS s oprávněními ke čtení pro přístup ke kontejneru. Pokud se položky STAC nebo jejich související aktiva nacházejí v kontejneru úložiště, ke kterému váš GeoCatalog nemá udělený přístup, pokus o ingestování selže.

Pokud chcete tento proces spustit, nejprve si vyžádáte token SAS z planetárního počítače, který nám udělí přístup pro čtení ke kontejneru, ve kterém se nacházejí image 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']}")

Dále se pokusíte zaregistrovat tento kontejner Azure Blob Storage a přidružený token SAS jako zdroj příjmu dat v GeoCatalogu. Existuje potenciál, že pro tento kontejner úložiště už existuje zdroj příjmu dat. Pokud ano, vyhledejte ID existujícího zdroje příjmu dat.

Výstraha

Pokud se najde duplicitní zdroj příjmu dat s tokenem, jehož platnost vyprší za dalších 15 minut, odstraní se a nahradí. Odstranění zdroje příjmu dat, který je aktuálně využíván probíhajícím zpracováním dat, může toto zpracování přerušit.

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

Ingestování položek STAC pomocí rozhraní API položek GeoCatalogu

Teď, když jste zaregistrovali zdroj příjmu dat nebo ověřili, že zdroj existuje, ingestujete položky STAC, které jste našli v planetárním počítači pomocí rozhraní API Položek GeoCatalogu. Provedete to odesláním každé položky do rozhraní API položek, které vytvoří novou operaci příjmu dat v rámci GeoCatalogu.

# 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']}")

Vzhledem k tomu, že příjem dat položek Sentinel-2 může nějakou dobu trvat, můžete tento kód spustit a zkontrolovat stav operací příjmu dat pomocí rozhraní API operací GeoCatalogu.

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

Měli byste být schopni aktualizovat webový prohlížeč a kliknutím na kartu Položky zobrazit tyto nově nahrané položky.

Správa kolekcí

Teď, když jste začlenili tyto položky STAC a jejich přidružené prostředky (obrázky) do kolekce STAC, musíte poskytnout GeoCatalogu některé další konfigurační soubory, než budete moci tyto položky vizualizovat ve webovém rozhraní GeoCatalogu.

Konfigurace vykreslování kolekce

Nejprve stáhněte konfigurační soubor vykreslení pro tuto kolekci z planetárního počítače. Tento konfigurační soubor může GeoCatalog přečíst, aby vykresloval obrázky různými způsoby v Průzkumníku. Je to proto, že položky STAC mohou obsahovat mnoho různých prostředků (obrázků), které lze kombinovat a vytvářet zcela nové obrázky dané oblasti, které zvýrazňují viditelné nebo nevisovatelné funkce. Například položky STAC Sentinel-2 mají více než 12 různých obrázků od různých částí elektromagnetického spektra. Tato konfigurace vykreslení dává geoCatalogu pokyn, jak tyto obrázky kombinovat, aby mohl zobrazovat obrázky v přirozené barvě nebo v nepravdivé barvě (Color Infrared).

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

Po přečtení této konfigurace možností vykreslení z planetárního počítače můžete povolit tyto možnosti vykreslování pro kolekci publikováním této konfigurace do koncového bodu možnosti vykreslování.

# Post render options config to GeoCatalog render-options API

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

for render_option in render_json['renderOptions']:

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

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

Definice mozaiky

Podobně jako konfigurace vykreslení, o které jsme mluvili výše, nám Průzkumník GeoCatalog umožňuje zadat jednu nebo více definic mozaiky pro kolekci. Tyto definice mozaiky nám umožňují instruovat Průzkumníka GeoCatalogu, jak filtrovat, které položky se zobrazují v Průzkumníkovi. Například jedna základní konfigurace vykreslení (zobrazená v další buňce) dává GeoCatalog pokyn k zobrazení nejnovějšího obrázku pro libovolnou danou oblast. Pokročilejší konfigurace vykreslování nám umožňují vykreslit různá zobrazení, jako je nejméně zamračený obraz pro danou lokalitu zachycený v říjnu 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}
)

Otevření webového rozhraní GeoCatalog

Gratulace! Vytvořili jste kolekci, přidali jste položky a prostředky STAC a aktualizovali jste kolekci tak, aby zahrnovala požadované konfigurační soubory, aby bylo možné ji zobrazit prostřednictvím Průzkumníka ve webovém rozhraní GeoCatalog.

Přejděte zpět do Průzkumníka GeoCatalogu ve webovém rozhraní a zobrazte kolekci.

Vyhledávání kolekcí prostřednictvím rozhraní STAC API

Teď, když jste si prohlédli kolekci v Průzkumníku GeoCatalogu, si ukážeme, jak pomocí rozhraní STAC API GeoCatalogu vyhledávat a načítat položky a prostředky STAC k další analýze.

Tento proces začíná publikováním vyhledávání do rozhraní STAC API geoCatalogu. Konkrétně budete hledat obrázky v kolekci, které spadají do původního ohraničujícího rámečku, který jste použili k extrakci obrázků z planetárního počítače.

Není překvapením, že tento dotaz vrátí všechny položky STAC, které jste dříve vložili do své kolekce.

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

V předchozím dotazu jste také zadali další parametr: sign:true. Tím dáte GeoCatalogu pokyn k vrácení podepsaného href (href položky + SAS token), který umožňuje číst dané assety ze služby 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

Vyčistěte zdroje

Odstranění položek

V tomto bodě jste vytvořili kolekci GeoCatalog, přidali do ní položky a prostředky, a pak jste tyto položky a prostředky načetli pomocí STAC API GeoCatalogu. V závěrečné fázi tohoto kurzu odeberete tyto položky a odstraníte kolekci.

# Delete all items

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

Odstranění všech položek můžete potvrdit spuštěním dalšího příkazu. Všimněte si, že úplné odstranění položek a jejich přidružených prostředků může trvat minutu nebo dvě.

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

Odstranění kolekce

Teď jako poslední krok možná budete chtít úplně odstranit kolekci z instance GeoCatalogu.

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

Další kroky

Vytvoření kolekce STAC prostřednictvím webového rozhraní

V tomto kompletním kurzu jste prošli procesem vytvoření nové kolekce STAC, ingestování obrázků Sentinel-2 do kolekce a dotazování těchto obrázků prostřednictvím rozhraní API GeoCatalogu. Pokud se chcete dozvědět více o každém z těchto témat, prozkoumejte tyto další materiály:

Váš názor

Byla tato stránka užitečná?

Last updated on 2025-05-20