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 にフォールバックします。

パイプライン内のデータ参照

パイプライン設定でデータソースを参照するには、これらのプレフィックスを使用します:

プレフィックス	例	説明
`db:`	`"db:measurement-id"`	アップロードされた測定を ID で参照
`file:`	`"file:data.csv"`	ローカル CSV ファイルを読み込み
`folder:`	`"folder:data_dir/"`	ローカルディレクトリから読み込み

パイプライン設定内のインライン DataFrame には 1,000 行の制限があります。より大きなデータセットはまず測定としてアップロードし、db:measurement-id で参照してください。

folder: スキームは time_series ファイルと steps ファイルを含むディレクトリを想定しています。.parquet と .csv の両方がサポートされ、両方が存在する場合は parquet が優先されます。例えば、time_series.parquet と steps.parquet（または .csv）を含むフォルダは正しく読み込まれます。

PyBaMM モデルのサポート

モデルを名前で指定する代わりに、PyBaMM モデルオブジェクトを目的関数に直接渡すこともできます。設定を構築する際に自動的にシリアライズされるため、登録手順も文字列名の参照も必要ありません。

import pybamm
import ionworks_schema as iws

objective = iws.objectives.CurrentDriven(
    data_input="path/to/discharge.csv",
    options={"model": pybamm.lithium_ion.DFN()},
)

モデルのオプションはオブジェクトと一緒に渡されるため、設定済みのモデルはそのまま設定済みの状態で届きます。

model = pybamm.lithium_ion.DFN(options={"thermal": "lumped"})

objective = iws.objectives.CurrentDriven(
    data_input="path/to/discharge.csv",
    options={"model": model},
)

完了待ち

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)

データフィットのモデル vs データプロットデータ

client.job.get_plot_data は、データフィットジョブのモデル vs データのトレース、つまり Studio がフィット結果ページに描画するのと同じオーバーレイを返します。データフィットはベストフィットのパラメータで検証を再実行し、そのオーバーレイを自身のメタデータに保存するため、別の Validation 要素を追加したり、生のメタデータ blob を解析したりせずに、フィットのジョブ ID から直接取得できます。

plot = client.job.get_plot_data(
    fit_job_id,
    objective_name="1C",   # DataFit の objectives マッピングで使用したキー
)

ノートブックやレポートでフィットのオーバーレイを再現したり、独自のプロットパイプラインに渡したりする際に使用してください。トレースはサーバー側で系列ごとに最大 max_points 点までダウンサンプリングされます（デフォルトは 2000、範囲は 100–10000）。ユーザーがズームインするたびにより詳細を再取得する セマンティックズーム を実装したい場合は、現在のビューポートに対応する x_min と x_max を渡します。

zoomed = client.job.get_plot_data(
    fit_job_id,
    objective_name="1C",
    x_min=1200.0,
    x_max=1800.0,
    max_points=5000,
)

パラメータ	説明
`job_id`	オーバーレイを取得するデータフィットジョブ。
`objective_name`	`DataFit` の `objectives` マッピングのキー（上記例では `"1C"`）。
`plot_index`	目的関数が複数のプロットを定義している場合の、そのリストへのインデックス。デフォルトは `0`。
`max_points`	トレースごとに返される最大点数（`100`–`10000`）。デフォルトは `2000`。
`x_min`, `x_max`	x 範囲の下限・上限（両端を含む）。全範囲の場合は省略します。

パイプライン一覧

# 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、ArrayDataFit、または Validation）のワークフロー向けの、Pipeline の軽量な代替手段です。ファイア・アンド・フォーゲット方式で実行され、parameter_values、cost、そして（検証の場合は）summary_stats を含むフラットな結果を返します。送信とポーリングは client.simple_pipeline を通じて行います。

設定の構築

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

import ionworks_schema as iws

objective = iws.objectives.CurrentDriven(
    data_input="file:path/to/discharge.csv", options={"model": "SPM"}
)
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、ArrayDataFit、または Validation 要素を渡すと、SimplePipeline は即座に ValueError を発生させます。サーバー側で拒否されるのを待つ必要はありません。

送信とポーリング

from ionworks import Ionworks

client = Ionworks()

# スキーマインスタンスを直接渡して送信（.to_config() の dict も可）
sp = client.simple_pipeline.create(pipeline, 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"])

client.pipeline.wait_for_completion と同様に、poll_interval（ポーリング間隔 [秒]、デフォルト 2）、verbose（ステータス出力、デフォルト True）、raise_on_failure（デフォルト True。False を指定すると、例外を発生させず失敗レスポンスを返します）も引数として受け付けます。実行が終端ステータス — completed、failed、canceled のいずれか — に達すると完了します。 client.simple_pipeline.create() は client.pipeline.create() と揃えられており、iws.SimplePipeline スキーマインスタンス、または .to_config() が返す dict のいずれも受け取れます。スキーマインスタンスを推奨します: 自分で .to_config() を呼び出す必要がなく、スキーマオブジェクトを構築した時点で形状エラーがローカルに表面化します。生の dict はそのまま送信されるため、不正な dict はサーバー側で HTTP 422 として初めて拒否されます。

一覧、フィルタリング、ソート

list は items、count、total を含むページ分割されたレスポンスを返します。文字列フィルタは、完全な値または ilike.%foo%（大文字小文字を区別しない部分一致）や in.(completed,failed)（集合のいずれかに一致）のような演算子プレフィックス付きの式を受け入れます。

# Newest 25 simple pipelines in the default project
page = client.simple_pipeline.list(limit=25)

# Only currently active runs
active = client.simple_pipeline.list(status="in.(pending,running)")

# Name contains "diffevo" (case-insensitive)
matches = client.simple_pipeline.list(name="ilike.%diffevo%")

# Created in the last week, oldest first
recent = client.simple_pipeline.list(
    created_at_gt="2026-05-08",
    created_at_lt="2026-05-15",
    order_by="created_at",
    order="asc",
)

実行オプション

シンプルパイプライン内の data_fit 要素は、通常のパイプラインと同じ分散ワーカープールを使って並列に評価されます。追加の設定は不要です。通常どおり optimizer.population_size を設定すれば、サーバーが集団の評価をワーカー間に分散します。

create に options 辞書を渡すことで、送信したパイプラインの実行時の挙動を制御できます。オプションは送信メタデータであり、サーバーがジョブをどのように実行するかに影響しますが、パイプライン設定の一部としては保存されません。

オプション	型	デフォルト	効果
`live_progress_updates`	`bool \| None`	`None`（ジョブタイプに応じてワーカーが適切なデフォルトを選択）	`True` の場合、ワーカーは実行中にチェックポイントの進捗をデータベースに書き込み、途中経過をポーリングできるようにします。`False` の場合、パフォーマンス向上のためチェックポイントをスキップします。

sp = client.simple_pipeline.create(
    config,
    name="NMC622 diffusivity fit",
    options={"live_progress_updates": False},  # 速度のためチェックポイントをスキップ
)

options（および project_id、name、description）は設定辞書に直接埋め込むこともできます。create は送信前にこれらを設定から取り出します。create に明示的に渡された引数は、設定内の値より優先されます。

config = {
    "options": {"live_progress_updates": True},
    "elements": { ... },
}

sp = client.simple_pipeline.create(config)

更新、キャンセル、削除

# Rename or add a description (PATCH — at least one field is required)
client.simple_pipeline.update(sp.id, name="Renamed", description="notes")

# Cancel a running pipeline
client.simple_pipeline.cancel(sp.id)

# Permanently delete the pipeline, its job, and stored config
client.simple_pipeline.delete(sp.id)

エラー処理

from ionworks.errors import IonworksError

try:
    result = client.simple_pipeline.wait_for_completion(sp.id)
except TimeoutError:
    # Still running after the timeout — fetch the latest status to decide
    current = client.simple_pipeline.get(sp.id)
    print(f"Still {current.status}")
except IonworksError as e:
    # Pipeline ended in "failed" — the message includes the server error
    print(e)

検証パイプライン

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

objective = iws.objectives.CurrentDriven(
    data_input="file:path/to/cycle.csv", options={"model": "SPM"}
)
validation_pipeline = iws.SimplePipeline(
    elements={
        "validate": iws.Validation(
            objectives={"cycle": objective},
        ),
    },
    name="Validate fitted model",
)

sp = client.simple_pipeline.create(
    validation_pipeline, 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 にシリアライズする

送信メタデータの上書き

パイプライン内のデータ参照

PyBaMM モデルのサポート

完了待ち

結果の取得

要素のメタデータ

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

データフィットのモデル vs データプロットデータ

パイプライン一覧

単一送信の取得

SimplePipeline

設定の構築

送信とポーリング

一覧、フィルタリング、ソート

実行オプション

更新、キャンセル、削除

エラー処理

検証パイプライン

エンドツーエンドの例

​パイプラインの送信

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

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

​パイプライン内のデータ参照

​PyBaMM モデルのサポート

​完了待ち

​結果の取得

​要素のメタデータ

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

​データフィットのモデル vs データプロットデータ

​パイプライン一覧

​単一送信の取得

​SimplePipeline

​設定の構築

​送信とポーリング

​一覧、フィルタリング、ソート

​実行オプション

​更新、キャンセル、削除

​エラー処理

​検証パイプライン

​エンドツーエンドの例

パイプラインの送信

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

送信メタデータの上書き

パイプライン内のデータ参照

PyBaMM モデルのサポート

完了待ち

結果の取得

要素のメタデータ

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

データフィットのモデル vs データプロットデータ

パイプライン一覧

単一送信の取得

SimplePipeline

設定の構築

送信とポーリング

一覧、フィルタリング、ソート

実行オプション

更新、キャンセル、削除

エラー処理

検証パイプライン

エンドツーエンドの例