Validation (cross-engine)

The :mod:groundfield.validation module contains :func:compare_engines, the cross-engine consistency check that underpins ADR-0001. Given two :class:Engine instances and a :class:World, it solves both and reports a structured :class:EngineComparison summarising the agreement of cluster impedances, electrode currents and surface potentials.

This is the post-solve counterpart of :mod:groundfield.diagnostics (pre-solve structural checks).

Frequency-order behaviour

Since 0.5.0 :class:Engine.frequencies is order-preserving. A non-monotonic frequency list raises a dedicated :class:EngineFrequencyOrderWarning (subclass of :class:UserWarning) so a single warnings.simplefilter("once", EngineFrequencyOrderWarning) will collapse a compare_engines 4 × 4 matrix to one emission. The opt-in :meth:Engine.with_frequencies(*, preserve_order=True) constructor silences the warning explicitly.

API reference

validation

Cross-engine comparison for self-validation.

This module provides :func:compare_engines — a small convenience helper that runs the same :class:World through several :class:Engine configurations and checks the consistency of the results. It implements ADR-0001 (docs/adr/0001-two-layer-method.md): two engines side by side, validating each other.

Usage

import groundfield as gf world = gf.create_world(soil=gf.HomogeneousSoil(resistivity=100.0)) gf.create_electrode(world, "rod", name="g1", ... position=(0, 0, 0.0), length=1.5) gf.create_source(world, attached_to="g1", magnitude=1.0) report = gf.compare_engines( ... world, ... engines={ ... "image": gf.create_engine(backend="image", segment_length=0.05), ... # "image_2layer": gf.create_engine(backend="image_2layer", ...), ... }, ... rel_tolerance=0.05, ... ) report.is_consistent True

EngineComparison dataclass

EngineComparison(
    results: dict[str, "FieldResult"],
    rel_tolerance: float,
    cluster_impedance_table: dict[
        str, dict[str, float]
    ] = dict(),
    deviations: dict[str, float] = dict(),
    is_consistent: bool = False,
    notes: list[str] = list(),
)

Outcome of a cross-engine comparison.

Attributes:

Name	Type	Description
results	dict[str, 'FieldResult']	Mapping engine_label -> FieldResult.
rel_tolerance	float	Relative tolerance the results were checked against.
cluster_impedance_table	dict[str, dict[str, float]]	{cluster_root: {engine_label: Z(f0).real}} where f0 is the first frequency in :attr:FieldResult.frequencies.
deviations	dict[str, float]	{cluster_root: max relative deviation}.
is_consistent	bool	True if and only if max(deviations) <= rel_tolerance and no engine returned a stub result.
notes	list[str]	Diagnostic strings (e.g. "stub backend", "frequency lists do not match").

summary

summary() -> str

Return a line-oriented textual summary.

Source code in src/groundfield/validation.py

def summary(self) -> str:
    """Return a line-oriented textual summary."""
    labels = list(self.results.keys())
    lines = [
        f"Cross-engine comparison, tolerance {self.rel_tolerance*100:.1f} %",
        f"Engines: {', '.join(labels)}",
        "",
        "Cluster impedances Re{Z(f0)} in Ω:",
    ]
    for cluster, by_engine in self.cluster_impedance_table.items():
        row = "  " + cluster.ljust(12) + " "
        row += " | ".join(f"{lbl}: {by_engine.get(lbl, float('nan')):.3f}"
                          for lbl in labels)
        row += f"  Δ_max = {self.deviations.get(cluster, 0)*100:5.2f} %"
        lines.append(row)
    lines.append("")
    lines.append(
        "Consistent." if self.is_consistent
        else "INCONSISTENT — see notes."
    )
    if self.notes:
        lines.append("")
        lines.append("Notes:")
        lines.extend(f"  - {n}" for n in self.notes)
    return "\n".join(lines)

compare_engines

compare_engines(
    world: "World",
    engines: dict[str, "Engine"],
    *,
    rel_tolerance: float = 0.05,
    sample_points: np.ndarray | None = None
) -> EngineComparison

Run world through every engine and compare the results.

Parameters:

Name	Type	Description	Default
world	'World'	World to evaluate. Not modified by this function.	required
engines	dict[str, 'Engine']	Mapping label -> Engine. Must contain at least two entries.	required
rel_tolerance	float	Maximum allowed relative deviation of the cluster impedances (default 5 %). The tolerance applies to every cluster present in the result.	0.05
sample_points	ndarray | None	Optional. Array of shape (M, 3): additional field points whose potentials will be compared. Deviations are reported in :attr:EngineComparison.notes.	None

Returns:

Type	Description
EngineComparison	Structured report (see :class:EngineComparison).

Notes

The check uses the real part of the cluster impedance at the first frequency. Clusters with Σ I = 0 (purely passive observers) have an undefined impedance and are skipped; the skip is recorded in notes.

Source code in src/groundfield/validation.py

def compare_engines(
    world: "World",
    engines: dict[str, "Engine"],
    *,
    rel_tolerance: float = 0.05,
    sample_points: np.ndarray | None = None,
) -> EngineComparison:
    """Run ``world`` through every engine and compare the results.

    Parameters
    ----------
    world
        World to evaluate. **Not modified** by this function.
    engines
        Mapping ``label -> Engine``. Must contain at least two entries.
    rel_tolerance
        Maximum allowed relative deviation of the cluster impedances
        (default 5 %). The tolerance applies to every cluster present
        in the result.
    sample_points
        Optional. Array of shape ``(M, 3)``: additional field points
        whose potentials will be compared. Deviations are reported in
        :attr:`EngineComparison.notes`.

    Returns
    -------
    EngineComparison
        Structured report (see :class:`EngineComparison`).

    Notes
    -----
    The check uses the **real part** of the cluster impedance at the
    first frequency. Clusters with ``Σ I = 0`` (purely passive
    observers) have an undefined impedance and are skipped; the skip
    is recorded in ``notes``.
    """
    if len(engines) < 2:
        raise ValueError("compare_engines requires at least 2 engines.")

    results: dict[str, FieldResult] = {
        label: engine.solve(world) for label, engine in engines.items()
    }

    cmp = EngineComparison(results=results, rel_tolerance=rel_tolerance)

    # Detect stub backends
    for label, res in results.items():
        if res.metadata.get("stub"):
            cmp.notes.append(
                f"Engine '{label}' returned a stub result "
                f"(metadata['stub']=True). Comparison not meaningful."
            )

    # Frequency lists must match
    freq_first = next(iter(results.values())).frequencies
    for label, res in results.items():
        if res.frequencies != freq_first:
            cmp.notes.append(
                f"Engine '{label}' has a different frequency list "
                f"({res.frequencies} vs. {freq_first}). Comparison "
                "uses the first frequency only."
            )

    # Cluster sets must be identical
    cluster_keys_first = {tuple(sorted(v))
                          for v in next(iter(results.values())).clusters.values()}
    for label, res in results.items():
        keys = {tuple(sorted(v)) for v in res.clusters.values()}
        if keys != cluster_keys_first:
            cmp.notes.append(
                f"Engine '{label}' reports a different cluster structure. "
                "Cluster comparison skipped."
            )
            cmp.is_consistent = False
            return cmp

    # Cluster impedances (one representative per cluster)
    representative: dict[str, str] = {}
    for label, res in results.items():
        for ename, members in res.clusters.items():
            root = sorted(members)[0]
            representative.setdefault(root, root)

    max_dev = 0.0
    for cluster_root in representative.values():
        per_engine: dict[str, float] = {}
        for label, res in results.items():
            try:
                Z = res.cluster_impedance(cluster_root)[0]
            except (KeyError, IndexError):
                continue
            if not np.isfinite(Z.real):
                continue
            per_engine[label] = float(Z.real)
        if len(per_engine) < 2:
            cmp.notes.append(
                f"Cluster '{cluster_root}': Σ I = 0 or Z undefined — "
                "skipped."
            )
            continue
        cmp.cluster_impedance_table[cluster_root] = per_engine
        zs = np.array(list(per_engine.values()))
        zmean = zs.mean()
        if zmean == 0.0:
            continue
        dev = float(np.max(np.abs(zs - zmean)) / abs(zmean))
        cmp.deviations[cluster_root] = dev
        max_dev = max(max_dev, dev)

    # Optional point-sample for the potential
    if sample_points is not None and len(results) >= 2:
        try:
            phi_table = {
                lbl: res.potential(sample_points).real
                for lbl, res in results.items()
            }
            phis = np.stack(list(phi_table.values()))
            mean = phis.mean(axis=0)
            with np.errstate(divide="ignore", invalid="ignore"):
                rel = np.where(np.abs(mean) > 0,
                               np.max(np.abs(phis - mean), axis=0) / np.abs(mean),
                               0.0)
            sample_max = float(rel.max()) if rel.size else 0.0
            cmp.notes.append(
                f"Potential point-sample at {len(sample_points)} points: "
                f"max relative deviation = {sample_max*100:.2f} %."
            )
            max_dev = max(max_dev, sample_max)
        except RuntimeError as e:
            cmp.notes.append(f"Potential sample not evaluated: {e}")

    cmp.is_consistent = (max_dev <= rel_tolerance) and not any(
        n.startswith("Engine '") and "stub result" in n for n in cmp.notes
    )
    return cmp