PyHydroGeophysX.workflows package

Submodules

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.EXIT_WITH_PARENT_ENV = 'PHGX_EXIT_WITH_PARENT'

Set to 0 to let a run with --result-file outlive the program that started it.

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.

One handler per registered workflow: it turns a recipe’s inputs into the objects and paths a domain function takes, calls it, and files what comes back as a WorkflowRunResult. Workflow IDs are registry identifiers; they do not imply that a same-named function exists in the scientific module.

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

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

A line of soundings inverted together (invert_line()).

inputs.data is the stored soundings; inputs.positions and inputs.heights, when the survey has them, the along-line distance and the sensor height of each, as arrays in an .npz. The parameters are the method, geometry, inversion and the rest of invert_line’s keywords.

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_forward_bodies(spec: WorkflowSpec, context: RunContext) → WorkflowRunResult

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

PyHydroGeophysX.workflows.domain.run_gravmag_process(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_mt_invert_1d(spec: WorkflowSpec, context: RunContext) → WorkflowRunResult

Occam 1D of one site, optionally with static shift, a TEM sounding and water content.

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

2D inversion of a profile’s TE and TM impedances on SimPEG.

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

Time series to impedance and tipper: process_mt on a site and its remote.

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

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

PyHydroGeophysX.workflows.em1d module

The 1D electromagnetic API in one namespace.

Nothing is implemented here. The readers, the example catalogue and the result writers are in PyHydroGeophysX.data_processing.em1d, the forward models in PyHydroGeophysX.forward.em1d, single-sounding inversion in PyHydroGeophysX.inversion.em1d, and line inversion with its amplitude calibration in PyHydroGeophysX.inversion.em1d_line. The examples, the generated walkthroughs and older scripts import them from here.

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 the scale returned is the one that ties the observed amplitudes of a few probe soundings to the forward response of a half-space at that resistivity - the geometric mean of forward/observed over their gates. That is one forward model and no inversion, so it is deterministic; the layered model recovered afterwards is not forced to ref_resistivity exactly, but its absolute level is pinned the same way for every dataset. 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.

The example soundings ship with the source repository (examples/data/EM), not with the pip package. They are looked for in the checkout this package runs from, then in the clone that PHGX_EXAMPLES_DIR names (the clone or its examples folder), then under the working folder and its parents. When none has them the paths still point at the checkout layout, so a caller can report the file as missing.

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, real/imag per frequency.

real/imag hold the field named by geom["component"]. For "both" they hold the secondary field and total_real/total_imag (with total_amplitude/total_phase_deg) the total field, all on the same frequencies.

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.gate_report(path: str, sounding: int = 0, *, moment: str = 'LM+HM', use_flags: bool = True, max_relative_std: float | None = None, gate_rejection: str = 'truncate', reject_negative: bool = False) → Dict[str, Any]

Every gate one station holds, and what the current selection does with it.

load_sounding() returns the gates that survive, which is what an inversion needs and less than a reader needs. Working out why a station came back with four gates out of twenty-five otherwise means changing a setting and watching the count move. This returns the whole list instead, each gate carrying its verdict from _gate_disposition(), so the same question is answered by looking.

sounding indexes the stations the loader would return under these same settings, so a station number here and one there mean the same station. A station the settings empty is absent from both.

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.

In simultaneous LCI, a positive inv["reference_resistivity"] sets the damping reference independently of auto_starting_model and initial_models, for both TDEM and FDEM.

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.

Complete project directories, under either the project.tiw layout that TEMImage 3 writes or the project.db one earlier releases wrote, and the self-describing *.xyz exports are all accepted. So is an acquisition folder holding the instrument’s own .stb raw stream, which needs no import to be read.

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.

The container is read whole once per file version and the soundings served from memory, because a line inversion asks for every one of them in turn.

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 an equivalent cut is already reflected in the stored 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.preset_inversion(name: str = 'generic') → Dict[str, Any]

Inversion settings for a named survey type, over the framework defaults.

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, soundings: List[Dict[str, Any]] | None = 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.

soundings are the survey already read under exactly these settings (read_survey()); the file is then not read again.

PyHydroGeophysX.workflows.em1d.survey_summary(path: str, *, moment: str = 'LM+HM', use_flags: bool = True, max_relative_std: float | None = None, gate_rejection: str = 'truncate', reject_negative: bool = False, reference_gate: int = 2) → Dict[str, Any]

One row per station: where it is, and how much of it survives the QC.

A line inversion reads every station anyway, so the question “what will this quality-control setting actually cost me” has an answer that costs one pass over the file rather than a run. That is what this returns: per station, the gates the file holds, the gates the given settings keep, and the stack error of what is left, alongside the coordinates and per-station geometry.

rows is a list of dicts, one per station, in file order. totals carries the survey-wide counts so a caller can report the cost of a setting without summing the table itself.

Each row also carries {moment}_signal and {moment}_noise, the stacked voltage and the absolute noise in volts at one gate common to every station, along with the {moment}_reference_time that gate sits at. They answer a question the gate counts cannot: whether a station returns fewer gates because the ground is resistive, which makes the response genuinely smaller, or because something raised the noise floor. The stored relative error is the second divided by the first, so it moves either way and cannot tell them apart; the two recorded separately can.

reference_gate is the index of that gate within the moment’s own table, counted from the earliest. The default of 2 is an early gate, which is where the shallow ground shows, while being late enough to sit clear of the turn-off. The gate table is fixed per project, so one index is one physical time at every station, which is what makes the column comparable along a line. A station whose value there is a dummy records NaN rather than a substitute.

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 and magnetics in one namespace.

Nothing is implemented here: preprocessing, gridding, profiles and the grid writer are in PyHydroGeophysX.data_processing.gravmag, the analytic bodies in PyHydroGeophysX.forward.gravmag, and the SimPEG inversion in PyHydroGeophysX.inversion.gravmag. The examples, the generated walkthroughs and older scripts import them from here.

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.

Observation x/y coordinates follow NumPy broadcasting, including a scalar coordinate shared by every station. The result has their broadcast shape.

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 | None = 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.

max_stations thins a larger survey to a spatially balanced subset; None or 0 inverts every station whose sensitivities fit in memory (sensitivity_station_limit()), and thins a survey larger than that. Stations with a missing coordinate or value are left out. Either reduction is named in the result’s warnings (and n_loaded, n_missing, n_input, n_data count it), since a model fitted to part of a survey looks no different from one fitted to all.

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.mt module

Magnetotelluric workflows: the public names of PyHydroGeophysX.data_processing.mt.

The registered workflows are mt.process (time series to transfer functions), mt.invert_1d (Occam 1D, with static shift and an optional TEM sounding) and mt.invert_profile (2D on SimPEG); this module gathers what their recipes and walkthroughs call.

class PyHydroGeophysX.workflows.mt.ProcessingConfig(window_length: int = 128, overlap: int = 32, decimation_factor: int = 4, n_levels: int | None = None, min_windows: int = 5, bands: List[Band] | None = None, bands_per_decade: int | None = None, taper: str = 'hamming', prewhiten: bool = True, huber: float = 1.5, redescend: float = 2.8, max_iterations: int = 10, redescend_iterations: int = 2, tolerance: float = 0.005, leverage_cutoff: float | None = None, min_points: int = 10, allow_uncalibrated: bool = False, tipper: bool = True, periodic_noise: Tuple[float, ...] = (), periodic_cycles: int = 31, periodic_where_found: bool = True, skip_notched: bool = True, widen_above: float = 0.0, powerline: float | None = None, drop_saturated: bool = True, bounded_influence: bool = False, jackknife: bool = False, min_coherence: float = 0.0, presort_windows: int = 10, min_reference_coherence: float = 0.0, tapers: int = 1, multitaper_above: float = 0.0, multitaper_tapers: int = 3, multitaper_periods: Tuple[float, float] | None = None, multi_window: bool = False, noise_classes: int = 0)

Bases: object

Every choice of process_mt(); the defaults are EMTF’s.

allow_uncalibrated: bool = False

bands: List[Band] | None = None

bands_per_decade: int | None = None

Overlapping bands, this many to a decade (spectra.log_bands()), in place of EMTF’s band set: continuous curves, errors correlated between neighbours.

bounded_influence: bool = False

Weight points by their leverage too, every iteration (robust): noise in the magnetic channels then cannot pull the estimate.

decimation_factor: int = 4

drop_saturated: bool = True

Leave out the windows holding a sample at the recorder’s full scale (timeseries.saturated()), where the reader knows it.

huber: float = 1.5

jackknife: bool = False

Errors from the delete-a-group jackknife (20 groups) instead of the parametric model (robust).

leverage_cutoff: float | None = None

max_iterations: int = 10

min_coherence: float = 0.0

in each band, stretches of presort_windows consecutive windows whose multiple coherence (squared) of the output with Hx, Hy is below this are left out of that output’s regression.

Type:

Coherence presorting

min_points: int = 10

min_reference_coherence: float = 0.0

where a band’s Hx or Hy is coherent with the remote’s same channel below this (squared coherence), the reference holds almost no natural signal; the elements estimated from that input (Hx: Zxx, Zyx; Hy: Zxy, Zyy; and the tipper’s) are then a ratio of cross-spectra near zero, and the jackknife understates their error. Their relative errors are raised to at least 100 %. 0 is off.

Type:

Remote reference

min_windows: int = 5

multi_window: bool = False

each band also takes the coefficients at its frequencies of the windows one level up or down - windows decimation_factor times longer or shorter - in one regression. Those overlap the band’s own data in time, so the errors are the jackknife’s.

Type:

Multi-window stacking (Ernst et al. 2001)

multitaper_above: float = 0.0

An output (Ex for Zxy, Ey for Zyx, Hz) whose relative error in a band is above this with one taper is estimated again there with multitaper_tapers Slepian tapers, and that estimate kept. Where the field is weakest the tapers’ extra points cut the scatter; where the errors are a few per cent, the tapers’ smoothing over neighbouring harmonics (a bias of a few per cent where the curve is steep, which the errors do not show) would be the larger error, so those bands keep one taper. Where the two estimates differ, which is right is not known: the difference is added to the errors (in quadrature), so the curve is the steadier one and its errors hold the other. 0 is off.

multitaper_periods: Tuple[float, float] | None = None

The periods (s, lowest and highest) whose bands multitaper_above may estimate again; None is every band. Tapers help where few windows are the limit, as in a weak dead band; in short bursts at high rates they can make the curve rougher.

multitaper_tapers: int = 3

n_levels: int | None = None

noise_classes: int = 0

up to this many classes of noise that grows with the signal, each normalised before the M-estimate; 0 is off.

Type:

Noise classes (robust)

overlap: int = 32

periodic_cycles: int = 31

fewer follow interference whose shape drifts, each notch then a little wider.

Type:

Cycles each interference waveform is averaged over

periodic_noise: Tuple[float, ...] = ()

Periods (s) of interference repeating exactly on the receiver’s clock, subtracted from every run before its spectra (timeseries.remove_periodic()).

periodic_where_found: bool = True

Remove a period from a run only where timeseries.periodic_interference() finds it in that run (when the run holds 30 cycles or more, enough to tell). The removal also takes the field at every harmonic of the period, and where a level’s windows cannot resolve those notches it biases the band: so one site’s interference is not taken out of another site’s data. Bursts too short to fold are cleaned for every period, as before.

powerline: float | None = None

Grid frequency (50 or 60 Hz) whose hum and harmonics are notched out of every run before its spectra (timeseries.remove_powerline()); None is off.

presort_windows: int = 10

prewhiten: bool = True

redescend: float = 2.8

redescend_iterations: int = 2

regression() → RegressionConfig

skip_notched: bool = True

Leave out of the bands the harmonics inside those notches, where windows are long enough to resolve them (twice the period or more): the notch takes the field there too, and what is left is the interference’s drift.

spectra() → SpectraConfig

taper: str = 'hamming'

tapers: int = 1

Slepian tapers per window (multitaper, spectra); 1 is taper alone.

tipper: bool = True

tolerance: float = 0.005

widen_above: float = 0.0

A band whose Zxy and Zyx relative errors average above this is estimated again from a band about twice as wide (Q 1.5 instead of 3, same level and centre) and the wider one kept if its errors are smaller: more data where the field is weakest, at the price of resolution. 0 is off.

window_length: int = 128

class PyHydroGeophysX.workflows.mt.TimeSeriesRun(channels: ~typing.Dict[str, ~PyHydroGeophysX.data_processing.mt.timeseries.Channel], sample_rate: float, start: ~typing.Any, station: str = '', latitude: float = nan, longitude: float = nan, elevation: float = nan, declination: float = 0.0, instrument: str = '', serial: str = '', metadata: ~typing.Dict[str, ~typing.Any] = <factory>)

Bases: object

Simultaneous, gap-free samples of several channels at one rate.

array(components: Sequence[str] | None = None, dtype: Any = <class 'float'>) → ndarray

The chosen channels stacked as (n_channels, n_samples).

calibrated(components: Sequence[str] | None = None, *, band: Tuple[float, float] | None = None) → TimeSeriesRun

The channels in field units (mV/km, nT), the response removed by FFT.

Each channel is demeaned and divided by its response at every Fourier frequency; band (Hz) zeroes the spectrum outside it, which a coil’s vanishing low-frequency response needs. This holds the whole series in memory as complex numbers: for long, fast recordings select a window first.

channels: Dict[str, Channel]

property components: List[str]

declination: float = 0.0

property duration: float

Seconds the samples span, n / sample_rate.

elevation: float = nan

property end: datetime64

The time just after the last sample.

instrument: str = ''

latitude: float = nan

longitude: float = nan

metadata: Dict[str, Any]

property n_samples: int

sample_rate: float

select(components: Iterable[str]) → TimeSeriesRun

serial: str = ''

slice(first: int, last: int | None = None) → TimeSeriesRun

Samples first:last (views, not copies).

start: Any

station: str = ''

summary() → str

times() → ndarray

Each sample’s UTC time as datetime64[ns].

window(start: Any = None, end: Any = None) → TimeSeriesRun

The samples between two UTC times (either may be None).

class PyHydroGeophysX.workflows.mt.TransferFunction(frequency: ndarray, z: ndarray | None = None, z_err: ndarray | None = None, tipper: ndarray | None = None, tipper_err: ndarray | None = None, rotation: Any = 0.0, inverse_signal_power: ndarray | None = None, z_residual_covariance: ndarray | None = None, tipper_residual_covariance: ndarray | None = None, station: str = '', latitude: float = nan, longitude: float = nan, elevation: float = nan, x: float = nan, y: float = nan, declination: float = nan, metadata: Dict[str, ~typing.Any]=<factory>)

Bases: object

Impedance and tipper of one MT site, against frequency.

z is (n, 2, 2) complex ohms, tipper (n, 1, 2) complex and dimensionless; either may be None. z_err and tipper_err are the standard errors of each real and imaginary part. inverse_signal_power (n, 2, 2) and the residual covariances - z_residual_covariance (n, 2, 2) for (Ex, Ey) and tipper_residual_covariance (n, 1, 1) for Hz - are the full error model where a file or the estimator supplies it: var(Z_ij) = residual_ii S_jj. Values that are missing are NaN.

property angular_frequency: ndarray

apparent_resistivity() → ndarray

|Z|^2 / (omega mu0) in ohm-m, (n, 2, 2).

apparent_resistivity_err() → ndarray

First-order standard error of the apparent resistivity, ohm-m.

conjugated() → TransferFunction

The site under the opposite time convention (e^{-i omega t} <-> e^{+i omega t}).

copy() → TransferFunction

declination: float = nan

determinant() → ndarray

The rotation-invariant determinant impedance sqrt(Zxx Zyy - Zxy Zyx), (n,).

The branch is the one with the phase of Zxy, so a half-space gives +45.

elevation: float = nan

frequency: ndarray

property has_impedance: bool

property has_tipper: bool

inverse_signal_power: ndarray | None = None

latitude: float = nan

longitude: float = nan

metadata: Dict[str, Any]

property n_frequencies: int

property period: ndarray

phase() → ndarray

Impedance phase in degrees, (n, 2, 2); Zyx sits near -135 for a half-space.

phase_err() → ndarray

First-order standard error of the phase, degrees.

rotated(angle: Any) → TransferFunction

The site with its x axis at azimuth angle (degrees from north, clockwise).

angle is one value or one per frequency. A missing element (Zxx and Zyy of a site given as Zxy and Zyx only) leaves missing only what it enters: a rotation by 0 or 90 degrees keeps the rest.

The errors are recomputed from the rotated covariance when the site’s errors are the covariance’s (EMTF XML, Z-files); otherwise - errors of their own, such as the jackknife’s - each rotated variance is sum (R_ik R_jl)^2 var_kl from the errors in the frame the site was read in. That assumes the elements independent, an approximation that does not depend on the path the rotations took and leaves the errors as they were at the frame they came in.

rotation: Any = 0.0

select(mask: Any) → TransferFunction

The frequencies a boolean mask or an index array selects.

sorted_by_period() → TransferFunction

Frequencies in order of increasing period.

station: str = ''

summary() → Dict[str, Any]

A JSON-safe description of the site, for logs and workflow results.

tipper: ndarray | None = None

tipper_err: ndarray | None = None

tipper_residual_covariance: ndarray | None = None

x: float = nan

Local coordinates (m), when the site belongs to a projected survey.

y: float = nan

z: ndarray | None = None

z_err: ndarray | None = None

property z_field_units: ndarray | None

The impedance in (mV/km)/nT.

z_residual_covariance: ndarray | None = None

PyHydroGeophysX.workflows.mt.apply_static_shift(tf: TransferFunction, factor_x: float = 1.0, factor_y: float | None = None) → TransferFunction

Divide the apparent resistivities of the Ex row by factor_x and Ey’s by factor_y.

factor_y defaults to factor_x. The phases are unchanged; errors and covariances scale with the rows.

PyHydroGeophysX.workflows.mt.estimate_static_shift(tf: TransferFunction, reference_rho_a: Any, *, period_range: Tuple[float, float] | None = None) → Dict[str, float]

Static-shift factors of the Ex and Ey rows against a reference apparent resistivity.

reference_rho_a is an array at tf.frequency or a callable of frequency. The factor is the geometric mean of rho_a / reference over period_range (s), where the reference holds - at the short periods a TEM sounding sees. Returns {"x", "y", "x_spread", "y_spread", "n"}.

PyHydroGeophysX.workflows.mt.load_runs(source: Any) → List[TimeSeriesRun]

The runs a save_runs() container holds.

PyHydroGeophysX.workflows.mt.niblett_bostick(tf: TransferFunction, *, mode: str = 'det') → Dict[str, ndarray]

depth (m) and resistivity (ohm m) by the Niblett-Bostick transform.

mode is "det" (the determinant impedance), "xy" or "yx". rho_NB = rho_a (pi / (2 phi) - 1), depth = sqrt(rho_a / (omega mu0)).

PyHydroGeophysX.workflows.mt.occam1d(data: Any, *, mode: str = 'det', n_layers: int = 50, thickness: Sequence[float] | None = None, target_rms: float = 1.0, max_iterations: int = 30, starting_resistivity: float | None = None, static_shift: bool = False, static_shift_prior: float = 0.3, error_floor: float = 0.05, period_range: Tuple[float, float] | None = None, tem: Dict[str, Any] | None = None, tem_weight: float = 1.0, log: Callable[[str], None] | None = None, resistivity_bounds: Tuple[float, float] = (0.01, 100000.0)) → Occam1DResult

Occam’s smoothest model fitting a site’s 1D data (and a TEM sounding).

data is a TransferFunction (its mode is taken, see sounding_from_tf()) or a Sounding1D. thickness fixes the layers; otherwise n_layers are spread over the skin-depth range. static_shift adds one offset of log10 rho_a per mode. tem={"data": ..., "geometry": ..., "inversion": ...} is a TDEM sounding as PyHydroGeophysX.inversion.em1d.tdem_invert() takes it, fitted with weight tem_weight; the RMS target then applies to all data together.

PyHydroGeophysX.workflows.mt.phase_tensor(tf: TransferFunction, *, n_realizations: int = 0, seed: int = 0) → Dict[str, ndarray]

The phase tensor and its invariants (degrees), per frequency.

Returns phi (n, 2, 2) (in the TF’s frame) and phi_max, phi_min, alpha, beta (skew), strike and ellipticity; alpha and strike are azimuths clockwise from north, modulo 180 (strike has the usual 90-degree ambiguity). With n_realizations, their standard errors (*_err) come from perturbing Z by its errors.

PyHydroGeophysX.workflows.mt.process_mt(runs: TimeSeriesRun | Sequence[TimeSeriesRun], *, remote: TimeSeriesRun | Sequence[TimeSeriesRun] | None = None, config: ProcessingConfig | None = None, station: str | None = None, log: Callable[[str], None] | None = None, **options: Any) → TransferFunction

Estimate a site’s impedance (and tipper, when it has Hz) from its runs.

remote holds the reference site’s runs (its Hx and Hy are used). config or keyword options set ProcessingConfig fields; log receives a line per sample rate.

PyHydroGeophysX.workflows.mt.read_band_setup(path: Any) → List[Band]

An EMTF band-setup file: the number of bands, then level first last rows.

PyHydroGeophysX.workflows.mt.read_timeseries(path: Any, *, azimuths: Mapping[str, float] | None = None, **options: Any) → List[TimeSeriesRun]

Read an instrument’s MT time series - a file or a recording folder - into runs.

The format is recognized from the files: Phoenix MTU-5C/5P/8A recordings (read_phoenix()), legacy Phoenix MTU-5A TS/TBL sites (read_phoenix_legacy()), Metronix ATS (read_metronix()), Zonge Z3D (read_zonge()) and LEMI-424 text files (read_lemi424()). options go to that reader; format forces one.

azimuths ({"ex": 180, "hy": 270, ...}, degrees clockwise from north) replaces the direction the files give those channels, in every run: for a layout that differs from what the receiver was told, such as a dipole laid out to the south or a coil pointing west. Processing expresses every channel in the frame of Hx, so a reversed channel needs nothing more.

PyHydroGeophysX.workflows.mt.read_transfer_function(path: Any, *, sign_convention: str = 'auto', z_units: str = 'auto') → TransferFunction

Read one site’s transfer functions from an EDI, EMTF XML, Z- or J-file.

sign_convention applies to EDI and J-files, which do not state it (see read_edi()), and z_units to J-files (see read_jfile()); EMTF XML and Z-files state both.

PyHydroGeophysX.workflows.mt.save_runs(destination: Any, runs: Sequence[TimeSeriesRun])

Write runs as one compressed container (.npz); returns its path.

PyHydroGeophysX.workflows.mt.water_content_profile(result: Occam1DResult, *, rhos: Any, n: Any, porosity: Any, sigma_sur: Any = 0.0) → Dict[str, ndarray]

Volumetric water content of each layer from its resistivity (Waxman-Smits).

The petrophysical link the rest of PyHydroGeophysX uses (PyHydroGeophysX.petrophysics.resistivity_models.resistivity_to_water_content()): rhos the saturated resistivity, n the saturation exponent, porosity and sigma_sur the surface conductivity - each one value or one per layer.

PyHydroGeophysX.workflows.mt.write_edi(tf: TransferFunction, path: Any) → Path

Write tf as a standard EDI file, impedances in (mV/km)/nT, e^{+i omega t}.

The impedance blocks carry Z**.VAR as the squared standard error, the tipper the experimental T*.EXP blocks every program reads, and both rotation blocks the site’s rotation.

PyHydroGeophysX.workflows.mt.write_emtf_xml(tf: TransferFunction, path: Any) → Path

Write tf as an EMTF XML file, impedances in (mV/km)/nT, e^{+i omega t}.

The full error model is written when the site carries it; otherwise only the variances, err^2. One rotation angle is written for the whole site, so a site whose frequencies are in different frames is first rotated to the frame of its first frequency.

PyHydroGeophysX.workflows.objects module

A workflow’s live objects across a process boundary.

A workflow run in a process of its own (the studio’s ProcessWorkflowWorker) hands its result back as JSON, and WorkflowRunResult.objects - the meshes, arrays, tables and fitted models a page shows - would stay behind in the child. save_objects() writes them into a folder beside the result file and returns a JSON manifest; load_objects() reads them back in the studio, so a page gets the same objects whether the run was in a thread or a process.

Every value is written in the format made for it: arrays as .npy, a PyGIMLi mesh as .bms with the sidecar that keeps what BMS drops, a data container in its own file format, a table with pandas, and a fitted PyGIMLi manager - which holds solver state no file can - as the parts a page draws (ModelResult). Mappings, sequences and dataclasses are taken apart and put back together. Only a plain Python object none of these fit is pickled, and one that cannot be written at all is named in the manifest’s skipped, not dropped silently.

Nothing here imports PyGIMLi or pandas until a value needs them.

PyHydroGeophysX.workflows.objects.OBJECTS_SCHEMA = 1

Manifest layout; a studio reading a newer one says so instead of guessing.

PyHydroGeophysX.workflows.objects.is_plain_data(value: Any, depth: int = 0) → bool

Numbers, text, lists and string-keyed mappings of them - what JSON holds, NaN and infinity included. Small, so a workflow process sends it back whether asked for or not (a result’s summary cannot hold a NaN).

PyHydroGeophysX.workflows.objects.load_objects(manifest: Mapping[str, Any], folder: str | Path) → Tuple[Dict[str, Any], List[str]]

Read back what save_objects() wrote; returns (objects, problems).

problems names every object that is missing: skipped by the workflow process, or unreadable here.

PyHydroGeophysX.workflows.objects.objects_folder(result_file: str | Path) → Path

Where the objects of the run whose result is result_file are written.

PyHydroGeophysX.workflows.objects.save_objects(objects: Mapping[str, Any], folder: str | Path) → Dict[str, Any]

Write objects into folder; returns the JSON manifest.

{"schema": 1, "entries": {name: description}, "skipped": {name: reason}}, the descriptions naming files relative to folder.

PyHydroGeophysX.workflows.recipe module

Versioned JSON recipe IO.

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

Read UTF-8 JSON and validate its registered workflow and seed contract.

Returns a WorkflowSpec without executing it or resolving input artifacts. Relative artifact paths are resolved later using the run’s project root, not the recipe file’s parent directory. Read/JSON/validation errors propagate.

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]

Import the module:attribute target on demand and require a callable.

module_key: str = ''

stochastic: bool = False

workflow_id: str

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

Look up an exact workflow ID; raise KeyError listing available IDs if absent.

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

Return registered descriptors ordered by workflow ID, without loading handlers.

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

Register metadata without importing its handler; reject duplicate IDs with KeyError unless replace=True. Mutates the process-local registry.

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 synchronously.

spec may be a WorkflowSpec or its JSON-compatible mapping. Stochastic workflows require an explicit seed. context supplies the project/output paths, progress callback and cancellation predicate; a default context writes under the current directory’s workflow_results folder.

The output directory is created before calling the handler. Return a WorkflowRunResult whose serializable view has been validated; optional in-process objects remain on the result. Validation, import and handler exceptions propagate to the caller rather than becoming a success result.

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 0x7f4ba0253740>, 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.

A path is the artifact’s own, relative to the run directory the script is written into - inputs/... for what a studio run stored - not just its name, which pointed beside the script where nothing is.

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]

Import the module:attribute target on demand and require a callable.

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

Look up an exact workflow ID; raise KeyError listing available IDs if absent.

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, ...]

Return registered descriptors ordered by workflow ID, without loading handlers.

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

Read UTF-8 JSON and validate its registered workflow and seed contract.

Returns a WorkflowSpec without executing it or resolving input artifacts. Relative artifact paths are resolved later using the run’s project root, not the recipe file’s parent directory. Read/JSON/validation errors propagate.

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

Register metadata without importing its handler; reject duplicate IDs with KeyError unless replace=True. Mutates the process-local registry.

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 synchronously.

spec may be a WorkflowSpec or its JSON-compatible mapping. Stochastic workflows require an explicit seed. context supplies the project/output paths, progress callback and cancellation predicate; a default context writes under the current directory’s workflow_results folder.

The output directory is created before calling the handler. Return a WorkflowRunResult whose serializable view has been validated; optional in-process objects remain on the result. Validation, import and handler exceptions propagate to the caller rather than becoming a success result.

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.