Source code for pm4py.convert

'''
    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/>.
'''
from typing import Union, Tuple

import pandas as pd

from pm4py.objects.bpmn.obj import BPMN
from pm4py.objects.heuristics_net.obj import HeuristicsNet
from pm4py.objects.log.obj import EventLog, EventStream
from pm4py.objects.petri_net.obj import PetriNet, Marking
from pm4py.objects.process_tree.obj import ProcessTree
from pm4py.utils import get_properties


[docs]def convert_to_event_log(obj: Union[pd.DataFrame, EventStream]) -> EventLog: """ Converts a log object to an event log Parameters ------------- obj Log object Returns ------------- log Event log object """ if type(obj) not in [pd.DataFrame, EventLog, EventStream]: raise Exception("the method can be applied only to a traditional event log!") from pm4py.objects.conversion.log import converter log = converter.apply(obj, variant=converter.Variants.TO_EVENT_LOG, parameters=get_properties(obj)) return log
[docs]def convert_to_event_stream(obj: Union[EventLog, pd.DataFrame]) -> EventStream: """ Converts a log object to an event stream Parameters -------------- obj Log object Returns -------------- stream Event stream object """ if type(obj) not in [pd.DataFrame, EventLog, EventStream]: raise Exception("the method can be applied only to a traditional event log!") from pm4py.objects.conversion.log import converter stream = converter.apply(obj, variant=converter.Variants.TO_EVENT_STREAM, parameters=get_properties(obj)) return stream
[docs]def convert_to_dataframe(obj: Union[EventStream, EventLog]) -> pd.DataFrame: """ Converts a log object to a dataframe Parameters -------------- obj Log object Returns -------------- df Dataframe """ if type(obj) not in [pd.DataFrame, EventLog, EventStream]: raise Exception("the method can be applied only to a traditional event log!") from pm4py.objects.conversion.log import converter df = converter.apply(obj, variant=converter.Variants.TO_DATA_FRAME, parameters=get_properties(obj)) return df
[docs]def convert_to_bpmn(*args: Union[Tuple[PetriNet, Marking, Marking], ProcessTree]) -> BPMN: """ Converts an object to a BPMN diagram Parameters -------------- *args Object (process tree) Returns -------------- bpmn_diagram BPMN diagram """ from pm4py.objects.process_tree.obj import ProcessTree from pm4py.objects.bpmn.obj import BPMN if isinstance(args[0], BPMN): # the object is already a BPMN return args[0] elif isinstance(args[0], ProcessTree): from pm4py.objects.conversion.process_tree.variants import to_bpmn return to_bpmn.apply(args[0]) else: # try to convert the object to a Petri net. Then, use the PM4Py PN-to-BPMN converter # to get the BPMN object try: net, im, fm = convert_to_petri_net(*args) from pm4py.objects.conversion.wf_net.variants import to_bpmn return to_bpmn.apply(net, im, fm) except: # don't do nothing and throw the following exception pass # if no conversion is done, then the format of the arguments is unsupported raise Exception("unsupported conversion of the provided object to BPMN")
[docs]def convert_to_petri_net(*args: Union[BPMN, ProcessTree, HeuristicsNet, dict]) -> Tuple[PetriNet, Marking, Marking]: """ Converts an object to an (accepting) Petri net Parameters -------------- *args Object (process tree, BPMN) Returns -------------- net Petri net im Initial marking fm Final marking """ if isinstance(args[0], PetriNet): # the object is already a Petri net return args[0], args[1], args[2] elif isinstance(args[0], ProcessTree): from pm4py.objects.conversion.process_tree.variants import to_petri_net return to_petri_net.apply(args[0]) elif isinstance(args[0], BPMN): from pm4py.objects.conversion.bpmn.variants import to_petri_net return to_petri_net.apply(args[0]) elif isinstance(args[0], HeuristicsNet): from pm4py.objects.conversion.heuristics_net.variants import to_petri_net return to_petri_net.apply(args[0]) elif isinstance(args[0], dict): # DFG from pm4py.objects.conversion.dfg.variants import to_petri_net_activity_defines_place return to_petri_net_activity_defines_place.apply(args[0], parameters={ to_petri_net_activity_defines_place.Parameters.START_ACTIVITIES: args[1], to_petri_net_activity_defines_place.Parameters.END_ACTIVITIES: args[2]}) # if no conversion is done, then the format of the arguments is unsupported raise Exception("unsupported conversion of the provided object to Petri net")
[docs]def convert_to_process_tree(*args: Union[Tuple[PetriNet, Marking, Marking], BPMN]) -> ProcessTree: """ Converts an object to a process tree Parameters -------------- *args Object (Petri net, BPMN) Returns -------------- tree Process tree (when the model is block-structured) """ from pm4py.objects.process_tree.obj import ProcessTree from pm4py.objects.petri_net.obj import PetriNet if isinstance(args[0], ProcessTree): # the object is already a process tree return args[0] if isinstance(args[0], PetriNet): net, im, fm = args[0], args[1], args[2] else: net, im, fm = convert_to_petri_net(*args) from pm4py.objects.conversion.wf_net.variants import to_process_tree tree = to_process_tree.apply(net, im, fm) if tree is not None: return tree raise Exception("the object represents a model that cannot be represented as a process tree!")