Python API

ionworks-api Python パッケージは、パイプラインの実行と管理のための client.pipeline を提供します。インストールと認証については Python API クライアントを参照してください。

パイプラインの送信

import ionworks_schema as iws
from ionworks import Ionworks

# 環境変数 IONWORKS_API_KEY と IONWORKS_PROJECT_ID を読み込みます
client = Ionworks()

pipeline = iws.Pipeline(
    {
        "known": iws.direct_entries.DirectEntry(
            parameters={"Ambient temperature [K]": 298.15},
        ),
        "Q_pos": iws.calculations.ElectrodeCapacity(electrode="positive"),
    },
    name="Capacity pipeline",
)

submission = client.pipeline.create(pipeline)
print(f"Pipeline ID: {submission.id}")
print(f"Status: {submission.status}")

client.pipeline.create() は iws.Pipeline インスタンスでも .to_config() が返す dict でも受け取れます。スキーマインスタンスは送信前にローカルで検証されるため、形状エラーは即座に表面化します。

パイプラインを JSON にシリアライズする

送信前のペイロード確認、キャッシュ、バージョン管理へのコミット、あるいは別プロセスへの受け渡しなど、パイプラインを JSON として扱いたい場合は .to_config() を使用してください。

import json

config = pipeline.to_config()

# 内容の確認や保存
with open("pipeline_config.json", "w") as f:
    json.dump(config, f, indent=2)

# 後で送信することも、別プロセスから送信することもできます
submission = client.pipeline.create(config)

.to_config() はサポートされている唯一のシリアライザです。パイプライン要素と目的関数に必要な判別子(トップレベルの要素は element_type、ネストされたスキーマはそれぞれの type)を出力し、スキーマのフィールド名マッピング(例: data_input → data)を適用します。API ペイロードの構築に Pydantic の model_dump() を 使用しないでください。model_dump() はこれらの判別子を落とし、フィールド名マッピングを適用しないため、API が拒否する可能性のある dict を生成します。

送信メタデータの上書き

create() は、スキーマに含まれる値を上書きするための project_id、name、description、options を kwargs として受け付けます。

submission = client.pipeline.create(
    pipeline,
    project_id="your-project-id",
    name="Custom run name",
    options={"live_progress_updates": True},
)

project_id を省略した場合、クライアントは Ionworks(...) のデフォルト、または環境変数 IONWORKS_PROJECT_ID にフォールバックします。

完了待ち

submission = client.pipeline.wait_for_completion(
    submission.id,
    timeout=600,        # 秒 (デフォルト: 600)
    poll_interval=2,    # ポーリング間隔 [秒] (デフォルト: 2)
    verbose=True,       # ステータス出力 (デフォルト: True)
)

パイプラインが失敗した際に例外を発生させず失敗レスポンスを返すには、raise_on_failure=False を指定します。

結果の取得

result = client.pipeline.result(submission.id)

# パイプラインが生成した最終パラメータ値
print(result.result)

# Pipeline 内で使用した要素名をキーとする出力
print(result.element_results["Q_pos"])

result.element_results のキーは iws.Pipeline(elements=...) に渡したキーと対応します。

要素のメタデータ

一部の要素(特に Validation)は element_results に含まれない追加メタデータを書き込みます。それらは次で取得できます。

metadata = client.pipeline.get_element_metadata(submission.id, "validate")

データフィッティングのパラメータトレース

データフィッティング要素では、オプティマイザの反復ごとの進行状況が要素ジョブのメタデータに記録されます。要素のジョブ ID を渡して client.job.get_parameter_trace で取得できます。詳細なスキーマはパラメータトレースの取得を参照してください。要素の job_id は element_results ではなくパイプラインの要素一覧に含まれているため、一覧を取得してデータフィッティング要素を名前で取り出します。

elements = client.get(f"/pipelines/{submission.id}/elements")
fit_job_id = next(e["job_id"] for e in elements if e["name"] == "fit")

trace = client.job.get_parameter_trace(fit_job_id)

パイプライン一覧

# Ionworks(...) のデフォルトまたは IONWORKS_PROJECT_ID を使用
pipelines = client.pipeline.list()

# 1 回の呼び出しだけプロジェクトを上書き
pipelines = client.pipeline.list(project_id="other-project-id")

# 件数を制限
pipelines = client.pipeline.list(limit=10)

単一送信の取得

submission = client.pipeline.get(submission.id)
print(submission.status)  # "pending", "running", "completed", "failed"

SimplePipeline

SimplePipeline は、高コストな要素が最大 1 つ（単一の DataFit または Validation）のワークフロー向けの、Pipeline の軽量な代替手段です。ファイア・アンド・フォーゲット方式で実行され、parameter_values、cost、そして（検証の場合は）summary_stats を含むフラットな結果を返します。送信とポーリングは client.simple_pipeline を通じて行います。

設定の構築

SimplePipeline は Pipeline からすべてを継承し、高コストな要素を複数含む設定を拒否するクライアントサイドの検証を追加します。

import ionworks_schema as iws

objective = iws.objectives.CurrentDriven(data_input="path/to/discharge.csv")

pipeline = iws.SimplePipeline(
    elements={
        "initial_params": iws.direct_entries.DirectEntry(
            parameters={"Negative particle diffusivity [m2.s-1]": 2e-14},
        ),
        "fit": iws.DataFit(
            objectives={"cycle": objective},
            parameters={
                "Negative particle diffusivity [m2.s-1]": iws.Parameter(
                    "Negative particle diffusivity [m2.s-1]",
                    bounds=(1e-14, 1e-13),
                    initial_value=2e-14,
                )
            },
            cost=iws.costs.RMSE(),
            optimizer=iws.parameter_estimators.ScipyDifferentialEvolution(
                maxiter=10
            ),
        ),
    },
    name="My Fit",
)

config = pipeline.to_config()

複数の DataFit または Validation 要素を渡すと、SimplePipeline は即座に ValueError を発生させます。サーバー側で拒否されるのを待つ必要はありません。

送信とポーリング

from ionworks import Ionworks

client = Ionworks()

# 設定を送信
sp = client.simple_pipeline.create(config, name=pipeline.name)
# sp.status == "pending"

# 完了を待機（自動的にポーリング）
result = client.simple_pipeline.wait_for_completion(sp.id, timeout=600)

# 結果を読み取り
print(result.result["parameter_values"])
# {"Negative particle diffusivity [m2.s-1]": 5.3e-14}
print(result.result["cost"])

検証パイプライン

SimplePipeline は単一の Validation 要素もサポートします。結果には parameter_values に加えて summary_stats が含まれます。

objective = iws.objectives.CurrentDriven(data_input="path/to/cycle.csv")

validation_pipeline = iws.SimplePipeline(
    elements={
        "validate": iws.Validation(
            objectives={"cycle": objective},
        ),
    },
    name="Validate fitted model",
)

sp = client.simple_pipeline.create(
    validation_pipeline.to_config(), name=validation_pipeline.name
)
result = client.simple_pipeline.wait_for_completion(sp.id, timeout=600)
print(result.result["summary_stats"])

エンドツーエンドの例

import pybamm
import ionworks_schema as iws
from ionworks import Ionworks

client = Ionworks()

pipeline = iws.Pipeline(
    {
        "known": iws.direct_entries.DirectEntry(
            parameters={"Ambient temperature [K]": 298.15},
        ),
        "fit": iws.DataFit(
            objectives={
                "1C": iws.objectives.CurrentDriven(
                    data_input="file:examples/data/chen_synthetic_1C/time_series.csv",
                    options={"model": pybamm.lithium_ion.SPMe()},
                ),
            },
            parameters={
                "Negative particle diffusivity [m2.s-1]": iws.Parameter(
                    "Negative particle diffusivity [m2.s-1]",
                    initial_value=2e-14,
                    bounds=(1e-14, 1e-13),
                ),
            },
            cost=iws.costs.RMSE(),
            optimizer=iws.optimizers.DifferentialEvolution(),
        ),
    },
    name="SPMe diffusivity fit",
)

submission = client.pipeline.create(pipeline)
client.pipeline.wait_for_completion(submission.id)

result = client.pipeline.result(submission.id)
print(result.element_results["fit"])

より多くのエンドツーエンド例(エントリのみ、計算のみ、データフィット、検証)は SDK リポジトリの packages/ionworks-api/examples/pipeline/ を参照してください。

はじめに

基本コンセプト

データ

モデル構築

パイプライン

シミュレーション

最適化

パイプラインの送信

パイプラインを JSON にシリアライズする

送信メタデータの上書き

完了待ち

結果の取得

要素のメタデータ

データフィッティングのパラメータトレース

パイプライン一覧

単一送信の取得

SimplePipeline

設定の構築

送信とポーリング

検証パイプライン

エンドツーエンドの例

​パイプラインの送信

​パイプラインを JSON にシリアライズする

​送信メタデータの上書き

​完了待ち

​結果の取得

​要素のメタデータ

​データフィッティングのパラメータトレース

​パイプライン一覧

​単一送信の取得

​SimplePipeline

​設定の構築

​送信とポーリング

​検証パイプライン

​エンドツーエンドの例

パイプラインの送信

パイプラインを JSON にシリアライズする

送信メタデータの上書き

完了待ち

結果の取得

要素のメタデータ

データフィッティングのパラメータトレース

パイプライン一覧

単一送信の取得

SimplePipeline

設定の構築

送信とポーリング

検証パイプライン

エンドツーエンドの例