2.20.4.1. pycsamt.forward.maxwell#

Maxwell solver contracts, meshes, adapters, backend discovery, benchmarks, batch execution, caching, and external solver integration.

Solver-neutral contracts and adapters for multidimensional Maxwell physics.

This package is the integration boundary for verified 2-D and 3-D EM solver backends. It will own common problem/result contracts, backend capability reporting, convergence diagnostics, and canonical benchmark definitions.

Existing pycsamt.forward.em2d and pycsamt.forward.em3d APIs are not re-exported here. A backend is exposed through this namespace only after its dimensional coupling and numerical validation gates are documented and tested. Importing the package must not import optional solver dependencies.

class pycsamt.forward.maxwell.MaxwellMesh(x_edges_m, z_edges_m, y_edges_m=None, crs=None)#

Bases: object

Describe a rectilinear finite-volume or finite-element mesh.

Parameters:

x_edges_m (array-like) – Strictly increasing cell-edge coordinates in metres. Depth z increases downward.
z_edges_m (array-like) – Strictly increasing cell-edge coordinates in metres. Depth z increases downward.
y_edges_m (array-like or None, optional) – Second horizontal axis. Omit for a 2-D mesh.
crs (str or None, optional) – Coordinate reference system for horizontal coordinates.

Examples

>>> mesh = MaxwellMesh([0, 100, 250], [0, 50, 150])
>>> mesh.shape, mesh.dimension
((2, 2), 2)
>>> mesh.cell_centres_m["x"].tolist()
[50.0, 175.0]

x_edges_m: ndarray#

z_edges_m: ndarray#

y_edges_m: ndarray | None = None#

crs: str | None = None#

property dimension: int#

Return the spatial dimension.

Returns:: Mesh dimension.
Return type:: {2, 3}

Examples

>>> MaxwellMesh([0, 1, 2], [0, 1, 2]).dimension
2

property shape: tuple[int, ...]#

Return canonical cell-array shape.

Returns:: (nz, nx) or (nz, ny, nx).
Return type:: tuple of int

Examples

>>> MaxwellMesh([0, 1, 2], [0, 1, 2, 3]).shape
(3, 2)

property cell_widths_m: Mapping[str, ndarray]#

Return read-only cell widths keyed by axis.

Returns:: Keys are x and z, plus y for 3-D.
Return type:: mapping

Examples

>>> mesh = MaxwellMesh([0, 2, 5], [0, 1, 3])
>>> mesh.cell_widths_m["x"].tolist()
[2.0, 3.0]

property cell_centres_m: Mapping[str, ndarray]#

Return read-only cell centres keyed by axis.

Returns:: Centre coordinates in metres.
Return type:: mapping

Examples

>>> mesh = MaxwellMesh([0, 2, 4], [0, 10, 20])
>>> mesh.cell_centres_m["z"].tolist()
[5.0, 15.0]

to_dict()#

Return a JSON-compatible mesh representation.

Returns:: Versioned mesh state.
Return type:: dict

Examples

>>> MaxwellMesh([0, 1, 2], [0, 1, 2]).to_dict()["schema_version"]
1

classmethod from_dict(data)#

Restore a validated mesh from serialized state.

Parameters:: data (mapping) – State returned by to_dict().
Returns:: Restored immutable mesh.
Return type:: MaxwellMesh

Examples

>>> mesh = MaxwellMesh([0, 1, 2], [0, 2, 4])
>>> MaxwellMesh.from_dict(mesh.to_dict()).shape
(2, 2)

class pycsamt.forward.maxwell.ReceiverSet(coordinates_m, names, orientation_deg=0.0)#

Bases: object

Define named receiver locations in mesh coordinates.

Parameters:

coordinates_m (array-like, shape (n, dimension)) – x,z locations for 2-D or x,y,z locations for 3-D.
names (sequence of str) – Unique receiver or station identifiers.
orientation_deg (float, default=0.0) – Clockwise rotation of receiver x/y axes from the model axes.

Examples

>>> receivers = ReceiverSet([[50, 0], [150, 0]], ["S00", "S01"])
>>> receivers.dimension, receivers.count
(2, 2)

coordinates_m: ndarray#

names: tuple[str, ...]#

orientation_deg: float = 0.0#

property count: int#

Return the number of receivers.

Returns:: Receiver count.
Return type:: int

Examples

>>> ReceiverSet([[0, 0]], ["S00"]).count
1

property dimension: int#

Return the coordinate dimension.

Returns:: Number of coordinate columns.
Return type:: {2, 3}

Examples

>>> ReceiverSet([[0, 0, 0]], ["S00"]).dimension
3

to_dict()#

Return JSON-compatible receiver state.

Returns:: Versioned receiver coordinates and names.
Return type:: dict

Examples

>>> ReceiverSet([[0, 0]], ["S00"]).to_dict()["names"]
['S00']

classmethod from_dict(data)#

Restore receivers from serialized state.

Parameters:: data (mapping) – State returned by to_dict().
Returns:: Validated receiver collection.
Return type:: ReceiverSet

Examples

>>> state = ReceiverSet([[0, 0]], ["S00"]).to_dict()
>>> ReceiverSet.from_dict(state).names
('S00',)

class pycsamt.forward.maxwell.MaxwellProblem(mesh, conductivity_s_m, frequencies_hz, receivers, components=('zxy', 'zyx'), active_cells=None, time_dependence='exp(+iwt)', magnetic_permeability_h_m=1.2566370614359173e-06, metadata=<factory>)#

Bases: object

Define one isotropic frequency-domain MT boundary-value problem.

Parameters:

mesh (MaxwellMesh) – Rectilinear simulation mesh.
conductivity_s_m (array-like) – Positive isotropic conductivity in S/m, shaped like mesh.
frequencies_hz (array-like, shape (n_frequency,)) – Positive, unique frequencies. Input order is retained.
receivers (ReceiverSet) – Observation locations with the same dimension as mesh.
components (sequence of {"zxx", "zxy", "zyx", "zyy"}) – Requested impedance components in explicit output order.
active_cells (array-like of bool or None, optional) – Cells participating in the physical model. Air cells can remain active with a small conductivity or be marked inactive for capable backends.
time_dependence ({"exp(+iwt)", "exp(-iwt)"}, default="exp(+iwt)") – Complex phasor convention.
magnetic_permeability_h_m (float, default=1.25663706212e-6) – Uniform scalar permeability in H/m.
metadata (mapping, optional) – Finite JSON-compatible provenance; never interpreted by a backend.

Examples

>>> mesh = MaxwellMesh([0, 100, 200], [0, 50, 100])
>>> receivers = ReceiverSet([[50, 0]], ["S00"])
>>> problem = MaxwellProblem(
...     mesh, np.full(mesh.shape, 0.01), [10, 1], receivers
... )
>>> problem.problem_hash == problem.problem_hash
True

mesh: MaxwellMesh#

conductivity_s_m: ndarray#

frequencies_hz: ndarray#

receivers: ReceiverSet#

components: tuple[str, ...] = ('zxy', 'zyx')#

active_cells: ndarray | None = None#

time_dependence: str = 'exp(+iwt)'#

magnetic_permeability_h_m: float = 1.2566370614359173e-06#

metadata: Mapping[str, Any]#

property problem_hash: str#

Return a deterministic SHA-256 digest of all physical inputs.

Returns:: Digest suitable for cache keys.
Return type:: str

Examples

>>> mesh = MaxwellMesh([0, 1, 2], [0, 1, 2])
>>> p = MaxwellProblem(
...     mesh, np.ones(mesh.shape), [1], ReceiverSet([[0, 0]], ["S"])
... )
>>> len(p.problem_hash)
64

provenance()#

Return JSON-compatible problem provenance excluding large arrays.

Returns:: Mesh, receiver, convention, components, and metadata state.
Return type:: dict

Examples

>>> mesh = MaxwellMesh([0, 1, 2], [0, 1, 2])
>>> p = MaxwellProblem(
...     mesh, np.ones(mesh.shape), [1], ReceiverSet([[0, 0]], ["S"])
... )
>>> p.provenance()["components"]
['zxy', 'zyx']

to_npz(path)#

Write a pickle-free problem archive.

Parameters:: path (str or pathlib.Path) – Destination archive.
Returns:: Destination path.
Return type:: pathlib.Path

Examples

>>> from tempfile import TemporaryDirectory
>>> p = MaxwellProblem(
...     MaxwellMesh([0, 1, 2], [0, 1, 2]),
...     np.ones((2, 2)),
...     [1],
...     ReceiverSet([[0, 0]], ["S"]),
... )
>>> with TemporaryDirectory() as d:
...     restored = MaxwellProblem.from_npz(p.to_npz(Path(d) / "p.npz"))
>>> restored.problem_hash == p.problem_hash
True

classmethod from_npz(path)#

Restore and validate a problem archive.

Parameters:: path (str or pathlib.Path) – Archive written by to_npz().
Returns:: Restored problem.
Return type:: MaxwellProblem

Examples

>>> from tempfile import TemporaryDirectory
>>> p = MaxwellProblem(
...     MaxwellMesh([0, 1, 2], [0, 1, 2]),
...     np.ones((2, 2)),
...     [1],
...     ReceiverSet([[0, 0]], ["S"]),
... )
>>> with TemporaryDirectory() as d:
...     q = MaxwellProblem.from_npz(p.to_npz(Path(d) / "p.npz"))
>>> np.array_equal(q.conductivity_s_m, p.conductivity_s_m)
True

class pycsamt.forward.maxwell.SolverDiagnostics(converged, iterations, relative_residual, runtime_s, messages=())#

Bases: object

Record convergence information for every frequency and source solve.

Parameters:

converged (array-like of bool, shape (n_frequency, n_source)) – Whether each linear solve met its tolerance.
iterations (array-like of int, same shape) – Iteration count; zero is valid for direct solvers.
relative_residual (array-like of float, same shape) – Final non-negative relative residual.
runtime_s (float) – Total non-negative solver runtime in seconds.
messages (sequence of str, optional) – Backend messages for failed or noteworthy solves.

Examples

>>> d = SolverDiagnostics(
...     [[True], [False]], [[4], [20]], [[1e-8], [1e-2]], 0.5
... )
>>> d.success, d.maximum_relative_residual
(False, 0.01)

converged: ndarray#

iterations: ndarray#

relative_residual: ndarray#

runtime_s: float#

messages: tuple[str, ...] = ()#

property success: bool#

Return whether every solve converged.

Returns:: True only when all convergence flags are true.
Return type:: bool

Examples

>>> SolverDiagnostics([[True]], [[1]], [[1e-9]], 0).success
True

property maximum_relative_residual: float#

Return the largest reported relative residual.

Returns:: Worst solve residual.
Return type:: float

Examples

>>> SolverDiagnostics(
...     [[True]], [[1]], [[1e-7]], 0
... ).maximum_relative_residual
1e-07

to_dict()#

Return JSON-compatible convergence diagnostics.

Returns:: Versioned diagnostic state.
Return type:: dict

Examples

>>> SolverDiagnostics([[True]], [[2]], [[1e-8]], 0.1).to_dict()[
...     "runtime_s"
... ]
0.1

classmethod from_dict(data)#

Restore validated convergence diagnostics.

Parameters:: data (mapping) – State returned by to_dict().
Returns:: Restored diagnostic record.
Return type:: SolverDiagnostics

Examples

>>> state = SolverDiagnostics([[True]], [[2]], [[1e-8]], 0.1).to_dict()
>>> SolverDiagnostics.from_dict(state).success
True

class pycsamt.forward.maxwell.ForwardResult(problem_hash, frequencies_hz, receiver_names, components, impedance_v_a, valid, backend_name, backend_version, diagnostics, metadata=<factory>)#

Bases: object

Store canonical impedance predictions from a Maxwell backend.

Parameters:

problem_hash (str) – Hash of the exact MaxwellProblem solved.
frequencies_hz (array-like) – Frequency vector in problem order.
receiver_names (sequence of str) – Explicit station and tensor-component axes.
components (sequence of str) – Explicit station and tensor-component axes.
impedance_v_a (complex array, shape (station, frequency, component)) – Predicted SI impedance.
valid (bool array or None, optional) – Validity mask with the same shape. Defaults to finite predictions.
backend_name (str) – Solver identity required for reproducibility.
backend_version (str) – Solver identity required for reproducibility.
diagnostics (SolverDiagnostics) – Per-solve convergence record.
metadata (mapping, optional) – Additional finite JSON-compatible backend provenance.

Examples

>>> d = SolverDiagnostics([[True]], [[3]], [[1e-9]], 0.01)
>>> r = ForwardResult(
...     "a" * 64, [1], ["S"], ["zxy"], [[[1 + 2j]]], None, "demo", "1", d
... )
>>> r.shape, r.success
((1, 1, 1), True)

problem_hash: str#

frequencies_hz: ndarray#

receiver_names: tuple[str, ...]#

components: tuple[str, ...]#

impedance_v_a: ndarray#

valid: ndarray | None#

backend_name: str#

backend_version: str#

diagnostics: SolverDiagnostics#

metadata: Mapping[str, Any]#

property shape: tuple[int, int, int]#

Return canonical impedance shape.

Returns:: (station, frequency, component).
Return type:: tuple of int

Examples

>>> d = SolverDiagnostics([[True]], [[0]], [[0]], 0)
>>> ForwardResult(
...     "0" * 64, [1], ["S"], ["zxy"], [[[1j]]], None, "b", "1", d
... ).shape
(1, 1, 1)

property success: bool#

Return whether all solves converged and predictions are valid.

Returns:: Combined numerical and observation validity status.
Return type:: bool

Examples

>>> d = SolverDiagnostics([[True]], [[0]], [[0]], 0)
>>> ForwardResult(
...     "0" * 64, [1], ["S"], ["zxy"], [[[1j]]], None, "b", "1", d
... ).success
True

validate_against(problem)#

Raise if this result does not exactly match a problem contract.

Parameters:: problem (MaxwellProblem) – Expected input problem.
Raises:: ValueError – If hash or any output axis differs.
Return type:: None

Examples

>>> mesh = MaxwellMesh([0, 1, 2], [0, 1, 2])
>>> p = MaxwellProblem(
...     mesh,
...     np.ones((2, 2)),
...     [1],
...     ReceiverSet([[0, 0]], ["S"]),
...     ("zxy",),
... )
>>> d = SolverDiagnostics([[True]], [[0]], [[0]], 0)
>>> ForwardResult(
...     p.problem_hash,
...     [1],
...     ["S"],
...     ["zxy"],
...     [[[1j]]],
...     None,
...     "b",
...     "1",
...     d,
... ).validate_against(p)

provenance()#

Return JSON-compatible solver and output-axis provenance.

Returns:: Problem identity, axes, backend, diagnostics, and metadata.
Return type:: dict

Examples

>>> d = SolverDiagnostics([[True]], [[0]], [[0]], 0)
>>> r = ForwardResult(
...     "0" * 64, [1], ["S"], ["zxy"], [[[1j]]], None, "b", "1", d
... )
>>> r.provenance()["backend_name"]
'b'

to_npz(path)#

Write a versioned, pickle-free result archive.

Parameters:: path (str or pathlib.Path) – Destination archive.
Returns:: Destination path.
Return type:: pathlib.Path

Examples

>>> from tempfile import TemporaryDirectory
>>> d = SolverDiagnostics([[True]], [[0]], [[0]], 0)
>>> r = ForwardResult(
...     "0" * 64, [1], ["S"], ["zxy"], [[[1j]]], None, "b", "1", d
... )
>>> with TemporaryDirectory() as directory:
...     restored = ForwardResult.from_npz(
...         r.to_npz(Path(directory) / "r.npz")
...     )
>>> restored.backend_name, restored.shape
('b', (1, 1, 1))

classmethod from_npz(path)#

Restore and validate a result archive without pickle.

Parameters:: path (str or pathlib.Path) – Archive written by to_npz().
Returns:: Restored canonical result.
Return type:: ForwardResult

Examples

>>> from tempfile import TemporaryDirectory
>>> d = SolverDiagnostics([[True]], [[0]], [[0]], 0)
>>> r = ForwardResult(
...     "0" * 64, [1], ["S"], ["zxy"], [[[1j]]], None, "b", "1", d
... )
>>> with TemporaryDirectory() as directory:
...     restored = ForwardResult.from_npz(
...         r.to_npz(Path(directory) / "r.npz")
...     )
>>> np.array_equal(restored.impedance_v_a, r.impedance_v_a)
True

class pycsamt.forward.maxwell.TriMesh(nodes_m, triangles, region_ids=None, boundary_segments=None, crs=None)#

Bases: object

Describe an unstructured 2-D triangular finite-element mesh.

Parameters:

nodes_m (array-like, shape (n_nodes, 2)) – Node (x, z) coordinates in metres. Depth z increases downward, matching MaxwellMesh.
triangles (array-like of int, shape (n_triangles, 3)) – 0-based node-index connectivity, one row per triangle.
region_ids (array-like of int, shape (n_triangles,), optional) – Material/region label per triangle. Defaults to all zeros (a single region) when omitted.
boundary_segments (array-like of int, shape (n_segments, 2), optional) – 0-based node-index pairs marking PSLG boundary edges (region divides, survey extent). Purely descriptive; not required for a valid mesh.
crs (str or None, optional) – Coordinate reference system for horizontal coordinates.

Examples

>>> mesh = TriMesh(
...     nodes_m=[[0, 0], [100, 0], [50, 50], [150, 60]],
...     triangles=[[0, 1, 2], [1, 3, 2]],
... )
>>> mesh.dimension, mesh.n_nodes, mesh.n_triangles
(2, 4, 2)
>>> mesh.shape
(2,)

nodes_m: ndarray#

triangles: ndarray#

region_ids: ndarray | None = None#

boundary_segments: ndarray | None = None#

crs: str | None = None#

property dimension: int#

Return the spatial dimension.

Returns:: Always 2 for this contract.
Return type:: int

Examples

>>> TriMesh([[0, 0], [1, 0], [0, 1]], [[0, 1, 2]]).dimension
2

property n_nodes: int#

Return the number of mesh nodes.

Examples

>>> TriMesh([[0, 0], [1, 0], [0, 1]], [[0, 1, 2]]).n_nodes
3

property n_triangles: int#

Return the number of mesh triangles.

Examples

>>> TriMesh([[0, 0], [1, 0], [0, 1]], [[0, 1, 2]]).n_triangles
1

property shape: tuple[int]#

Return canonical per-triangle array shape.

Returns:: (n_triangles,), mirroring shape’s role for TriProblem conductivity/active-cell arrays.
Return type:: tuple of int

Examples

>>> TriMesh([[0, 0], [1, 0], [0, 1]], [[0, 1, 2]]).shape
(1,)

property triangle_centroids_m: ndarray#

Return the (x, z) centroid of every triangle.

Returns:: Mean of each triangle’s three node coordinates.
Return type:: ndarray, shape (n_triangles, 2)

Examples

>>> mesh = TriMesh([[0, 0], [3, 0], [0, 3]], [[0, 1, 2]])
>>> mesh.triangle_centroids_m.tolist()
[[1.0, 1.0]]

property triangle_areas_m2: ndarray#

Return the area of every triangle.

Returns:: Positive triangle areas in square metres.
Return type:: ndarray, shape (n_triangles,)

Examples

>>> mesh = TriMesh([[0, 0], [4, 0], [0, 3]], [[0, 1, 2]])
>>> mesh.triangle_areas_m2.tolist()
[6.0]

to_dict()#

Return a JSON-compatible mesh representation.

Returns:: Versioned mesh state.
Return type:: dict

Examples

>>> TriMesh([[0, 0], [1, 0], [0, 1]], [[0, 1, 2]]).to_dict()[
...     "schema_version"
... ]
1

classmethod from_dict(data)#

Restore a validated mesh from serialized state.

Parameters:: data (mapping) – State returned by to_dict().
Returns:: Restored immutable mesh.
Return type:: TriMesh

Examples

>>> mesh = TriMesh([[0, 0], [1, 0], [0, 1]], [[0, 1, 2]])
>>> TriMesh.from_dict(mesh.to_dict()).n_triangles
1

class pycsamt.forward.maxwell.TriProblem(mesh, conductivity_s_m, frequencies_hz, receivers, components=('zxy', 'zyx'), active_cells=None, time_dependence='exp(+iwt)', magnetic_permeability_h_m=1.2566370614359173e-06, metadata=<factory>)#

Bases: object

Define one isotropic frequency-domain MT problem on a triangular mesh.

Parameters:

mesh (TriMesh) – Unstructured simulation mesh.
conductivity_s_m (array-like, shape (n_triangles,)) – Positive isotropic conductivity in S/m, one value per triangle.
frequencies_hz (array-like, shape (n_frequency,)) – Positive, unique frequencies. Input order is retained.
receivers (ReceiverSet) – Observation locations with the same dimension as mesh.
components (sequence of {"zxy", "zyx"}) – Requested impedance components; only the 2-D pair is valid since TriMesh is 2-D-only in this release.
active_cells (array-like of bool or None, optional) – Triangles participating in the physical model.
time_dependence ({"exp(+iwt)", "exp(-iwt)"}, default="exp(+iwt)") – Complex phasor convention.
magnetic_permeability_h_m (float, default=1.25663706212e-6) – Uniform scalar permeability in H/m.
metadata (mapping, optional) – Finite JSON-compatible provenance; never interpreted by a backend.

Examples

>>> from pycsamt.forward.maxwell.contracts import ReceiverSet
>>> mesh = TriMesh([[0, 0], [200, 0], [100, 100]], [[0, 1, 2]])
>>> receivers = ReceiverSet([[100, 0]], ["S00"])
>>> problem = TriProblem(mesh, [0.01], [10, 1], receivers)
>>> problem.problem_hash == problem.problem_hash
True

mesh: TriMesh#

conductivity_s_m: ndarray#

frequencies_hz: ndarray#

receivers: ReceiverSet#

components: tuple[str, ...] = ('zxy', 'zyx')#

active_cells: ndarray | None = None#

time_dependence: str = 'exp(+iwt)'#

magnetic_permeability_h_m: float = 1.2566370614359173e-06#

metadata: Mapping[str, Any]#

property problem_hash: str#

Return a deterministic SHA-256 digest of all physical inputs.

Returns:: Digest suitable for cache keys.
Return type:: str

Examples

>>> from pycsamt.forward.maxwell.contracts import ReceiverSet
>>> mesh = TriMesh([[0, 0], [1, 0], [0, 1]], [[0, 1, 2]])
>>> p = TriProblem(mesh, [1], [1], ReceiverSet([[0, 0]], ["S"]))
>>> len(p.problem_hash)
64

provenance()#

Return JSON-compatible problem provenance excluding large arrays.

Returns:: Mesh, receiver, convention, components, and metadata state.
Return type:: dict

Examples

>>> from pycsamt.forward.maxwell.contracts import ReceiverSet
>>> mesh = TriMesh([[0, 0], [1, 0], [0, 1]], [[0, 1, 2]])
>>> p = TriProblem(mesh, [1], [1], ReceiverSet([[0, 0]], ["S"]))
>>> p.provenance()["components"]
['zxy', 'zyx']

to_npz(path)#

Write a pickle-free problem archive.

Parameters:: path (str or pathlib.Path) – Destination archive.
Returns:: Destination path.
Return type:: pathlib.Path

Examples

>>> from tempfile import TemporaryDirectory
>>> from pycsamt.forward.maxwell.contracts import ReceiverSet
>>> mesh = TriMesh([[0, 0], [1, 0], [0, 1]], [[0, 1, 2]])
>>> p = TriProblem(mesh, [1], [1], ReceiverSet([[0, 0]], ["S"]))
>>> with TemporaryDirectory() as d:
...     restored = TriProblem.from_npz(p.to_npz(Path(d) / "p.npz"))
>>> restored.problem_hash == p.problem_hash
True

classmethod from_npz(path)#

Restore and validate a problem archive.

Parameters:: path (str or pathlib.Path) – Archive written by to_npz().
Returns:: Restored problem.
Return type:: TriProblem

Examples

>>> from tempfile import TemporaryDirectory
>>> from pycsamt.forward.maxwell.contracts import ReceiverSet
>>> mesh = TriMesh([[0, 0], [1, 0], [0, 1]], [[0, 1, 2]])
>>> p = TriProblem(mesh, [1], [1], ReceiverSet([[0, 0]], ["S"]))
>>> with TemporaryDirectory() as d:
...     q = TriProblem.from_npz(p.to_npz(Path(d) / "p.npz"))
>>> np.array_equal(q.conductivity_s_m, p.conductivity_s_m)
True

class pycsamt.forward.maxwell.BackendCapabilities(name, version, dimensions, components, time_conventions=('exp(+iwt)',), supports_nonuniform_mesh=True, supports_inactive_cells=False, supports_topography=False, supports_anisotropy=False, maximum_cells=None, maximum_frequencies=None, verified_benchmarks=())#

Bases: object

Declare a Maxwell adapter’s supported physical and numerical scope.

Parameters:

name (str) – Stable backend identifier and adapter/solver version.
version (str) – Stable backend identifier and adapter/solver version.
dimensions (tuple containing 2 and/or 3) – Spatial problem dimensions genuinely supported.
components (tuple of str) – Impedance tensor components the adapter can return.
time_conventions (tuple of str, default=("exp(+iwt)",)) – Phasor conventions accepted without conversion.
supports_nonuniform_mesh (bool, default=True) – Whether variable cell widths are supported.
supports_inactive_cells (bool, default=False) – Whether MaxwellProblem.active_cells is honored.
supports_topography (bool, default=False) – Whether an inactive/air mask may describe non-flat terrain.
supports_anisotropy (bool, default=False) – Reserved declaration for future tensor-conductivity contracts.
maximum_cells (int or None, optional) – Enforced adapter limits. None means no declared limit.
maximum_frequencies (int or None, optional) – Enforced adapter limits. None means no declared limit.
verified_benchmarks (tuple of str, optional) – Stable benchmark identifiers passed by this adapter version.

Examples

>>> capability = BackendCapabilities("mt2d", "1.0", (2,), ("zxy", "zyx"))
>>> capability.supports_dimension(2), capability.supports_component("zxy")
(True, True)

name: str#

version: str#

dimensions: tuple[int, ...]#

components: tuple[str, ...]#

time_conventions: tuple[str, ...] = ('exp(+iwt)',)#

supports_nonuniform_mesh: bool = True#

supports_inactive_cells: bool = False#

supports_topography: bool = False#

supports_anisotropy: bool = False#

maximum_cells: int | None = None#

maximum_frequencies: int | None = None#

verified_benchmarks: tuple[str, ...] = ()#

supports_dimension(dimension)#

Return whether a spatial dimension is supported.

Parameters:: dimension (int) – Requested dimension.
Returns:: Capability status.
Return type:: bool

Examples

>>> BackendCapabilities("b", "1", (2,), ("zxy",)).supports_dimension(3)
False

supports_component(component)#

Return whether an impedance component is supported.

Parameters:: component (str) – Canonical tensor component name.
Returns:: Capability status.
Return type:: bool

Examples

>>> BackendCapabilities("b", "1", (2,), ("zxy",)).supports_component(
...     "ZYX"
... )
False

assess(problem)#

Assess a problem without invoking the numerical backend.

Parameters:: problem (MaxwellProblem or TriProblem) – Validated problem contract, rectilinear or triangular. Both expose the same mesh/frequencies_hz/components/ time_dependence/active_cells shape this method reads; only cell_widths_m is rectilinear-only, so the nonuniform-mesh check below is skipped (not applicable) for an unstructured TriMesh.
Returns:: All hard errors plus advisory validation warnings.
Return type:: CompatibilityReport

Examples

>>> from .contracts import MaxwellMesh, ReceiverSet
>>> mesh = MaxwellMesh([0, 1, 2], [0, 1, 2])
>>> problem = MaxwellProblem(
...     mesh, np.ones((2, 2)), [1], ReceiverSet([[0, 0]], ["S"])
... )
>>> BackendCapabilities("b", "1", (2,), ("zxy", "zyx")).assess(
...     problem
... ).compatible
True

to_dict()#

Return a JSON-compatible capability declaration.

Returns:: Versioned capability state.
Return type:: dict

Examples

>>> BackendCapabilities("b", "1", (2,), ("zxy",)).to_dict()[
...     "dimensions"
... ]
[2]

classmethod from_dict(data)#

Restore a validated capability declaration.

Parameters:: data (mapping) – State returned by to_dict().
Returns:: Restored declaration.
Return type:: BackendCapabilities

Examples

>>> cap = BackendCapabilities("b", "1", (2,), ("zxy",))
>>> BackendCapabilities.from_dict(cap.to_dict()).name
'b'

class pycsamt.forward.maxwell.CompatibilityReport(backend_name, compatible, errors=(), warnings=())#

Bases: object

Describe whether a backend can solve a particular problem.

Parameters:

backend_name (str) – Normalized backend identifier.
compatible (bool) – Whether all hard capability requirements are satisfied.
errors (tuple of str, optional) – Hard incompatibilities and advisory concerns.
warnings (tuple of str, optional) – Hard incompatibilities and advisory concerns.

Examples

>>> report = CompatibilityReport("demo", False, ("3-D unsupported",))
>>> report.require()
Traceback (most recent call last):
...
ValueError: backend 'demo' is incompatible: 3-D unsupported

backend_name: str#

compatible: bool#

errors: tuple[str, ...] = ()#

warnings: tuple[str, ...] = ()#

require()#

Raise a consolidated error when the report is incompatible.

Raises:: ValueError – If one or more hard incompatibilities were found.
Return type:: None

Examples

>>> CompatibilityReport("demo", True).require()

to_dict()#

Return a JSON-compatible report.

Returns:: Compatibility state and messages.
Return type:: dict

Examples

>>> CompatibilityReport("demo", True).to_dict()["compatible"]
True

class pycsamt.forward.maxwell.MaxwellBackend(*args, **kwargs)#

Bases: Protocol

Runtime-checkable interface implemented by Maxwell adapters.

Examples

A conforming adapter exposes immutable capabilities and a solve method.

>>> class Demo:
...     capabilities = BackendCapabilities(
...         "demo", "1", (2,), ("zxy", "zyx")
...     )
...
...     def solve(self, problem):
...         raise NotImplementedError
>>> isinstance(Demo(), MaxwellBackend)
True

property capabilities: BackendCapabilities#: Return the adapter’s immutable capability declaration.

solve(problem)#

Solve a compatible problem and return canonical output.

Parameters:: problem (MaxwellProblem)
Return type:: ForwardResult

class pycsamt.forward.maxwell.BackendRegistration(capabilities, factory, availability_probe=None)#

Bases: object

Store one lazy backend factory and its availability probe.

Parameters:

capabilities (BackendCapabilities) – Static capability declaration, available without importing the solver.
factory (callable) – Factory returning a MaxwellBackend.
availability_probe (callable or None, optional) – Lightweight function returning (available, reason).

Examples

>>> cap = BackendCapabilities("demo", "1", (2,), ("zxy",))
>>> registration = BackendRegistration(cap, lambda: object())
>>> registration.availability()
(True, None)

capabilities: BackendCapabilities#

factory: Callable[[...], MaxwellBackend]#

availability_probe: Callable[[], tuple[bool, str | None]] | None = None#

availability()#

Return whether the optional backend can currently be created.

Returns:

available (bool) – Probe status.
reason (str or None) – Human-readable reason when unavailable.

Return type:

tuple[bool, str | None]

Examples

>>> cap = BackendCapabilities("demo", "1", (2,), ("zxy",))
>>> BackendRegistration(
...     cap, lambda: None, lambda: (False, "missing")
... ).availability()
(False, 'missing')

create(**options)#

Create and validate a backend instance.

Parameters:: **options (Any) – Backend-specific constructor options.
Returns:: Conforming adapter instance.
Return type:: MaxwellBackend
Raises:: RuntimeError – If unavailable or the factory violates its registration.

Examples

>>> cap = BackendCapabilities("demo", "1", (2,), ("zxy",))
>>> class Demo:
...     capabilities = cap
...
...     def solve(self, problem):
...         raise NotImplementedError
>>> BackendRegistration(cap, Demo).create().capabilities.name
'demo'

class pycsamt.forward.maxwell.BackendRegistry#

Bases: object

Thread-safe registry of lazy Maxwell backend factories.

Examples

>>> registry = BackendRegistry()
>>> cap = BackendCapabilities("demo", "1", (2,), ("zxy",))
>>> registry.register(BackendRegistration(cap, lambda: None))
>>> registry.names()
('demo',)

register(registration, *, replace=False)#

Register a backend under its declared capability name.

Parameters:

registration (BackendRegistration) – Lazy backend definition.
replace (bool, default=False) – Explicitly replace an existing registration.

Raises:

ValueError – If the name already exists and replacement was not requested.

Return type:

None

Examples

>>> registry = BackendRegistry()
>>> cap = BackendCapabilities("demo", "1", (2,), ("zxy",))
>>> registry.register(BackendRegistration(cap, lambda: None))

unregister(name)#

Remove and return a registration.

Parameters:: name (str) – Backend identifier.
Returns:: Removed registration.
Return type:: BackendRegistration

Examples

>>> registry = BackendRegistry()
>>> cap = BackendCapabilities("demo", "1", (2,), ("zxy",))
>>> registry.register(BackendRegistration(cap, lambda: None))
>>> registry.unregister("demo").capabilities.name
'demo'

get(name)#

Return one registration without creating its backend.

Parameters:: name (str) – Backend identifier.
Returns:: Lazy registration.
Return type:: BackendRegistration

Examples

>>> registry = BackendRegistry()
>>> cap = BackendCapabilities("demo", "1", (2,), ("zxy",))
>>> registry.register(BackendRegistration(cap, lambda: None))
>>> registry.get("DEMO").capabilities.version
'1'

names(*, available_only=False)#

Return sorted registered backend names.

Parameters:: available_only (bool, default=False) – Exclude registrations whose availability probe fails.
Returns:: Stable sorted names.
Return type:: tuple of str

Examples

>>> BackendRegistry().names()
()

describe()#

Return immutable availability and capability summaries.

Returns:: Backend names mapped to JSON-compatible summaries.
Return type:: mapping

Examples

>>> BackendRegistry().describe() == {}
True

create(name, **options)#

Create a named backend through its lazy factory.

Parameters:

name (str) – Registered backend identifier.
**options – Backend-specific constructor options.

Returns:

Validated adapter.

Return type:

MaxwellBackend

Examples

See BackendRegistration.create() for a complete adapter example.

pycsamt.forward.maxwell.register_backend(registration, *, replace=False)#

Register a lazy backend in the process-wide registry.

Parameters:

registration (BackendRegistration) – Backend definition.
replace (bool, default=False) – Explicitly replace an existing name.

Return type:

None

Examples

Prefer a private BackendRegistry in isolated applications and tests; this function is intended for adapter package initialization.

pycsamt.forward.maxwell.unregister_backend(name)#

Remove a backend from the process-wide registry.

Parameters:: name (str) – Backend identifier.
Returns:: Removed definition.
Return type:: BackendRegistration

Examples

This operation is primarily useful for plugin teardown and tests.

pycsamt.forward.maxwell.create_backend(name, **options)#

Create a registered Maxwell adapter lazily.

Parameters:

name (str) – Backend identifier.
**options – Backend-specific constructor options.

Returns:

Validated backend instance.

Return type:

MaxwellBackend

Examples

Backend packages register their factories before this function is called.

pycsamt.forward.maxwell.list_backends(*, available_only=False)#

Describe registered Maxwell backends without creating them.

Parameters:: available_only (bool, default=False) – Exclude unavailable optional backends.
Returns:: Immutable capability and availability summaries.
Return type:: mapping

Examples

>>> isinstance(list_backends(), Mapping)
True

exception pycsamt.forward.maxwell.MaxwellAdapterError#

Bases: RuntimeError

Base exception raised by the validated adapter execution layer.

Examples

>>> error = MaxwellAdapterError("solver failed")
>>> str(error)
'solver failed'

exception pycsamt.forward.maxwell.IncompatibleProblemError#

Bases: MaxwellAdapterError

Indicate that declared backend capabilities reject a problem.

Examples

>>> isinstance(
...     IncompatibleProblemError("unsupported"), MaxwellAdapterError
... )
True

exception pycsamt.forward.maxwell.BackendExecutionError#

Bases: MaxwellAdapterError

Wrap an exception raised inside a numerical backend.

Examples

>>> isinstance(BackendExecutionError("failed"), MaxwellAdapterError)
True

exception pycsamt.forward.maxwell.InvalidBackendResultError#

Bases: MaxwellAdapterError

Indicate malformed, mislabeled, or mismatched backend output.

Examples

>>> isinstance(InvalidBackendResultError("bad axes"), MaxwellAdapterError)
True

exception pycsamt.forward.maxwell.SolverConvergenceError#

Bases: MaxwellAdapterError

Indicate that a valid result violates the convergence policy.

Examples

>>> isinstance(
...     SolverConvergenceError("residual too large"), MaxwellAdapterError
... )
True

class pycsamt.forward.maxwell.AdapterPolicy(require_convergence=True, maximum_relative_residual=None, require_all_valid=True, emit_capability_warnings=True, wrap_backend_exceptions=True)#

Bases: object

Configure solver-independent result acceptance rules.

Parameters:

require_convergence (bool, default=True) – Reject a result when any solve reports non-convergence.
maximum_relative_residual (float or None, optional) – Reject a result whose worst reported residual exceeds this value. None delegates residual acceptance entirely to the backend.
require_all_valid (bool, default=True) – Reject results containing masked or non-finite observations.
emit_capability_warnings (bool, default=True) – Emit advisory messages from capability assessment.
wrap_backend_exceptions (bool, default=True) – Wrap ordinary backend exceptions in BackendExecutionError.

Examples

>>> policy = AdapterPolicy(maximum_relative_residual=1e-6)
>>> policy.maximum_relative_residual
1e-06

require_convergence: bool = True#

maximum_relative_residual: float | None = None#

require_all_valid: bool = True#

emit_capability_warnings: bool = True#

wrap_backend_exceptions: bool = True#

class pycsamt.forward.maxwell.BaseMaxwellAdapter(capabilities, policy=None)#

Bases: ABC

Base class enforcing common preflight and postflight validation.

Parameters:

capabilities (BackendCapabilities) – Immutable declaration for the exact backend version.
policy (AdapterPolicy or None, optional) – Result acceptance policy. Defaults to AdapterPolicy.

Notes

Implementations override only _solve_backend(). They must return a canonical ForwardResult; all validation is performed by solve().

Examples

See CallableMaxwellAdapter for a minimal concrete adapter.

property capabilities: BackendCapabilities#

Return the immutable backend capability declaration.

Returns:: Physical and numerical scope of this adapter.
Return type:: BackendCapabilities

Examples

>>> cap = BackendCapabilities("demo", "1", (2,), ("zxy",))
>>> CallableMaxwellAdapter(
...     cap, lambda problem: None
... ).capabilities is cap
True

property policy: AdapterPolicy#

Return the immutable result acceptance policy.

Returns:: Policy applied after every solve.
Return type:: AdapterPolicy

Examples

>>> cap = BackendCapabilities("demo", "1", (2,), ("zxy",))
>>> CallableMaxwellAdapter(
...     cap, lambda problem: None
... ).policy.require_convergence
True

assess(problem)#

Assess a problem against declared backend capabilities.

Parameters:: problem (MaxwellProblem) – Candidate simulation problem.
Returns:: Consolidated errors and warnings without invoking the solver.
Return type:: CompatibilityReport

Examples

>>> from .contracts import MaxwellMesh, ReceiverSet
>>> mesh = MaxwellMesh([0, 1, 2], [0, 1, 2])
>>> problem = MaxwellProblem(
...     mesh,
...     np.ones((2, 2)),
...     [1],
...     ReceiverSet([[0, 0]], ["S"]),
...     ("zxy",),
... )
>>> cap = BackendCapabilities("demo", "1", (2,), ("zxy",))
>>> CallableMaxwellAdapter(cap, lambda problem: None).assess(
...     problem
... ).compatible
True

solve(problem)#

Validate, execute, and verify one Maxwell problem.

Parameters:

problem (MaxwellProblem) – Solver-neutral simulation input.

Returns:

Canonical, problem-matched impedance result.

Return type:

ForwardResult

Raises:

IncompatibleProblemError – If preflight capability checks fail.
BackendExecutionError – If the numerical backend raises an ordinary exception.
InvalidBackendResultError – If returned output violates the result contract or problem axes.
SolverConvergenceError – If convergence, residual, or validity policy fails.

Examples

Concrete execution examples require a backend callback; see CallableMaxwellAdapter.

solve_many(problems)#

Solve problems sequentially while preserving input order.

Parameters:: problems (iterable of MaxwellProblem) – Finite problem stream. Execution stops at the first failure.
Returns:: Results in exactly the supplied order.
Return type:: tuple of ForwardResult

Examples

An empty collection performs no backend calls:

>>> cap = BackendCapabilities("demo", "1", (2,), ("zxy",))
>>> CallableMaxwellAdapter(cap, lambda problem: None).solve_many([])
()

class pycsamt.forward.maxwell.CallableMaxwellAdapter(capabilities, solver, policy=None)#

Bases: BaseMaxwellAdapter

Adapt a trusted callable to the validated Maxwell backend interface.

Parameters:

capabilities (BackendCapabilities) – Static declaration matching callback output identity.
solver (callable) – Function accepting MaxwellProblem and returning ForwardResult.
policy (AdapterPolicy or None, optional) – Solver-independent acceptance policy.

Examples

>>> from .contracts import MaxwellMesh, ReceiverSet, SolverDiagnostics
>>> mesh = MaxwellMesh([0, 1, 2], [0, 1, 2])
>>> problem = MaxwellProblem(
...     mesh, np.ones((2, 2)), [1], ReceiverSet([[0, 0]], ["S"]), ("zxy",)
... )
>>> cap = BackendCapabilities(
...     "demo", "1", (2,), ("zxy",), verified_benchmarks=("half-space",)
... )
>>> def solver(value):
...     diagnostics = SolverDiagnostics([[True]], [[0]], [[0]], 0)
...     return ForwardResult(
...         value.problem_hash,
...         value.frequencies_hz,
...         value.receivers.names,
...         value.components,
...         [[[1j]]],
...         None,
...         "demo",
...         "1",
...         diagnostics,
...     )
>>> CallableMaxwellAdapter(cap, solver).solve(problem).success
True

class pycsamt.forward.maxwell.MeshDesign(horizontal_padding_cells=6, bottom_padding_cells=8, air_layers=8, padding_expansion=1.35, air_expansion=1.25, air_conductivity_s_m=1e-08, minimum_cells_per_skin_depth=4.0, maximum_adjacent_ratio=1.5, maximum_aspect_ratio=20.0)#

Bases: object

Configure geometric padding, air treatment, and quality targets.

Parameters:

horizontal_padding_cells (int or pair of int, default=6) – Number of padding cells before/after each horizontal core axis. A scalar is applied symmetrically.
bottom_padding_cells (int, default=8) – Number of cells beneath the geological model.
air_layers (int, default=8) – Number of cells above the geological reference surface.
padding_expansion (float, default=1.35) – Geometric growth factor away from the core mesh.
air_expansion (float, default=1.25) – Geometric growth factor upward through the air.
air_conductivity_s_m (float, default=1e-8) – Positive numerical conductivity assigned to air cells.
minimum_cells_per_skin_depth (float, default=4.0) – Advisory resolution target evaluated at the smallest skin depth.
maximum_adjacent_ratio (float, default=1.5) – Advisory upper bound for adjacent cell-width ratios.
maximum_aspect_ratio (float, default=20.0) – Advisory upper bound across cell widths.

Examples

>>> design = MeshDesign(horizontal_padding_cells=(3, 5), air_layers=4)
>>> design.horizontal_padding
(3, 5)

horizontal_padding_cells: int | tuple[int, int] = 6#

bottom_padding_cells: int = 8#

air_layers: int = 8#

padding_expansion: float = 1.35#

air_expansion: float = 1.25#

air_conductivity_s_m: float = 1e-08#

minimum_cells_per_skin_depth: float = 4.0#

maximum_adjacent_ratio: float = 1.5#

maximum_aspect_ratio: float = 20.0#

property horizontal_padding: tuple[int, int]#

Return normalized before/after horizontal padding counts.

Returns:: Padding cells on the low and high sides of each horizontal axis.
Return type:: tuple of int

Examples

>>> MeshDesign(horizontal_padding_cells=3).horizontal_padding
(3, 3)

to_dict()#

Return a JSON-compatible design representation.

Returns:: Versioned design state.
Return type:: dict

Examples

>>> MeshDesign(air_layers=2).to_dict()["air_layers"]
2

classmethod from_dict(data)#

Restore a validated mesh design.

Parameters:: data (mapping) – State returned by to_dict().
Returns:: Restored design.
Return type:: MeshDesign

Examples

>>> design = MeshDesign(horizontal_padding_cells=2)
>>> MeshDesign.from_dict(design.to_dict()).horizontal_padding
(2, 2)

class pycsamt.forward.maxwell.MeshQuality(cell_count, minimum_cell_width_m, maximum_cell_width_m, maximum_aspect_ratio, maximum_adjacent_ratio, minimum_skin_depth_m, cells_per_minimum_skin_depth, warnings=())#

Bases: object

Summarize numerical mesh quality and skin-depth resolution.

Parameters:

cell_count (int) – Total number of cells including air and padding.
minimum_cell_width_m (float) – Extreme cell widths over all axes.
maximum_cell_width_m (float) – Extreme cell widths over all axes.
maximum_aspect_ratio (float) – Ratio of maximum to minimum cell width.
maximum_adjacent_ratio (float) – Worst neighboring width expansion on any axis.
minimum_skin_depth_m (float) – Smallest skin depth across the requested physics range.
cells_per_minimum_skin_depth (float) – Skin depth divided by the largest core cell width.
warnings (tuple of str) – Advisory quality violations.

Examples

>>> quality = MeshQuality(10, 1, 5, 5, 1.2, 100, 20, ())
>>> quality.acceptable
True

cell_count: int#

minimum_cell_width_m: float#

maximum_cell_width_m: float#

maximum_aspect_ratio: float#

maximum_adjacent_ratio: float#

minimum_skin_depth_m: float#

cells_per_minimum_skin_depth: float#

warnings: tuple[str, ...] = ()#

property acceptable: bool#

Return whether no advisory quality limits were violated.

Returns:: True when warnings is empty.
Return type:: bool

Examples

>>> MeshQuality(1, 1, 1, 1, 1, 1, 1, ("coarse",)).acceptable
False

to_dict()#

Return JSON-compatible mesh-quality diagnostics.

Returns:: Numeric diagnostics and warnings.
Return type:: dict

Examples

>>> MeshQuality(1, 1, 1, 1, 1, 1, 1).to_dict()["cell_count"]
1

class pycsamt.forward.maxwell.SolverMeshModel(mesh, conductivity_s_m, earth_mask, core_slices, design, quality, source_shape)#

Bases: object

Store a padded mesh, conductivity, regions, and construction record.

Parameters:

mesh (MaxwellMesh) – Solver-neutral padded mesh.
conductivity_s_m (ndarray) – Positive conductivity shaped like mesh. Air cells contain the configured small numerical conductivity.
earth_mask (ndarray of bool) – True for cells on or below local terrain.
core_slices (tuple of slice) – Geological core location in canonical array order.
design (MeshDesign) – Construction settings.
quality (MeshQuality) – Mesh-quality diagnostics for the requested frequency range.
source_shape (tuple of int) – Original geological model shape.

Examples

Instances are normally created with build_solver_mesh().

mesh: MaxwellMesh#

conductivity_s_m: ndarray#

earth_mask: ndarray#

core_slices: tuple[slice, ...]#

design: MeshDesign#

quality: MeshQuality#

source_shape: tuple[int, ...]#

property air_mask: ndarray#

Return the read-only complement of the earth mask.

Returns:: Air-region mask shaped like mesh.
Return type:: ndarray of bool

Examples

air_mask and earth_mask always partition the complete mesh.

property model_hash: str#

Return a deterministic digest of mesh, model, regions, and design.

Returns:: SHA-256 digest suitable for provenance checks.
Return type:: str

Examples

A valid model hash always contains 64 hexadecimal characters.

assess_receivers(receivers)#

Return receiver-placement errors without modifying coordinates.

Parameters:: receivers (ReceiverSet) – Candidate locations in the mesh coordinate system.
Returns:: Empty when all receivers lie inside mesh bounds and no receiver is below the discretized local terrain surface.
Return type:: tuple of str

Examples

Use this check before to_problem() when receiver coordinates are assembled independently from the mesh.

to_problem(frequencies_hz, receivers, *, components=('zxy', 'zyx'), mark_air_inactive=False, time_dependence='exp(+iwt)', magnetic_permeability_h_m=1.2566370614359173e-06, metadata=None)#

Create a validated Maxwell problem from this mesh model.

Parameters:

frequencies_hz (sequence of float) – Positive simulation frequencies.
receivers (ReceiverSet) – Receiver locations matching the mesh dimension.
components (sequence of str, default=("zxy", "zyx")) – Requested canonical impedance components.
mark_air_inactive (bool, default=False) – Use earth_mask as active cells. Keep false for formulations that solve conductive air explicitly.
time_dependence (str, default="exp(+iwt)") – Complex phasor convention.
magnetic_permeability_h_m (float, default=4e-7*pi) – Uniform magnetic permeability.
metadata (mapping or None, optional) – Additional problem provenance.

Returns:

Solver-neutral problem ready for adapter assessment.

Return type:

MaxwellProblem

Examples

The generated problem includes mesh_model_hash in its metadata.

provenance()#

Return JSON-compatible mesh-construction provenance.

Returns:: Mesh, design, quality, source shape, and core slices.
Return type:: dict

Examples

The returned schema version is currently one.

to_npz(path)#

Persist a solver mesh model without enabling pickle.

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

Examples

Archives can be restored with from_npz().

classmethod from_npz(path)#

Restore and validate a solver mesh archive without pickle.

Parameters:: path (str or pathlib.Path) – Archive written by to_npz().
Returns:: Restored immutable mesh model.
Return type:: SolverMeshModel

Examples

Restored arrays remain read-only after construction.

pycsamt.forward.maxwell.skin_depth_m(resistivity_ohm_m, frequency_hz)#

Calculate electromagnetic skin depth for a non-magnetic conductor.

Parameters:

resistivity_ohm_m (array-like) – Positive resistivity and frequency, broadcast using NumPy rules.
frequency_hz (array-like) – Positive resistivity and frequency, broadcast using NumPy rules.

Returns:

Skin depth in metres, sqrt(rho / (pi * mu0 * f)).

Return type:

ndarray

Examples

>>> round(float(skin_depth_m(100, 1)))
5033
>>> skin_depth_m([100, 400], 1).shape
(2,)

pycsamt.forward.maxwell.build_solver_mesh(grid, *, conductivity_s_m=None, resistivity_ohm_m=None, frequencies_hz, topography=None, design=None)#

Build a padded Maxwell mesh from a geological cell-centre model.

Parameters:

grid (pycsamt.ai.geology.GeologyGrid) – Source grid in canonical geological order. Its upper cell edge must be depth zero, the reference used by topography and receiver coordinates.
conductivity_s_m (array-like or None) – Supply exactly one positive model shaped like grid.
resistivity_ohm_m (array-like or None) – Supply exactly one positive model shaped like grid.
frequencies_hz (sequence of float) – Frequencies used only for skin-depth quality diagnostics.
topography (pycsamt.ai.geology.TopographicSurface or None, optional) – Terrain aligned to grid. When omitted, the geological top edge is treated as a flat earth surface.
design (MeshDesign or None, optional) – Padding and quality configuration.

Returns:

Padded conductivity, earth/air regions, core mapping, and diagnostics.

Return type:

SolverMeshModel

Examples

>>> from pycsamt.ai.geology import GeologyGrid
>>> grid = GeologyGrid.regular_2d(nx=3, nz=2, dx_m=100, dz_m=50)
>>> model = build_solver_mesh(
...     grid,
...     resistivity_ohm_m=np.full(grid.shape, 100),
...     frequencies_hz=[10, 1],
...     design=MeshDesign(
...         horizontal_padding_cells=1,
...         bottom_padding_cells=1,
...         air_layers=1,
...     ),
... )
>>> model.mesh.shape, model.core_slices
((4, 5), (slice(1, 3, None), slice(1, 4, None)))

exception pycsamt.forward.maxwell.CacheCorruptionError#

Bases: RuntimeError

Indicate that a cached archive failed integrity validation.

Examples

>>> isinstance(CacheCorruptionError("bad checksum"), RuntimeError)
True

exception pycsamt.forward.maxwell.CacheLockTimeoutError#

Bases: TimeoutError

Indicate that a cache-key lock could not be acquired in time.

Examples

>>> isinstance(CacheLockTimeoutError("busy"), TimeoutError)
True

class pycsamt.forward.maxwell.CacheEntry(key, archive_path, size_bytes, modified_time_s)#

Bases: object

Describe one complete cache entry.

Parameters:

key (str) – Problem SHA-256 digest.
archive_path (pathlib.Path) – Result archive location.
size_bytes (int) – Combined archive and checksum size.
modified_time_s (float) – Archive modification time as Unix seconds.

Examples

>>> entry = CacheEntry("0" * 64, Path("result.npz"), 10, 1.0)
>>> entry.size_bytes
10

key: str#

archive_path: Path#

size_bytes: int#

modified_time_s: float#

property checksum_path: Path#

Return the SHA-256 sidecar path.

Returns:: Archive path with .sha256 appended.
Return type:: pathlib.Path

Examples

>>> entry = CacheEntry("0" * 64, Path("r.npz"), 0, 0)
>>> entry.checksum_path.name
'r.npz.sha256'

to_dict()#

Return JSON-compatible entry metadata.

Returns:: Key, path, size, and modification time.
Return type:: dict

Examples

>>> entry = CacheEntry("0" * 64, Path("r.npz"), 5, 2)
>>> entry.to_dict()["size_bytes"]
5

class pycsamt.forward.maxwell.CacheStatistics(entry_count, total_bytes, orphan_count, corrupt_count)#

Bases: object

Summarize the current on-disk cache state.

Parameters:

entry_count (int) – Counts and storage for normal, incomplete, and quarantined files.
total_bytes (int) – Counts and storage for normal, incomplete, and quarantined files.
orphan_count (int) – Counts and storage for normal, incomplete, and quarantined files.
corrupt_count (int) – Counts and storage for normal, incomplete, and quarantined files.

Examples

>>> CacheStatistics(2, 100, 0, 1).entry_count
2

entry_count: int#

total_bytes: int#

orphan_count: int#

corrupt_count: int#

to_dict()#

Return JSON-compatible cache statistics.

Returns:: Entry, byte, orphan, and corruption counts.
Return type:: dict

Examples

>>> CacheStatistics(1, 20, 0, 0).to_dict()["total_bytes"]
20

class pycsamt.forward.maxwell.MaxwellResultCache(root, *, lock_timeout_s=300.0, poll_interval_s=0.05, stale_lock_s=3600.0, quarantine_corrupt=True)#

Bases: object

Manage a validated, content-addressed result cache.

Parameters:

root (str or pathlib.Path) – Dedicated cache directory. It is created when absent.
lock_timeout_s (float, default=300) – Maximum wait for another worker holding the same problem key.
poll_interval_s (float, default=0.05) – Delay between lock acquisition attempts.
stale_lock_s (float, default=3600) – Age after which an abandoned lock can be removed.
quarantine_corrupt (bool, default=True) – Move corrupt files under root/corrupt. When false, reads raise CacheCorruptionError and leave the entry untouched.

Examples

>>> from tempfile import TemporaryDirectory
>>> with TemporaryDirectory() as directory:
...     cache = MaxwellResultCache(directory)
...     cache.statistics().entry_count
0

property root: Path#

Return the resolved cache root.

Returns:: Dedicated cache directory.
Return type:: pathlib.Path

Examples

The returned path is always absolute.

contains(problem)#

Return whether a complete entry exists for a problem.

Parameters:: problem (MaxwellProblem) – Problem whose content hash identifies the entry.
Returns:: True when both archive and checksum sidecar exist.
Return type:: bool

Examples

A newly created cache contains no problems.

get(problem)#

Load and validate a cached result.

Parameters:: problem (MaxwellProblem) – Exact problem expected by the caller.
Returns:: Valid result, or None when no complete entry exists. Corruption is quarantined and treated as a miss when configured.
Return type:: ForwardResult or None
Raises:: CacheCorruptionError – If validation fails and quarantine is disabled.

Examples

Cache misses return None rather than raising KeyError.

put(problem, result, *, overwrite=False)#

Validate and atomically store one result.

Parameters:

problem (MaxwellProblem) – Exact simulation input.
result (ForwardResult) – Canonical result matching problem.
overwrite (bool, default=False) – Replace an existing complete entry. Otherwise the validated existing entry is retained.

Returns:

Metadata for the stored or retained entry.

Return type:

CacheEntry

Examples

Invalid problem/result pairs are rejected before writing an archive.

get_or_solve(problem, backend)#

Return a hit or solve and cache one problem under a key lock.

Parameters:

problem (MaxwellProblem) – Simulation input and cache identity.
backend (MaxwellBackend) – Conforming backend used only after a cache miss.

Returns:

Valid cached or newly computed result.

Return type:

ForwardResult

Notes

The second read after locking prevents duplicate concurrent work.

Examples

Backend invocation is skipped whenever a validated hit exists.

entry(key)#

Return filesystem metadata for a complete entry.

Parameters:: key (str) – Problem SHA-256 digest.
Returns:: Entry paths, size, and modification time.
Return type:: CacheEntry
Raises:: KeyError – If archive or checksum sidecar is missing.

Examples

This method does not deserialize the result archive.

entries()#

Return complete entries sorted by problem key.

Returns:: Stable snapshot of complete cache entries.
Return type:: tuple of CacheEntry

Examples

An empty cache returns an empty tuple.

remove(problem)#

Remove one problem entry under its key lock.

Parameters:: problem (MaxwellProblem) – Exact problem identifying the entry.
Returns:: True when at least one entry file was removed.
Return type:: bool

Examples

Removing an absent problem is a no-op returning False.

prune(maximum_bytes)#

Remove oldest entries until storage is within a byte budget.

Parameters:: maximum_bytes (int) – Non-negative archive and checksum budget.
Returns:: Entries removed, oldest first.
Return type:: tuple of CacheEntry

Examples

prune(0) removes every complete entry but leaves infrastructure.

statistics()#

Inspect complete, orphaned, and quarantined cache files.

Returns:: Current cache counts and complete-entry storage.
Return type:: CacheStatistics

Examples

Statistics inspect metadata without deserializing archives.

class pycsamt.forward.maxwell.BenchmarkThresholds(maximum_normalized_rms=0.05, maximum_amplitude_relative_error=0.05, maximum_phase_error_deg=2.0, minimum_valid_fraction=1.0, require_convergence=True)#

Bases: object

Define quantitative acceptance limits for one benchmark.

Parameters:

maximum_normalized_rms (float, default=0.05) – Maximum complex root-sum-square error normalized by the reference.
maximum_amplitude_relative_error (float, default=0.05) – Maximum pointwise relative impedance-amplitude error.
maximum_phase_error_deg (float, default=2.0) – Maximum absolute circular phase error in degrees.
minimum_valid_fraction (float, default=1.0) – Minimum fraction of output values marked valid.
require_convergence (bool, default=True) – Require every solve in backend diagnostics to converge.

Examples

>>> limits = BenchmarkThresholds(maximum_phase_error_deg=1)
>>> limits.maximum_phase_error_deg
1.0

maximum_normalized_rms: float = 0.05#

maximum_amplitude_relative_error: float = 0.05#

maximum_phase_error_deg: float = 2.0#

minimum_valid_fraction: float = 1.0#

require_convergence: bool = True#

to_dict()#

Return JSON-compatible acceptance limits.

Returns:: Versioned threshold state.
Return type:: dict

Examples

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

classmethod from_dict(data)#

Restore validated benchmark thresholds.

Parameters:: data (mapping) – State returned by to_dict().
Returns:: Restored limits.
Return type:: BenchmarkThresholds

Examples

>>> limits = BenchmarkThresholds(maximum_normalized_rms=0.1)
>>> BenchmarkThresholds.from_dict(limits.to_dict()) == limits
True

class pycsamt.forward.maxwell.BenchmarkMetrics(normalized_rms, maximum_amplitude_relative_error, maximum_phase_error_deg, valid_fraction, converged)#

Bases: object

Store errors measured against one analytic reference.

Parameters:

normalized_rms (float) – Complex normalized root-sum-square error.
maximum_amplitude_relative_error (float) – Worst pointwise relative amplitude error.
maximum_phase_error_deg (float) – Worst circular phase difference.
valid_fraction (float) – Fraction of output values marked valid and finite.
converged (bool) – Whether all backend solves converged.

Examples

>>> BenchmarkMetrics(0.01, 0.02, 0.5, 1, True).converged
True

normalized_rms: float#

maximum_amplitude_relative_error: float#

maximum_phase_error_deg: float#

valid_fraction: float#

converged: bool#

to_dict()#

Return JSON-compatible benchmark metrics.

Returns:: Error values, validity, and convergence status.
Return type:: dict

Examples

>>> BenchmarkMetrics(0, 0, 0, 1, True).to_dict()["valid_fraction"]
1.0

class pycsamt.forward.maxwell.BenchmarkOutcome(benchmark_name, benchmark_hash, backend_name, backend_version, passed, metrics, failures=())#

Bases: object

Record the auditable outcome of one backend benchmark.

Parameters:

benchmark_name (str) – Stable case identity and full content digest.
benchmark_hash (str) – Stable case identity and full content digest.
backend_name (str) – Exact adapter identity from the result.
backend_version (str) – Exact adapter identity from the result.
passed (bool) – Whether every configured acceptance criterion passed.
metrics (BenchmarkMetrics) – Quantitative comparison with the reference.
failures (tuple of str) – Human-readable failed criteria.

Examples

>>> metrics = BenchmarkMetrics(0, 0, 0, 1, True)
>>> outcome = BenchmarkOutcome(
...     "half-space", "0" * 64, "demo", "1", True, metrics
... )
>>> outcome.passed
True

benchmark_name: str#

benchmark_hash: str#

backend_name: str#

backend_version: str#

passed: bool#

metrics: BenchmarkMetrics#

failures: tuple[str, ...] = ()#

to_dict()#

Return a JSON-compatible benchmark outcome.

Returns:: Case identity, backend identity, metrics, and failures.
Return type:: dict

Examples

>>> metrics = BenchmarkMetrics(0, 0, 0, 1, True)
>>> value = BenchmarkOutcome(
...     "case", "0" * 64, "demo", "1", True, metrics
... )
>>> value.to_dict()["passed"]
True

class pycsamt.forward.maxwell.MaxwellBenchmark(name, description, problem, reference_impedance_v_a, thresholds=BenchmarkThresholds(maximum_normalized_rms=0.05, maximum_amplitude_relative_error=0.05, maximum_phase_error_deg=2.0, minimum_valid_fraction=1.0, require_convergence=True), tags=(), metadata=<factory>)#

Bases: object

Define one immutable problem and its expected impedance.

Parameters:

name (str) – Stable identifier and scientific purpose.
description (str) – Stable identifier and scientific purpose.
problem (MaxwellProblem) – Exact solver input.
reference_impedance_v_a (complex ndarray) – Analytic reference with canonical problem output shape.
thresholds (BenchmarkThresholds, optional) – Quantitative acceptance criteria.
tags (sequence of str, optional) – Searchable labels such as analytic and half-space.
metadata (mapping, optional) – Finite JSON-compatible provenance.

Examples

Cases are normally built with half_space_benchmark() or layered_earth_benchmark().

name: str#

description: str#

problem: MaxwellProblem#

reference_impedance_v_a: ndarray#

thresholds: BenchmarkThresholds = BenchmarkThresholds(maximum_normalized_rms=0.05, maximum_amplitude_relative_error=0.05, maximum_phase_error_deg=2.0, minimum_valid_fraction=1.0, require_convergence=True)#

tags: tuple[str, ...] = ()#

metadata: Mapping[str, Any]#

property benchmark_hash: str#

Return a deterministic digest of case inputs and thresholds.

Returns:: SHA-256 benchmark identity.
Return type:: str

Examples

Benchmark hashes contain 64 hexadecimal characters.

evaluate(result)#

Compare one canonical result with the analytic reference.

Parameters:: result (ForwardResult) – Backend result for this exact problem.
Returns:: Metrics and every failed acceptance criterion.
Return type:: BenchmarkOutcome
Raises:: ValueError – If the result belongs to a different problem or output axes.

Examples

Exact references produce zero error when wrapped as backend results.

run(backend)#

Execute and evaluate this case with a conforming backend.

Parameters:: backend (MaxwellBackend) – Backend compatible with the benchmark problem.
Returns:: Auditable validation outcome.
Return type:: BenchmarkOutcome

Examples

Backend exceptions propagate so infrastructure failures cannot be mistaken for numerical benchmark failures.

provenance()#

Return JSON-compatible benchmark provenance.

Returns:: Case identity, problem hash, limits, tags, and metadata.
Return type:: dict

Examples

The full conductivity model remains identified by problem_hash.

class pycsamt.forward.maxwell.BenchmarkReport(outcomes)#

Bases: object

Aggregate ordered outcomes from one backend benchmark run.

Parameters:: outcomes (sequence of BenchmarkOutcome) – Non-empty ordered outcomes from one backend version.

Examples

>>> metrics = BenchmarkMetrics(0, 0, 0, 1, True)
>>> outcome = BenchmarkOutcome(
...     "case", "0" * 64, "demo", "1", True, metrics
... )
>>> BenchmarkReport((outcome,)).passed
True

outcomes: tuple[BenchmarkOutcome, ...]#

property passed: bool#

Return whether every benchmark passed.

Returns:: Aggregate acceptance status.
Return type:: bool

Examples

A report fails if any contained outcome fails.

property pass_fraction: float#

Return the fraction of cases that passed.

Returns:: Passed case count divided by total cases.
Return type:: float

Examples

The value lies in the closed interval [0, 1].

to_dict()#

Return a JSON-compatible benchmark report.

Returns:: Backend identity, summary, and ordered outcomes.
Return type:: dict

Examples

Serialized reports retain every individual failure message.

pycsamt.forward.maxwell.half_space_impedance(resistivity_ohm_m, frequencies_hz, *, time_dependence='exp(+iwt)')#

Return analytic plane-wave impedance of a uniform half-space.

Parameters:

resistivity_ohm_m (float) – Positive half-space resistivity.
frequencies_hz (array-like) – Positive frequencies.
time_dependence ({"exp(+iwt)", "exp(-iwt)"}) – Complex phasor convention.

Returns:

sqrt(i*omega*mu0*rho) or its conjugate convention.

Return type:

ndarray of complex

Examples

>>> round(float(np.angle(half_space_impedance(100, 1), deg=True)))
45

pycsamt.forward.maxwell.layered_earth_impedance(resistivity_ohm_m, thickness_m, frequencies_hz, *, time_dependence='exp(+iwt)')#

Return analytic 1-D MT impedance by upward layer recursion.

Parameters:

resistivity_ohm_m (sequence of float) – Layer resistivities from surface to basal half-space.
thickness_m (sequence of float) – Thickness of every layer except the basal half-space.
frequencies_hz (array-like) – Positive evaluation frequencies.
time_dependence ({"exp(+iwt)", "exp(-iwt)"}) – Complex phasor convention.

Returns:

Surface impedance in frequency order.

Return type:

ndarray of complex

Examples

A single layer reduces exactly to the half-space expression:

>>> np.allclose(
...     layered_earth_impedance([100], [], [1, 10]),
...     half_space_impedance(100, [1, 10]),
... )
True

pycsamt.forward.maxwell.half_space_benchmark(mesh, receivers, frequencies_hz, *, resistivity_ohm_m=100.0, components=('zxy', 'zyx'), time_dependence='exp(+iwt)', thresholds=None)#

Build a uniform-earth analytic benchmark.

Parameters:

mesh (MaxwellMesh, ReceiverSet) – Solver geometry and observation locations.
receivers (MaxwellMesh, ReceiverSet) – Solver geometry and observation locations.
frequencies_hz (sequence of float) – Positive benchmark frequencies.
resistivity_ohm_m (float, default=100) – Uniform earth resistivity.
components (sequence of str, default=("zxy", "zyx")) – Requested components. Diagonal 3-D components are excluded because their analytic reference is zero and relative metrics are undefined.
time_dependence (str, default="exp(+iwt)") – Complex phasor convention.
thresholds (BenchmarkThresholds or None, optional) – Acceptance limits.

Returns:

Executable half-space case.

Return type:

MaxwellBenchmark

Examples

>>> mesh = MaxwellMesh([0, 1, 2], [0, 1, 2])
>>> receivers = ReceiverSet([[0.5, 0]], ["S"])
>>> half_space_benchmark(mesh, receivers, [1]).name
'half-space'

pycsamt.forward.maxwell.layered_earth_benchmark(mesh, receivers, frequencies_hz, resistivity_ohm_m, thickness_m, *, components=('zxy', 'zyx'), time_dependence='exp(+iwt)', thresholds=None)#

Build a laterally uniform layered-earth benchmark.

Parameters:

mesh (MaxwellMesh) – Solver geometry, observations, and frequencies.
receivers (ReceiverSet) – Solver geometry, observations, and frequencies.
frequencies_hz (Sequence[float]) – Solver geometry, observations, and frequencies.
resistivity_ohm_m (sequence of float) – Layer resistivities ending with a basal half-space.
thickness_m (sequence of float) – Finite-layer thicknesses. Every cumulative interface must coincide with a mesh z edge.
components (Sequence[str]) – Output components, phasor convention, and acceptance limits.
time_dependence (str) – Output components, phasor convention, and acceptance limits.
thresholds (BenchmarkThresholds | None) – Output components, phasor convention, and acceptance limits.

Returns:

Executable analytic layered-earth case.

Return type:

MaxwellBenchmark

Examples

>>> mesh = MaxwellMesh([0, 1, 2], [0, 1, 2])
>>> receivers = ReceiverSet([[0.5, 0]], ["S"])
>>> case = layered_earth_benchmark(mesh, receivers, [1], [10, 100], [1])
>>> case.name
'layered-earth'

pycsamt.forward.maxwell.run_benchmarks(backend, benchmarks)#

Run an ordered benchmark collection with one backend version.

Parameters:

backend (MaxwellBackend) – Backend under validation.
benchmarks (sequence of MaxwellBenchmark) – Non-empty cases with unique names.

Returns:

Aggregate and per-case outcomes.

Return type:

BenchmarkReport

Examples

Backend and numerical exceptions propagate rather than becoming false benchmark failures.

exception pycsamt.forward.maxwell.ExecutableNotFoundError#

Bases: MaxwellAdapterError

Indicate that a configured external solver executable is missing.

Examples

>>> isinstance(ExecutableNotFoundError("missing"), MaxwellAdapterError)
True

exception pycsamt.forward.maxwell.ExternalProcessError(message, attempts)#

Bases: BackendExecutionError

Indicate that every attempt to run an external solver process failed.

Parameters:

message (str) – Human-readable summary, normally including the last attempt’s exit code and a tail of its captured stderr.
attempts (tuple of ExternalRunResult) – Every attempt made, in order, including the failing ones.

Return type:

None

Examples

>>> error = ExternalProcessError("occam2d failed after 1 attempt", ())
>>> error.attempts
()

class pycsamt.forward.maxwell.ExternalRunPolicy(executable, search_paths=(), timeout_s=None, max_attempts=1, retry_backoff_s=1.0, workdir=None, keep_workdir_on_failure=True, extra_env=<factory>, capture_output=True)#

Bases: object

Configure how an external solver executable is located and run.

Parameters:

executable (str) – Executable name resolved via PATH and search_paths, or an absolute/relative path to the external solver binary.
search_paths (sequence of str, optional) – Additional directories checked, in order, after PATH and before giving up.
timeout_s (float or None, default=None) – Maximum wall-clock time allowed per attempt. None disables the per-attempt timeout.
max_attempts (int, default=1) – Total attempts per solve, including the first. Values above one retry a failed or timed-out run.
retry_backoff_s (float, default=1.0) – Base delay before each retry; attempt n (n > 1) waits retry_backoff_s * n seconds before relaunching.
workdir (str or pathlib.Path or None, optional) – Fixed working directory reused across solves and owned by the caller; it is created if missing and never deleted by this adapter. None creates and deletes a private temporary directory per solve.
keep_workdir_on_failure (bool, default=True) – Preserve a private temporary working directory (see workdir) when every attempt fails, so its contents can be inspected. Has no effect when workdir is caller-supplied, since that directory is always preserved.
extra_env (mapping of str to str, optional) – Extra environment variables merged over the current process environment for the subprocess only.
capture_output (bool, default=True) – Capture stdout/stderr for diagnostics instead of inheriting the parent process streams.

Examples

>>> policy = ExternalRunPolicy("occam2d", max_attempts=2, timeout_s=60.0)
>>> policy.max_attempts, policy.timeout_s
(2, 60.0)

executable: str#

search_paths: tuple[str, ...] = ()#

timeout_s: float | None = None#

max_attempts: int = 1#

retry_backoff_s: float = 1.0#

workdir: str | None = None#

keep_workdir_on_failure: bool = True#

extra_env: Mapping[str, str]#

capture_output: bool = True#

class pycsamt.forward.maxwell.ExternalRunResult(command, returncode, stdout, stderr, runtime_s, attempt, workdir)#

Bases: object

Record one external-process execution attempt.

Parameters:

command (sequence of str) – Exact argv executed.
returncode (int) – Process exit status; -1 denotes a timeout.
stdout (str) – Captured output; empty when capture_output was False or the process timed out before producing output.
stderr (str) – Captured output; empty when capture_output was False or the process timed out before producing output.
runtime_s (float) – Wall-clock duration of this attempt, in seconds.
attempt (int) – 1-based attempt number.
workdir (str or pathlib.Path) – Directory the process was launched from.

Examples

>>> result = ExternalRunResult(("occam2d",), 0, "done", "", 0.5, 1, ".")
>>> result.success
True

command: tuple[str, ...]#

returncode: int#

stdout: str#

stderr: str#

runtime_s: float#

attempt: int#

workdir: str#

property success: bool#

Return whether the process exited with status zero.

Returns:: True only for a normal, non-timed-out, zero exit status.
Return type:: bool

Examples

>>> ExternalRunResult(("a",), 1, "", "", 0.0, 1, ".").success
False

tail(*, stream='stderr', lines=20)#

Return the last lines of captured stdout or stderr.

Parameters:

stream ({"stderr", "stdout"}, default="stderr") – Which captured stream to summarize.
lines (int, default=20) – Maximum number of trailing lines returned.

Returns:

Newline-joined tail, or an empty string when nothing was captured.

Return type:

str

Raises:

ValueError – If stream is not "stderr" or "stdout".

Examples

>>> result = ExternalRunResult(
...     ("a",), 1, "", "line1\nline2\nline3", 0.0, 1, "."
... )
>>> result.tail(lines=2)
'line2\nline3'

to_dict()#

Return a JSON-compatible representation of this attempt.

Returns:: Command, exit status, captured output, timing, and attempt number. Useful for batch-run failure manifests.
Return type:: dict

Examples

>>> ExternalRunResult(("a",), 0, "ok", "", 1.0, 1, ".").to_dict()[
...     "success"
... ]
True

class pycsamt.forward.maxwell.BaseExternalMaxwellAdapter(capabilities, run_policy, policy=None)#

Bases: BaseMaxwellAdapter

Base class for adapters that run a trusted external solver process.

Concrete adapters wrap file-based external tools (for example ModEM, Occam2D, or MARE2DEM) launched as subprocesses rather than called as an in-process Python function. This class owns the shared, solver-independent mechanics: working-directory lifecycle, executable resolution, subprocess execution with timeout and retry, and captured diagnostics. A concrete subclass implements only three solver-specific extension points:

_prepare_inputs(problem, workdir): Write the external tool’s input files for problem into workdir and return a context object (any value) carried through to the other two extension points.
_build_command(problem, workdir, executable, context): Return the argv sequence that runs the external tool against the files written by _prepare_inputs.
_parse_result(problem, workdir, run_result, context): Read the external tool’s output files from workdir and return a canonical ForwardResult. run_result is the successful ExternalRunResult.

Parameters:

capabilities (BackendCapabilities) – Immutable declaration for the exact external solver version.
run_policy (ExternalRunPolicy) – Executable resolution, timeout, retry, and working-directory rules.
policy (AdapterPolicy or None, optional) – Solver-independent result acceptance policy (convergence, residual, and validity checks applied after _parse_result returns).

Examples

A minimal concrete adapter (see the module docstring for context) would look like:

class DemoExternalAdapter(BaseExternalMaxwellAdapter):
    def _prepare_inputs(self, problem, workdir):
        (workdir / "input.txt").write_text(str(problem.problem_hash))
        return None

    def _build_command(self, problem, workdir, executable, context):
        return [str(executable), "input.txt", "output.txt"]

    def _parse_result(
        self, problem, workdir, run_result, context
    ): ...  # read workdir / "output.txt" and build a ForwardResult

property run_policy: ExternalRunPolicy#

Return the immutable external-process execution policy.

Returns:: Executable resolution, timeout, retry, and working-directory rules used by every solve() call.
Return type:: ExternalRunPolicy

Examples

See BaseExternalMaxwellAdapter for a complete subclass example.

resolve_executable()#

Resolve this adapter’s configured executable to a concrete path.

Returns:: Resolved executable path.
Return type:: pathlib.Path
Raises:: ExecutableNotFoundError – If the executable cannot be found on PATH or in ExternalRunPolicy.search_paths.

Examples

See BaseExternalMaxwellAdapter for a complete subclass example.

pycsamt.forward.maxwell.resolve_executable(name_or_path, *, search_paths=())#

Resolve an external solver executable to a concrete file path.

Parameters:

name_or_path (str) – Executable name looked up on PATH, or an absolute/relative path checked directly.
search_paths (sequence of str, optional) – Extra directories checked, in order, after PATH and before giving up.

Returns:

Resolved, existing executable path.

Return type:

pathlib.Path

Raises:

ExecutableNotFoundError – If the executable cannot be found as a direct path, on PATH, or in any of search_paths.

Examples

>>> resolve_executable("nope")
Traceback (most recent call last):
...
pycsamt.forward.maxwell.external.ExecutableNotFoundError: ...

pycsamt.forward.maxwell.probe_executable_version(executable, *, version_args=('--version',), version_pattern=None, timeout_s=5.0)#

Best-effort external solver version string for provenance.

Parameters:

executable (str) – Resolved or resolvable executable path or name.
version_args (sequence of str, default=("--version",)) – Arguments appended to the executable when probing.
version_pattern (str, optional) – Regular expression whose first capture group extracts the version from the combined stdout/stderr. When omitted, the first non-empty output line is returned verbatim.
timeout_s (float, default=5.0) – Maximum time allowed for the probe process.

Returns:

Detected version string, or None when the probe process cannot be started, times out, or produces no matching output. This function never raises for an ordinary probe failure; a concrete adapter must still supply a non-empty backend_version to ForwardResult when this returns None.

Return type:

str or None

Examples

>>> probe_executable_version("does-not-exist-xyz") is None
True

pycsamt.forward.maxwell.make_availability_probe(name_or_path, *, search_paths=())#

Build a zero-argument availability probe for backend registration.

Parameters:

name_or_path (str) – Executable resolved the same way as resolve_executable().
search_paths (sequence of str, optional) – Extra directories checked before giving up.

Returns:

Zero-argument function returning (available, reason), directly usable as availability_probe.

Return type:

callable

Examples

>>> probe = make_availability_probe("does-not-exist-xyz")
>>> probe()[0]
False

class pycsamt.forward.maxwell.MT2DAdapter(*, version='1.0', policy=None, verbose=False)#

Bases: BaseMaxwellAdapter

Validated 2-D MT adapter over the in-repo finite-difference solver.

Parameters:

version (str, default="1.0") – Adapter version reported in every ForwardResult. Bump this when the translation logic in this module changes in a way that could alter numerical output.
policy (AdapterPolicy or None, optional) – Solver-independent result acceptance policy. Defaults to AdapterPolicy.
verbose (bool, default=False) – Forwarded to MT2DForward; prints per-frequency progress when true.

Examples

>>> import numpy as np
>>> from pycsamt.forward.maxwell import (
...     MaxwellMesh,
...     MaxwellProblem,
...     ReceiverSet,
... )
>>> mesh = MaxwellMesh(
...     np.linspace(0, 10_000, 41), np.linspace(0, 5_000, 31)
... )
>>> problem = MaxwellProblem(
...     mesh,
...     np.full(mesh.shape, 1.0 / 100.0),
...     [10.0, 1.0],
...     ReceiverSet([[5_000.0, 0.0]], ["S00"]),
...     ("zxy", "zyx"),
... )
>>> adapter = MT2DAdapter(verbose=False)
>>> result = adapter.solve(problem)
>>> result.shape
(1, 2, 2)

assess(problem)#

Assess a problem, adding this solver’s surface-only checks.

Parameters:: problem (MaxwellProblem) – Candidate simulation problem.
Returns:: The generic capability report from assess(), extended with two solver-specific checks: every receiver must be at the surface, and the permeability must be the vacuum value the wrapped solver hardcodes.
Return type:: CompatibilityReport

Examples

See MT2DAdapter for a complete solve example; a problem with a buried receiver is rejected before the solver runs.

pycsamt.forward.maxwell.register_mt2d_backend(*, replace=False)#

Register MT2DAdapter in the process-wide backend registry.

Parameters:: replace (bool, default=False) – Explicitly replace an existing "mt2d" registration.
Return type:: None

Examples

>>> from pycsamt.forward.maxwell import create_backend, list_backends
>>> register_mt2d_backend(replace=True)
>>> "mt2d" in list_backends()
True
>>> create_backend("mt2d").capabilities.name
'mt2d'

class pycsamt.forward.maxwell.MT3DAdapter(*, version='1.0-research', policy=None, max_cells=6000)#

Bases: BaseMaxwellAdapter

Research-only 3-D MT adapter (see module docstring for scope).

Parameters:

version (str, default="1.0-research") – Adapter version reported in every ForwardResult.
policy (AdapterPolicy or None, optional) – Solver-independent result acceptance policy.
max_cells (int, default=6000) – Safety ceiling on total mesh cells (maximum_cells). A direct sparse solve becomes impractically slow well before typical production 3-D mesh sizes; see docs/source/development/adr/AI-INVERSION-M6-3D-ADR.md. Raise this only if you have confirmed the resulting solve time is acceptable for your use.

Examples

>>> import numpy as np
>>> from pycsamt.forward.maxwell import (
...     MaxwellMesh,
...     MaxwellProblem,
...     ReceiverSet,
... )
>>> mesh = MaxwellMesh(
...     np.linspace(0, 4000, 9),
...     np.linspace(0, 3000, 11),
...     np.linspace(0, 4000, 9),
... )
>>> problem = MaxwellProblem(
...     mesh,
...     np.full(mesh.shape, 1.0 / 100.0),
...     [1.0],
...     ReceiverSet([[2000.0, 2000.0, 0.0]], ["S00"]),
...     ("zxy", "zyx"),
... )
>>> result = MT3DAdapter().solve(problem)
>>> result.shape
(1, 1, 2)

assess(problem)#

Assess a problem, adding this solver’s research-only checks.

Parameters:: problem (MaxwellProblem) – Candidate simulation problem.
Returns:: The generic capability report from assess(), extended with surface-receiver, horizontal-bounds, and vacuum-permeability checks.
Return type:: CompatibilityReport

Examples

See MT3DAdapter for a complete solve example.

class pycsamt.forward.maxwell.ModEm3DAdapter(*, config=None, run_policy=None, policy=None, predicted_data_filename=None, version='modem-v6.2.6-adapter-1.0')#

Bases: BaseExternalMaxwellAdapter

Adapter wrapping the external ModEM 3-D forward solver.

Parameters:

config (ModEmConfig or None, optional) – ModEM configuration (executable names, MPI settings, units, sign convention). Defaults to ModEmConfig(mode="3d").
run_policy (ExternalRunPolicy or None, optional) – Executable resolution, timeout, retry, and working-directory rules. When omitted, a policy is built from config (config.mpi_command if config.use_mpi else config.binary_3d), with the vendored pycsamt/models/modem/_source/3D directory as a search path fallback.
policy (AdapterPolicy or None, optional) – Solver-independent result acceptance policy.
predicted_data_filename (str or None, optional) – Exact name of ModEM’s predicted-response output file within the run’s working directory. When omitted, the adapter auto-detects it as the only *.dat file other than the one it wrote itself, and raises a clear error if that is ambiguous (zero or more than one candidate) — set this explicitly once you know your ModEM build’s naming convention.
version (str, default="modem-v6.2.6-adapter-1.0") – Adapter version reported in every ForwardResult. Encodes both the vendored ModEM release this adapter targets and this adapter code’s own revision.

Examples

>>> from pycsamt.forward.maxwell import ExternalRunPolicy
>>> adapter = ModEm3DAdapter(
...     run_policy=ExternalRunPolicy("does-not-exist-xyz")
... )
>>> adapter.capabilities.name
'modem3d'

assess(problem)#

Assess a problem, adding this adapter’s mapping checks.

Parameters:: problem (MaxwellProblem) – Candidate simulation problem.
Returns:: The generic capability report from assess(), extended with surface-receiver, horizontal-bounds, surface-aligned-mesh, and vacuum-permeability checks.
Return type:: CompatibilityReport

Examples

See ModEm3DAdapter for construction; a problem with a buried receiver is rejected before any file is written.

pycsamt.forward.maxwell.register_modem3d_backend(*, config=None, replace=False)#

Register ModEm3DAdapter in the process-wide registry.

Parameters:

config (ModEmConfig or None, optional) – Configuration used both to build the registered adapter and to derive its availability probe (whether the configured executable can currently be resolved).
replace (bool, default=False) – Explicitly replace an existing "modem3d" registration.

Return type:

None

Examples

>>> register_modem3d_backend(replace=True)
>>> from pycsamt.forward.maxwell import list_backends
>>> "modem3d" in list_backends()
True
>>> list_backends()["modem3d"]["available"]
False

class pycsamt.forward.maxwell.Mare2DEMAdapter(*, config=None, run_policy=None, policy=None, std_err=1.0, response_filename=None, version='mare2dem-adapter-1.0')#

Bases: BaseExternalMaxwellAdapter

Adapter wrapping the external MARE2DEM 2.5-D triangular-mesh solver.

Parameters:

config (Mare2DEMConfig or None, optional) – MARE2DEM configuration (executable name, MPI settings). Defaults to Mare2DEMConfig().
run_policy (ExternalRunPolicy or None, optional) – Executable resolution, timeout, retry, and working-directory rules. Built from config when omitted (config.mpi_command if config.use_mpi else config.binary).
policy (AdapterPolicy or None, optional) – Solver-independent result acceptance policy.
std_err (float, default=1.0) – Placeholder standard error written for every forward-request DATA row (MARE2DEM’s file format requires a nonzero value even though it is unused for a zero-iteration forward evaluation).
response_filename (str or None, optional) – Exact name of MARE2DEM’s predicted-response output file. When omitted, the adapter globs the working directory for a *.resp file, falling back to *_MARE2DEM.emdata, and raises a clear error if that is ambiguous.
version (str, default="mare2dem-adapter-1.0") – Adapter version reported in every ForwardResult.

Examples

>>> from pycsamt.forward.maxwell import ExternalRunPolicy
>>> adapter = Mare2DEMAdapter(
...     run_policy=ExternalRunPolicy("does-not-exist-xyz")
... )
>>> adapter.capabilities.name
'mare2dem'

assess(problem)#

Assess a problem, adding this adapter’s mesh/region checks.

Parameters:: problem (TriProblem) – Candidate simulation problem.
Returns:: The generic capability report from assess(), extended with a triangular-mesh-type check and a per-region uniform-conductivity check.
Return type:: CompatibilityReport

Examples

See Mare2DEMAdapter for construction; a problem whose conductivity varies within one mesh region is rejected before any file is written.

pycsamt.forward.maxwell.register_mare2dem_backend(*, config=None, replace=False)#

Register Mare2DEMAdapter in the process-wide registry.

Parameters:

config (Mare2DEMConfig or None, optional) – Configuration used both to build the registered adapter and to derive its availability probe (whether the configured executable can currently be resolved).
replace (bool, default=False) – Explicitly replace an existing "mare2dem" registration.

Return type:

None

Examples

>>> register_mare2dem_backend(replace=True)
>>> from pycsamt.forward.maxwell import list_backends
>>> "mare2dem" in list_backends()
True

class pycsamt.forward.maxwell.TriFEM2DAdapter(*, version='1.0', policy=None, node_tolerance_m=0.001, verbose=False)#

Bases: BaseMaxwellAdapter

In-house 2-D triangular-mesh MT FEM adapter (research-only).

Parameters:

version (str, default="1.0") – Adapter version reported in every ForwardResult.
policy (AdapterPolicy or None, optional) – Solver-independent result acceptance policy.
node_tolerance_m (float, default=1e-3) – Maximum distance between a receiver and its assumed mesh node.
verbose (bool, default=False) – Forwarded to Tri2DFEMForward.

Examples

>>> import numpy as np
>>> from pycsamt.forward.maxwell import ReceiverSet
>>> from pycsamt.forward.maxwell.contracts_tri import TriMesh, TriProblem
>>> nodes = [[-1000, 0], [1000, 0], [1000, 1000], [-1000, 1000]]
>>> mesh = TriMesh(
...     nodes, [[0, 1, 2], [0, 2, 3]],
...     boundary_segments=[[0, 1], [1, 2], [2, 3], [3, 0]],
... )
>>> problem = TriProblem(
...     mesh, np.full(2, 0.01), [1.0], ReceiverSet([[-1000, 0]], ["S00"])
... )
>>> adapter = TriFEM2DAdapter()
>>> adapter.capabilities.name
'trifem2d'

assess(problem)#

Assess a problem, adding this solver’s mesh/receiver checks.

Parameters:: problem (TriProblem) – Candidate simulation problem.
Returns:: The generic capability report from assess(), extended with boundary-segment, receiver-node-matching, surface, and permeability checks.
Return type:: CompatibilityReport

Examples

See TriFEM2DAdapter for construction; a receiver not coinciding with any mesh node is rejected before the solver runs.

pycsamt.forward.maxwell.register_trifem2d_backend(*, replace=False)#

Register TriFEM2DAdapter in the process-wide backend registry.

Parameters:: replace (bool, default=False) – Explicitly replace an existing "trifem2d" registration.
Return type:: None

Examples

>>> from pycsamt.forward.maxwell import create_backend, list_backends
>>> register_trifem2d_backend(replace=True)
>>> "trifem2d" in list_backends()
True
>>> create_backend("trifem2d").capabilities.name
'trifem2d'

pycsamt.forward.maxwell.build_graded_tri_mesh(x_range_m, z_range_m, station_x_m, *, surface_cell_m, growth_rate=1.3, max_cell_m=None, min_angle=30.0, topo_x_m=None, topo_z_m=None)#

Build a real, graded, quality FEM mesh via Shewchuk’s Triangle.

Parameters:

x_range_m ((float, float)) – Domain extent. z_range_m[0] should be 0 (surface) when topo_z_m is omitted; ignored (the top follows the topography instead) when it is given. z_range_m[1] (the domain bottom) must always be deeper than every topo_z_m sample.
z_range_m ((float, float)) – Domain extent. z_range_m[0] should be 0 (surface) when topo_z_m is omitted; ignored (the top follows the topography instead) when it is given. z_range_m[1] (the domain bottom) must always be deeper than every topo_z_m sample.
station_x_m (array-like) – Receiver x-positions; included as explicit PSLG vertices so every receiver sits exactly on a mesh node. Placed at z=0 unless topo_z_m is given, in which case each station’s z is interpolated from the topography polyline instead.
topo_x_m (array-like, optional) – Topography polyline (z positive down, so a ridge is negative) sampled at topo_x_m. When given, the mesh’s top boundary follows this polyline (merged with station_x_m, sorted and deduplicated by x, each station’s elevation interpolated from it) instead of a flat surface. Both must be given together, the same length, and finite. This only builds the mesh; using stations away from z=0 for a real solve also needs TriFEM2DAdapter’s local-surface-aware boundary conditions (already the default there – see that module’s own docstring).
topo_z_m (array-like, optional) – Topography polyline (z positive down, so a ridge is negative) sampled at topo_x_m. When given, the mesh’s top boundary follows this polyline (merged with station_x_m, sorted and deduplicated by x, each station’s elevation interpolated from it) instead of a flat surface. Both must be given together, the same length, and finite. This only builds the mesh; using stations away from z=0 for a real solve also needs TriFEM2DAdapter’s local-surface-aware boundary conditions (already the default there – see that module’s own docstring).
surface_cell_m (float) – Target triangle edge length at a station (the finest part of the mesh). Smaller values give a finer, more expensive mesh near the receivers. Choose this relative to skin depth for FEM solver accuracy, not just picture quality: a first layer around ~0.03-0.05 skin depths (at the highest simulated frequency) is what TriFEM2DAdapter’s own analytic half-space benchmark needs to stay under 5% relative impedance error (see that module’s “Station field extraction” docstring section for the same sensitivity on a structured mesh).
growth_rate (float, default=1.3) – Geometric growth factor of the target edge length per surface_cell_m of distance from the nearest station. Must be greater than 1 (grading away from the receivers, never toward them).
max_cell_m (float or None, optional) – Upper bound on the graded target edge length, reached far from every station / at depth. Defaults to 25 * surface_cell_m.
min_angle (float, default=30.0) – Minimum triangle angle in degrees, Triangle’s own quality constraint (same default as pycsamt.models.mare2dem.triangle_exec.run_triangle()).

Returns:

Graded mesh with one region per triangle (region_ids = 1..n_triangles, matching the old build_delaunay_mesh contract so Mare2DEMAdapter’s per-triangle-region convention still holds), ready for a per-triangle heterogeneous resistivity field.

Return type:

TriMesh

Raises:

ValueError – If the ranges are not increasing, a size parameter is non-positive, or station_x_m is empty or outside x_range_m.

Examples

>>> mesh = build_graded_tri_mesh(
...     (0.0, 1000.0), (0.0, 500.0), [200.0, 500.0, 800.0],
...     surface_cell_m=20.0,
... )
>>> mesh.n_triangles > 0
True
>>> sorted(mesh.region_ids.tolist()) == list(
...     range(1, mesh.n_triangles + 1)
... )
True

exception pycsamt.forward.maxwell.BatchAbortedError(report)#

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.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)#

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.BatchReport(total, solved, cache_hits, failed)#

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#

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()#

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

class pycsamt.forward.maxwell.FailureManifest(failures=())#

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]#

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()#

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)#

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)#

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)#

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.ProblemFailure(problem_hash, attempts, error_type, message)#

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()#

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)#

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'

pycsamt.forward.maxwell.solve_batch(problems, backend, *, cache=None, policy=None, on_result=None, on_failure=None)#

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

Maxwell Modules#

`pycsamt.forward.maxwell.contracts`	Solver-neutral contracts for frequency-domain Maxwell simulations.
`pycsamt.forward.maxwell.mesh`	Construct solver meshes from geological models and topography.
`pycsamt.forward.maxwell.adapters`	Validated execution layer for solver-specific Maxwell adapters.
`pycsamt.forward.maxwell.backends`	Capability checks and lazy registration for Maxwell solver backends.
`pycsamt.forward.maxwell.benchmarks`	Canonical analytic benchmarks for Maxwell backend validation.
`pycsamt.forward.maxwell.mt2d`	Validated adapter for the in-repo 2-D MT finite-difference solver.
`pycsamt.forward.maxwell.mt3d`	Research-only, small-grid 3-D MT finite-difference adapter.
`pycsamt.forward.maxwell.modem3d`	External ModEM adapter: the trusted production 3-D backend.
`pycsamt.forward.maxwell.external`	Adapter foundation for trusted external Maxwell solver executables.
`pycsamt.forward.maxwell.batch`	Robust, resumable batch solving with retries and failure manifests.
`pycsamt.forward.maxwell.cache`	Content-addressed cache for canonical Maxwell forward results.