stac-assetのサンプルコード集

概要

STACのItemのAssetをダウンロードするツールであるstac-assetのサンプルコードを紹介する。

環境

• Windows 10
• Python 3.11.5
• AWS CLI aws-cli/2.17.42 Python/3.11.9 Windows/10 exe/AMD64

環境構築

1. 以下のコマンドを実行し、Pythonの仮想環境をセットアップする。

python -m venv venv
.\venv\Scripts\activate
pip install pystac-client stac-asset

2. 以下のコマンドを実行し、リクエスタ支払いが可能なAWSのクレデンシャルをセットアップする。

aws configure

サンプルコード

Sentinel-2 Level-2Aのダウンロード

import asyncio

import pystac
import pystac_client
import stac_asset

STAC_API_URL = "https://earth-search.aws.element84.com/v1/"


def get_item() -> pystac.Item:
    client = pystac_client.Client.open(url=STAC_API_URL)
    items = client.search(
        collections=["sentinel-2-l2a"],
        max_items=1,
    )
    item = list(items.items())[0]
    return item


async def download_item(item: pystac.Item) -> None:
    config = stac_asset.Config()
    client = await stac_asset.S3Client.from_config(config=config)
    item = await stac_asset.download_item(item, ".", clients=[client])


if __name__ == "__main__":
    item = get_item()
    asyncio.run(download_item(item))

Sentinel-1 Level-1C Ground Range Detected (GRD)のダウンロード

import asyncio

import pystac
import pystac_client
import stac_asset

STAC_API_URL = "https://earth-search.aws.element84.com/v1/"


def get_item() -> pystac.Item:
    client = pystac_client.Client.open(url=STAC_API_URL)
    items = client.search(
        collections=["sentinel-1-grd"],
        max_items=1,
    )
    item = list(items.items())[0]
    return item


async def download_item(item: pystac.Item) -> None:
    config = stac_asset.Config()
    config.s3_requester_pays = True
    client = await stac_asset.S3Client.from_config(config=config)
    item = await stac_asset.download_item(item, ".", clients=[client])


if __name__ == "__main__":
    item = get_item()
    asyncio.run(download_item(item))

MODIS Surface Reflectance 8-Day (500m)のダウンロード

import asyncio

import pystac
import pystac_client
import stac_asset

STAC_API_URL = "https://planetarycomputer.microsoft.com/api/stac/v1/"


def get_item() -> pystac.Item:
    client = pystac_client.Client.open(url=STAC_API_URL)
    items = client.search(
        collections=["modis-09A1-061"],
        max_items=1,
    )
    item = list(items.items())[0]
    return item


async def download_item(item: pystac.Item) -> None:
    config = stac_asset.Config()
    client = await stac_asset.PlanetaryComputerClient.from_config(config=config)
    item = await stac_asset.download_item(item, ".", clients=[client])


if __name__ == "__main__":
    item = get_item()
    asyncio.run(download_item(item))

ESA WorldCoverのダウンロード

import asyncio

import pystac
import pystac_client
import stac_asset

STAC_API_URL = "https://planetarycomputer.microsoft.com/api/stac/v1/"


def get_item() -> pystac.Item:
    client = pystac_client.Client.open(url=STAC_API_URL)
    items = client.search(
        collections=["esa-worldcover"],
        max_items=1,
    )
    item = list(items.items())[0]
    return item


async def download_item(item: pystac.Item) -> None:
    config = stac_asset.Config()
    client = await stac_asset.PlanetaryComputerClient.from_config(config=config)
    item = await stac_asset.download_item(item, ".", clients=[client])


if __name__ == "__main__":
    item = get_item()
    asyncio.run(download_item(item))

ALOS World 3D-30mのダウンロード

import asyncio

import pystac
import pystac_client
import stac_asset

STAC_API_URL = "https://planetarycomputer.microsoft.com/api/stac/v1/"


def get_item() -> pystac.Item:
    client = pystac_client.Client.open(url=STAC_API_URL)
    items = client.search(
        collections=["alos-dem"],
        max_items=1,
    )
    item = list(items.items())[0]
    return item


async def download_item(item: pystac.Item) -> None:
    config = stac_asset.Config()
    client = await stac_asset.PlanetaryComputerClient.from_config(config=config)
    item = await stac_asset.download_item(item, ".", clients=[client])


if __name__ == "__main__":
    item = get_item()
    asyncio.run(download_item(item))

SENTINEL-1のSLCのダウンロード

import asyncio
import os
from urllib.parse import urlparse, urlunparse

import pystac
import pystac_client
import stac_asset
from aiohttp_oauth2_client.models.grant import GrantType

os.environ["COPERNICUS_USER"] = "your_user"  # Copernicusのユーザー
os.environ["COPERNICUS_PASSWORD"] = "your_password"  # Copernicusのパスワード

STAC_API_URL = "https://catalogue.dataspace.copernicus.eu/stac"


def get_item() -> pystac.Item:
    client = pystac_client.Client.open(url=STAC_API_URL)
    items = client.search(
        collections=["SENTINEL-1"],
    )
    for item in items.items():
        if "SLC" in item.id:
            return item


def modify_assets(item: pystac.Item) -> pystac.Item:
    url = item.assets["PRODUCT"].href
    parsed_url = urlparse(url)
    new_netloc = "zipper.dataspace.copernicus.eu"
    new_url = urlunparse(parsed_url._replace(netloc=new_netloc))
    item.assets["PRODUCT"].href = new_url
    return item


async def download_item(item: pystac.Item) -> None:
    config = stac_asset.Config()
    config.oauth2_grant = GrantType.RESOURCE_OWNER_PASSWORD_CREDENTIALS
    config.oauth2_token_url = "https://identity.dataspace.copernicus.eu/auth/realms/CDSE/protocol/openid-connect/token"
    config.oauth2_username = os.getenv("COPERNICUS_USER")
    config.oauth2_password = os.getenv("COPERNICUS_PASSWORD")
    config.oauth2_client_id = "cdse-public"
    config.http_client_timeout = 3600
    client = await stac_asset.HttpClient.from_config(config=config)
    await client.download_href(
        href=item.assets["PRODUCT"].href,
        path=os.path.basename(
            item.to_dict()["assets"]["PRODUCT"]["alternate"]["s3"]["href"]
        ),
    )


if __name__ == "__main__":
    item = get_item()
    item = modify_assets(item)
    asyncio.run(download_item(item))

Precipitation Rate (Monthly)のダウンロード

import asyncio

import pystac
import stac_asset
from aiohttp_oauth2_client.models.grant import GrantType

ITEM_URL = "https://s3.ap-northeast-1.wasabisys.com/je-pds/cog/v1/JAXA.EORC_GSMaP_standard.Gauge.00Z-23Z.v6_monthly/2024-04/1/E000.00-E090.00/N00.00-N90.00.json"


def get_item() -> pystac.Item:
    item = pystac.read_file(href=ITEM_URL)
    assert isinstance(item, pystac.Item)
    return item


async def download_item(item: pystac.Item) -> None:
    config = stac_asset.Config()
    client = await stac_asset.HttpClient.from_config(config=config)
    item = await stac_asset.download_item(item, ".", clients=[client])


if __name__ == "__main__":
    item = get_item()
    asyncio.run(download_item(item))

参考文献

• MODIS Version 6.1 Products
• STAC product catalogue