NetGraph API Reference (Auto-Generated)¶

This is the complete auto-generated API documentation for NetGraph. For a curated, example-driven API guide, see api.md.

Quick links:

Generated from source code on: August 20, 2025 at 22:00 UTC

Modules auto-discovered: 73

ngraph.cli¶

Command-line interface for NetGraph.

main(argv: 'Optional[List[str]]' = None) -> 'None'¶

Entry point for the ngraph command.

Args: argv: Optional list of command-line arguments. If None, sys.argv is used.

ngraph.components¶

Component and ComponentsLibrary classes for hardware capex/power modeling.

Component¶

A generic component that can represent chassis, line cards, optics, etc. Components can have nested children, each with their own capex, power, etc.

Attributes: name (str): Name of the component (e.g., "SpineChassis" or "400G-LR4"). component_type (str): A string label (e.g., "chassis", "linecard", "optic"). description (str): A human-readable description of this component. capex (float): Monetary capex of a single instance of this component. power_watts (float): Typical/nominal power usage (watts) for one instance. power_watts_max (float): Maximum/peak power usage (watts) for one instance. capacity (float): A generic capacity measure (e.g., platform capacity). ports (int): Number of ports if relevant for this component. count (int): How many identical copies of this component are present. attrs (Dict[str, Any]): Arbitrary key-value attributes for extra metadata. children (Dict[str, Component]): Nested child components (e.g., line cards inside a chassis), keyed by child name.

Attributes:

name (str)
component_type (str) = generic
description (str)
capex (float) = 0.0
power_watts (float) = 0.0
power_watts_max (float) = 0.0
capacity (float) = 0.0
ports (int) = 0
count (int) = 1
attrs (Dict[str, Any]) = {}
children (Dict[str, Component]) = {}

Methods:

as_dict(self, include_children: 'bool' = True) -> 'Dict[str, Any]' - Returns a dictionary containing all properties of this component.
total_capacity(self) -> 'float' - Computes the total (recursive) capacity of this component,
total_capex(self) -> 'float' - Computes total capex including children, multiplied by count.
total_power(self) -> 'float' - Computes the total typical (recursive) power usage of this component,
total_power_max(self) -> 'float' - Computes the total peak (recursive) power usage of this component,

ComponentsLibrary¶

Holds a collection of named Components. Each entry is a top-level "template" that can be referenced for cost/power/capacity lookups, possibly with nested children.

Example (YAML-like): components: BigSwitch: component_type: chassis cost: 20000 power_watts: 1750 capacity: 25600 children: PIM16Q-16x200G: component_type: linecard cost: 1000 power_watts: 10 ports: 16 count: 8 200G-FR4: component_type: optic cost: 2000 power_watts: 6 power_watts_max: 6.5

Attributes:

components (Dict[str, Component]) = {}

Methods:

clone(self) -> 'ComponentsLibrary' - Creates a deep copy of this ComponentsLibrary.
from_dict(data: 'Dict[str, Any]') -> 'ComponentsLibrary' - Constructs a ComponentsLibrary from a dictionary of raw component definitions.
from_yaml(yaml_str: 'str') -> 'ComponentsLibrary' - Constructs a ComponentsLibrary from a YAML string. If the YAML contains
get(self, name: 'str') -> 'Optional[Component]' - Retrieves a Component by its name from the library.
merge(self, other: 'ComponentsLibrary', override: 'bool' = True) -> 'ComponentsLibrary' - Merges another ComponentsLibrary into this one. By default (override=True),

resolve_link_end_components(attrs: 'Dict[str, Any]', library: 'ComponentsLibrary') -> 'tuple[tuple[Optional[Component], float, bool], tuple[Optional[Component], float, bool], bool]'¶

Resolve per-end hardware components for a link.

Input format inside link.attrs:

Structured mapping under hardware key only: {"hardware": {"source": {"component": NAME, "count": N}, "target": {"component": NAME, "count": N}}}

Args: attrs: Link attributes mapping. library: Components library for lookups.

Exclusive usage:

Optional exclusive: true per end indicates unsharable usage.

For exclusive ends, validation and BOM counting should round-up counts to integers.

Returns: ((src_comp, src_count, src_exclusive), (dst_comp, dst_count, dst_exclusive), per_end_specified) where components may be None if name is absent/unknown. per_end_specified is True when a structured per-end mapping is present.

resolve_node_hardware(attrs: 'Dict[str, Any]', library: 'ComponentsLibrary') -> 'Tuple[Optional[Component], float]'¶

Resolve node hardware from attrs['hardware'].

Expects the mapping: {"hardware": {"component": NAME, "count": N}}. count defaults to 1 if missing or invalid. If component is missing or unknown, returns (None, 1.0).

Args: attrs: Node attributes mapping. library: Component library used for lookups.

Returns: Tuple of (component or None, positive multiplier).

totals_with_multiplier(comp: 'Component', hw_count: 'float') -> 'Tuple[float, float, float]'¶

Return (capex, power_watts, capacity) totals multiplied by hw_count.

Args: comp: Component definition (may include nested children and internal count). hw_count: External multiplier (e.g., number of modules used for a link or node).

Returns: Tuple of total capex, total power (typical), and total capacity as floats.

ngraph.config¶

Configuration classes for NetGraph components.

TrafficManagerConfig¶

Configuration for traffic demand placement estimation.

Attributes:

default_rounds (int) = 5
min_rounds (int) = 5
max_rounds (int) = 100
ratio_base (int) = 5
ratio_multiplier (int) = 5

Methods:

estimate_rounds(self, demand_capacity_ratio: float) -> int - Calculate placement rounds based on demand to capacity ratio.

ngraph.explorer¶

NetworkExplorer class for analyzing network hierarchy and structure.

ExternalLinkBreakdown¶

Holds stats for external links to a particular other subtree.

Attributes: link_count (int): Number of links to that other subtree. link_capacity (float): Sum of capacities for those links.

Attributes:

link_count (int) = 0
link_capacity (float) = 0.0

LinkCapacityIssue¶

Represents a link capacity constraint violation in active topology.

Attributes: source: Source node name. target: Target node name. capacity: Configured link capacity. limit: Effective capacity limit from per-end hardware (min of ends). reason: Brief reason tag.

Attributes:

source (str)
target (str)
capacity (float)
limit (float)
reason (str)

NetworkExplorer¶

Provides hierarchical exploration of a Network, computing statistics in two modes: 'all' (ignores disabled) and 'active' (only enabled).

Methods:

explore_network(network: 'Network', components_library: 'Optional[ComponentsLibrary]' = None, strict_validation: 'bool' = True) -> 'NetworkExplorer' - Build a NetworkExplorer, constructing a tree plus 'all' and 'active' stats.
get_bom(self, include_disabled: 'bool' = True) -> 'Dict[str, float]' - Return aggregated hardware BOM for the whole network.
get_bom_by_path(self, path: 'str', include_disabled: 'bool' = True) -> 'Dict[str, float]' - Return the hardware BOM for a specific hierarchy path.
get_bom_map(self, include_disabled: 'bool' = True, include_root: 'bool' = True, root_label: 'str' = '') -> 'Dict[str, Dict[str, float]]' - Return a mapping from hierarchy path to BOM for each subtree.
get_link_issues(self) -> 'List[LinkCapacityIssue]' - Return recorded link capacity issues discovered in non-strict mode.
get_node_utilization(self, include_disabled: 'bool' = True) -> 'List[NodeUtilization]' - Return hardware utilization per node based on active topology.
print_tree(self, node: 'Optional[TreeNode]' = None, indent: 'int' = 0, max_depth: 'Optional[int]' = None, skip_leaves: 'bool' = False, detailed: 'bool' = False, include_disabled: 'bool' = True, max_external_lines: 'Optional[int]' = None, line_prefix: 'str' = '') -> 'None' - Print the hierarchy from 'node' down (default: root).

NodeUtilization¶

Per-node hardware utilization snapshot based on active topology.

Attributes: node_name: Fully qualified node name. component_name: Hardware component name if present. hw_count: Hardware multiplicity used for capacity/power scaling. capacity_supported: Total capacity supported by node hardware. attached_capacity_active: Sum of capacities of enabled adjacent links where the opposite endpoint is also enabled. capacity_utilization: Ratio of attached to supported capacity (0.0 when N/A). ports_available: Total port equivalents available on the node (0.0 when N/A). ports_used: Sum of port equivalents used by per-end link optics attached to this node on active links. ports_utilization: Ratio of used to available ports (0.0 when N/A). capacity_violation: True if attached capacity exceeds supported capacity. ports_violation: True if used ports exceed available ports. disabled: True if the node itself is disabled.

Attributes:

node_name (str)
component_name (Optional[str])
hw_count (float)
capacity_supported (float)
attached_capacity_active (float)
capacity_utilization (float)
ports_available (float)
ports_used (float)
ports_utilization (float)
capacity_violation (bool)
ports_violation (bool)
disabled (bool)

TreeNode¶

Represents a node in the hierarchical tree.

Attributes: name (str): Name/label of this node. parent (Optional[TreeNode]): Pointer to the parent tree node. children (Dict[str, TreeNode]): Mapping of child name -> child TreeNode. subtree_nodes (Set[str]): Node names in the subtree (all nodes, ignoring disabled). active_subtree_nodes (Set[str]): Node names in the subtree (only enabled). stats (TreeStats): Aggregated stats for "all" view. active_stats (TreeStats): Aggregated stats for "active" (only enabled) view. raw_nodes (List[Node]): Direct Node objects at this hierarchy level.

Attributes:

name (str)
parent (Optional[TreeNode])
children (Dict[str, TreeNode]) = {}
subtree_nodes (Set[str]) = set()
active_subtree_nodes (Set[str]) = set()
stats (TreeStats) = TreeStats(node_count=0, internal_link_count=0, internal_link_capacity=0.0, external_link_count=0, external_link_capacity=0.0, external_link_details={}, total_capex=0.0, total_power=0.0, bom={}, active_bom={})
active_stats (TreeStats) = TreeStats(node_count=0, internal_link_count=0, internal_link_capacity=0.0, external_link_count=0, external_link_capacity=0.0, external_link_details={}, total_capex=0.0, total_power=0.0, bom={}, active_bom={})
raw_nodes (List[Node]) = []

Methods:

add_child(self, child_name: 'str') -> 'TreeNode' - Ensure a child node named 'child_name' exists and return it.
is_leaf(self) -> 'bool' - Return True if this node has no children.

TreeStats¶

Aggregated statistics for a single tree node (subtree).

Attributes: node_count (int): Total number of nodes in this subtree. internal_link_count (int): Number of internal links in this subtree. internal_link_capacity (float): Sum of capacities for those internal links. external_link_count (int): Number of external links from this subtree to another. external_link_capacity (float): Sum of capacities for those external links. external_link_details (Dict[str, ExternalLinkBreakdown]): Breakdown by other subtree path. total_capex (float): Cumulative capex (nodes + links). total_power (float): Cumulative power (nodes + links).

Attributes:

node_count (int) = 0
internal_link_count (int) = 0
internal_link_capacity (float) = 0.0
external_link_count (int) = 0
external_link_capacity (float) = 0.0
external_link_details (Dict[str, ExternalLinkBreakdown]) = {}
total_capex (float) = 0.0
total_power (float) = 0.0
bom (Dict[str, float]) = {}
active_bom (Dict[str, float]) = {}

ngraph.logging¶

Centralized logging configuration for NetGraph.

disable_debug_logging() -> None¶

Disable debug logging, set to INFO level.

enable_debug_logging() -> None¶

Enable debug logging for the entire package.

get_logger(name: str) -> logging.Logger¶

Get a logger with NetGraph's standard configuration.

This is the main function that should be used throughout the package. All loggers will inherit from the root 'ngraph' logger configuration.

Args: name: Logger name (typically name from calling module).

Returns: Configured logger instance.

reset_logging() -> None¶

Reset logging configuration (mainly for testing).

set_global_log_level(level: int) -> None¶

Set the log level for all NetGraph loggers.

Args: level: Logging level (e.g., logging.DEBUG, logging.INFO).

setup_root_logger(level: int = 20, format_string: Optional[str] = None, handler: Optional[logging.Handler] = None) -> None¶

Set up the root NetGraph logger with a single handler.

This should only be called once to avoid duplicate handlers.

Args: level: Logging level (default: INFO). format_string: Custom format string (optional). handler: Custom handler (optional, defaults to StreamHandler).

ngraph.report¶

Standalone report generation for NetGraph analysis results.

Generates Jupyter notebooks and optional HTML reports from results.json. This module is separate from workflow execution to allow independent analysis in notebooks.

ReportGenerator¶

Generate notebooks and HTML reports from a results document.

The notebook includes environment setup, results loading, overview, and per-step analysis sections chosen via the analysis registry.

Methods:

generate_html_report(self, notebook_path: 'Path' = PosixPath('analysis.ipynb'), html_path: 'Path' = PosixPath('analysis_report.html'), include_code: 'bool' = False) -> 'Path' - Render the notebook to HTML using nbconvert.
generate_notebook(self, output_path: 'Path' = PosixPath('analysis.ipynb')) -> 'Path' - Create a Jupyter notebook with analysis scaffold.
load_results(self) -> 'None' - Load and validate the JSON results file into memory.

ngraph.scenario¶

Scenario class for defining network analysis workflows from YAML.

Scenario¶

Represents a complete scenario for building and executing network workflows.

This scenario includes:

A network (nodes/links), constructed via blueprint expansion.
A failure policy set (one or more named failure policies).
A traffic matrix set containing one or more named traffic matrices.
A list of workflow steps to execute.
A results container for storing outputs.
A components_library for hardware/optics definitions.
A seed for reproducible random operations (optional).

Typical usage example:

scenario = Scenario.from_yaml(yaml_str, default_components=default_lib)
scenario.run()
# Inspect scenario.results

Attributes:

network (Network)
workflow (List[WorkflowStep])
failure_policy_set (FailurePolicySet) = FailurePolicySet(policies={})
traffic_matrix_set (TrafficMatrixSet) = TrafficMatrixSet(matrices={})
results (Results) = Results(_store={}, _metadata={}, _active_step=None, _scenario={})
components_library (ComponentsLibrary) = ComponentsLibrary(components={})
seed (Optional[int])

Methods:

from_yaml(yaml_str: 'str', default_components: 'Optional[ComponentsLibrary]' = None) -> 'Scenario' - Constructs a Scenario from a YAML string, optionally merging
run(self) -> 'None' - Executes the scenario's workflow steps in order.

ngraph.seed_manager¶

Deterministic seed derivation to avoid global random.seed() order dependencies.

SeedManager¶

Manages deterministic seed derivation for isolated component reproducibility.

Global random.seed() creates order dependencies and component interference. SeedManager derives unique seeds per component from a master seed using SHA-256, ensuring reproducible results regardless of execution order or parallelism.

Usage: seed_mgr = SeedManager(42) failure_seed = seed_mgr.derive_seed("failure_policy", "default") analysis_seed = seed_mgr.derive_seed("workflow", "capacity_analysis", 0)

Methods:

create_random_state(self, *components: 'Any') -> 'random.Random' - Create a new Random instance with derived seed.
derive_seed(self, *components: 'Any') -> 'Optional[int]' - Derive a deterministic seed from master seed and component identifiers.
seed_global_random(self, *components: 'Any') -> 'None' - Seed the global random module with derived seed.

ngraph.yaml_utils¶

Utilities for handling YAML parsing quirks and common operations.

normalize_yaml_dict_keys(data: Dict[Any, ~V]) -> Dict[str, ~V]¶

Normalize dictionary keys from YAML parsing to ensure consistent string keys.

YAML 1.1 boolean keys (e.g., true, false, yes, no, on, off) get converted to Python True/False boolean values. This function converts them to predictable string representations ("True"/"False") and ensures all keys are strings.

Args: data: Dictionary that may contain boolean or other non-string keys from YAML parsing

Returns: Dictionary with all keys converted to strings, boolean keys converted to "True"/"False"

Examples: >>> normalize_yaml_dict_keys({True: "value1", False: "value2", "normal": "value3"}) {"True": "value1", "False": "value2", "normal": "value3"}

>>> # In YAML: true:, yes:, on: all become Python True
>>> # In YAML: false:, no:, off: all become Python False

ngraph.graph.convert¶

Graph conversion utilities between StrictMultiDiGraph and NetworkX graphs.

Functions in this module consolidate or expand multi-edges and can preserve original edge data for reversion through a special _uv_edges attribute.

from_digraph(nx_graph: networkx.classes.digraph.DiGraph) -> ngraph.graph.strict_multidigraph.StrictMultiDiGraph¶

Convert a revertible NetworkX DiGraph to a StrictMultiDiGraph.

This function reconstructs the original StrictMultiDiGraph by restoring multi-edge information from the '_uv_edges' attribute of each edge.

Args: nx_graph: A revertible NetworkX DiGraph with _uv_edges attributes.

Returns: A StrictMultiDiGraph reconstructed from the input DiGraph.

from_graph(nx_graph: networkx.classes.graph.Graph) -> ngraph.graph.strict_multidigraph.StrictMultiDiGraph¶

Convert a revertible NetworkX Graph to a StrictMultiDiGraph.

Restores the original multi-edge structure from the '_uv_edges' attribute stored in each consolidated edge.

Args: nx_graph: A revertible NetworkX Graph with _uv_edges attributes.

Returns: A StrictMultiDiGraph reconstructed from the input Graph.

to_digraph(graph: ngraph.graph.strict_multidigraph.StrictMultiDiGraph, edge_func: Optional[Callable[[ngraph.graph.strict_multidigraph.StrictMultiDiGraph, Hashable, Hashable, dict], dict]] = None, revertible: bool = True) -> networkx.classes.digraph.DiGraph¶

Convert a StrictMultiDiGraph to a NetworkX DiGraph.

This function consolidates multi-edges between nodes into a single edge. Optionally, a custom edge function can be provided to compute edge attributes. If revertible is True, the original multi-edge data is stored in the '_uv_edges' attribute of each consolidated edge, allowing for later reversion.

Args: graph: The StrictMultiDiGraph to convert. edge_func: Optional function to compute consolidated edge attributes. The callable receives (graph, u, v, edges) and returns a dict. revertible: If True, store the original multi-edge data for reversion.

Returns: A NetworkX DiGraph representing the input graph.

to_graph(graph: ngraph.graph.strict_multidigraph.StrictMultiDiGraph, edge_func: Optional[Callable[[ngraph.graph.strict_multidigraph.StrictMultiDiGraph, Hashable, Hashable, dict], dict]] = None, revertible: bool = True) -> networkx.classes.graph.Graph¶

Convert a StrictMultiDiGraph to a NetworkX Graph.

This function works similarly to to_digraph but returns an undirected graph.

Args: graph: The StrictMultiDiGraph to convert. edge_func: Optional function to compute consolidated edge attributes. revertible: If True, store the original multi-edge data for reversion.

Returns: A NetworkX Graph representing the input graph.

ngraph.graph.io¶

Graph serialization functions for node-link and edge-list formats.

edgelist_to_graph(lines: 'Iterable[str]', columns: 'List[str]', separator: 'str' = ' ', graph: 'Optional[StrictMultiDiGraph]' = None, source: 'str' = 'src', target: 'str' = 'dst', key: 'str' = 'key') -> 'StrictMultiDiGraph'¶

Build or update a StrictMultiDiGraph from an edge list.

Each line in the input is split by the specified separator into tokens. These tokens are mapped to column names provided in columns. The tokens corresponding to source and target become the node IDs. If a key column exists, its token is used as the edge ID; remaining tokens are added as edge attributes.

Args: lines: An iterable of strings, each representing one edge. columns: A list of column names, e.g. ["src", "dst", "cost"]. separator: The separator used to split each line (default is a space). graph: An existing StrictMultiDiGraph to update; if None, a new graph is created. source: The column name for the source node ID. target: The column name for the target node ID. key: The column name for a custom edge ID (if present).

Returns: The updated (or newly created) StrictMultiDiGraph.

Raises: RuntimeError: If a line does not match the expected number of columns.

graph_to_edgelist(graph: 'StrictMultiDiGraph', columns: 'Optional[List[str]]' = None, separator: 'str' = ' ', source_col: 'str' = 'src', target_col: 'str' = 'dst', key_col: 'str' = 'key') -> 'List[str]'¶

Convert a StrictMultiDiGraph into an edge-list text representation.

Each line in the output represents one edge with tokens joined by the given separator. By default, the output columns are: [source_col, target_col, key_col] + sorted(edge_attribute_names)

If an explicit list of columns is provided, those columns (in that order) are used, and any missing values are output as an empty string.

Args: graph: The StrictMultiDiGraph to export. columns: Optional list of column names. If None, they are auto-generated. separator: The string used to join tokens (default is a space). source_col: The column name for the source node (default "src"). target_col: The column name for the target node (default "dst"). key_col: The column name for the edge key (default "key").

Returns: A list of strings, each representing one edge in the specified column format.

graph_to_node_link(graph: 'StrictMultiDiGraph') -> 'Dict[str, Any]'¶

Convert a StrictMultiDiGraph into a node-link dict representation.

This representation is suitable for JSON serialization (e.g., for D3.js or Nx formats).

The returned dict has the following structure: { "graph": { ... top-level graph attributes ... }, "nodes": [ {"id": node_id, "attr": { ... node attributes ... }}, ... ], "links": [ { "source": , "target": , "key": , "attr": { ... edge attributes ... } }, ... ] }

Args: graph: The StrictMultiDiGraph to convert.

Returns: A dict containing the 'graph' attributes, list of 'nodes', and list of 'links'.

node_link_to_graph(data: 'Dict[str, Any]') -> 'StrictMultiDiGraph'¶

Reconstruct a StrictMultiDiGraph from its node-link dict representation.

Expected input format: { "graph": { ... graph attributes ... }, "nodes": [ {"id": , "attr": { ... node attributes ... }}, ... ], "links": [ { "source": , "target": , "key": , "attr": { ... edge attributes ... } }, ... ] }

Args: data: A dict representing the node-link structure.

Returns: A StrictMultiDiGraph reconstructed from the provided data.

Raises: KeyError: If required keys (e.g., "id" or "attr" on nodes) are missing.

ngraph.graph.strict_multidigraph¶

Strict multi-directed graph with validation and convenience APIs.

StrictMultiDiGraph extends networkx.MultiDiGraph to enforce explicit node management, unique edge identifiers, and predictable error handling. It exposes helpers to access nodes/edges as dictionaries and to serialize in node-link format via to_dict().

StrictMultiDiGraph¶

A custom multi-directed graph with strict rules and unique edge IDs.

This class enforces:

No automatic creation of missing nodes when adding an edge.
No duplicate nodes (raises ValueError on duplicates).
No duplicate edges by key (raises ValueError on duplicates).
Removing non-existent nodes or edges raises ValueError.
Each edge key must be unique; by default, a Base64-UUID is generated

if none is provided.
copy() can perform a pickle-based deep copy that may be faster

than the NetworkX default.

Inherits from: networkx.MultiDiGraph

Methods:

add_edge(self, u_for_edge: 'NodeID', v_for_edge: 'NodeID', key: 'Optional[EdgeID]' = None, **attr: 'Any') -> 'EdgeID' - Add a directed edge from u_for_edge to v_for_edge.
add_edges_from(self, ebunch_to_add, **attr) - Add all the edges in ebunch_to_add.
add_node(self, node_for_adding: 'NodeID', **attr: 'Any') -> 'None' - Add a single node, disallowing duplicates.
add_nodes_from(self, nodes_for_adding, **attr) - Add multiple nodes.
add_weighted_edges_from(self, ebunch_to_add, weight='weight', **attr) - Add weighted edges in ebunch_to_add with specified weight attr
adjacency(self) - Returns an iterator over (node, adjacency dict) tuples for all nodes.
clear(self) - Remove all nodes and edges from the graph.
clear_edges(self) - Remove all edges from the graph without altering nodes.
copy(self, as_view: 'bool' = False, pickle: 'bool' = True) -> 'StrictMultiDiGraph' - Create a copy of this graph.
edge_subgraph(self, edges) - Returns the subgraph induced by the specified edges.
edges_between(self, u: 'NodeID', v: 'NodeID') -> 'List[EdgeID]' - List all edge keys from node u to node v.
get_edge_attr(self, key: 'EdgeID') -> 'AttrDict' - Retrieve the attribute dictionary of a specific edge.
get_edge_data(self, u, v, key=None, default=None) - Returns the attribute dictionary associated with edge (u, v,
get_edges(self) -> 'Dict[EdgeID, EdgeTuple]' - Retrieve a dictionary of all edges by their keys.
get_nodes(self) -> 'Dict[NodeID, AttrDict]' - Retrieve all nodes and their attributes as a dictionary.
has_edge(self, u, v, key=None) - Returns True if the graph has an edge between nodes u and v.
has_edge_by_id(self, key: 'EdgeID') -> 'bool' - Check whether an edge with the given key exists.
has_node(self, n) - Returns True if the graph contains the node n.
has_predecessor(self, u, v) - Returns True if node u has predecessor v.
has_successor(self, u, v) - Returns True if node u has successor v.
is_directed(self) - Returns True if graph is directed, False otherwise.
is_multigraph(self) - Returns True if graph is a multigraph, False otherwise.
nbunch_iter(self, nbunch=None) - Returns an iterator over nodes contained in nbunch that are
neighbors(self, n) - Returns an iterator over successor nodes of n.
new_edge_key(self, u, v) - Returns an unused key for edges between nodes u and v.
number_of_edges(self, u=None, v=None) - Returns the number of edges between two nodes.
number_of_nodes(self) - Returns the number of nodes in the graph.
order(self) - Returns the number of nodes in the graph.
predecessors(self, n) - Returns an iterator over predecessor nodes of n.
remove_edge(self, u: 'NodeID', v: 'NodeID', key: 'Optional[EdgeID]' = None) -> 'None' - Remove an edge (or edges) between nodes u and v.
remove_edge_by_id(self, key: 'EdgeID') -> 'None' - Remove a directed edge by its unique key.
remove_edges_from(self, ebunch) - Remove all edges specified in ebunch.
remove_node(self, n: 'NodeID') -> 'None' - Remove a single node and all incident edges.
remove_nodes_from(self, nodes) - Remove multiple nodes.
reverse(self, copy=True) - Returns the reverse of the graph.
size(self, weight=None) - Returns the number of edges or total of all edge weights.
subgraph(self, nodes) - Returns a SubGraph view of the subgraph induced on nodes.
successors(self, n) - Returns an iterator over successor nodes of n.
to_dict(self) -> 'Dict[str, Any]' - Convert the graph to a dictionary representation suitable for JSON serialization.
to_directed(self, as_view=False) - Returns a directed representation of the graph.
to_directed_class(self) - Returns the class to use for empty directed copies.
to_undirected(self, reciprocal=False, as_view=False) - Returns an undirected representation of the digraph.
to_undirected_class(self) - Returns the class to use for empty undirected copies.
update(self, edges=None, nodes=None) - Update the graph using nodes/edges/graphs as input.
update_edge_attr(self, key: 'EdgeID', **attr: 'Any') -> 'None' - Update attributes on an existing edge by key.

ngraph.model.network¶

Network topology modeling with Node, Link, RiskGroup, and Network classes.

Link¶

Represents one directed link between two nodes.

The model stores a single direction (source -> target). When building the working graph for analysis, a reverse edge is added by default to provide bidirectional connectivity. Disable with add_reverse=False in Network.to_strict_multidigraph.

Attributes: source (str): Name of the source node. target (str): Name of the target node. capacity (float): Link capacity (default 1.0). cost (float): Link cost (default 1.0). disabled (bool): Whether the link is disabled. risk_groups (Set[str]): Set of risk group names this link belongs to. attrs (Dict[str, Any]): Additional metadata (e.g., distance). id (str): Auto-generated unique identifier: "{source}|{target}|".

Attributes:

source (str)
target (str)
capacity (float) = 1.0
cost (float) = 1.0
disabled (bool) = False
risk_groups (Set[str]) = set()
attrs (Dict[str, Any]) = {}
id (str)

Network¶

A container for network nodes and links.

Network represents the scenario-level topology with persistent state (nodes/links that are disabled in the scenario configuration). For temporary exclusion of nodes/links during analysis (e.g., failure simulation), use NetworkView instead of modifying the Network's disabled states.

Attributes: nodes (Dict[str, Node]): Mapping from node name -> Node object. links (Dict[str, Link]): Mapping from link ID -> Link object. risk_groups (Dict[str, RiskGroup]): Top-level risk groups by name. attrs (Dict[str, Any]): Optional metadata about the network.

Attributes:

nodes (Dict[str, Node]) = {}
links (Dict[str, Link]) = {}
risk_groups (Dict[str, RiskGroup]) = {}
attrs (Dict[str, Any]) = {}

Methods:

add_link(self, link: 'Link') -> 'None' - Add a link to the network (keyed by the link's auto-generated ID).
add_node(self, node: 'Node') -> 'None' - Add a node to the network (keyed by node.name).
disable_all(self) -> 'None' - Mark all nodes and links as disabled.
disable_link(self, link_id: 'str') -> 'None' - Mark a link as disabled.
disable_node(self, node_name: 'str') -> 'None' - Mark a node as disabled.
disable_risk_group(self, name: 'str', recursive: 'bool' = True) -> 'None' - Disable all nodes/links that have 'name' in their risk_groups.
enable_all(self) -> 'None' - Mark all nodes and links as enabled.
enable_link(self, link_id: 'str') -> 'None' - Mark a link as enabled.
enable_node(self, node_name: 'str') -> 'None' - Mark a node as enabled.
enable_risk_group(self, name: 'str', recursive: 'bool' = True) -> 'None' - Enable all nodes/links that have 'name' in their risk_groups.
find_links(self, source_regex: 'Optional[str]' = None, target_regex: 'Optional[str]' = None, any_direction: 'bool' = False) -> 'List[Link]' - Search for links using optional regex patterns for source or target node names.
get_links_between(self, source: 'str', target: 'str') -> 'List[str]' - Retrieve all link IDs that connect the specified source node
k_shortest_paths(self, source_path: 'str', sink_path: 'str', mode: 'str' = 'pairwise', *, max_k: 'int' = 3, max_path_cost: 'float' = inf, max_path_cost_factor: 'Optional[float]' = None, split_parallel_edges: 'bool' = False) -> 'Dict[Tuple[str, str], List[_NGPath]]' - Return up to K shortest paths per group pair.
max_flow(self, source_path: 'str', sink_path: 'str', mode: 'str' = 'combine', shortest_path: 'bool' = False, flow_placement: 'FlowPlacement' = <FlowPlacement.PROPORTIONAL: 1>) -> 'Dict[Tuple[str, str], float]' - Compute maximum flow between node groups in this network.
max_flow_detailed(self, source_path: 'str', sink_path: 'str', mode: 'str' = 'combine', shortest_path: 'bool' = False, flow_placement: 'FlowPlacement' = <FlowPlacement.PROPORTIONAL: 1>) -> 'Dict[Tuple[str, str], Tuple[float, FlowSummary, StrictMultiDiGraph]]' - Compute maximum flow with both analytics summary and graph.
max_flow_with_graph(self, source_path: 'str', sink_path: 'str', mode: 'str' = 'combine', shortest_path: 'bool' = False, flow_placement: 'FlowPlacement' = <FlowPlacement.PROPORTIONAL: 1>) -> 'Dict[Tuple[str, str], Tuple[float, StrictMultiDiGraph]]' - Compute maximum flow and return flow-assigned graphs.
max_flow_with_summary(self, source_path: 'str', sink_path: 'str', mode: 'str' = 'combine', shortest_path: 'bool' = False, flow_placement: 'FlowPlacement' = <FlowPlacement.PROPORTIONAL: 1>) -> 'Dict[Tuple[str, str], Tuple[float, FlowSummary]]' - Compute maximum flow and return per-pair analytics summary.
saturated_edges(self, source_path: 'str', sink_path: 'str', mode: 'str' = 'combine', tolerance: 'float' = 1e-10, shortest_path: 'bool' = False, flow_placement: 'FlowPlacement' = <FlowPlacement.PROPORTIONAL: 1>) -> 'Dict[Tuple[str, str], List[Tuple[str, str, str]]]' - Identify saturated edges in max flow solutions.
select_node_groups_by_path(self, path: 'str') -> 'Dict[str, List[Node]]' - Select and group nodes by regex on name or by attribute directive.
sensitivity_analysis(self, source_path: 'str', sink_path: 'str', mode: 'str' = 'combine', change_amount: 'float' = 1.0, shortest_path: 'bool' = False, flow_placement: 'FlowPlacement' = <FlowPlacement.PROPORTIONAL: 1>) -> 'Dict[Tuple[str, str], Dict[Tuple[str, str, str], float]]' - Perform sensitivity analysis for capacity changes.
shortest_path_costs(self, source_path: 'str', sink_path: 'str', mode: 'str' = 'combine') -> 'Dict[Tuple[str, str], float]' - Return minimal path costs between node groups in this network.
shortest_paths(self, source_path: 'str', sink_path: 'str', mode: 'str' = 'combine', *, split_parallel_edges: 'bool' = False) -> 'Dict[Tuple[str, str], List[_NGPath]]' - Return concrete shortest path(s) between selected node groups.
to_strict_multidigraph(self, add_reverse: 'bool' = True) -> 'StrictMultiDiGraph' - Create a StrictMultiDiGraph representation of this Network.

Node¶

Represents a node in the network.

Each node is uniquely identified by its name, which is used as the key in the Network's node dictionary.

Attributes: name (str): Unique identifier for the node. disabled (bool): Whether the node is disabled in the scenario configuration. risk_groups (Set[str]): Set of risk group names this node belongs to. attrs (Dict[str, Any]): Additional metadata (e.g., coordinates, region).

Attributes:

name (str)
disabled (bool) = False
risk_groups (Set[str]) = set()
attrs (Dict[str, Any]) = {}

RiskGroup¶

Represents a shared-risk or failure domain, which may have nested children.

Attributes: name (str): Unique name of this risk group. children (List[RiskGroup]): Subdomains in a nested structure. disabled (bool): Whether this group was declared disabled on load. attrs (Dict[str, Any]): Additional metadata for the risk group.

Attributes:

name (str)
children (List[RiskGroup]) = []
disabled (bool) = False
attrs (Dict[str, Any]) = {}

ngraph.model.view¶

Read-only view of a Network with temporary exclusions.

This module defines a view over Network objects that can exclude nodes and links for analysis without mutating the base network. It supports what-if analysis, including failure simulations.

NetworkView¶

Read-only overlay that hides selected nodes/links from a base Network.

NetworkView provides filtered access to a Network where both scenario-disabled elements (Node.disabled, Link.disabled) and analysis-excluded elements are hidden from algorithms. This enables failure simulation and what-if analysis without mutating the base Network.

Multiple NetworkView instances can safely operate on the same base Network concurrently, each with different exclusion sets.

Example:

# Create view excluding specific nodes for failure analysis
view = NetworkView.from_excluded_sets(
    base_network,
    excluded_nodes=["node1", "node2"],
    excluded_links=["link1"]
)

# Run analysis on filtered topology
flows = view.max_flow("source.*", "sink.*")

Attributes: _base: The underlying Network object. _excluded_nodes: Frozen set of node names to exclude from analysis. _excluded_links: Frozen set of link IDs to exclude from analysis.

Attributes:

_base ('Network')
_excluded_nodes (frozenset[str]) = frozenset()
_excluded_links (frozenset[str]) = frozenset()

Methods:

from_excluded_sets(base: "'Network'", excluded_nodes: 'Iterable[str]' = (), excluded_links: 'Iterable[str]' = ()) -> "'NetworkView'" - Create a NetworkView with specified exclusions.
is_link_hidden(self, link_id: 'str') -> 'bool' - Check if a link is hidden in this view.
is_node_hidden(self, name: 'str') -> 'bool' - Check if a node is hidden in this view.
k_shortest_paths(self, source_path: 'str', sink_path: 'str', mode: 'str' = 'pairwise', *, max_k: 'int' = 3, max_path_cost: 'float' = inf, max_path_cost_factor: 'Optional[float]' = None, split_parallel_edges: 'bool' = False) -> 'Dict[Tuple[str, str], List[_NGPath]]' - Return up to K shortest paths per group pair.
max_flow(self, source_path: 'str', sink_path: 'str', mode: 'str' = 'combine', shortest_path: 'bool' = False, flow_placement: "Optional['FlowPlacement']" = None) -> 'Dict[Tuple[str, str], float]' - Compute maximum flow between node groups in this view.
max_flow_detailed(self, source_path: 'str', sink_path: 'str', mode: 'str' = 'combine', shortest_path: 'bool' = False, flow_placement: "Optional['FlowPlacement']" = None) -> "Dict[Tuple[str, str], Tuple[float, 'FlowSummary', 'StrictMultiDiGraph']]" - Compute maximum flow with complete analytics and graph.
max_flow_with_graph(self, source_path: 'str', sink_path: 'str', mode: 'str' = 'combine', shortest_path: 'bool' = False, flow_placement: "Optional['FlowPlacement']" = None) -> "Dict[Tuple[str, str], Tuple[float, 'StrictMultiDiGraph']]" - Compute maximum flow and return flow-assigned graph.
max_flow_with_summary(self, source_path: 'str', sink_path: 'str', mode: 'str' = 'combine', shortest_path: 'bool' = False, flow_placement: "Optional['FlowPlacement']" = None) -> "Dict[Tuple[str, str], Tuple[float, 'FlowSummary']]" - Compute maximum flow with detailed analytics summary.
saturated_edges(self, source_path: 'str', sink_path: 'str', mode: 'str' = 'combine', tolerance: 'float' = 1e-10, shortest_path: 'bool' = False, flow_placement: "Optional['FlowPlacement']" = None) -> 'Dict[Tuple[str, str], List[Tuple[str, str, str]]]' - Identify saturated edges in max flow solutions.
select_node_groups_by_path(self, path: 'str') -> "Dict[str, List['Node']]" - Select and group visible nodes by regex or attribute directive.
sensitivity_analysis(self, source_path: 'str', sink_path: 'str', mode: 'str' = 'combine', change_amount: 'float' = 1.0, shortest_path: 'bool' = False, flow_placement: "Optional['FlowPlacement']" = None) -> 'Dict[Tuple[str, str], Dict[Tuple[str, str, str], float]]' - Perform sensitivity analysis on capacity changes.
shortest_path_costs(self, source_path: 'str', sink_path: 'str', mode: 'str' = 'combine') -> 'Dict[Tuple[str, str], float]' - Return minimal path costs between node groups in this view.
shortest_paths(self, source_path: 'str', sink_path: 'str', mode: 'str' = 'combine', *, split_parallel_edges: 'bool' = False) -> 'Dict[Tuple[str, str], List[_NGPath]]' - Return concrete shortest path(s) between selected node groups.
to_strict_multidigraph(self, add_reverse: 'bool' = True) -> "'StrictMultiDiGraph'" - Create a StrictMultiDiGraph representation of this view.

ngraph.algorithms.base¶

Base classes and enums for network analysis algorithms.

EdgeSelect¶

Edge selection criteria.

Determines which edges are considered for path-finding between a node and its neighbor(s).

FlowPlacement¶

Strategies to distribute flow across parallel equal-cost paths.

PathAlg¶

Path-finding algorithm types.

ngraph.algorithms.capacity¶

Capacity calculation algorithms for network analysis.

This module computes feasible flow given a predecessor DAG from a shortest-path routine and supports two placement strategies: proportional and equal-balanced in reversed orientation. Functions follow a Dinic-like blocking-flow approach for proportional placement.

calc_graph_capacity(flow_graph: 'StrictMultiDiGraph', src_node: 'NodeID', dst_node: 'NodeID', pred: 'Dict[NodeID, Dict[NodeID, List[EdgeID]]]', flow_placement: 'FlowPlacement' = , capacity_attr: 'str' = 'capacity', flow_attr: 'str' = 'flow') -> 'Tuple[float, Dict[NodeID, Dict[NodeID, float]]]'¶

Calculate feasible flow and flow fractions between two nodes.

In PROPORTIONAL mode (similar to Dinic in reversed orientation):

Build the reversed residual graph from dst_node (via _init_graph_data).
Use BFS (in _set_levels_bfs) to build a level graph and DFS (_push_flow_dfs)

to push blocking flows, repeating until no more flow can be pushed.
The net flow found is stored in reversed orientation. Convert final flows

to forward orientation by negating and normalizing by the total.

In EQUAL_BALANCED mode:

Build reversed adjacency from dst_node (also via _init_graph_data),

ignoring capacity checks in that BFS.
Perform a BFS pass from src_node (_equal_balance_bfs) to distribute a

nominal flow of 1.0 equally among parallel edges.
Determine the scaling ratio so that no edge capacity is exceeded.

Scale the flow assignments accordingly, then normalize to the forward sense.

Args: flow_graph: The multigraph with capacity and flow attributes. src_node: The source node in the forward graph. dst_node: The destination node in the forward graph. pred: Forward adjacency mapping (node -> (adjacent node -> list of EdgeIDs)), typically produced by spf(..., multipath=True). Must be a DAG. flow_placement: The flow distribution strategy (PROPORTIONAL or EQUAL_BALANCED). capacity_attr: Name of the capacity attribute on edges. flow_attr: Name of the flow attribute on edges.

Returns: tuple[float, dict[NodeID, dict[NodeID, float]]]:

Total feasible flow from src_node to dst_node.
Normalized flow fractions in forward orientation ([u][v] >= 0).

Raises: ValueError: If src_node or dst_node is not in the graph, or the flow_placement is unsupported.

ngraph.algorithms.edge_select¶

Edge selection algorithms for routing.

Provides selection routines used by SPF to choose candidate edges between neighbors according to cost and capacity constraints.

edge_select_fabric(edge_select: ngraph.algorithms.base.EdgeSelect, select_value: Optional[Any] = None, edge_select_func: Optional[Callable[[ngraph.graph.strict_multidigraph.StrictMultiDiGraph, Hashable, Hashable, Dict[Hashable, Dict[str, Any]], Optional[Set[Hashable]], Optional[Set[Hashable]]], Tuple[Union[int, float], List[Hashable]]]] = None, excluded_edges: Optional[Set[Hashable]] = None, excluded_nodes: Optional[Set[Hashable]] = None, cost_attr: str = 'cost', capacity_attr: str = 'capacity', flow_attr: str = 'flow') -> Callable[[ngraph.graph.strict_multidigraph.StrictMultiDiGraph, Hashable, Hashable, Dict[Hashable, Dict[str, Any]], Optional[Set[Hashable]], Optional[Set[Hashable]]], Tuple[Union[int, float], List[Hashable]]]¶

Create an edge-selection callable for SPF.

Args: edge_select: An EdgeSelect enum specifying the selection strategy. select_value: An optional numeric threshold or scaling factor for capacity checks. edge_select_func: A user-supplied function if edge_select=USER_DEFINED. excluded_edges: A set of edges to ignore entirely. excluded_nodes: A set of nodes to skip (if the destination node is in this set). cost_attr: The edge attribute name representing cost. capacity_attr: The edge attribute name representing capacity. flow_attr: The edge attribute name representing current flow.

Returns: Callable: Function with signature (graph, src, dst, edges_dict, excluded_edges, excluded_nodes) -> (selected_cost, [edge_ids]).

ngraph.algorithms.flow_init¶

Flow graph initialization utilities.

Ensures nodes and edges carry aggregate (flow_attr) and per-flow (flows_attr) attributes, optionally resetting existing values.

init_flow_graph(flow_graph: 'StrictMultiDiGraph', flow_attr: 'str' = 'flow', flows_attr: 'str' = 'flows', reset_flow_graph: 'bool' = True) -> 'StrictMultiDiGraph'¶

Ensure that nodes and edges expose flow-related attributes.

For each node and edge:

The attribute named flow_attr (default: "flow") is set to 0.
The attribute named flows_attr (default: "flows") is set to an empty dict.

If reset_flow_graph is True, any existing flow values in these attributes are overwritten; otherwise they are only created if missing.

Args: flow_graph: The StrictMultiDiGraph whose nodes and edges should be prepared for flow assignment. flow_attr: The attribute name to track a numeric flow value per node/edge. flows_attr: The attribute name to track multiple flow identifiers (and flows). reset_flow_graph: If True, reset existing flows (set to 0). If False, do not overwrite.

Returns: StrictMultiDiGraph: The same graph instance after attribute checks.

ngraph.algorithms.max_flow¶

Maximum-flow computation via iterative shortest-path augmentation.

Implements a practical Edmonds-Karp-like procedure using SPF with capacity constraints and configurable flow-splitting across equal-cost parallel edges. Provides helpers for saturated-edge detection and simple sensitivity analysis.

calc_max_flow(graph: ngraph.graph.strict_multidigraph.StrictMultiDiGraph, src_node: Hashable, dst_node: Hashable, *, return_summary: bool = False, return_graph: bool = False, flow_placement: ngraph.algorithms.base.FlowPlacement = , shortest_path: bool = False, reset_flow_graph: bool = False, capacity_attr: str = 'capacity', flow_attr: str = 'flow', flows_attr: str = 'flows', copy_graph: bool = True, tolerance: float = 1e-10) -> Union[float, tuple]¶

Compute max flow between two nodes in a directed multi-graph.

Uses iterative shortest-path augmentation with capacity-aware SPF and configurable flow placement.

By default, this function:

Creates or re-initializes a flow-aware copy of the graph (via init_flow_graph).
Repeatedly finds a path from src_node to dst_node using spf with

capacity constraints (EdgeSelect.ALL_MIN_COST_WITH_CAP_REMAINING).
Places flow along that path (via place_flow_on_graph) until no augmenting path

remains or the capacities are exhausted.

If shortest_path=True, the function performs only one iteration (single augmentation) and returns the flow placed along that single path (not the true max flow).

Args: graph (StrictMultiDiGraph): The original graph containing capacity/flow attributes on each edge. src_node (NodeID): The source node for flow. dst_node (NodeID): The destination node for flow. return_summary (bool): If True, return a FlowSummary with detailed flow analytics. Defaults to False. return_graph (bool): If True, return the mutated flow graph along with other results. Defaults to False. flow_placement (FlowPlacement): Determines how flow is split among parallel edges of equal cost. Defaults to FlowPlacement.PROPORTIONAL. shortest_path (bool): If True, place flow only once along the first shortest path found and return immediately, rather than iterating for the true max flow. reset_flow_graph (bool): If True, reset any existing flow data (e.g., flow_attr, flows_attr). Defaults to False. capacity_attr (str): The name of the capacity attribute on edges. Defaults to "capacity". flow_attr (str): The name of the aggregated flow attribute on edges. Defaults to "flow". flows_attr (str): The name of the per-flow dictionary attribute on edges. Defaults to "flows". copy_graph (bool): If True, work on a copy of the original graph so it remains unmodified. Defaults to True. tolerance (float): Tolerance for floating-point comparisons when determining saturated edges and residual capacity. Defaults to 1e-10.

Returns: Union[float, tuple]:

If neither flag: float total flow.
If return_summary only: tuple[float, FlowSummary].
If both flags: tuple[float, FlowSummary, StrictMultiDiGraph].

Notes:

When using return_summary or return_graph, the return value is a tuple.

Examples: >>> g = StrictMultiDiGraph() >>> g.add_node('A') >>> g.add_node('B') >>> g.add_node('C') >>> g.add_edge('A', 'B', capacity=10.0, flow=0.0, flows={}, cost=1) >>> g.add_edge('B', 'C', capacity=5.0, flow=0.0, flows={}, cost=1) >>> >>> # Basic usage (scalar return) >>> max_flow_value = calc_max_flow(g, 'A', 'C') >>> print(max_flow_value) 5.0 >>> >>> # With flow summary analytics >>> flow, summary = calc_max_flow(g, 'A', 'C', return_summary=True) >>> print(f"Min-cut edges: {summary.min_cut}") >>> >>> # With both summary and mutated graph >>> flow, summary, flow_graph = calc_max_flow( ... g, 'A', 'C', return_summary=True, return_graph=True ... ) >>> # flow_graph contains the flow assignments

run_sensitivity(graph: ngraph.graph.strict_multidigraph.StrictMultiDiGraph, src_node: Hashable, dst_node: Hashable, , capacity_attr: str = 'capacity', flow_attr: str = 'flow', change_amount: float = 1.0, *kwargs) -> dict[tuple, float]¶

Simple sensitivity analysis for per-edge capacity changes.

Tests changing each saturated edge capacity by change_amount and measures the resulting change in total flow. Positive values increase capacity, negative values decrease capacity (with validation to prevent negative capacities).

Args: graph: The graph to analyze src_node: Source node dst_node: Destination node capacity_attr: Name of capacity attribute flow_attr: Name of flow attribute change_amount: Amount to change capacity for testing (positive=increase, negative=decrease) **kwargs: Additional arguments passed to calc_max_flow

Returns: dict[tuple, float]: Flow delta per modified edge.

saturated_edges(graph: ngraph.graph.strict_multidigraph.StrictMultiDiGraph, src_node: Hashable, dst_node: Hashable, , capacity_attr: str = 'capacity', flow_attr: str = 'flow', tolerance: float = 1e-10, *kwargs) -> list[tuple]¶

Identify saturated edges in the max-flow solution.

Args: graph: The graph to analyze src_node: Source node dst_node: Destination node capacity_attr: Name of capacity attribute flow_attr: Name of flow attribute tolerance: Tolerance for considering an edge saturated **kwargs: Additional arguments passed to calc_max_flow

Returns: list[tuple]: Edges (u, v, k) with residual capacity <= tolerance.

ngraph.algorithms.paths¶

Path manipulation utilities.

Provides helpers to enumerate realized paths from a predecessor map produced by SPF/KSP, with optional expansion of parallel edges into distinct paths.

resolve_to_paths(src_node: 'NodeID', dst_node: 'NodeID', pred: 'Dict[NodeID, Dict[NodeID, List[EdgeID]]]', split_parallel_edges: 'bool' = False) -> 'Iterator[PathTuple]'¶

Enumerate all paths from a predecessor map.

Args: src_node: Source node ID. dst_node: Destination node ID. pred: Predecessor map from SPF or KSP. split_parallel_edges: If True, expand parallel edges into distinct paths.

Yields: PathTuple: Sequence of (node_id, (edge_ids,)) pairs from source to dest.

ngraph.algorithms.placement¶

Flow placement for routing over equal-cost predecessor DAGs.

Places feasible flow on a graph given predecessor relations and a placement strategy, updating aggregate and per-flow attributes.

FlowPlacementMeta¶

Metadata describing how flow was placed on the graph.

Attributes: placed_flow: The amount of flow actually placed. remaining_flow: The portion of flow that could not be placed due to capacity limits. nodes: Set of node IDs that participated in the flow. edges: Set of edge IDs that carried some portion of this flow.

Attributes:

placed_flow (float)
remaining_flow (float)
nodes (Set[NodeID]) = set()
edges (Set[EdgeID]) = set()

place_flow_on_graph(flow_graph: 'StrictMultiDiGraph', src_node: 'NodeID', dst_node: 'NodeID', pred: 'Dict[NodeID, Dict[NodeID, List[EdgeID]]]', flow: 'float' = inf, flow_index: 'Optional[Hashable]' = None, flow_placement: 'FlowPlacement' = , capacity_attr: 'str' = 'capacity', flow_attr: 'str' = 'flow', flows_attr: 'str' = 'flows') -> 'FlowPlacementMeta'¶

Place flow from src_node to dst_node on flow_graph.

Uses a precomputed flow_dict from calc_graph_capacity to figure out how much flow can be placed. Updates the graph's edges and nodes with the placed flow.

Args: flow_graph: The graph on which flow will be placed. src_node: The source node. dst_node: The destination node. pred: A dictionary of node->(adj_node->list_of_edge_IDs) giving path adjacency. flow: Requested flow amount; can be infinite. flow_index: Identifier for this flow (used to track multiple flows). flow_placement: Strategy for distributing flow among parallel equal cost paths. capacity_attr: Attribute name on edges for capacity. flow_attr: Attribute name on edges/nodes for aggregated flow. flows_attr: Attribute name on edges/nodes for per-flow tracking.

Returns: FlowPlacementMeta: Amount placed, remaining amount, and touched nodes/edges.

remove_flow_from_graph(flow_graph: 'StrictMultiDiGraph', flow_index: 'Optional[Hashable]' = None, flow_attr: 'str' = 'flow', flows_attr: 'str' = 'flows') -> 'None'¶

Remove one or all flows from the graph.

Args: flow_graph: Graph whose edge flow attributes will be modified. flow_index: If provided, remove only the specified flow; otherwise remove all. flow_attr: Aggregate flow attribute name on edges. flows_attr: Per-flow attribute name on edges.

ngraph.algorithms.spf¶

Shortest-path-first (SPF) algorithms.

Implements Dijkstra-like SPF with pluggable edge-selection policies and a Yen-like KSP generator. Specialized fast paths exist for common selection strategies without exclusions.

Notes: When a destination node is known, SPF supports an optimized mode that terminates once the destination's minimal distance is settled. In this mode:

The destination node is not expanded (no neighbor relaxation from dst).
The algorithm continues processing any nodes with equal distance to capture

equal-cost predecessors (needed by proportional flow placement).

ksp(graph: ngraph.graph.strict_multidigraph.StrictMultiDiGraph, src_node: Hashable, dst_node: Hashable, edge_select: ngraph.algorithms.base.EdgeSelect = , edge_select_func: Optional[Callable[[ngraph.graph.strict_multidigraph.StrictMultiDiGraph, Hashable, Hashable, Dict[Hashable, Dict[str, Any]], Set[Hashable], Set[Hashable]], Tuple[Union[int, float], List[Hashable]]]] = None, max_k: Optional[int] = None, max_path_cost: Union[int, float] = inf, max_path_cost_factor: Optional[float] = None, multipath: bool = True, excluded_edges: Optional[Set[Hashable]] = None, excluded_nodes: Optional[Set[Hashable]] = None) -> Iterator[Tuple[Dict[Hashable, Union[int, float]], Dict[Hashable, Dict[Hashable, List[Hashable]]]]]¶

Yield up to k shortest paths using a Yen-like algorithm.

The initial SPF (shortest path) is computed; subsequent paths are found by systematically excluding edges/nodes used by previously generated paths. Each iteration yields a (costs, pred) describing one path. Stops if there are no more valid paths or if max_k is reached.

Args: graph: The directed graph (StrictMultiDiGraph). src_node: The source node. dst_node: The destination node. edge_select: The edge selection strategy. Defaults to ALL_MIN_COST. edge_select_func: Optional override of the default edge selection function. max_k: If set, yields at most k distinct paths. max_path_cost: If set, do not yield any path whose total cost > max_path_cost. max_path_cost_factor: If set, updates max_path_cost to: min(max_path_cost, best_path_cost * max_path_cost_factor). multipath: Whether to consider multiple same-cost expansions in SPF. excluded_edges: Set of edge IDs to exclude globally. excluded_nodes: Set of node IDs to exclude globally.

Yields: Tuple of (costs, pred) per discovered path in ascending cost order.

spf(graph: ngraph.graph.strict_multidigraph.StrictMultiDiGraph, src_node: Hashable, edge_select: ngraph.algorithms.base.EdgeSelect = , edge_select_func: Optional[Callable[[ngraph.graph.strict_multidigraph.StrictMultiDiGraph, Hashable, Hashable, Dict[Hashable, Dict[str, Any]], Set[Hashable], Set[Hashable]], Tuple[Union[int, float], List[Hashable]]]] = None, multipath: bool = True, excluded_edges: Optional[Set[Hashable]] = None, excluded_nodes: Optional[Set[Hashable]] = None, dst_node: Optional[Hashable] = None) -> Tuple[Dict[Hashable, Union[int, float]], Dict[Hashable, Dict[Hashable, List[Hashable]]]]¶

Compute shortest paths from a source node.

By default, uses EdgeSelect.ALL_MIN_COST. If multipath=True, multiple equal-cost paths to the same node will be recorded in the predecessor structure. If no excluded edges/nodes are given and edge_select is one of the specialized (ALL_MIN_COST or ALL_MIN_COST_WITH_CAP_REMAINING), it uses a fast specialized routine.

Args: graph: The directed graph (StrictMultiDiGraph). src_node: The source node from which to compute shortest paths. edge_select: The edge selection strategy. Defaults to ALL_MIN_COST. edge_select_func: If provided, overrides the default edge selection function. Must return (cost, list_of_edges) for the given node->neighbor adjacency. multipath: Whether to record multiple same-cost paths. excluded_edges: A set of edge IDs to ignore in the graph. excluded_nodes: A set of node IDs to ignore in the graph. dst_node: Optional destination node. If provided, SPF avoids expanding from the destination and performs early termination once the next candidate in the heap would exceed the settled distance for dst_node. This preserves equal-cost predecessors while avoiding unnecessary relaxations beyond the destination.

Returns: tuple[dict[NodeID, Cost], dict[NodeID, dict[NodeID, list[EdgeID]]]]: Costs and predecessor mapping.

Raises: KeyError: If src_node does not exist in graph.

ngraph.algorithms.types¶

Types and data structures for algorithm analytics.

Defines immutable summary containers and aliases for algorithm outputs.

FlowSummary¶

Summary of max-flow computation results.

Captures edge flows, residual capacities, reachable set, and min-cut.

Attributes: total_flow: Maximum flow value achieved. edge_flow: Flow amount per edge, indexed by (src, dst, key). residual_cap: Remaining capacity per edge after placement. reachable: Nodes reachable from source in residual graph. min_cut: Saturated edges crossing the s-t cut. cost_distribution: Mapping of path cost to flow volume placed at that cost.

Attributes:

total_flow (float)
edge_flow (Dict[Edge, float])
residual_cap (Dict[Edge, float])
reachable (Set[str])
min_cut (List[Edge])
cost_distribution (Dict[Cost, float])

ngraph.paths.bundle¶

Utilities for compact representation of equal-cost path sets.

This module defines PathBundle, a structure that represents one or more equal-cost paths between two nodes using a predecessor map. It supports concatenation, containment checks, sub-bundle extraction with cost recalculation, and enumeration into concrete Path instances.

PathBundle¶

A collection of equal-cost paths between two nodes.

This class encapsulates one or more parallel paths (all of the same cost) between src_node and dst_node. The predecessor map pred associates each node with the node(s) from which it can be reached, along with a list of edge IDs used in that step. The constructor performs a reverse traversal from dst_node to src_node to collect all edges, nodes, and store them in this bundle.

The constructor assumes the predecessor relation forms a DAG between src_node and dst_node. No cycle detection is performed. If cycles are present, traversal may not terminate.

Methods:

add(self, other: 'PathBundle') -> 'PathBundle' - Concatenate this bundle with another bundle (end-to-start).
contains(self, other: 'PathBundle') -> 'bool' - Check if this bundle's edge set contains all edges of other.
from_path(path: 'Path', resolve_edges: 'bool' = False, graph: 'Optional[StrictMultiDiGraph]' = None, edge_select: 'Optional[EdgeSelect]' = None, cost_attr: 'str' = 'cost', capacity_attr: 'str' = 'capacity') -> 'PathBundle' - Construct a PathBundle from a single Path object.
get_sub_path_bundle(self, new_dst_node: 'NodeID', graph: 'StrictMultiDiGraph', cost_attr: 'str' = 'cost') -> 'PathBundle' - Create a sub-bundle ending at new_dst_node with correct minimal cost.
is_disjoint_from(self, other: 'PathBundle') -> 'bool' - Check if this bundle shares no edges with other.
is_subset_of(self, other: 'PathBundle') -> 'bool' - Check if this bundle's edge set is contained in other's edge set.
resolve_to_paths(self, split_parallel_edges: 'bool' = False) -> 'Iterator[Path]' - Generate all concrete Path objects contained in this PathBundle.

ngraph.paths.path¶

Lightweight representation of a single routing path.

The Path dataclass stores a node-and-parallel-edges sequence and a numeric cost. Cached properties expose derived sequences for nodes and edges, and helpers provide equality, ordering by cost, and sub-path extraction with cost recalculation.

Path¶

Represents a single path in the network.

Attributes: path_tuple (PathTuple): A sequence of path elements. Each element is a tuple of the form (node_id, (edge_id_1, edge_id_2, ...)), where the final element typically has an empty tuple. cost (Cost): The total numeric cost (e.g., distance or metric) of the path. edges (Set[EdgeID]): A set of all edge IDs encountered in the path. nodes (Set[NodeID]): A set of all node IDs encountered in the path. edge_tuples (Set[Tuple[EdgeID, ...]]): A set of all tuples of parallel edges from each path element (including the final empty tuple).

Attributes:

path_tuple (PathTuple)
cost (Cost)
edges (Set[EdgeID]) = set()
nodes (Set[NodeID]) = set()
edge_tuples (Set[Tuple[EdgeID, ...]]) = set()

Methods:

get_sub_path(self, dst_node: 'NodeID', graph: 'StrictMultiDiGraph', cost_attr: 'str' = 'cost') -> 'Path' - Create a sub-path ending at the specified destination node, recalculating the cost.

ngraph.flows.flow¶

Flow and FlowIndex classes for traffic flow representation.

Flow¶

Represents a fraction of demand routed along a given PathBundle.

In traffic-engineering scenarios, a Flow object can model:

MPLS LSPs/tunnels with explicit paths,
IP forwarding behavior (with ECMP or UCMP),
Or anything that follows a specific set of paths.

Methods:

place_flow(self, flow_graph: 'StrictMultiDiGraph', to_place: 'float', flow_placement: 'FlowPlacement') -> 'Tuple[float, float]' - Place or update this flow on the graph.
remove_flow(self, flow_graph: 'StrictMultiDiGraph') -> 'None' - Remove this flow from the graph.

FlowIndex¶

Unique identifier for a flow.

Attributes: src_node: Source node. dst_node: Destination node. flow_class: Flow class label (hashable). flow_id: Monotonic integer id for this flow.

ngraph.flows.policy¶

FlowPolicy and FlowPolicyConfig classes for traffic routing algorithms.

FlowPolicy¶

Create, place, rebalance, and remove flows on a network graph.

Converts a demand into one or more Flow objects subject to capacity constraints and configuration: path selection, edge selection, and flow placement method.

Methods:

deep_copy(self) -> 'FlowPolicy' - Return a deep copy of this policy including flows.
get_metrics(self) -> 'Dict[str, float]' - Return cumulative placement metrics for this policy instance.
place_demand(self, flow_graph: 'StrictMultiDiGraph', src_node: 'NodeID', dst_node: 'NodeID', flow_class: 'Hashable', volume: 'float', target_flow_volume: 'Optional[float]' = None, min_flow: 'Optional[float]' = None) -> 'Tuple[float, float]' - Place demand volume on the graph by splitting or creating flows as needed.
rebalance_demand(self, flow_graph: 'StrictMultiDiGraph', src_node: 'NodeID', dst_node: 'NodeID', flow_class: 'Hashable', target_flow_volume: 'float') -> 'Tuple[float, float]' - Rebalance demand across existing flows towards the target volume per flow.
remove_demand(self, flow_graph: 'StrictMultiDiGraph') -> 'None' - Removes all flows from the network graph without clearing internal state.

FlowPolicyConfig¶

Enumerates supported flow policy configurations.

get_flow_policy(flow_policy_config: 'FlowPolicyConfig') -> 'FlowPolicy'¶

Create a policy instance from a configuration preset.

Args: flow_policy_config: A FlowPolicyConfig enum value specifying the desired policy.

Returns: FlowPolicy: Pre-configured policy instance.

Raises: ValueError: If an unknown FlowPolicyConfig value is provided.

ngraph.solver.helpers¶

ngraph.solver.maxflow¶

Problem-level max-flow API bound to the model layer.

Functions here operate on a model context that provides:

to_strict_multidigraph(add_reverse: bool = True) -> StrictMultiDiGraph
select_node_groups_by_path(path: str) -> dict[str, list[Node]]

They accept either a Network or a NetworkView. The input context is not mutated. Pseudo source and sink nodes are attached on a working graph when computing flows between groups.

max_flow(context: 'Any', source_path: 'str', sink_path: 'str', *, mode: 'str' = 'combine', shortest_path: 'bool' = False, flow_placement: 'FlowPlacement' = ) -> 'Dict[Tuple[str, str], float]'¶

Compute max flow between groups selected from the context.

Creates a working graph from the context, adds a pseudo source attached to the selected source nodes and a pseudo sink attached to the selected sink nodes, then runs the max-flow routine.

Args: context: Network or NetworkView providing selection and graph APIs. source_path: Selection expression for source groups. sink_path: Selection expression for sink groups. mode: Aggregation strategy. "combine" considers all sources as one group and all sinks as one group. "pairwise" evaluates each source-label and sink-label pair separately. shortest_path: If True, perform a single augmentation along the first shortest path instead of the full max-flow. flow_placement: Strategy for splitting flow among equal-cost parallel edges.

Returns: Dict[Tuple[str, str], float]: Total flow per (source_label, sink_label).

Raises: ValueError: If no matching sources or sinks are found, or if mode is not one of {"combine", "pairwise"}.

max_flow_detailed(context: 'Any', source_path: 'str', sink_path: 'str', *, mode: 'str' = 'combine', shortest_path: 'bool' = False, flow_placement: 'FlowPlacement' = ) -> "Dict[Tuple[str, str], Tuple[float, FlowSummary, 'StrictMultiDiGraph']]"¶

Compute max flow, return summary and flow graph for each pair.

Args: context: Network or NetworkView providing selection and graph APIs. source_path: Selection expression for source groups. sink_path: Selection expression for sink groups. mode: "combine" or "pairwise". See max_flow. shortest_path: If True, perform only one augmentation step. flow_placement: Strategy for splitting among equal-cost parallel edges.

Returns: Dict[Tuple[str, str], Tuple[float, FlowSummary, StrictMultiDiGraph]]: For each (source_label, sink_label), the total flow, a summary, and the flow-assigned graph.