Skip to content

Abstract Source Sink Graph

AbstractSourceSinkGraph 

AbstractSourceSinkGraph(
    base_graph: DiGraph,
    additional_starts: (
        list | None
    ) = None,
    additional_ends: (
        list | None
    ) = None,
)

Bases: DiGraph

Base class for s-t augmented graphs (internal).

This class (introduced when unifying :class:stDAG and :class:stDiGraph) factors out logic previously duplicated in both classes.

Core responsibilities

  • Store and expose the original base_graph plus user supplied additional_starts / additional_ends.
  • Validate that all nodes are strings and that any additional start/end nodes belong to base_graph.
  • Create unique global source / sink node identifiers (self.source / self.sink).
  • Attach the global source to every (in-degree 0) source or additional start; attach every (out-degree 0) sink or additional end to the global sink.
  • Expose convenience collections: source_edges, sink_edges, source_sink_edges.
  • Provide shared flow helper utilities: :meth:get_non_zero_flow_edges and :meth:get_max_flow_value_and_check_non_negative_flow.

Extension hooks

Subclasses customise behaviour via two lightweight hooks:

  • _pre_build_validate - extra validation before augmentation (e.g. acyclicity for stDAG).
  • _post_build - populate subclass specific derived structures (e.g. condensation for stDiGraph).

Backwards compatibility

External code should keep instantiating stDAG or stDiGraph; their public APIs are unchanged. AbstractSourceSinkGraph is an internal implementation detail and may change without notice.

Source code in flowpaths/abstractsourcesinkgraph.py 
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
def __init__(
    self,
    base_graph: nx.DiGraph,
    additional_starts: list | None = None,
    additional_ends: list | None = None,
):
    if not all(isinstance(node, str) for node in base_graph.nodes()):
        utils.logger.error(f"{__name__}: Every node of the graph must be a string.")
        raise ValueError("Every node of the graph must be a string.")

    super().__init__()
    self.base_graph = base_graph
    if "id" in base_graph.graph:
        self.id = str(base_graph.graph["id"])
    else:
        self.id = str(id(self))

    self.additional_starts = set(additional_starts or [])
    self.additional_ends = set(additional_ends or [])

    # Ensure any declared additional start/end nodes are in the base graph
    if not self.additional_starts.issubset(base_graph.nodes()):
        utils.logger.error(f"{__name__}: Some nodes in additional_starts are not in the base graph.")
        raise ValueError("Some nodes in additional_starts are not in the base graph.")
    if not self.additional_ends.issubset(base_graph.nodes()):
        utils.logger.error(f"{__name__}: Some nodes in additional_ends are not in the base graph.")
        raise ValueError("Some nodes in additional_ends are not in the base graph.")

    self.source = f"source_{id(self)}"
    self.sink = f"sink_{id(self)}"

    # Hooks
    self._pre_build_validate()
    self._augment_with_source_sink()
    self._post_build()

    nx.freeze(self)

get_max_flow_value_and_check_non_negative_flow 

get_max_flow_value_and_check_non_negative_flow(
    flow_attr: str,
    edges_to_ignore: set,
) -> float

Return maximum value of flow_attr over edges (ignoring some) verifying non-negativity.

Raises ValueError if any required attribute missing or negative.

Source code in flowpaths/abstractsourcesinkgraph.py 
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
def get_max_flow_value_and_check_non_negative_flow(
    self, flow_attr: str, edges_to_ignore: set
) -> float:
    """Return maximum value of `flow_attr` over edges (ignoring some) verifying non-negativity.

    Raises ValueError if any required attribute missing or negative.
    """
    w_max = float("-inf")
    if edges_to_ignore is None:
        edges_to_ignore = set()
    for u, v, data in self.edges(data=True):
        if (u, v) in edges_to_ignore:
            continue
        if flow_attr not in data:
            utils.logger.error(
                f"Edge ({u},{v}) does not have the required flow attribute '{flow_attr}'."
            )
            raise ValueError(
                f"Edge ({u},{v}) does not have the required flow attribute '{flow_attr}'."
            )
        if data[flow_attr] < 0:
            utils.logger.error(
                f"Edge ({u},{v}) has negative flow value {data[flow_attr]}. All flow values must be >=0."
            )
            raise ValueError(
                f"Edge ({u},{v}) has negative flow value {data[flow_attr]}. All flow values must be >=0."
            )
        w_max = max(w_max, data[flow_attr])
    return w_max

get_non_zero_flow_edges 

get_non_zero_flow_edges(
    flow_attr: str,
    edges_to_ignore: set = set(),
) -> set

Return set of edges whose attribute flow_attr is non-zero and not ignored.

Source code in flowpaths/abstractsourcesinkgraph.py 
 97
 98
 99
100
101
102
103
104
105
def get_non_zero_flow_edges(
    self, flow_attr: str, edges_to_ignore: set = set()
) -> set:
    """Return set of edges whose attribute `flow_attr` is non-zero and not ignored."""
    non_zero_flow_edges = set()
    for u, v, data in self.edges(data=True):
        if (u, v) not in edges_to_ignore and data.get(flow_attr, 0) != 0:
            non_zero_flow_edges.add((u, v))
    return non_zero_flow_edges