Skip to content

Graph Algorithms

The voids.graph sub-package provides graph-theoretic utilities operating on the pore-throat topology stored in a Network.


Connectivity

voids.graph.connectivity

adjacency_matrix

adjacency_matrix(net)

Build the undirected pore adjacency matrix.

Parameters:

Name Type Description Default
net Network

Network whose pore connectivity is to be represented.

required

Returns:

Type Description
csr_matrix

Sparse symmetric matrix A with A[i, j] = 1 when pores i and j are connected by at least one throat.

Source code in src/voids/graph/connectivity.py
def adjacency_matrix(net: Network) -> sparse.csr_matrix:
    """Build the undirected pore adjacency matrix.

    Parameters
    ----------
    net :
        Network whose pore connectivity is to be represented.

    Returns
    -------
    scipy.sparse.csr_matrix
        Sparse symmetric matrix ``A`` with ``A[i, j] = 1`` when pores ``i`` and
        ``j`` are connected by at least one throat.
    """

    i = net.throat_conns[:, 0]
    j = net.throat_conns[:, 1]
    data = np.ones(net.Nt, dtype=float)
    A = sparse.coo_matrix((data, (i, j)), shape=(net.Np, net.Np))
    return (A + A.T).tocsr()

connected_components

connected_components(net)

Compute connected components of the pore graph.

Parameters:

Name Type Description Default
net Network

Network whose pore graph is analyzed.

required

Returns:

Type Description
int

Number of connected components.

ndarray

Integer component labels with shape (Np,).

Source code in src/voids/graph/connectivity.py
def connected_components(net: Network) -> tuple[int, np.ndarray]:
    """Compute connected components of the pore graph.

    Parameters
    ----------
    net :
        Network whose pore graph is analyzed.

    Returns
    -------
    int
        Number of connected components.
    numpy.ndarray
        Integer component labels with shape ``(Np,)``.
    """

    A = adjacency_matrix(net)
    n, labels = _cc(A, directed=False, return_labels=True)
    return int(n), labels.astype(np.int64)

spanning_component_ids

spanning_component_ids(net, axis, labels=None)

Return component identifiers that span a given sample axis.

Parameters:

Name Type Description Default
net Network

Network to analyze.

required
axis str

Axis whose inlet and outlet boundaries define the spanning criterion.

required
labels ndarray | None

Optional precomputed connected-component labels.

None

Returns:

Type Description
ndarray

Sorted array of component identifiers touching both the inlet and outlet boundary sets for the requested axis.

Raises:

Type Description
KeyError

If the required inlet or outlet labels are missing.

Source code in src/voids/graph/connectivity.py
def spanning_component_ids(net: Network, axis: str, labels: np.ndarray | None = None) -> np.ndarray:
    """Return component identifiers that span a given sample axis.

    Parameters
    ----------
    net :
        Network to analyze.
    axis :
        Axis whose inlet and outlet boundaries define the spanning criterion.
    labels :
        Optional precomputed connected-component labels.

    Returns
    -------
    numpy.ndarray
        Sorted array of component identifiers touching both the inlet and outlet
        boundary sets for the requested axis.

    Raises
    ------
    KeyError
        If the required inlet or outlet labels are missing.
    """

    if labels is None:
        _, labels = connected_components(net)
    inlet_name, outlet_name = _axis_boundary_labels(axis)
    if inlet_name not in net.pore_labels or outlet_name not in net.pore_labels:
        raise KeyError(f"Missing pore labels '{inlet_name}'/'{outlet_name}'")
    inlet_mask = net.pore_labels[inlet_name]
    outlet_mask = net.pore_labels[outlet_name]
    inlet_ids = np.unique(labels[inlet_mask])
    outlet_ids = np.unique(labels[outlet_mask])
    return np.asarray(np.intersect1d(inlet_ids, outlet_ids))

spanning_component_mask

spanning_component_mask(net, axis, labels=None)

Return a pore mask selecting axis-spanning connected components.

Parameters:

Name Type Description Default
net Network

Network to analyze.

required
axis str

Axis whose boundary labels define the spanning criterion.

required
labels ndarray | None

Optional precomputed connected-component labels.

None

Returns:

Type Description
ndarray

Boolean array with shape (Np,) selecting pores that belong to any spanning component.

Source code in src/voids/graph/connectivity.py
def spanning_component_mask(
    net: Network, axis: str, labels: np.ndarray | None = None
) -> np.ndarray:
    """Return a pore mask selecting axis-spanning connected components.

    Parameters
    ----------
    net :
        Network to analyze.
    axis :
        Axis whose boundary labels define the spanning criterion.
    labels :
        Optional precomputed connected-component labels.

    Returns
    -------
    numpy.ndarray
        Boolean array with shape ``(Np,)`` selecting pores that belong to any
        spanning component.
    """

    if labels is None:
        _, labels = connected_components(net)
    comp_ids = spanning_component_ids(net, axis=axis, labels=labels)
    return np.isin(labels, comp_ids)

induced_subnetwork

induced_subnetwork(net, pore_mask)

Return the induced subnetwork associated with a pore subset.

Parameters:

Name Type Description Default
net Network

Network to subset.

required
pore_mask ndarray

Boolean mask with shape (Np,) selecting retained pores.

required

Returns:

Type Description
tuple

(subnet, pore_indices, throat_mask) where subnet is the induced network, pore_indices are the retained pore indices in the original network, and throat_mask selects retained throats in the original network.

Source code in src/voids/graph/connectivity.py
def induced_subnetwork(
    net: Network, pore_mask: np.ndarray
) -> tuple[Network, np.ndarray, np.ndarray]:
    """Return the induced subnetwork associated with a pore subset.

    Parameters
    ----------
    net :
        Network to subset.
    pore_mask :
        Boolean mask with shape ``(Np,)`` selecting retained pores.

    Returns
    -------
    tuple
        ``(subnet, pore_indices, throat_mask)`` where ``subnet`` is the induced
        network, ``pore_indices`` are the retained pore indices in the original
        network, and ``throat_mask`` selects retained throats in the original network.
    """

    pore_mask = np.asarray(pore_mask, dtype=bool)
    if pore_mask.shape != (net.Np,):
        raise ValueError("pore_mask must have shape (Np,)")
    pore_indices = np.flatnonzero(pore_mask)
    local = -np.ones(net.Np, dtype=int)
    local[pore_indices] = np.arange(pore_indices.size)
    throat_mask = pore_mask[net.throat_conns[:, 0]] & pore_mask[net.throat_conns[:, 1]]
    throat_conns = local[net.throat_conns[throat_mask]]

    # Build pore/throat data and label dicts, subsetting only arrays whose first
    # dimension matches the number of pores/throats in the original network.
    pore_data: dict[str, np.ndarray] = {}
    for k, v in net.pore.items():
        arr = np.asarray(v)
        if arr.shape and arr.shape[0] == net.Np:
            pore_data[k] = arr[pore_indices]
        else:
            pore_data[k] = v

    throat_data: dict[str, np.ndarray] = {}
    for k, v in net.throat.items():
        arr = np.asarray(v)
        if arr.shape and arr.shape[0] == net.Nt:
            throat_data[k] = arr[throat_mask]
        else:
            throat_data[k] = v

    pore_labels: dict[str, np.ndarray] = {}
    for k, v in net.pore_labels.items():
        arr = np.asarray(v)
        if arr.shape and arr.shape[0] == net.Np:
            pore_labels[k] = arr[pore_indices]
        else:
            pore_labels[k] = v

    throat_labels: dict[str, np.ndarray] = {}
    for k, v in net.throat_labels.items():
        arr = np.asarray(v)
        if arr.shape and arr.shape[0] == net.Nt:
            throat_labels[k] = arr[throat_mask]
        else:
            throat_labels[k] = v

    subnet = Network(
        throat_conns=throat_conns,
        pore_coords=net.pore_coords[pore_indices],
        sample=net.sample,
        provenance=net.provenance,
        schema_version=net.schema_version,
        pore=pore_data,
        throat=throat_data,
        pore_labels=pore_labels,
        throat_labels=throat_labels,
        extra={**net.extra},
    )
    return subnet, pore_indices, throat_mask

spanning_subnetwork

spanning_subnetwork(net, axis, labels=None)

Return the induced subnetwork formed by axis-spanning components.

Parameters:

Name Type Description Default
net Network

Network to subset.

required
axis str

Axis whose inlet/outlet labels define the spanning criterion.

required
labels ndarray | None

Optional connected-component labels.

None

Returns:

Type Description
tuple

(subnet, pore_indices, throat_mask) for the axis-spanning subnetwork.

Source code in src/voids/graph/connectivity.py
def spanning_subnetwork(
    net: Network, axis: str, labels: np.ndarray | None = None
) -> tuple[Network, np.ndarray, np.ndarray]:
    """Return the induced subnetwork formed by axis-spanning components.

    Parameters
    ----------
    net :
        Network to subset.
    axis :
        Axis whose inlet/outlet labels define the spanning criterion.
    labels :
        Optional connected-component labels.

    Returns
    -------
    tuple
        ``(subnet, pore_indices, throat_mask)`` for the axis-spanning subnetwork.
    """

    pore_mask = spanning_component_mask(net, axis=axis, labels=labels)
    return induced_subnetwork(net, pore_mask)

Incidence

voids.graph.incidence

incidence_matrix

incidence_matrix(net)

Build the throat-to-pore incidence matrix.

Parameters:

Name Type Description Default
net Network

Network whose incidence structure is requested.

required

Returns:

Type Description
csr_matrix

Sparse matrix B with shape (Nt, Np). For each throat t connecting pores i and j, row t stores +1 at one endpoint and -1 at the other.

Notes

The orientation is arbitrary but fixed by the ordering in net.throat_conns. This is sufficient to define discrete pressure differences and fluxes consistently.

Source code in src/voids/graph/incidence.py
def incidence_matrix(net: Network) -> sparse.csr_matrix:
    """Build the throat-to-pore incidence matrix.

    Parameters
    ----------
    net :
        Network whose incidence structure is requested.

    Returns
    -------
    scipy.sparse.csr_matrix
        Sparse matrix ``B`` with shape ``(Nt, Np)``. For each throat ``t``
        connecting pores ``i`` and ``j``, row ``t`` stores ``+1`` at one
        endpoint and ``-1`` at the other.

    Notes
    -----
    The orientation is arbitrary but fixed by the ordering in
    ``net.throat_conns``. This is sufficient to define discrete pressure
    differences and fluxes consistently.
    """

    rows = np.repeat(np.arange(net.Nt), 2)
    cols = net.throat_conns.reshape(-1)
    data = np.tile(np.array([1.0, -1.0]), net.Nt)
    return sparse.coo_matrix((data, (rows, cols)), shape=(net.Nt, net.Np)).tocsr()

Metrics

voids.graph.metrics

ConnectivitySummary dataclass

Summarize graph-level connectivity statistics for a pore network.

Attributes:

Name Type Description
n_components int

Number of connected components in the pore graph.

giant_component_fraction float

Fraction of pores belonging to the largest connected component.

isolated_pore_fraction float

Fraction of pores with coordination number zero.

dead_end_fraction float

Fraction of pores with coordination number one.

mean_coordination float

Mean pore coordination number.

coordination_histogram dict[int, int]

Histogram mapping coordination number to pore count.

spans dict[str, bool]

Dictionary indicating whether any component spans each principal axis.

Source code in src/voids/graph/metrics.py
@dataclass(slots=True)
class ConnectivitySummary:
    """Summarize graph-level connectivity statistics for a pore network.

    Attributes
    ----------
    n_components :
        Number of connected components in the pore graph.
    giant_component_fraction :
        Fraction of pores belonging to the largest connected component.
    isolated_pore_fraction :
        Fraction of pores with coordination number zero.
    dead_end_fraction :
        Fraction of pores with coordination number one.
    mean_coordination :
        Mean pore coordination number.
    coordination_histogram :
        Histogram mapping coordination number to pore count.
    spans :
        Dictionary indicating whether any component spans each principal axis.
    """

    n_components: int
    giant_component_fraction: float
    isolated_pore_fraction: float
    dead_end_fraction: float
    mean_coordination: float
    coordination_histogram: dict[int, int]
    spans: dict[str, bool]

coordination_numbers

coordination_numbers(net)

Compute pore coordination numbers.

Parameters:

Name Type Description Default
net Network

Network whose pore degrees are requested.

required

Returns:

Type Description
ndarray

Integer array with shape (Np,) where each entry equals the number of throats incident on the corresponding pore.

Source code in src/voids/graph/metrics.py
def coordination_numbers(net: Network) -> np.ndarray:
    """Compute pore coordination numbers.

    Parameters
    ----------
    net :
        Network whose pore degrees are requested.

    Returns
    -------
    numpy.ndarray
        Integer array with shape ``(Np,)`` where each entry equals the number of
        throats incident on the corresponding pore.
    """

    deg = np.zeros(net.Np, dtype=np.int64)
    np.add.at(deg, net.throat_conns[:, 0], 1)
    np.add.at(deg, net.throat_conns[:, 1], 1)
    return deg

connectivity_metrics

connectivity_metrics(net)

Compute a compact set of connectivity diagnostics for a pore graph.

Parameters:

Name Type Description Default
net Network

Network to analyze.

required

Returns:

Type Description
ConnectivitySummary

Aggregate summary of component counts, degree statistics, and axis spanning information.

Source code in src/voids/graph/metrics.py
def connectivity_metrics(net: Network) -> ConnectivitySummary:
    """Compute a compact set of connectivity diagnostics for a pore graph.

    Parameters
    ----------
    net :
        Network to analyze.

    Returns
    -------
    ConnectivitySummary
        Aggregate summary of component counts, degree statistics, and axis
        spanning information.
    """

    n_comp, labels = connected_components(net)
    counts = np.bincount(labels, minlength=n_comp)
    deg = coordination_numbers(net)
    hist_keys, hist_counts = np.unique(deg, return_counts=True)
    spans: dict[str, bool] = {}
    for ax in ("x", "y", "z"):
        try:
            spans[ax] = spanning_component_ids(net, ax, labels=labels).size > 0
        except KeyError:
            continue
    return ConnectivitySummary(
        n_components=n_comp,
        giant_component_fraction=float(counts.max() / net.Np if net.Np else 0.0),
        isolated_pore_fraction=float(np.mean(deg == 0)),
        dead_end_fraction=float(np.mean(deg == 1)),
        mean_coordination=float(np.mean(deg) if deg.size else 0.0),
        coordination_histogram={int(k): int(v) for k, v in zip(hist_keys, hist_counts)},
        spans=spans,
    )