Skip to content

Core models

Pydantic v2 data classes describing the physical network elements, the configured faults and sources, the per-frequency results and the ComplexNumber helper used throughout the package.

Physical / modelling context

groundinsight represents a grounding network as a labelled, undirected graph. The model layer owns four kinds of objects:

  • TypesBusType, BranchType — carry the SymPy formula strings that are compiled to vectorised callables and evaluated per (f, rho, l) triple. Optional lumped RLC formulas (R_formula, L_formula, C_formula, R_self_formula, L_self_formula, C_self_formula, R_mutual_formula, M_mutual_formula) parameterise the state-space transient solver in groundinsight.simulation.transient.
  • InstancesBus, Branch — carry concrete numerical values (specific_earth_resistance, length, parallel_coefficient) plus the per-frequency impedance dict \(Z(f) \in \mathbb{C}\) that is built from the formulas at network-build time.
  • ExcitationSource (current or voltage source per bus) and Fault (which bus, which scaling per frequency).
  • ResultsResult, ResultBus, ResultBranch, ResultReductionFactor, ResultGroundingImpedance — the outcome of run_fault. Both per-frequency components and the RMS over all frequencies (computed as \(\sqrt{\sum_f |X(f)|^2}\)) are stored.

ComplexNumber is a small Pydantic wrapper over the native complex type. It exists because complex is not natively JSON-serialisable; the wrapper exposes overloaded arithmetic and serialises as a {"real": ..., "imag": ...} dict.

Example

import groundinsight as gi
from groundinsight.models.core_models import (
    Bus, BusType, Branch, BranchType, Source, Fault,
    Network, ComplexNumber,
)

# Build a model directly (notebook style)
bt = BusType(name="GroundRod", system_type="Substation",
             voltage_level=20.0,
             impedance_formula="rho/(2*3.14159*1.5)*(1 + j*0.01*f)")
brt = BranchType(name="ShieldCable", grounding_conductor=True,
                 self_impedance_formula="(0.2 + j*0.4*f/50)*l",
                 mutual_impedance_formula="(0.0 + j*0.4*f/50)*l")

net = Network(name="demo", frequencies=[50.0, 150.0])
b1 = Bus(name="b1", type=bt, impedance={},
         specific_earth_resistance=100.0)
b2 = Bus(name="b2", type=bt, impedance={},
         specific_earth_resistance=100.0)
ln = Branch(name="ln", type=brt, from_bus="b1", to_bus="b2",
            length=2.0, self_impedance={}, mutual_impedance={})
src = Source(name="s1", bus="b1", values={50.0: 1.0, 150.0: 0.05})
flt = Fault(name="f1", bus="b2",
            scalings={50.0: 1.0, 150.0: 0.05})

# JSON round-trip — ComplexNumber serialises as {real, imag}
payload = net.model_dump_json(indent=2)
restored = Network.model_validate_json(payload)

Polars accessors net.res_buses(), net.res_branches() and net.res_all_impedances() produce DataFrames suitable for plotting and reporting.

Active subset / cache invalidation

Bus.active and Branch.active are plain Pydantic fields that flip an instance in or out of the topology used by PathFinder. Two callers therefore matter when a flag is flipped in-place after define_paths() has already populated network.paths:

  • network.paths itself, which mirrors the previously active topology.
  • The module-level _GRAPH_CACHE / _FIND_PATHS_CACHE in groundinsight.pathfinder, which mirror the same topology fingerprint for fast re-use.

Network.invalidate_paths() is the explicit hook for that case:

# Flip a branch out of service mid-notebook and rebuild paths.
net.branches["LN_main"].active = False
net.invalidate_paths()  # drops self.paths + this network's cache
gi.create_paths(net)    # rebuilds with the new topology

The invalidation is scoped to the calling Network instance: cache entries belonging to other live networks in the same process are preserved, so dashboards iterating over a set of feeders do not pay a global cache eviction every time a single network mutates.

Since 0.5.0 the invalidation is also an atomic rebind (self.paths = {} instead of self.paths.clear()), so an external snapshot saved = dict(network.paths) taken before the call keeps its entries:

saved = dict(network.paths)        # external snapshot
network.invalidate_paths()
# saved still holds the previously-enumerated paths.

Frequency validation and order warning

Network.frequencies is validated at construction time:

  • Empty / nan / inf / strictly-negative inputs are rejected with a clear ValueError. DC (f = 0) is permitted because the FFT transient solver in groundinsight.simulation.transient uses the zero-frequency bin to carry the steady-state offset.
  • Duplicate frequencies are rejected — the same f twice in the list silently doubled the work in solve_network and doubled the amplitude of the corresponding spectral bin in the FFT transient solver.
  • Non-strictly-monotone-increasing inputs accept the order but emit a NetworkFrequencyOrderWarning(UserWarning) (added in 0.5.0). The FFT transient solver maps spectral bins by position in Network.frequencies, so a shuffled or descending list is almost always a user error. Mirrors groundfield.solver.engine.EngineFrequencyOrderWarning so the three earthing-platform packages share one convention.
import warnings
import groundinsight as gi

with warnings.catch_warnings():
    warnings.simplefilter("error", gi.NetworkFrequencyOrderWarning)
    # Raises instead of just warning — use during validation.
    gi.create_network(name="net", frequencies=[100.0, 50.0])

Top-level set_active_fault factory

The keep_results= keyword on Network.set_active_fault is also reachable via the top-level factory wrapper:

import groundinsight as gi

# Re-plot the previously cached Result without recomputing it.
gi.set_active_fault(net, "F1", keep_results=True)

Field-level validation guards are documented inline (frequency duplicate / NaN / negative rejection on Network.frequencies, int-vs-float key coercion on Fault.scalings, …). See the mkdocstrings dump below for the authoritative list.

API reference

core_models

Bus

Bases: BaseModel

Represents a grounding bus within the network.

Attributes:

Name Type Description
name str

The name of the bus.

description (str, optional)

A brief description of the bus.

type BusType

The type of the bus.

impedance dict of float to ComplexNumber

Mapping of frequency to grounding impedance values.

specific_earth_resistance float

The specific earth resistance associated with the bus (Ohm * m).

active bool

Whether the bus participates in the solve. Inactive buses are removed from the admittance matrix; paths traversing them are dropped. Defaults to True. Used by the outage / what-if machinery in :mod:groundinsight.simulation.outage.

R dict of float to float, optional

Evaluated lumped resistance per frequency in Ohm. Populated only if type.R_formula is set. Consumed by the transient solvers.

L dict of float to float, optional

Evaluated lumped inductance per frequency in Henry. Populated only if type.L_formula is set.

C dict of float to float, optional

Evaluated lumped capacitance to remote earth per frequency in Farad. Populated only if type.C_formula is set.

calculate_impedance

calculate_impedance(frequencies: List[float])

Calculates impedance and -- if specified by the type -- the lumped RLC parameters for each frequency.

impedance is always recomputed from type.impedance_formula. Each of R, L and C is recomputed only if the matching formula is set on the type; otherwise the attribute is left as None. Utilizes the external impedance_calculator to avoid storing non-pickleable functions.

Parameters:

Name Type Description Default
frequencies List[float]

A list of frequencies at which to evaluate the formulas.

required
Source code in src/groundinsight/models/core_models.py
def calculate_impedance(self, frequencies: List[float]):
    """
    Calculates impedance and -- if specified by the type -- the lumped
    RLC parameters for each frequency.

    ``impedance`` is always recomputed from ``type.impedance_formula``.
    Each of ``R``, ``L`` and ``C`` is recomputed only if the matching
    formula is set on the type; otherwise the attribute is left as
    ``None``. Utilizes the external ``impedance_calculator`` to avoid
    storing non-pickleable functions.

    Parameters
    ----------
    frequencies : List[float]
        A list of frequencies at which to
        evaluate the formulas.
    """
    rho = self.specific_earth_resistance
    params = {"rho": rho}

    # Mandatory frequency-domain impedance.
    impedance = compute_impedance(
        formula_str=self.type.impedance_formula,
        frequencies=frequencies,
        params=params,
    )
    # The grounding impedance is inverted into a diagonal admittance, so a
    # value that cannot become an admittance has to be rejected here rather
    # than silently dropped by the solver. See check_passive_impedance.
    check_passive_impedance(
        impedance,
        element=f"bus '{self.name}' (grounding impedance)",
        formula_str=self.type.impedance_formula,
        params=params,
    )
    self.impedance = impedance

    # Optional lumped RLC -- skipped silently when no formula is set.
    if self.type.R_formula is not None:
        self.R = compute_real_value(
            self.type.R_formula, frequencies, params, name=f"{self.name}.R"
        )
    if self.type.L_formula is not None:
        self.L = compute_real_value(
            self.type.L_formula, frequencies, params, name=f"{self.name}.L"
        )
    if self.type.C_formula is not None:
        self.C = compute_real_value(
            self.type.C_formula, frequencies, params, name=f"{self.name}.C"
        )

BusType

Bases: BaseModel

Represents the type of a bus, including its default impedance formula.

The mandatory impedance_formula is used by the frequency-domain solver (Y(f) u = i) and is the only required parameter for stationary studies.

For transient simulations the type can additionally carry an explicit lumped-element decomposition R_formula / L_formula / C_formula. These are parallel to impedance_formula: the frequency-domain solver ignores them, the FFT- and state-space-based transient solvers consume them. The duplication is intentional so that the stationary model and the transient equivalent can be parameterised independently — a substation grounding, for example, may be modelled as a constant R in the stationary formula while the transient model uses the full R + j*omega*L plus an HF capacitance to remote earth.

Attributes:

Name Type Description
name str

The name of the bus type.

description (str, optional)

A brief description of the bus type.

system_type str

The system type associated with the bus, e.g. 'Tower' or 'Substation'.

voltage_level float

The voltage level of the bus, in kV.

impedance_formula str

SymPy formula for the frequency-domain grounding impedance Z(f, rho). Mandatory.

R_formula (str, optional)

SymPy formula for the lumped resistance R(rho, f) in Ohm. Used only by the transient solvers.

L_formula (str, optional)

SymPy formula for the lumped inductance L(rho, f) in Henry. Used only by the transient solvers.

C_formula (str, optional)

SymPy formula for the lumped capacitance to remote earth C(rho, f) in Farad. Used only by the transient solvers (typically only relevant for HF studies).

earthing_conductor_material ({Cu, Al, Steel}, optional)

Material of the earthing conductor (Erdungsleiter) — the lumped connection that carries the earth-fault current from the installation into the grounding system. Consumed only by :func:groundinsight.check_node_limits.

earthing_conductor_cross_section_mm2 (float, optional)

Cross-section of the earthing conductor in mm². Must be strictly positive when given.

earthing_conductor_theta_initial_C float

Initial earthing-conductor temperature in °C. Defaults to 20.0 (ambient).

earthing_conductor_theta_final_C (float, optional)

Maximum permissible earthing-conductor temperature in °C. Defaults to the material value in :data:groundinsight.analysis.thermal.IEC60949_MATERIALS.

earthing_conductor_current_split float

Share of the bus injection this conductor carries, in (0, 1]. 1.0 (default) is a single conductor carrying everything; 1/N splits the current across N equal parallel conductors. See :func:groundinsight.check_node_limits.

electrode_material ({Cu, Al, Steel}, optional)

Material of the earth electrode (Erder) — the part buried in the soil, which only carries the share of the current that is actually dissipated to earth at this bus.

electrode_cross_section_mm2 (float, optional)

Cross-section of the earth electrode in mm². Must be strictly positive when given.

electrode_theta_initial_C float

Initial electrode temperature in °C. Defaults to 20.0.

electrode_theta_final_C (float, optional)

Maximum permissible electrode temperature in °C. Buried electrodes are usually limited well below a free-air conductor to protect the surrounding soil and any coating; EN 50522 Table 2 is the reference.

electrode_current_split float

Share of the dissipated current a single electrode carries, in (0, 1]. Use 1/N for N equal parallel electrodes at the same bus.

Notes

The thermal fields are optional throughout. A bus is only assessed by :func:groundinsight.check_node_limits once both the material and the cross-section of the respective element are set; the two elements are independent, so a bus may declare only its electrode or only its earthing conductor.

validate_impedance_formula

validate_impedance_formula(value)

Validate the SymPy impedance_formula string.

Source code in src/groundinsight/models/core_models.py
@field_validator("impedance_formula")
def validate_impedance_formula(cls, value):
    """Validate the SymPy ``impedance_formula`` string."""
    return validate_impedance_formula_value(value)

validate_rlc_formula

validate_rlc_formula(value)

Validate the optional lumped RLC formula strings; None is allowed.

Source code in src/groundinsight/models/core_models.py
@field_validator("R_formula", "L_formula", "C_formula")
def validate_rlc_formula(cls, value):
    """Validate the optional lumped RLC formula strings; ``None`` is allowed."""
    if value is None:
        return value
    return validate_impedance_formula_value(value)

Branch

Bases: BaseModel

Represents a branch (conductor) connecting two buses in the network.

Attributes:

Name Type Description
name str

The name of the branch.

description (str, optional)

A brief description of the branch.

type BranchType

The type of the branch.

length float

The length of the branch (km).

from_bus str

The name of the originating bus.

to_bus str

The name of the destination bus.

self_impedance dict of float to ComplexNumber

Self-impedance values mapped by frequency.

mutual_impedance dict of float to ComplexNumber

Mutual-impedance values mapped by frequency.

specific_earth_resistance float

The specific earth resistance associated with the branch (Ohm * m).

parallel_coefficient (float, optional)

The parallel coefficient between 0 and 1, if any. Defaults to 1.0.

active bool

Whether the branch participates in the solve. An inactive branch behaves like an open circuit: it contributes neither to the admittance matrix nor to the mutual-coupling injection, paths traversing it are dropped, and its branch current in the result is forced to zero. Defaults to True. Used by the outage / what-if machinery in :mod:groundinsight.simulation.outage.

calculate_impedance

calculate_impedance(frequencies: List[float])

Calculate self/mutual impedance and -- if specified by the type -- the phase impedance and the lumped RLCM parameters for each frequency.

self_impedance and mutual_impedance are always recomputed. Each of phase_impedance, R_self, L_self, C_self, R_mutual, M_mutual is recomputed only if the matching formula is set on the branch type; otherwise the attribute is left as None.

Parameters:

Name Type Description Default
frequencies List[float]

A list of frequencies at which to evaluate the formulas.

required
Source code in src/groundinsight/models/core_models.py
def calculate_impedance(self, frequencies: List[float]):
    """
    Calculate self/mutual impedance and -- if specified by the type --
    the phase impedance and the lumped RLCM parameters for each frequency.

    ``self_impedance`` and ``mutual_impedance`` are always recomputed.
    Each of ``phase_impedance``, ``R_self``, ``L_self``, ``C_self``,
    ``R_mutual``, ``M_mutual`` is recomputed only if the matching
    formula is set on the branch type; otherwise the attribute is left
    as ``None``.

    Parameters
    ----------
    frequencies : List[float]
        A list of frequencies at which to
        evaluate the formulas.
    """
    self._calculate_self_impedance(frequencies)
    self._calculate_mutual_impedance(frequencies)
    self._calculate_phase_impedance(frequencies)
    self._calculate_rlc_parameters(frequencies)

validate_phase_impedance

validate_phase_impedance(value)

Accept None or the same shapes as the other impedance dicts.

Source code in src/groundinsight/models/core_models.py
@field_validator("phase_impedance", mode="before")
def validate_phase_impedance(cls, value):
    """Accept ``None`` or the same shapes as the other impedance dicts."""
    if value is None:
        return None
    if not isinstance(value, dict):
        raise TypeError(
            "Impedance must be a dictionary of frequency to impedance values."
        )
    new_value = {}
    for freq, imp in value.items():
        freq = float(freq)
        new_value[freq] = ComplexNumber.validate_complex(imp)
    return new_value

BranchType

Bases: BaseModel

Represents the type of a branch, including its impedance formulas.

The mandatory self_impedance_formula and mutual_impedance_formula drive the frequency-domain solver.

The optional phase_impedance_formula describes the phase conductor -- the faulted conductor whose current induces the longitudinal EMF on the shield -- and is what the automatic phase-current distribution solves on. It is only needed when the network contains rings, meshes or parallel branches, because only then does the source current have more than one way to reach the fault. Without it the distribution falls back to a documented proxy and warns; see :meth:~groundinsight.electrical_network. ElectricalNetwork._compute_phase_currents_auto.

For transient simulations the type can additionally carry a lumped RLCM decomposition: R_self_formula, L_self_formula, C_self_formula (shunt-to-ground capacitance per branch, only relevant for HF studies), R_mutual_formula (Carson earth-return resistance term) and M_mutual_formula (mutual inductance to the parallel phase conductor). These are parallel to the impedance formulas: the frequency-domain solver ignores them, the state-space and FFT-based transient solvers consume them. The duplication is intentional so the stationary and transient parameterisations can be maintained independently.

Attributes:

Name Type Description
name str

The name of the branch type.

description (str, optional)

A brief description of the branch type.

grounding_conductor bool

Indicates whether the branch has a grounding wire or cable shield.

self_impedance_formula str

SymPy formula used to calculate self-impedance per branch.

mutual_impedance_formula str

SymPy formula used to calculate mutual impedance.

phase_impedance_formula (str, optional)

SymPy formula for the series impedance of the phase conductor of this branch, in the same symbols f, rho, l as the other formulas. Consumed only by the automatic phase-current distribution; the admittance matrix never sees it.

R_self_formula (str, optional)

Per-branch series resistance in Ohm. Used only by the transient solvers.

L_self_formula (str, optional)

Per-branch series inductance in Henry. Used only by the transient solvers.

C_self_formula (str, optional)

Per-branch shunt capacitance to remote earth in Farad. Used only by the transient solvers.

R_mutual_formula (str, optional)

Per-branch mutual resistance (Carson earth-return) in Ohm. Used only by the transient solvers.

M_mutual_formula (str, optional)

Per-branch mutual inductance in Henry. Used only by the transient solvers.

validate_impedance_formula

validate_impedance_formula(value)

Validate the SymPy self / mutual impedance formula strings.

Source code in src/groundinsight/models/core_models.py
@field_validator("self_impedance_formula", "mutual_impedance_formula")
def validate_impedance_formula(cls, value):
    """Validate the SymPy self / mutual impedance formula strings."""
    return validate_impedance_formula_value(value)

validate_rlc_formula

validate_rlc_formula(value)

Validate the optional lumped RLC formula strings; None is allowed.

Source code in src/groundinsight/models/core_models.py
@field_validator(
    "phase_impedance_formula",
    "R_self_formula",
    "L_self_formula",
    "C_self_formula",
    "R_mutual_formula",
    "M_mutual_formula",
)
def validate_rlc_formula(cls, value):
    """Validate the optional lumped RLC formula strings; ``None`` is allowed."""
    if value is None:
        return value
    return validate_impedance_formula_value(value)

Source

Bases: BaseModel

Represents a current or Thevenin (voltage) source within the network.

For stationary grounding analyses the default is a current source with a fixed injected current per frequency (source_type="current"). This is equivalent to a Norton source with infinite parallel impedance and matches the conventional planning practice of grounding engineering, where the prospective fault current is treated as a constant input.

For transient simulations the source can alternatively be expressed as a Thevenin equivalent (source_type="voltage") with a frequency-dependent EMF voltage and a finite source_impedance. In that case the grounding network sees the loop impedance Z_src + Z_loop and the effective fault current results from the solution rather than being prescribed.

Attributes:

Name Type Description
name str

The name of the source.

description (str, optional)

A brief description of the source.

bus str

The name of the bus where the source is located.

source_type {current, voltage}

"current" (default) for a classic current-source injection; "voltage" for a Thevenin equivalent (EMF in series with source_impedance).

values dict of float to ComplexNumber, optional

Frequency-dependent current injection. Required when source_type == "current" and must be None otherwise.

voltage dict of float to ComplexNumber, optional

Frequency-dependent Thevenin EMF. Required when source_type == "voltage" and must be None otherwise.

source_impedance dict of float to ComplexNumber, optional

Frequency-dependent internal impedance of the Thevenin source. Required when source_type == "voltage" and must be None otherwise. Must be non-zero at every frequency in order to be invertible into a Norton equivalent.

i_k_a (float, optional)

Initial symmetrical short-circuit current I_k'' at the fault location as seen from this source, in amperes. Provenance metadata for the IEC 60909 characteristics — the solve always uses values / voltage. Typically filled by :func:groundinsight.io.apply_shortcircuit_characteristics from a pandapower calc_sc run.

r_to_x (float, optional)

Ratio R/X of the short-circuit loop feeding this source, used to derive kappa when the latter is not given explicitly. For a single line-to-earth fault the relevant loop is 2*Z1 + Z0, so r_to_x = (2*R1 + R0) / (2*X1 + X0).

kappa (float, optional)

IEC 60909-0 peak factor kappa of this source. Takes precedence over r_to_x when both are set, which lets a topology-aware value (e.g. pandapower's method C) override the closed-form 1.02 + 0.98 * exp(-3 * R/X). Physically bounded to (1, 2].

Notes

i_k_a, r_to_x and kappa are characteristic quantities. They do not enter the linear solve at all: the network equations superpose the RMS injections in values as before. The non-linear IEC 60909 factors are applied afterwards, to the aggregated branch current, by :func:groundinsight.analysis.shortcircuit.resolve_fault_sc_characteristics. Superposing i_p or I_th of individual sources directly would be wrong; see that module's docstring for the derivation.

Fault

Bases: BaseModel

Represents a fault within the network.

Attributes:

Name Type Description
name str

The name of the fault.

description (str, optional)

A brief description of the fault.

bus str

The name of the bus where the fault occurs.

scalings dict of float to float

Scaling factors for sources at different frequencies.

t_k_s (float, optional)

Short-circuit duration (clearing time) T_k in seconds, as defined in IEC 60909-0. Drives both the DC heat factor m of the thermally equivalent short-circuit current I_th and the adiabatic conductor rating I_adm = k * S / sqrt(t_k). None means "not specified"; the thermal check then requires an explicit t_k argument.

n_factor float, default 1.0

AC heat factor n of IEC 60909-0. 1.0 is the far-from-generator case (no AC decay), which is the normal situation for grounding studies. Values below 1.0 apply when the fault is near a generator and the AC component decays during T_k.

Notes

The private _active attribute indicates whether the fault is the currently active one in the network and is exposed read-only through the :attr:active computed property.

t_k_s and n_factor live on the fault, not on the sources, because the clearing time is a property of the protection scheme reacting to that fault. The IEC 60909 quantities that describe the feeding side (I_k'', R/X, kappa) live on :class:Source instead.

active property

active: bool

Whether this fault is currently the active one in the network.

Network

Bases: BaseModel

Top-level container for an entire grounding network.

Holds every physical element (buses, branches, sources, faults), the enumerated source-to-fault paths, the per-fault result objects and a private :class:ElectricalNetwork helper that owns the numerical working arrays.

Attributes:

Name Type Description
name str

The name of the network.

description (str, optional)

A brief description of the network.

frequencies list of float

Frequencies (in Hz) used in calculations.

buses dict of str to Bus

Buses keyed by name.

branches dict of str to Branch

Branches keyed by name.

faults dict of str to Fault

Faults keyed by name.

sources dict of str to Source

Sources keyed by name.

results dict of str to Result

Per-fault calculation results keyed by fault name.

paths dict of str to Path

Source-to-fault paths keyed by path name.

active_fault (str, optional)

Name of the currently active fault.

add_branch

add_branch(branch: Branch, overwrite: bool = False)

Adds a branch to the network.

Parameters:

Name Type Description Default
branch Branch

The branch instance to add.

required
overwrite bool

If True, overwrites an existing branch with the same name. Defaults to False.

False

Raises:

Type Description
ValueError

If a branch with the same name already exists, or if the connected buses are not in the network.

Source code in src/groundinsight/models/core_models.py
def add_branch(self, branch: Branch, overwrite: bool = False):
    """
    Adds a branch to the network.

    Parameters
    ----------
    branch : Branch
        The branch instance to add.
    overwrite : bool, optional
        If True, overwrites an existing branch with the same name. Defaults to False.

    Raises
    ------
    ValueError
        If a branch with the same name already exists, or if the connected buses are not in the network.
    """
    if branch.name in self.branches:
        if overwrite:
            logger.warning(
                "Branch '%s' already exists in the network. Overwriting.",
                branch.name,
            )
        else:
            raise ValueError(
                f"Branch with name '{branch.name}' already exists in the network '{self.name}'. If you want to overwrite, set overwrite=True."
            )

    # Validate that the from_bus and to_bus are in the network
    if branch.from_bus not in self.buses:
        raise ValueError(
            f"from_bus '{branch.from_bus}' is not in the network '{self.name}'"
        )
    if branch.to_bus not in self.buses:
        raise ValueError(
            f"to_bus '{branch.to_bus}' is not in the network '{self.name}'"
        )
    self.branches[branch.name] = branch
    # Trigger impedance calculation when a branch is added
    branch.calculate_impedance(self.frequencies)

add_bus

add_bus(bus: Bus, overwrite: bool = False)

Adds a bus to the network.

Parameters:

Name Type Description Default
bus Bus

The bus instance to add.

required
overwrite bool

If True, overwrites an existing bus with the same name. Defaults to False.

False

Raises:

Type Description
ValueError

If a bus with the same name already exists and overwrite is False.

Source code in src/groundinsight/models/core_models.py
def add_bus(self, bus: Bus, overwrite: bool = False):
    """
    Adds a bus to the network.

    Parameters
    ----------
    bus : Bus
        The bus instance to add.
    overwrite : bool, optional
        If True, overwrites an existing bus with the same name. Defaults to False.

    Raises
    ------
    ValueError
        If a bus with the same name already exists and overwrite is False.
    """
    if bus.name in self.buses:
        if overwrite:
            logger.warning(
                "Bus '%s' already exists in the network. Overwriting.",
                bus.name,
            )
        else:
            raise ValueError(
                f"Bus with name '{bus.name}' already exists in the network '{self.name}'. If you want to overwrite, set overwrite=True."
            )

    self.buses[bus.name] = bus
    # Trigger impedance calculation when a bus is added
    bus.calculate_impedance(self.frequencies)

add_fault

add_fault(fault: Fault, overwrite: bool = False)

Adds a fault to the network.

Parameters:

Name Type Description Default
fault Fault

The fault instance to add.

required
overwrite bool

If True, overwrites an existing fault with the same name. Defaults to False.

False

Raises:

Type Description
ValueError

If a fault with the same name already exists, or if the associated bus is not in the network.

Source code in src/groundinsight/models/core_models.py
def add_fault(self, fault: Fault, overwrite: bool = False):
    """
    Adds a fault to the network.

    Parameters
    ----------
    fault : Fault
        The fault instance to add.
    overwrite : bool, optional
        If True, overwrites an existing fault with the same name. Defaults to False.

    Raises
    ------
    ValueError
        If a fault with the same name already exists, or if the associated bus is not in the network.
    """
    if fault.bus not in self.buses:
        raise ValueError(f"bus '{fault.bus}' is not in the network '{self.name}'")

    if fault.name in self.faults:
        if overwrite:
            logger.warning(
                "Fault '%s' already exists in the network. Overwriting.",
                fault.name,
            )
        else:
            raise ValueError(
                f"Fault with name '{fault.name}' already exists in the network '{self.name}'. If you want to overwrite, set overwrite=True."
            )

    self.faults[fault.name] = fault

add_path

add_path(path: Path)

Adds a path to the network.

Parameters:

Name Type Description Default
path Path

The path instance to add.

required
Source code in src/groundinsight/models/core_models.py
def add_path(self, path: Path):
    """
    Adds a path to the network.

    Parameters
    ----------
    path : Path
        The path instance to add.
    """
    self.paths[path.name] = path

add_source

add_source(source: Source, overwrite: bool = False)

Adds a source to the network.

Parameters:

Name Type Description Default
source Source

The source instance to add.

required
overwrite bool

If True, overwrites an existing source with the same name. Defaults to False.

False

Raises:

Type Description
ValueError

If a source with the same name already exists, or if the associated bus is not in the network.

Source code in src/groundinsight/models/core_models.py
def add_source(self, source: Source, overwrite: bool = False):
    """
    Adds a source to the network.

    Parameters
    ----------
    source : Source
        The source instance to add.
    overwrite : bool, optional
        If True, overwrites an existing source with the same name. Defaults to False.

    Raises
    ------
    ValueError
        If a source with the same name already exists, or if the associated bus is not in the network.
    """
    if source.bus not in self.buses:
        raise ValueError(f"bus '{source.bus}' is not in the network '{self.name}'")

    if source.name in self.sources:
        if overwrite:
            logger.warning(
                "Source '%s' already exists in the network. Overwriting.",
                source.name,
            )
        else:
            raise ValueError(
                f"Source with name '{source.name}' already exists in the network '{self.name}'. If you want to overwrite, set overwrite=True."
            )
    self.sources[source.name] = source

define_paths

define_paths()

Identifies all paths from all sources to all faults in the network and adds them to the network's paths.

This method utilizes the PathFinder to locate paths and ensures that each path is unique before adding it to the network.

Source code in src/groundinsight/models/core_models.py
def define_paths(self):
    """
    Identifies all paths from all sources to all faults in the network and adds them to the network's paths.

    This method utilizes the `PathFinder` to locate paths and ensures that each path is unique
    before adding it to the network.
    """
    from groundinsight.pathfinder import PathFinder  # Import locally

    pathfinder = PathFinder(self)
    path_counter = 1  # To create unique path names
    seen_paths = set()  # To track unique paths

    for source_name, source in self.sources.items():
        source_bus_name = source.bus
        for fault_name, fault in self.faults.items():
            fault_bus_name = fault.bus
            # Find all paths between this source and fault
            paths = pathfinder.find_paths(source_bus_name, fault_bus_name)
            for path in paths:
                # Create a hashable representation of the path to check for duplicates
                path_signature = (
                    source_name,
                    fault_name,
                    tuple(branch.name for branch in path.segments),
                )
                if path_signature not in seen_paths:
                    seen_paths.add(path_signature)
                    # Assign a unique name to each path
                    path.name = f"path_{path_counter}"
                    path.description = f"Path from {source_name} to {fault_name}"
                    path.source = source_name
                    path.fault = fault_name
                    path_counter += 1
                    # Add the path to the network
                    self.add_path(path)

    # Record the active-topology fingerprint the paths were built for so
    # ``run_fault`` can detect stale paths after an in-place ``active``
    # flip or a rewiring and rebuild them instead of silently reusing them.
    self._paths_fingerprint = self._active_topology_fingerprint()

invalidate_paths

invalidate_paths() -> None

Drop the cached pathfinder results for this network.

Rebinds self.paths to a fresh empty dictionary (atomic) and drops the module-level :mod:groundinsight.pathfinder cache entries whose key is scoped to this :class:Network instance. Other networks' cache entries are preserved. This matters as soon as the user runs more than one network in the same Python process (notebooks that compare two scenarios, dashboards iterating over a set of feeders, …).

Earlier revisions called self.paths.clear() in place. Callers that had snapshot the mapping with saved = dict(network.paths) before the call observed the snapshot lose its entries because the snapshot dictionary shared its Path values with self.paths until the snapshot was deep-copied. The current atomic-rebind form — self.paths = {} — leaves the snapshot mapping intact and mirrors the atomic-rebind pattern in :func:groundinsight.analysis.inverse_rho_f.evaluate_max_epr_under_k.

Call this whenever the user has flipped Bus.active / Branch.active flags or added / removed branches outside of a context manager that performs its own rollback.

Source code in src/groundinsight/models/core_models.py
def invalidate_paths(self) -> None:
    """Drop the cached pathfinder results *for this network*.

    Rebinds ``self.paths`` to a fresh empty dictionary (atomic) and
    drops the module-level :mod:`groundinsight.pathfinder` cache
    entries whose key is scoped to this :class:`Network` instance.
    **Other networks' cache entries are preserved.** This matters
    as soon as the user runs more than one network in the same
    Python process (notebooks that compare two scenarios,
    dashboards iterating over a set of feeders, …).

    Earlier revisions called ``self.paths.clear()`` in place.
    Callers that had snapshot the mapping with
    ``saved = dict(network.paths)`` before the call observed the
    snapshot lose its entries because the snapshot dictionary
    shared its ``Path`` *values* with ``self.paths`` until the
    snapshot was deep-copied. The current atomic-rebind form
    — ``self.paths = {}`` — leaves the snapshot mapping intact and
    mirrors the atomic-rebind pattern in
    :func:`groundinsight.analysis.inverse_rho_f.evaluate_max_epr_under_k`.

    Call this whenever the user has flipped ``Bus.active`` /
    ``Branch.active`` flags or added / removed branches outside of
    a context manager that performs its own rollback.
    """
    from groundinsight.pathfinder import clear_pathfinder_cache  # local import

    # Atomic rebind so external snapshots survive.
    self.paths = {}
    clear_pathfinder_cache(self)

res_all_impedances

res_all_impedances() -> pl.DataFrame

Returns a Polars DataFrame containing the grounding impedance and reduction factor for each fault, bus, and frequency.

The DataFrame includes grounding impedance magnitude and angle, as well as the reduction factor.

Returns:

Type Description
DataFrame

A DataFrame containing grounding impedance and reduction factors.

Notes
- Faults without results are skipped.
- Missing grounding impedance or reduction factor results are noted.
Source code in src/groundinsight/models/core_models.py
def res_all_impedances(self) -> pl.DataFrame:
    """
    Returns a Polars DataFrame containing the grounding impedance and reduction factor
    for each fault, bus, and frequency.

    The DataFrame includes grounding impedance magnitude and angle, as well as the reduction factor.

    Returns
    -------
    pl.DataFrame
        A DataFrame containing grounding impedance and reduction factors.

    Notes
    -----
        - Faults without results are skipped.
        - Missing grounding impedance or reduction factor results are noted.
    """
    data = []
    for fault_name, fault in self.faults.items():
        if fault_name not in self.results:
            logger.warning(
                "No results available for fault '%s'. Skipping.",
                fault_name,
            )
            continue
        result = self.results[fault_name]
        fault_bus = fault.bus

        # Grounding Impedance
        grounding_impedance = result.grounding_impedance
        if not grounding_impedance:
            logger.warning(
                "No grounding impedance results for fault '%s'.",
                fault_name,
            )
            continue

        # Reduction Factor
        reduction_factor = result.reduction_factor
        if not reduction_factor:
            logger.warning(
                "No reduction factor results for fault '%s'.",
                fault_name,
            )
            continue

        for freq in self.frequencies:
            gi = grounding_impedance.value.get(freq)
            rf = reduction_factor.value.get(freq)
            rf_current = reduction_factor.value_current.get(freq)
            if gi:
                gi_real = gi.real
                gi_imag = gi.imag
                gi_magnitude = abs(complex(gi.real, gi.imag))
                gi_angle = np.degrees(np.angle(complex(gi.real, gi.imag)))
            else:
                gi_real = None
                gi_imag = None
                gi_magnitude = None
                gi_angle = None

            data.append(
                {
                    "fault_name": fault_name,
                    "fault_bus": fault_bus,
                    "frequency_Hz": freq,
                    "grounding_impedance_Ohm": gi_magnitude,
                    "grounding_impedance_deg": gi_angle,
                    "reduction_factor": rf,
                    "reduction_factor_current": rf_current,
                }
            )
    df = pl.DataFrame(data)
    return df

res_branches

res_branches(fault: Optional[str] = None) -> pl.DataFrame

Returns a Polars DataFrame with branch results for the specified fault.

If no fault is specified, returns results for the active fault.

Parameters:

Name Type Description Default
fault Optional[str]

The name of the fault. Defaults to None.

None

Returns:

Type Description
DataFrame

A DataFrame containing branch results.

Raises:

Type Description
ValueError

If no active fault is set or if results for the specified fault are unavailable.

Source code in src/groundinsight/models/core_models.py
def res_branches(self, fault: Optional[str] = None) -> pl.DataFrame:
    """
    Returns a Polars DataFrame with branch results for the specified fault.

    If no fault is specified, returns results for the active fault.

    Parameters
    ----------
    fault : Optional[str], optional
        The name of the fault. Defaults to None.

    Returns
    -------
    pl.DataFrame
        A DataFrame containing branch results.

    Raises
    ------
    ValueError
        If no active fault is set or if results for the specified fault are unavailable.
    """
    if fault is None:
        fault = self.active_fault
        if fault is None:
            raise ValueError("No active fault set in the network.")

    if fault not in self.results:
        raise ValueError(f"No results available for fault '{fault}'.")

    result = self.results[fault]
    data = []
    for result_branch in result.branches:
        # Add frequency-specific data
        for freq, current in result_branch.i_s_freq.items():
            if current:
                current_abs = abs(complex(current.real, current.imag))
                current_ang = (
                    np.angle(complex(current.real, current.imag)) * 180 / np.pi
                )
                data.append(
                    {
                        "branch_name": result_branch.name,
                        "fault": fault,
                        "frequency_Hz": freq,
                        "I_branch_A": current_abs,
                        "I_branch_degree": current_ang,
                    }
                )
        # Add RMS current
        data.append(
            {
                "branch_name": result_branch.name,
                "fault": fault,
                "frequency_Hz": "RMS",
                "I_branch_A": result_branch.i_s,
                "I_branch_degree": None,
            }
        )

    df = pl.DataFrame(data)
    return df

res_buses

res_buses(fault: Optional[str] = None) -> pl.DataFrame

Returns a Polars DataFrame with bus results for the specified fault.

If no fault is specified, returns results for the active fault.

Parameters:

Name Type Description Default
fault Optional[str]

The name of the fault. Defaults to None.

None

Returns:

Type Description
DataFrame

A DataFrame containing bus results.

Raises:

Type Description
ValueError

If no active fault is set or if results for the specified fault are unavailable.

Source code in src/groundinsight/models/core_models.py
def res_buses(self, fault: Optional[str] = None) -> pl.DataFrame:
    """
    Returns a Polars DataFrame with bus results for the specified fault.

    If no fault is specified, returns results for the active fault.

    Parameters
    ----------
    fault : Optional[str], optional
        The name of the fault. Defaults to None.

    Returns
    -------
    pl.DataFrame
        A DataFrame containing bus results.

    Raises
    ------
    ValueError
        If no active fault is set or if results for the specified fault are unavailable.
    """
    if fault is None:
        fault = self.active_fault
        if fault is None:
            raise ValueError("No active fault set in the network.")

    if fault not in self.results:
        raise ValueError(f"No results available for fault '{fault}'.")

    result = self.results[fault]
    data = []
    for result_bus in result.buses:
        # Add frequency-specific data
        for freq, voltage in result_bus.uepr_freq.items():
            current = result_bus.ia_freq.get(freq)
            voltage_abs = abs(complex(voltage.real, voltage.imag))
            current_abs = abs(complex(current.real, current.imag))
            voltage_ang = (
                np.angle(complex(voltage.real, voltage.imag)) * 180 / np.pi
            )
            current_ang = (
                np.angle(complex(current.real, current.imag)) * 180 / np.pi
            )
            data.append(
                {
                    "bus_name": result_bus.name,
                    "fault": fault,
                    "frequency_Hz": freq,
                    "EPR_V": voltage_abs,
                    "EPR_degree": voltage_ang,
                    "I_bus_A": current_abs,
                    "I_bus_degree": current_ang,
                }
            )
        # Add RMS values
        data.append(
            {
                "bus_name": result_bus.name,
                "fault": fault,
                "frequency_Hz": "RMS",
                "EPR_V": result_bus.uepr,
                "EPR_degree": None,
                "I_bus_A": result_bus.ia,
                "I_bus_degree": None,
            }
        )

    df = pl.DataFrame(data)
    return df

set_active_fault

set_active_fault(
    fault_name: str, keep_results: bool = False
)

Set the specified fault as active and deactivate all other faults.

Parameters:

Name Type Description Default
fault_name str

The name of the fault to activate.

required
keep_results bool

If False (historic behaviour), any previously cached :class:Result for fault_name is dropped so the next solve starts from a clean slate. If True, the cached result is preserved — useful when re-using the same network to plot a previous solve without recomputing it.

``False``

Raises:

Type Description
ValueError

If the specified fault does not exist in the network.

Source code in src/groundinsight/models/core_models.py
def set_active_fault(self, fault_name: str, keep_results: bool = False):
    """Set the specified fault as active and deactivate all other faults.

    Parameters
    ----------
    fault_name : str
        The name of the fault to activate.
    keep_results : bool, default ``False``
        If ``False`` (historic behaviour), any previously cached
        :class:`Result` for ``fault_name`` is dropped so the next
        solve starts from a clean slate. If ``True``, the cached
        result is preserved — useful when re-using the same
        network to plot a previous solve without recomputing it.

    Raises
    ------
    ValueError
        If the specified fault does not exist in the network.
    """
    if fault_name not in self.faults:
        raise ValueError(f"Fault '{fault_name}' does not exist in the network.")

    # Deactivate all faults
    for fault in self.faults.values():
        fault._set_active(False)

    # Activate the specified fault
    fault = self.faults[fault_name]
    fault._set_active(True)
    self.active_fault = fault_name

    # Clear previous results for the fault unless the caller
    # explicitly asks us to keep them.
    if not keep_results and fault_name in self.results:
        del self.results[fault_name]

Path

Bases: BaseModel

Ordered branch list connecting a source bus to a fault bus.

Attributes:

Name Type Description
name str

The name of the path.

description (str, optional)

A brief description of the path.

source str

The name of the source at the start of the path.

fault str

The name of the fault at the end of the path.

segments list of Branch

Ordered list of branches that make up the path, traversed from source to fault.

Result

Bases: BaseModel

Overall result of a single fault calculation.

Attributes:

Name Type Description
buses list of ResultBus

Per-bus results.

branches list of ResultBranch

Per-branch results.

reduction_factor (ResultReductionFactor, optional)

The reduction factor result, if available.

grounding_impedance (ResultGroundingImpedance, optional)

The grounding impedance result, if available.

fault str

The name of the fault that was active during the calculation.

ResultBus

Bases: BaseModel

Result data for a bus after running a fault calculation.

Three physically distinct currents meet at a grounding bus, and mixing them up is the classic sizing error EN 50522 / IEC 61936-1 guard against:

  • i_inj — the current injected into the grounding system at this bus by the sources (at a source bus) or drawn out of it by the fault (at the fault bus). It flows through a lumped connection, the earthing conductor (Erdungsleiter), which therefore has to be sized for the full earth-fault current. Zero at every other bus.
  • ia — the share of that current which is actually dissipated into the soil at this bus, u_EPR / Z_B. It flows through the earth electrode (Erder) and is generally much smaller.
  • the branch shield currents, reported per branch on :class:ResultBranch.

i_inj deliberately excludes the mutual Norton-equivalent injections of the inductively coupled branches: those model a distributed induced EMF along the line, not a current entering the node through a lumped conductor. The full nodal balance including them is ia = i_vector + sum_branches (u_other - u_self) * Y_self.

Attributes:

Name Type Description
name str

The name of the bus.

uepr float

RMS earth potential rise at the bus, in volts.

ia float

RMS bus current dissipated into the soil through the earth electrode, in amperes.

i_inj float

RMS source-side injection at the bus, in amperes — the current carried by the earthing conductor. Defaults to 0.0 so that results stored before this field existed still validate.

uepr_freq dict of float to ComplexNumber

Mapping of frequency to complex voltage values.

ia_freq dict of float to ComplexNumber

Mapping of frequency to complex electrode current values.

i_inj_freq dict of float to ComplexNumber

Mapping of frequency to complex injection values. Defaults to an empty mapping for backwards compatibility.

ResultBranch

Bases: BaseModel

Result data for a branch after running a fault calculation.

Attributes:

Name Type Description
name str

The name of the branch.

i_s float

RMS shield (grounding-conductor) current in the branch, in amperes.

i_s_freq dict of float to ComplexNumber

Mapping of frequency to complex shield current values.

ResultReductionFactor

Bases: BaseModel

Reduction-factor result at the fault bus.

Attributes:

Name Type Description
name (str, optional)

The name of the reduction factor result.

fault_bus str

The bus where the fault occurred.

value dict of float to float, optional

Mapping from frequency to the EPR-based reduction factor r(f) = |u_fault with mutual| / |u_fault without mutual|. This is the quantity that reproduces the familiar closed form r = |1 - Z_mutual / Z_self| for a single shielded line, and it is what ResultGroundingImpedance divides by. It is structurally independent of the impedance at the fault bus.

value_current dict of float to float, optional

Mapping from frequency to the current-based reduction factor r_I(f) = |I_E| / |I_F| -- the measured definition. I_E is the sum of the electrode currents of every bus that feeds the soil, not the electrode current at the faulted station: where the stations are bonded through continuous cable shields the current spreads along them and leaks into the soil at every one of them. Taking only the faulted station understates it by a factor of 1.8 to 4.7 on the verification feeder, depending on where the fault sits. Unlike value this factor responds to the fault-bus characteristic, which is what makes it usable as the ordinate of a rho-f sensitivity study. None per frequency where the fault bus carries no injection or no bus is earthed.

i_earth dict of float to complex, optional

The earth-return current I_E itself, per frequency, so a study can read the ampere value rather than only the ratio.

earth_buses dict of float to list of str, optional

The buses counted as feeding the soil at each frequency -- from the fault outwards, in every direction, up to where the potential profile turns. Reported because the split is a modelling statement and should be inspectable; see :mod:groundinsight.utils.earth_current.

u_earthing dict of float to complex, optional

The earthing voltage U_E of that bonded group: the mean potential weighted by each station's electrode current.

z_earthing dict of float to complex, optional

The earthing impedance Z_E = U_E / I_E of the group. Together with the fields above this closes the EN 50522 chain U_E = 3*I_0 * Z_E * r, which holds in the model to machine precision and can be checked from the result alone. Note that Z_E is not simply the electrodes in parallel: the shield sections between the stations add to it (3.30 Ohm against 2.50 Ohm on the verification feeder).

ResultGroundingImpedance

Bases: BaseModel

Grounding impedance result at the fault bus.

Attributes:

Name Type Description
name (str, optional)

The name of the grounding impedance result.

fault_bus str

The bus where the fault occurred.

value dict of float to ComplexNumber, optional

Mapping from frequency to grounding impedance Z_G(f).

ComplexNumber

Bases: BaseModel

Pydantic-compatible complex number with real and imaginary parts.

Wraps the native :class:complex type so that Pydantic models can serialise and deserialise complex numbers through JSON.

Attributes:

Name Type Description
real float

The real part of the complex number.

imag float

The imaginary part of the complex number.

convert_to_float

convert_to_float(value: Any) -> float

Coerce real / imag inputs to float; None becomes NaN.

Parameters:

Name Type Description Default
value Any

Numeric input to coerce. None is interpreted as numpy.nan.

required

Returns:

Type Description
float

The coerced value.

Source code in src/groundinsight/models/core_models.py
@field_validator("real", "imag", mode="before")
def convert_to_float(cls, value: Any) -> float:
    """Coerce ``real`` / ``imag`` inputs to ``float``; ``None`` becomes ``NaN``.

    Parameters
    ----------
    value : Any
        Numeric input to coerce. ``None`` is interpreted
        as ``numpy.nan``.

    Returns
    -------
    float
        The coerced value.
    """
    if value is None:
        return np.nan
    return float(value)

validate_complex classmethod

validate_complex(value: Any) -> Union[ComplexNumber, dict]

Validates and converts the input value to a ComplexNumber instance.

Parameters:

Name Type Description Default
value Any

The value to validate and convert. Can be a ComplexNumber, complex, float, int, dict or str.

required

Returns:

Type Description
Union[ComplexNumber, dict]

Either the original

``ComplexNumber`` instance (passed through) or a dictionary
with ``real`` and ``imag`` keys ready for Pydantic to
instantiate ``ComplexNumber``.

Raises:

Type Description
ValueError

If the input string cannot be parsed as a complex number.

TypeError

If the input type is unsupported.

Source code in src/groundinsight/models/core_models.py
@model_validator(mode="before")
@classmethod
def validate_complex(cls, value: Any) -> Union["ComplexNumber", dict]:
    """
    Validates and converts the input value to a ComplexNumber instance.

    Parameters
    ----------
    value : Any
        The value to validate and convert. Can be a
        ``ComplexNumber``, ``complex``, ``float``, ``int``,
        ``dict`` or ``str``.

    Returns
    -------
    Union[ComplexNumber, dict]
        Either the original
    ``ComplexNumber`` instance (passed through) or a dictionary
    with ``real`` and ``imag`` keys ready for Pydantic to
    instantiate ``ComplexNumber``.

    Raises
    ------
    ValueError
        If the input string cannot be parsed as a complex number.
    TypeError
        If the input type is unsupported.
    """
    if isinstance(value, cls):
        return value
    elif isinstance(value, complex):
        return {"real": value.real, "imag": value.imag}
    elif isinstance(value, (float, int)):
        return {"real": float(value), "imag": 0.0}
    elif isinstance(value, dict):
        return value
    elif isinstance(value, str):
        try:
            c = complex(value.replace(" ", "").replace("i", "j"))
            return {"real": c.real, "imag": c.imag}
        except ValueError:
            raise ValueError(f"Invalid complex number string: {value}")
    else:
        raise TypeError(f"Cannot parse ComplexNumber from type {type(value)}")