Skip to content

ragraph.analysis.comparison

Comparison analysis

Comparison provides classes for comparing Graph objects to find the commonalities (sigma) and differences (delta).

Comparison methods

  • sigma_graph to calculate the commonalities (i.e. sum) of a set of graphs with comparable node names.
  • delta_graph to calculate the differences and overlap between two graphs with comparable node names.

EdgeDescriptor dataclass

Bases: EdgeDescriptorLike

Describe a node based on its source name, target name, kind, labels, and weights.

EdgeDescriptorLike

Bases: ABC

A class that takes an edge and describes it by returning a fixed (hashable) output.

Hashable means that the description can be used to determine uniqueness by set operations.

from_edge abstractmethod classmethod 

1
from_edge(edge: Edge) -> EdgeDescriptor

Create an edge descriptor from an edge. Ignores annotations.

Source code in ragraph/analysis/comparison/utils.py 
71
72
73
74
75
@classmethod
@abstractmethod
def from_edge(cls, edge: Edge) -> "EdgeDescriptor":
    """Create an edge descriptor from an edge. Ignores annotations."""
    ...

to_edge abstractmethod 

1
to_edge(graph: Graph) -> Edge

Create an edge from a descriptor in a graph.

Source code in ragraph/analysis/comparison/utils.py 
77
78
79
80
@abstractmethod
def to_edge(self, graph: Graph) -> Edge:
    """Create an edge from a descriptor in a graph."""
    ...

NodeDescriptor dataclass

Bases: NodeDescriptorLike

Describe a node based on its name, kind, labels and weights.

NodeDescriptorLike

Bases: ABC

A class that takes a node and describes it by returning a fixed (hashable) output.

Hashable means that the description can be used to determine uniqueness by set operations.

from_node abstractmethod classmethod 

1
from_node(node: Node) -> NodeDescriptorLike

Create a node descriptor from a node.

Source code in ragraph/analysis/comparison/utils.py 
24
25
26
27
28
@classmethod
@abstractmethod
def from_node(cls, node: Node) -> "NodeDescriptorLike":
    """Create a node descriptor from a node."""
    raise NotImplementedError

to_node abstractmethod 

1
to_node(graph: Graph) -> Node

Create a node from a descriptor in a graph and return it for convenience.

Source code in ragraph/analysis/comparison/utils.py 
30
31
32
33
@abstractmethod
def to_node(self, graph: Graph) -> Node:
    """Create a node from a descriptor in a graph and return it for convenience."""
    raise NotImplementedError

SigmaMode

Bases: StrEnum

Aggregation mode for sigma analysis, absolute counted occurrences or an average per graph.

TagMode

Bases: StrEnum

How to tag nodes as unique to one of the given inputs or common.

add_edges 

1
2
3
add_edges(
    accumulator: Graph, item: Graph, count_prefix: str
)

Add observed edges in item to the accumulator in-place.

Based on node names.

Metadata for existing nodes is added using add_meta method.

Source code in ragraph/analysis/comparison/_sigma.py 
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
def add_edges(accumulator: Graph, item: Graph, count_prefix: str):
    """Add observed edges in item to the accumulator in-place.

    Based on node names.

    Metadata for existing nodes is added using
    [`add_meta` method][ragraph.analysis.comparison.add_meta].
    """
    for source_name, tgt_map in item.directed_edges.items():
        for target_name, item_edges in tgt_map.items():
            # Take a deepcopy of the list of edges in the item and aggregate those first.
            item_edges = deepcopy(item_edges)
            item_edge = item_edges.pop()
            for e in item_edges:
                add_meta(item_edge, e, count_prefix)

            # Now check whether we already accumulated something here, or add a new entry.
            edges = accumulator.directed_edges[source_name][target_name]
            if len(edges):
                add_meta(edges[0], item_edge, count_prefix)
            else:
                source_node = accumulator.node_dict[source_name]
                target_node = accumulator.node_dict[target_name]
                edge = Edge(source_node, target_node)
                add_meta(edge, item_edge, count_prefix, is_first=True)
                accumulator.add_edge(edge)

add_graph 

1
2
3
4
5
6
add_graph(
    accumulator: Graph,
    item: Graph,
    count_prefix: str,
    is_first: bool = False,
) -> Graph

Add a graph to an accumulator graph. Occurrences are counted under the 'count_prefix'.

Source code in ragraph/analysis/comparison/_sigma.py 
62
63
64
65
66
67
def add_graph(accumulator: Graph, item: Graph, count_prefix: str, is_first: bool = False) -> Graph:
    """Add a graph to an accumulator graph. Occurrences are counted under the 'count_prefix'."""
    add_meta(accumulator, item, count_prefix, is_first=is_first)
    add_nodes(accumulator, item, count_prefix)
    add_edges(accumulator, item, count_prefix)
    return accumulator

add_meta 

1
2
3
4
5
6
add_meta(
    accumulator: Metadata,
    item: Metadata,
    count_prefix: str,
    is_first: bool = False,
)

Add metadata of another item to the accumulator in-place. Annotations are updated and the last observed value for any key is kept. Occurrence value is tracked under the 'count_prefix'.

Source code in ragraph/analysis/comparison/_sigma.py 
70
71
72
73
74
75
76
77
def add_meta(accumulator: Metadata, item: Metadata, count_prefix: str, is_first: bool = False):
    """Add metadata of another item to the accumulator in-place. Annotations are updated and the
    last observed value for any key is kept. Occurrence value is tracked under the 'count_prefix'.
    """
    if is_first:
        initialize_meta(accumulator, item, count_prefix)
    else:
        increment_meta(accumulator, item, count_prefix)

add_nodes 

1
2
3
add_nodes(
    accumulator: Graph, item: Graph, count_prefix: str
)

Add observed nodes in item to the accumulator in-place and keep track of an occurrence count.

Based on node names.

Metadata for existing nodes is added using add_meta method.

Source code in ragraph/analysis/comparison/_sigma.py 
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
def add_nodes(accumulator: Graph, item: Graph, count_prefix: str):
    """Add observed nodes in item to the accumulator in-place and keep track of an occurrence count.

    Based on node names.

    Metadata for existing nodes is added using
    [`add_meta` method][ragraph.analysis.comparison.add_meta].
    """
    for n in item.leafs:
        try:
            add_meta(accumulator[n.name], n, count_prefix)
        except KeyError:
            node = Node(name=n.name)
            add_meta(node, n, count_prefix, is_first=True)
            accumulator.add_node(node)

delta_graph 

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
delta_graph(
    graph_a: Graph,
    graph_b: Graph,
    delta_a: str = "delta_a",
    delta_b: str = "delta_b",
    common: str = "common",
    tag_mode: TagMode = TagMode.KIND,
    node_descriptor: Type[
        NodeDescriptorLike
    ] = NodeDescriptor,
    edge_descriptor: Type[
        EdgeDescriptorLike
    ] = EdgeDescriptor,
) -> Graph

Get the delta graph between two graphs.

Parameters:

Name Type Description Default
graph_a Graph

First graph to compare.

 required
graph_b Graph

Second graph to compare.

 required
delta_a str

Name for nodes and edges unique to the first graph.

 'delta_a'
delta_b str

Name for nodes and edges unique to the second graph.

 'delta_b'
common str

Name for the common nodes and edges occurring in both graphs.

 'common'
Note

Graphs are compared on leaf node level and compared by name only. Please provide appropriate input as such and look to get_graph_slice for example on how to obtain a subgraph to your liking.

Source code in ragraph/analysis/comparison/_delta.py 
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
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
71
72
73
74
75
76
def delta_graph(
    graph_a: Graph,
    graph_b: Graph,
    delta_a: str = "delta_a",
    delta_b: str = "delta_b",
    common: str = "common",
    tag_mode: TagMode = TagMode.KIND,
    node_descriptor: Type[NodeDescriptorLike] = NodeDescriptor,
    edge_descriptor: Type[EdgeDescriptorLike] = EdgeDescriptor,
) -> Graph:
    """Get the delta graph between two graphs.

    Arguments:
        graph_a: First graph to compare.
        graph_b: Second graph to compare.
        delta_a: Name for nodes and edges unique to the first graph.
        delta_b: Name for nodes and edges unique to the second graph.
        common: Name for the common nodes and edges occurring in both graphs.

    Note:
        Graphs are compared on leaf node level and compared by name only.
        Please provide appropriate input as such and look to
        [`get_graph_slice`][ragraph.graph.Graph.get_graph_slice] for example on how to obtain a
        subgraph to your liking.
    """
    graph = Graph(name="delta", kind="delta")

    # Compare leaf nodes by name. Set to unique A first and change to common if found again.
    nds_a = set(node_descriptor.from_node(n) for n in graph_a.leafs)
    nds_b = set(node_descriptor.from_node(n) for n in graph_b.leafs)
    for n in nds_a.difference(nds_b):
        node = n.to_node(graph)
        tag(node, delta_a, tag_mode)
    for n in nds_b.difference(nds_a):
        node = n.to_node(graph)
        tag(node, delta_b, tag_mode)
    for n in nds_a.intersection(nds_b):
        node = n.to_node(graph)
        tag(node, common, tag_mode)

    # Compare edges between the nodes in the newly created graph.
    eds_a = set(
        edge_descriptor.from_edge(e)
        for e in graph_a.edges_between_all(graph_a.leafs, graph_a.leafs, inherit=False, loops=True)
    )
    eds_b = set(
        edge_descriptor.from_edge(e)
        for e in graph_b.edges_between_all(graph_b.leafs, graph_b.leafs, inherit=False, loops=True)
    )
    for e in eds_a.difference(eds_b):
        edge = e.to_edge(graph)
        tag(edge, delta_a, tag_mode)
    for e in eds_b.difference(eds_a):
        edge = e.to_edge(graph)
        tag(edge, delta_b, tag_mode)
    for e in eds_a.intersection(eds_b):
        edge = e.to_edge(graph)
        tag(edge, common, tag_mode)

    return graph

increment_meta 

1
2
3
increment_meta(
    accumulator: Metadata, item: Metadata, count_prefix: str
)

Increment the occurrence and label counts as weights in the accumulator with an item.

Source code in ragraph/analysis/comparison/_sigma.py 
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
def increment_meta(accumulator: Metadata, item: Metadata, count_prefix: str):
    """Increment the occurrence and label counts as weights in the accumulator with an item."""
    if accumulator.kind != item.kind:
        raise ValueError(f"kinds don't match: '{accumulator.kind}' and '{item.kind}'")

    # Update labels to superset.
    accumulator.labels = sorted(set(accumulator.labels + item.labels))

    # Sum weights
    for k, v in item.weights.items():
        accumulator.weights[k] = accumulator.weights.get(k, 0) + v
    accumulator.annotations.update(item.annotations)

    if count_prefix in item.weights:
        pass  # Is already incremented when summing weights above.
    else:
        accumulator.weights[count_prefix] = accumulator.weights.get(count_prefix, 0) + 1

    # Increment label counts.
    for label in item.labels:
        k = f"{count_prefix}_label_{label}"
        if k in item.weights:
            continue  # Is already summed with weights.
        accumulator.weights[k] = accumulator.weights.get(k, 0) + 1

initialize_meta 

1
2
3
initialize_meta(
    accumulator: Metadata, item: Metadata, count_prefix: str
)

Initialize an accumulator instance with the info from the first item.

Source code in ragraph/analysis/comparison/_sigma.py 
80
81
82
83
84
85
86
87
88
89
def initialize_meta(accumulator: Metadata, item: Metadata, count_prefix: str):
    """Initialize an accumulator instance with the info from the first item."""
    accumulator.kind = item.kind
    accumulator.labels = deepcopy(item.labels)
    accumulator.weights = deepcopy(item.weights)
    accumulator.weights[count_prefix] = 1
    for label in item.labels:
        k = f"{count_prefix}_label_{label}"
        accumulator.weights[k] = item.weights.get(k, 1)
    accumulator.annotations = deepcopy(item.annotations)

sigma_graph 

1
2
3
4
5
sigma_graph(
    graphs: Iterable[Graph],
    count_prefix: str = "sigma",
    mode: SigmaMode = SigmaMode.ABSOLUTE,
) -> Graph

Get the sigma (summed) graph based on the leaf nodes and the edges in multiple graphs.

Parameters:

Name Type Description Default
graphs Iterable[Graph]

Graphs to sum and count edge occurrences in.

 required
mode SigmaMode

Whether to count absolute occurrence values or an average per graph.

 ABSOLUTE
count_prefix str

Which weight key (prefix) to store the occurrence values under. Also used as a prefix for label counts that are stored as weights using as well.

 'sigma'
Note

Summations are done over (leaf) node names. Edges in an input graph are aggregated first using the add_meta method before being aggregated into the resulting graph.

Source code in ragraph/analysis/comparison/_sigma.py 
24
25
26
27
28
29
30
31
32
33
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
def sigma_graph(
    graphs: Iterable[Graph], count_prefix: str = "sigma", mode: SigmaMode = SigmaMode.ABSOLUTE
) -> Graph:
    """Get the sigma (summed) graph based on the leaf nodes and the edges in multiple graphs.

    Arguments:
        graphs: Graphs to sum and count edge occurrences in.
        mode: Whether to count absolute occurrence values or an average per graph.
        count_prefix: Which weight key (prefix) to store the occurrence values under. Also used as
            a prefix for label counts that are stored as weights using as well.

    Note:
        Summations are done over (leaf) node names. Edges in an input graph are aggregated first
        using the [`add_meta` method][ragraph.analysis.comparison.add_meta]
        before being aggregated into the resulting graph.
    """
    accumulator = Graph(name="sigma")
    num = 0
    for graph in graphs:
        # Override that first one's results because of default values for a Graph.
        add_graph(accumulator, graph, count_prefix, num == 0)
        num += 1
    if mode == SigmaMode.AVERAGE and num > 0:
        div = 1.0 / num
        for k, v in accumulator.weights.items():
            if k.startswith(f"{count_prefix}_") or k == count_prefix:
                accumulator.weights[k] = div * v
        for n in accumulator.nodes:
            for k, v in n.weights.items():
                if k.startswith(f"{count_prefix}_") or k == count_prefix:
                    n.weights[k] = div * v
        for e in accumulator.edges:
            for k, v in e.weights.items():
                if k.startswith(f"{count_prefix}_") or k == count_prefix:
                    e.weights[k] = div * v
    return accumulator

tag 

1
tag(item: Metadata, tag: str, mode: TagMode)

Tag an item using a given tagging mode.

Source code in ragraph/analysis/comparison/utils.py 
123
124
125
126
127
128
129
130
131
132
def tag(item: Metadata, tag: str, mode: TagMode):
    """Tag an item using a given tagging mode."""
    if mode == TagMode.KIND:
        item.kind = tag
    elif mode == TagMode.LABEL:
        item.labels = item.labels + [tag]
    elif mode == TagMode.ANNOTATION:
        item.annotations[tag] = True
    else:
        raise ValueError("Unknown tagging mode.")

_delta

Delta analysis

delta_graph 

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
delta_graph(
    graph_a: Graph,
    graph_b: Graph,
    delta_a: str = "delta_a",
    delta_b: str = "delta_b",
    common: str = "common",
    tag_mode: TagMode = TagMode.KIND,
    node_descriptor: Type[
        NodeDescriptorLike
    ] = NodeDescriptor,
    edge_descriptor: Type[
        EdgeDescriptorLike
    ] = EdgeDescriptor,
) -> Graph

Get the delta graph between two graphs.

Parameters:

Name Type Description Default
graph_a Graph

First graph to compare.

 required
graph_b Graph

Second graph to compare.

 required
delta_a str

Name for nodes and edges unique to the first graph.

 'delta_a'
delta_b str

Name for nodes and edges unique to the second graph.

 'delta_b'
common str

Name for the common nodes and edges occurring in both graphs.

 'common'
Note

Graphs are compared on leaf node level and compared by name only. Please provide appropriate input as such and look to get_graph_slice for example on how to obtain a subgraph to your liking.

Source code in ragraph/analysis/comparison/_delta.py 
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
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
71
72
73
74
75
76
def delta_graph(
    graph_a: Graph,
    graph_b: Graph,
    delta_a: str = "delta_a",
    delta_b: str = "delta_b",
    common: str = "common",
    tag_mode: TagMode = TagMode.KIND,
    node_descriptor: Type[NodeDescriptorLike] = NodeDescriptor,
    edge_descriptor: Type[EdgeDescriptorLike] = EdgeDescriptor,
) -> Graph:
    """Get the delta graph between two graphs.

    Arguments:
        graph_a: First graph to compare.
        graph_b: Second graph to compare.
        delta_a: Name for nodes and edges unique to the first graph.
        delta_b: Name for nodes and edges unique to the second graph.
        common: Name for the common nodes and edges occurring in both graphs.

    Note:
        Graphs are compared on leaf node level and compared by name only.
        Please provide appropriate input as such and look to
        [`get_graph_slice`][ragraph.graph.Graph.get_graph_slice] for example on how to obtain a
        subgraph to your liking.
    """
    graph = Graph(name="delta", kind="delta")

    # Compare leaf nodes by name. Set to unique A first and change to common if found again.
    nds_a = set(node_descriptor.from_node(n) for n in graph_a.leafs)
    nds_b = set(node_descriptor.from_node(n) for n in graph_b.leafs)
    for n in nds_a.difference(nds_b):
        node = n.to_node(graph)
        tag(node, delta_a, tag_mode)
    for n in nds_b.difference(nds_a):
        node = n.to_node(graph)
        tag(node, delta_b, tag_mode)
    for n in nds_a.intersection(nds_b):
        node = n.to_node(graph)
        tag(node, common, tag_mode)

    # Compare edges between the nodes in the newly created graph.
    eds_a = set(
        edge_descriptor.from_edge(e)
        for e in graph_a.edges_between_all(graph_a.leafs, graph_a.leafs, inherit=False, loops=True)
    )
    eds_b = set(
        edge_descriptor.from_edge(e)
        for e in graph_b.edges_between_all(graph_b.leafs, graph_b.leafs, inherit=False, loops=True)
    )
    for e in eds_a.difference(eds_b):
        edge = e.to_edge(graph)
        tag(edge, delta_a, tag_mode)
    for e in eds_b.difference(eds_a):
        edge = e.to_edge(graph)
        tag(edge, delta_b, tag_mode)
    for e in eds_a.intersection(eds_b):
        edge = e.to_edge(graph)
        tag(edge, common, tag_mode)

    return graph

_sigma

Sigma analysis

SigmaMode

Bases: StrEnum

Aggregation mode for sigma analysis, absolute counted occurrences or an average per graph.

add_edges 

1
2
3
add_edges(
    accumulator: Graph, item: Graph, count_prefix: str
)

Add observed edges in item to the accumulator in-place.

Based on node names.

Metadata for existing nodes is added using add_meta method.

Source code in ragraph/analysis/comparison/_sigma.py 
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
def add_edges(accumulator: Graph, item: Graph, count_prefix: str):
    """Add observed edges in item to the accumulator in-place.

    Based on node names.

    Metadata for existing nodes is added using
    [`add_meta` method][ragraph.analysis.comparison.add_meta].
    """
    for source_name, tgt_map in item.directed_edges.items():
        for target_name, item_edges in tgt_map.items():
            # Take a deepcopy of the list of edges in the item and aggregate those first.
            item_edges = deepcopy(item_edges)
            item_edge = item_edges.pop()
            for e in item_edges:
                add_meta(item_edge, e, count_prefix)

            # Now check whether we already accumulated something here, or add a new entry.
            edges = accumulator.directed_edges[source_name][target_name]
            if len(edges):
                add_meta(edges[0], item_edge, count_prefix)
            else:
                source_node = accumulator.node_dict[source_name]
                target_node = accumulator.node_dict[target_name]
                edge = Edge(source_node, target_node)
                add_meta(edge, item_edge, count_prefix, is_first=True)
                accumulator.add_edge(edge)

add_graph 

1
2
3
4
5
6
add_graph(
    accumulator: Graph,
    item: Graph,
    count_prefix: str,
    is_first: bool = False,
) -> Graph

Add a graph to an accumulator graph. Occurrences are counted under the 'count_prefix'.

Source code in ragraph/analysis/comparison/_sigma.py 
62
63
64
65
66
67
def add_graph(accumulator: Graph, item: Graph, count_prefix: str, is_first: bool = False) -> Graph:
    """Add a graph to an accumulator graph. Occurrences are counted under the 'count_prefix'."""
    add_meta(accumulator, item, count_prefix, is_first=is_first)
    add_nodes(accumulator, item, count_prefix)
    add_edges(accumulator, item, count_prefix)
    return accumulator

add_meta 

1
2
3
4
5
6
add_meta(
    accumulator: Metadata,
    item: Metadata,
    count_prefix: str,
    is_first: bool = False,
)

Add metadata of another item to the accumulator in-place. Annotations are updated and the last observed value for any key is kept. Occurrence value is tracked under the 'count_prefix'.

Source code in ragraph/analysis/comparison/_sigma.py 
70
71
72
73
74
75
76
77
def add_meta(accumulator: Metadata, item: Metadata, count_prefix: str, is_first: bool = False):
    """Add metadata of another item to the accumulator in-place. Annotations are updated and the
    last observed value for any key is kept. Occurrence value is tracked under the 'count_prefix'.
    """
    if is_first:
        initialize_meta(accumulator, item, count_prefix)
    else:
        increment_meta(accumulator, item, count_prefix)

add_nodes 

1
2
3
add_nodes(
    accumulator: Graph, item: Graph, count_prefix: str
)

Add observed nodes in item to the accumulator in-place and keep track of an occurrence count.

Based on node names.

Metadata for existing nodes is added using add_meta method.

Source code in ragraph/analysis/comparison/_sigma.py 
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
def add_nodes(accumulator: Graph, item: Graph, count_prefix: str):
    """Add observed nodes in item to the accumulator in-place and keep track of an occurrence count.

    Based on node names.

    Metadata for existing nodes is added using
    [`add_meta` method][ragraph.analysis.comparison.add_meta].
    """
    for n in item.leafs:
        try:
            add_meta(accumulator[n.name], n, count_prefix)
        except KeyError:
            node = Node(name=n.name)
            add_meta(node, n, count_prefix, is_first=True)
            accumulator.add_node(node)

increment_meta 

1
2
3
increment_meta(
    accumulator: Metadata, item: Metadata, count_prefix: str
)

Increment the occurrence and label counts as weights in the accumulator with an item.

Source code in ragraph/analysis/comparison/_sigma.py 
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
def increment_meta(accumulator: Metadata, item: Metadata, count_prefix: str):
    """Increment the occurrence and label counts as weights in the accumulator with an item."""
    if accumulator.kind != item.kind:
        raise ValueError(f"kinds don't match: '{accumulator.kind}' and '{item.kind}'")

    # Update labels to superset.
    accumulator.labels = sorted(set(accumulator.labels + item.labels))

    # Sum weights
    for k, v in item.weights.items():
        accumulator.weights[k] = accumulator.weights.get(k, 0) + v
    accumulator.annotations.update(item.annotations)

    if count_prefix in item.weights:
        pass  # Is already incremented when summing weights above.
    else:
        accumulator.weights[count_prefix] = accumulator.weights.get(count_prefix, 0) + 1

    # Increment label counts.
    for label in item.labels:
        k = f"{count_prefix}_label_{label}"
        if k in item.weights:
            continue  # Is already summed with weights.
        accumulator.weights[k] = accumulator.weights.get(k, 0) + 1

initialize_meta 

1
2
3
initialize_meta(
    accumulator: Metadata, item: Metadata, count_prefix: str
)

Initialize an accumulator instance with the info from the first item.

Source code in ragraph/analysis/comparison/_sigma.py 
80
81
82
83
84
85
86
87
88
89
def initialize_meta(accumulator: Metadata, item: Metadata, count_prefix: str):
    """Initialize an accumulator instance with the info from the first item."""
    accumulator.kind = item.kind
    accumulator.labels = deepcopy(item.labels)
    accumulator.weights = deepcopy(item.weights)
    accumulator.weights[count_prefix] = 1
    for label in item.labels:
        k = f"{count_prefix}_label_{label}"
        accumulator.weights[k] = item.weights.get(k, 1)
    accumulator.annotations = deepcopy(item.annotations)

sigma_graph 

1
2
3
4
5
sigma_graph(
    graphs: Iterable[Graph],
    count_prefix: str = "sigma",
    mode: SigmaMode = SigmaMode.ABSOLUTE,
) -> Graph

Get the sigma (summed) graph based on the leaf nodes and the edges in multiple graphs.

Parameters:

Name Type Description Default
graphs Iterable[Graph]

Graphs to sum and count edge occurrences in.

 required
mode SigmaMode

Whether to count absolute occurrence values or an average per graph.

 ABSOLUTE
count_prefix str

Which weight key (prefix) to store the occurrence values under. Also used as a prefix for label counts that are stored as weights using as well.

 'sigma'
Note

Summations are done over (leaf) node names. Edges in an input graph are aggregated first using the add_meta method before being aggregated into the resulting graph.

Source code in ragraph/analysis/comparison/_sigma.py 
24
25
26
27
28
29
30
31
32
33
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
def sigma_graph(
    graphs: Iterable[Graph], count_prefix: str = "sigma", mode: SigmaMode = SigmaMode.ABSOLUTE
) -> Graph:
    """Get the sigma (summed) graph based on the leaf nodes and the edges in multiple graphs.

    Arguments:
        graphs: Graphs to sum and count edge occurrences in.
        mode: Whether to count absolute occurrence values or an average per graph.
        count_prefix: Which weight key (prefix) to store the occurrence values under. Also used as
            a prefix for label counts that are stored as weights using as well.

    Note:
        Summations are done over (leaf) node names. Edges in an input graph are aggregated first
        using the [`add_meta` method][ragraph.analysis.comparison.add_meta]
        before being aggregated into the resulting graph.
    """
    accumulator = Graph(name="sigma")
    num = 0
    for graph in graphs:
        # Override that first one's results because of default values for a Graph.
        add_graph(accumulator, graph, count_prefix, num == 0)
        num += 1
    if mode == SigmaMode.AVERAGE and num > 0:
        div = 1.0 / num
        for k, v in accumulator.weights.items():
            if k.startswith(f"{count_prefix}_") or k == count_prefix:
                accumulator.weights[k] = div * v
        for n in accumulator.nodes:
            for k, v in n.weights.items():
                if k.startswith(f"{count_prefix}_") or k == count_prefix:
                    n.weights[k] = div * v
        for e in accumulator.edges:
            for k, v in e.weights.items():
                if k.startswith(f"{count_prefix}_") or k == count_prefix:
                    e.weights[k] = div * v
    return accumulator

utils

Comparison analysis utilities

EdgeDescriptor dataclass

Bases: EdgeDescriptorLike

Describe a node based on its source name, target name, kind, labels, and weights.

EdgeDescriptorLike

Bases: ABC

A class that takes an edge and describes it by returning a fixed (hashable) output.

Hashable means that the description can be used to determine uniqueness by set operations.

from_edge abstractmethod classmethod 

1
from_edge(edge: Edge) -> EdgeDescriptor

Create an edge descriptor from an edge. Ignores annotations.

Source code in ragraph/analysis/comparison/utils.py 
71
72
73
74
75
@classmethod
@abstractmethod
def from_edge(cls, edge: Edge) -> "EdgeDescriptor":
    """Create an edge descriptor from an edge. Ignores annotations."""
    ...

to_edge abstractmethod 

1
to_edge(graph: Graph) -> Edge

Create an edge from a descriptor in a graph.

Source code in ragraph/analysis/comparison/utils.py 
77
78
79
80
@abstractmethod
def to_edge(self, graph: Graph) -> Edge:
    """Create an edge from a descriptor in a graph."""
    ...

NodeDescriptor dataclass

Bases: NodeDescriptorLike

Describe a node based on its name, kind, labels and weights.

NodeDescriptorLike

Bases: ABC

A class that takes a node and describes it by returning a fixed (hashable) output.

Hashable means that the description can be used to determine uniqueness by set operations.

from_node abstractmethod classmethod 

1
from_node(node: Node) -> NodeDescriptorLike

Create a node descriptor from a node.

Source code in ragraph/analysis/comparison/utils.py 
24
25
26
27
28
@classmethod
@abstractmethod
def from_node(cls, node: Node) -> "NodeDescriptorLike":
    """Create a node descriptor from a node."""
    raise NotImplementedError

to_node abstractmethod 

1
to_node(graph: Graph) -> Node

Create a node from a descriptor in a graph and return it for convenience.

Source code in ragraph/analysis/comparison/utils.py 
30
31
32
33
@abstractmethod
def to_node(self, graph: Graph) -> Node:
    """Create a node from a descriptor in a graph and return it for convenience."""
    raise NotImplementedError

TagMode

Bases: StrEnum

How to tag nodes as unique to one of the given inputs or common.

tag 

1
tag(item: Metadata, tag: str, mode: TagMode)

Tag an item using a given tagging mode.

Source code in ragraph/analysis/comparison/utils.py 
123
124
125
126
127
128
129
130
131
132
def tag(item: Metadata, tag: str, mode: TagMode):
    """Tag an item using a given tagging mode."""
    if mode == TagMode.KIND:
        item.kind = tag
    elif mode == TagMode.LABEL:
        item.labels = item.labels + [tag]
    elif mode == TagMode.ANNOTATION:
        item.annotations[tag] = True
    else:
        raise ValueError("Unknown tagging mode.")