pycsamt.forward.maxwell.external#

Adapter foundation for trusted external Maxwell solver executables.

Some verified EM forward/inversion codes worth wrapping (for example ModEM, Occam2D, or MARE2DEM; see pycsamt.models.modem, pycsamt.models.occam2d, pycsamt.models.mare2dem) are external executables driven by input files and read back from output files, not in-process Python callables. CallableMaxwellAdapter in pycsamt.forward.maxwell.adapters cannot wrap that shape of solver directly. This module provides the shared, solver-independent mechanics for that integration:

resolving a configured executable from PATH or declared search directories (resolve_executable());
a reusable, zero-argument availability probe for backend registration (make_availability_probe());
a best-effort external solver version probe for provenance (probe_executable_version());
BaseExternalMaxwellAdapter, which owns working-directory lifecycle, subprocess execution with timeout and retry, and diagnostic capture, so a concrete adapter implements only three solver-specific steps: writing input files, building the command line, and parsing output files back into a canonical ForwardResult.

No external solver is imported or executed by importing this module.

Functions

`make_availability_probe`(name_or_path, *[, ...])	Build a zero-argument availability probe for backend registration.
`probe_executable_version`(executable, *[, ...])	Best-effort external solver version string for provenance.
`resolve_executable`(name_or_path, *[, ...])	Resolve an external solver executable to a concrete file path.

Classes

`BaseExternalMaxwellAdapter`(capabilities, ...)	Base class for adapters that run a trusted external solver process.
`ExternalRunPolicy`(executable[, ...])	Configure how an external solver executable is located and run.
`ExternalRunResult`(command, returncode, ...)	Record one external-process execution attempt.

Exceptions

`ExecutableNotFoundError`	Indicate that a configured external solver executable is missing.
`ExternalProcessError`(message, attempts)	Indicate that every attempt to run an external solver process failed.

exception pycsamt.forward.maxwell.external.ExecutableNotFoundError[source]

Bases: MaxwellAdapterError

Indicate that a configured external solver executable is missing.

Examples

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

exception pycsamt.forward.maxwell.external.ExternalProcessError(message, attempts)[source]

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.external.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)[source]

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.external.ExternalRunResult(command, returncode, stdout, stderr, runtime_s, attempt, workdir)[source]

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[source]

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)[source]

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()[source]

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.external.BaseExternalMaxwellAdapter(capabilities, run_policy, policy=None)[source]

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[source]

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()[source]

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.external.resolve_executable(name_or_path, *, search_paths=())[source]

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.external.probe_executable_version(executable, *, version_args=('--version',), version_pattern=None, timeout_s=5.0)[source]

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.external.make_availability_probe(name_or_path, *, search_paths=())[source]

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