チュートリアル: Microsoft プラネタリー コンピューター Pro API を使用してデータを取り込んで視覚化する

STAC (SpatioTemporal Asset Catalog) コレクションは、GeoCatalog 内で関連する時空間アセットのインデックス作成と格納に使用されます。 このエンド ツー エンドのチュートリアルでは、新しい STAC コレクションを作成し、Sentinel-2 イメージをコレクションに取り込み、GeoCatalog の API を使用してそれらのイメージにクエリを実行します。

このチュートリアルでは、次の操作を行います。

• 惑星コンピュータ プロ ジオカタログ内にあなただけのSTACコレクションを作成します。
• 欧州宇宙機関から衛星画像をそのコレクションに取り込む
• コレクション内の画像をプラネタリー コンピューター Pro の Web インターフェイスで視覚化できるようにコレクションを構成する
• プラネタリー コンピューター Pro の STAC API を使用して STAC コレクション内からデータを照会する

このチュートリアルでは、コード スニペットを使用して機能を示し、説明します。対話型ノートブック スタイルのエクスペリエンスでは、 このチュートリアルを Jupyter ノートブックとしてダウンロードします。

[前提条件]

このチュートリアルを実行する前に、Azure サブスクリプションにデプロイされたプラネタリー コンピューター Pro GeoCatalog が必要です。 また、このノートブックを実行し、必要なパッケージをインストールするための環境も必要です。 このチュートリアルは、Azure Machine Learning 仮想マシンまたは Python 仮想環境での Visual Studio Code のノートブック統合を通じて実行することをお勧めします。 ただし、このノートブックは、次の要件が満たされていれば、Jupyter Notebook を実行できる場所であればどこでも実行する必要があります。

• Python 3.10 以降
• Azure CLI がインストールされ、az login を実行して Azure アカウントにログインしました
• 「チュートリアル オプション」セクションに記載されている必要な要件がインストールされている

Azure Machine Learning または VS Code で Jupyter ノートブックを開く

Azure CLI を使用して Azure にログインする

次のコマンドは、Azure CLI を使用して Azure にログインします。 コマンドを実行し、指示に従ってログインします。

!az login

チュートリアル オプションを選択する

このチュートリアルを実行する前に、既存の GeoCatalog インスタンスへの共同作成者アクセス権が必要です。 geocatalog_url変数に GeoCatalog インスタンスの URL を入力します。 このチュートリアルでは、現在 Microsoft の惑星コンピューター データ カタログに格納されている欧州宇宙機関 (ESA) によって提供される 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

必要なパッケージをインポートする

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 のプラネタリー コンピューター内の 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

惑星コンピューター Pro GeoCatalog にサムネイルを追加する

サムネイルを読んだ後、必要なアセット 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)

惑星コンピューター 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 の Items API を使用して新しい STAC 項目を STAC コレクションに取り込む準備ができました。 このプロセスを実行するには、次の手順を実行します。

1. Microsoft のプラネタリー コンピューターから SAS トークンを取得する
2. そのトークンを GeoCatalog 内のインジェスト ソースとして登録する
3. そのコレクションから GeoCatalog の Item API に STAC 項目を投稿する
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)

惑星コンピューターのクエリを実行する

まず、プラネタリーコンピュータにクエリを実行して、特定の要件に一致する 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 Storage コンテナー) を追跡するために使用されます。

インジェスト ソースの登録は、GeoCatalog にストレージ コンテナーの場所と、コンテナーにアクセスするための読み取りアクセス許可を持つ SAS トークンを提供することによって実現されます。 STAC 項目またはその関連資産がストレージ コンテナーに配置されている場合、GeoCatalog に取り込みへのアクセス権が付与されていないと、失敗します。

このプロセスを開始するには、まず、Sentinel-2 イメージが存在するコンテナーへの読み取りアクセスを許可する SAS トークンを惑星コンピューターに要求します。

# 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 Storage コンテナーと関連付けられている SAS トークンを、GeoCatalog のインジェスト ソースとして登録します。 このストレージ コンテナーにはインジェスト ソースが既に存在する可能性があります。 その場合は、既存のインジェスト ソースの ID を見つけます。

Warnung

次の 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 の Items API を使用して STAC 項目を取り込む

インジェスト ソースを登録するか、ソースが存在することを検証したら、GeoCatalog の Items API を使用して、惑星コンピューター内で見つかった STAC 項目を取り込みます。 これを実現するには、GeoCatalog 内に新しいインジェスト操作を作成する Items API に各項目を投稿します。

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

Web ブラウザーを更新し、[アイテム] タブをクリックすると、新しくアップロードされたアイテムが表示されます。

コレクション管理

これらの STAC 項目とそれに関連付けられている資産 (イメージ) を STAC コレクションに取り込んだので、GeoCatalog Web インターフェイスでこれらの項目を視覚化する前に、GeoCatalog に他のいくつかの構成ファイルを提供する必要があります。

コレクションのレンダリング構成

最初に、このコレクションのレンダリング構成ファイルを惑星コンピューターからダウンロードします。 この構成ファイルを 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']

このレンダリング オプションの構成をプラネタリー コンピューターから読み取った後、この構成を render-options エンドポイントに投稿することで、コレクションに対してこれらのレンダリング オプションを有効にすることができます。

# 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 のエクスプローラーでは、コレクションに対して 1 つ以上のモザイク定義を指定できます。 これらのモザイク定義を使用すると、エクスプローラー内に表示される項目をフィルター処理する方法について GeoCatalog のエクスプローラーに指示できます。 たとえば、1 つの基本的なレンダリング構成 (次のセルに示されています) は、特定の領域の最新の画像を表示するように 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 に検索を投稿することから始めます。 具体的には、惑星コンピューターから画像を抽出するために使用した元の境界ボックス内にある画像をコレクション内で検索します。

当然ながら、このクエリは、コレクション内に以前に配置したすべての 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 も指定しました。 これにより、Azure Blob Storage から特定の資産を読み取る署名付き href (項目 href + SAS トークン) を返すように GeoCatalog に指示されます。

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

次のコマンドを実行すると、すべてのアイテムが削除されたことを確認できます。 アイテムとそれに関連付けられている資産を完全に削除するには、1 ~ 2 分かかる場合があることに注意してください。

# 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 項目の取り込み
• レンダリング構成を作成する
• コレクションタイルの設定を構成する
• モザイク構成
• Queryables の構成