教程：使用 Microsoft 行星计算机专业 API 引入和可视化数据

STAC（时空资产目录）集合在 GeoCatalog 中用于索引和存储相关的时空资产。 在本端到端教程中，你将创建一个新的 STAC 集合，将 Sentinel-2 图像引入到集合中，并通过 GeoCatalog 的 API 查询这些图像。

在本教程中，你将：

• 将在 Planetary Computer Pro GeoCatalog 中创建您自己的 STAC 集合
• 从欧洲航天局将卫星图像引入该集合
• 配置集合，以便在行星计算机专业版的 Web 界面中可视化集合中的图像
• 使用行星计算机 Pro 的 STAC API 从 STAC 集合中查询数据

本教程通过代码片段演示和说明功能，以获取交互式笔记本样式体验， 以 Jupyter 笔记本的形式下载本教程。

先决条件

在运行本教程之前，需要部署在 Azure 订阅中的行星计算机 Pro GeoCatalog。 还需要一个环境来执行此笔记本并安装所需的包。 建议通过 Azure 机器学习虚拟机或 Visual Studio Code 的笔记本集成在 Python 虚拟环境中运行本教程。 但是，满足以下要求的情况下，应该可以在任何可以运行 Jupyter 笔记本的地方运行此笔记本。

• Python 3.10 或更高版本
• 已安装 Azure CLI，并且已运行 az login 以登录到 Azure 帐户
• 安装了“教程选项”部分中列出的必要要求项

在 Azure 机器学习或 VS Code 中打开 Jupyter 笔记本

使用 Azure CLI 登录到 Azure

以下命令使用 Azure CLI 将你登录到 Azure。 运行该命令，并按照说明登录。

!az login

选择教程选项

在运行本教程之前，需要对现有 GeoCatalog 实例拥有参与者访问权限。 在geocatalog_url变量中输入 GeoCatalog 实例的 URL。 在本教程中，你将为欧洲航天局（ESA）提供的 Sentinel-2 图像创建集合，该图像目前存储在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

导入所需的包

在创建 STAC 集合之前，需要导入几个 python 包并定义帮助程序函数来检索所需的访问令牌。

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 项。 对于本教程，在 Microsoft 的 Planetary Computer 中使用 Sentinel-2-l2a 集合的现有 STAC 集合 JSON。 为集合分配随机 ID 和标题，以免与其他现有集合冲突。

# 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 界面，应会看到“集合”选项卡下列出的新集合。

访问集合缩略图

接下来，需要向集合添加要与集合一起显示的缩略图。 出于此演示的目的，请使用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

将缩略图添加到您的 Planetary Computer Pro 地理目录中

读取缩略图后，可以将其与所需的资产 json 一起发布到 GeoCatalogs 的集合资产 API 终结点，从而将其添加到集合中。

# 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 集合！ 通过以下方式完成此过程：

1. 从Microsoft的行星计算机获取 SAS 令牌
2. 在 GeoCatalog 中将该令牌注册为引入源
3. 将该集合中的 STAC 条目提交到 GeoCatalog 的条目 API
4. 验证项是否已成功引入

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 日期间收集
• 兴趣区域 - 在冰岛南部收集的图像（定义为边界框）

通过执行此搜索，可以在行星计算机中找到匹配的 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']}")

接下来，尝试将此 Azure Blob 存储容器和关联的 SAS 令牌注册为 GeoCatalog 的引入源。 对于此存储容器，可能已存在引入源。 如果是，请查找现有引入源的 ID。

警告

如果发现一个令牌将在接下来 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 的项目 API 引入行星计算机中找到的 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 的操作 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)

应该能够刷新 Web 浏览器，然后单击“项目”选项卡以查看这些新上传的项目。

集合管理

将这些 STAC 项及其关联的资产（图像）引入 STAC 集合后，需要向 GeoCatalog 提供一些其他配置文件，然后才能在 GeoCatalog Web 界面中可视化这些项目。

集合呈现配置

首先从 Planetary Computer 数据平台下载此集合的渲染配置文件。 GeoCatalog 可以读取此配置文件，以在资源管理器中以不同方式呈现图像。 这是因为 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}
    )

马赛克定义

类似于上述讨论的渲染配置，GeoCatalog 的浏览器允许我们为集合指定一个或多个镶嵌定义。 通过这些马赛克定义，我们可以指示 GeoCatalog 的资源管理器如何筛选资源管理器中显示哪些项。 例如，一个基本呈现配置（在下一个单元格中所示）指示 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 资源管理器以查看集合！

通过 STAC API 进行查询收集

在 GeoCatalog 资源管理器中查看集合后，你将逐步了解如何使用 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 返回带有签名的链接（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 查询这些图像的过程。 如果想要了解有关这些主题的详细信息，请浏览以下其他材料：

• 创建 GeoCatalog
• 创建集合
• 引入 STAC 项
• 创建呈现配置
• 配置集合磁贴设置
• 马赛克配置
• Queryable 对象配置