Run¶

Run lifecycle: the content-addressed bundle a Driver writes per solve and a Run opens for reading.

Modules:

• artifacts –

  The run side of the built-in artifacts: rebinding a cached mesh's groups, the bundle file policy, and rehydration.

• bundle –

  Read and write the runs/<run_id>/ bundle: a content-addressed run directory holding the manifest, the typed result and the pinned geometry.

• convergence –

  Mesh convergence: re-run one configuration at increasing refinement, then Richardson-extrapolate the quantity of interest.

• driver –

  The driver: assemble one run end-to-end (identity, reuse, mesh + sif + solve, parse, bundle, publish, retention); the generators (Mesher, Sim) stay run-ignorant.

• electrical –

  Per-phase electrical quantities of a current-conduction run, from the electrode flux columns in Result.raw and boundaries.yml.

• entities –

  Named bodies and boundaries of a run bundle, resolved to the mesh ids the VTU's GeometryIds cell array uses.

• fields –

  Fluent numpy queries over a solved VTU's fields, keyed by body names, tags, and mesh ids.

• inspect –

  The discoverable face of a run bundle: Run(sim_dir) opens typed results, named entities, components, and the VTU field query.

• kinds –

  Run identity and the built-in artifact kinds: the STEP a geometry builds, the Elmer mesh a config meshes, and the bundle a run leaves.

• magnetic –

  Per-component magnetic quantities of a magnetostatics run, joined from the SaveScalars columns in Result.raw and components.yml.

• manifest –

  Run provenance: the self-describing run manifest and the git version stamp.

• parse –

  Parse Elmer outputs into a typed Result: SaveScalars columns for the numbers, the solver log for status, residual and wall time.

• purge –

  Post-hoc retention over existing run bundles: one bundle or a whole runs directory, dry-run by default.

• reap –

  Abandoned runs: the bundles left running with no live claim on them, marked so and their claims broken.

• result –

  The typed outcome of a solve: Elmer outputs parsed into validated, pint-typed fields.

• retention –

  Run retention: which classes of bundle files leave after a run, the policy that says so, and the engine that applies it.

• stages –

  Pipeline stage vocabulary: the driver's stages in execution order, plus the timed event a run records per stage.

Classes:

• BundlePolicy –

  Which bundle files a published run bundle carries.

• MeshGroupsFile –

  groups.json: the group records and element count a mesh artifact carries beside ElmerGrid's files.

• MeshRebindError –

  A cached mesh cannot be bound to the current config's domain.

• ConvergencePoint –

  One rung of the refinement ladder: the refinement level, the resulting quantity, and its relative change from the previous (coarser) rung.

• RichardsonEstimate –

  Richardson extrapolation of a ladder's finest three rungs, assuming q(h) = q* + C·hᵖ.

• Driver –

  Runs each config through the pipeline into a runs/<run_id>/ bundle and returns its typed Result.

• Bodies –

  Name-keyed collection of a run's bodies.

• Body –

  One named body group of a run.

• Boundaries –

  Name-keyed collection of a run's boundaries, with adjacency lookups powered by the A__B name convention.

• Boundary –

  One named boundary group of a run.

• BoundaryKind –

  What a boundary group separates: two bodies, a body's external shell, or its trace on the 2D symmetry axis.

• Entities –

  A bundle's named bodies and boundaries.

• FieldMesh –

  A solved VTU as cell-space numpy arrays with named-body selection; open one via load or Run.mesh().

• FieldQuery –

  One field array under an immutable cell mask: filters return refined copies, terminals return numpy values.

• Run –

  One run bundle (runs/<run_id>/) opened for reading.

• SifComponent –

  One sif Component block with its lumped force and torque.

• MeshSpec –

  Everything that decides the Elmer mesh files and their group names, and nothing that only decides the solve.

• RunSpec –

  The bundle kind's spec: the config a run solved.

• RunManifest –

  The self-describing record of one run.

• ReapedRun –

  One run reap_runs settled.

• MeshStats –

  Per-run mesh size/quality summary, recorded in the manifest so a sweep can be audited for mesh consistency.

• Result –

  The typed outcome of one solve.

• BundleEntry –

  One file found in a bundle, with the retention class its name falls in.

• Cleanup –

  When a file class leaves the bundle; definition order is escalation order.

• CleanupRecord –

  One cleanup pass over a bundle: when, by whom, under which policy, and what left.

• CleanupReport –

  The outcome of applying a plan: what left (or would have, in a dry run) and what could not be removed.

• Preset –

  Named retention policies, from keep-everything to removing the run directory.

• RetentionPolicy –

  One Cleanup rule per bundle file class; the default keeps everything the run cannot re-fetch.

• Stage –

  The driver's pipeline stages, in execution order.

• StageEvent –

  One stage's timing record: created at stage start, mutated with finished_at when the stage completes.

Functions:

• bundle_files –

  The bundle-relative posix paths policy publishes from sim_dir.

• rebind_groups –

  Rebuild a cached mesh's groups against this config's domain, matching materials by tag and EntityTag objects by selector; ids stay placeholders until reconcile_ids.

• rehydrate_bundle –

  Fetch every store-sourced input the manifest names back into the bundle.

• load_manifest –

  Read a bundle's manifest.json back into a typed RunManifest — a delegate to RunManifest.load.

• load_result –

  Read a bundle's result.json back into a typed Result — a delegate to Result.load.

• write_atomic –

  Replace path atomically: write a .tmp sibling, then os.replace — a reader never sees a half-written file.

• write_bundle –

  Write manifest.json and result.json into sim_dir, pin the source STEP as geometry.step unless the run already holds one, and return the bundle.

• write_manifest –

  Atomically write (or overwrite) manifest.json in sim_dir (created if needed) and return its path.

• convergence_ladder –

  Run base_config at each refinement level and report the quantity of interest plus its relative change between rungs.

• richardson –

  Richardson-extrapolate the finest three rungs of a ladder whose scales are mesh-size multipliers (larger is coarser) paired with values.

• boundary_fluxes –

  Elmer bc index -> integrated conductive flux (A) from the SaveScalars columns in Result.raw.

• electrode_currents –

  Boundary block name -> terminal current (A), joining the flux columns against the bc-index name map.

• phase_currents –

  Phase index ("0", "1", ...) -> terminal current (A): the largest-magnitude current among each phase's named electrode currents.

• read_boundary_names –

  The boundaries.yml map the Sim writes (<bc index>: <boundary block name>), as a dict.

• resistance –

  R = ΔV / I for one phase; raises on a (numerically) zero current.

• load_entities –

  Read a bundle's bodies/boundaries: entities.json when present, else mesh.names enriched from the manifest config, else empty.

• write_entities –

  Write entities.json mapping each group's reconciled mesh id to its named record; call after reconcile_ids.

• select_array –

  Resolve a VTU array name for a solver variable: exact (nodal), then the elemental "<name> e", then substring — VTU names come back lowercased.

• find_runs –

  The bundles under runs_dir whose metadata matches, newest first; this reads every manifest, so prefer RunIndex over thousands of bundles.

• mesh_spec –

  The mesh artifact spec of config (step_key_override when the driver already resolved the geometry).

• step_digest –

  SHA-256 of a STEP with its FILE_NAME header (the export timestamp) masked; of the path string when the file is absent, so pure-logic keys stay deterministic.

• component_forces –

  Component name -> lumped magnetic force (N, per radian in an axisymmetric run), joining the force triples against the components.yml map.

• component_torques –

  Component name -> lumped magnetic torque (N*m about the authored origin/axis, per radian in an axisymmetric run), joining the torque triples against the components.yml map.

• eddy_current_power –

  The total eddy current power (W, per radian in an axisymmetric run), or None when the run exported none (static solves have no eddy losses).

• force_triples –

  Column stem -> per-component force Vec3 (N, per radian in an axisymmetric run) from the SaveScalars columns in Result.raw.

• magnetic_field_energy –

  The total magnetic field energy (J, per radian in an axisymmetric run), or None when the run exported none.

• read_component_names –

  The components.yml map the Sim writes (<component id>: <Component name>) — the same flat format as boundaries.yml.

• cache_key –

  Full-run identity: every result-affecting input plus RUN_PRODUCER_VERSION; never the code commit, the machine, or the run metadata.

• git_sha –

  The current commit, with -dirty appended when the working tree has uncommitted changes; None without git or outside a repository.

• mesh_key –

  The mesh artifact key of config: invariant to materials' properties, conditions, physics, and metadata.

• step_key –

  The solve-frame geometry identity a run and its mesh key on: the STEP artifact key of step_spec(config).

• parse_result –

  Parse a finished run directory into a typed Result, revolving force and torque to full-revolution totals under an Axisymmetric frame.

• result_vtu –

  The run's newest ParaView output: the .pvtu master when the solve was partitioned, else the newest .vtu; None when there is none.

• reap_runs –

  Mark every bundle under runs_dir left running with no live claim on it as abandoned, and break every claim that went silent.

• apply_cleanup –

  Remove every planned path; never raises, so a path that cannot be removed lands in report.failed.

• bundle_inventory –

  Every file under a bundle with the retention class its name matches; a name no class claims is filed under RunFile.RUN_DIR, the only rule that reaches it.

• plan_cleanup –

  Resolve policy against one bundle into the sized paths each applying rule removes; refuses a symlinked or junctioned directory.

Attributes:

• SolverFn –

  A solve step that fills sim_dir with Elmer outputs and reports the mesh it built; injecting one replaces the gmsh and Elmer pipeline.

• GEOMETRY_IDS_ARRAY –

  The Int cell array ResultOutputSolver (Save Geometry Ids) writes: each cell's mesh id, bodies and boundaries alike.

• ELMER_MESH (ArtifactKind[MeshSpec]) –

  ElmerGrid's mesh files plus groups.json; produced by the driver, which owns gmsh and ElmerGrid.

• RUN_BUNDLE (ArtifactKind[RunSpec]) –

  A finished run bundle, keyed by cache_key so its store key and its run_id agree.

• STEP (ArtifactKind[StepSpec]) –

  The solve-frame STEP of a parametric geometry or an authored file, keyed by its source (HashExclude fields left out) and the frame.

• ManifestStatus –

  A manifest's lifecycle status: "running" while the pipeline executes, then the terminal RunStatus, or "abandoned" when its claim went silent first.

• RunStatus –

  The terminal state of a solve: completed cleanly, failed to converge, or errored out.

SolverFn = Callable[[SimulationConfig, Path], MeshStats]

GEOMETRY_IDS_ARRAY = 'GeometryIds'

ELMER_MESH: ArtifactKind[MeshSpec] = ArtifactKind(name=BuiltinKind.ELMER_MESH, spec_type=MeshSpec, producer_version=MESH_PRODUCER_VERSION, required_files=(BundleFile.MESH_HEADER, BundleFile.MESH_NODES, BundleFile.MESH_ELEMENTS, BundleFile.MESH_BOUNDARY, BundleFile.MESH_NAMES, BundleFile.GROUPS))

RUN_BUNDLE: ArtifactKind[RunSpec] = ArtifactKind(name=BuiltinKind.RUN_BUNDLE, spec_type=RunSpec, producer_version=RUN_PRODUCER_VERSION, required_files=(BundleFile.MANIFEST, BundleFile.RESULT), key=lambda spec: cache_key(spec.config), in_place=True)

STEP: ArtifactKind[StepSpec] = ArtifactKind(name=BuiltinKind.STEP, spec_type=StepSpec, producer_version=STEP_PRODUCER_VERSION, required_files=(BundleFile.GEOMETRY_STEP,), produce=_build_step)

ManifestStatus = Literal['running', 'abandoned'] | RunStatus

RunStatus = Literal['ok', 'diverged', 'error']

CORE = 'core'

FIELDS = 'fields'

FULL = 'full'

classes: tuple[RunFile, ...]

bodies: list[BodyGroupRecord]

boundaries: list[BoundaryGroupRecord]

element_count: int = 0

capture(physical_groups: Iterable[_BodySource], boundary_groups: Iterable[_BoundarySource], element_count: int) -> MeshGroupsFile

215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247

@classmethod
def capture(cls, physical_groups: Iterable[_BodySource], boundary_groups: Iterable[_BoundarySource], element_count: int) -> MeshGroupsFile:
    """Record the mesher's groups.

    Args:
        physical_groups: The body groups, in mesher order.
        boundary_groups: The boundary groups, in mesher order.
        element_count: Elements of the body dimension in the mesh.
    """
    return cls(
        bodies=[
            BodyGroupRecord(
                dim=group.dim,
                name=group.name,
                material=group.material.tag,
                selectors=[dict(tag.selector) for tag in group.tags],
                entity_tags=list(group.entity_tags),
                authored=group.authored,
            )
            for group in physical_groups
        ],
        boundaries=[
            BoundaryGroupRecord(
                dim=group.dim,
                name=group.name,
                bodies=tuple(group.bodies),
                face_tags=list(group.entity_tags),
                on_axis=group.on_axis,
            )
            for group in boundary_groups
        ],
        element_count=element_count,
    )

load(sim_dir: str | Path) -> MeshGroupsFile

249
250
251
252

@classmethod
def load(cls, sim_dir: str | Path) -> MeshGroupsFile:
    """Read ``groups.json`` from ``sim_dir``."""
    return cls.model_validate_json((Path(sim_dir) / BundleFile.GROUPS).read_text(encoding="utf-8"))

write(sim_dir: str | Path) -> Path

254
255
256
257
258

def write(self, sim_dir: str | Path) -> Path:
    """Write ``groups.json`` into ``sim_dir`` and return its path."""
    path: Path = Path(sim_dir) / BundleFile.GROUPS
    _ = path.write_text(self.model_dump_json(indent=2), encoding="utf-8")
    return path

refinement: float

quantity: Quantity

delta_vs_previous: float | None = None

observed_order: float

extrapolated_value: float

uncertainty: float

Driver(*, runs_dir: str | Path = 'runs', code_sha: str | None = None, solver: SolverFn | None = None, mpi_processes: int = 1, backend: ExecutionBackend | None = None, mesher: MeshBackend | None = None, docker: bool | None = None, docker_image: str | None = None, retention: RetentionPolicy | None = None, artifacts: ArtifactCache | None = None, reuse: bool = True, bundle_policy: BundlePolicy = FIELDS, telemetry: str | Path | None = None)

132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193

def __init__(
    self,
    *,
    runs_dir: str | Path = "runs",
    code_sha: str | None = None,
    solver: SolverFn | None = None,
    mpi_processes: int = 1,
    backend: ExecutionBackend | None = None,
    mesher: MeshBackend | None = None,
    docker: bool | None = None,
    docker_image: str | None = None,
    retention: RetentionPolicy | None = None,
    artifacts: ArtifactCache | None = None,
    reuse: bool = True,
    bundle_policy: BundlePolicy = BundlePolicy.FIELDS,
    telemetry: str | Path | None = None,
) -> None:
    """Set where runs land and where each stage executes.

    Args:
        runs_dir: The directory each run's bundle is written under.
        code_sha: The code version each manifest records; None reads it from git.
        solver: A replacement for the staged pipeline, for tests that run without gmsh or Elmer.
        mpi_processes: The ElmerSolver rank count; more than 1 needs a partitioned backend.
        backend: Where ElmerGrid and ElmerSolver run; None picks Docker or native Elmer from ``docker``, ``docker_image`` and ``mpi_processes``.
        mesher: Where gmsh meshes each STEP; None meshes in this process, and a [`WorkerMesher`][femforge.meshing.backend.WorkerMesher] meshes in a child process.
        docker: True runs Elmer in Docker and False runs it natively; it cannot be combined with ``backend``.
        docker_image: An image tag to run Elmer in instead of femforge's own image; it cannot be combined with ``backend``.
        retention: What leaves each bundle after its evaluation; ``evaluate(retention=...)`` overrides it for one call.
        artifacts: The cache the STEP, mesh and bundle stages read from and publish to; None keeps runs store-free.
        reuse: Whether to return a finished bundle with the same identity instead of solving again.
        bundle_policy: Which bundle files a publish carries, and so which files a reused bundle must still hold.
        telemetry: The telemetry root for each run's index row and log records; None falls back to ``FEMFORGE_TELEMETRY``.
    """
    self.runs_dir: Path = Path(runs_dir)

    self.code_sha: str | None = code_sha if code_sha is not None else git_sha()

    self._solver: SolverFn = solver if solver is not None else self._default_solve

    if mpi_processes < 1:
        raise ValueError(f"mpi_processes must be >= 1, got {mpi_processes}")

    if backend is not None and (docker is not None or docker_image is not None):
        raise ValueError("Pass either backend or docker/docker_image, not both.")

    self.backend: ExecutionBackend = backend if backend is not None else default_backend(docker=docker, docker_image=docker_image, mpi_processes=mpi_processes)
    self.mesher: MeshBackend = mesher if mesher is not None else GmshMesher()

    if mpi_processes > 1 and not self.backend.partitioned:
        raise ValueError(f"mpi_processes > 1 needs a partitioned backend such as DockerElmer; {type(self.backend).__name__} is not one.")

    self.mpi_processes: int = mpi_processes
    self.retention: RetentionPolicy = retention if retention is not None else RetentionPolicy.default()
    self.artifacts: ArtifactCache | None = artifacts
    self.reuse: bool = reuse
    self.bundle_policy: BundlePolicy = bundle_policy
    self.telemetry: str | Path | None = telemetry

    self._active: tuple[RunManifest, Path] | None = None
    """The in-flight run [`on_stage_event`][] records into — a Driver executes one solve at a time; set by
    [`evaluate`][] before its first manifest write and cleared once the cleanup stage has recorded."""

evaluate(config: SimulationConfig, *, retention: RetentionPolicy | None = None) -> Result

197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235

def evaluate(self, config: SimulationConfig, *, retention: RetentionPolicy | None = None) -> Result:
    """
    Return the finished bundle for ``config`` when one is reusable, else mesh + sif + solve it under a claim on its
    run directory, parse the result, write and publish the bundle, then apply the retention policy.

    The run directory is ``bundle_name(cache_key)`` — a memorable name derived from the content hash — so
    re-evaluating identical inputs still targets the same bundle. ``retention`` overrides the driver's policy for this call.
    """
    # Every record of this evaluation lands in this driver's telemetry root, not the process default.
    with telemetry_scope(self.telemetry):
        policy: RetentionPolicy = retention if retention is not None else self.retention

        key: ArtifactKey[RunSpec] = ArtifactKey[RunSpec](cache_key(config))
        run_id: str = bundle_name(key)
        sim_dir: Path = self.runs_dir / run_id

        reused: Result | None = self._reuse(config, key, sim_dir)
        if reused is not None:
            return reused

        # One producer per bundle: a second process asking for the same identity waits, then reuses what the first left.
        claim = DirLock(self.runs_dir / f"{run_id}{LOCK_SUFFIX}")
        while not claim.try_acquire():
            time.sleep(CLAIM_POLL_S)
            reused = self._reuse(config, key, sim_dir)
            if reused is not None:
                return reused

        try:
            with keep_alive(claim, HEARTBEAT_S):
                return self._solve_bundle(
                    config,
                    key,
                    run_id,
                    sim_dir,
                    policy,
                )
        finally:
            claim.release()

run_index() -> RunIndex | None

366
367
368
369
370
371
372
373
374
375

def run_index(self) -> RunIndex | None:
    """The index this driver records into, or None when telemetry is off — which is also what keeps polars unimported."""
    if self.telemetry is None and not os.environ.get(TELEMETRY_ENV):
        return None

    from femforge.telemetry.paths import Telemetry, resolve_root
    from femforge.telemetry.runs import RunIndex

    # The guard above already proved one of the two sources resolve_root reads is set.
    return RunIndex(Telemetry(cast("Path", resolve_root(self.telemetry))), self.runs_dir)

publish(manifest: RunManifest, sim_dir: Path, config: SimulationConfig, result: Result) -> ArtifactRef

390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432

def publish(self, manifest: RunManifest, sim_dir: Path, config: SimulationConfig, result: Result) -> ArtifactRef:
    """The [`Stage.PUBLISH`][] default: commit the files `bundle_policy` selects, store-sourced inputs excluded, to the artifact cache under the run identity.

    The store keeps the manifest as it stands on disk at publish time.

    Args:
        manifest: The run's manifest, which records the published reference.
        sim_dir: The run directory the files come from.
        config: The config whose identity keys the published bundle.
        result: The parsed result, whose Elmer version the store records.
    """
    if self.artifacts is None:
        raise RuntimeError("publish requires an artifact cache.")

    files: list[str] = bundle_files(
        sim_dir,
        self.bundle_policy,
        exclude=manifest.cached_inputs(),
    )
    key: ArtifactKey[RunSpec] = self.artifacts.key(RUN_BUNDLE, RunSpec(config=config))
    manifest.published = ArtifactRef(
        kind=RUN_BUNDLE.name,
        key=key,
        origin=ArtifactOrigin.PRODUCED,
        files=files,
    )
    write_manifest(sim_dir, manifest)

    handle: ArtifactHandle[RunSpec] = self.artifacts.publish(
        RUN_BUNDLE,
        RunSpec(config=config),
        sim_dir,
        files,
        ToolVersions(code_sha=self.code_sha, elmer=result.elmer_version),
    )
    manifest.published = handle.ref
    log.bind(
        run_id=manifest.run_id,
        origin=handle.origin,
        files=len(files),
        policy=self.bundle_policy,
    ).info("Bundle published.")
    return handle.ref

cleanup(manifest: RunManifest, sim_dir: Path, policy: RetentionPolicy) -> CleanupReport

460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499

def cleanup(self, manifest: RunManifest, sim_dir: Path, policy: RetentionPolicy) -> CleanupReport:
    """The [`Stage.CLEANUP`][] default: apply ``policy`` against the run's terminal status and record what left on the manifest.

    Override it to push the bundle elsewhere first, then call ``super()``.

    Args:
        manifest: The run's manifest, which records the cleanup.
        sim_dir: The run directory to clean.
        policy: The retention policy to apply.
    """
    plan = plan_cleanup(
        sim_dir,
        policy,
        status=manifest.status,
        cached=manifest.cached_inputs(),
    )
    report: CleanupReport = apply_cleanup(plan)

    if report.removed:
        manifest.cleanups.append(
            CleanupRecord(
                applied_at=datetime.now(timezone.utc),
                trigger=CleanupTrigger.DRIVER,
                status=manifest.status,
                policy=policy,
                removed=report.removed,
            )
        )

    for relative, error in report.failed:
        log.bind(run_id=manifest.run_id, path=relative, error=error).warning("Cleanup could not remove a file.")

    log.bind(
        run_id=manifest.run_id,
        files=len(report.removed),
        bytes_freed=report.bytes_freed,
        remove_run_dir=plan.remove_run_dir,
    ).info("Cleanup applied.")

    return report

on_stage_event(event: StageEvent) -> None

533
534
535
536
537
538
539
540
541
542
543
544
545
546
547

def on_stage_event(self, event: StageEvent) -> None:
    """Default stage-event sink: record the event on the active run's manifest and persist it (override + ``super()`` to add behavior)."""
    if self._active is None:
        return

    manifest, sim_dir = self._active

    if event.finished_at is None:
        log.info(f"[{event.stage.name}] started.")
        manifest.stage_history.append(event)
    else:
        time_taken = (event.finished_at - event.started_at).total_seconds()
        log.info(f"[{event.stage.name}] completed ({round(time_taken, 2)}s).")

    write_manifest(sim_dir, manifest)

prepare_geometry(state: SolveState) -> None

558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574

def prepare_geometry(self, state: SolveState) -> None:
    """Materialize the solve-frame STEP into the run directory (from the artifact cache when one is configured) and repoint ``state.config`` at it."""
    spec: StepSpec = step_spec(state.config)
    state.step_key = artifact_key(STEP, spec)

    if self.artifacts is None or spec.passes_through:
        state.config = materialize_geometry(state.config, state.sim_dir)
        return

    handle: ArtifactHandle[StepSpec] = self.artifacts.get_or_produce(
        STEP,
        spec,
        into=state.sim_dir,
    )
    self._record_artifact(handle)
    state.step_key = handle.key
    state.config = repoint_step(state.config, state.sim_dir / BundleFile.GEOMETRY_STEP)

pre_mesh(state: SolveState) -> None

576
577

def pre_mesh(self, state: SolveState) -> None:
    """Hook before meshing (e.g. adjust ``state.config`` or drop extra inputs into ``state.sim_dir``)."""

mesh(state: SolveState) -> None

579
580
581
582
583
584
585
586
587
588
589
590
591
592
593

def mesh(self, state: SolveState) -> None:
    """Mesh with `mesher` and convert with ElmerGrid, served from the artifact cache when one is configured, then bind the groups to this config."""
    spec: MeshSpec = mesh_spec(state.config, state.step_key)
    if self.artifacts is None:
        _ = self._mesh_into(state, spec, state.sim_dir)
    else:
        handle: ArtifactHandle[MeshSpec] = self.artifacts.get_or_produce(
            ELMER_MESH,
            spec,
            into=state.sim_dir,
            produce=partial(self._produce_mesh, state),
        )
        self._record_artifact(handle)

    self._adopt_mesh(state, MeshGroupsFile.load(state.sim_dir))

post_mesh(state: SolveState) -> None

634
635

def post_mesh(self, state: SolveState) -> None:
    """Hook between meshing and sim construction (e.g. amend the Elmer mesh files in ``state.sim_dir``)."""

build_sim(state: SolveState) -> None

637
638
639
640
641
642
643
644
645
646
647
648
649
650

def build_sim(self, state: SolveState) -> None:
    """Populate ``state.sim``: the typed `Sim` from the config's resolved solver stack, MPI-adjusted when partitioned."""
    from femforge.elmer.config import prepare_for_mpi

    sim: Sim = Sim(
        state.physical_groups,
        physics=state.config.physics.selected,
        boundary_groups=state.boundary_groups,
        config=state.config.resolved_stack(),
        frame=state.config.frame,
    )
    if self.mpi_processes > 1:
        prepare_for_mpi(sim.config)
    state.sim = sim

configure_sim(state: SolveState) -> None

652
653
654
655
656

def configure_sim(self, state: SolveState) -> None:
    """
    Hook to edit the typed solver stack (``state.sim.config``) before the sif is generated. Prefer authoring
    ``config.physics.overrides`` instead — edits made here happen after the run identity is computed.
    """

write_sif(state: SolveState) -> None

658
659
660
661
662
663

def write_sif(self, state: SolveState) -> None:
    """Generate the sif object graph from the typed stack and write the sif into the run directory."""
    if state.sim is None:
        raise RuntimeError("write_sif requires build_sim to have populated state.sim.")

    state.sim.generate().write(str(state.sim_dir))

run_solver(state: SolveState) -> None

665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693

def run_solver(self, state: SolveState) -> None:
    """Partition the mesh when the solve is parallel (a partitioned backend is checked at construction), then run ElmerSolver on `backend`."""
    from femforge.telemetry.convergence import ConvergenceTargetsFile

    if state.sim is None:
        raise RuntimeError("run_solver requires build_sim to have populated state.sim.")

    if self.mpi_processes > 1:
        # -halobc only when the stack needs it (coil electrode integrals): halo elements double-count the Component
        # force lumping of partition-cut bodies, and partition-split electrode boundaries bias the coil drive without them.
        halobc: bool = needs_halobc(state.sim.config)
        field_solvers = [solver for solver in state.sim.config.solvers if isinstance(solver, MagnetoDynamicsCalcFieldsSolver)]
        if halobc and any(solver.calculate_nodal_forces for solver in field_solvers):
            log.warning("Partitioned coil solve computes nodal forces: halo elements corrupt the Component force lumping — extract forces from a serial solve.")

        # The flat bundle is the Elmer mesh dir itself.
        _ = self.backend.grid_partition(state.sim_dir, self.mpi_processes, halobc=halobc)

    # The sif-authored iteration caps and tolerances contextualize the streamed convergence lines (iter/max, relc → tol, %),
    # and the file keeps them beside convergence.jsonl so the log can be read back the same way later.
    targets = convergence_targets(state.sim.config)
    _ = ConvergenceTargetsFile.of(targets).write(state.sim_dir)

    _ = solve_case(
        self.backend,
        state.sim_dir,
        n_procs=self.mpi_processes,
        targets=targets,
    )

mesh_stats(state: SolveState) -> MeshStats

695
696
697
698
699
700
701
702
703
704
705
706

def mesh_stats(self, state: SolveState) -> MeshStats:
    """The built mesh's [`MeshStats`][]: the meshed element count plus the config's size settings."""
    config: SimulationConfig = state.config
    sizes: list[float] = [tag.mesh_size for tag in config.domain.tags if tag.mesh_size is not None]
    if not sizes:
        sizes = [config.mesh.global_mesh_size]

    return MeshStats(
        element_count=state.element_count,
        min_element_size=min(sizes),
        max_element_size=max(sizes),
    )

Bodies(bodies: Iterable[Body])

183
184
185

def __init__(self, bodies: Iterable[Body]) -> None:
    """Collect ``bodies`` by name."""
    self._by_name: dict[str, Body] = {body.name: body for body in bodies}

get(name: str) -> Body | None

202
203
204

def get(self, name: str) -> Body | None:
    """The body by exact name, or None."""
    return self._by_name.get(name)

with_material(tag: str) -> list[Body]

206
207
208
209

def with_material(self, tag: str) -> list[Body]:
    """Every body whose material tag matches (case-insensitive)."""
    want: str = tag.casefold()
    return [body for body in self if body.material is not None and body.material.casefold() == want]

with_tag(key: TagKey | str, value: str) -> list[Body]

211
212
213

def with_tag(self, key: TagKey | str, value: str) -> list[Body]:
    """Every body whose selector tags carry ``key=value``."""
    return [body for body in self if body.tags.get(str(key)) == value]

Body(name: str, mesh_id: int, material: str | None = None, tags: Mapping[str, str] = dict[str, str]())

name: str

mesh_id: int

material: str | None = None

tags: Mapping[str, str] = field(default_factory=dict[str, str])

Boundaries(boundaries: Iterable[Boundary])

219
220
221

def __init__(self, boundaries: Iterable[Boundary]) -> None:
    """Collect ``boundaries`` by name."""
    self._by_name: dict[str, Boundary] = {boundary.name: boundary for boundary in boundaries}

get(name: str) -> Boundary | None

238
239
240

def get(self, name: str) -> Boundary | None:
    """The boundary by exact name, or None."""
    return self._by_name.get(name)

between(a: Body | str, b: Body | str) -> Boundary

242
243
244
245
246
247
248
249
250
251

def between(self, a: Body | str, b: Body | str) -> Boundary:
    """The interface boundary between two bodies, whichever order the mesher named it."""
    first, second = _name_of(a), _name_of(b)
    for candidate in (f"{first}{BOUNDARY_SEP}{second}", f"{second}{BOUNDARY_SEP}{first}"):
        found: Boundary | None = self._by_name.get(candidate)
        if found is not None:
            return found

    near: list[str] = [boundary.name for boundary in self.of_body(first) + self.of_body(second)]
    raise KeyError(f"no boundary between {first!r} and {second!r}; their boundaries: {sorted(set(near))}")

external(body: Body | str) -> Boundary

253
254
255

def external(self, body: Body | str) -> Boundary:
    """The body's external shell boundary (``<BODY>__EXT``)."""
    return self[f"{_name_of(body)}{BOUNDARY_SEP}{EXTERNAL_SUFFIX}"]

axis(body: Body | str) -> Boundary | None

257
258
259

def axis(self, body: Body | str) -> Boundary | None:
    """The body's symmetry-axis boundary (``<BODY>__AXIS``), or None when it has none."""
    return self.get(f"{_name_of(body)}{BOUNDARY_SEP}{AXIS_SUFFIX}")

of_body(body: Body | str) -> list[Boundary]

261
262
263
264

def of_body(self, body: Body | str) -> list[Boundary]:
    """Every boundary bordering the body."""
    name: str = _name_of(body)
    return [boundary for boundary in self if name in boundary.bodies]

Boundary(name: str, mesh_id: int, bodies: tuple[str, ...], kind: BoundaryKind, sif_index: int | None = None)

name: str

mesh_id: int

bodies: tuple[str, ...]

kind: BoundaryKind

sif_index: int | None = None

INTERFACE = 'Interface'

EXTERNAL = 'External'

AXIS = 'Axis'

Entities(bodies: Bodies, boundaries: Boundaries)

bodies: Bodies

boundaries: Boundaries

FieldMesh(*, arrays: Mapping[str, FloatArray], measure: FloatArray, geometry_ids: IntArray | None = None, centers: FloatArray | None = None, bodies: Bodies | None = None, grid: UnstructuredGrid | None = None)