PyHydroGeophysX.workflows package

Submodules

PyHydroGeophysX.workflows.builtin module

Initial built-in workflow handlers.

Workflow IDs are registry identifiers. They intentionally do not imply that a same-named function exists in the scientific module.

PyHydroGeophysX.workflows.builtin.run_gravmag_forward_bodies(spec: WorkflowSpec, context: RunContext) → WorkflowRunResult

PyHydroGeophysX.workflows.builtin.run_gravmag_process(spec: WorkflowSpec, context: RunContext) → WorkflowRunResult

PyHydroGeophysX.workflows.builtin.run_srt_inversion(spec: WorkflowSpec, context: RunContext) → WorkflowRunResult

PyHydroGeophysX.workflows.bundle module

Export a validated recipe together with the code that reproduces it.

PyHydroGeophysX.workflows.bundle.export_workflow_bundle(spec: WorkflowSpec, directory: str | Path, *, stem: str = 'workflow', teaching: bool = True) → Tuple[Path, Path]

Write the recipe and the runner, plus a walkthrough script and notebook.

Four files land in directory:

<stem>_recipe.json

The run configuration, the single source both generators read from.

run_<stem>.py

A runner that calls the workflow engine, for a byte-identical rerun.

<stem>_walkthrough.py and <stem>_walkthrough.ipynb

The same run written as domain-level calls with named parameters and prose, for reading, teaching and editing.

Returns (recipe_path, script_path). The walkthrough files are written as a side effect and a failure to produce them never blocks the reproducible pair: a workflow with no walkthrough definition still exports a valid recipe and runner.

Validation occurs before any writer mutates its destination, so a non-serializable or seedless stochastic spec cannot produce a runnable- looking partial export.

PyHydroGeophysX.workflows.bundle.teaching_paths(script_path: str | Path) → Tuple[Path | None, Path | None]

Locate the walkthrough pair export_workflow_bundle() wrote.

Takes the runner path because that is what a caller already holds, and returns (walkthrough_py, walkthrough_ipynb) with None for whichever is absent. The naming convention lives in this module, so resolving by convention here is a lookup rather than a guess.

PyHydroGeophysX.workflows.cli module

Command-line interface for versioned workflows.

PyHydroGeophysX.workflows.cli.main(argv: Sequence[str] | None = None) → int

PyHydroGeophysX.workflows.codegen module

Deterministic Python generation from a workflow recipe.

PyHydroGeophysX.workflows.codegen.generate_python(spec: WorkflowSpec, path: str | Path) → Path

Generate a standalone, editable script using only the public workflow API.

PyHydroGeophysX.workflows.domain module

Adapters from serializable workflow contracts to canonical domain APIs.

PyHydroGeophysX.workflows.domain.run_em_inversion(spec: WorkflowSpec, context: RunContext) → WorkflowRunResult

PyHydroGeophysX.workflows.domain.run_ert3d_forward(spec: WorkflowSpec, context: RunContext) → WorkflowRunResult

PyHydroGeophysX.workflows.domain.run_ert_single(spec: WorkflowSpec, context: RunContext) → WorkflowRunResult

PyHydroGeophysX.workflows.domain.run_ert_timelapse(spec: WorkflowSpec, context: RunContext) → WorkflowRunResult

PyHydroGeophysX.workflows.domain.run_geo_hydrology(spec: WorkflowSpec, context: RunContext) → WorkflowRunResult

PyHydroGeophysX.workflows.domain.run_gravmag_inversion(spec: WorkflowSpec, context: RunContext) → WorkflowRunResult

PyHydroGeophysX.workflows.domain.run_hydro_geophysics(spec: WorkflowSpec, context: RunContext) → WorkflowRunResult

PyHydroGeophysX.workflows.domain.run_joint(spec: WorkflowSpec, context: RunContext) → WorkflowRunResult

PyHydroGeophysX.workflows.domain.run_mesh3d(spec: WorkflowSpec, context: RunContext) → WorkflowRunResult

PyHydroGeophysX.workflows.domain.run_seismic3d(spec: WorkflowSpec, context: RunContext) → WorkflowRunResult

PyHydroGeophysX.workflows.em1d module

High-level 1D electromagnetic workflows and compatibility facade.

exception PyHydroGeophysX.workflows.em1d.BackendUnavailable

Bases: RuntimeError

Raised when an optional numerical backend cannot be used.

PyHydroGeophysX.workflows.em1d.backend_status(method: str | None = None) → Dict[str, Any]

Report whether the requested EM forward/inversion backend is usable.

The check imports the same method-specific forward class used by inversion. This prevents the UI and AQUAH from announcing a background inversion that cannot start because SimPEG or one of its runtime dependencies is missing.

PyHydroGeophysX.workflows.em1d.build_em_config(method: str, model: Dict[str, Any], geom: Dict[str, Any], inv: Dict[str, Any]) → Dict[str, Any]

PyHydroGeophysX.workflows.em1d.calibrate_to_reference(path: str, method: str, geom: ~typing.Dict[str, ~typing.Any], inv: ~typing.Dict[str, ~typing.Any], ref_resistivity: float, *, max_probe: int = 6, log: ~typing.Callable[[str], None] = <function noop>) → float

Find the data_scale that makes the recovered near-surface resistivity match a known/expected value.

The amplitude scale and the absolute resistivity level are degenerate — the EM data alone cannot fix the level (any data_scale fits the data, with the resistivity shifting to compensate). This breaks the degeneracy with EXTERNAL information: the user supplies ref_resistivity (a known background, e.g. from a borehole or regional geology), and this inverts a few probe soundings at two trial scales, fits the near-surface resistivity’s log-linear response to the scale, and solves for the scale that yields ref_resistivity. Returns the current data_scale unchanged if calibration is not possible.

PyHydroGeophysX.workflows.em1d.estimate_data_scale(path: str, method: str, geom: ~typing.Dict[str, ~typing.Any], *, max_soundings: int = 8, log: ~typing.Callable[[str], None] = <function noop>) → float

Estimate the amplitude calibration (data_scale) for normalized data.

Normalized airborne responses (e.g. moment-normalized dB/dt) differ from the studio’s 1D forward by a near-constant amplitude factor. This fits each sounding’s decay SHAPE to a grid of half-space forward responses at the current geometry and takes the geometric-mean amplitude ratio forward/observed at the best-fitting resistivity. Returns 1.0 if it cannot be estimated (so the caller can fall back to no scaling).

PyHydroGeophysX.workflows.em1d.example_catalog() → Dict[str, Dict[str, Any]]

Return the desktop EM examples and their documented settings.

Prefer the source checkout for development and fall back to the compact package-data copy installed from a wheel.

PyHydroGeophysX.workflows.em1d.fdem_forward(model: ~typing.Dict[str, ~typing.Any], geom: ~typing.Dict[str, ~typing.Any], log: ~typing.Callable[[str], None] = <function noop>) → Dict[str, Any]

1D FDEM forward response (secondary field, real/imag per frequency).

PyHydroGeophysX.workflows.em1d.fdem_invert(data: ~typing.Dict[str, ~typing.Any], geom: ~typing.Dict[str, ~typing.Any], inv: ~typing.Dict[str, ~typing.Any], log: ~typing.Callable[[str], None] = <function noop>) → Dict[str, Any]

Invert an FDEM sounding for a layered resistivity model (Occam 1D).

PyHydroGeophysX.workflows.em1d.invert_line(path: str, method: str, geom: ~typing.Dict[str, ~typing.Any], inv: ~typing.Dict[str, ~typing.Any], *, spacing: float = 50.0, positions: ~numpy.ndarray | None = None, heights: ~numpy.ndarray | None = None, max_soundings: int = 12, lines: ~typing.Sequence[int] | None = None, doi_blank: bool = True, doi_factor: float = 0.5, ref_resistivity: float = 0.0, out_dir: ~pathlib.Path | None = None, initial_models: ~numpy.ndarray | None = None, log: ~typing.Callable[[str], None] = <function noop>) → Dict[str, Any]

Invert a line on a shared fixed-layer grid.

inv["lci_mode"] selects how the soundings are coupled:

simultaneous (the default whenever lateral_smoothness is positive)

Solves the whole line as one system, with the lateral constraint part of what is being minimized. See PyHydroGeophysX.inversion.em1d_lci.

sequential

The older block-coordinate passes: each station is re-inverted on its own against the distance-weighted model its neighbours had at the end of the previous pass. Kept because it needs no analytic Jacobian, so it still runs against a forward operator that cannot supply one.

off

Independent 1D inversion per sounding, no lateral coupling.

The models are laid side by side to form a resistivity(position, depth) section ready for Model3DView.show_model(): edges = (ex, ey, ez) (ez is elevation, increasing upward) and model3d of shape (n_pos, 1, n_depth).

positions gives the along-line distance of each sounding (the section x-axis); heights overrides the sensor height per sounding. When doi_blank is set, cells below a per-sounding depth of investigation (a diffusion-depth estimate scaled by doi_factor) are blanked (NaN) so the unconstrained deep part of an early-time sounding is not shown as railed.

inv["robust_errors"] retains all imported gates and iteratively inflates effective errors for large residuals. It overrides hard rejection. The main chi2 uses ORIGINAL errors; result["robust"] records effective errors and a separate effective chi2. Import-time flags and QC still apply.

inv["auto_lambda"] re-solves the line at other smoothness weights to reach target_chi2. inv["reject_outliers"] drops the gates the converged model cannot explain (beyond outlier_threshold sigma, over outlier_passes cycles, never below min_data_fraction of the gates) and solves again; what it removed is reported under result["outliers"]. They address different causes, so they can be used together: relaxing the smoothness helps when the model is too stiff for the data, rejection helps when a minority of gates are simply wrong.

lines restricts the run to the named survey lines, so a line whose data is thinner than the rest can be given its own settings instead of one set having to suit every line. Passing None runs from the first station, as before. The lateral constraint already groups by line, so a line inverted on its own is tied exactly as it would be inside a whole-survey run; what changes is which settings reach it, and that the other lines are not re-solved. max_soundings then counts within the selection.

Stations arrive ordered by line, so a selection is a contiguous block. A set of lines that is not contiguous is refused rather than quietly widened to the span that encloses it, which would invert the lines in between under settings chosen for their neighbours.

PyHydroGeophysX.workflows.em1d.is_temcompany_source(path: str) → bool

Return whether path looks like a TEMcompany/TEM2Go export.

Both complete project directories and the self-describing *.xyz exports written by TEMImage are accepted.

PyHydroGeophysX.workflows.em1d.is_ttem_source(path: str) → bool

Return whether path contains a TEMcompany tTEM raw acquisition.

PyHydroGeophysX.workflows.em1d.load_line_geometry(path: str) → Dict[str, Any]

Load per-sounding line geometry: along-line positions (m), optional sensor heights (m), and optional map coordinates x/y (e.g. easting/northing) for plan-view depth slices. Recognizes header names (distance/position for the position; alt/height for the height; easting/northing for the map coordinates, which also derive the distance when no distance column is present). A header-less file is read by column order (1 column = position; 2+ = position, height). positions is shifted to start at 0.

PyHydroGeophysX.workflows.em1d.load_sounding(path: str, method: str, sounding: int = 0, *, moment: str = 'HM', use_flags: bool = True, max_relative_std: float | None = None, gate_rejection: str = 'truncate', reject_negative: bool = False, min_gates_per_moment: Mapping[str, int] | None = None, ttem_loop_area: float | None = None, ttem_gex_path: str | None = None, ttem_tfi_path: str | None = None) → Dict[str, Any]

Load one sounding from a sounding file.

The first column is the abscissa (FDEM: frequency Hz; TDEM: time s). The remaining columns hold the response(s) — a single sounding, or several stacked side by side so one file can carry a whole survey line (common for airborne EM exports). sounding picks which one (0-based):

• TDEM: each extra column is one sounding’s response → column 1 + sounding.

• FDEM: response columns come in (real, imag) pairs → one sounding is the pair starting at 1 + 2*sounding; a lone trailing real column gives imag = 0.

The returned dict also reports n_soundings so the caller can offer a picker. use_flags applies only to TEMcompany project databases; see load_temcompany_sounding().

PyHydroGeophysX.workflows.em1d.load_sounding_container(path: str | Path, method: str, sounding: int = 0) → Dict[str, Any]

Return one sounding from a container written by save_sounding_container().

The moment and flag settings are not arguments here: they were applied when the container was written, and the stored arrays are what they produced.

PyHydroGeophysX.workflows.em1d.load_temcompany_sounding(path: str, sounding: int = 0, moment: str = 'HM', *, use_flags: bool = True, max_relative_std: float | None = None, gate_rejection: str = 'truncate', reject_negative: bool = False, min_gates_per_moment: Mapping[str, int] | None = None) → Dict[str, Any]

Load a TEMcompany/TEM2Go sounding from a project folder or XYZ export.

The defaults reproduce the gate selection the project itself records, which was measured rather than assumed. Over 1,503 station-moment datasets of one project, the gates the stored inversion used (InversionModel.Datasets) are exactly the gates whose InUseFlags are set and whose value is finite and not a dummy, with 100 percent agreement. There is no further sign test: 87 low- moment and 251 high-moment datasets keep a non-positive gate. There is no further error cut either, and none is needed, because the largest relative error among the kept gates is exactly 0.250, so TEMImage applied that cut upstream when it wrote the flags. The selection is not even contiguous, so nor truncation: only 36 percent of the high-moment selections are a single run of gates.

So max_relative_std=None and reject_negative=False are the defaults. Both arguments remain, because a survey whose flags were written by an older an older acquisition release, or one being deliberately treated more strictly, still needs them. Note that the sign test only runs when max_relative_std is set: it condemns a gate alongside a noisy one rather than on its own.

use_flags=False ignores the project’s in-use flags and returns every gate with a finite, non-dummy value. Only a project database records those flags, so it makes no difference to an XYZ export.

min_gates_per_moment applies only to a joint LM+HM read of a project database, which is the only path where dropping one moment still leaves a sounding; see _load_temcompany_joint_database().

PyHydroGeophysX.workflows.em1d.load_ttem_sounding(path: str, sounding: int = 0, moment: str = 'LM+HM', *, max_relative_std: float | None = None, stack_seconds: float = 2.0, loop_area: float | None = None, gex_path: str | None = None, tfi_path: str | None = None) → Dict[str, Any]

Load and stack one sounding from a raw TEMcompany tTEM survey.

Data are normalized by measured transmitter current and the loop area from the selected GEX (or an explicit UI override). A selected TFI is applied to each sign-corrected transient sequence before stacking.

PyHydroGeophysX.workflows.em1d.model_arrays(model: Dict[str, Any]) → tuple[ndarray, ndarray, ndarray]

Return (thicknesses, resistivity, conductivity) from a model dict.

PyHydroGeophysX.workflows.em1d.model_depth_profile(thicknesses: ndarray, resistivity: ndarray, pad: float = 20.0) → tuple[ndarray, ndarray]

Step profile (depth, resistivity) for plotting a layered model.

PyHydroGeophysX.workflows.em1d.save_inversion(result: Dict[str, Any], out_dir: Path) → List[str]

Save recovered model + data fit to npy/csv; return written paths.

PyHydroGeophysX.workflows.em1d.save_line_csv(result: Dict[str, Any], out_dir: Path) → List[str]

Write the section as two flat tables; return the paths written.

model_cells.csv is one row per layer per sounding, which is the form a GIS or a gridding package wants: every row carries its own map coordinate and its own elevation, so the section can be reconstructed without knowing anything about the layer grid. soundings.csv is the per-station summary that would otherwise have to be recovered by grouping the first table.

Depths are below each station’s own ground level, and z is the elevation of the cell centre where the survey carries ground elevations. A cell below the depth of investigation is written out with its resistivity and flagged rather than dropped: what the inversion produced there is still the answer to a question the data cannot settle, and a reader filtering on the flag can decide for themselves.

PyHydroGeophysX.workflows.em1d.save_sounding_container(destination: str | Path, path: str, method: str, *, moment: str = 'HM', use_flags: bool = True, max_relative_std: float | None = None, gate_rejection: str = 'truncate', reject_negative: bool = False, min_gates_per_moment: Mapping[str, int] | None = None, ttem_loop_area: float | None = None, ttem_gex_path: str | None = None, ttem_tfi_path: str | None = None, progress=None) → Path

Materialize every sounding in path into one compressed container.

A recorded run used to keep its input by copying the acquisition folder, which for a TEMcompany project is hundreds of megabytes per inversion. The soundings themselves are a few hundred kilobytes, and they are the only part the inversion reads, so this parses the survey once and stores the result.

The load settings are baked in, because they decide what the arrays contain: a container written for LM+HM holds joint moments, and re-reading it under another moment would silently return the wrong gates. They travel in the manifest so a reader can report what it was given.

PyHydroGeophysX.workflows.em1d.tdem_forward(model: ~typing.Dict[str, ~typing.Any], geom: ~typing.Dict[str, ~typing.Any], log: ~typing.Callable[[str], None] = <function noop>) → Dict[str, Any]

1D TDEM forward response (dB/dt or H per time channel).

PyHydroGeophysX.workflows.em1d.tdem_invert(data: ~typing.Dict[str, ~typing.Any], geom: ~typing.Dict[str, ~typing.Any], inv: ~typing.Dict[str, ~typing.Any], log: ~typing.Callable[[str], None] = <function noop>) → Dict[str, Any]

Invert a TDEM sounding for a layered resistivity model (Occam 1D).

PyHydroGeophysX.workflows.em1d.tdem_joint_invert(data: ~typing.Dict[str, ~typing.Any], geom: ~typing.Dict[str, ~typing.Any], inv: ~typing.Dict[str, ~typing.Any], log: ~typing.Callable[[str], None] = <function noop>) → Dict[str, Any]

Invert all available LM/HM gates at one station for one shared 1D model.

PyHydroGeophysX.workflows.gravmag module

Gravity/magnetics workflow facade and result export.

exception PyHydroGeophysX.workflows.gravmag.InversionBackendUnavailable

Bases: BackendUnavailable

SimPEG / discretize / a usable solver could not be imported.

PyHydroGeophysX.workflows.gravmag.backend_status() → Dict[str, Any]

Report whether the SimPEG potential-field inversion stack is available.

PyHydroGeophysX.workflows.gravmag.build_gravmag_config(kind: str, settings: Dict[str, Any], bodies: List[Dict[str, Any]], field: Dict[str, Any] | None = None) → Dict[str, Any]

PyHydroGeophysX.workflows.gravmag.extract_profile(grid: Dict[str, ndarray], p1: Sequence[float], p2: Sequence[float], n: int = 200) → Dict[str, ndarray]

Sample a gridded field along the line p1 -> p2 (bilinear).

PyHydroGeophysX.workflows.gravmag.forward_bodies(xobs: ~numpy.ndarray, yobs: ~numpy.ndarray, kind: str, bodies: ~typing.List[~typing.Dict[str, ~typing.Any]], field: ~typing.Dict[str, ~typing.Any] | None = None, log: ~typing.Callable[[str], None] = <function noop>) → ndarray

Sum the anomaly of a list of bodies. kind = ‘gravity’ or ‘magnetics’.

PyHydroGeophysX.workflows.gravmag.gravity_prism(xobs: ndarray, yobs: ndarray, body: Dict[str, Any]) → ndarray

Vertical gravity (mGal) of a right rectangular prism (Nagy 1966). z down.

PyHydroGeophysX.workflows.gravmag.gravity_sphere(xobs: ndarray, yobs: ndarray, body: Dict[str, Any]) → ndarray

Vertical gravity (mGal) of a buried sphere. z positive down, obs at z=0.

PyHydroGeophysX.workflows.gravmag.grid_data(x: ndarray, y: ndarray, value: ndarray, nx: int = 120, ny: int = 120, method: str = 'linear') → Dict[str, ndarray]

Grid scattered station values onto a regular map. Returns xx, yy, zz.

PyHydroGeophysX.workflows.gravmag.invert_gravmag(x, y, value, kind: str, *, z: ~numpy.ndarray | None = None, field: ~typing.Dict[str, ~typing.Any] | None = None, detrend: int = 0, n_xy: int = 22, n_z: int = 12, max_iterations: int = 20, beta0_ratio: float = 1.0, max_stations: int = 600, relative_error: float = 0.03, noise_floor: float | None = None, solver: str = 'simpeg', auto_beta: bool = True, target_chi2: float = 1.0, chi2_tolerance: float = 0.2, max_beta_trials: int = 6, sensitivity_power: float = 1.0, out_dir: str | None = None, random_seed: int | None = 42, log: ~typing.Callable[[str], None] = <function noop>) → Dict[str, Any]

Run a SimPEG 3D potential-field inversion under the survey.

gravity recovers a density-contrast model (g/cc); magnetics recovers a susceptibility model (SI) and needs field = {inclination, declination, strength_nT}. z is optional per-station elevation (m, positive upward); a missing value falls back to 1 m. detrend (0..3) removes a polynomial regional trend before inversion. The returned grid uses elevation increasing upward. random_seed makes SimPEG’s eigenvalue-based beta estimate reproducible. Raises InversionBackendUnavailable if SimPEG is missing.

PyHydroGeophysX.workflows.gravmag.magnetic_dipole(xobs: ndarray, yobs: ndarray, body: Dict[str, Any], field: Dict[str, Any]) → ndarray

Total-field magnetic anomaly (nT) of an induced/magnetized sphere (a dipole).

PyHydroGeophysX.workflows.gravmag.qc_products(x: ndarray, y: ndarray, value: ndarray, *, detrend: int = 1, nx: int = 120, ny: int = 120) → Dict[str, Any]

Calculate observed, regional and residual products for map/profile QC.

PyHydroGeophysX.workflows.gravmag.regional_residual(x: ndarray, y: ndarray, value: ndarray, degree: int = 1) → Tuple[ndarray, ndarray]

Fit a polynomial regional trend of degree (1..3); return (regional, residual).

PyHydroGeophysX.workflows.gravmag.save_grid(grid: ~typing.Dict[str, ~numpy.ndarray], out_dir: ~pathlib.Path, name: str = 'anomaly', log: ~typing.Callable[[str], None] = <function noop>) → List[str]

Save a grid to npy + CSV + VTK (best-effort). Return written paths.

PyHydroGeophysX.workflows.gravmag.spatially_balanced_indices(x: ndarray, y: ndarray, max_stations: int) → ndarray

Return deterministic farthest-point indices for a spatially balanced subset.

The previous evenly spaced file-row selection could over-sample a survey segment when input rows were ordered by flight line or acquisition order. Farthest-point selection starts near the survey centroid and repeatedly adds the station furthest from the selected set, preserving map coverage without a random seed.

PyHydroGeophysX.workflows.models module

Serializable workflow contracts shared by the GUI, CLI, and Python API.

class PyHydroGeophysX.workflows.models.ArtifactRef(artifact_id: str, kind: str, path: str, format: str, checksum: str = '', metadata: Mapping[str, ~typing.Any]=<factory>)

Bases: object

A serializable reference to a workflow input or output artifact.

artifact_id: str

checksum: str = ''

format: str

classmethod from_dict(value: Mapping[str, Any]) → ArtifactRef

classmethod from_path(path: Path, *, artifact_id: str, kind: str, format: str | None = None, base_dir: Path | None = None, metadata: Mapping[str, Any] | None = None, checksum: bool = True) → ArtifactRef

kind: str

metadata: Mapping[str, Any]

path: str

resolve(base_dir: Path) → Path

to_dict() → Dict[str, Any]

class PyHydroGeophysX.workflows.models.RunContext(project_root: ~pathlib.Path = <factory>, output_dir: ~pathlib.Path = <factory>, progress: ~typing.Callable[[str], None] = <function _noop_progress>, cancelled: ~typing.Callable[[], bool] = <function _never_cancelled>, object_cache: ~typing.MutableMapping[str, ~typing.Any] = <factory>)

Bases: object

Process-local execution settings that are deliberately absent from recipes.

static cache_key(ref: ArtifactRef) → str

cancelled() → bool

load_object(ref: ArtifactRef, loader: Callable[[Path], Any]) → Any

Load and cache an artifact by artifact_id + checksum.

object_cache: MutableMapping[str, Any]

output_dir: Path

prepare() → None

progress() → None

project_root: Path

resolve_artifact(ref: ArtifactRef) → Path

class PyHydroGeophysX.workflows.models.WorkflowRunResult(status: str, summary: Mapping[str, ~typing.Any]=<factory>, metrics: Mapping[str, ~typing.Any]=<factory>, artifacts: List[ArtifactRef] = <factory>, warnings: List[str] = <factory>, provenance: Mapping[str, ~typing.Any]=<factory>, objects: MutableMapping[str, ~typing.Any]=<factory>)

Bases: object

Public serializable result plus an optional same-process object channel.

artifacts: List[ArtifactRef]

classmethod from_dict(value: Mapping[str, Any]) → WorkflowRunResult

legacy_payload() → Dict[str, Any]

Return a same-process mapping for existing viewers during migration.

Unlike to_dict(), this deliberately merges the live objects channel back into the summary. It must never be persisted or emitted by CLI/bridge code.

metrics: Mapping[str, Any]

objects: MutableMapping[str, Any]

provenance: Mapping[str, Any]

status: str

summary: Mapping[str, Any]

to_dict() → Dict[str, Any]

Return the process-safe view. objects is intentionally excluded.

warnings: List[str]

class PyHydroGeophysX.workflows.models.WorkflowSpec(workflow_id: str, inputs: Mapping[str, ~typing.Any]=<factory>, parameters: Mapping[str, ~typing.Any]=<factory>, seed: int | None = None, dependencies: List[str] = <factory>, metadata: Mapping[str, ~typing.Any]=<factory>, schema_version: str = '1')

Bases: object

Everything required to reproduce one workflow in another process.

dependencies: List[str]

classmethod from_dict(value: Mapping[str, Any]) → WorkflowSpec

inputs: Mapping[str, Any]

metadata: Mapping[str, Any]

parameters: Mapping[str, Any]

schema_version: str = '1'

seed: int | None = None

to_dict() → Dict[str, Any]

validate(*, stochastic: bool = False) → None

workflow_id: str

exception PyHydroGeophysX.workflows.models.WorkflowValidationError

Bases: ValueError

A recipe cannot cross a process boundary safely.

PyHydroGeophysX.workflows.models.file_checksum(path: Path, algorithm: str = 'sha256') → str

Return a stable checksum for an artifact on disk.

PyHydroGeophysX.workflows.models.iter_artifact_refs(value: Any) → Iterable[ArtifactRef]

Yield every artifact nested in a workflow input/parameter structure.

PyHydroGeophysX.workflows.recipe module

Versioned JSON recipe IO.

PyHydroGeophysX.workflows.recipe.load_recipe(path: str | Path) → WorkflowSpec

PyHydroGeophysX.workflows.recipe.save_recipe(spec: WorkflowSpec, path: str | Path) → Path

Validate and save a workflow recipe as UTF-8 JSON.

PyHydroGeophysX.workflows.registry module

Workflow and desktop-module registries.

class PyHydroGeophysX.workflows.registry.ModuleDescriptor(navigation_key: 'str', result_key: 'str', workflow_ids: 'Tuple[str, ...]', aliases: 'Tuple[str, ...]' = ())

Bases: object

aliases: Tuple[str, ...] = ()

navigation_key: str

result_key: str

workflow_ids: Tuple[str, ...]

class PyHydroGeophysX.workflows.registry.WorkflowDescriptor(workflow_id: 'str', handler_path: 'str', description: 'str', stochastic: 'bool' = False, module_key: 'str' = '')

Bases: object

description: str

handler_path: str

load_handler() → Callable[[WorkflowSpec, RunContext], WorkflowRunResult]

module_key: str = ''

stochastic: bool = False

workflow_id: str

PyHydroGeophysX.workflows.registry.get_workflow(workflow_id: str) → WorkflowDescriptor

PyHydroGeophysX.workflows.registry.list_workflows() → Tuple[WorkflowDescriptor, ...]

PyHydroGeophysX.workflows.registry.module_descriptor_for(key: str) → ModuleDescriptor | None

Find the descriptor a key belongs to, whichever spelling it uses.

A module is addressed by two names on purpose: the navigation key (ert) is what the CLI flag, the agent’s navigate tool, and the Streamlit bridge pass around, while the result key (ert_processing) is what a page publishes its results under and what the stored run records carry. Anything that receives one and needs the other goes through here, so the pairing lives in this table rather than in each caller’s head.

PyHydroGeophysX.workflows.registry.navigation_key_for(key: str) → str

Return the key that navigates to a module; unknown keys pass through.

PyHydroGeophysX.workflows.registry.register_workflow(descriptor: WorkflowDescriptor, *, replace: bool = False) → None

PyHydroGeophysX.workflows.registry.result_key_for(key: str) → str

Return the key a module publishes results under; unknown keys pass through.

PyHydroGeophysX.workflows.runner module

Public workflow execution entry point.

PyHydroGeophysX.workflows.runner.run_workflow(spec: WorkflowSpec | Mapping[str, Any], context: RunContext | None = None) → WorkflowRunResult

Validate and execute one registered workflow.

PyHydroGeophysX.workflows.walkthrough module

Readable, step-by-step Python and notebooks from a workflow recipe.

PyHydroGeophysX.workflows.generate_python() emits a runner: one run_workflow(spec) call plus a nested-dict blob. That reproduces a run exactly, which is what CI and batch reruns need, but it teaches nothing and its parameters cannot really be edited.

This module emits the other half: the same run written the way a person would write it, with named parameters, the actual domain calls, printed intermediates and a figure. Both files come from one recipe, so a walkthrough always describes the run it was generated from, and tests/test_walkthroughs.py::test_gravmag_walkthrough_matches_the_engine pins the two to identical artifacts.

Each Walkthrough is the single prose description of what a workflow does. Parameters are rendered into named module-level constants, so step code is plain text that never needs escaping: it refers to those constants by name.

class PyHydroGeophysX.workflows.walkthrough.Step(title: str, code: str, note: str = '')

Bases: object

One numbered section of a walkthrough.

note carries the reasoning a reader needs: what the step is for, how to tell whether it worked, and which parameter to reach for when it did not. code is emitted verbatim and refers to the constants that the parameter block defines.

code: str

note: str = ''

title: str

class PyHydroGeophysX.workflows.walkthrough.Walkthrough(summary: 'str', imports: 'Sequence[str]', steps: 'Sequence[Step]', parameters: 'ParameterRenderer' = <function Walkthrough.<lambda> at 0x7f72a8195940>, inputs: 'InputRenderer | None' = None, reading: 'Sequence[str]' = <factory>)

Bases: object

imports: Sequence[str]

inputs: Callable[[WorkflowSpec], Sequence[str]] | None = None

parameters()

reading: Sequence[str]

steps: Sequence[Step]

summary: str

PyHydroGeophysX.workflows.walkthrough.default_inputs(spec: WorkflowSpec) → List[str]

Load array bundles as named variables and other artifacts as paths.

PyHydroGeophysX.workflows.walkthrough.generate_notebook(spec: WorkflowSpec, path: str | Path) → Path

Write the same transcript as a Jupyter notebook.

Prose becomes markdown cells and each step becomes one code cell, so a reader can run the steps one at a time and inspect what each produced.

PyHydroGeophysX.workflows.walkthrough.generate_walkthrough(spec: WorkflowSpec, path: str | Path) → Path

Write a readable, runnable transcript of spec as a .py script.

Module contents

Qt-free workflow, recipe, CLI, and code-generation API.

class PyHydroGeophysX.workflows.ArtifactRef(artifact_id: str, kind: str, path: str, format: str, checksum: str = '', metadata: Mapping[str, ~typing.Any]=<factory>)

Bases: object

A serializable reference to a workflow input or output artifact.

artifact_id: str

checksum: str = ''

format: str

classmethod from_dict(value: Mapping[str, Any]) → ArtifactRef

classmethod from_path(path: Path, *, artifact_id: str, kind: str, format: str | None = None, base_dir: Path | None = None, metadata: Mapping[str, Any] | None = None, checksum: bool = True) → ArtifactRef

kind: str

metadata: Mapping[str, Any]

path: str

resolve(base_dir: Path) → Path

to_dict() → Dict[str, Any]

class PyHydroGeophysX.workflows.ModuleDescriptor(navigation_key: 'str', result_key: 'str', workflow_ids: 'Tuple[str, ...]', aliases: 'Tuple[str, ...]' = ())

Bases: object

aliases: Tuple[str, ...] = ()

navigation_key: str

result_key: str

workflow_ids: Tuple[str, ...]

class PyHydroGeophysX.workflows.RunContext(project_root: ~pathlib.Path = <factory>, output_dir: ~pathlib.Path = <factory>, progress: ~typing.Callable[[str], None] = <function _noop_progress>, cancelled: ~typing.Callable[[], bool] = <function _never_cancelled>, object_cache: ~typing.MutableMapping[str, ~typing.Any] = <factory>)

Bases: object

Process-local execution settings that are deliberately absent from recipes.

static cache_key(ref: ArtifactRef) → str

cancelled() → bool

load_object(ref: ArtifactRef, loader: Callable[[Path], Any]) → Any

Load and cache an artifact by artifact_id + checksum.

object_cache: MutableMapping[str, Any]

output_dir: Path

prepare() → None

progress() → None

project_root: Path

resolve_artifact(ref: ArtifactRef) → Path

class PyHydroGeophysX.workflows.WorkflowDescriptor(workflow_id: 'str', handler_path: 'str', description: 'str', stochastic: 'bool' = False, module_key: 'str' = '')

Bases: object

description: str

handler_path: str

load_handler() → Callable[[WorkflowSpec, RunContext], WorkflowRunResult]

module_key: str = ''

stochastic: bool = False

workflow_id: str

class PyHydroGeophysX.workflows.WorkflowRunResult(status: str, summary: Mapping[str, ~typing.Any]=<factory>, metrics: Mapping[str, ~typing.Any]=<factory>, artifacts: List[ArtifactRef] = <factory>, warnings: List[str] = <factory>, provenance: Mapping[str, ~typing.Any]=<factory>, objects: MutableMapping[str, ~typing.Any]=<factory>)

Bases: object

Public serializable result plus an optional same-process object channel.

artifacts: List[ArtifactRef]

classmethod from_dict(value: Mapping[str, Any]) → WorkflowRunResult

legacy_payload() → Dict[str, Any]

Return a same-process mapping for existing viewers during migration.

Unlike to_dict(), this deliberately merges the live objects channel back into the summary. It must never be persisted or emitted by CLI/bridge code.

metrics: Mapping[str, Any]

objects: MutableMapping[str, Any]

provenance: Mapping[str, Any]

status: str

summary: Mapping[str, Any]

to_dict() → Dict[str, Any]

Return the process-safe view. objects is intentionally excluded.

warnings: List[str]

class PyHydroGeophysX.workflows.WorkflowSpec(workflow_id: str, inputs: Mapping[str, ~typing.Any]=<factory>, parameters: Mapping[str, ~typing.Any]=<factory>, seed: int | None = None, dependencies: List[str] = <factory>, metadata: Mapping[str, ~typing.Any]=<factory>, schema_version: str = '1')

Bases: object

Everything required to reproduce one workflow in another process.

dependencies: List[str]

classmethod from_dict(value: Mapping[str, Any]) → WorkflowSpec

inputs: Mapping[str, Any]

metadata: Mapping[str, Any]

parameters: Mapping[str, Any]

schema_version: str = '1'

seed: int | None = None

to_dict() → Dict[str, Any]

validate(*, stochastic: bool = False) → None

workflow_id: str

exception PyHydroGeophysX.workflows.WorkflowValidationError

Bases: ValueError

A recipe cannot cross a process boundary safely.

PyHydroGeophysX.workflows.export_workflow_bundle(spec: WorkflowSpec, directory: str | Path, *, stem: str = 'workflow', teaching: bool = True) → Tuple[Path, Path]

Write the recipe and the runner, plus a walkthrough script and notebook.

Four files land in directory:

<stem>_recipe.json

The run configuration, the single source both generators read from.

run_<stem>.py

A runner that calls the workflow engine, for a byte-identical rerun.

<stem>_walkthrough.py and <stem>_walkthrough.ipynb

The same run written as domain-level calls with named parameters and prose, for reading, teaching and editing.

Returns (recipe_path, script_path). The walkthrough files are written as a side effect and a failure to produce them never blocks the reproducible pair: a workflow with no walkthrough definition still exports a valid recipe and runner.

Validation occurs before any writer mutates its destination, so a non-serializable or seedless stochastic spec cannot produce a runnable- looking partial export.

PyHydroGeophysX.workflows.generate_notebook(spec: WorkflowSpec, path: str | Path) → Path

Write the same transcript as a Jupyter notebook.

Prose becomes markdown cells and each step becomes one code cell, so a reader can run the steps one at a time and inspect what each produced.

PyHydroGeophysX.workflows.generate_python(spec: WorkflowSpec, path: str | Path) → Path

Generate a standalone, editable script using only the public workflow API.

PyHydroGeophysX.workflows.generate_walkthrough(spec: WorkflowSpec, path: str | Path) → Path

Write a readable, runnable transcript of spec as a .py script.

PyHydroGeophysX.workflows.get_workflow(workflow_id: str) → WorkflowDescriptor

PyHydroGeophysX.workflows.iter_artifact_refs(value: Any) → Iterable[ArtifactRef]

Yield every artifact nested in a workflow input/parameter structure.

PyHydroGeophysX.workflows.list_workflows() → Tuple[WorkflowDescriptor, ...]

PyHydroGeophysX.workflows.load_recipe(path: str | Path) → WorkflowSpec

PyHydroGeophysX.workflows.module_descriptor_for(key: str) → ModuleDescriptor | None

Find the descriptor a key belongs to, whichever spelling it uses.

A module is addressed by two names on purpose: the navigation key (ert) is what the CLI flag, the agent’s navigate tool, and the Streamlit bridge pass around, while the result key (ert_processing) is what a page publishes its results under and what the stored run records carry. Anything that receives one and needs the other goes through here, so the pairing lives in this table rather than in each caller’s head.

PyHydroGeophysX.workflows.navigation_key_for(key: str) → str

Return the key that navigates to a module; unknown keys pass through.

PyHydroGeophysX.workflows.register_workflow(descriptor: WorkflowDescriptor, *, replace: bool = False) → None

PyHydroGeophysX.workflows.result_key_for(key: str) → str

Return the key a module publishes results under; unknown keys pass through.

PyHydroGeophysX.workflows.run_workflow(spec: WorkflowSpec | Mapping[str, Any], context: RunContext | None = None) → WorkflowRunResult

Validate and execute one registered workflow.

PyHydroGeophysX.workflows.save_recipe(spec: WorkflowSpec, path: str | Path) → Path

Validate and save a workflow recipe as UTF-8 JSON.

PyHydroGeophysX.workflows.teaching_paths(script_path: str | Path) → Tuple[Path | None, Path | None]

Locate the walkthrough pair export_workflow_bundle() wrote.

Takes the runner path because that is what a caller already holds, and returns (walkthrough_py, walkthrough_ipynb) with None for whichever is absent. The naming convention lives in this module, so resolving by convention here is a lookup rather than a guess.