pycsamt.forward.maxwell.batch#

Robust, resumable batch solving with retries and failure manifests.

solve_batch() is the single entry point. It solves many MaxwellProblem instances against one MaxwellBackend, optionally through a MaxwellResultCache for resumability, and never lets one bad problem silently corrupt or halt the whole run: every terminal failure is recorded in a FailureManifest instead of entering a training dataset unnoticed.

Three concerns are deliberately kept separate:

Resumability: Pass a MaxwellResultCache; a repeated run over the same problems and cache directory skips everything already solved, including across process restarts.
Retries: BatchPolicy retries only exceptions considered transient (by default BackendExecutionError and SolverConvergenceError). A deterministic failure such as IncompatibleProblemError is recorded immediately without wasting attempts.
Failure manifests: Every problem that exhausts its attempts becomes a ProblemFailure inside the returned BatchReport, which can be persisted with FailureManifest.to_json_file() and inspected or filtered back out of a future run’s input problems.

Functions

solve_batch(problems, backend, *[, cache, ...])

Solve many problems robustly, with retries and a failure manifest.

Classes

`BatchPolicy`([max_attempts, retry_backoff_s, ...])	Configure retries and concurrency for `solve_batch()`.
`BatchReport`(total, solved, cache_hits, failed)	Summarize one `solve_batch()` run.
`FailureManifest`([failures])	Ordered, JSON-persistable record of every terminal batch failure.
`ProblemFailure`(problem_hash, attempts, ...)	Record one problem's terminal solve failure.

Exceptions

BatchAbortedError(report)

Raised when BatchPolicy.stop_on_first_failure aborts early.

exception pycsamt.forward.maxwell.batch.BatchAbortedError(report)[source]

Bases: RuntimeError

Raised when BatchPolicy.stop_on_first_failure aborts early.

Parameters:: report (BatchReport) – Partial report covering every problem resolved before the abort; every problem submitted after the triggering failure (in the sequential case) was never attempted.
Return type:: None

Examples

>>> report = BatchReport(1, (), (), FailureManifest())
>>> error = BatchAbortedError(report)
>>> error.report is report
True

class pycsamt.forward.maxwell.batch.BatchPolicy(max_attempts=1, retry_backoff_s=1.0, retry_on=(<class 'pycsamt.forward.maxwell.adapters.BackendExecutionError'>, <class 'pycsamt.forward.maxwell.adapters.SolverConvergenceError'>), stop_on_first_failure=False, max_workers=1)[source]

Bases: object

Configure retries and concurrency for solve_batch().

Parameters:

max_attempts (int, default=1) – Attempts per problem, including the first. Values above one retry a solve that raised an exception in retry_on.
retry_backoff_s (float, default=1.0) – Base delay before each retry; attempt n (n > 1) waits retry_backoff_s * n seconds before the next attempt.
retry_on (tuple of exception types, optional) – Exception types treated as transient and worth retrying. Defaults to BackendExecutionError and SolverConvergenceError (which covers ExternalProcessError as a subclass). Anything else, including IncompatibleProblemError and InvalidBackendResultError, is recorded as a terminal failure after its first occurrence. Note that BaseMaxwellAdapter wraps ordinary exceptions raised inside a solve into BackendExecutionError by default (its own AdapterPolicy.wrap_backend_exceptions), so a deterministic bug in a backend’s own code is retried like any other BackendExecutionError unless that adapter was built with wrap_backend_exceptions=False.
stop_on_first_failure (bool, default=False) – Raise BatchAbortedError as soon as any problem exhausts its attempts, instead of recording it and continuing. With max_workers > 1 this only stops further submissions; futures already in flight are still allowed to finish.
max_workers (int, default=1) – Number of solves run concurrently in a thread pool. Values above one only help when the backend releases the GIL during its work (true of scipy sparse solves and of any BaseExternalMaxwellAdapter, which spends most of its time waiting on a subprocess).

Examples

>>> policy = BatchPolicy(max_attempts=3, retry_backoff_s=0.5)
>>> policy.max_attempts
3

max_attempts: int = 1

retry_backoff_s: float = 1.0

retry_on: tuple[type[Exception], ...] = (<class 'pycsamt.forward.maxwell.adapters.BackendExecutionError'>, <class 'pycsamt.forward.maxwell.adapters.SolverConvergenceError'>)

stop_on_first_failure: bool = False

max_workers: int = 1

class pycsamt.forward.maxwell.batch.ProblemFailure(problem_hash, attempts, error_type, message)[source]

Bases: object

Record one problem’s terminal solve failure.

Parameters:

problem_hash (str) – Identity of the MaxwellProblem that failed.
attempts (int) – Number of attempts made before this failure was recorded.
error_type (str) – Exception class name, for quick triage without deserializing.
message (str) – Human-readable exception message.

Examples

>>> failure = ProblemFailure("a" * 64, 2, "BackendExecutionError", "x")
>>> failure.attempts
2

problem_hash: str

attempts: int

error_type: str

message: str

to_dict()[source]

Return a JSON-serializable representation.

Returns:: Problem identity, attempt count, and exception details.
Return type:: dict

Examples

>>> ProblemFailure("a" * 64, 1, "X", "y").to_dict()["attempts"]
1

classmethod from_dict(data)[source]

Restore a validated failure record.

Parameters:: data (mapping) – State previously returned by to_dict().
Returns:: Restored, validated record.
Return type:: ProblemFailure

Examples

>>> state = ProblemFailure("a" * 64, 1, "X", "y").to_dict()
>>> ProblemFailure.from_dict(state).error_type
'X'

class pycsamt.forward.maxwell.batch.FailureManifest(failures=())[source]

Bases: object

Ordered, JSON-persistable record of every terminal batch failure.

Parameters:: failures (sequence of ProblemFailure, optional) – Failures in the order they were recorded. Problem hashes must be unique within one manifest.

Examples

>>> failure = ProblemFailure("a" * 64, 1, "X", "boom")
>>> manifest = FailureManifest((failure,))
>>> len(manifest), bool(manifest)
(1, True)

failures: tuple[ProblemFailure, ...] = ()

property hashes: frozenset[str][source]

Return every failed problem’s hash.

Returns:: Set suitable for filtering a future run’s input problems.
Return type:: frozenset of str

Examples

>>> failure = ProblemFailure("a" * 64, 1, "X", "boom")
>>> "a" * 64 in FailureManifest((failure,)).hashes
True

to_dict()[source]

Return a JSON-serializable, schema-versioned representation.

Returns:: Every failure, in recorded order.
Return type:: dict

Examples

>>> FailureManifest().to_dict()["schema_version"]
1

classmethod from_dict(data)[source]

Restore a validated manifest.

Parameters:: data (mapping) – State previously returned by to_dict().
Returns:: Restored manifest.
Return type:: FailureManifest
Raises:: ValueError – If the schema version is unsupported.

Examples

>>> state = FailureManifest().to_dict()
>>> FailureManifest.from_dict(state).failures
()

to_json_file(path)[source]

Write this manifest as indented, deterministic JSON.

Parameters:: path (str or pathlib.Path) – Destination file.
Returns:: The destination path.
Return type:: pathlib.Path

Examples

>>> from tempfile import TemporaryDirectory
>>> failure = ProblemFailure("a" * 64, 1, "X", "boom")
>>> manifest = FailureManifest((failure,))
>>> with TemporaryDirectory() as directory:
...     target = Path(directory) / "failures.json"
...     _ = manifest.to_json_file(target)
...     restored = FailureManifest.from_json_file(target)
>>> restored == manifest
True

classmethod from_json_file(path)[source]

Load a manifest written by to_json_file().

Parameters:: path (str or pathlib.Path) – Source file.
Returns:: Restored manifest.
Return type:: FailureManifest

Examples

See to_json_file() for a complete round trip.

class pycsamt.forward.maxwell.batch.BatchReport(total, solved, cache_hits, failed)[source]

Bases: object

Summarize one solve_batch() run.

Parameters:

total (int) – Number of problems submitted to the batch. Can exceed len(solved) + len(failed) when BatchPolicy.stop_on_first_failure ended the run before every problem was attempted.
solved (tuple of str) – Problem hashes with a valid result, whether freshly computed or already cached.
cache_hits (tuple of str) – Subset of solved that were already present in the cache (skipped re-solving). Empty when no cache was used.
failed (FailureManifest) – Every problem that exhausted its attempts.

Examples

>>> report = BatchReport(1, ("a" * 64,), (), FailureManifest())
>>> report.success_fraction
1.0

total: int

solved: tuple[str, ...]

cache_hits: tuple[str, ...]

failed: FailureManifest

property success_fraction: float[source]

Return the fraction of submitted problems that were solved.

Returns:: len(solved) / total; 1.0 when total is zero.
Return type:: float

Examples

>>> BatchReport(0, (), (), FailureManifest()).success_fraction
1.0

to_dict()[source]

Return a JSON-serializable representation.

Returns:: Totals, solved/cache-hit hashes, and the failure manifest.
Return type:: dict

Examples

>>> BatchReport(0, (), (), FailureManifest()).to_dict()["total"]
0

pycsamt.forward.maxwell.batch.solve_batch(problems, backend, *, cache=None, policy=None, on_result=None, on_failure=None)[source]

Solve many problems robustly, with retries and a failure manifest.

Parameters:

problems (iterable of MaxwellProblem) – Problems to solve. Consumed fully before returning; order is preserved in the returned report only through per-problem hashes, not positionally.
backend (MaxwellBackend) – Conforming backend used for every problem.
cache (MaxwellResultCache or None, optional) – When given, problems already cached are skipped (resumability) and freshly computed results are written back to it.
policy (BatchPolicy or None, optional) – Retry, concurrency, and abort configuration. Defaults to BatchPolicy.
on_result (callable, optional) – Invoked with (problem, result) for every problem that succeeds, in completion order. Useful for streaming results into a dataset without holding them all in memory.
on_failure (callable, optional) – Invoked with (problem, failure) for every problem that exhausts its attempts, in completion order.

Returns:

Totals, solved/cache-hit problem hashes, and a FailureManifest of every terminal failure.

Return type:

BatchReport

Raises:

TypeError – If backend or policy has the wrong type.
BatchAbortedError – If policy.stop_on_first_failure is set and any problem exhausts its attempts.

Examples

>>> import numpy as np
>>> from pycsamt.forward.maxwell import (
...     CallableMaxwellAdapter,
...     BackendCapabilities,
...     ForwardResult,
...     MaxwellMesh,
...     MaxwellProblem,
...     ReceiverSet,
...     SolverDiagnostics,
... )
>>> mesh = MaxwellMesh([0, 1, 2], [0, 1, 2])
>>> problem = MaxwellProblem(
...     mesh, np.ones((2, 2)), [1], ReceiverSet([[0.5, 0]], ["S"])
... )
>>> def solver(value):
...     diagnostics = SolverDiagnostics([[True]], [[0]], [[0]], 0)
...     return ForwardResult(
...         value.problem_hash,
...         value.frequencies_hz,
...         value.receivers.names,
...         value.components,
...         [[[1j, 1j]]],
...         None,
...         "demo",
...         "1",
...         diagnostics,
...     )
>>> cap = BackendCapabilities("demo", "1", (2,), ("zxy", "zyx"))
>>> backend = CallableMaxwellAdapter(cap, solver)
>>> report = solve_batch([problem], backend)
>>> report.success_fraction
1.0