processes.processes_graph_model()   B
last analyzed

Complexity

Conditions 6

Size

Total Lines 36
Code Lines 21

Duplication

Lines 0
Ratio 0 %

Importance

Changes 0
Metric Value
cc 6
eloc 21
nop 0
dl 0
loc 36
rs 8.4426
c 0
b 0
f 0
1
import operator
2
from collections.abc import Iterable
3
4
import networkx as nx
5
import psutil
6
7
import graphinate
8
9
10
def processes_graph_model():
11
    """
12
    Create a graph model representing processes and their parent-child relationships.
13
14
    Returns:
15
        GraphModel: A graph model representing processes and their parent-child relationships.
16
    """
17
18
    graph_model = graphinate.model("Processes Graph")
19
20
    def processes() -> Iterable[psutil.Process]:
21
        for pid in psutil.pids():
22
            if psutil.pid_exists(pid):
23
                yield psutil.Process(pid)
24
25
    processes_list = [
26
        {
27
            'pid': p.pid,
28
            'name': p.name(),
29
            'parent_pid': p.parent().pid if p.parent() else None
30
        }
31
        for p in processes()
32
    ]
33
34
    @graph_model.node(key=operator.itemgetter('pid'), label=operator.itemgetter('name'))
35
    def process():
36
        yield from processes_list
37
38
    @graph_model.edge()
39
    def edge():
40
        for p in processes_list:
41
            parent_pid = p.get('parent_pid')
42
            if parent_pid:
43
                yield {'source': p.get('pid'), 'target': parent_pid}
44
45
    return graph_model
46
47
48
model = processes_graph_model()
49
50
if __name__ == '__main__':
51
    # 1. Define Graph Builder
52
    builder = graphinate.builders.NetworkxBuilder(model=model)
53
54
    # Then
55
    # 2. Build the Graph object
56
    graph: nx.Graph = builder.build()
57
58
    # Then
59
    # 3. Option A - Output to console
60
    print(graph)
61
62
    # Or
63
    # 3. Option B - Output as a plot
64
    graphinate.materializers.plot(graph)
65