Source code for pm4py.visualization.network_analysis.variants.performance

'''
    This file is part of PM4Py (More Info: https://pm4py.fit.fraunhofer.de).

    PM4Py is free software: you can redistribute it and/or modify
    it under the terms of the GNU General Public License as published by
    the Free Software Foundation, either version 3 of the License, or
    (at your option) any later version.

    PM4Py is distributed in the hope that it will be useful,
    but WITHOUT ANY WARRANTY; without even the implied warranty of
    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
    GNU General Public License for more details.

    You should have received a copy of the GNU General Public License
    along with PM4Py.  If not, see <https://www.gnu.org/licenses/>.
'''
import sys
import uuid
from enum import Enum
from pm4py.util import exec_utils
import tempfile
from graphviz import Digraph
from pm4py.util import vis_utils
from typing import Dict, Optional, Any, Tuple
from statistics import mean, median, stdev


[docs]class Parameters(Enum): FORMAT = "format" BGCOLOR = "bgcolor" ACTIVITY_THRESHOLD = "activity_threshold" EDGE_THRESHOLD = "edge_threshold" AGGREGATION_MEASURE = "aggregation_measure"
[docs]def apply(network_analysis_edges0: Dict[Tuple[str, str], Dict[str, Any]], parameters: Optional[Dict[Any, Any]] = None) -> Digraph: """ Creates a visualization of the network analysis (performance view) Parameters ----------------- network_analysis_edges0 Edges of the network analysis parameters Parameters of the algorithm, including: - Parameters.FORMAT => the format of the visualization - Parameters.BGCOLOR => the background color - Parameters.ACTIVITY_THRESHOLD => the minimum number of occurrences for an activity to be included (default: 1) - Parameters.EDGE_THRESHOLD => the minimum number of occurrences for an edge to be included (default: 1) - Parameters.AGGREGATION_MEASURE => the aggregation measure (default: mean) Returns ------------------ digraph Graphviz graph """ if parameters is None: parameters = {} image_format = exec_utils.get_param_value(Parameters.FORMAT, parameters, "png") bgcolor = exec_utils.get_param_value(Parameters.BGCOLOR, parameters, "transparent") activity_threshold = exec_utils.get_param_value(Parameters.ACTIVITY_THRESHOLD, parameters, 1) edge_threshold = exec_utils.get_param_value(Parameters.EDGE_THRESHOLD, parameters, 1) aggregation_measure = exec_utils.get_param_value(Parameters.AGGREGATION_MEASURE, parameters, "mean") aggregation_f = mean if aggregation_measure == "median": aggregation_f = median elif aggregation_measure == "min": aggregation_f = min elif aggregation_measure == "max": aggregation_f = max elif aggregation_measure == "stdev": aggregation_f = stdev elif aggregation_measure == "sum": aggregation_f = sum filename = tempfile.NamedTemporaryFile(suffix='.gv') viz = Digraph("pt", filename=filename.name, engine='dot', graph_attr={'bgcolor': bgcolor}) viz.attr('node', shape='ellipse', fixedsize='false') network_analysis_edges = {} network_analysis_edges_agg_performance = {} for x in network_analysis_edges0: network_analysis_edges[x] = {} network_analysis_edges_agg_performance[x] = {} for y in network_analysis_edges0[x]: network_analysis_edges[x][y] = len(network_analysis_edges0[x][y]) network_analysis_edges_agg_performance[x][y] = aggregation_f(network_analysis_edges0[x][y]) nodes = set(x[0] for x in network_analysis_edges).union(set(x[1] for x in network_analysis_edges)) nodes_in_degree = {x: 0 for x in nodes} nodes_out_degree = {x: 0 for x in nodes} for edge in network_analysis_edges: for edge_value in network_analysis_edges[edge]: if network_analysis_edges[edge][edge_value] >= edge_threshold: nodes_in_degree[edge[1]] += network_analysis_edges[edge][edge_value] nodes_out_degree[edge[0]] += network_analysis_edges[edge][edge_value] nodes_max_degree = {x: max(nodes_in_degree[x], nodes_out_degree[x]) for x in nodes} max_node_value = sys.maxsize min_node_value = -sys.maxsize nodes_dict = {} for node in nodes_max_degree: if nodes_max_degree[node] >= activity_threshold: nodes_dict[node] = str(uuid.uuid4()) viz.node(nodes_dict[node], node+"\n(in="+str(nodes_in_degree[node])+"; out="+str(nodes_out_degree[node])+")", style="filled", fillcolor=vis_utils.get_trans_freq_color(nodes_max_degree[node], max_node_value, max_node_value)) count = nodes_max_degree[node] if count > max_node_value: max_node_value = count elif count < min_node_value: min_node_value = count min_edge_value = sys.maxsize max_edge_value = -sys.maxsize for edge in network_analysis_edges: if edge[0] in nodes_dict and edge[1] in nodes_dict: for edge_value in network_analysis_edges[edge]: count = network_analysis_edges[edge][edge_value] if count > max_edge_value: max_edge_value = count elif count < min_edge_value: min_edge_value = count for edge in network_analysis_edges: if edge[0] in nodes_dict and edge[1] in nodes_dict: for edge_value in network_analysis_edges[edge]: if network_analysis_edges[edge][edge_value] >= edge_threshold: viz.edge(nodes_dict[edge[0]], nodes_dict[edge[1]], label=edge_value+"\n"+vis_utils.human_readable_stat(network_analysis_edges_agg_performance[edge][edge_value])+"", penwidth=str(vis_utils.get_arc_penwidth(network_analysis_edges[edge][edge_value], min_edge_value, max_edge_value))) viz.format = image_format return viz