linpershey/process_mining

1import logging2import os, sys3import unittest4 5from pm4py.objects.conversion.log import converter as log_conversion6from pm4py.algo.conformance.tokenreplay import algorithm as token_replay7from pm4py.algo.conformance.tokenreplay.variants.token_replay import NoConceptNameException8from pm4py.algo.discovery.inductive import algorithm as inductive_miner9from pm4py.objects import petri_net10from pm4py.objects.log.util import dataframe_utils11from pm4py.util import constants, pandas_utils12from pm4py.objects.log.importer.xes import importer as xes_importer13from pm4py.objects.log.util import sampling, sorting, index_attribute14from pm4py.objects.petri_net.exporter import exporter as petri_exporter15from pm4py.visualization.petri_net.common import visualize as pn_viz16from pm4py.objects.conversion.process_tree import converter as process_tree_converter17 18# from tests.constants import INPUT_DATA_DIR, OUTPUT_DATA_DIR, PROBLEMATIC_XES_DIR19 20INPUT_DATA_DIR = "input_data"21OUTPUT_DATA_DIR = "test_output_data"22PROBLEMATIC_XES_DIR = "xes_importer_tests"23COMPRESSED_INPUT_DATA = "compressed_input_data"24 25 26class InductiveMinerTest(unittest.TestCase):27    def obtain_petri_net_through_im(self, log_name, variant=inductive_miner.Variants.IM):28        # to avoid static method warnings in tests,29        # that by construction of the unittest package have to be expressed in such way30        self.dummy_variable = "dummy_value"31        if ".xes" in log_name:32            log = xes_importer.apply(log_name)33        else:34            df = pandas_utils.read_csv(log_name)35            df = dataframe_utils.convert_timestamp_columns_in_df(df, timest_format=constants.DEFAULT_TIMESTAMP_PARSE_FORMAT)36            log = log_conversion.apply(df, variant=log_conversion.Variants.TO_EVENT_LOG)37        process_tree = inductive_miner.apply(log)38        net, marking, final_marking = process_tree_converter.apply(process_tree)39 40        return log, net, marking, final_marking41 42    def test_applyImdfToXES(self):43        # to avoid static method warnings in tests,44        # that by construction of the unittest package have to be expressed in such way45        self.dummy_variable = "dummy_value"46        # calculate and compare Petri nets obtained on the same log to verify that instances47        # are working correctly48        log1, net1, marking1, fmarking1 = self.obtain_petri_net_through_im(49            os.path.join(INPUT_DATA_DIR, "running-example.xes"))50        log2, net2, marking2, fmarking2 = self.obtain_petri_net_through_im(51            os.path.join(INPUT_DATA_DIR, "running-example.xes"))52        log1 = sorting.sort_timestamp(log1)53        log1 = sampling.sample(log1)54        log1 = index_attribute.insert_trace_index_as_event_attribute(log1)55        log2 = sorting.sort_timestamp(log2)56        log2 = sampling.sample(log2)57        log2 = index_attribute.insert_trace_index_as_event_attribute(log2)58        petri_exporter.apply(net1, marking1, os.path.join(OUTPUT_DATA_DIR, "running-example.pnml"))59        os.remove(os.path.join(OUTPUT_DATA_DIR, "running-example.pnml"))60        self.assertEqual(len(net1.places), len(net2.places))61        final_marking = petri_net.obj.Marking()62        for p in net1.places:63            if not p.out_arcs:64                final_marking[p] = 165        aligned_traces = token_replay.apply(log1, net1, marking1, final_marking)66        del aligned_traces67 68    def test_applyImdfToCSV(self):69        # to avoid static method warnings in tests,70        # that by construction of the unittest package have to be expressed in such way71        self.dummy_variable = "dummy_value"72        # calculate and compare Petri nets obtained on the same log to verify that instances73        # are working correctly74        log1, net1, marking1, fmarking1 = self.obtain_petri_net_through_im(75            os.path.join(INPUT_DATA_DIR, "running-example.csv"))76        log2, net2, marking2, fmarking2 = self.obtain_petri_net_through_im(77            os.path.join(INPUT_DATA_DIR, "running-example.csv"))78        log1 = sorting.sort_timestamp(log1)79        log1 = sampling.sample(log1)80        log1 = index_attribute.insert_trace_index_as_event_attribute(log1)81        log2 = sorting.sort_timestamp(log2)82        log2 = sampling.sample(log2)83        log2 = index_attribute.insert_trace_index_as_event_attribute(log2)84        petri_exporter.apply(net1, marking1, os.path.join(OUTPUT_DATA_DIR, "running-example.pnml"))85        os.remove(os.path.join(OUTPUT_DATA_DIR, "running-example.pnml"))86        self.assertEqual(len(net1.places), len(net2.places))87        final_marking = petri_net.obj.Marking()88        for p in net1.places:89            if not p.out_arcs:90                final_marking[p] = 191        aligned_traces = token_replay.apply(log1, net1, marking1, final_marking)92        del aligned_traces93 94    def test_imdfVisualizationFromXES(self):95        # to avoid static method warnings in tests,96        # that by construction of the unittest package have to be expressed in such way97        self.dummy_variable = "dummy_value"98        log, net, marking, fmarking = self.obtain_petri_net_through_im(99            os.path.join(INPUT_DATA_DIR, "running-example.xes"))100        log = sorting.sort_timestamp(log)101        log = sampling.sample(log)102        log = index_attribute.insert_trace_index_as_event_attribute(log)103        petri_exporter.apply(net, marking, os.path.join(OUTPUT_DATA_DIR, "running-example.pnml"))104        os.remove(os.path.join(OUTPUT_DATA_DIR, "running-example.pnml"))105        gviz = pn_viz.graphviz_visualization(net)106        final_marking = petri_net.obj.Marking()107        for p in net.places:108            if not p.out_arcs:109                final_marking[p] = 1110        aligned_traces = token_replay.apply(log, net, marking, final_marking)111        del gviz112        del aligned_traces113 114    def test_inductive_miner_new_log(self):115        import pm4py116        log = pm4py.read_xes("input_data/running-example.xes", return_legacy_log_object=True)117        tree = pm4py.discover_process_tree_inductive(log, noise_threshold=0.2)118 119    def test_inductive_miner_new_df(self):120        import pm4py121        log = pm4py.read_xes("input_data/running-example.xes")122        tree = pm4py.discover_process_tree_inductive(log, noise_threshold=0.2)123 124    def test_inductive_miner_new_log_dfg(self):125        import pm4py126        from pm4py.objects.dfg.obj import DFG127        log = pm4py.read_xes("input_data/running-example.xes", return_legacy_log_object=True)128        dfg, sa, ea = pm4py.discover_dfg(log)129        typed_dfg = DFG(dfg, sa, ea)130        tree = pm4py.discover_process_tree_inductive(typed_dfg, noise_threshold=0.2)131 132    def test_inductive_miner_new_df_dfg(self):133        import pm4py134        log = pm4py.read_xes("input_data/running-example.xes", return_legacy_log_object=False)135        typed_dfg = pm4py.discover_dfg_typed(log)136        tree = pm4py.discover_process_tree_inductive(typed_dfg, noise_threshold=0.2)137 138    def test_inductive_miner_new_log_variants(self):139        import pm4py140        from pm4py.util.compression.dtypes import UVCL141        from pm4py.algo.discovery.inductive.variants.imf import IMFUVCL142        from pm4py.algo.discovery.inductive.dtypes.im_ds import IMDataStructureUVCL143        log = pm4py.read_xes("input_data/running-example.xes", return_legacy_log_object=True)144        variants = pm4py.get_variants(log)145        uvcl = UVCL()146        for var, occ in variants.items():147            uvcl[var] = len(occ)148        parameters = {"noise_threshold": 0.2}149        imfuvcl = IMFUVCL(parameters)150 151        tree = imfuvcl.apply(IMDataStructureUVCL(uvcl), parameters=parameters)152 153 154    def test_inductive_miner_new_df_variants(self):155        import pm4py156        from pm4py.util.compression.dtypes import UVCL157        from pm4py.algo.discovery.inductive.variants.imf import IMFUVCL158        from pm4py.algo.discovery.inductive.dtypes.im_ds import IMDataStructureUVCL159        log = pm4py.read_xes("input_data/running-example.xes", return_legacy_log_object=True)160        variants = pm4py.get_variants(log)161        uvcl = UVCL()162        for var, occ in variants.items():163            uvcl[var] = len(occ)164        parameters = {"noise_threshold": 0.2}165        imfuvcl = IMFUVCL(parameters)166 167        tree = imfuvcl.apply(IMDataStructureUVCL(uvcl), parameters=parameters)168 169 170 171if __name__ == "__main__":172    unittest.main()173