`ionworks.pipeline`¶

ionworks.pipeline provides Pydantic schemas for declaratively building configurations for the Ionworks battery parameterization and simulation stack. Every runtime class in the private ionworkspipeline engine has a matching schema here, so you can assemble a pipeline configuration from typed objects, serialize it to a config dict, and submit it to the Ionworks API without installing it.

For conceptual / mathematical background, see the Technical Guide.

Note

.to_config() is the only supported serializer. Pydantic’s own .model_dump() is not a wire format — it omits the type discriminator and emits Python attribute names rather than the wire names the API expects (data_input rather than data), so it returns a dict the Ionworks API may reject. Always build API payloads with .to_config().

Core classes¶

The top-level entry points for assembling a pipeline configuration:

class ionworks.pipeline.BaseSchema[source]

Bases: BaseModel

Shared parent of every schema class in this package.

Construct a concrete subclass (iw.pipeline.Pipeline, iw.pipeline.DataFit, …) rather than this. It rejects unknown fields, so a typo fails at construction, and provides .to_config(), the only supported serializer: .model_dump() omits the type discriminator and uses attribute names (data_input) rather than wire names (data).

model_config = {'arbitrary_types_allowed': True, 'extra': 'forbid', 'populate_by_name': True, 'validate_assignment': True}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

to_config() → dict[source]

Build the dict you submit through ionworks.api.

Nested schemas convert recursively, None fields are dropped, and each component the server must re-identify gains a "type" key.

Notes

The only supported serializer. .model_dump() (even with by_alias=True) omits the "type" discriminator, so it is not a payload: iw.pipeline.stats.Uniform(lb=0.0, ub=1.0).to_config() gives {"distribution": "Uniform", "lb": 0.0, "ub": 1.0}, whereas .model_dump(exclude_none=True) gives {"lb": 0.0, "ub": 1.0}.

run(*args, **kwargs)[source]

model_post_init(context: Any, /) → None

This function is meant to behave like a BaseModel method to initialize private attributes.

It takes context as an argument since that’s what pydantic-core passes when calling it.

Args:: self: The BaseModel instance. context: The context.

class ionworks.pipeline.Pipeline(elements: dict | None = None, output_file: str | None = None, name: str | None = None, description: str | None = None)[source]

Bases: BaseSchema

A sequence of steps that together produce a parameterised cell model.

Each element pulls parameters from a source (DirectEntry), fits them to data (DataFit), derives quantities (Calculation) or checks them against held-out data (Validation), and passes its parameters to the next, so order matters. Submit .to_config() through ionworks.api to run it server-side.

Parameters:

elements (dict, optional) – Mapping of element name to pipeline element (DataFit, DirectEntry, Validation, Calculation, …). The name is used to identify the element in the pipeline report. None is treated as an empty pipeline by to_config.
output_file (str, optional) – Optional file path for persisting the final parameter values produced by the pipeline. If None, parameters are not written to disk.
name (str, optional) – Human-readable name for the pipeline, used in reports and logs.
description (str, optional) – Free-text description of what the pipeline does.

Examples

>>> # known parameters (e.g. ambient temperature)
>>> known = iw.pipeline.direct_entries.DirectEntry(
...     parameters={"Ambient temperature [K]": 298.15},
... )
>>> # fit one parameter against an OCP measurement
>>> obj = iw.pipeline.objectives.OCPHalfCell(
...     electrode="positive",
...     data_input="path/to/ocp.csv",
... )
>>> fit = iw.pipeline.DataFit(
...     objectives={"ocp": obj},
...     parameters={"Positive electrode capacity [A.h]": iw.pipeline.Parameter(
...         "Positive electrode capacity [A.h]", initial_value=3.0, bounds=(2.0, 4.0),
...     )},
... )
>>> # validate the fit against held-out cycling data
>>> val_obj = iw.pipeline.objectives.CurrentDriven(
...     data_input="path/to/cycle.csv", options={"model": "SPM"}
... )
>>> val = iw.pipeline.Validation(objectives={"cycle": val_obj})
>>> pipeline = iw.pipeline.Pipeline(
...     {"known": known, "ocp fit": fit, "validate": val}
... )
>>> config = pipeline.to_config()
>>> # then submit `config` via ionworks.api

elements: dict | None

output_file: str | None

name: str | None

description: str | None

to_config() → dict[source]

Build the pipeline payload you submit through ionworks.api.

Walks every element you added to elements, serialises it (each element’s own to_config), and tags it with its kind ("entry", "data_fit", "validation", …) so the server knows how to dispatch it. Pass the returned dict to the Ionworks API to run the full pipeline.

model_config = {'arbitrary_types_allowed': True, 'extra': 'forbid', 'populate_by_name': True, 'validate_assignment': True}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_post_init(context: Any, /) → None

This function is meant to behave like a BaseModel method to initialize private attributes.

It takes context as an argument since that’s what pydantic-core passes when calling it.

Args:: self: The BaseModel instance. context: The context.

class ionworks.pipeline.DataFit(objectives, source='', parameters=None, cost=None, initial_guesses=None, optimizer=None, cost_logger=None, multistarts=None, objective_parallelism='auto', initial_guess_sampler=None, priors=None, options=None)[source]

Bases: BaseSchema

Fit a model’s parameters to measured experimental data.

objectives name the experiments and measured curves to match, parameters the values free to move, and optional priors what you already believe about them. The other fields (cost, optimizer, initial_guesses, multistarts, …) tune how the fit runs; their defaults suit most fits.

Parameters:

objectives (objective or dict, or a mapping of name to either) – What to fit against. Either a single objective (a CurrentDriven, MSMRHalfCell, … from iw.pipeline.objectives) or a dict of named objectives if the fit spans multiple experiments.
source (str, optional) – Free-text label for the data source (paper, dataset name, instrument). Shown in reports and provenance records.
parameters (dict[str, Parameter | pybamm.Symbol | callable] | None, optional) –
Which parameters are being fitted, and (optionally) how they relate to each other through pybamm expressions. At least one of parameters or priors must be set. Each value can be:
- an iw.pipeline.Parameter object, e.g. iw.pipeline.Parameter("x")
- a pybamm expression, in which case other referenced parameters must also be supplied as iw.pipeline.Parameter objects via pybamm.Parameter wrapping. For example:
```
{
    "param": 2 * pybamm.Parameter("half-param"),
    "half-param": iw.pipeline.Parameter("half-param"),
}
```
  works, but {"param": 2 * iw.pipeline.Parameter("half-param")} does not.
- a function that constructs a pybamm expression referencing other parameters, which must again be explicitly supplied as iw.pipeline.Parameter objects:
```
{
    "main parameter": lambda x: (
        pybamm.Parameter("other parameter") * x**2
    ),
    "other parameter": iw.pipeline.Parameter("other parameter"),
}
```
The dict key does not need to match the underlying pybamm parameter name — DataFit figures out which variable to fit from the iw.pipeline.Parameter reference.
cost (Cost or dict or None, optional) – How disagreement between model and data is summed up into a single number (e.g. sum-of-squares, log-likelihood). Give a cost schema instance (e.g. iw.pipeline.costs.SSE()) or a config dict ({"type": "SSE"}). Leave unset for a sensible default.
initial_guesses (dict[str, float] or list[dict[str, float]] or None, optional) – Starting point(s) for the optimiser. One dict applies to every restart; a list of dicts provides one starting point per restart.
optimizer (Optimizer or dict or None, optional) – Which optimisation algorithm to use (e.g. CMAES, PSO, ScipyMinimize). Leave unset for the default.
cost_logger (BaseSchema or dict or None, optional) – Optional logger that records the cost trajectory and parameter values across the fit, for later inspection.
multistarts (int | None, optional) – Number of independent restarts from different initial guesses. More restarts is more robust but takes longer.
objective_parallelism ({"auto", "on", "off"}, optional) – Whether to evaluate objectives in parallel. "auto" (default) lets the engine decide; "on"/"off" force or disable it. A debugging escape hatch.
initial_guess_sampler (DistributionSampler or dict or None, optional) – How to spread the multistart guesses across the parameter space (LatinHypercube by default).
priors (Prior or list[Prior] or dict or None, optional) – What you already believe about the parameter values. Acts as a regulariser on the fit. May be supplied alone (the prior names become the fit parameters) or alongside parameters (priors regularise the listed fit parameters).
options (DataFitOptions or dict or None, optional) – Runtime options for the fit — seed, low_memory, max_iterations, maxtime, validate, and skip_objective_callbacks. Pass a DataFitOptions or a plain dict; unknown keys are rejected. See DataFitOptions for each option’s meaning and default.

Examples

>>> # build the schema with the fields you care about
>>> obj = iw.pipeline.objectives.OCPHalfCell(
...     electrode="positive",
...     data_input="path/to/ocp.csv",
... )
>>> fit = iw.pipeline.DataFit(
...     objectives={"ocp": obj},
...     parameters={"Q_pe": iw.pipeline.Parameter(
...         "Positive electrode capacity [A.h]", initial_value=3.0, bounds=(2.0, 4.0),
...     )},
...     priors={"Q_pe": iw.pipeline.priors.Prior("Q_pe", iw.pipeline.stats.Normal(3.0, 0.2))},
... )
>>> config = iw.pipeline.Pipeline({"ocp fit": fit}).to_config()
>>> # then submit `config` via ionworks.api

objectives: Annotated[dict[str, Annotated[CalendarAgeing | CurrentDriven | CycleAgeing | EIS | ElectrodeBalancing | ElectrodeBalancingHalfCell | MSMRFullCell | MSMRHalfCell | OCPHalfCell | Objective | Pulse | Resistance, FieldInfo(annotation=NoneType, required=True, discriminator='type'), WrapValidator(func=_resolve_objective, json_schema_input_type=PydanticUndefined)]] | Annotated[CalendarAgeing | CurrentDriven | CycleAgeing | EIS | ElectrodeBalancing | ElectrodeBalancingHalfCell | MSMRFullCell | MSMRHalfCell | OCPHalfCell | Objective | Pulse | Resistance, FieldInfo(annotation=NoneType, required=True, discriminator='type'), WrapValidator(func=_resolve_objective, json_schema_input_type=PydanticUndefined)], FieldInfo(annotation=NoneType, required=True, metadata=[_PydanticGeneralMetadata(union_mode='left_to_right')])]

source: str

parameters: dict[str, Any] | None

cost: Annotated[RMSE | MAE | MSE | Max | SSE | Wasserstein | ChiSquare | MultiCost | GaussianLogLikelihood | DesignFunction, FieldInfo(annotation=NoneType, required=True, discriminator='type')] | None

initial_guesses: dict[str, int | float] | list[dict[str, int | float]] | None

optimizer: Annotated[DifferentialEvolution | CMAES | PSO | XNES | BayesianOptimization | SOBER | TuRBO | AskTellOptimizer | Nested | ScipyDifferentialEvolution | ScipyMinimize | ScipyLeastSquares | ScipyLsqLinear | ScipyBasinhopping | ScipyDualAnnealing | ScipyShgo | GridSearch | PointEstimateOptimizer | PointEstimateSampler | DummyOptimizer | DummySampler | PintsSampler, FieldInfo(annotation=NoneType, required=True, discriminator='type')] | None

cost_logger: Annotated[dict[str, Any] | BaseSchema | Any, FieldInfo(annotation=NoneType, required=True, metadata=[_PydanticGeneralMetadata(union_mode='left_to_right')])] | None

multistarts: int | None

objective_parallelism: Literal['auto', 'on', 'off']

initial_guess_sampler: Annotated[LatinHypercube | Uniform, FieldInfo(annotation=NoneType, required=True, discriminator='type'), WrapValidator(func=_resolve_sampler, json_schema_input_type=PydanticUndefined)] | None

priors: Annotated[dict[str, Any] | Prior | Any, FieldInfo(annotation=NoneType, required=True, metadata=[_PydanticGeneralMetadata(union_mode='left_to_right')])] | list[Annotated[dict[str, Any] | Prior | Any, FieldInfo(annotation=NoneType, required=True, metadata=[_PydanticGeneralMetadata(union_mode='left_to_right')])]] | None

options: DataFitOptions | None

wrap_bare_objective()[source]

Wrap a bare objective in a dict for a consistent internal shape.

Only applies to DataFit, not ArrayDataFit (which requires a dict keyed by independent variable values).

validate_parameters_or_priors()[source]

At least one of parameters or priors must be supplied.

The runtime accepts both together — priors then act as regularizers on the listed fit parameters — so we mirror the runtime here rather than enforce a stricter mutual exclusion at the schema boundary.

model_config = {'arbitrary_types_allowed': True, 'extra': 'forbid', 'populate_by_name': True, 'validate_assignment': True}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_post_init(context: Any, /) → None

This function is meant to behave like a BaseModel method to initialize private attributes.

It takes context as an argument since that’s what pydantic-core passes when calling it.

Args:: self: The BaseModel instance. context: The context.

class ionworks.pipeline.ArrayDataFit(objectives, source='', parameters=None, cost=None, initial_guesses=None, optimizer=None, cost_logger=None, multistarts=None, objective_parallelism='auto', initial_guess_sampler=None, priors=None, options=None)[source]

Bases: DataFit

Fit the same model separately at each value of an independent variable.

Use this when you have one experiment repeated at different conditions — typically temperatures, C-rates, or pulse indices — and you want one fitted parameter set per condition rather than one global fit. objectives is keyed by the independent variable value ({298.15: ..., 313.15: ...}); each entry is fitted independently and the results can be post-processed to extract how parameters depend on the variable.

All other fields behave the same as DataFit.

objectives: Annotated[dict[Any, Annotated[CalendarAgeing | CurrentDriven | CycleAgeing | EIS | ElectrodeBalancing | ElectrodeBalancingHalfCell | MSMRFullCell | MSMRHalfCell | OCPHalfCell | Objective | Pulse | Resistance, FieldInfo(annotation=NoneType, required=True, discriminator='type'), WrapValidator(func=_resolve_objective, json_schema_input_type=PydanticUndefined)]] | Annotated[CalendarAgeing | CurrentDriven | CycleAgeing | EIS | ElectrodeBalancing | ElectrodeBalancingHalfCell | MSMRFullCell | MSMRHalfCell | OCPHalfCell | Objective | Pulse | Resistance, FieldInfo(annotation=NoneType, required=True, discriminator='type'), WrapValidator(func=_resolve_objective, json_schema_input_type=PydanticUndefined)], FieldInfo(annotation=NoneType, required=True, metadata=[_PydanticGeneralMetadata(union_mode='left_to_right')])]

model_config = {'arbitrary_types_allowed': True, 'extra': 'forbid', 'populate_by_name': True, 'validate_assignment': True}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_post_init(context: Any, /) → None

This function is meant to behave like a BaseModel method to initialize private attributes.

It takes context as an argument since that’s what pydantic-core passes when calling it.

Args:: self: The BaseModel instance. context: The context.

class ionworks.pipeline.Parameter(name, initial_value=None, bounds=None, prior=None, normalize=None, check_bounds=None, check_initial_value=None, initial_guess_distribution=None)[source]

Bases: BaseSchema

A model parameter that is free to move during a data fit.

Wrap any quantity you want the fit to adjust in a Parameter so the pipeline knows it’s a free variable: its starting value, its plausible range, and (optionally) what you already believe about its likely value. Anything not wrapped in a Parameter is treated as a fixed input.

Parameters:

name (str) – The name of the parameter. Should match the name used inside the model (e.g. "Particle diffusion time [s]").
initial_value (float | None, optional) – The value the optimiser starts from. If you leave it unset, the midpoint of finite bounds is used (or 1 when there are no bounds).
bounds (tuple[float, float] | list[float] | None, optional) – (lower, upper) tuple bracketing where you believe the true value lies. Defaults to no bounds. The upper bound must be strictly greater than the lower bound.
prior (Distribution or Prior or None, optional) – A probability distribution describing what you already believe about the parameter. Used by Bayesian and regularised fits.
normalize (bool | None, optional) – Rescale by the initial value before optimisation so the optimiser sees comparable magnitudes. Defaults to True.
check_bounds (bool | None, optional) – Validate the bounds at construction time. Defaults to True.
check_initial_value (bool | None, optional) – Validate that the initial value falls inside the bounds at construction time. Defaults to True.
initial_guess_distribution (Distribution or None, optional) – When running multistart fits, this is the distribution the starting points are drawn from. Defaults to a uniform distribution over the bounds.

Examples

>>> # build the parameter with bounds, a prior, and a log transform
>>> raw = iw.pipeline.Parameter(
...     "Negative particle diffusivity [m2.s-1]",
...     initial_value=1e-14,
...     bounds=(1e-16, 1e-12),
...     prior=iw.pipeline.stats.LogNormal(mean=-32.2, std=2.0),
... )
>>> param = iw.pipeline.transforms.Log10(parameter=raw)
>>> # slot it into a DataFit
>>> obj = iw.pipeline.objectives.Pulse(
...     data_input="path/to/gitt.csv", options={"model": "SPM"}
... )
>>> fit = iw.pipeline.DataFit(objectives={"gitt": obj}, parameters={raw.name: param})

name: str

initial_value: int | float | None

bounds: tuple[int | float, int | float] | list[int | float] | None

prior: Any | None

normalize: bool | None

check_bounds: bool | None

check_initial_value: bool | None

initial_guess_distribution: Any | None

upper_bound_greater_than_lower()[source]: Validate that upper bound is greater than lower bound.

model_config = {'arbitrary_types_allowed': True, 'extra': 'forbid', 'populate_by_name': True, 'validate_assignment': True}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_post_init(context: Any, /) → None

This function is meant to behave like a BaseModel method to initialize private attributes.

It takes context as an argument since that’s what pydantic-core passes when calling it.

Args:: self: The BaseModel instance. context: The context.

class ionworks.pipeline.Library[source]

Bases: BaseModel

Material library access.

Exposes the bundled set of reference materials (e.g. graphite, NMC, LFP) via static lookups by name.

static list_materials() → list[str][source]: Return the names of all bundled materials.

static get_material(name: str) → Material[source]

Return the Material registered under name.

Raises:: KeyError – If name is not registered in the library.

model_config = {}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class ionworks.pipeline.Material(name: str, description: str | None = None, parameter_values: dict[str, Any] | None = None)[source]

Bases: BaseModel

Material configuration.

Holds a named parameter set — the three thermodynamic host-site fields for MSMR-style materials, or any other pybamm-compatible parameter values for the material.

Parameters:

name (str) – Human-readable material identifier (e.g. "NMC - Verbrugge 2017"). Used as the lookup key in the library.
description (str, optional) – Free-text description of the material — typically includes the citation or source the parameter values were taken from.
parameter_values (dict, optional) – Mapping of pybamm parameter names to their values for this material. For MSMR materials typically contains the host-site standard potentials, occupancy fractions, and ideality factors.

model_config = {'extra': 'forbid'}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

name: str

description: str | None

parameter_values: dict[str, Any]

classmethod from_library(name: str) → Material[source]

Bases: BaseSchema

Check a fitted model against held-out experimental data.

A Validation step takes the parameters produced earlier in the pipeline, simulates the experiments listed in objectives, and compares those simulations to the measured data. The result tells you how well the model generalises beyond the data you fit on.

Each objective describes one comparison (e.g. “current vs. time for this discharge”). The summary_stats list controls which scalar error metrics — RMSE, MAE, max error, … — get reported alongside the full time-series comparison.

Parameters:

objectives (dict of name to objective or dict) – One entry per experiment you want to compare against. The key is a human-readable label (used in the report); the value is the objective describing what to simulate and what to compare.
summary_stats (list[Cost | dict], optional) – Which scalar error metrics to report (e.g. RMSE(), MAE(), Max()). If you leave this unset, sensible defaults are filled in for fitting-style objectives so the report carries the same physical units as the measurements.

Examples

>>> obj1 = iw.pipeline.objectives.CurrentDriven(
...     data_input="path/to/cycle_1C.csv", options={"model": "SPM"}
... )
>>> obj2 = iw.pipeline.objectives.CurrentDriven(
...     data_input="path/to/cycle_C2.csv", options={"model": "SPM"}
... )
>>> val = iw.pipeline.Validation(
...     objectives={"1C": obj1, "C/2": obj2},
...     summary_stats=[iw.pipeline.costs.RMSE(), iw.pipeline.costs.MAE()],
... )
>>> config = iw.pipeline.Pipeline({"validate": val}).to_config()
>>> # then submit `config` via ionworks.api

objectives: dict[str, Annotated[CalendarAgeing | CurrentDriven | CycleAgeing | EIS | ElectrodeBalancing | ElectrodeBalancingHalfCell | MSMRFullCell | MSMRHalfCell | OCPHalfCell | Objective | Pulse | Resistance, FieldInfo(annotation=NoneType, required=True, discriminator='type'), WrapValidator(func=_resolve_objective, json_schema_input_type=PydanticUndefined)]]

summary_stats: list[Annotated[RMSE | MAE | MSE | Max | SSE | Wasserstein | ChiSquare | MultiCost | GaussianLogLikelihood | DesignFunction, FieldInfo(annotation=NoneType, required=True, discriminator='type')] | str] | None

model_config = {'arbitrary_types_allowed': True, 'extra': 'forbid', 'populate_by_name': True, 'validate_assignment': True}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_post_init(context: Any, /) → None

This function is meant to behave like a BaseModel method to initialize private attributes.

It takes context as an argument since that’s what pydantic-core passes when calling it.

Args:: self: The BaseModel instance. context: The context.

Examples

Examples

ionworks.pipeline¶

Core classes¶

Submodule reference¶

`ionworks.pipeline`¶