STAC Task (stac-task)

• 创建新任务快速入门
• 任务输入
  • ProcessDefinition 对象
    • UploadOptions 对象
      • path_template
      • collections
    • tasks
    • TaskConfig 对象
• 完整流程定义示例
• 迁移
  • 0.4.x -> 0.5.x
• 开发
• 贡献

这个Python库由Task类组成，用于根据“STAC In，STAC Out”方法创建自定义任务。Task类作为自定义代码的包装器，并为修改STAC Items、创建派生Items和提供CLI提供了一些便利方法。

这个库基于cirrus-lib的一个分支，但旨在更加通用。

创建新任务快速入门

from typing import Any

from stactask import Task, DownloadConfig

class MyTask(Task):
    name = "my-task"
    description = "this task does it all"

    def validate(self, payload: dict[str, Any]) -> bool:
        return len(self.items) == 1

    def process(self, **kwargs: Any) -> list[dict[str, Any]]:
        item = self.items[0]

        # download a datafile
        item = self.download_item_assets(
            item,
            config=DownloadConfig(include=['data'])
        )

        # operate on the local file to create a new asset
        item = self.upload_item_assets_to_s3(item)

        # this task returns a single item
        return [item.to_dict(include_self_link=True, transform_hrefs=False)]

任务输入

ProcessDefinition 对象

STAC任务可以通过输入ItemCollection中的'process'字段提供额外的配置。

UploadOptions 对象

path_template

path_template字符串是一种使用项目本身的元数据来控制从STAC项目上传的资源的输出位置的方法。模板可以包含固定字符串以及用于替换的变量。有关支持的模板变量及其含义的列表，请参阅PySTAC文档中的LayoutTemplate。

collections

collections字典提供集合ID和JSONPath模式，用于与STAC项目匹配。在处理完成后，在返回最终STAC项目之前，可以使用Task类将所有项目分配给特定的集合ID。对于每个项目，将比较所有集合的JSONPath模式。第一个匹配将导致项目的集合ID设置为提供的值。

例如

"collections": {
    "landsat-c2l2": "$[?(@.id =~ 'LC08.*')]"
}

在这个例子中，该任务将ID以"LC08"开头的任何STAC项目设置为landsat-c2l2集合。

请参阅JSONPath在线评估器以实验JSONPath和regex101以实验正则表达式。

tasks

tasks字段是一个字典，可选的键对应每个任务。如果存在，它包含一个字典，该字典被转换为关键字集并传递给Task的process函数。每个任务的文档将提供可用参数的列表。

{
    "tasks": {
        "task-a": {
            "param1": "value1"
        },
        "task-c": {
            "param2": "value2"
        }
    }
}

在上面的例子中，名为task-a的任务将传递param1=value1作为关键字，而task-c将传递param2=value2。如果有一个要运行的task-b，则不会传递任何关键字。

TaskConfig 对象

已弃用：tasks应是一个参数字典，以任务名称为键。有关更多信息，请参阅tasks。

任务配置包含运行特定任务的信息。

完整流程定义示例

进程定义有时被称为“有效载荷”

{
    "description": "My process configuration",
    "collections": {
        "landsat-c2l2": "$[?(@.id =~ 'LC08.*')]"
    },
    "upload_options": {
        "path_template": "s3://my-bucket/${collection}/${year}/${month}/${day}/${id}"
    },
    "tasks": {
        "task-name": {
            "param": "value"
        }
    }
}

迁移

0.4.x -> 0.5.x

在0.5.0版本中，之前使用fsspec下载项目资源的方法已被stac-asset库所取代。这需要更改下载方法接受的参数。

主要变化是，Task方法download_item_assets和download_items_assets（复数项）现在接受更少的显式和隐式（kwargs）参数。

之前的方法看起来像

  def download_item_assets(
        self,
        item: Item,
        path_template: str = "${collection}/${id}",
        keep_original_filenames: bool = False,
        **kwargs: Any,
    ) -> Item:

但现在看起来像

    def download_item_assets(
        self,
        item: Item,
        path_template: str = "${collection}/${id}",
        config: Optional[DownloadConfig] = None,
    ) -> Item:

同样，asset_io包的方法之前是

async def download_item_assets(
    item: Item,
    assets: Optional[list[str]] = None,
    save_item: bool = True,
    overwrite: bool = False,
    path_template: str = "${collection}/${id}",
    absolute_path: bool = False,
    keep_original_filenames: bool = False,
    **kwargs: Any,
) -> Item:

现在是

async def download_item_assets(
    item: Item,
    path_template: str = "${collection}/${id}",
    config: Optional[DownloadConfig] = None,
) -> Item:

此外，kwargs键被设置为将配置传递给fsspec。最常用的参数是requester_pays，用于设置AWS S3请求中的请求者付费标志。

许多这些参数可以直接转换为传递给DownloadConfig对象的配置，它只是对stac_asset.Config对象的包装。

将这些各种参数迁移到DownloadConfig的方式如下

• 资源：assets 设置为 include
• requester_pays：将 s3_requester_pays 设置为 True
• keep_original_filenames：如果为 True，则将 file_name_strategy 设置为 FileNameStrategy.FILE_NAME；如果为 False，则设置为 FileNameStrategy.KEY
• overwrite：设置 overwrite
• save_item：无，项总是保存
• absolute_path：无。要创建或检索资产 href 作为绝对路径，请使用 Item#make_all_asset_hrefs_absolute() 或 Asset#get_absolute_href()

0.5.x -> 0.6.0

之前，validate 方法是一个 classmethod，用于验证传递的负载参数。现在，它已被改为实例方法，用于验证负载的 self._payload 复制，该复制用于构造 Task 实例。行为上相同，即如果验证失败，构造将失败，但允许实现者利用实例方法的便利功能。

validate 的先前实现可能如下所示

    @classmethod
    def validate(payload: dict[str, Any]) -> bool:
        # Check The Things™
        return isinstance(payload, dict)

现在需要更新为以下形式

    def validate(self) -> bool:
        # Check The Things™
        return isinstance(self._payload, dict)

开发

克隆，以可编辑模式安装带有开发需求，并安装 pre-commit 挂钩

git clone https://github.com/stac-utils/stac-task
cd stac-task
pip install -e '.[dev]'
pre-commit install

运行测试

pytest

检查所有文件

pre-commit run --all-files

贡献

使用 Github 问题 和 拉取请求。