linpershey/process_mining

1import os2import unittest3 4from pm4py.algo.conformance.alignments.petri_net import algorithm as align_alg5from pm4py.algo.conformance.tokenreplay import algorithm as tr_alg6from pm4py.algo.discovery.alpha import algorithm as alpha_miner7from pm4py.algo.discovery.dfg import algorithm as dfg_mining8from pm4py.algo.discovery.heuristics import algorithm as heuristics_miner9from pm4py.algo.discovery.inductive import algorithm as inductive_miner10from pm4py.algo.discovery.transition_system import algorithm as ts_disc11from pm4py.algo.evaluation import algorithm as eval_alg12from pm4py.algo.evaluation.generalization import algorithm as generalization13from pm4py.algo.evaluation.precision import algorithm as precision_evaluator14from pm4py.algo.evaluation.replay_fitness import algorithm as rp_fit15from pm4py.algo.evaluation.simplicity import algorithm as simplicity16from pm4py.objects.conversion.log import converter as log_conversion17from pm4py.objects.log.exporter.xes import exporter as xes_exporter18from pm4py.objects.log.importer.xes import importer as xes_importer19from pm4py.objects.log.util import dataframe_utils20from pm4py.util import constants, pandas_utils21from pm4py.objects.conversion.process_tree import converter as process_tree_converter22 23 24class MainFactoriesTest(unittest.TestCase):25    def test_nonstandard_exporter(self):26        log = xes_importer.apply(os.path.join("input_data", "running-example.xes"))27        xes_exporter.apply(log, os.path.join("test_output_data", "running-example.xes"),28                           variant=xes_exporter.Variants.LINE_BY_LINE)29        os.remove(os.path.join("test_output_data", "running-example.xes"))30 31    def test_alphaminer_log(self):32        log = xes_importer.apply(os.path.join("input_data", "running-example.xes"))33        net, im, fm = alpha_miner.apply(log)34        aligned_traces_tr = tr_alg.apply(log, net, im, fm)35        aligned_traces_alignments = align_alg.apply(log, net, im, fm)36        evaluation = eval_alg.apply(log, net, im, fm)37        fitness = rp_fit.apply(log, net, im, fm)38        precision = precision_evaluator.apply(log, net, im, fm)39        gen = generalization.apply(log, net, im, fm)40        sim = simplicity.apply(net)41 42    def test_memory_efficient_iterparse(self):43        log = xes_importer.apply(os.path.join("input_data", "running-example.xes"),44                                 variant=xes_importer.Variants.ITERPARSE_MEM_COMPRESSED)45 46    def test_alphaminer_stream(self):47        df = pandas_utils.read_csv(os.path.join("input_data", "running-example.csv"))48        df = dataframe_utils.convert_timestamp_columns_in_df(df, timest_format=constants.DEFAULT_TIMESTAMP_PARSE_FORMAT)49        stream = log_conversion.apply(df, variant=log_conversion.TO_EVENT_STREAM)50        net, im, fm = alpha_miner.apply(stream)51        aligned_traces_tr = tr_alg.apply(stream, net, im, fm)52        aligned_traces_alignments = align_alg.apply(stream, net, im, fm)53        evaluation = eval_alg.apply(stream, net, im, fm)54        fitness = rp_fit.apply(stream, net, im, fm)55        precision = precision_evaluator.apply(stream, net, im, fm)56        gen = generalization.apply(stream, net, im, fm)57        sim = simplicity.apply(net)58 59    def test_alphaminer_df(self):60        log = pandas_utils.read_csv(os.path.join("input_data", "running-example.csv"))61        log = dataframe_utils.convert_timestamp_columns_in_df(log, timest_format=constants.DEFAULT_TIMESTAMP_PARSE_FORMAT)62        net, im, fm = alpha_miner.apply(log)63        aligned_traces_tr = tr_alg.apply(log, net, im, fm)64        aligned_traces_alignments = align_alg.apply(log, net, im, fm)65        evaluation = eval_alg.apply(log, net, im, fm)66        fitness = rp_fit.apply(log, net, im, fm)67        precision = precision_evaluator.apply(log, net, im, fm)68        gen = generalization.apply(log, net, im, fm)69        sim = simplicity.apply(net)70 71    def test_inductiveminer_log(self):72        log = xes_importer.apply(os.path.join("input_data", "running-example.xes"))73        process_tree = inductive_miner.apply(log)74        net, im, fm = process_tree_converter.apply(process_tree)75        aligned_traces_tr = tr_alg.apply(log, net, im, fm)76        aligned_traces_alignments = align_alg.apply(log, net, im, fm)77        evaluation = eval_alg.apply(log, net, im, fm)78        fitness = rp_fit.apply(log, net, im, fm)79        precision = precision_evaluator.apply(log, net, im, fm)80        gen = generalization.apply(log, net, im, fm)81        sim = simplicity.apply(net)82 83    def test_inductiveminer_df(self):84        log = pandas_utils.read_csv(os.path.join("input_data", "running-example.csv"))85        log = dataframe_utils.convert_timestamp_columns_in_df(log, timest_format=constants.DEFAULT_TIMESTAMP_PARSE_FORMAT)86        process_tree = inductive_miner.apply(log)87        net, im, fm = process_tree_converter.apply(process_tree)88        aligned_traces_tr = tr_alg.apply(log, net, im, fm)89        aligned_traces_alignments = align_alg.apply(log, net, im, fm)90        evaluation = eval_alg.apply(log, net, im, fm)91        fitness = rp_fit.apply(log, net, im, fm)92        precision = precision_evaluator.apply(log, net, im, fm)93        gen = generalization.apply(log, net, im, fm)94        sim = simplicity.apply(net)95 96    def test_heu_log(self):97        log = xes_importer.apply(os.path.join("input_data", "running-example.xes"))98        net, im, fm = heuristics_miner.apply(log)99        aligned_traces_tr = tr_alg.apply(log, net, im, fm)100        aligned_traces_alignments = align_alg.apply(log, net, im, fm)101        evaluation = eval_alg.apply(log, net, im, fm)102        fitness = rp_fit.apply(log, net, im, fm)103        precision = precision_evaluator.apply(log, net, im, fm)104        gen = generalization.apply(log, net, im, fm)105        sim = simplicity.apply(net)106 107    def test_heu_stream(self):108        df = pandas_utils.read_csv(os.path.join("input_data", "running-example.csv"))109        df = dataframe_utils.convert_timestamp_columns_in_df(df, timest_format=constants.DEFAULT_TIMESTAMP_PARSE_FORMAT)110        stream = log_conversion.apply(df, variant=log_conversion.TO_EVENT_STREAM)111        net, im, fm = heuristics_miner.apply(stream)112        aligned_traces_tr = tr_alg.apply(stream, net, im, fm)113        aligned_traces_alignments = align_alg.apply(stream, net, im, fm)114        evaluation = eval_alg.apply(stream, net, im, fm)115        fitness = rp_fit.apply(stream, net, im, fm)116        precision = precision_evaluator.apply(stream, net, im, fm)117        gen = generalization.apply(stream, net, im, fm)118        sim = simplicity.apply(net)119 120    def test_heu_df(self):121        log = pandas_utils.read_csv(os.path.join("input_data", "running-example.csv"))122        log = dataframe_utils.convert_timestamp_columns_in_df(log, timest_format=constants.DEFAULT_TIMESTAMP_PARSE_FORMAT)123        net, im, fm = heuristics_miner.apply(log)124        aligned_traces_tr = tr_alg.apply(log, net, im, fm)125        aligned_traces_alignments = align_alg.apply(log, net, im, fm)126        evaluation = eval_alg.apply(log, net, im, fm)127        fitness = rp_fit.apply(log, net, im, fm)128        precision = precision_evaluator.apply(log, net, im, fm)129        gen = generalization.apply(log, net, im, fm)130        sim = simplicity.apply(net)131 132    def test_dfg_log(self):133        log = xes_importer.apply(os.path.join("input_data", "running-example.xes"))134        dfg = dfg_mining.apply(log)135 136    def test_dfg_stream(self):137        df = pandas_utils.read_csv(os.path.join("input_data", "running-example.csv"))138        df = dataframe_utils.convert_timestamp_columns_in_df(df, timest_format=constants.DEFAULT_TIMESTAMP_PARSE_FORMAT)139        stream = log_conversion.apply(df, variant=log_conversion.TO_EVENT_STREAM)140        dfg = dfg_mining.apply(stream)141 142    def test_dfg_df(self):143        df = pandas_utils.read_csv(os.path.join("input_data", "running-example.csv"))144        df = dataframe_utils.convert_timestamp_columns_in_df(df, timest_format=constants.DEFAULT_TIMESTAMP_PARSE_FORMAT)145        dfg = dfg_mining.apply(df)146 147    def test_ts_log(self):148        log = xes_importer.apply(os.path.join("input_data", "running-example.xes"))149        ts = ts_disc.apply(log)150 151    def test_ts_stream(self):152        df = pandas_utils.read_csv(os.path.join("input_data", "running-example.csv"))153        df = dataframe_utils.convert_timestamp_columns_in_df(df, timest_format=constants.DEFAULT_TIMESTAMP_PARSE_FORMAT)154        stream = log_conversion.apply(df, variant=log_conversion.TO_EVENT_STREAM)155        ts = ts_disc.apply(stream)156 157    def test_ts_df(self):158        df = pandas_utils.read_csv(os.path.join("input_data", "running-example.csv"))159        df = dataframe_utils.convert_timestamp_columns_in_df(df, timest_format=constants.DEFAULT_TIMESTAMP_PARSE_FORMAT)160        ts = ts_disc.apply(df)161 162    def test_csvimp_xesexp(self):163        df = pandas_utils.read_csv(os.path.join("input_data", "running-example.csv"))164        df = dataframe_utils.convert_timestamp_columns_in_df(df, timest_format=constants.DEFAULT_TIMESTAMP_PARSE_FORMAT)165        log0 = log_conversion.apply(df, variant=log_conversion.TO_EVENT_STREAM)166        log = log_conversion.apply(log0, variant=log_conversion.TO_EVENT_LOG)167        stream = log_conversion.apply(log0, variant=log_conversion.TO_EVENT_STREAM)168        df = log_conversion.apply(log0, variant=log_conversion.TO_DATA_FRAME)169        xes_exporter.apply(log, "ru.xes")170        xes_exporter.apply(stream, "ru.xes")171        xes_exporter.apply(df, "ru.xes")172        os.remove('ru.xes')173 174    def test_xesimp_xesexp(self):175        log0 = xes_importer.apply(os.path.join("input_data", "running-example.xes"))176        log = log_conversion.apply(log0, variant=log_conversion.TO_EVENT_LOG)177        stream = log_conversion.apply(log0, variant=log_conversion.TO_EVENT_STREAM)178        df = log_conversion.apply(log0, variant=log_conversion.TO_DATA_FRAME)179        xes_exporter.apply(log, "ru.xes")180        xes_exporter.apply(stream, "ru.xes")181        xes_exporter.apply(df, "ru.xes")182        os.remove('ru.xes')183 184    def test_pdimp_xesexp(self):185        log0 = pandas_utils.read_csv(os.path.join("input_data", "running-example.csv"))186        log0 = dataframe_utils.convert_timestamp_columns_in_df(log0, timest_format=constants.DEFAULT_TIMESTAMP_PARSE_FORMAT)187        log = log_conversion.apply(log0, variant=log_conversion.TO_EVENT_LOG)188        stream = log_conversion.apply(log0, variant=log_conversion.TO_EVENT_STREAM)189        df = log_conversion.apply(log0, variant=log_conversion.TO_DATA_FRAME)190        xes_exporter.apply(log, "ru.xes")191        xes_exporter.apply(stream, "ru.xes")192        xes_exporter.apply(df, "ru.xes")193        os.remove('ru.xes')194 195 196if __name__ == "__main__":197    unittest.main()198