IO

The io subpackage is the boundary between groundfield and the rest of the software family. It carries no numerical core; every function here is a translator between an in-memory groundfield artefact (a fit, a result) and an external representation (a JSON file, a sister-project Pydantic model).

Mathematical and physical context

The reduction pipeline reads

\[ \text{PDE / field model} \;\longrightarrow\; \text{reduced } \rho\text{-}f \text{ model} \;\longrightarrow\; \text{measurement-anchored network model}. \]

The third step is owned by groundinsight (network-grade fault analysis on the reduced equivalent model). The interface that closes the loop is BusType.impedance_formula — a SymPy-compatible string in the two free symbols f (frequency in Hz) and rho (Bus.specific_earth_resistance in \(\Omega\,\mathrm{m}\)).

io.groundinsight produces exactly that string from either of the two rho-f fits supported by groundfield:

• Standard form (RhoFStandardFit) — the five-coefficient ansatz

$$ Z(\rho, f) \;=\; k_1 \rho \;+\; (k_2 + j k_3)\,f \;+\; (k_4 + j k_5)\,f\,\rho, $$

fitted across a \(\rho\)-sweep. Already in the canonical (rho, f) symbol set.

• Vector fit (VectorFitResult) — a Gustavsen/Semlyen rational \(Z(s)\) in the Laplace variable \(s = j\,2\pi f\) at fixed soil. The exporter performs the symbolic substitution

$$ s \;\longrightarrow\; j\,2\pi f $$

and renders the formula in \(f\) alone. The underlying soil resistivity is recorded in the metadata block (rho_at_fit_Ohm_m) but does not appear in the formula — a vector fit is by construction bound to a specific soil.

Validity

• Frequency range \(f \le 1\,\mathrm{kHz}\) — both fit families inherit the quasi-static assumption. The exported BusType is meaningful only inside this band; groundinsight evaluates the formula for whatever frequencies are stored on its Network, so the user is responsible for staying within the band.
• Linearity — the underlying field model is solved on the \(\nabla \cdot (\sigma\,\nabla\Phi) = 0\) regime; saturation effects in electrodes (e.g. ionisation under impulse currents) are not modelled. Exports are therefore appropriate for protection, reduction-factor and EPR studies, not for transient surge/lightning models.
• Multi-port reduction — the current API exports a single BusType, i.e. the driving-point impedance of one cluster. BranchType (mutual impedance) is out of scope for ADR-0008 and reserved for a follow-up.

Transport

Two equally supported paths, see ADR-0008 for the design rationale and the exact JSON schema (v1):

Path	Function	groundinsight install required?
JSON file	save_bustype_json / load_bustype_json	no
Python API	to_bustype / save_bustype_to_db	yes (extras group [groundinsight])
Hybrid	to_bustype_dict (JSON-ready dict; build & inspect)	no

CSV exports — groundfield.io.csv

Three convenience writers that turn a :class:FieldResult (and optionally its companion :class:World) into machine-readable, tool-agnostic CSV files. They wrap the existing postprocess helpers — no new science here, just a clean disk format for sharing typical results across notebooks, spreadsheets, and downstream pipelines.

import groundfield as gf

# 1. Sample the potential along a path and dump (s, x, y, z, freq, phi).
gf.save_potential_path_csv(
    result, "out/phi_radial.csv",
    start=(1.0, 0.0, 0.0), direction=(1, 0, 0),
    distance=30.0, n=200,
)
# 2. Dump the per-electrode summary.
gf.save_electrode_table_csv(result, "out/electrodes.csv", world=world)
# 3. Dump the per-cluster summary (members are flattened to a string).
gf.save_cluster_impedances_csv(result, "out/clusters.csv")

All writers use UTF-8, comma-separated, with a header row; floating-point values are written at full precision (%.17g) so files round-trip without loss of accuracy.

API reference — CSV

csv

CSV exports for groundfield results.

This module provides three convenience writers that turn a :class:FieldResult (and optionally its companion :class:World) into machine-readable, tool-agnostic CSV files. They wrap the existing postprocess helpers — no new science here, just a clean disk format for sharing typical results across notebooks, spreadsheets, and downstream pipelines.

Functions:

Name	Description
:func:`save_potential_path_csv`	Sample :meth:FieldResult.potential along a straight line on or below the soil surface and write (s, x, y, z, frequency_Hz, phi_re, phi_im, abs_phi) rows.
:func:`save_electrode_table_csv`	Wrap :func:groundfield.postprocess.electrode_current_table and dump the per-electrode summary.
:func:`save_cluster_impedances_csv`	Wrap :func:groundfield.postprocess.cluster_current_balance and dump the per-cluster summary.

Column-name convention

All complex-valued quantities follow the same <symbol>_re, <symbol>_im, abs_<symbol> triple where symbol is the physical symbol of the quantity:

• potential — phi_re / phi_im / abs_phi;
• current — I_re / I_im / abs_I;
• impedance — Z_re / Z_im / abs_Z.

The three writers therefore deliberately use different magnitude columns (abs_phi for the potential path, abs_I for the electrode table, abs_Z for the cluster table). They are not intended to be concat/merge-ed on the magnitude column without an explicit rename; the column constants below (:data:POTENTIAL_PATH_COLUMNS, etc.) lock the schema for downstream consumers.

All writers use UTF-8, comma-separated, with a header row; floating-point values are written at full precision so the files round-trip without loss of accuracy.

save_cluster_impedances_csv

save_cluster_impedances_csv(
    result: "FieldResult",
    path: str | Path,
    *,
    frequency_index: int = 0
) -> Path

Wrap :func:cluster_current_balance and write to CSV.

Note that the members column of :func:cluster_current_balance carries Python lists; CSV is a flat format, so the lists are joined into ';'-separated strings before writing.

Parameters:

Name	Type	Description	Default
result	'FieldResult'	Solver output.	required
path	str | Path	Destination CSV path.	required
frequency_index	int	Index into :attr:FieldResult.frequencies. Default 0.	0

Returns:

Type	Description
Path

Source code in src/groundfield/io/csv.py

def save_cluster_impedances_csv(
    result: "FieldResult",
    path: str | Path,
    *,
    frequency_index: int = 0,
) -> Path:
    """Wrap :func:`cluster_current_balance` and write to CSV.

    Note that the ``members`` column of
    :func:`cluster_current_balance` carries Python lists; CSV is
    a flat format, so the lists are joined into ``';'``-separated
    strings before writing.

    Parameters
    ----------
    result
        Solver output.
    path
        Destination CSV path.
    frequency_index
        Index into :attr:`FieldResult.frequencies`. Default 0.

    Returns
    -------
    pathlib.Path
    """
    from groundfield.postprocess.current_balance import cluster_current_balance

    df = cluster_current_balance(result, frequency_index=frequency_index)
    if "members" in df.columns:
        df = df.copy()
        df["members"] = df["members"].apply(lambda lst: ";".join(lst))
    out = _ensure_path(path)
    df.to_csv(out, index=False, float_format="%.17g")
    return out

save_electrode_table_csv

save_electrode_table_csv(
    result: "FieldResult",
    path: str | Path,
    *,
    world: "World | None" = None,
    frequency_index: int = 0
) -> Path

Wrap :func:electrode_current_table and write to CSV.

Parameters:

Name	Type	Description	Default
result	'FieldResult'	Solver output.	required
path	str | Path	Destination CSV path.	required
world	'World | None'	Optional companion world; when given, the table includes the kind and depth_m columns.	None
frequency_index	int	Index into :attr:FieldResult.frequencies. Default 0.	0

Returns:

Type	Description
Path

Source code in src/groundfield/io/csv.py

def save_electrode_table_csv(
    result: "FieldResult",
    path: str | Path,
    *,
    world: "World | None" = None,
    frequency_index: int = 0,
) -> Path:
    """Wrap :func:`electrode_current_table` and write to CSV.

    Parameters
    ----------
    result
        Solver output.
    path
        Destination CSV path.
    world
        Optional companion world; when given, the table includes
        the ``kind`` and ``depth_m`` columns.
    frequency_index
        Index into :attr:`FieldResult.frequencies`. Default 0.

    Returns
    -------
    pathlib.Path
    """
    from groundfield.postprocess.current_balance import electrode_current_table

    df = electrode_current_table(
        result, world=world, frequency_index=frequency_index
    )
    out = _ensure_path(path)
    df.to_csv(out, index=False, float_format="%.17g")
    return out

save_potential_path_csv

save_potential_path_csv(
    result: "FieldResult",
    path: str | Path,
    *,
    start: tuple[float, float, float],
    direction: tuple[float, float, float] = (1.0, 0.0, 0.0),
    distance: float = 30.0,
    n: int = 200,
    frequency_indices: Sequence[int] | None = None
) -> Path

Sample the potential along a line and write to CSV.

Builds an evenly spaced set of n field points starting at start and walking distance metres along the unit vector of direction, evaluates :meth:FieldResult.potential at every selected frequency index, and writes the result to path.

Parameters:

Name	Type	Description	Default
result	'FieldResult'	Solver output. Must carry point_sources (i.e. not a stub backend).	required
path	str | Path	Destination CSV path. Parent directories are created automatically.	required
start	tuple[float, float, float]	Path start point (x, y, z) in metres.	required
direction	tuple[float, float, float]	Direction vector. Will be normalised to unit length. Default (1, 0, 0).	(1.0, 0.0, 0.0)
distance	float	Path length in metres. Default 30.	30.0
n	int	Number of sample points along the path. Default 200.	200
frequency_indices	Sequence[int] | None	Iterable of integer indices into :attr:FieldResult.frequencies. None (default) uses every available frequency.	None

Returns:

Type	Description
Path	The path the file was written to.

Raises:

Type	Description
ValueError	If distance <= 0, n < 2, direction is the zero vector, or any frequency_indices value is out of range.

Source code in src/groundfield/io/csv.py

def save_potential_path_csv(
    result: "FieldResult",
    path: str | Path,
    *,
    start: tuple[float, float, float],
    direction: tuple[float, float, float] = (1.0, 0.0, 0.0),
    distance: float = 30.0,
    n: int = 200,
    frequency_indices: Sequence[int] | None = None,
) -> Path:
    """Sample the potential along a line and write to CSV.

    Builds an evenly spaced set of ``n`` field points starting at
    ``start`` and walking ``distance`` metres along the unit
    vector of ``direction``, evaluates
    :meth:`FieldResult.potential` at every selected frequency
    index, and writes the result to ``path``.

    Parameters
    ----------
    result
        Solver output. Must carry ``point_sources`` (i.e. not a
        stub backend).
    path
        Destination CSV path. Parent directories are created
        automatically.
    start
        Path start point ``(x, y, z)`` in metres.
    direction
        Direction vector. Will be normalised to unit length.
        Default ``(1, 0, 0)``.
    distance
        Path length in metres. Default 30.
    n
        Number of sample points along the path. Default 200.
    frequency_indices
        Iterable of integer indices into
        :attr:`FieldResult.frequencies`. ``None`` (default)
        uses every available frequency.

    Returns
    -------
    pathlib.Path
        The path the file was written to.

    Raises
    ------
    ValueError
        If ``distance <= 0``, ``n < 2``, ``direction`` is the zero
        vector, or any ``frequency_indices`` value is out of range.
    """
    if distance <= 0.0:
        raise ValueError(f"distance must be > 0, got {distance!r}.")
    if n < 2:
        raise ValueError(f"n must be >= 2, got {n!r}.")
    direction_arr = np.asarray(direction, dtype=float)
    norm = float(np.linalg.norm(direction_arr))
    if norm == 0.0:
        raise ValueError(f"direction must be non-zero, got {tuple(direction)}.")
    direction_arr = direction_arr / norm

    if frequency_indices is None:
        f_idx = list(range(len(result.frequencies)))
    else:
        f_idx = list(frequency_indices)
        for k in f_idx:
            if not (0 <= k < len(result.frequencies)):
                raise ValueError(
                    f"frequency index {k} out of range "
                    f"[0, {len(result.frequencies)})."
                )

    s = np.linspace(0.0, float(distance), int(n))
    pts = np.column_stack(
        [
            start[0] + s * direction_arr[0],
            start[1] + s * direction_arr[1],
            start[2] + s * direction_arr[2],
        ]
    )

    rows = []
    for k in f_idx:
        f_hz = float(result.frequencies[k])
        phi = result.potential(pts, frequency_index=k)
        for i in range(len(s)):
            rows.append(
                {
                    "s": float(s[i]),
                    "x": float(pts[i, 0]),
                    "y": float(pts[i, 1]),
                    "z": float(pts[i, 2]),
                    "frequency_Hz": f_hz,
                    "phi_re": float(phi[i].real),
                    "phi_im": float(phi[i].imag),
                    "abs_phi": float(abs(phi[i])),
                }
            )

    out = _ensure_path(path)
    pd.DataFrame(rows).to_csv(out, index=False, float_format="%.17g")
    return out

VTK exports — groundfield.io.vtk

Legacy ASCII VTK writers (no pyvista / vtk Python bindings required). Two file flavours:

Function	VTK dataset	Use case
export_geometry_vtk	POLYDATA	inspect electrodes + conductors in ParaView
export_field_vtk	STRUCTURED_POINTS	render the surface potential as a heatmap / iso

gf.export_geometry_vtk(world, "out/world.vtk")
gf.export_field_vtk(
    result, "out/phi_surface.vtk",
    extent=(-30, 30, -20, 20), z=0.0, n=(200, 150),
)

The geometry export carries an integer role cell-data scalar (0 = electrode, 1 = conductor) so colour-by-role works in ParaView without further configuration. The field export includes potential_re and potential_im so above-DC typical studies remain visible.

API reference — VTK

vtk

Legacy ASCII VTK exports for ParaView / VisIt / Mayavi.

This module writes two flavours of the legacy ASCII VTK file format (.vtk), keeping the dependency surface zero — no pyvista or vtk Python bindings are required:

• :func:export_geometry_vtk — POLYDATA with the electrode wires (rod, ring, strip, mesh, grid_mesh) and the conductor line segments. Useful for "drag the world into ParaView and rotate it" inspections of large typical networks.
• :func:export_field_vtk — STRUCTURED_POINTS with the soil surface (or any horizontal slice) sampled on a regular (N_x, N_y) grid. The potential is exported as a single scalar field; ParaView's contour / colour-map filters take it from there.

Why legacy ASCII VTK?

The legacy format is the smallest common denominator: every VTK reader since the 90s opens it, the on-disk layout is plain text (diff-able, version-controllable), and the writer is ~30 lines of pure-Python without any external library. For large research runs prefer pyvista once you need binary I/O and unstructured grids; for production-grade networks at notebook scale this format is fast enough.

References

• Schroeder, W., Martin, K., Lorensen, B. (2006). The Visualization Toolkit, 4th ed., Kitware. Section 19.5 (Legacy file format).
• VTK File Formats reference: https://vtk.org/wp-content/uploads/2015/04/file-formats.pdf

export_field_vtk

export_field_vtk(
    result: "FieldResult",
    path: str | Path,
    *,
    extent: tuple[float, float, float, float],
    z: float = 0.0,
    n: tuple[int, int] = (120, 120),
    frequency_index: int = 0
) -> Path

Sample the potential on a horizontal grid and write a VTK file.

Evaluates :meth:FieldResult.potential on a regular :math:N_x \times N_y grid in the plane :math:z = z_0 and writes a DATASET STRUCTURED_POINTS (a.k.a. uniform grid) with a single scalar field potential_re. The imaginary part is included as a second scalar potential_im for above-DC typical studies.

Parameters:

Name	Type	Description	Default
result	'FieldResult'	Solver output. Must carry point_sources (not a stub).	required
path	str | Path	Destination .vtk path.	required
extent	tuple[float, float, float, float]	Plot extent (x_min, x_max, y_min, y_max) in metres.	required
z	float	Slice depth in metres (default 0.0 — soil surface). Positive z is below ground.	0.0
n	tuple[int, int]	Grid resolution (n_x, n_y). Default (120, 120).	(120, 120)
frequency_index	int	Index into :attr:FieldResult.frequencies. Default 0.	0

Returns:

Type	Description
Path

Raises:

Type	Description
ValueError	On bad extent or non-positive grid sizes.

Source code in src/groundfield/io/vtk.py

def export_field_vtk(
    result: "FieldResult",
    path: str | Path,
    *,
    extent: tuple[float, float, float, float],
    z: float = 0.0,
    n: tuple[int, int] = (120, 120),
    frequency_index: int = 0,
) -> Path:
    """Sample the potential on a horizontal grid and write a VTK file.

    Evaluates :meth:`FieldResult.potential` on a regular
    :math:`N_x \\times N_y` grid in the plane :math:`z = z_0` and
    writes a ``DATASET STRUCTURED_POINTS`` (a.k.a. uniform grid)
    with a single scalar field ``potential_re``. The imaginary
    part is included as a second scalar ``potential_im`` for
    above-DC typical studies.

    Parameters
    ----------
    result
        Solver output. Must carry ``point_sources`` (not a stub).
    path
        Destination ``.vtk`` path.
    extent
        Plot extent ``(x_min, x_max, y_min, y_max)`` in metres.
    z
        Slice depth in metres (default ``0.0`` — soil surface).
        Positive ``z`` is below ground.
    n
        Grid resolution ``(n_x, n_y)``. Default ``(120, 120)``.
    frequency_index
        Index into :attr:`FieldResult.frequencies`. Default 0.

    Returns
    -------
    pathlib.Path

    Raises
    ------
    ValueError
        On bad extent or non-positive grid sizes.
    """
    x_min, x_max, y_min, y_max = extent
    if x_max <= x_min or y_max <= y_min:
        raise ValueError(
            f"extent must satisfy x_max > x_min and y_max > y_min, "
            f"got {extent!r}."
        )
    n_x, n_y = n
    if n_x < 2 or n_y < 2:
        raise ValueError(f"n must be >= (2, 2), got {n!r}.")

    xs = np.linspace(float(x_min), float(x_max), int(n_x))
    ys = np.linspace(float(y_min), float(y_max), int(n_y))
    X, Y = np.meshgrid(xs, ys, indexing="xy")
    pts = np.column_stack(
        [X.ravel(), Y.ravel(), np.full(X.size, float(z))]
    )
    phi = result.potential(pts, frequency_index=frequency_index)
    phi = phi.reshape(X.shape)

    dx = (x_max - x_min) / (n_x - 1)
    dy = (y_max - y_min) / (n_y - 1)

    out = _ensure_path(path)
    buf = StringIO()
    buf.write("# vtk DataFile Version 3.0\n")
    buf.write(
        f"groundfield potential @ z={z} m, "
        f"f={result.frequencies[frequency_index]} Hz\n"
    )
    buf.write("ASCII\n")
    buf.write("DATASET STRUCTURED_POINTS\n")
    buf.write(f"DIMENSIONS {n_x} {n_y} 1\n")
    buf.write(f"ORIGIN {x_min:.17g} {y_min:.17g} {z:.17g}\n")
    buf.write(f"SPACING {dx:.17g} {dy:.17g} 1\n")

    n_pts = n_x * n_y
    buf.write(f"POINT_DATA {n_pts}\n")
    buf.write("SCALARS potential_re float 1\n")
    buf.write("LOOKUP_TABLE default\n")
    # The order in STRUCTURED_POINTS is x fastest, then y, then z.
    # numpy meshgrid with indexing='xy' gives Y[i, j], X[i, j];
    # ravel() with C-order then walks j (x) before i (y) — exactly
    # what VTK expects.
    for v in phi.real.ravel():
        buf.write(f"{float(v):.17g}\n")

    buf.write("SCALARS potential_im float 1\n")
    buf.write("LOOKUP_TABLE default\n")
    for v in phi.imag.ravel():
        buf.write(f"{float(v):.17g}\n")

    out.write_text(buf.getvalue(), encoding="utf-8")
    return out

export_geometry_vtk

export_geometry_vtk(
    world: "World", path: str | Path
) -> Path

Export the world geometry as a legacy ASCII VTK PolyData file.

Writes electrodes (rods, rings, strips, mesh / grid_mesh perimeters) and conductors as 3-D polylines. The on-disk layout uses VTK's DATASET POLYDATA with a LINES block; cell data carries an integer role field (0 = electrode, 1 = conductor) so colour-by-role works directly in ParaView.

Parameters:

Name	Type	Description	Default
world	'World'	World to export. May be empty (the writer still produces a syntactically valid header-only file).	required
path	str | Path	Destination .vtk path. Parent directories are created automatically.	required

Returns:

Type	Description
Path	The path the file was written to.

Source code in src/groundfield/io/vtk.py

def export_geometry_vtk(world: "World", path: str | Path) -> Path:
    """Export the world geometry as a legacy ASCII VTK PolyData file.

    Writes electrodes (rods, rings, strips, mesh / grid_mesh
    perimeters) **and** conductors as 3-D polylines. The on-disk
    layout uses VTK's ``DATASET POLYDATA`` with a ``LINES`` block;
    cell data carries an integer ``role`` field (0 = electrode,
    1 = conductor) so colour-by-role works directly in ParaView.

    Parameters
    ----------
    world
        World to export. May be empty (the writer still produces
        a syntactically valid header-only file).
    path
        Destination ``.vtk`` path. Parent directories are created
        automatically.

    Returns
    -------
    pathlib.Path
        The path the file was written to.
    """
    polylines: list[np.ndarray] = []
    roles: list[int] = []

    for e in world.electrodes:
        for poly in _polylines_for_electrode(e):
            polylines.append(poly)
            roles.append(0)

    for c in world.conductors:
        seg = np.array([list(c.start), list(c.end)], dtype=float)
        polylines.append(seg)
        roles.append(1)

    out = _ensure_path(path)
    buf = StringIO()
    buf.write("# vtk DataFile Version 3.0\n")
    buf.write(f"groundfield geometry: {world.name}\n")
    buf.write("ASCII\n")
    buf.write("DATASET POLYDATA\n")

    if not polylines:
        buf.write("POINTS 0 float\n")
        out.write_text(buf.getvalue(), encoding="utf-8")
        return out

    n_points = sum(p.shape[0] for p in polylines)
    n_lines = len(polylines)
    n_line_data = n_lines + n_points  # one count + entries per line

    buf.write(f"POINTS {n_points} float\n")
    for poly in polylines:
        for x, y, z in poly:
            buf.write(f"{x:.17g} {y:.17g} {z:.17g}\n")

    buf.write(f"LINES {n_lines} {n_line_data}\n")
    offset = 0
    for poly in polylines:
        n = poly.shape[0]
        idx_list = " ".join(str(offset + k) for k in range(n))
        buf.write(f"{n} {idx_list}\n")
        offset += n

    # Cell data: role per polyline (0 = electrode, 1 = conductor).
    buf.write(f"CELL_DATA {n_lines}\n")
    buf.write("SCALARS role int 1\n")
    buf.write("LOOKUP_TABLE default\n")
    for r in roles:
        buf.write(f"{r}\n")

    out.write_text(buf.getvalue(), encoding="utf-8")
    return out

Example

JSON path (no groundinsight install needed)

import numpy as np
import groundfield as gf
from groundfield.postprocess.vector_fitting import rho_f_from_field_result
from groundfield.io.groundinsight import (
    save_bustype_json, load_bustype_json, to_bustype_dict,
)

# 1. Run a field study and produce a vector fit.
result = ...  # FieldResult from world.solve(engine)
fit = rho_f_from_field_result(result, electrode_name="g1", n_poles=3)

# 2. Export to a versioned JSON file.
save_bustype_json(
    fit,
    path="bus_type_substation.json",
    name="SubstationBus",
    description="Substation grounding grid (vector-fitted, typical ref).",
    system_type="Substation",
    voltage_level=20,
)

# 3. Load and inspect later (no groundinsight needed).
spec = load_bustype_json("bus_type_substation.json")
print(spec.name, spec.metadata["fit_method"])

Python API path (live groundinsight.BusType)

from groundfield.io.groundinsight import to_bustype, save_bustype_to_db

# Requires: pip install groundfield[groundinsight]
bus_type = to_bustype(
    fit,
    name="SubstationBus",
    description="Substation grounding grid (vector-fitted, typical ref).",
    system_type="Substation",
    voltage_level=20,
)
# bus_type is a live groundinsight.models.core_models.BusType instance
# and can be wired straight into a Network:
import groundinsight as gi
net = gi.create_network(name="reference", frequencies=[50.0, 250.0])
gi.create_bus(name="bus_substation", type=bus_type, network=net,
              specific_earth_resistance=100.0)

# Or persist to the groundinsight SQLite database in one call:
save_bustype_to_db(fit, db_path="grounding.db",
                   name="SubstationBus", system_type="Substation",
                   voltage_level=20)

evaluate_spec(spec, frequencies, rho) re-evaluates an exported formula at arbitrary \((f, \rho)\) points without round-tripping through groundinsight. Both evaluate_spec and the companion diagnostics helper fit_quality_summary(spec) are reachable directly as groundfield.evaluate_spec / groundfield.fit_quality_summary (top-level re-exports, listed in groundfield.__all__).

CSV writer column-name convention

The three CSV writers in groundfield.io.csv use a single naming convention: every complex quantity is written as the <symbol>_re / <symbol>_im / abs_<symbol> triple, where the symbol is the physical letter of the quantity (phi for potentials, I for currents, Z for impedances). The magnitude columns therefore differ between writers by design — abs_phi for the potential path, abs_I for the electrode table, abs_Z for the cluster table — and merging two tables on the magnitude column requires an explicit rename. The frozen column tuples POTENTIAL_PATH_COLUMNS, ELECTRODE_TABLE_REQUIRED_COLUMNS and CLUSTER_IMPEDANCE_REQUIRED_COLUMNS expose the schema for regression testing.

Errors

groundfield.io.groundinsight.evaluate_spec validates the :class:BusTypeSpec it receives before lambdifying the impedance_formula. Any malformed input — non-BusTypeSpec argument, empty formula, unparseable SymPy expression, unknown free symbol — raises a typed :class:EvaluateSpecError (subclass of :class:ValueError). Downstream groundinsight consumers can catch the typed class instead of substring-matching on str(exc). Existing except ValueError blocks keep working unchanged.

from groundfield.io.groundinsight import (
    EvaluateSpecError, evaluate_spec, load_bustype_json,
)

spec = load_bustype_json("my-bustype.json")
try:
    Z = evaluate_spec(spec, frequencies=[50.0, 1000.0], rho=100.0)
except EvaluateSpecError as exc:
    print("Spec rejected:", exc)

API reference

groundinsight

Export of reduced rho-f fits to groundinsight.

This module is the canonical bridge between the field-grade reference computation in groundfield and the reduced equivalent network model in groundinsight.

Two equally supported transports are provided:

• JSON file — neutral, language-agnostic schema versioned via schema_version. Produce with :func:to_bustype_dict / :func:save_bustype_json; read back with :func:load_bustype_json.
• Python API — :func:to_bustype returns a live :class:groundinsight.BusType Pydantic instance via a lazy import. groundinsight is therefore an optional dependency of groundfield (extras group [groundinsight]). The JSON path does not require groundinsight at all.

Mathematical background

groundinsight.BusType.impedance_formula is parsed with two free symbols, f (frequency in Hz) and rho (Bus.specific_earth_resistance), and the imaginary unit j. Both fit families exported here are projected onto that symbol set:

• :class:~groundfield.postprocess.rho_f_standard.RhoFStandardFit is already in the canonical (rho, f) form

$$ Z(\rho, f) \;=\; k_1\rho \;+\; (k_2 + j k_3)\,f \;+\; (k_4 + j k_5)\,f\,\rho. $$

• :class:~groundfield.postprocess.vector_fitting.VectorFitResult is a rational function of the Laplace variable \(s = j\,2\pi f\). The export substitutes \(s\to j\,2\pi f\) symbolically so the resulting expression is in f only (independent of rho); the underlying soil is recorded in the metadata block.

See also

docs/adr/0008-groundinsight-bridge.md — full design rationale, JSON schema, and validation programme.

SCHEMA_VERSION module-attribute

SCHEMA_VERSION = 1

Schema name and version of the JSON document produced by this module.

BusTypeSpec dataclass

BusTypeSpec(
    name: str,
    description: Optional[str],
    system_type: str,
    voltage_level: float,
    impedance_formula: str,
    samples: dict[str, list[float]],
    metadata: dict[str, Any],
)

Neutral, in-memory representation of an exported BusType.

The spec is what :func:to_bustype_dict returns under the hood and what :func:load_bustype_json reads back. It carries everything necessary to either build a groundinsight.BusType (via :func:to_bustype) or to write the on-disk JSON document.

Attributes:

Name	Type	Description
name, description, system_type, voltage_level		The four scalar fields that the groundinsight.BusType Pydantic model carries.
impedance_formula	str	The fitted Z(rho, f) expression as a SymPy-compatible string. Free symbols: f (Hz), rho (\(\Omega\,\mathrm{m}\)), and the imaginary unit j.
samples	dict[str, list[float]]	Dict with the parallel tabular representation. Keys "frequency_Hz", "rho_Ohm_m", "Z_real_Ohm", "Z_imag_Ohm" each map to a list of floats of equal length. Provided so a future tabular ingest path on the groundinsight side can be served immediately.
metadata	dict[str, Any]	Free-form metadata dict; populated by the exporters with the fit method, fit quality, fit-method-specific details (poles/residues for vector fits, \(k_1\dots k_5\) for the standard form), the source groundfield version and a UTC timestamp.

from_dict classmethod

from_dict(payload: dict[str, Any]) -> 'BusTypeSpec'

Build a :class:BusTypeSpec from a JSON-loaded dict.

Source code in src/groundfield/io/groundinsight.py

@classmethod
def from_dict(cls, payload: dict[str, Any]) -> "BusTypeSpec":
    """Build a :class:`BusTypeSpec` from a JSON-loaded ``dict``."""
    if payload.get("schema") != SCHEMA_NAME:
        raise ValueError(
            f"unexpected schema name: got {payload.get('schema')!r}, "
            f"expected {SCHEMA_NAME!r}"
        )
    version = int(payload.get("schema_version", 0))
    if version != SCHEMA_VERSION:
        raise ValueError(
            f"unsupported schema_version {version}; this groundfield "
            f"build only reads v{SCHEMA_VERSION}"
        )
    samples = payload.get("samples") or {}
    return cls(
        name=str(payload["name"]),
        description=payload.get("description"),
        system_type=str(payload["system_type"]),
        voltage_level=float(payload["voltage_level"]),
        impedance_formula=str(payload["impedance_formula"]),
        samples={
            "frequency_Hz": list(samples.get("frequency_Hz", [])),
            "rho_Ohm_m": list(samples.get("rho_Ohm_m", [])),
            "Z_real_Ohm": list(samples.get("Z_real_Ohm", [])),
            "Z_imag_Ohm": list(samples.get("Z_imag_Ohm", [])),
        },
        metadata=dict(payload.get("metadata", {})),
    )

to_dict

to_dict() -> dict[str, Any]

Return a JSON-ready dict matching schema v1.

Source code in src/groundfield/io/groundinsight.py

def to_dict(self) -> dict[str, Any]:
    """Return a JSON-ready ``dict`` matching schema v1."""
    return {
        "schema": SCHEMA_NAME,
        "schema_version": SCHEMA_VERSION,
        "name": self.name,
        "description": self.description,
        "system_type": self.system_type,
        "voltage_level": float(self.voltage_level),
        "impedance_formula": self.impedance_formula,
        "samples": {
            "frequency_Hz": [float(x) for x in self.samples["frequency_Hz"]],
            "rho_Ohm_m": [float(x) for x in self.samples["rho_Ohm_m"]],
            "Z_real_Ohm": [float(x) for x in self.samples["Z_real_Ohm"]],
            "Z_imag_Ohm": [float(x) for x in self.samples["Z_imag_Ohm"]],
        },
        "metadata": dict(self.metadata),
    }

EvaluateSpecError

Bases: ValueError

Raised when :func:evaluate_spec rejects a :class:BusTypeSpec.

Subclass of :class:ValueError so legacy except ValueError blocks keep working unchanged; downstream consumers (notably the groundinsight consumer of the bridge format) can now catch the typed exception instead of matching against str(exc) substrings.

evaluate_spec

evaluate_spec(
    spec: BusTypeSpec,
    frequencies: Sequence[float],
    rho: float,
) -> np.ndarray

Evaluate a :class:BusTypeSpec at arbitrary frequencies.

Re-evaluates the stored impedance_formula with the provided rho and f values, using the same SymPy-based path groundinsight.utils.impedance_calculator follows. Useful for cheap sanity checks that do not require groundinsight to be installed.

Parameters:

Name	Type	Description	Default
spec	BusTypeSpec	Loaded spec (e.g. from :func:load_bustype_json).	required
frequencies	Sequence[float]	Frequencies in Hz.	required
rho	float	Soil resistivity in \(\Omega\,\mathrm{m}\). Ignored if the formula does not depend on rho (vector-fit case).	required

Returns:

Type	Description
ndarray	Complex-valued \(Z\) at every requested frequency.

Raises:

Type

Description

EvaluateSpecError

Subclass of :class:ValueError; raised when spec is not a :class:BusTypeSpec instance, impedance_formula is empty / missing on the spec, the formula cannot be parsed by SymPy, or the formula contains unrecognised free symbols (anything other than f, rho and j). The exception is typed so downstream consumers (notably groundinsight) can catch it without substring matching on the message. Existing except ValueError blocks keep working because :class:EvaluateSpecError inherits from :class:ValueError.

Source code in src/groundfield/io/groundinsight.py

def evaluate_spec(
    spec: BusTypeSpec,
    frequencies: Sequence[float],
    rho: float,
) -> np.ndarray:
    """Evaluate a :class:`BusTypeSpec` at arbitrary frequencies.

    Re-evaluates the stored ``impedance_formula`` with the provided
    ``rho`` and ``f`` values, using the same SymPy-based path
    ``groundinsight.utils.impedance_calculator`` follows. Useful for
    cheap sanity checks that do not require ``groundinsight`` to be
    installed.

    Parameters
    ----------
    spec
        Loaded spec (e.g. from :func:`load_bustype_json`).
    frequencies
        Frequencies in Hz.
    rho
        Soil resistivity in $\\Omega\\,\\mathrm{m}$. Ignored if the
        formula does not depend on ``rho`` (vector-fit case).

    Returns
    -------
    np.ndarray
        Complex-valued $Z$ at every requested frequency.

    Raises
    ------
    EvaluateSpecError
        Subclass of :class:`ValueError`; raised when ``spec`` is not a
        :class:`BusTypeSpec` instance, ``impedance_formula`` is empty
        / missing on the spec, the formula cannot be parsed by SymPy,
        or the formula contains unrecognised free symbols (anything
        other than ``f``, ``rho`` and ``j``). The exception is typed
        so downstream consumers (notably ``groundinsight``) can catch
        it without substring matching on the message. Existing
        ``except ValueError`` blocks keep working because
        :class:`EvaluateSpecError` inherits from :class:`ValueError`.
    """
    import sympy as sp

    # Defensive entry-point validation. Hand-rolled BusTypeSpec-like
    # dicts (e.g. constructed in a CLI helper from a YAML campaign
    # config) routinely miss the impedance_formula; the previous
    # implementation produced an opaque KeyError / AttributeError
    # several frames deep in sympy. Surface the real problem here.
    if not isinstance(spec, BusTypeSpec):
        raise EvaluateSpecError(
            "evaluate_spec: expected a BusTypeSpec instance, got "
            f"{type(spec).__name__}. Build one via load_bustype_json "
            "or BusTypeSpec.from_dict before calling evaluate_spec."
        )
    formula = getattr(spec, "impedance_formula", None)
    if formula is None or not str(formula).strip():
        raise EvaluateSpecError(
            "evaluate_spec: spec.impedance_formula is missing or "
            "empty. A BusTypeSpec must carry a non-empty "
            "impedance_formula string with free symbols in "
            "{f, rho, j}."
        )

    f_sym = sp.Symbol("f", real=True, positive=True)
    rho_sym = sp.Symbol("rho", real=True, positive=True)
    j_sym = sp.Symbol("j")
    try:
        parsed = sp.sympify(str(formula))
    except (sp.SympifyError, SyntaxError, TypeError) as exc:
        raise EvaluateSpecError(
            "evaluate_spec: could not parse spec.impedance_formula "
            f"({formula!r}) as a SymPy expression: {exc}. Expected "
            "free symbols are f (Hz), rho (Ohm·m) and j (imag. unit)."
        ) from exc
    expr = parsed.subs(j_sym, sp.I)
    allowed_symbols = {f_sym, rho_sym}
    unknown = {
        s for s in expr.free_symbols
        if s.name not in {"f", "rho"}
    }
    if unknown:
        unknown_names = sorted(s.name for s in unknown)
        raise EvaluateSpecError(
            "evaluate_spec: spec.impedance_formula contains "
            f"unrecognised free symbols {unknown_names!r}. Only "
            "f (frequency in Hz), rho (resistivity in Ohm·m) and "
            "the imaginary unit j are allowed."
        )
    func = sp.lambdify((f_sym, rho_sym), expr, modules=["numpy"])
    f_arr = np.asarray(frequencies, dtype=float)
    out = func(f_arr, float(rho))
    out = np.asarray(out, dtype=complex)
    if out.ndim == 0:
        out = np.full(f_arr.shape, complex(out), dtype=complex)
    return out

fit_quality_summary

fit_quality_summary(spec: BusTypeSpec) -> str

One-line human-readable summary of fit quality / kind.

Source code in src/groundfield/io/groundinsight.py

def fit_quality_summary(spec: BusTypeSpec) -> str:
    """One-line human-readable summary of fit quality / kind."""
    md = spec.metadata
    method = md.get("fit_method", "unknown")
    fq = md.get("fit_quality", {})
    rms = fq.get("rms_error_Ohm", math.nan)
    rel = fq.get("rms_relative")
    if rel is not None:
        return (
            f"BusTypeSpec(name={spec.name!r}, method={method}, "
            f"rms={rms:.3e} Ω, rms_rel={rel:.2%})"
        )
    return (
        f"BusTypeSpec(name={spec.name!r}, method={method}, "
        f"rms={rms:.3e} Ω)"
    )

load_bustype_json

load_bustype_json(path: Union[str, Path]) -> BusTypeSpec

Load a BusType JSON document into a :class:BusTypeSpec.

Validates the schema name and version. Forward-compatibility for future schema versions is the explicit responsibility of this function: when schema_version differs from :data:SCHEMA_VERSION, dispatch to the appropriate loader. Today only v1 exists.

Source code in src/groundfield/io/groundinsight.py

def load_bustype_json(path: Union[str, Path]) -> BusTypeSpec:
    """Load a ``BusType`` JSON document into a :class:`BusTypeSpec`.

    Validates the schema name and version. Forward-compatibility for
    future schema versions is the explicit responsibility of this
    function: when ``schema_version`` differs from
    :data:`SCHEMA_VERSION`, dispatch to the appropriate loader. Today
    only v1 exists.
    """
    p = Path(path)
    with p.open("r", encoding="utf-8") as fh:
        payload = json.load(fh)
    return BusTypeSpec.from_dict(payload)

save_bustype_json

save_bustype_json(
    fit: FitLike,
    path: Union[str, Path],
    *,
    name: str,
    system_type: str,
    voltage_level: float,
    description: Optional[str] = None,
    decimals: int = 12,
    rho_at_fit: Optional[float] = None,
    electrode_name: Optional[str] = None,
    soil_summary: Optional[str] = None,
    indent: int = 2
) -> Path

Write the schema-v1 JSON file to disk.

Returns:

Type	Description
Path	The path the file was actually written to (as a :class:pathlib.Path), for chaining.

Source code in src/groundfield/io/groundinsight.py

def save_bustype_json(
    fit: FitLike,
    path: Union[str, Path],
    *,
    name: str,
    system_type: str,
    voltage_level: float,
    description: Optional[str] = None,
    decimals: int = 12,
    rho_at_fit: Optional[float] = None,
    electrode_name: Optional[str] = None,
    soil_summary: Optional[str] = None,
    indent: int = 2,
) -> Path:
    """Write the schema-v1 JSON file to disk.

    Returns
    -------
    pathlib.Path
        The path the file was actually written to (as a
        :class:`pathlib.Path`), for chaining.
    """
    payload = to_bustype_dict(
        fit,
        name=name,
        system_type=system_type,
        voltage_level=voltage_level,
        description=description,
        decimals=decimals,
        rho_at_fit=rho_at_fit,
        electrode_name=electrode_name,
        soil_summary=soil_summary,
    )
    p = Path(path)
    p.parent.mkdir(parents=True, exist_ok=True)
    with p.open("w", encoding="utf-8") as fh:
        json.dump(payload, fh, indent=indent, sort_keys=False)
        fh.write("\n")
    return p

save_bustype_to_db

save_bustype_to_db(
    fit: FitLike,
    *,
    name: str,
    system_type: str,
    voltage_level: float,
    description: Optional[str] = None,
    decimals: int = 12,
    rho_at_fit: Optional[float] = None,
    electrode_name: Optional[str] = None,
    soil_summary: Optional[str] = None,
    overwrite: bool = False
) -> None

Convenience wrapper: build a BusType and save it to the groundinsight SQLite store opened by gi.start_dbsession().

Requires groundinsight to be installed and a session to be active. Raises ImportError if the package is missing and a plain RuntimeError (from groundinsight) if no session is open.

Source code in src/groundfield/io/groundinsight.py

def save_bustype_to_db(
    fit: FitLike,
    *,
    name: str,
    system_type: str,
    voltage_level: float,
    description: Optional[str] = None,
    decimals: int = 12,
    rho_at_fit: Optional[float] = None,
    electrode_name: Optional[str] = None,
    soil_summary: Optional[str] = None,
    overwrite: bool = False,
) -> None:
    """Convenience wrapper: build a BusType and save it to the
    ``groundinsight`` SQLite store opened by ``gi.start_dbsession()``.

    Requires ``groundinsight`` to be installed *and* a session to be
    active. Raises ``ImportError`` if the package is missing and a
    plain ``RuntimeError`` (from ``groundinsight``) if no session
    is open.
    """
    bustype = to_bustype(
        fit,
        name=name,
        system_type=system_type,
        voltage_level=voltage_level,
        description=description,
        decimals=decimals,
        rho_at_fit=rho_at_fit,
        electrode_name=electrode_name,
        soil_summary=soil_summary,
    )
    import groundinsight as gi  # already verified by to_bustype above

    gi.save_bustype_to_db(bustype, overwrite=overwrite)

to_bustype

to_bustype(
    fit: FitLike,
    *,
    name: str,
    system_type: str,
    voltage_level: float,
    description: Optional[str] = None,
    decimals: int = 12,
    rho_at_fit: Optional[float] = None,
    electrode_name: Optional[str] = None,
    soil_summary: Optional[str] = None
)

Build a live :class:groundinsight.BusType from a fit.

Performs the same conversion as :func:to_bustype_dict, but returns a fully-validated BusType Pydantic instance instead of a JSON-ready dict. groundinsight is imported lazily; a missing install raises an :class:ImportError with a clear pointer to the optional install.

The metadata block is not carried over into the returned BusType (the BusType schema has no metadata field) — use :func:to_bustype_dict or :func:save_bustype_json if metadata must be persisted.

Source code in src/groundfield/io/groundinsight.py

def to_bustype(
    fit: FitLike,
    *,
    name: str,
    system_type: str,
    voltage_level: float,
    description: Optional[str] = None,
    decimals: int = 12,
    rho_at_fit: Optional[float] = None,
    electrode_name: Optional[str] = None,
    soil_summary: Optional[str] = None,
):
    """Build a live :class:`groundinsight.BusType` from a fit.

    Performs the same conversion as :func:`to_bustype_dict`, but
    returns a fully-validated ``BusType`` Pydantic instance instead
    of a JSON-ready ``dict``. ``groundinsight`` is imported lazily;
    a missing install raises an :class:`ImportError` with a clear
    pointer to the optional install.

    The metadata block is **not** carried over into the returned
    ``BusType`` (the ``BusType`` schema has no metadata field) — use
    :func:`to_bustype_dict` or :func:`save_bustype_json` if metadata
    must be persisted.
    """
    BusType = _import_groundinsight()
    spec = _build_spec(
        fit,
        name=name,
        system_type=system_type,
        voltage_level=voltage_level,
        description=description,
        decimals=decimals,
        rho_at_fit=rho_at_fit,
        electrode_name=electrode_name,
        soil_summary=soil_summary,
    )
    return BusType(
        name=spec.name,
        description=spec.description,
        system_type=spec.system_type,
        voltage_level=spec.voltage_level,
        impedance_formula=spec.impedance_formula,
    )

to_bustype_dict

to_bustype_dict(
    fit: FitLike,
    *,
    name: str,
    system_type: str,
    voltage_level: float,
    description: Optional[str] = None,
    decimals: int = 12,
    rho_at_fit: Optional[float] = None,
    electrode_name: Optional[str] = None,
    soil_summary: Optional[str] = None
) -> dict[str, Any]

Convert a fit into the JSON-ready schema-v1 dict.

Parameters:

Name	Type	Description	Default
fit	FitLike	A :class:~groundfield.postprocess.rho_f_standard.RhoFStandardFit or :class:~groundfield.postprocess.vector_fitting.VectorFitResult.	required
name	str	Match the four scalar fields of groundinsight.BusType.	required
system_type	str	Match the four scalar fields of groundinsight.BusType.	required
voltage_level	str	Match the four scalar fields of groundinsight.BusType.	required
description	str	Match the four scalar fields of groundinsight.BusType.	required
decimals	int	Number of significant digits to keep per coefficient when rendering the SymPy formula. Default 12 — chosen so that the round-trip through groundinsight.compute_impedance reproduces fit.evaluate(...) to better than \(10^{-9}\) relative on typical impedance magnitudes. Reduce to 6 if you want a shorter, human-readable formula at the cost of ~\(10^{-7}\) round-trip drift.	12
rho_at_fit	Optional[float]	Soil resistivity at which a :class:VectorFitResult was produced, in \(\Omega\,\mathrm{m}\). Required for vector fits; ignored for standard-form fits.	None
electrode_name	Optional[str]	Free-form metadata entries; recommended for traceability.	None
soil_summary	Optional[str]	Free-form metadata entries; recommended for traceability.	None

Returns:

Type	Description
dict	JSON-ready dict matching schema_version = 1 of groundfield.bustype.

Source code in src/groundfield/io/groundinsight.py

def to_bustype_dict(
    fit: FitLike,
    *,
    name: str,
    system_type: str,
    voltage_level: float,
    description: Optional[str] = None,
    decimals: int = 12,
    rho_at_fit: Optional[float] = None,
    electrode_name: Optional[str] = None,
    soil_summary: Optional[str] = None,
) -> dict[str, Any]:
    """Convert a fit into the JSON-ready schema-v1 ``dict``.

    Parameters
    ----------
    fit
        A :class:`~groundfield.postprocess.rho_f_standard.RhoFStandardFit`
        or :class:`~groundfield.postprocess.vector_fitting.VectorFitResult`.
    name, system_type, voltage_level, description
        Match the four scalar fields of ``groundinsight.BusType``.
    decimals
        Number of significant digits to keep per coefficient when
        rendering the SymPy formula. Default 12 — chosen so that the
        round-trip through ``groundinsight.compute_impedance``
        reproduces ``fit.evaluate(...)`` to better than $10^{-9}$
        relative on typical impedance magnitudes. Reduce to 6 if
        you want a shorter, human-readable formula at the cost of
        ~$10^{-7}$ round-trip drift.
    rho_at_fit
        Soil resistivity at which a :class:`VectorFitResult` was
        produced, in $\\Omega\\,\\mathrm{m}$. Required for vector fits;
        ignored for standard-form fits.
    electrode_name, soil_summary
        Free-form metadata entries; recommended for traceability.

    Returns
    -------
    dict
        JSON-ready ``dict`` matching ``schema_version = 1`` of
        ``groundfield.bustype``.
    """
    spec = _build_spec(
        fit,
        name=name,
        system_type=system_type,
        voltage_level=voltage_level,
        description=description,
        decimals=decimals,
        rho_at_fit=rho_at_fit,
        electrode_name=electrode_name,
        soil_summary=soil_summary,
    )
    return spec.to_dict()