PyHydroGeophysX.data_processing package

Submodules

PyHydroGeophysX.data_processing.em1d module

1D electromagnetic sounding and line-geometry readers.

PyHydroGeophysX.data_processing.em1d.SOUNDING_CONTAINER_KIND = 'em_soundings'

What save_sounding_container() writes and load_sounding() reads back, so a recorded run carries the soundings it imported rather than the acquisition folder they were parsed out of.

PyHydroGeophysX.data_processing.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.data_processing.em1d.is_ttem_source(path: str) → bool

Return whether path contains a TEMcompany tTEM raw acquisition.

PyHydroGeophysX.data_processing.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.data_processing.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.data_processing.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.data_processing.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.data_processing.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.data_processing.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.data_processing.ert_data_agent module

PyHydroGeophysX.data_processing.ert_formats module

PyHydroGeophysX.data_processing.ert_io module

Robust ERT file loading for the desktop studio (Qt-free).

A single source of truth for turning an ERT data file into a pygimli DataContainerERT with correct geometry, topography, and apparent resistivity. Both the single-inversion ERT module and the time-lapse pipeline use this so they behave identically.

Why this exists: pygimli’s native ert.load cannot parse several common field formats. The E4D survey export (used here with a .ohm extension), for example, carries a leading index column and no # a b m n token header, so ert.load misreads the index column as coordinates, drops the topography, and discards every measurement (size() == 0). The device-specific parsers in PyHydroGeophysX.data_processing.ert_data_agent (resipy or the embedded fallback) handle those layouts; this module wires them to pygimli and falls back across loaders so a file never silently loads empty.

PyHydroGeophysX.data_processing.ert_io.electrode_elevation(electrodes) → ndarray

Per-electrode elevation aligned with electrodes.

load_ert_resipy carries 2D elevation in y (its _normalize_elevation_axis moves a flat-y/varying-z profile so elevation lives in y); fall back to z if y is flat.

PyHydroGeophysX.data_processing.ert_io.load_ert_container(path: str, instrument: str | None = None, electrode_file: str | None = None, spacing: float | None = None, log: ~typing.Callable[[str], None] = <function noop>)

Load one ERT file into a pygimli DataContainerERT, robustly.

An explicit instrument uses the device parsers (handles index-prefixed E4D, BERT topography, Syscal, etc.); None/"auto" uses pygimli’s own reader. If the chosen path yields no measurements the loader falls back: an explicit instrument retries native pygimli; a still-empty result triggers a short sweep of count-prefixed device formats so an E4D-style file never loads empty. Raises ValueError if nothing parses.

PyHydroGeophysX.data_processing.ert_io.measurement_times_for(files: Sequence[str]) → Tuple[List[float], List[str]]

Derive numeric measurement times + display labels from filenames.

When every filename embeds a distinct date, times are elapsed days from the first acquisition and labels are YYYY-MM-DD. Otherwise falls back to a sequential 1..n with index labels.

PyHydroGeophysX.data_processing.ert_io.normalize_for_timelapse(files: ~typing.Sequence[str], instrument: str | None, out_dir: str, log: ~typing.Callable[[str], None] = <function noop>, max_error: float | None = None)

Load each file robustly and write a clean pygimli .dat into one folder.

The core TimeLapseERTInversion reloads files with ert.load; the normalized files are written in pygimli’s native unified format (proper token headers, geometric factors, rhoa = R*k, topography in z) so that reload is correct. As in the published AD-TLERT real-data workflow, one common quality mask is intersected across every time step before anything is written. This guarantees that every window sees the same ABMN rows and prevents a datum that is bad at one time from changing the inverse problem only at that time. Returns (clean_dir, basenames, containers); all clean files share one folder so the windowed inversion (which takes a directory + filenames) works directly.

PyHydroGeophysX.data_processing.ert_io.save_edited_ert_container(data: Any, destination: str | Path, electrodes: Sequence[dict]) → str

Persist QC-filtered ERT data with the current electrode edits applied.

original_index is zero-based for retained sensors and None for a newly added sensor. Deleting an original sensor removes measurements that reference it through PyGIMLi’s own removeSensorIdx implementation. Retained sensors must preserve their original relative order; the current Qt editor moves/adds/deletes but does not expose arbitrary reordering.

PyHydroGeophysX.data_processing.ert_io.standard_to_pg(std)

Build a pygimli DataContainerERT from a StandardERT for inversion.

Most instrument loaders report transfer resistance (V/I), not apparent resistivity, so apparent resistivity is recovered with geometric factors: rhoa = R * k. When the source already provides apparent resistivity it is used as-is. pygimli’s forward operator uses the same data["k"], so the inversion stays self-consistent. Returns None if pygimli is unavailable or no measurement maps onto the electrode set.

PyHydroGeophysX.data_processing.field_formats module

Field data IO helpers for seismic and EM workflows.

PyHydroGeophysX.data_processing.field_formats.export_results_to_csv(result: Any, filename: str) → str

Export inversion result arrays to a tabular CSV file.

PyHydroGeophysX.data_processing.field_formats.export_to_vtk(result: Any, mesh: Any, filename: str) → str

Export result/model values to a lightweight VTK PolyData file.

The export writes cell-center points with a single scalar field.

PyHydroGeophysX.data_processing.field_formats.read_seg2_seismic(file: str) → Dict[str, ndarray]

Read SEG-2 seismic file into arrays.

If ObsPy is available, uses ObsPy SEG-2 reader. Otherwise falls back to text loading for already exported arrays.

PyHydroGeophysX.data_processing.field_formats.read_tem_fast(file: str) → Dict[str, ndarray]

Read TEM-FAST style text data into SimPEG-ready arrays.

PyHydroGeophysX.data_processing.gravmag module

Gravity/magnetics preprocessing, QC, gridding, and profiles.

PyHydroGeophysX.data_processing.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.data_processing.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.data_processing.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.data_processing.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.data_processing.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.data_processing.joint_io module

Process-safe persistence for joint-inversion observations.

The array packing lives in PyHydroGeophysX.data_processing.run_inputs, which every module now writes its run inputs through. The manifest keys below stay as they were so joint runs saved before that move still load; only the ERT/SRT branches, which hand off to PyGIMLi’s own writers, are specific to this module.

PyHydroGeophysX.data_processing.joint_io.load_joint_observations(method: str, source: str | Path) → Any

Load a sidecar written by save_joint_observations().

PyHydroGeophysX.data_processing.joint_io.save_joint_observations(method: str, payload: Any, destination: str | Path) → Path

Persist one method’s observations without pickle or Qt dependencies.

PyHydroGeophysX.data_processing.model_csv module

Flat CSV views of mesh-based inversion models.

Every 2-D inversion in this package stores its result as a PyGIMLi mesh plus a value vector ordered by cell. That pair is exact, but reading it back needs PyGIMLi installed and the knowledge that the ordering is by paraDomain cell. A collaborator who only wants to redraw the section in matplotlib, R, or a GIS should not have to acquire either.

The tables written here carry the coordinate on every row, so a section can be redrawn with nothing but the file:

import pandas as pd, matplotlib.pyplot as plt d = pd.read_csv(“model_cells.csv”) plt.tricontourf(d.x, d.z, d.resistivity_ohm_m)

mesh_nodes.csv and mesh_cell_nodes.csv are written alongside for anyone who wants the true cell polygons rather than an interpolation through the centroids. They are optional to read and cost one pass over the mesh to write.

Coordinates follow the convention used elsewhere in this package: a 2-D PyGIMLi mesh stores the section plane in its first two components, so those are written as x and z (elevation, positive up) and no y column appears. A 3-D mesh writes all three.

PyHydroGeophysX.data_processing.model_csv.column_name(value_name: str, units: str = '') → str

Build a self-describing column header, e.g. resistivity_ohm_m.

The unit belongs in the header rather than in a sidecar, because a table that travels on its own is the point of writing it.

PyHydroGeophysX.data_processing.model_csv.export_model_csv(out_dir: str | Path, mesh: Any, values: Any, *, value_name: str = 'value', units: str = '', coverage: Any = None, step_labels: Sequence[Any] | None = None, extra_columns: Dict[str, Any] | None = None, geometry: bool = True) → List[str]

Write the per-cell table and, unless disabled, the mesh geometry.

Returns the paths written, in the order a reader should meet them.

PyHydroGeophysX.data_processing.model_csv.model_cell_table(mesh: Any, values: Any, *, value_name: str = 'value', units: str = '', coverage: Any = None, step_labels: Sequence[Any] | None = None, extra_columns: Dict[str, Any] | None = None) → Tuple[List[str], List[List[str]]]

Build the header and rows of the per-cell table without writing anything.

Exposed separately so callers can add a column or inspect the table in a test without going through the filesystem.

PyHydroGeophysX.data_processing.model_csv.write_grid_model_csv(path: str | Path, edges: Sequence[Any], values: Any, *, value_name: str = 'value', units: str = '', order: str = 'F') → Path

Write a regular 3-D grid model as one row per cell.

The gravity, magnetics, and joint potential-field inversions solve on a rectilinear grid rather than a PyGIMLi mesh, so they cannot use write_model_csv(). edges is the three edge vectors in x, y, z, and values is shaped (nx, ny, nz) or flat in order (SimPEG’s tensor meshes are Fortran-ordered, which is why that is the default).

Cell extents are written alongside the centre, so a reader can draw the true voxels instead of guessing the spacing back from the centres.

PyHydroGeophysX.data_processing.model_csv.write_layered_model_csv(path: str | Path, thicknesses: Any, values: Any, *, value_name: str = 'value', units: str = '', positions: Sequence[Any] | None = None) → Path

Write a 1-D layered model as one row per layer per sounding.

The bottom layer of a layered inversion is a half-space, so its lower depth is left empty rather than invented.

PyHydroGeophysX.data_processing.model_csv.write_mesh_geometry_csv(out_dir: str | Path, mesh: Any) → List[Path]

Write the node coordinates and each cell’s node ids.

Together with model_cells.csv this is enough to draw the true cell polygons, which a centroid interpolation cannot reproduce near the topography or along a sharp region boundary.

PyHydroGeophysX.data_processing.model_csv.write_model_csv(path: str | Path, mesh: Any, values: Any, *, value_name: str = 'value', units: str = '', coverage: Any = None, step_labels: Sequence[Any] | None = None, extra_columns: Dict[str, Any] | None = None) → Path

Write one row per mesh cell: coordinate, marker, size, and value(s).

PyHydroGeophysX.data_processing.run_inputs module

One compact on-disk format for what a run keeps about its inputs.

A run used to record its inputs by copying whatever the user had selected, so a run cost as much disk as the survey it read: a TEMcompany project folder runs to hundreds of megabytes and every inversion took another copy of it. What a run needs to be reproducible is the data the page imported, which is arrays and a little metadata, and not the acquisition those arrays were parsed out of.

The format is the one PyHydroGeophysX.data_processing.joint_io already used for its non-PyGIMLi observations, lifted out so every module writes the same thing: a single .npz holding the arrays, plus a JSON manifest stored under __manifest__ that describes the structure they came from. Nesting is preserved, so a loader’s return value can be handed straight to save_container() and comes back the same shape.

np.savez_compressed does the shrinking. Field data is mostly smooth float arrays and compresses well, and no pickle is involved, so the file crosses a process boundary into an isolated workflow run the same way a plain array does.

PyHydroGeophysX.data_processing.run_inputs.SCHEMA_VERSION = 1

Bumped only when an older reader could misread a newer file. Additive metadata does not need it.

PyHydroGeophysX.data_processing.run_inputs.bundle_file_names(source: str | Path) → list

The names expand_file_bundle() would write.

PyHydroGeophysX.data_processing.run_inputs.expand_file_bundle(source: str | Path, destination: str | Path) → Path

Write a bundle back out as loose files and return the directory.

PyHydroGeophysX.data_processing.run_inputs.is_container(source: str | Path) → bool

True when source is one of our containers.

Reads the zip directory rather than the arrays, so this stays cheap enough to sit in a loader’s dispatch chain in front of the file sniffers.

PyHydroGeophysX.data_processing.run_inputs.is_file_bundle(source: str | Path) → bool

True when source is a bundle written by save_file_bundle().

PyHydroGeophysX.data_processing.run_inputs.load_container(source: str | Path, *, kind: str | None = None) → Any

Load a container written by save_container().

Passing kind checks the file is the one the caller expects, which turns a mismatched path into a clear error instead of a downstream KeyError.

PyHydroGeophysX.data_processing.run_inputs.load_sequence_item(source: str | Path, index: int, *, kind: str | None = None) → Any

Return one item from a container written by save_sequence_container().

PyHydroGeophysX.data_processing.run_inputs.pack(value: Any, arrays: Dict[str, ndarray]) → Dict[str, Any]

Split value into a JSON-safe skeleton and a flat array table.

PyHydroGeophysX.data_processing.run_inputs.read_manifest(source: str | Path) → Dict[str, Any]

Return the manifest without unpacking any array.

PyHydroGeophysX.data_processing.run_inputs.save_container(destination: str | Path, payload: Any, *, kind: str, meta: Mapping[str, Any] | None = None) → Path

Write payload as one compressed container and return its path.

kind names what the file holds ("em_soundings", "hydrology_arrays", …) so a reader can refuse a file it was not meant to open. meta carries anything a reader needs before it unpacks the arrays, such as the method the data was imported as.

PyHydroGeophysX.data_processing.run_inputs.save_file_bundle(destination: str | Path, files: Mapping[str, str | Path], *, kind: str, meta: Mapping[str, Any] | None = None) → Path

Store a set of files as one compressed container, keyed by file name.

Some inputs cannot be reduced to arrays on the way in: a PyGIMLi mesh and a BERT data file are read back by PyGIMLi’s own loaders, which want a real path. Their bytes are kept verbatim, so what a run saves here is the compression and the file count rather than a change of representation. ASCII data files shrink several-fold; .npy arrays shrink by whatever deflate finds in them.

Pair with expand_file_bundle(), which puts the files back under their original names so the reader sees the directory it expects.

PyHydroGeophysX.data_processing.run_inputs.save_sequence_container(destination: str | Path, items: list, *, kind: str, meta: Mapping[str, Any] | None = None) → Path

Write a list of like-shaped payloads, storing what they share once.

PyHydroGeophysX.data_processing.run_inputs.sequence_length(source: str | Path) → int

How many items load_sequence_item() can return.

PyHydroGeophysX.data_processing.run_inputs.unpack(node: Mapping[str, Any], archive: Any) → Any

Rebuild what pack() took apart.

PyHydroGeophysX.data_processing.seismic module

Raw seismic data processing helpers.

SEG-Y reading prefers a robust third-party library and degrades gracefully so local workflows keep working with no extra installs. The reader order is: segyio (broadest coverage: endianness, sample-format codes, byte locations, SEG-Y rev 1/2), then ObsPy, then a small built-in big-endian reader that handles the common SEG-Y layout used by the bundled example. Install segyio (pip install segyio) for the most reliable reads of arbitrary field data.

class PyHydroGeophysX.data_processing.seismic.FirstBreakPick(source_id: int, receiver_id: int, time_s: float, source_x: float, source_z: float, receiver_x: float, receiver_z: float, field_record: int, trace_number: int, trace_index: int, amplitude: float)

Bases: object

One first-break pick.

amplitude: float

field_record: int

receiver_id: int

receiver_x: float

receiver_z: float

source_id: int

source_x: float

source_z: float

time_s: float

to_dict() → Dict[str, Any]

Return a CSV/JSON friendly representation.

trace_index: int

trace_number: int

class PyHydroGeophysX.data_processing.seismic.SegyMetadata(sample_interval_us: int, samples_per_trace: int, format_code: int, trace_count: int, file_size_bytes: int)

Bases: object

Summary metadata from a SEG-Y file.

file_size_bytes: int

format_code: int

property sample_interval_s: float

Sampling interval in seconds.

sample_interval_us: int

samples_per_trace: int

trace_count: int

class PyHydroGeophysX.data_processing.seismic.SeismicDataset(path: str, traces: ndarray, time: ndarray, headers: List[SeismicTraceHeader], metadata: SegyMetadata)

Bases: object

In-memory SEG-Y dataset with traces arranged as samples by traces.

property field_records: List[int]

Sorted shot/field-record identifiers available in this dataset.

get_gather(field_record: int | None = None) → SeismicShotGather

Return one shot gather.

Parameters:

field_record (int, optional) – Field-record id. If omitted, the first available record is used.

Returns:

Gather with traces, channel ids, offsets, and headers.

Return type:

SeismicShotGather

headers: List[SeismicTraceHeader]

metadata: SegyMetadata

path: str

time: ndarray

traces: ndarray

class PyHydroGeophysX.data_processing.seismic.SeismicShotGather(field_record: int, trace_indices: ndarray, traces: ndarray, time: ndarray, channels: ndarray, offsets: ndarray, headers: List[SeismicTraceHeader])

Bases: object

One shot gather extracted from a seismic dataset.

channels: ndarray

field_record: int

headers: List[SeismicTraceHeader]

offsets: ndarray

time: ndarray

trace_indices: ndarray

traces: ndarray

class PyHydroGeophysX.data_processing.seismic.SeismicTraceHeader(field_record: int, trace_number: int, energy_source_point: int, source_x: float, source_y: float, source_z: float, receiver_x: float, receiver_y: float, receiver_z: float, offset: float)

Bases: object

Subset of SEG-Y trace-header fields used by the GUI workflow.

energy_source_point: int

field_record: int

offset: float

receiver_x: float

receiver_y: float

receiver_z: float

source_x: float

source_y: float

source_z: float

trace_number: int

PyHydroGeophysX.data_processing.seismic.apply_agc(data: ndarray, dt: float, window: float = 0.05, rms: float = 1.0) → ndarray

Apply gate-based automatic gain control to traces.

The implementation follows the MATLAB example’s gate/interpolation logic while handling zero-energy windows safely.

PyHydroGeophysX.data_processing.seismic.bandpass_filter(data: ndarray, dt: float, f1: float, f2: float, f3: float, f4: float) → ndarray

Apply a zero-phase Butterworth bandpass using the pass-band edges.

f1 and f4 are retained for API compatibility with Ormsby-style four-corner filters; this implementation uses f2 and f3 as the pass-band edges.

PyHydroGeophysX.data_processing.seismic.export_first_breaks(picks: Iterable[FirstBreakPick | Dict[str, Any]], filename: str) → str

Export first-break picks to CSV.

PyHydroGeophysX.data_processing.seismic.export_traveltime_container(data: Any, filename: str) → str

Persist a loaded PyGIMLi travel-time container in portable BERT format.

PyHydroGeophysX.data_processing.seismic.first_breaks_to_traveltime(picks: Iterable[FirstBreakPick | Dict[str, Any]], filename: str, receiver_spacing: float = 1.0, shot_spacing: float | None = None) → str

Export first breaks to a PyGIMLi/BERT travel-time .dat file.

PyHydroGeophysX.data_processing.seismic.normalize_traces(data: ndarray, trace_axis: int = 1, eps: float = 1e-12) → ndarray

Normalize each seismic trace by its maximum absolute amplitude.

PyHydroGeophysX.data_processing.seismic.pick_first_breaks(data: SeismicDataset | SeismicShotGather | ndarray, dt: float | None = None, headers: Sequence[SeismicTraceHeader] | None = None, threshold: float = 0.2, noise_multiplier: float = 5.0, min_time: float = 0.0, max_time: float | None = None, polarity: float = 1.0) → List[FirstBreakPick]

Pick first arrivals using a simple amplitude/noise threshold.

This assisted picker is intended as a starting point for GUI review rather than a final scientific picking algorithm.

PyHydroGeophysX.data_processing.seismic.read_segy(file: str, max_traces: int | None = None, load_traces: bool = True, prefer_obspy: bool = True) → SeismicDataset

Read a SEG-Y file into traces, headers, and metadata.

Reader order is segyio -> ObsPy -> built-in: the most robust library available is used, and the built-in conservative reader is the final fallback so reads keep working with no third-party SEG-Y dependency installed.

Parameters:

• file (str) – SEG-Y path.

• max_traces (int, optional) – Maximum number of traces to read. Useful for responsive GUI previews.

• load_traces (bool, optional) – If False, parse headers but skip sample arrays (built-in reader only).

• prefer_obspy (bool, optional) – When True (default), try ObsPy if segyio is unavailable or fails.

Returns:

Parsed seismic dataset.

Return type:

SeismicDataset

PyHydroGeophysX.data_processing.seismic.tukey_taper(n_samples: int, taper_samples: int) → ndarray

Return a Tukey-style taper with taper length in samples.

PyHydroGeophysX.data_processing.table_io module

Lightweight numeric table I/O shared by core and desktop workflows.

PyHydroGeophysX.data_processing.table_io.ensure_dir(path: str | Path) → Path

Create path and its parents if needed, then return it.

PyHydroGeophysX.data_processing.table_io.load_2d_array(path: str | Path) → ndarray

Load an array from NPY, NPZ, CSV, TXT, or DAT input.

PyHydroGeophysX.data_processing.table_io.load_xyz_table(path: str | Path, min_cols: int = 2) → ndarray

Load a two-dimensional table with at least min_cols columns.

PyHydroGeophysX.data_processing.table_io.npy_shape(path: str | Path) → Tuple[int, ...]

Read an .npy file’s shape from its header, without opening the data.

np.load(..., mmap_mode="r") is the usual way to ask an array how big it is, but a mapping keeps the file open for as long as the array lives, and Windows then refuses to let anything overwrite it. Reading the header costs one short read, closes immediately, and works even while another process is rewriting the file.

PyHydroGeophysX.data_processing.table_io.read_json(path: str | Path) → dict[str, Any] | None

Read JSON, returning None for a missing or malformed document.

PyHydroGeophysX.data_processing.table_io.save_npy_atomic(path: str | Path, array: Any) → Path

Write an .npy via a sibling temp file, then swap it into place.

A direct np.save over an existing result truncates it first, so a write that fails part way leaves a corrupt file that still looks like a result. Staging keeps the previous file intact on failure and never publishes a half-written array.

This does not defeat a lock: replacing a file another process holds mapped still raises, by design. It makes that failure clean rather than destructive.

PyHydroGeophysX.data_processing.table_io.write_csv(path: str | Path, rows: Sequence[Sequence[Any]], header: Iterable[str] | None = None) → Path

Write rows to a CSV file.

PyHydroGeophysX.data_processing.table_io.write_json(path: str | Path, obj: Any) → Path

Atomically write a JSON document.

PyHydroGeophysX.data_processing.temcompany_reference module

Read inversion results stored inside a TEMcompany project file.

project.db keeps more than the recorded transients. Where a survey has already been inverted, InversionModel holds the recovered model, the depth of investigation, the data fit, and a Datasets blob carrying, per moment, the gate windows used, the data the inversion was given (InputData), the uncertainty assigned to it (InputSTD) and the response computed for the recovered model (ForwardData).

This module reads those records so that results produced earlier can be loaded alongside the soundings they came from: plotted on a section, compared with a new inversion, or used to check that this package’s own reader returns the same gates, values and errors the file holds. Measured over the 882 inverted stations of one survey, InputData is identical to LM/HM_VoltageValues and InputSTD to LM/HM_VoltageValues_STD to the last bit, which is the property tests.test_temcompany_alignment asserts.

Nothing here executes anything. It parses a database the acquisition software wrote.

PyHydroGeophysX.data_processing.temcompany_reference.REFERENCE_MOMENTS: Tuple[str, ...] = ('LM', 'HM')

Moments a Datasets blob can name.

PyHydroGeophysX.data_processing.temcompany_reference.has_reference_models(path: str | Path) → bool

Whether path holds inversion results as well as soundings.

PyHydroGeophysX.data_processing.temcompany_reference.iter_reference_stations(path: str | Path, inversion_name: str | None = None) → Iterator[Dict[str, Any]]

Yield one dict per station of a stored inversion, in file order.

inversion_name selects among several saved runs; the default is whichever name carries the most rows, which is the full-survey inversion in every project seen so far.

PyHydroGeophysX.data_processing.temcompany_reference.load_reference_models(path: str | Path, inversion_name: str | None = None) → Dict[str, Any]

Every station of one stored inversion, with the project’s own settings.

Returns stations alongside the RxTxSpecs block and the acquisition protocol, so a caller has the full geometry, waveform and gate description that produced the stored responses.

PyHydroGeophysX.data_processing.temcompany_reference.reference_inversion_names(path: str | Path) → List[str]

Every named inversion the project holds, the current one first.

“Current” is the project’s own answer where it has one: InverseSettings records LastInversionName, which is the run the application would show. Only where that is missing does the order fall back to a guess, and the guess is row count rather than timestamp because a later run is often a re-inversion of a few stations while the one worth comparing against covers the survey.

PyHydroGeophysX.data_processing.ttem module

Reader for TEMcompany tTEM SKB/SPS raw acquisitions.

The acquisition logger writes alternating-polarity transient series to SKB and navigation/transmitter telemetry to SPS. This module indexes those files lazily, stacks a few consecutive cycles into a sounding, and exposes the same dictionary contract as the TEM2Go reader.

PyHydroGeophysX.data_processing.ttem.is_ttem_source(path: str) → bool

Return whether path contains a TEMcompany tTEM raw acquisition.

PyHydroGeophysX.data_processing.ttem.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.

Module contents

Lazy data-processing exports.

Importing a lightweight reader must not initialize ERT plotting code or probe optional geophysics engines. Public objects are therefore loaded only when they are requested.

class PyHydroGeophysX.data_processing.FirstBreakPick(source_id: int, receiver_id: int, time_s: float, source_x: float, source_z: float, receiver_x: float, receiver_z: float, field_record: int, trace_number: int, trace_index: int, amplitude: float)

Bases: object

One first-break pick.

amplitude: float

field_record: int

receiver_id: int

receiver_x: float

receiver_z: float

source_id: int

source_x: float

source_z: float

time_s: float

to_dict() → Dict[str, Any]

Return a CSV/JSON friendly representation.

trace_index: int

trace_number: int

class PyHydroGeophysX.data_processing.SegyMetadata(sample_interval_us: int, samples_per_trace: int, format_code: int, trace_count: int, file_size_bytes: int)

Bases: object

Summary metadata from a SEG-Y file.

file_size_bytes: int

format_code: int

property sample_interval_s: float

Sampling interval in seconds.

sample_interval_us: int

samples_per_trace: int

trace_count: int

class PyHydroGeophysX.data_processing.SeismicDataset(path: str, traces: ndarray, time: ndarray, headers: List[SeismicTraceHeader], metadata: SegyMetadata)

Bases: object

In-memory SEG-Y dataset with traces arranged as samples by traces.

property field_records: List[int]

Sorted shot/field-record identifiers available in this dataset.

get_gather(field_record: int | None = None) → SeismicShotGather

Return one shot gather.

Parameters:

field_record (int, optional) – Field-record id. If omitted, the first available record is used.

Returns:

Gather with traces, channel ids, offsets, and headers.

Return type:

SeismicShotGather

headers: List[SeismicTraceHeader]

metadata: SegyMetadata

path: str

time: ndarray

traces: ndarray

class PyHydroGeophysX.data_processing.SeismicShotGather(field_record: int, trace_indices: ndarray, traces: ndarray, time: ndarray, channels: ndarray, offsets: ndarray, headers: List[SeismicTraceHeader])

Bases: object

One shot gather extracted from a seismic dataset.

channels: ndarray

field_record: int

headers: List[SeismicTraceHeader]

offsets: ndarray

time: ndarray

trace_indices: ndarray

traces: ndarray

class PyHydroGeophysX.data_processing.SeismicTraceHeader(field_record: int, trace_number: int, energy_source_point: int, source_x: float, source_y: float, source_z: float, receiver_x: float, receiver_y: float, receiver_z: float, offset: float)

Bases: object

Subset of SEG-Y trace-header fields used by the GUI workflow.

energy_source_point: int

field_record: int

offset: float

receiver_x: float

receiver_y: float

receiver_z: float

source_x: float

source_y: float

source_z: float

trace_number: int

class PyHydroGeophysX.data_processing.TravelTimeModel(segments: List[TravelTimeModelSegment], residual_rms_s: float, branch_ids: List[str], predicted_times: ndarray | None = None, message: str = '')

Bases: object

Piecewise-linear 1-D velocity model used for first-arrival guidance.

branch_ids: List[str]

message: str = ''

predicted_times: ndarray | None = None

residual_rms_s: float

segments: List[TravelTimeModelSegment]

class PyHydroGeophysX.data_processing.TravelTimeModelSegment(branch_id: str, segment_index: int, x_min: float, x_max: float, slope_s_per_m: float, intercept_s: float, apparent_velocity_m_s: float, crossover_offset_m: float | None = None, intercept_time_s: float | None = None, depth_estimate_m: float | None = None)

Bases: object

One fitted straight travel-time branch in offset space.

apparent_velocity_m_s: float

branch_id: str

crossover_offset_m: float | None = None

depth_estimate_m: float | None = None

intercept_s: float

intercept_time_s: float | None = None

segment_index: int

slope_s_per_m: float

x_max: float

x_min: float

PyHydroGeophysX.data_processing.apply_agc(data: ndarray, dt: float, window: float = 0.05, rms: float = 1.0) → ndarray

Apply gate-based automatic gain control to traces.

The implementation follows the MATLAB example’s gate/interpolation logic while handling zero-energy windows safely.

PyHydroGeophysX.data_processing.bandpass_filter(data: ndarray, dt: float, f1: float, f2: float, f3: float, f4: float) → ndarray

Apply a zero-phase Butterworth bandpass using the pass-band edges.

f1 and f4 are retained for API compatibility with Ormsby-style four-corner filters; this implementation uses f2 and f3 as the pass-band edges.

PyHydroGeophysX.data_processing.ensure_dir(path: str | Path) → Path

Create path and its parents if needed, then return it.

PyHydroGeophysX.data_processing.export_first_breaks(picks: Iterable[FirstBreakPick | Dict[str, Any]], filename: str) → str

Export first-break picks to CSV.

PyHydroGeophysX.data_processing.export_model_csv(out_dir: str | Path, mesh: Any, values: Any, *, value_name: str = 'value', units: str = '', coverage: Any = None, step_labels: Sequence[Any] | None = None, extra_columns: Dict[str, Any] | None = None, geometry: bool = True) → List[str]

Write the per-cell table and, unless disabled, the mesh geometry.

Returns the paths written, in the order a reader should meet them.

PyHydroGeophysX.data_processing.export_results_to_csv(result: Any, filename: str) → str

Export inversion result arrays to a tabular CSV file.

PyHydroGeophysX.data_processing.export_to_vtk(result: Any, mesh: Any, filename: str) → str

Export result/model values to a lightweight VTK PolyData file.

The export writes cell-center points with a single scalar field.

PyHydroGeophysX.data_processing.export_traveltime_container(data: Any, filename: str) → str

Persist a loaded PyGIMLi travel-time container in portable BERT format.

PyHydroGeophysX.data_processing.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.data_processing.first_breaks_to_traveltime(picks: Iterable[FirstBreakPick | Dict[str, Any]], filename: str, receiver_spacing: float = 1.0, shot_spacing: float | None = None) → str

Export first breaks to a PyGIMLi/BERT travel-time .dat file.

PyHydroGeophysX.data_processing.fit_velocity_traveltime_model(source_x: Sequence[float], receiver_x: Sequence[float], times_s: Sequence[float], weights: Sequence[float] | None = None, anchor_mask: Sequence[bool] | None = None, max_segments: int = 3, velocity_bounds: Tuple[float, float] = (100.0, 8000.0)) → TravelTimeModel

Fit a simple piecewise-linear 1-D travel-time model.

The fitted x-axis is absolute source-receiver offset. If the source lies inside a receiver spread, left and right receiver branches are fitted independently.

PyHydroGeophysX.data_processing.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.data_processing.has_reference_models(path: str | Path) → bool

Whether path holds inversion results as well as soundings.

PyHydroGeophysX.data_processing.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.data_processing.is_ttem_source(path: str) → bool

Return whether path contains a TEMcompany tTEM raw acquisition.

PyHydroGeophysX.data_processing.load_2d_array(path: str | Path) → ndarray

Load an array from NPY, NPZ, CSV, TXT, or DAT input.

PyHydroGeophysX.data_processing.load_joint_observations(method: str, source: str | Path) → Any

Load a sidecar written by save_joint_observations().

PyHydroGeophysX.data_processing.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.data_processing.load_reference_models(path: str | Path, inversion_name: str | None = None) → Dict[str, Any]

Every station of one stored inversion, with the project’s own settings.

Returns stations alongside the RxTxSpecs block and the acquisition protocol, so a caller has the full geometry, waveform and gate description that produced the stored responses.

PyHydroGeophysX.data_processing.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.data_processing.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.data_processing.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.data_processing.load_xyz_table(path: str | Path, min_cols: int = 2) → ndarray

Load a two-dimensional table with at least min_cols columns.

PyHydroGeophysX.data_processing.model_cell_table(mesh: Any, values: Any, *, value_name: str = 'value', units: str = '', coverage: Any = None, step_labels: Sequence[Any] | None = None, extra_columns: Dict[str, Any] | None = None) → Tuple[List[str], List[List[str]]]

Build the header and rows of the per-cell table without writing anything.

Exposed separately so callers can add a column or inspect the table in a test without going through the filesystem.

PyHydroGeophysX.data_processing.normalize_traces(data: ndarray, trace_axis: int = 1, eps: float = 1e-12) → ndarray

Normalize each seismic trace by its maximum absolute amplitude.

PyHydroGeophysX.data_processing.pick_first_breaks(data: SeismicDataset | SeismicShotGather | ndarray, dt: float | None = None, headers: Sequence[SeismicTraceHeader] | None = None, threshold: float = 0.2, noise_multiplier: float = 5.0, min_time: float = 0.0, max_time: float | None = None, polarity: float = 1.0) → List[FirstBreakPick]

Pick first arrivals using a simple amplitude/noise threshold.

This assisted picker is intended as a starting point for GUI review rather than a final scientific picking algorithm.

PyHydroGeophysX.data_processing.predict_velocity_traveltimes(model: TravelTimeModel, source_x: Sequence[float], receiver_x: Sequence[float]) → ndarray

Predict travel times from a fitted 1-D velocity model.

PyHydroGeophysX.data_processing.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.data_processing.read_geometrics_dat(path: str, max_traces: int | None = None, load_traces: bool = True) → SeismicDataset

Read Geometrics binary DAT seismic records.

Parameters:

• path (str) – Either a single Geometrics .dat file or a directory containing one .dat file per shot gather.

• max_traces (int, optional) – Maximum number of traces to read across all files.

• load_traces (bool, optional) – If False, parse headers but skip sample arrays.

Returns:

Dataset with all shots concatenated by trace column and separated by field_record.

Return type:

SeismicDataset

PyHydroGeophysX.data_processing.read_json(path: str | Path) → dict[str, Any] | None

Read JSON, returning None for a missing or malformed document.

PyHydroGeophysX.data_processing.read_seg2_seismic(file: str) → Dict[str, ndarray]

Read SEG-2 seismic file into arrays.

If ObsPy is available, uses ObsPy SEG-2 reader. Otherwise falls back to text loading for already exported arrays.

PyHydroGeophysX.data_processing.read_segy(file: str, max_traces: int | None = None, load_traces: bool = True, prefer_obspy: bool = True) → SeismicDataset

Read a SEG-Y file into traces, headers, and metadata.

Reader order is segyio -> ObsPy -> built-in: the most robust library available is used, and the built-in conservative reader is the final fallback so reads keep working with no third-party SEG-Y dependency installed.

Parameters:

• file (str) – SEG-Y path.

• max_traces (int, optional) – Maximum number of traces to read. Useful for responsive GUI previews.

• load_traces (bool, optional) – If False, parse headers but skip sample arrays (built-in reader only).

• prefer_obspy (bool, optional) – When True (default), try ObsPy if segyio is unavailable or fails.

Returns:

Parsed seismic dataset.

Return type:

SeismicDataset

PyHydroGeophysX.data_processing.read_tem_fast(file: str) → Dict[str, ndarray]

Read TEM-FAST style text data into SimPEG-ready arrays.

PyHydroGeophysX.data_processing.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.data_processing.save_edited_ert_container(data: Any, destination: str | Path, electrodes: Sequence[dict]) → str

Persist QC-filtered ERT data with the current electrode edits applied.

original_index is zero-based for retained sensors and None for a newly added sensor. Deleting an original sensor removes measurements that reference it through PyGIMLi’s own removeSensorIdx implementation. Retained sensors must preserve their original relative order; the current Qt editor moves/adds/deletes but does not expose arbitrary reordering.

PyHydroGeophysX.data_processing.save_joint_observations(method: str, payload: Any, destination: str | Path) → Path

Persist one method’s observations without pickle or Qt dependencies.

PyHydroGeophysX.data_processing.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.data_processing.tukey_taper(n_samples: int, taper_samples: int) → ndarray

Return a Tukey-style taper with taper length in samples.

PyHydroGeophysX.data_processing.write_csv(path: str | Path, rows: Sequence[Sequence[Any]], header: Iterable[str] | None = None) → Path

Write rows to a CSV file.

PyHydroGeophysX.data_processing.write_grid_model_csv(path: str | Path, edges: Sequence[Any], values: Any, *, value_name: str = 'value', units: str = '', order: str = 'F') → Path

Write a regular 3-D grid model as one row per cell.

The gravity, magnetics, and joint potential-field inversions solve on a rectilinear grid rather than a PyGIMLi mesh, so they cannot use write_model_csv(). edges is the three edge vectors in x, y, z, and values is shaped (nx, ny, nz) or flat in order (SimPEG’s tensor meshes are Fortran-ordered, which is why that is the default).

Cell extents are written alongside the centre, so a reader can draw the true voxels instead of guessing the spacing back from the centres.

PyHydroGeophysX.data_processing.write_json(path: str | Path, obj: Any) → Path

Atomically write a JSON document.

PyHydroGeophysX.data_processing.write_layered_model_csv(path: str | Path, thicknesses: Any, values: Any, *, value_name: str = 'value', units: str = '', positions: Sequence[Any] | None = None) → Path

Write a 1-D layered model as one row per layer per sounding.

The bottom layer of a layered inversion is a half-space, so its lower depth is left empty rather than invented.

PyHydroGeophysX.data_processing.write_mesh_geometry_csv(out_dir: str | Path, mesh: Any) → List[Path]

Write the node coordinates and each cell’s node ids.

Together with model_cells.csv this is enough to draw the true cell polygons, which a centroid interpolation cannot reproduce near the topography or along a sharp region boundary.

PyHydroGeophysX.data_processing.write_model_csv(path: str | Path, mesh: Any, values: Any, *, value_name: str = 'value', units: str = '', coverage: Any = None, step_labels: Sequence[Any] | None = None, extra_columns: Dict[str, Any] | None = None) → Path

Write one row per mesh cell: coordinate, marker, size, and value(s).