linpershey/process_mining

1import os2import unittest3 4from pm4py.objects.petri_net.importer import importer as pnml_importer5from pm4py.algo.simulation.playout.petri_net import algorithm as simulator6from tests.constants import INPUT_DATA_DIR7from datetime import datetime8 9 10class SimulationTest(unittest.TestCase):11    def test_simulate_petrinet(self):12        net, im, fm = pnml_importer.apply(13            os.path.join(INPUT_DATA_DIR, "running-example.pnml"))14        number_of_traces = 1015        eventlog = simulator.apply(net, im, fm, variant=simulator.Variants.BASIC_PLAYOUT,16                                   parameters={17                                       simulator.Variants.BASIC_PLAYOUT.value.Parameters.NO_TRACES: number_of_traces})18        self.assertEqual(len(eventlog), number_of_traces)19        case_id_default = 020        last_case_id = case_id_default + number_of_traces - 121        timestamp_default = 1000000022        self.assertEqual(eventlog[0].attributes['concept:name'], str(case_id_default))23        self.assertEqual(datetime.timestamp(eventlog[0][0]['time:timestamp']), timestamp_default)24        self.assertEqual(eventlog[-1].attributes['concept:name'], str(last_case_id))25 26    def test_simulate_petrinet_start_params(self):27        net, im, fm = pnml_importer.apply(28            os.path.join(INPUT_DATA_DIR, "running-example.pnml"))29        number_of_traces = 1030        timestamp = 5000000031        case_id = 532        eventlog = simulator.apply(net, im, fm, variant=simulator.Variants.BASIC_PLAYOUT,33                                   parameters={34                                       simulator.Variants.BASIC_PLAYOUT.value.Parameters.NO_TRACES: number_of_traces,35                                       simulator.Variants.BASIC_PLAYOUT.value.Parameters.INITIAL_TIMESTAMP:36                                           timestamp,37                                       simulator.Variants.BASIC_PLAYOUT.value.Parameters.INITIAL_CASE_ID: case_id})38        self.assertEqual(len(eventlog), number_of_traces)39        last_case_id = case_id + number_of_traces - 140        self.assertEqual(eventlog[0].attributes['concept:name'], str(case_id))41        self.assertEqual(datetime.timestamp(eventlog[0][0]['time:timestamp']), timestamp)42        self.assertEqual(eventlog[-1].attributes['concept:name'], str(last_case_id))43 44 45if __name__ == "__main__":46    unittest.main()47