linpershey/process_mining

1import os2from typing import List, Optional, Tuple, Any3from collections import OrderedDict4 5import pandas as pd6from loguru import logger7import pm4py8import plotly.graph_objects as go9import networkx as nx10import matplotlib.pyplot as plt11from PIL import Image12from pydantic import BaseModel13 14 15class ProcessMap(BaseModel):16    net: Any17    start_activities: List | None18    end_activities: List | None19    img: Any | None20 21 22def dfg2networkx( dfg, start, end):23    """Dfg to networkx 24    Argument25        dfg: a list of dict of edges from directly-follow-graph26        start: a dict of start activities27        end: a dict of end activities28    Return29        nx: networkx graph object30    """31    PROCESS_START = '#Start#'32    PROCESS_END = '#End#'33    nodes = { PROCESS_START: 0, PROCESS_END: 1}34    node_idx = 235    for activity in start:36        assert activity not in nodes, f"#ERROR: {activity} exists"37        nodes[activity] = node_idx38        node_idx += 139    for activity in end:40        assert activity not in nodes, f"#ERROR: {activity} exists"41        nodes[activity] = node_idx42        node_idx += 143    for node in dfg:44        left_activity = node[0]45        if left_activity not in nodes:46            nodes[left_activity] = node_idx47            node_idx +=1 48        right_activity = node[1]49        if right_activity not in nodes:50            nodes[right_activity] = node_idx51            node_idx +=1 52    nodes = list(nodes.keys())53    54    edges = []55    for activity in start:56        from_id = str(PROCESS_START)57        to_id = str(activity)58        edges.append( ( PROCESS_START, activity) ) 59    for activity in end:60        from_id = str(activity)61        to_id = str(PROCESS_END)62        edges.append( ( activity, PROCESS_END) ) 63    for transition in dfg:64        edges.append( ( transition[0], transition[1]) ) 65    nx_graph = nx.DiGraph()66    nx_graph.add_nodes_from( nodes)67    nx_graph.add_edges_from(edges)68    return nx_graph69 70 71def discover_process_map_variants( df, top_k: int = 0, type: str = 'dfg'):72    """Discover process map from data frame (raw event log)73    Argument74        df: a pandas dataframe75        top_k: top k variants76        type: dfg or petri77    Return78        dfg, start_activities, end_activities79    """80    event_log = pm4py.format_dataframe( df, case_id='case_id', activity_key='activity', timestamp_key='timestamp')81    if top_k > 0:82        event_log = pm4py.filter_variants_top_k( event_log, k = top_k)83    dfg, start_activities, end_activities = pm4py.discover_dfg(event_log)84    pm4py.view_dfg(dfg, start_activities=start_activities, end_activities=end_activities)85    return dfg, start_activities, end_activities86 87 88def discover_process_map_activities_connections( df, activity_rank: int = 0, connection_rank: int = 0, state: dict = {}, type: str = 'dfg'):89    """Discover process map from data frame (raw event log)90    Argument91        df: a pandas dataframe92        top_k: top k variants93        type: dfg or petri94    Return95        dfg, start_activities, end_activities96    """97    event_log = pm4py.format_dataframe( df, case_id='case_id', activity_key='activity', timestamp_key='timestamp')98    full_dfg, _, __ = pm4py.discover_dfg(event_log)99    ranked_connections = OrderedDict(sorted(full_dfg.items(), key=lambda item: item[1], reverse=True))100 101    if activity_rank > 0:102        pass103    if connection_rank > 0:104        top_variant_connections = state.get('top_variant_connections', [])105        filtered_connections = list(ranked_connections.keys())[ : (connection_rank+ len(ranked_connections))]106    else:107        filtered_connections = list(ranked_connections.keys())108    event_log = pm4py.filter_directly_follows_relation( event_log, relations = filtered_connections)109    dfg, start_activities, end_activities = pm4py.discover_dfg(event_log)110    pm4py.view_dfg(dfg, start_activities=start_activities, end_activities=end_activities)111    return dfg, start_activities, end_activities112 113 114def discover_process_map( df: pd.DataFrame, type: str = 'dfg'):115    """116    """117    event_log = pm4py.format_dataframe( df, case_id='case_id', activity_key='activity', timestamp_key='timestamp')118    if type=='dfg':119        dfg, start_activities, end_activities = pm4py.discover_dfg(event_log)120        pm4py.view_dfg(dfg, start_activities=start_activities, end_activities=end_activities)121        return dfg, start_activities, end_activities122    elif type=='petrinet':123        net, im, fm = pm4py.discover_petri_net_inductive(event_log)124        pm4py.view_petri_net( petri_net=net, initial_marking=im, final_marking=fm)125        file_path = 'output/petri_net.png'126        pm4py.save_vis_petri_net( net, im, fm, file_path)127        img = Image.open(file_path)128        return net, img129    elif type=='bpmn':130        net = pm4py.discover_bpmn_inductive(event_log)131        pm4py.view_bpmn(net, format='png')132        file_path = 'output/bpmn.png'133        pm4py.save_vis_bpmn( net, file_path)134        img = Image.open(file_path)135        return net, img136    else:137        raise Exception(f"Invalid type: {type}")138 139 140def view_networkx( nx_graph, layout):141    """142    Argument143        nx_graph144    Return 145        graph object146    fig.update_xaxes(showticklabels=False)147    fig.update_yaxes(showticklabels=False)148    """149    # Create node scatter plot150    node_trace = go.Scatter(151        x=[layout[n][0] for n in nx_graph.nodes],152        y=[layout[n][1] for n in nx_graph.nodes],153        text=list(nx_graph.nodes),154        mode='markers+text',155        hovertext = [n for n in nx_graph.nodes],156        textposition='top center',157        marker=dict(size=20, color='LightSkyBlue', line=dict(width=2),opacity=0.5)158    )159    160    # Create edge lines161    edge_trace = go.Scatter(162        x=(),163        y=(),164        line=dict(width=1.5, color='#888'),165        hoverinfo='none',166        mode='lines'167    )168    169    # Add arrows for directed edges170    annotations = []171    for edge in nx_graph.edges:172        x0, y0 = layout[edge[0]]173        x1, y1 = layout[edge[1]]174        edge_trace['x'] += (x0, x1, None)175        edge_trace['y'] += (y0, y1, None)176    177        # Calculate direction of the arrow178        annotations.append(179            dict(180                ax=x0,181                ay=y0,182                axref='x',183                ayref='y',184                x=x1,185                y=y1,186                xref='x',187                yref='y',188                showarrow=True,189                arrowhead=2,190                arrowsize=1,191                arrowwidth=5,  # 增加箭头的宽度192                arrowcolor='rgba(128, 128, 128, 0.5)' 193            )194        )195    196    # Draw the figure197    fig = go.Figure(data=[edge_trace, node_trace],198        layout=go.Layout( 199        showlegend=False,200        hovermode='closest',201        margin=dict(b=0, l=0, r=0, t=0),202        annotations=annotations,203        xaxis=dict(showgrid=False, zeroline=False),204        yaxis=dict(showgrid=False, zeroline=False)205    ))206    fig = fig.update_xaxes(showticklabels=False)207    fig = fig.update_yaxes(showticklabels=False)208    return fig209 210 211def view_process_map( nx_graph, process_type: str = 'dfg', layout_type: str = 'sfdp'):212    """213    """214    layout = nx.nx_agraph.graphviz_layout( nx_graph, prog=layout_type)215    # min_x, max_x = min([ node_loc[0] for node, node_loc in layout.items()]), max([ node_loc[0] for node, node_loc in layout.items()])216    # min_y, max_y = min([ node_loc[1] for node, node_loc in layout.items()]), max([ node_loc[1] for node, node_loc in layout.items()])217    # layout['#Start#'] = ( min_x, min_y)218    # layout['#End#'] = (max_x, max_y)219    fig = view_networkx(nx_graph, layout)220    return fig