kipe.forward_solver module

The forward solver: the interface it implements, and its construction from a study file.

Forward solvers are external. They conform to ForwardSolver structurally: no inheritance from, or import of, kipe required. See kipe.examples.fitzhugh_nagumo for a reference implementation.

build_forward_solver() constructs the solver from the forward_solver section of a study file and checks that it conforms.

type kipe.forward_solver.State = dict[str, NDArray_f64]

Model state as named fields: field name -> rank-local array of the field state

type kipe.forward_solver.Parameters = dict[str, float]

Parameter name -> physical value.

class kipe.forward_solver.FieldSpec(*args, **kwargs)

Bases: Protocol

Layout of one state field on the current MPI rank.

Any object with these two attributes conforms, e.g., a solver’s own frozen dataclass.

property local_size: int

the owned entries for a distributed field, the full size for a replicated one.

Type:

Number of entries of the field held by this rank

property kind: Literal['distributed', 'replicated']

each entry is owned by exactly one rank (e.g., the owned dofs of a finite element function). replicated: every rank holds the full, identical field (e.g., the state of a small lumped-parameter model).

Type:

distributed

class kipe.forward_solver.ForwardSolver(*args, **kwargs)

Bases: Protocol

Forward model as seen by kipe.

From kipe’s point of view the solver is stateless: state, parameters and time are passed in on every call of propagate() and results are returned. Whatever the solver keeps internally (meshes, assembled operators, parameter objects it updates in place) must not make a call’s result depend on previous calls.

States are exchanged as named numpy fields, local to each MPI rank (see FieldSpec). Both directions are lenient for solvers: a state passed in is handed over (the solver may modify or reuse its arrays), and returned arrays may share memory with the solver’s internal data. They only need to stay valid until the next call, kipe copies whatever it keeps. kipe passes parameters as physical values, any reparameterization is kipe’s business.

The state must be complete: everything a time step depends on (e.g., previous time levels of a multistep scheme, states of lumped-parameter boundary models) is a field of the state, so that a step can be restarted from any given state.

initial_state() → tuple[float, State]

Return the start time and the initial state.

Returns:

• start time t0

• initial state, valid until the next call

property state_spec: Mapping[str, FieldSpec]

Names and layout of the state fields.

Every state returned or accepted by the solver has exactly these fields, with these local sizes.

nominal_parameters() → Parameters

Return the estimable parameters and their nominal values.

The nominal values are those the solver was configured with (e.g., by its input file or constructor). kipe uses the names to validate the parameters selected in a study, and the values as default initial guesses.

Returns:

parameter name -> nominal physical value dict (a new dict that the caller may modify)

propagate(t0: float, t1: float, state: State, parameters: Parameters) → State

Advance the state from t0 to exactly t1.

The solver may take any number of internal steps, but must land exactly on t1. Adaptive solvers rebuild their integrator from state on every call rather than keeping step-size history across calls.

Parameters:

• t0 – time of state

• t1 – target time, t1 > t0

• state – state at t0 (handed over, may be modified or reused)

• parameters – physical values of the selected parameters, applied for this call. Parameters not in parameters keep their nominal values.

Returns:

state at t1, valid until the next call

Raises:

ValueError – if t1 cannot be reached exactly (e.g., not on a fixed time grid)

class kipe.forward_solver.SupportsCheckpoints(*args, **kwargs)

Bases: Protocol

A forward solver that reads states from the checkpoints of an earlier run.

Synthesis uses it (synthesis.truth: {type: checkpoints, path: ...}) to generate data from an existing run instead of running the solver. The run may differ from the solver’s own configuration (e.g., in its time step), as long as its checkpoints can be read into the solver’s state layout.

read_checkpoint(path: str, t: float) → State

Return the state at time t from the checkpoints at path.

The checkpoint need not hold the full state: the fields it has, in the layout of ForwardSolver.state_spec. Synthesis checks that the fields a measurement observes are among them.

Parameters:

• path – location of the checkpoints, as the solver understands it (e.g., the output directory of the run)

• t – time of the checkpoint

Returns:

the fields of the checkpoint at t, valid until the next call

Raises:

ValueError – if there is no checkpoint at t, or none can be read at path

exception kipe.forward_solver.ForwardSolverError

Bases: Exception

The forward solver cannot be constructed from the given options.

kipe.forward_solver.build_forward_solver(options: ForwardSolverOptions) → ForwardSolver

Import the factory, check the arguments against its signature and call it.

Parameters:

options – forward_solver section of the study file

Returns:

the forward solver

Raises:

ForwardSolverError – if the factory cannot be imported or does not accept the arguments