linpershey/process_mining

1import unittest2from pm4py.objects.log.util import dataframe_utils3from pm4py.objects.log.importer.xes import importer as xes_importer4from pm4py.objects.conversion.process_tree import converter as process_tree_converter5from pm4py.util import constants, pandas_utils6import os7 8 9class AlgorithmTest(unittest.TestCase):10    def test_importing_xes(self):11        from pm4py.objects.log.importer.xes import importer as xes_importer12        log = xes_importer.apply(os.path.join("input_data", "running-example.xes"),13                                 variant=xes_importer.Variants.ITERPARSE)14        log = xes_importer.apply(os.path.join("input_data", "running-example.xes"),15                                 variant=xes_importer.Variants.LINE_BY_LINE)16        log = xes_importer.apply(os.path.join("input_data", "running-example.xes"),17                                 variant=xes_importer.Variants.ITERPARSE_MEM_COMPRESSED)18        log = xes_importer.apply(os.path.join("input_data", "running-example.xes"),19                                 variant=xes_importer.Variants.CHUNK_REGEX)20 21    """def test_hiearch_clustering(self):22        from pm4py.algo.clustering.trace_attribute_driven import algorithm as clust_algorithm23        log = xes_importer.apply(os.path.join("input_data", "receipt.xes"), variant=xes_importer.Variants.LINE_BY_LINE,24                                 parameters={xes_importer.Variants.LINE_BY_LINE.value.Parameters.MAX_TRACES: 50})25        # raise Exception("%d" % (len(log)))26        clust_algorithm.apply(log, "responsible", variant=clust_algorithm.Variants.VARIANT_DMM_VEC)"""27 28    def test_log_skeleton(self):29        log = xes_importer.apply(os.path.join("input_data", "running-example.xes"))30        from pm4py.algo.discovery.log_skeleton import algorithm as lsk_discovery31        model = lsk_discovery.apply(log)32        from pm4py.algo.conformance.log_skeleton import algorithm as lsk_conformance33        conf = lsk_conformance.apply(log, model)34 35    def test_alignment(self):36        log = xes_importer.apply(os.path.join("input_data", "running-example.xes"))37        from pm4py.algo.discovery.alpha import algorithm as alpha_miner38        net, im, fm = alpha_miner.apply(log)39        from pm4py.algo.conformance.alignments.petri_net import algorithm as alignments40        aligned_traces = alignments.apply(log, net, im, fm, variant=alignments.Variants.VERSION_STATE_EQUATION_A_STAR)41        aligned_traces = alignments.apply(log, net, im, fm, variant=alignments.Variants.VERSION_DIJKSTRA_NO_HEURISTICS)42        from pm4py.algo.evaluation.replay_fitness import algorithm as rp_fitness_evaluator43        fitness = rp_fitness_evaluator.apply(log, net, im, fm, variant=rp_fitness_evaluator.Variants.ALIGNMENT_BASED)44        evaluation = rp_fitness_evaluator.evaluate(aligned_traces,45                                                   variant=rp_fitness_evaluator.Variants.ALIGNMENT_BASED)46        from pm4py.algo.evaluation.precision import algorithm as precision_evaluator47        precision = precision_evaluator.apply(log, net, im, fm, variant=rp_fitness_evaluator.Variants.ALIGNMENT_BASED)48 49    def test_decomp_alignment(self):50        log = xes_importer.apply(os.path.join("input_data", "running-example.xes"))51        from pm4py.algo.discovery.alpha import algorithm as alpha_miner52        net, im, fm = alpha_miner.apply(log)53        from pm4py.algo.conformance.alignments.decomposed import algorithm as decomp_align54        aligned_traces = decomp_align.apply(log, net, im, fm, variant=decomp_align.Variants.RECOMPOS_MAXIMAL)55 56    def test_tokenreplay(self):57        log = xes_importer.apply(os.path.join("input_data", "running-example.xes"))58        from pm4py.algo.discovery.alpha import algorithm as alpha_miner59        net, im, fm = alpha_miner.apply(log)60        from pm4py.algo.conformance.tokenreplay import algorithm as token_replay61        replayed_traces = token_replay.apply(log, net, im, fm, variant=token_replay.Variants.TOKEN_REPLAY)62        replayed_traces = token_replay.apply(log, net, im, fm, variant=token_replay.Variants.BACKWARDS)63        from pm4py.algo.evaluation.replay_fitness import algorithm as rp_fitness_evaluator64        fitness = rp_fitness_evaluator.apply(log, net, im, fm, variant=rp_fitness_evaluator.Variants.TOKEN_BASED)65        evaluation = rp_fitness_evaluator.evaluate(replayed_traces, variant=rp_fitness_evaluator.Variants.TOKEN_BASED)66        from pm4py.algo.evaluation.precision import algorithm as precision_evaluator67        precision = precision_evaluator.apply(log, net, im, fm,68                                              variant=precision_evaluator.Variants.ETCONFORMANCE_TOKEN)69        from pm4py.algo.evaluation.generalization import algorithm as generalization_evaluation70        generalization = generalization_evaluation.apply(log, net, im, fm,71                                                         variant=generalization_evaluation.Variants.GENERALIZATION_TOKEN)72 73    def test_evaluation(self):74        log = xes_importer.apply(os.path.join("input_data", "running-example.xes"))75        from pm4py.algo.discovery.alpha import algorithm as alpha_miner76        net, im, fm = alpha_miner.apply(log)77        from pm4py.algo.evaluation.simplicity import algorithm as simplicity78        simp = simplicity.apply(net)79        from pm4py.algo.evaluation import algorithm as evaluation_method80        eval = evaluation_method.apply(log, net, im, fm)81 82    def test_playout(self):83        log = xes_importer.apply(os.path.join("input_data", "running-example.xes"))84        from pm4py.algo.discovery.alpha import algorithm as alpha_miner85        net, im, fm = alpha_miner.apply(log)86        from pm4py.algo.simulation.playout.petri_net import algorithm87        log2 = algorithm.apply(net, im, fm)88 89    def test_tree_generation(self):90        from pm4py.algo.simulation.tree_generator import algorithm as tree_simulator91        tree1 = tree_simulator.apply(variant=tree_simulator.Variants.BASIC)92        tree2 = tree_simulator.apply(variant=tree_simulator.Variants.PTANDLOGGENERATOR)93 94    def test_alpha_miner_log(self):95        log = xes_importer.apply(os.path.join("input_data", "running-example.xes"))96        from pm4py.algo.discovery.alpha import algorithm as alpha_miner97        net1, im1, fm1 = alpha_miner.apply(log, variant=alpha_miner.Variants.ALPHA_VERSION_CLASSIC)98        net2, im2, fm2 = alpha_miner.apply(log, variant=alpha_miner.Variants.ALPHA_VERSION_PLUS)99        from pm4py.algo.discovery.dfg import algorithm as dfg_discovery100        dfg = dfg_discovery.apply(log)101        net3, im3, fm3 = alpha_miner.apply_dfg(dfg, variant=alpha_miner.Variants.ALPHA_VERSION_CLASSIC)102 103    def test_alpha_miner_dataframe(self):104        df = pandas_utils.read_csv(os.path.join("input_data", "running-example.csv"))105        df = dataframe_utils.convert_timestamp_columns_in_df(df, timest_format=constants.DEFAULT_TIMESTAMP_PARSE_FORMAT)106        from pm4py.algo.discovery.alpha import algorithm as alpha_miner107        net, im, fm = alpha_miner.apply(df, variant=alpha_miner.Variants.ALPHA_VERSION_CLASSIC)108 109    def test_tsystem(self):110        log = xes_importer.apply(os.path.join("input_data", "running-example.xes"))111        from pm4py.algo.discovery.transition_system import algorithm as ts_system112        tsystem = ts_system.apply(log, variant=ts_system.Variants.VIEW_BASED)113 114    def test_inductive_miner(self):115        log = xes_importer.apply(os.path.join("input_data", "running-example.xes"))116        from pm4py.algo.discovery.inductive import algorithm as inductive_miner117        process_tree = inductive_miner.apply(log)118        net, im, fm = process_tree_converter.apply(process_tree)119 120    def test_performance_spectrum(self):121        log = xes_importer.apply(os.path.join("input_data", "running-example.xes"))122        from pm4py.algo.discovery.performance_spectrum import algorithm as pspectrum123        ps = pspectrum.apply(log, ["register request", "decide"])124        df = pandas_utils.read_csv(os.path.join("input_data", "running-example.csv"))125        df = dataframe_utils.convert_timestamp_columns_in_df(df, timest_format=constants.DEFAULT_TIMESTAMP_PARSE_FORMAT)126        ps = pspectrum.apply(df, ["register request", "decide"])127 128if __name__ == "__main__":129    unittest.main()130