教學課程：使用Microsoft行星計算機專業版 API 來內嵌和可視化數據

STAC（時空資產目錄）集合會在 GeoCatalog 中用於索引及儲存相關的時空資產。在本端對端教學課程中，您將建立新的 STAC 集合、將 Sentinel-2 影像內嵌到集合中，並透過 GeoCatalog 的 API 查詢這些影像。

在本教學課程中，您會：

將在 Planetary Computer Pro 地理目錄中建立您專屬的 STAC 集合。
從歐洲航太局將衛星圖像內嵌到該集合中
設定集合，讓集合中的影像可以在行星計算機專業版的 Web 介面中可視化
使用行星電腦專業版的 STAC API 從 STAC 集合內查詢數據

本教學課程會透過代碼段示範及說明功能，以取得互動式筆記本樣式體驗，以 Jupyter 筆記本的形式下載本教學課程。

先決條件

在執行本教學課程之前，您需要擁有在 Azure 訂閱中部署的"Planetary Computer Pro GeoCatalog"。您也需要環境來執行此筆記本，並安裝必要的套件。建議您透過 Azure Machine Learning 虛擬機或 Visual Studio Code 的筆記本整合，在 Python 虛擬環境中執行本教學課程。不過，只要符合下列需求，此筆記本就應該在您可以執行 Jupyter Notebook 的位置執行：

Python 3.10 或更新版本
已安裝 Azure CLI，而且您已執行 az login 來登入您的 Azure 帳戶
教學選項部分列出的必要需求已安裝

在 Azure Machine Learning 或 VS Code 中開啟 Jupyter Notebook

使用 Azure CLI 登入 Azure

下列命令會使用 Azure CLI 將您登入 Azure。執行命令，並依照指示登入。

!az login

選取教學課程選項

執行本教學課程之前，您需要現有 GeoCatalog 實例的參與者存取權。在 geocatalog_url變數中輸入 GeoCatalog 實例的 URL。在本教學課程中，您將建立歐洲航太局（ESA）目前儲存在Microsoft行星計算機數據目錄中的 Sentinel-2 影像集合。

# 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

匯入所需的軟體套件

您必須先匯入幾個 Python 套件，並定義協助程式函式來擷取必要的存取令牌，才能建立 STAC 集合。

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

建立 STAC 集合

定義 STAC 集合 JSON

接下來，您會將 STAC 集合定義為 JSON 項目。在本教程中，我們將使用微軟行星電腦中 Sentinel-2-l2a 資料集的現有 STAC 資料集 JSON。您的集合會指派隨機標識碼和標題，以免與其他現有的集合衝突。

# 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

在 GeoCatalog 中建立集合時，集合 JSON 不能有任何與集合相關聯的集合層級資產（例如集合縮圖），因此請先移除這些現有的資產（別擔心您稍後再新增縮圖）。

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

開啟您的 GeoCatalog Web 介面，您應該會看到新的集合列在 [集合] 索引標籤底下。

Access 集合縮圖

接下來，您需要在我們的收藏集中新增縮圖，以便與我們的收藏集一起顯示。基於此示範的目的，請使用Microsoft行星計算機內現有 Sentinel-2 集合中的縮圖。

# Read thumbnail for your collection

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

將縮圖新增至行星電腦專業版 GeoCatalog

讀取縮圖後，您可以將它張貼至 GeoCatalogs 的集合資產 API 端點，並附上所需的資產 JSON，將該縮圖新增至我們的集合中。

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

從您的 Planetary Computer Pro GeoCatalog 中閱讀新收藏

請重新整理您的瀏覽器，您應該可以看到縮圖。您也可以對集合端點進行下列呼叫，以程序設計方式擷取集合 JSON：

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

內嵌 STAC 專案&資產

建立此集合之後，您就可以使用 GeoCatalog 的專案 API，將新的 STAC 專案內嵌至 STAC 集合！完成此過程的方法如下：

從Microsoft的行星計算機取得 SAS 令牌
將該令牌註冊為 GeoCatalog 內的擷取來源
將 STAC 專案從該集合張貼至 GeoCatalog 的專案 API
確認物品已成功匯入

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)

查詢 Planetary Computer

首先，您需要查詢行星計算機，以搜尋符合我們特定需求的 Sentinel-2 影像。在此情況下，您要在行星計算機中尋找 Sentinel-2 影像，其符合下列準則：

集合 - 來自 Sentinel-2-l2a 資料集的影像
時間範圍 - 收集於 2024 年 2 月 4 日和 2 月 11 日之間
研究區域－冰島南部 (以邊界框定義) 所採集的影像

透過執行此搜尋，您可以看到在 Planetary Computer 中找到的相符 STAC 項目。

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

註冊匯入來源

您需要先判斷是否需要註冊新的匯入來源，才能將這些 STAC 項目及其相關資產影像匯入到 GeoCatalog 集合。 GeoCatalog 使用攝取來源來追蹤它有權存取哪些儲存位置 (Azure Blob 儲存體容器)。

註冊匯入來源可藉由提供 GeoCatalog 儲存容器的位置，以及具有讀取權限的 SAS 令牌來完成。如果 STAC 專案或其相關資產位於 GeoCatalog 尚未獲得存取權的儲存容器中，內嵌過程將會失敗。

若要啟動此程式，您必須先向行星計算機要求 SAS 令牌，以授與我們 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']}")

接下來，嘗試使用 GeoCatalog 將此 Azure Blob 記憶體容器和相關聯的 SAS 令牌註冊為擷取來源。此儲存容器可能已經有資料引入來源。如果是，請尋找現有擷取來源的標識碼。

警告

如果找到重複的擷取來源，且權杖會在接下來 15 分鐘內到期，則會將其刪除並取代。刪除目前執行擷取所使用的擷取來源可能會中斷這些擷取。

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

使用 GeoCatalog 的項目 API 導入 STAC 項目

現在您已經註冊了攝取來源或驗證了來源的存在，您將使用 GeoCatalog 的 Items API 攝取您在 Planetary Computer 中找到的 STAC 項目。將每個項目發送至 Items API，以在 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']}")

鑑於 Sentinel-2 專案攝取可能需要一些時間，您可以執行此程式碼，使用 GeoCatalog 的 Operations API 檢查攝取操作的狀態。

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

您應該可以刷新網頁瀏覽器，然後點擊項目索引標籤以查看這些新上傳的項目。

集合管理

現在你已經將這些 STAC 項目及其相關資產（圖片）匯入 STAC 集合，接下來你需要先提供一些其他設定檔，才能在 GeoCatalog 網頁介面中視覺化這些項目。

集合渲染設定檔

首先從行星計算機下載此集合的轉譯組態檔。 GeoCatalog 可以讀取此組態檔，以在 Explorer 內以不同方式轉譯影像。這是因為 STAC 項目可能包含許多不同的資產（影像），這些資產可以合併，創建該區域全新的影像，以強調可見或不可見的特徵。例如，Sentinel-2 STAC 項目有來自電磁光譜不同部分的超過 12 種不同影像。此渲染配置會指示 GeoCatalog 如何結合這些影像，以便顯示影像為自然色彩或假色彩（色彩紅外線）。

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

從行星電腦讀取此轉譯選項設定之後，您可以將此設定張貼到轉譯選項端點，以啟用集合的這些轉譯選項。

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

馬賽克定義

類似於上述的 Render Config，GeoCatalog 的 Explorer 可讓我們指定集合的一或多個馬賽克定義。這些馬賽克定義能讓我們指導 GeoCatalog 的 Explorer 如何篩選出在 Explorer 中顯示的項目。例如，一個基本轉譯組態（在下一個單元格中所示）會指示 GeoCatalog 顯示任何指定區域的最新影像。更進階的轉譯設定可讓我們針對在 2023 年 10 月擷取的指定位置呈現不同的檢視，例如最不多雲的影像。

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

開啟 GeoCatalog Web 介面

恭喜！您已建立集合、新增 STAC 專案和資產，並更新您的集合以包含必要的組態檔，以便透過 GeoCatalog Web 介面內的檔案總管加以檢視。

流覽回 Web 介面中的 GeoCatalog Explorer，以檢視您的集合！

透過 STAC API 查詢集合

在 GeoCatalog Explorer 中檢視您的集合後，您將學習如何使用 GeoCatalog 的 STAC API 來搜尋和擷取 STAC 專案和資產，以進行進一步的分析。

流程首先向您的 GeoCatalog 的 STAC API 發送搜尋請求。具體來說，您需要在收藏集中搜尋位於您用於從 Planetary Computer 提取圖像的原始邊界框內的圖像。

不出所料，此查詢會傳回您先前放在集合中的所有 STAC 項目。

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

在先前的查詢中，您也提供了另一個參數 ：sign：true。這會指示 GeoCatalog 傳回已簽署的 href（item href + SAS 令牌），這可讓您從 Azure Blob 儲存體讀取指定的資產。

# 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

清理資源

刪除項目

此時，您已建立 GeoCatalog 集合、將專案和資產新增至集合，並使用 GeoCatalog 的 STAC API 擷取這些項目和資產。在本教學課程的最後階段，您將移除這些項目並刪除您的收藏。

# Delete all items

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

您可以執行下一個命令來確認所有專案都已刪除。請注意，可能需要一兩分鐘的時間才能完全刪除專案及其相關聯的資產。

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

刪除集合

現在作為最後一個步驟，您可能想要從 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}")

後續步驟

透過 Web 介面建立 STAC 集合

在本教程中，您逐步了解如何建立新的 STAC 集合、將 Sentinel-2 圖像匯入集合以及透過 GeoCatalog 的 API 查詢這些圖像。如果您想要深入了解這些主題，請探索下列其他數據：

意見反應

此頁面對您有幫助嗎？

Last updated on 2025-12-19