tilebox-workflow-authoring

Tilebox ワークフローの作成

Python ワークフローのソースコードと、ローカル／ランタイムでの動作を作成または変更します。プラットフォーム API は Tilebox のリファレンスに、移植可能な科学処理と IO は地理空間関連のリファレンスに、エンドツーエンドの組み合わせはレシピに記載してください。

最新の API と適切なスキルから始める

API が不明な場合や、記憶している動作が古くなっている可能性がある場合は、インストール済みのバージョンを確認するか、最新のドキュメントを検索してください。

tilebox docs search "Task ExecutionContext submit_subtasks"
tilebox docs search "logging tracing context.logger context.tracer"
tilebox docs search "caches job_cache"

関連する作業は、次の担当に委任してください。

• tilebox-workflow-releases：プロジェクトの初期化、pyproject.toml のひな型、ビルド、公開、デプロイ、ランナーの設定。
• tilebox-workflow-jobs：ジョブの送信、待機、一覧表示、確認、再試行、キャンセル、クラスター操作。
• tilebox-datasets：ソース／プロダクトの推奨、正確なデータセットのスラッグ、プロバイダーの認証情報、補助カタログ、CLI による調査。
• tilebox-cli：認証、CLI の確認、JSON 出力、ドキュメントの検索。
• tilebox-workflow-automations：cron とストレージをトリガーとする自動化。

新しいプロジェクトでは、tilebox-workflow-releases を通じて tilebox workflow init を使用してください。プロジェクト設定を手作業でひな型から作成しないでください。初期化後は、生成されたコードを編集できます。

コーディングの前に計画する

成果を重視する地球観測の作業では、reference/geospatial/project-planning.md を読んでください。解像度、時間的な対象範囲、観測品質、比較可能性、必要な入力、検証、成果物を確認してください。ソースの選定と調査には tilebox-datasets を使用し、そのカタログや認証情報に関するガイダンスを重複して記載しないでください。

タスクグラフの概要を作成します。

1. ルート、ワーカー、集約の各ステージを特定します。
2. ファンアウトの軸を選択します。シーン、プロダクト、AOI、時間ウィンドウ、空間チャンク、モデルのタイルなどが該当します。
3. 実際に必要な依存関係を depends_on で示し、不要な連鎖は避けます。
4. タスクのフィールド、context.job_cache、永続オブジェクト／Zarr アーティファクトのどれを使うか選択します。
5. 冪等性のあるネットワーク／ストレージ操作に対する再試行の動作を選択します。

実装を小さく保つ

• リクエストを満たす最もシンプルな結果から始めます。正確性のため、または反復作業で求められた場合に限り、より具体的な選択、品質ポリシー、処理ステージを追加します。
• ワークフローを求められた場合は、実行可能な Tilebox タスクとランナー登録を提供します。ローカルプレビューで処理を途中検証してもかまいませんが、単独のスクリプトだけで終わらせたり、ユーザーが実装の各段階を個別に求めることが前提になったりしないようにします。
• 高コストな処理の前に、求められた結果に必要な入力を選択します。デフォルトですべての候補プロダクトを計算しないでください。
• 同じデータポイントを二度クエリしないようにします。選択したアセット／メタデータを利用側で使えるようキャッシュするか、時間範囲を分割してリーフタスク内でクエリします。
• 作業セットがメモリに収まる場合は、各アセットの AOI 全体を一度だけ読み込みます。読み込みの分割は、明示的な空間チャンク化の一部として行い、小さな出力ブロックに合わせるためだけに行わないでください。
• 実行ごとの選択肢はタスク入力として公開します。ユーザーが設定する必要のない実装や表示のデフォルト値は、それらを扱う関数内に保持します。
• 可読性、テスト容易性、再利用性が向上する場合は処理関数を抽出します。単純な処理は execute 内に置いてもかまいません。実際に共通する操作には関数を再利用し、差異はパラメーター化します。単純な API ラッパーはインライン化し、先走ってフレームワークを作らないでください。
• グリッド、配列のマスキング、再投影、画像の合成、エンコードには、独自実装を書く前に既存ライブラリの操作を使います。キャリブレーション、nodata、位置合わせ、科学的な検証は明示します。
• 外部入力と科学的な前提条件を検証します。予期しないエラーは Tilebox に伝播させ、根拠のないフォールバックや独自の再試行レイヤーを追加しないでください。
• 求められた成果物と、グラフで使用される中間成果物を生成します。ユーザーから求められない限り、マニフェスト、来歴／品質のサイドカー、追加のエクスポートは追加しないでください。役立つソース ID、パラメーター、品質情報は、既存のアーティファクトのメタデータまたは構造化ログに保持します。

責務に基づく構成

複数段階のワークフローでは、runner.py はタスクの登録とランナーの設定に限定してください。タスクコードと処理関数は責務ごとにまとめます。固定のレイアウトに従ったり、タスクごとにファイルを作成したりするのではなく、有用な境界となる場合にのみモジュールを追加してください。小規模なプロトタイプなら、単一ファイルでも問題ありません。

選択、マスキング、変換、エンコードなどの実質的な処理関数は、ExecutionContext をモックせずにテストしてください。テストのためだけに単純なヘルパーを切り出さないでください。オーケストレーションは、リリースビルドとタスクの検出、および代表的なジョブの実際のグラフを通じて検証してください。サブミッション、進捗、ログ記録を対象とするタスククラスのテストでは検証しないでください。実行時に必要なすべてのモジュールをリリースに含め、ビルド済み成果物からインポートできることを確認してください。

タスクレベルの並列処理を必須にする

タスクレベルのファンアウトはワークフロー設計上の要件であり、任意の最適化ではありません。要求される出力とワークロードに基づいて、シーン、期間、AOI、製品、空間チャンク、モデルタイルなどのファンアウト軸を選択してください。その軸上の独立した単位は、それぞれ context.submit_subtask(s) でサブミットする個別の Tilebox タスクとして表現します。この分解は成果物から推測してください。submit_subtasks の指定、タスク数の指定、DAG の指示をユーザーに求めてはいけません。たとえば、すべてのフレームを結合する 12 シーンのタイムラプスには、12 回のループを実行する 1 つのタスクではなく、1 root + 12 scene workers + 1 encoder のようなグラフが必要です。

同じシーンや期間でも、異なるソースアセットを使って独立して計算する出力は、それぞれ別のタスクに分けてください。メタデータや共通の QA マスクを共有するからといって、逐次実行にする必要はありません。同じ高コストな読み取りや計算を再利用する出力は、まとめてください。

独立した出力ラスタが複数ある小規模なワークフローでは、比較対象の日付ごとに RGB や NBR などの TIFF 出力単位でファンアウトします。各ワーカーの出力、読み取り、処理用一時データが、想定するランナーの同時実行数におけるワーカーあたりのメモリ予算に収まる場合、ワーカーは選択したグラニュールから複数のバンドを読み込み、メモリ上でモザイクを作成できます。これにより、グラニュールごとのタスクやモザイク用の Zarr 中間データを必要とせずに、出力レベルの並列処理を維持できます。

より広い領域や入力セットを扱うために分散モザイク処理が必要な場合は、グラニュール単位でファンアウトします。大きすぎるグラニュールの処理は上限のあるウィンドウに分割し、その後、出力チャンクごとに集約します。グラフを変更する際も、要求された合成ルールは変更しないでください。チャンク単位のリダクションについては、reference/tilebox/geospatial-task-graphs.md を参照してください。

オーケストレーションタスクを使って独立したワーカーをサブミットし、出力を結合する場合は depends_on を使って集約タスクや公開タスクを設定してください。出力ワーカー内でのループや非同期読み取りは許可されていますが、独立した出力間のファンアウトや、より大きなモザイクに必要なグラニュール／チャンク単位のファンアウトの代わりにはなりません。このファンアウトを、1 つのタスク内の asyncio.gather、スレッド、ローカルのマルチプロセス処理で置き換えてはいけません。追加のランナーが並列化できるのは、個別の Tilebox タスクだけです。初回実行の規模が小さいことやスケジューリングのオーバーヘッドを理由に、ワークフローを 1 つのタスクにまとめてはいけません。粒度が細かすぎる作業は、負荷のバランスが取れた複数のワーカーにまとめてください。単一タスクのワークフローを使えるのは、作業を本質的に分割できない場合、依存関係の順序が決まっている場合、またはユーザーが明示的に要求した場合に限ります。

コーディングの前に、想定するグラフの構成を記録してください。代表的なジョブの実行後に、execution_stats.total_tasks と task_summaries をその構成と照合してください。進捗単位や子トレーススパンはタスクではなく、ファンアウトの証明にもなりません。グラフが一致しない場合は、完了を報告する前にワークフローを修正してください。ランナーの処理能力に余裕がある場合は、タスクのタイムスタンプやスパンを使って、独立したワーカーが重複して実行されたことを確認してください。

タスクの基本的な意味については reference/tilebox/tasks-and-graphs.md、地理空間のステージの対応付けについては reference/tilebox/geospatial-task-graphs.md、境界については reference/tilebox/state-and-artifacts.md を参照してください。

型付きタスクを定義する

Task はデータクラスの振る舞いを適用します。フィールドはシリアライズされる入力です。

各フィールドには、その内容を表現できる最もシンプルで慣用的な型を使用してください。表現を新たに考案する前に、Python、Tilebox、または処理ライブラリに既存の値型がないか確認してください。シリアライズ可能にするためだけに、確立された型をカスタムデータクラス、名前付きタプル、場当たり的なタプル／リスト／辞書形式として再現しないでください。時間範囲を表す tuple[datetime, datetime] のように、自然な Python の形式であれば通常のコンテナを使用できます。対応する型と例については、reference/tilebox/tasks-and-graphs.md をお読みください。

from tilebox.workflows import ExecutionContext, Task


class ProcessScene(Task):
    scene_id: str

    @staticmethod
    def identifier() -> tuple[str, str]:
        return "tilebox.com/example/ProcessScene", "v1.0"

    def execute(self, context: ExecutionContext) -> None:
        context.current_task.display = f"ProcessScene({self.scene_id})"
        context.logger.info("Processing scene", scene_id=self.scene_id)

• プロトタイプでは、デフォルトの v0.0 識別子を使用できます。安定した識別子は (name, vX.Y) を返します。
• リリース前の反復中は初期バージョンを維持してください。互換性が重要になったら、マイナーバージョンは後方互換とし、入力または動作に互換性のない変更を加える場合はメジャーバージョンを上げてください。
• シリアライズされる入力全体を 2048 バイト以下にしてください。コンパクトなプレーンな Python 値、対応ライブラリの値、ID、キー、小規模な設定値を渡し、配列、データフレーム、xarray データセット、大きなジオメトリ、マニフェスト、認証情報、クライアント、開いたファイル、ローカルパスは渡さないでください。reference/tilebox/state-and-artifacts.md に従って、context.job_cache または永続ストレージを使用してください。
• ジョブで使用するすべてのタスククラスをランナーに登録してください。

タスク内で非同期 I/O を直接待機する

tilebox.storage.aio、非同期 HTTP クライアント、上限付きの並行リクエストなど、非同期 API を使用するタスクでは必ず async def execute(self, context: ExecutionContext) -> None を定義してください。これらの API は直接待機し、asyncio.run(...) でラップしないでください。ワークフロー実行エンジンは非同期の execute メソッドを待機し、通常の失敗、再試行、トレース、完了の動作を維持します。

import asyncio

from niquests import AsyncSession


class FetchMetadata(Task):
    urls: list[str]

    async def execute(self, context: ExecutionContext) -> None:
        async with AsyncSession() as client:
            responses = await asyncio.gather(
                *(client.get(url) for url in self.urls)
            )
        for response in responses:
            response.raise_for_status()
        context.logger.info("Metadata fetched", count=len(responses))

クライアントのライフタイムは、そのイベントループでの動作に応じて選択してください。niquests.AsyncSession は execute 内で作成し、tilebox.storage.aio.Client はタスクの実行をまたいで再利用できます。詳細については、reference/tilebox/tasks-and-graphs.md を参照してください。大量のリクエストは、セマフォまたは上限付きバッチで制限してください。非同期の並行処理は、1 つのシーンのバンドやバイト範囲、またはメモリ上の上限付きモザイクなど、選択したワーカーユニット内で複数の I/O リクエストを処理するのに適しています。これは必須のタスクレベルのファンアウトに代わるものではありません。runner.run_all() と runner.run_forever() は同期ランナーのエントリーポイントに置き、非同期コードから呼び出さないでください。

シンプルな動的グラフを構築する

class ProcessScenes(Task):
    scene_ids: list[str]

    def execute(self, context: ExecutionContext) -> None:
        context.current_task.display = f"ProcessScenes(n={len(self.scene_ids)})"
        workers = context.submit_subtasks(
            [ProcessScene(scene_id) for scene_id in self.scene_ids],
            max_retries=3,
        )
        context.submit_subtask(PublishSummary(), depends_on=workers)

• 子タスクが 1 つの場合は submit_subtask を、同種のバッチには submit_subtasks を使用してください。
• ステージ間のバリアを共有する方法を優先してください。すべての A タスクを送信してから、アイテムごとに個別の B[i] → A[i] 依存関係グループを作成するのではなく、depends_on=all_a_handles を指定して B のバッチを送信してください。下流のタスクが 1 つのプロデューサーの出力だけを使用する場合は、関係のない他のタスクではなく、そのプロデューサーに依存させてください。
• 重要度の低い作業にのみ optional=True を使用してください。
• 副作用は冪等にしてください。決定的なキー、上書き可能な書き込み、有効な出力のチェック、またはアトミックなコミットを使用します。再試行では元のタスク入力が使用されます。

進捗と可観測性

コストの高い処理を始める前に、current_task.display に簡潔なラベルを設定してください。意味のあるファンアウトでは、送信側のタスクが context.progress(name).add(n) を呼び出し、各ワーカーは自身が担当するユニットの処理を完了した後に .done(1) を呼び出します。合計数と完了数は一致させてください。

構造化フィールドには context.logger を使用し、繰り返し使うコンテキストには logger.bind を使用してください。コストの高い処理段階は context.tracer.span(...) で囲み、子スパンが必要なヘルパーにはそのトレーサーを渡してください。グローバルな OpenTelemetry トレーサーを作成しないでください。例外ハンドラーは、復旧または有用なコンテキストの追加が必要な場合にのみ設けてください。例外は logger.exception で記録してから再送出します。コンソールログはタスククラスではなく、ランナーのエントリーポイントで設定してください。

データセット、アセット、ストレージの境界

• ソース選択後のクエリ処理については、reference/tilebox/datasets-and-datapoints.mdを参照してください。
• アセットのデコードとメタデータについては、reference/tilebox/assets.mdを参照してください。
• アセットのバイトデータへのアクセスと、範囲を限定したCOGウィンドウについては、reference/tilebox/storage-access.mdを参照してください。

プロバイダーのパスを再構築したり、ストレージAPIのスニペットを繰り返したり、ソースやプロバイダーのセットアップを埋め込んだりしないでください。スケール／オフセット、nodata、マスク、アラインメント、再投影を明示的に適用してください。ダウンロードは利用側のタスクで行い、別のタスクにはローカルパスではなく、永続的なIDやキーを渡してください。

依存関係と運用

実行時の依存関係をpyproject.tomlに宣言してください。タスクフィールドに型が現れるオプションライブラリも含めます。Tileboxはそれらのシリアライザーを遅延読み込みしますが、ライブラリ自体はインストールしません。uv syncを実行し、編集可能な依存関係やローカルパスへの依存関係は避けてください。詳しくはreference/tilebox/dependencies-and-packaging.mdを参照してください。ソースの作成と、対象を絞ったローカルチェックにのみオーサリングを使用してください。初期化／ビルド／デプロイにはtilebox-workflow-releasesを、ジョブの送信／待機／クラスターにはtilebox-workflow-jobsを使用してください。

リファレンスの案内

Tilebox プラットフォーム

リファレンス	使用する場面
reference/tilebox/tasks-and-graphs.md	タスクの定義／バージョン管理、サポートされる入力型の選択、入力上限の遵守、依存関係の送信、再試行、進捗状況、可観測性、登録、ランナーモード。
reference/tilebox/datasets-and-datapoints.md	選択済みのデータセットへのクエリ、サンプルの確認、データポイントを正確に1つ選択、またはデータポイントの反復処理。
reference/tilebox/assets.md	AssetCollectionのデコード、セマンティックキー、ラスターメタデータ、スケール／オフセット、明示的なオーバーライド。
reference/tilebox/storage-access.md	Tileboxアセットの解決／読み取り／ストリーミング／ダウンロード／オープン、アクセス・ポリシー、匿名アクセス、ウィンドウ、並行処理。
reference/tilebox/state-and-artifacts.md	タスク入力、ジョブキャッシュ、オブジェクト／Zarrアーティファクト、ローカルのスクラッチ領域、再試行に安全なキーの選択。
reference/tilebox/dependencies-and-packaging.md	uv互換のワークフロー依存関係と、リリースに対応したパッケージの宣言。
reference/tilebox/geospatial-task-graphs.md	シーン、ウィンドウ、チャンク、リダクション、進捗状況をTileboxタスクに割り当てる方法。

Tileboxと地理空間処理を組み合わせたレシピ

リファレンス	使用する場面
reference/tilebox/recipes/sentinel-2-cog.md	公開L2AのRGBとSCLを読み込み、20 mのクラスを10 mのRGBにアラインメントする。
reference/tilebox/recipes/sentinel-1-sar.md	SARによる変化検出、洪水、または海事処理のタスクグラフを構築する。
reference/tilebox/recipes/time-series-composite.md	アラインメント済みのシーンを合成する、またはタイムラプスを生成する。
reference/tilebox/recipes/geospatial-ml.md	モデルタイルに処理を分散し、地理空間予測を集約する。

ポータブルな地理空間データ処理

Reference	使用する場面
reference/geospatial/project-planning.md	成果を、実現可能なデータ、検証、出力の要件に落とし込む場合。
reference/geospatial/raster-fundamentals.md	CRS、変換、形状、dtype、nodata、マスク、スケール、単位を扱う場合。
reference/geospatial/io/cloud-native-raster.md	正規のアセット以外から、クラウドネイティブなGeoTIFF/COGを直接読み込む場合。
reference/geospatial/io/object-storage.md	S3、GCS、Azure、ローカルストレージ、または互換ストレージでobstoreを使用する場合。
reference/geospatial/io/zarr.md	Zarrのスキーマ、チャンク、領域書き込み、ラベル付き読み込みを設計する場合。
reference/geospatial/io/cog-output.md	正しいCOGラスターを書き込み、必要に応じて入力データやライター設定をテストする場合。
reference/geospatial/io/formats-and-encoding.md	COG/Zarr/NetCDF/GeoParquetを選び、不可欠なエンコーディングの意味を保持する場合。
reference/geospatial/processing/grids-and-reprojection.md	グリッド、再投影、意味に即したリサンプリングを選ぶ場合。
reference/geospatial/processing/masking-and-qa.md	nodata、マスク、QA、クラス、モルフォロジーを扱う場合。
reference/geospatial/processing/time-series.md	時系列画像を位置合わせ、合成、スタイル設定、検証する場合。
reference/geospatial/processing/ml-inference.md	タイリング、モデルの読み込み、正規化、予測出力を扱う場合。
reference/geospatial/products/sentinel-1-sar.md	SAR製品の選択、ジオメトリ、単位、注意事項を理解する場合。
reference/geospatial/products/sentinel-2-l2a.md	L2Aのバンド、解像度、スケーリング、SCLを理解する場合。
reference/geospatial/products/landsat-collection-2.md	Landsat Level-2のバンド、スケーリング、QAビットを理解する場合。

検証チェックリスト

1. 提出されたすべてのタスクが登録され、識別子とバージョンが一致していること。
2. タスクのフィールドには、構造をカスタムデータクラスや場当たり的なコンテナーで重複して表現するのではなく、慣用的なPythonの型または既存ライブラリの値型を使用すること。
3. シリアライズされる各タスク入力が2048バイト以下であること。
4. タスク間のデータには適切な状態／アーティファクトの境界と決定論的なキーを使用すること。
5. 再試行可能な実行が再入可能で、入力との互換性を保つこと。
6. ファンアウトの大きいタスクにはラベル、有用な進捗情報、構造化ログを設定し、必要に応じてスパンを使用すること。
7. 依存関係を同期し、想定される環境でタスクモジュールをインポートできること。
8. スキャフォールドされたリリースプロジェクトでは、生成されたタスクコードを編集した後、tilebox workflow build-release --debug --json が成功すること。
9. 複雑なワークフローでは一貫性のあるパッケージモジュールを使用し、ランナーのエントリーポイントは登録とランナーレベルの設定に限定すること。
10. 有用な場合は、基盤となる重要な処理関数に焦点を絞ったテストを用意すること。Tileboxのタスクオーケストレーションは、タスククラスの単体テストではなく、ビルド／タスクの検出と代表的なジョブによって検証すること。
11. 地球観測の出力には、代表的な科学的検証、カバレッジ／位置合わせの確認、来歴、明記された制限事項を含めること。
12. 代表的なジョブの実際のタスク概要が、計画されたルート、ファンアウト、集約の各段階と一致すること。進捗カウンターや子スパンをタスクのファンアウトとして扱わないこと。