Database

SQLAlchemy-based CRUD helpers for persisting bus types, branch types and entire networks to a SQLite database. The ORM mirror classes live next to the Pydantic models in groundinsight.models.database_models and expose from_pydantic / to_pydantic converters.

Physical / modelling context

The persistence layer is a side-channel to the in-memory Pydantic model. Every Pydantic class has an ORM counterpart with suffix DB (BusDB, BranchDB, NetworkDB, ...). Many-to-many relationships (e.g. a BranchType shared by several Network instances, or a Path composed of many Branch segments) are modelled through association tables (network_buses, path_segments, ...). Two design choices are worth noting:

• Frequency lists are stored as PickleType blobs. They are conceptually atomic — a network has one frequency list, never shared, never queried element-wise — so the round-trip price of storing them as JSON or as a child table is not justified.
• Impedance dicts \(Z(f) \in \mathbb{C}\) are stored as JSON with stringified frequency keys, because SQLite has no native complex type. The ComplexNumber Pydantic helper guarantees symmetric round-trips on read-back.

A typical workflow opens a session via gi.start_dbsession(path), reads or writes types or networks via the CRUD helpers, and closes the session at process exit. Sessions are not thread-safe; use one session per process.

Example

import groundinsight as gi

gi.start_dbsession("library.db")

# 1. Persist a re-usable BusType
bus_type = ...  # gi.BusType instance
gi.save_bustype_to_db(bus_type=bus_type, overwrite=True)

# 2. Persist an entire fault-ready network
gi.save_network_to_db(network=net, overwrite=True)

# 3. Read back later
recovered_types = gi.load_bustypes_from_db()  # dict[name -> BusType]
recovered_net = gi.load_network_from_db(name="demo")

gi.close_dbsession()

The high-level helpers re-exported on the top-level package (gi.save_bustype_to_db, gi.load_bustypes_from_db, gi.save_branchtype_to_db, gi.load_branchtypes_from_db, gi.save_network_to_db, gi.load_network_from_db) wrap the session lifecycle. The lower-level CRUD functions in groundinsight.database.crud take an explicit SQLAlchemy Session argument and are intended for callers that already manage their own session scope.

Session globals: gi.session and gi.db_session

The module exposes two equivalent names for the scoped session:

• gi.session — canonical name.
• gi.db_session — historic alias, kept to support from groundinsight import db_session written before 0.4.0.

Both names point at the same :class:sqlalchemy.orm.scoped_session instance while a session is active, and both are reset to None when no session is active. The lock-step contract is pinned by a single private helper, groundinsight._set_session(new). Every code path that rebinds the session — start_dbsession, close_dbsession and any future swap_dbsession context manager — must route through _set_session instead of assigning the module global directly. This removes the cross-API drift risk between the two names.

Cross-repo version helper: gi.show_versions()

Returns a dictionary with the installed versions of groundinsight, the active Python and platform, and — if importable — the peer packages groundfield and groundmeas. Intended as the ADR-0013-aligned cross-repo convention, so a bug report can include the full earthing-platform stack with a single call:

>>> import groundinsight as gi
>>> gi.show_versions()
{'groundinsight': '0.5.0',
 'python': '3.11.7',
 'platform': 'Linux-...-x86_64-...'}

CRUD functions

crud

CRUD Operations Module.

This module provides functions for creating, reading, updating, and deleting (CRUD) entities in the GroundInsight database using SQLAlchemy sessions. It facilitates the management of core electrical network components such as BusTypes, BranchTypes, Networks, Buses, Branches, Faults, Sources, and Paths. The functions convert between Pydantic models and SQLAlchemy database models to ensure seamless data manipulation and persistence.

ensure_current_schema

ensure_current_schema(session: Session)

Reject databases written by the pre-network-scoped schema.

Up to and including the association-table schema, buses, branches, faults, sources and paths were keyed by element name alone and linked to their network through network_buses & co. Two networks containing an element of the same name therefore shared one row, so saving one network silently rewrote the other. Those tables are now keyed by (network_name, name).

Base.metadata.create_all only ever creates missing tables -- it never adds a column to an existing one -- so an old database file opens without complaint and only fails deep inside a query with a bare OperationalError: no such column: buses.network_name. This helper turns that into an actionable message before any row is read or written.

This is the last line of defence, not the normal path: :func:groundinsight.start_dbsession converts such a file automatically (keeping a .bak copy) before the engine is bound. The error is reached when the caller passed migrate=False, or built its own engine and session without going through start_dbsession.

Parameters:

Name	Type	Description	Default
session	Session	The SQLAlchemy session whose bind is inspected.	required

Raises:

Type	Description
RuntimeError	If the connected database still uses the legacy, globally-keyed element tables. The message names :func:groundinsight.migrate_database.

See Also

groundinsight.database.migration.migrate_database : performs the conversion this function refuses to do implicitly.

Source code in src/groundinsight/database/crud.py

def ensure_current_schema(session: Session):
    """
    Reject databases written by the pre-network-scoped schema.

    Up to and including the association-table schema, ``buses``, ``branches``,
    ``faults``, ``sources`` and ``paths`` were keyed by element name alone and
    linked to their network through ``network_buses`` & co. Two networks
    containing an element of the same name therefore shared one row, so saving
    one network silently rewrote the other. Those tables are now keyed by
    ``(network_name, name)``.

    ``Base.metadata.create_all`` only ever creates *missing* tables -- it never
    adds a column to an existing one -- so an old database file opens without
    complaint and only fails deep inside a query with a bare
    ``OperationalError: no such column: buses.network_name``. This helper turns
    that into an actionable message before any row is read or written.

    This is the *last* line of defence, not the normal path:
    :func:`groundinsight.start_dbsession` converts such a file automatically
    (keeping a ``.bak`` copy) before the engine is bound. The error is reached
    when the caller passed ``migrate=False``, or built its own engine and
    session without going through ``start_dbsession``.

    Parameters
    ----------
    session : Session
        The SQLAlchemy session whose bind is inspected.

    Raises
    ------
    RuntimeError
        If the connected database still uses the legacy, globally-keyed
        element tables. The message names
        :func:`groundinsight.migrate_database`.

    See Also
    --------
    groundinsight.database.migration.migrate_database : performs the
        conversion this function refuses to do implicitly.
    """
    inspector = inspect(session.connection())
    table_names = set(inspector.get_table_names())
    if "buses" not in table_names:
        # Nothing written yet; ``create_all`` will lay out the current schema.
        return
    if any(column["name"] == "network_name" for column in inspector.get_columns("buses")):
        return

    legacy_tables = [name for name in _LEGACY_ASSOCIATION_TABLES if name in table_names]
    database = getattr(session.get_bind().url, "database", None) or "<your file>.db"
    raise RuntimeError(
        "This database uses the legacy groundinsight schema, in which buses, "
        "branches, faults, sources and paths were keyed by name alone and "
        "shared between networks"
        + (f" (found: {', '.join(legacy_tables)})" if legacy_tables else "")
        + ". Element tables are now keyed by (network_name, name). Convert the "
        f"file with gi.migrate_database('{database}') -- it copies the "
        "unmodified file to a .bak sibling first and reports anything it could "
        "not recover -- or let gi.start_dbsession() do it for you, which is "
        "the default."
    )

load_branchtypes

load_branchtypes(session: Session) -> Dict[str, BranchType]

Load all BranchTypes from the database.

This function retrieves all BranchType entries from the database and converts them into a dictionary mapping BranchType names to their corresponding Pydantic models.

Parameters:

Name	Type	Description	Default
session	Session	The SQLAlchemy session used for database operations.	required

Returns:

Type	Description
Dict[str, BranchType]	A dictionary where keys are BranchType names and values are BranchType instances.

Source code in src/groundinsight/database/crud.py

def load_branchtypes(session: Session) -> Dict[str, BranchType]:
    """
    Load all BranchTypes from the database.

    This function retrieves all BranchType entries from the database and converts them
    into a dictionary mapping BranchType names to their corresponding Pydantic models.

    Parameters
    ----------
    session : Session
        The SQLAlchemy session used for database operations.

    Returns
    -------
    Dict[str, BranchType]
        A dictionary where keys are BranchType names and values are BranchType instances.
    """
    branch_types = session.query(BranchTypeDB).all()
    return {bt.name: bt.to_pydantic() for bt in branch_types}

load_bustypes

load_bustypes(session: Session) -> Dict[str, BusType]

Load all BusTypes from the database.

This function retrieves all BusType entries from the database and converts them into a dictionary mapping BusType names to their corresponding Pydantic models.

Parameters:

Name	Type	Description	Default
session	Session	The SQLAlchemy session used for database operations.	required

Returns:

Type	Description
Dict[str, BusType]	A dictionary where keys are BusType names and values are BusType instances.

Source code in src/groundinsight/database/crud.py

def load_bustypes(session: Session) -> Dict[str, BusType]:
    """
    Load all BusTypes from the database.

    This function retrieves all BusType entries from the database and converts them
    into a dictionary mapping BusType names to their corresponding Pydantic models.

    Parameters
    ----------
    session : Session
        The SQLAlchemy session used for database operations.

    Returns
    -------
    Dict[str, BusType]
        A dictionary where keys are BusType names and values are BusType instances.
    """
    bus_types = session.query(BusTypeDB).all()
    return {bt.name: bt.to_pydantic() for bt in bus_types}

load_network

load_network(name: str, session: Session) -> Network

Load a Network from the database.

This function retrieves a Network instance by its name from the database and converts it into a Pydantic Network model. It ensures that all related entities such as Buses, Branches, Faults, Sources, and Paths are properly associated.

Parameters:

Name	Type	Description	Default
name	str	The name of the network to load.	required
session	Session	The SQLAlchemy session used for database operations.	required

Returns:

Type	Description
Network	The loaded Network instance.

Raises:

Type	Description
ValueError	If the specified network does not exist in the database.
RuntimeError	If the database still uses the legacy, globally-keyed element tables.

Source code in src/groundinsight/database/crud.py

def load_network(name: str, session: Session) -> Network:
    """
    Load a Network from the database.

    This function retrieves a `Network` instance by its name from the database and converts it
    into a Pydantic `Network` model. It ensures that all related entities such as Buses, Branches,
    Faults, Sources, and Paths are properly associated.

    Parameters
    ----------
    name : str
        The name of the network to load.
    session : Session
        The SQLAlchemy session used for database operations.

    Returns
    -------
    Network
        The loaded `Network` instance.

    Raises
    ------
    ValueError
        If the specified network does not exist in the database.
    RuntimeError
        If the database still uses the legacy, globally-keyed element tables.
    """
    ensure_current_schema(session)
    network_db = session.get(NetworkDB, name)
    if not network_db:
        raise ValueError(f"Network '{name}' not found.")
    network = network_db.to_pydantic()
    return network

save_branchtype

save_branchtype(branch_type: BranchType, session: Session)

Save a BranchType to the database.

This function converts a Pydantic BranchType model to its corresponding SQLAlchemy BranchTypeDB model and saves it to the database. If a BranchType with the same name already exists, it will be updated.

Parameters:

Name	Type	Description	Default
branch_type	BranchType	The BranchType instance to be saved.	required
session	Session	The SQLAlchemy session used for database operations.	required

Raises:

Type	Description
Exception	If there is an error during the database commit.

Source code in src/groundinsight/database/crud.py

def save_branchtype(branch_type: BranchType, session: Session):
    """
    Save a BranchType to the database.

    This function converts a Pydantic `BranchType` model to its corresponding SQLAlchemy
    `BranchTypeDB` model and saves it to the database. If a BranchType with the same name
    already exists, it will be updated.

    Parameters
    ----------
    branch_type : BranchType
        The BranchType instance to be saved.
    session : Session
        The SQLAlchemy session used for database operations.

    Raises
    ------
    Exception
        If there is an error during the database commit.
    """
    branch_type_db = BranchTypeDB.from_pydantic(branch_type)
    session.merge(branch_type_db)
    session.commit()

save_bustype

save_bustype(bus_type: BusType, session: Session)

Save a BusType to the database.

This function converts a Pydantic BusType model to its corresponding SQLAlchemy BusTypeDB model and saves it to the database. If a BusType with the same name already exists, it will be updated.

Parameters:

Name	Type	Description	Default
bus_type	BusType	The BusType instance to be saved.	required
session	Session	The SQLAlchemy session used for database operations.	required

Raises:

Type	Description
Exception	If there is an error during the database commit.

Source code in src/groundinsight/database/crud.py

def save_bustype(bus_type: BusType, session: Session):
    """
    Save a BusType to the database.

    This function converts a Pydantic `BusType` model to its corresponding SQLAlchemy
    `BusTypeDB` model and saves it to the database. If a BusType with the same name
    already exists, it will be updated.

    Parameters
    ----------
    bus_type : BusType
        The BusType instance to be saved.
    session : Session
        The SQLAlchemy session used for database operations.

    Raises
    ------
    Exception
        If there is an error during the database commit.

    """
    bus_type_db = BusTypeDB.from_pydantic(bus_type)
    session.merge(bus_type_db)
    session.commit()

save_network

save_network(
    network: Network,
    session: Session,
    overwrite: bool = False,
)

Save a Network to the database.

This function saves a comprehensive Network instance to the database, including all associated BusTypes, BranchTypes, Buses, Branches, Faults, Sources, and Paths. It handles the creation or updating of related entities and ensures referential integrity. If overwrite is set to True, an existing network with the same name will be deleted and replaced.

BusTypes and BranchTypes are a global catalogue: they are merged, exactly as :func:save_bustype and :func:save_branchtype do, so re-saving a network with an edited type definition updates the stored type instead of silently keeping the old one. Every other element is scoped to this network and is written under the composite key (network.name, element.name), so saving one network can never rewrite another's buses, branches, faults, sources or paths.

The whole operation runs in a single transaction. On overwrite, the delete of the previous revision is flushed but not committed before the replacement rows are written, so a failure anywhere in the save rolls the delete back as well and leaves the stored network untouched.

Parameters:

Name	Type	Description	Default
network	Network	The Network instance to be saved.	required
session	Session	The SQLAlchemy session used for database operations.	required
overwrite	bool	If True, existing network data with the same name will be overwritten. Defaults to False.	False

Raises:

Type	Description
ValueError	If the network already exists and overwrite is set to False, or if a path references a branch that is not part of network.branches.
RuntimeError	If the database still uses the legacy, globally-keyed element tables.
Exception	If there is an error during the database commit. The transaction is rolled back before the exception is re-raised.

Source code in src/groundinsight/database/crud.py

def save_network(network: Network, session: Session, overwrite: bool = False):
    """
    Save a Network to the database.

    This function saves a comprehensive `Network` instance to the database, including all
    associated BusTypes, BranchTypes, Buses, Branches, Faults, Sources, and Paths. It handles
    the creation or updating of related entities and ensures referential integrity. If `overwrite`
    is set to `True`, an existing network with the same name will be deleted and replaced.

    BusTypes and BranchTypes are a *global catalogue*: they are merged, exactly
    as :func:`save_bustype` and :func:`save_branchtype` do, so re-saving a
    network with an edited type definition updates the stored type instead of
    silently keeping the old one. Every other element is scoped to this network
    and is written under the composite key ``(network.name, element.name)``, so
    saving one network can never rewrite another's buses, branches, faults,
    sources or paths.

    The whole operation runs in a single transaction. On overwrite, the delete
    of the previous revision is *flushed but not committed* before the
    replacement rows are written, so a failure anywhere in the save rolls the
    delete back as well and leaves the stored network untouched.

    Parameters
    ----------
    network : Network
        The Network instance to be saved.
    session : Session
        The SQLAlchemy session used for database operations.
    overwrite : bool, optional
        If `True`, existing network data with the same name will be overwritten.
        Defaults to `False`.

    Raises
    ------
    ValueError
        If the network already exists and `overwrite` is set to `False`, or if a
        path references a branch that is not part of ``network.branches``.
    RuntimeError
        If the database still uses the legacy, globally-keyed element tables.
    Exception
        If there is an error during the database commit. The transaction is
        rolled back before the exception is re-raised.
    """
    ensure_current_schema(session)

    # Check for existing network
    existing_network = session.get(NetworkDB, network.name)
    if existing_network and not overwrite:
        raise ValueError(
            f"Network '{network.name}' already exists. Use overwrite=True to overwrite."
        )

    # Fail before touching the database rather than half-way through the write.
    _validate_path_segments(network)

    try:
        # Save BusTypes / BranchTypes -- merge, so an edited type definition
        # replaces the stored one instead of being ignored when the name exists.
        for bus in network.buses.values():
            session.merge(BusTypeDB.from_pydantic(bus.type))
        for branch in network.branches.values():
            session.merge(BranchTypeDB.from_pydantic(branch.type))

        if existing_network is not None:
            # Drop the previous revision including its child rows (the
            # relationships cascade), then flush so the primary keys are free
            # for the replacement rows. This stays inside the transaction
            # opened above -- no commit happens until the new rows are written.
            session.delete(existing_network)
            session.flush()

        network_db = NetworkDB.from_pydantic(network)
        network_db.active_fault_name = network.active_fault

        # ``Network`` element dictionaries are keyed by element name (see
        # ``Network.add_bus`` & co.), and ``to_pydantic`` rebuilds them from
        # the stored names, so the values -- not the dictionary keys -- are the
        # source of truth here. The enumeration index preserves the dictionary
        # order across the round-trip.
        network_db.buses = [
            BusDB.from_pydantic(bus, network.name, position)
            for position, bus in enumerate(network.buses.values())
        ]
        network_db.branches = [
            BranchDB.from_pydantic(branch, network.name, position)
            for position, branch in enumerate(network.branches.values())
        ]
        network_db.faults = [
            FaultDB.from_pydantic(fault, network.name, position)
            for position, fault in enumerate(network.faults.values())
        ]
        network_db.sources = [
            SourceDB.from_pydantic(source, network.name, position)
            for position, source in enumerate(network.sources.values())
        ]

        path_dbs = []
        for position, path in enumerate(network.paths.values()):
            path_db = PathDB.from_pydantic(path, network.name, position)
            # Segment order is semantically meaningful, so it is stored
            # explicitly instead of being left to the database.
            path_db.segments = [
                PathSegmentDB(
                    network_name=network.name,
                    path_name=path.name,
                    position=segment_position,
                    branch_name=segment.name,
                )
                for segment_position, segment in enumerate(path.segments)
            ]
            path_dbs.append(path_db)
        network_db.paths = path_dbs

        session.add(network_db)
        session.flush()

        # Commit the session only once every replacement row is on disk.
        session.commit()
    except Exception:
        # Roll the delete and the partial write back as one unit so a failed
        # overwrite cannot destroy the stored network.
        session.rollback()
        raise

ORM models

database_models

Database Models Module.

This module defines the SQLAlchemy ORM (Object-Relational Mapping) models corresponding to the core electrical network components in the GroundInsight package. Each database model facilitates the storage, retrieval, and manipulation of data related to BusTypes, BranchTypes, Buses, Branches, Faults, Sources, Paths, and Networks. The models include methods to convert between Pydantic models and SQLAlchemy database models, ensuring seamless data integration and persistence.

Ownership model

BusTypeDB and BranchTypeDB form a global catalogue: a type is identified by its name alone and is deliberately shared between networks, so re-saving an edited type updates every network that references it.

Every other element -- BusDB, BranchDB, FaultDB, SourceDB, PathDB and the PathSegmentDB rows -- belongs to exactly one network. This mirrors the Pydantic :class:~groundinsight.models.core_models.Network, whose element dictionaries only require uniqueness within a network. The primary key of those tables is therefore the composite (network_name, name); the relationships from NetworkDB are plain one-to-many collections with delete-orphan cascade, so dropping a network (or shrinking one on overwrite) removes its child rows instead of leaving them behind.

Earlier revisions keyed the child tables by name alone and linked them to their network through network_buses / network_branches / network_faults / network_sources / network_paths association tables. Two networks that happened to contain an element of the same name -- the default case for create_paths, which names every path path_1, path_2, ... -- then shared a single row, so saving one network silently rewrote the other. Databases written by those revisions are not readable by this schema; they are detected and rejected with an actionable error by :func:groundinsight.database.crud.ensure_current_schema.

Collection order is preserved explicitly through position columns rather than being left to the database. PathDB.segments in particular is order-sensitive: the solver walks the segments from the source bus onwards and fails if consecutive segments do not connect.

BranchDB

Bases: Base

BranchDB Model.

Represents a Branch in the database, including its properties, type, connected buses, the frequency-domain impedance dicts and -- when the type defines them -- the lumped RLCM dicts used by the transient solvers.

A branch belongs to exactly one network; the primary key is the composite (network_name, name). from_bus_name / to_bus_name reference buses of the same network, hence the composite foreign keys.

from_pydantic classmethod

from_pydantic(
    branch: Branch,
    network_name: str = None,
    position: int = 0,
)

Build a BranchDB row from a Pydantic :class:Branch.

Parameters:

Name	Type	Description	Default
branch	Branch	The branch to convert.	required
network_name	str	Name of the owning network. Part of the composite primary key.	None
position	int	Zero-based index of the branch inside Network.branches, used to restore the dictionary order on load. Defaults to 0.	0

Returns:

Type	Description
BranchDB	The unattached database row.

Source code in src/groundinsight/models/database_models.py

@classmethod
def from_pydantic(cls, branch: Branch, network_name: str = None, position: int = 0):
    """Build a ``BranchDB`` row from a Pydantic :class:`Branch`.

    Parameters
    ----------
    branch : Branch
        The branch to convert.
    network_name : str, optional
        Name of the owning network. Part of the composite primary key.
    position : int, optional
        Zero-based index of the branch inside ``Network.branches``, used
        to restore the dictionary order on load. Defaults to ``0``.

    Returns
    -------
    BranchDB
        The unattached database row.
    """
    # Convert impedance to JSON serializable format
    self_impedance = (
        {
            str(freq): {"real": imp.real, "imag": imp.imag}
            for freq, imp in branch.self_impedance.items()
        }
        if branch.self_impedance
        else {}
    )

    mutual_impedance = (
        {
            str(freq): {"real": imp.real, "imag": imp.imag}
            for freq, imp in branch.mutual_impedance.items()
        }
        if branch.mutual_impedance
        else {}
    )

    phase_impedance = (
        {
            str(freq): {"real": imp.real, "imag": imp.imag}
            for freq, imp in branch.phase_impedance.items()
        }
        if branch.phase_impedance
        else None
    )

    return cls(
        network_name=network_name,
        position=position,
        name=branch.name,
        description=branch.description,
        type_name=branch.type.name,
        length=branch.length,
        from_bus_name=branch.from_bus,
        to_bus_name=branch.to_bus,
        self_impedance=self_impedance,
        mutual_impedance=mutual_impedance,
        phase_impedance=phase_impedance,
        specific_earth_resistance=branch.specific_earth_resistance,
        parallel_coefficient=branch.parallel_coefficient,
        active=branch.active,
        R_self=_real_dict_to_json(branch.R_self),
        L_self=_real_dict_to_json(branch.L_self),
        C_self=_real_dict_to_json(branch.C_self),
        R_mutual=_real_dict_to_json(branch.R_mutual),
        M_mutual=_real_dict_to_json(branch.M_mutual),
    )

to_pydantic

to_pydantic()

Convert the row back into a Pydantic :class:Branch.

Returns:

Type	Description
Branch	The reconstructed branch.

Raises:

Type	Description
ValueError	If type_name does not resolve to a stored BranchTypeDB. See :meth:BusDB.to_pydantic for the rationale.

Source code in src/groundinsight/models/database_models.py

def to_pydantic(self):
    """Convert the row back into a Pydantic :class:`Branch`.

    Returns
    -------
    Branch
        The reconstructed branch.

    Raises
    ------
    ValueError
        If ``type_name`` does not resolve to a stored
        ``BranchTypeDB``. See :meth:`BusDB.to_pydantic` for the
        rationale.
    """
    if self.type is None:
        raise ValueError(
            f"Branch '{self.name}' of network '{self.network_name}' "
            f"references branch type '{self.type_name}', which is not "
            "stored in the database. The database is inconsistent -- "
            "re-save that branch type or point the branch at an existing "
            "one."
        )

    # Convert impedance JSON to Dict[float, ComplexNumber]
    self_impedance = (
        {
            float(freq): ComplexNumber(**value)
            for freq, value in self.self_impedance.items()
        }
        if self.self_impedance
        else {}
    )

    mutual_impedance = (
        {
            float(freq): ComplexNumber(**value)
            for freq, value in self.mutual_impedance.items()
        }
        if self.mutual_impedance
        else {}
    )

    phase_impedance = (
        {
            float(freq): ComplexNumber(**value)
            for freq, value in self.phase_impedance.items()
        }
        if self.phase_impedance
        else None
    )

    return Branch(
        name=self.name,
        description=self.description,
        type=self.type.to_pydantic(),
        length=self.length,
        from_bus=self.from_bus_name,
        to_bus=self.to_bus_name,
        self_impedance=self_impedance,
        mutual_impedance=mutual_impedance,
        phase_impedance=phase_impedance,
        specific_earth_resistance=self.specific_earth_resistance,
        parallel_coefficient=self.parallel_coefficient,
        active=True if self.active is None else bool(self.active),
        R_self=_real_dict_from_json(self.R_self),
        L_self=_real_dict_from_json(self.L_self),
        C_self=_real_dict_from_json(self.C_self),
        R_mutual=_real_dict_from_json(self.R_mutual),
        M_mutual=_real_dict_from_json(self.M_mutual),
    )

BranchTypeDB

Bases: Base

BranchTypeDB Model.

Represents a BranchType in the database, including the mandatory self_impedance_formula / mutual_impedance_formula and the optional lumped RLCM formulas for the transient solvers (R_self_formula, L_self_formula, C_self_formula, R_mutual_formula, M_mutual_formula).

BusDB

Bases: Base

BusDB Model.

Represents a Bus in the database. Carries the frequency-domain impedance dict and -- when the type defines them -- the lumped R, L and C dicts used by the transient solvers.

A bus belongs to exactly one network; the primary key is the composite (network_name, name) so two networks may each own a bus of the same name. position records the insertion order of Network.buses so a round-trip through the database returns the dictionary in the order it was written.

from_pydantic classmethod

from_pydantic(
    bus: Bus, network_name: str = None, position: int = 0
)

Build a BusDB row from a Pydantic :class:Bus.

Parameters:

Name	Type	Description	Default
bus	Bus	The bus to convert.	required
network_name	str	Name of the owning network. Part of the composite primary key; may be left out when the row is appended to NetworkDB.buses, in which case SQLAlchemy fills it in.	None
position	int	Zero-based index of the bus inside Network.buses, used to restore the dictionary order on load. Defaults to 0.	0

Returns:

Type	Description
BusDB	The unattached database row.

Source code in src/groundinsight/models/database_models.py

@classmethod
def from_pydantic(cls, bus: Bus, network_name: str = None, position: int = 0):
    """Build a ``BusDB`` row from a Pydantic :class:`Bus`.

    Parameters
    ----------
    bus : Bus
        The bus to convert.
    network_name : str, optional
        Name of the owning network. Part of the composite primary key;
        may be left out when the row is appended to
        ``NetworkDB.buses``, in which case SQLAlchemy fills it in.
    position : int, optional
        Zero-based index of the bus inside ``Network.buses``, used to
        restore the dictionary order on load. Defaults to ``0``.

    Returns
    -------
    BusDB
        The unattached database row.
    """
    # Convert impedance to JSON serializable format
    impedance = (
        {
            str(freq): {"real": imp.real, "imag": imp.imag}
            for freq, imp in bus.impedance.items()
        }
        if bus.impedance
        else {}
    )

    return cls(
        network_name=network_name,
        position=position,
        name=bus.name,
        description=bus.description,
        type_name=bus.type.name,
        specific_earth_resistance=bus.specific_earth_resistance,
        impedance=impedance,
        active=bus.active,
        R=_real_dict_to_json(bus.R),
        L=_real_dict_to_json(bus.L),
        C=_real_dict_to_json(bus.C),
    )

to_pydantic

to_pydantic()

Convert the row back into a Pydantic :class:Bus.

Returns:

Type	Description
Bus	The reconstructed bus.

Raises:

Type	Description
ValueError	If type_name does not resolve to a stored BusTypeDB -- a bus type deleted from a shared database, or a hand-edited row. Without the check the unresolved relationship surfaced as AttributeError: 'NoneType' object has no attribute 'to_pydantic', which names neither the bus nor the type.

Source code in src/groundinsight/models/database_models.py

def to_pydantic(self):
    """Convert the row back into a Pydantic :class:`Bus`.

    Returns
    -------
    Bus
        The reconstructed bus.

    Raises
    ------
    ValueError
        If ``type_name`` does not resolve to a stored ``BusTypeDB`` --
        a bus type deleted from a shared database, or a hand-edited
        row. Without the check the unresolved relationship surfaced as
        ``AttributeError: 'NoneType' object has no attribute
        'to_pydantic'``, which names neither the bus nor the type.
    """
    if self.type is None:
        raise ValueError(
            f"Bus '{self.name}' of network '{self.network_name}' references "
            f"bus type '{self.type_name}', which is not stored in the "
            "database. The database is inconsistent -- re-save that bus "
            "type or point the bus at an existing one."
        )

    # Convert impedance JSON to Dict[float, ComplexNumber]
    impedance = (
        {
            float(freq): ComplexNumber(**value)
            for freq, value in self.impedance.items()
        }
        if self.impedance
        else {}
    )

    return Bus(
        name=self.name,
        description=self.description,
        type=self.type.to_pydantic(),
        impedance=impedance,
        specific_earth_resistance=self.specific_earth_resistance,
        active=True if self.active is None else bool(self.active),
        R=_real_dict_from_json(self.R),
        L=_real_dict_from_json(self.L),
        C=_real_dict_from_json(self.C),
    )

BusTypeDB

Bases: Base

BusTypeDB Model.

Represents a BusType in the database, including its properties, the mandatory frequency-domain impedance_formula, the optional lumped RLC formulas for the transient solvers (R_formula, L_formula, C_formula) and the optional thermal-limit data of the earthing conductor and the earth electrode (EN 50522 / IEC 60949), consumed by :func:groundinsight.check_node_limits.

ComplexNumberDB

Bases: Base

ComplexNumberDB Model.

Represents a complex number with real and imaginary parts for storage in the database.

FaultDB

Bases: Base

FaultDB Model.

Represents a Fault in the database, including its properties and associated bus.

A fault belongs to exactly one network; the primary key is the composite (network_name, name) and bus_name references a bus of the same network.

from_pydantic classmethod

from_pydantic(
    fault: Fault,
    network_name: str = None,
    position: int = 0,
)

Build a FaultDB row from a Pydantic :class:Fault.

Parameters:

Name	Type	Description	Default
fault	Fault	The fault to convert.	required
network_name	str	Name of the owning network. Part of the composite primary key.	None
position	int	Zero-based index of the fault inside Network.faults, used to restore the dictionary order on load. Defaults to 0.	0

Returns:

Type	Description
FaultDB	The unattached database row.

Source code in src/groundinsight/models/database_models.py

@classmethod
def from_pydantic(cls, fault: Fault, network_name: str = None, position: int = 0):
    """Build a ``FaultDB`` row from a Pydantic :class:`Fault`.

    Parameters
    ----------
    fault : Fault
        The fault to convert.
    network_name : str, optional
        Name of the owning network. Part of the composite primary key.
    position : int, optional
        Zero-based index of the fault inside ``Network.faults``, used to
        restore the dictionary order on load. Defaults to ``0``.

    Returns
    -------
    FaultDB
        The unattached database row.
    """
    scalings = {str(freq): scale for freq, scale in fault.scalings.items()}
    return cls(
        network_name=network_name,
        position=position,
        name=fault.name,
        description=fault.description,
        bus_name=fault.bus,
        scalings=scalings,
        active=fault.active,
        t_k_s=fault.t_k_s,
        n_factor=fault.n_factor,
    )

NetworkDB

Bases: Base

NetworkDB Model.

Represents a Network in the database, including its properties and associated components such as buses, branches, faults, sources, and paths. It also tracks the active fault within the network.

The element collections are one-to-many with delete-orphan cascade: a bus, branch, fault, source or path row is owned by exactly one network, so deleting a network -- or replacing a collection on overwrite -- removes the rows it no longer contains instead of orphaning them in a global table. Each collection is ordered by the element's position column so the Pydantic dictionaries come back in insertion order.

PathDB

Bases: Base

PathDB Model.

Represents a Path in the database, including its properties, associated source and fault, and connected branches (segments).

A path belongs to exactly one network; the primary key is the composite (network_name, name). This matters in practice because :func:groundinsight.create_paths names every path path_1, path_2, ..., so path-name collisions between two saved networks are the rule rather than the exception.

from_pydantic classmethod

from_pydantic(
    path: Path, network_name: str = None, position: int = 0
)

Build a PathDB row from a Pydantic :class:Path.

Parameters:

Name	Type	Description	Default
path	Path	The path to convert.	required
network_name	str	Name of the owning network. Part of the composite primary key.	None
position	int	Zero-based index of the path inside Network.paths, used to restore the dictionary order on load. Defaults to 0.	0

Returns:

Type	Description
PathDB	The unattached database row. segments is left empty; the caller fills it once the branches of the network are known.

Source code in src/groundinsight/models/database_models.py

@classmethod
def from_pydantic(cls, path: Path, network_name: str = None, position: int = 0):
    """Build a ``PathDB`` row from a Pydantic :class:`Path`.

    Parameters
    ----------
    path : Path
        The path to convert.
    network_name : str, optional
        Name of the owning network. Part of the composite primary key.
    position : int, optional
        Zero-based index of the path inside ``Network.paths``, used to
        restore the dictionary order on load. Defaults to ``0``.

    Returns
    -------
    PathDB
        The unattached database row. ``segments`` is left empty; the
        caller fills it once the branches of the network are known.
    """
    return cls(
        network_name=network_name,
        position=position,
        name=path.name,
        description=path.description,
        source_name=path.source,
        fault_name=path.fault,
        # Segments will be added separately after the branches are saved
    )

to_pydantic

to_pydantic()

Convert the row back into a Pydantic :class:Path.

Returns:

Type	Description
Path	The reconstructed path, with segments in stored order.

Raises:

Type	Description
ValueError	If a stored segment references a branch that is not present in the owning network -- a corrupt or hand-edited database.

Source code in src/groundinsight/models/database_models.py

def to_pydantic(self):
    """Convert the row back into a Pydantic :class:`Path`.

    Returns
    -------
    Path
        The reconstructed path, with ``segments`` in stored order.

    Raises
    ------
    ValueError
        If a stored segment references a branch that is not present in
        the owning network -- a corrupt or hand-edited database.
    """
    segments = []
    for segment in self.segments:
        if segment.branch is None:
            raise ValueError(
                f"Path '{self.name}' of network '{self.network_name}' references "
                f"branch '{segment.branch_name}', which is not stored in that "
                "network. The database is inconsistent."
            )
        segments.append(segment.branch.to_pydantic())

    return Path(
        name=self.name,
        description=self.description,
        source=self.source_name,
        fault=self.fault_name,
        segments=segments,
    )

PathSegmentDB

Bases: Base

PathSegmentDB Model.

One ordered segment of a :class:PathDB, pointing at a branch of the same network. This used to be a plain association table without an ordering column, so a round-trip returned Path.segments in an unspecified order. Segment order is semantically meaningful -- the solver walks the segments from the source bus onwards and raises if a segment does not connect to the current bus -- hence the explicit position column, which is part of the primary key and drives PathDB.segments' order_by.

SourceDB

Bases: Base

SourceDB Model.

Represents a Source in the database. Both the legacy current-source mode and the Thevenin (voltage-source) mode are persisted in the same row. The source_type column selects which of values and the pair (voltage, source_impedance) carries the actual data; the other fields are stored as NULL.

A source belongs to exactly one network; the primary key is the composite (network_name, name) and bus_name references a bus of the same network.

from_pydantic classmethod

from_pydantic(
    source: Source,
    network_name: str = None,
    position: int = 0,
)

Build a SourceDB row from a Pydantic :class:Source.

Parameters:

Name	Type	Description	Default
source	Source	The source to convert.	required
network_name	str	Name of the owning network. Part of the composite primary key.	None
position	int	Zero-based index of the source inside Network.sources, used to restore the dictionary order on load. Defaults to 0.	0

Returns:

Type	Description
SourceDB	The unattached database row.

Source code in src/groundinsight/models/database_models.py

@classmethod
def from_pydantic(cls, source: Source, network_name: str = None, position: int = 0):
    """Build a ``SourceDB`` row from a Pydantic :class:`Source`.

    Parameters
    ----------
    source : Source
        The source to convert.
    network_name : str, optional
        Name of the owning network. Part of the composite primary key.
    position : int, optional
        Zero-based index of the source inside ``Network.sources``, used
        to restore the dictionary order on load. Defaults to ``0``.

    Returns
    -------
    SourceDB
        The unattached database row.
    """
    return cls(
        network_name=network_name,
        position=position,
        name=source.name,
        description=source.description,
        bus_name=source.bus,
        source_type=source.source_type,
        values=cls._freq_dict_to_json(source.values),
        voltage=cls._freq_dict_to_json(source.voltage),
        source_impedance=cls._freq_dict_to_json(source.source_impedance),
        i_k_a=source.i_k_a,
        r_to_x=source.r_to_x,
        kappa=source.kappa,
    )