linpershey/process_mining
0
1import importlib.util2import os3import unittest4 5import pm4py6from pm4py.objects.bpmn.obj import BPMN7from pm4py.objects.petri_net.obj import PetriNet8from pm4py.objects.process_tree.obj import ProcessTree9from pm4py.util import constants, pandas_utils10from pm4py.objects.log.util import dataframe_utils11from pm4py.objects.log.importer.xes import importer as xes_importer12 13 14class SimplifiedInterfaceTest(unittest.TestCase):15 def test_csv(self):16 df = pandas_utils.read_csv("input_data/running-example.csv")17 df = dataframe_utils.convert_timestamp_columns_in_df(df, timest_format=constants.DEFAULT_TIMESTAMP_PARSE_FORMAT, timest_columns=["time:timestamp"])18 df["case:concept:name"] = df["case:concept:name"].astype("string")19 20 log2 = pm4py.convert_to_event_log(df)21 stream1 = pm4py.convert_to_event_stream(log2)22 df2 = pm4py.convert_to_dataframe(log2)23 pm4py.write_xes(log2, "test_output_data/log.xes")24 os.remove("test_output_data/log.xes")25 26 def test_alpha_miner(self):27 for legacy_obj in [True, False]:28 log = pm4py.read_xes("input_data/running-example.xes", return_legacy_log_object=legacy_obj)29 net, im, fm = pm4py.discover_petri_net_alpha(log)30 31 def test_alpha_miner_plus(self):32 for legacy_obj in [True, False]:33 log = pm4py.read_xes("input_data/running-example.xes", return_legacy_log_object=legacy_obj)34 net, im, fm = pm4py.discover_petri_net_alpha_plus(log)35 36 def test_inductive_miner(self):37 for legacy_obj in [True, False]:38 log = pm4py.read_xes("input_data/running-example.xes", return_legacy_log_object=legacy_obj)39 net, im, fm = pm4py.discover_petri_net_inductive(log)40 41 def test_inductive_miner_noise(self):42 for legacy_obj in [True, False]:43 log = pm4py.read_xes("input_data/running-example.xes", return_legacy_log_object=legacy_obj)44 net, im, fm = pm4py.discover_petri_net_inductive(log, noise_threshold=0.5)45 46 def test_heuristics_miner(self):47 for legacy_obj in [True, False]:48 log = pm4py.read_xes("input_data/running-example.xes", return_legacy_log_object=legacy_obj)49 net, im, fm = pm4py.discover_petri_net_heuristics(log)50 51 def test_inductive_miner_tree(self):52 for legacy_obj in [True, False]:53 log = pm4py.read_xes("input_data/running-example.xes", return_legacy_log_object=legacy_obj)54 tree = pm4py.discover_process_tree_inductive(log)55 tree = pm4py.discover_process_tree_inductive(log, noise_threshold=0.2)56 57 def test_heuristics_miner_heu_net(self):58 for legacy_obj in [True, False]:59 log = pm4py.read_xes("input_data/running-example.xes", return_legacy_log_object=legacy_obj)60 heu_net = pm4py.discover_heuristics_net(log)61 62 def test_dfg(self):63 for legacy_obj in [True, False]:64 log = pm4py.read_xes("input_data/running-example.xes", return_legacy_log_object=legacy_obj)65 dfg, sa, ea = pm4py.discover_directly_follows_graph(log)66 67 def test_read_petri(self):68 net, im, fm = pm4py.read_pnml("input_data/running-example.pnml")69 70 def test_read_tree(self):71 tree = pm4py.read_ptml("input_data/running-example.ptml")72 73 def test_read_dfg(self):74 dfg, sa, ea = pm4py.read_dfg("input_data/running-example.dfg")75 76 def test_alignments_simpl_interface(self):77 for legacy_obj in [True, False]:78 for diagn_df in [True, False]:79 log = pm4py.read_xes("input_data/running-example.xes", return_legacy_log_object=legacy_obj)80 net, im, fm = pm4py.discover_petri_net_inductive(log)81 aligned_traces = pm4py.conformance_diagnostics_alignments(log, net, im, fm, return_diagnostics_dataframe=diagn_df)82 83 def test_tbr_simpl_interface(self):84 for legacy_obj in [True, False]:85 for diagn_df in [True, False]:86 log = pm4py.read_xes("input_data/running-example.xes", return_legacy_log_object=legacy_obj)87 net, im, fm = pm4py.discover_petri_net_inductive(log)88 replayed_traces = pm4py.conformance_diagnostics_token_based_replay(log, net, im, fm, return_diagnostics_dataframe=diagn_df)89 90 def test_fitness_alignments(self):91 for legacy_obj in [True, False]:92 log = pm4py.read_xes("input_data/running-example.xes", return_legacy_log_object=legacy_obj)93 net, im, fm = pm4py.discover_petri_net_inductive(log)94 fitness_ali = pm4py.fitness_alignments(log, net, im, fm)95 96 def test_fitness_tbr(self):97 for legacy_obj in [True, False]:98 log = pm4py.read_xes("input_data/running-example.xes", return_legacy_log_object=legacy_obj)99 net, im, fm = pm4py.discover_petri_net_inductive(log)100 fitness_tbr = pm4py.fitness_token_based_replay(log, net, im, fm)101 102 def test_precision_alignments(self):103 for legacy_obj in [True, False]:104 log = pm4py.read_xes("input_data/running-example.xes", return_legacy_log_object=legacy_obj)105 net, im, fm = pm4py.discover_petri_net_inductive(log)106 precision_ali = pm4py.precision_alignments(log, net, im, fm)107 108 def test_precision_tbr(self):109 for legacy_obj in [True, False]:110 log = pm4py.read_xes("input_data/running-example.xes", return_legacy_log_object=legacy_obj)111 net, im, fm = pm4py.discover_petri_net_inductive(log)112 precision_tbr = pm4py.precision_token_based_replay(log, net, im, fm)113 114 def test_convert_to_tree_from_petri(self):115 net, im, fm = pm4py.read_pnml("input_data/running-example.pnml")116 tree = pm4py.convert_to_process_tree(net, im, fm)117 self.assertTrue(isinstance(tree, ProcessTree))118 119 def test_convert_to_tree_from_bpmn(self):120 bpmn = pm4py.read_bpmn("input_data/running-example.bpmn")121 tree = pm4py.convert_to_process_tree(bpmn)122 self.assertTrue(isinstance(tree, ProcessTree))123 124 def test_convert_to_net_from_tree(self):125 tree = pm4py.read_ptml("input_data/running-example.ptml")126 net, im, fm = pm4py.convert_to_petri_net(tree)127 self.assertTrue(isinstance(net, PetriNet))128 129 def test_convert_to_net_from_bpmn(self):130 bpmn = pm4py.read_bpmn("input_data/running-example.bpmn")131 net, im, fm = pm4py.convert_to_petri_net(bpmn)132 self.assertTrue(isinstance(net, PetriNet))133 134 def test_convert_to_net_from_dfg(self):135 dfg, sa, ea = pm4py.read_dfg("input_data/running-example.dfg")136 net, im, fm = pm4py.convert_to_petri_net(dfg, sa, ea)137 self.assertTrue(isinstance(net, PetriNet))138 139 def test_convert_to_net_from_heu(self):140 for legacy_obj in [True, False]:141 log = pm4py.read_xes("input_data/running-example.xes", return_legacy_log_object=legacy_obj)142 heu_net = pm4py.discover_heuristics_net(log)143 net, im, fm = pm4py.convert_to_petri_net(heu_net)144 self.assertTrue(isinstance(net, PetriNet))145 146 def test_convert_to_bpmn_from_tree(self):147 tree = pm4py.read_ptml("input_data/running-example.ptml")148 bpmn = pm4py.convert_to_bpmn(tree)149 self.assertTrue(isinstance(bpmn, BPMN))150 151 def test_statistics_log(self):152 for legacy_obj in [True, False]:153 log = pm4py.read_xes("input_data/running-example.xes", return_legacy_log_object=legacy_obj)154 pm4py.get_start_activities(log)155 pm4py.get_end_activities(log)156 pm4py.get_event_attributes(log)157 pm4py.get_trace_attributes(log)158 pm4py.get_event_attribute_values(log, "org:resource")159 pm4py.get_variants_as_tuples(log)160 161 def test_statistics_df(self):162 df = pandas_utils.read_csv("input_data/running-example-transformed.csv")163 df = dataframe_utils.convert_timestamp_columns_in_df(df, timest_format=constants.DEFAULT_TIMESTAMP_PARSE_FORMAT, timest_columns=["Timestamp"])164 df["CaseID"] = df["CaseID"].astype("string")165 166 pm4py.get_start_activities(df, case_id_key="CaseID", activity_key="Activity", timestamp_key="Timestamp")167 pm4py.get_end_activities(df, case_id_key="CaseID", activity_key="Activity", timestamp_key="Timestamp")168 pm4py.get_event_attributes(df)169 pm4py.get_event_attribute_values(df, "Resource", case_id_key="CaseID")170 pm4py.get_variants_as_tuples(df, case_id_key="CaseID", activity_key="Activity", timestamp_key="Timestamp")171 172 def test_playout(self):173 net, im, fm = pm4py.read_pnml("input_data/running-example.pnml")174 pm4py.play_out(net, im, fm)175 176 def test_generator(self):177 pm4py.generate_process_tree()178 179 def test_mark_em_equation(self):180 for legacy_obj in [True, False]:181 log = xes_importer.apply("input_data/running-example.xes")182 net, im, fm = pm4py.read_pnml("input_data/running-example.pnml")183 sync_net, sync_im, sync_fm = pm4py.construct_synchronous_product_net(log[0], net, im, fm)184 m_h = pm4py.solve_marking_equation(sync_net, sync_im, sync_fm)185 em_h = pm4py.solve_extended_marking_equation(log[0], sync_net, sync_im, sync_fm)186 187 def test_new_statistics_log(self):188 for legacy_obj in [True, False]:189 log = pm4py.read_xes("input_data/running-example.xes", return_legacy_log_object=legacy_obj)190 pm4py.get_trace_attribute_values(log, "case:creator")191 pm4py.discover_eventually_follows_graph(log)192 pm4py.get_case_arrival_average(log)193 194 def test_new_statistics_df(self):195 df = pandas_utils.read_csv("input_data/running-example-transformed.csv")196 df = dataframe_utils.convert_timestamp_columns_in_df(df, timest_format=constants.DEFAULT_TIMESTAMP_PARSE_FORMAT,197 timest_columns=["Timestamp"])198 df["CaseID"] = df["CaseID"].astype("string")199 200 pm4py.discover_eventually_follows_graph(df, case_id_key="CaseID", activity_key="Activity", timestamp_key="Timestamp")201 pm4py.get_case_arrival_average(df, case_id_key="CaseID", activity_key="Activity", timestamp_key="Timestamp")202 203 def test_serialization_log(self):204 if importlib.util.find_spec("pyarrow"):205 for legacy_obj in [True, False]:206 log = pm4py.read_xes("input_data/running-example.xes", return_legacy_log_object=legacy_obj)207 ser = pm4py.serialize(log)208 log2 = pm4py.deserialize(ser)209 210 def test_serialization_dataframe(self):211 if importlib.util.find_spec("pyarrow"):212 df = pandas_utils.read_csv("input_data/running-example.csv")213 df = dataframe_utils.convert_timestamp_columns_in_df(df, timest_format=constants.DEFAULT_TIMESTAMP_PARSE_FORMAT, timest_columns=["time:timestamp"])214 ser = pm4py.serialize(df)215 df2 = pm4py.deserialize(ser)216 217 def test_serialization_petri_net(self):218 net, im, fm = pm4py.read_pnml("input_data/running-example.pnml")219 ser = pm4py.serialize(net, im, fm)220 net2, im2, fm2 = pm4py.deserialize(ser)221 222 def test_serialization_process_tree(self):223 tree = pm4py.read_ptml("input_data/running-example.ptml")224 ser = pm4py.serialize(tree)225 tree2 = pm4py.deserialize(ser)226 227 def test_serialization_bpmn(self):228 bpmn = pm4py.read_bpmn("input_data/running-example.bpmn")229 ser = pm4py.serialize(bpmn)230 bpmn2 = pm4py.deserialize(ser)231 232 def test_serialization_dfg(self):233 dfg, sa, ea = pm4py.read_dfg("input_data/running-example.dfg")234 ser = pm4py.serialize(dfg, sa, ea)235 dfg2, sa2, ea2 = pm4py.deserialize(ser)236 237 def test_minimum_self_distance(self):238 import pm4py239 for legacy_obj in [True, False]:240 log = pm4py.read_xes(os.path.join("input_data", "running-example.xes"), return_legacy_log_object=legacy_obj)241 msd = pm4py.get_minimum_self_distances(log)242 243 def test_minimum_self_distance_2(self):244 import pm4py245 for legacy_obj in [True, False]:246 log = pm4py.read_xes(os.path.join("input_data", "running-example.xes"), return_legacy_log_object=legacy_obj)247 msd = pm4py.get_minimum_self_distance_witnesses(log)248 249 def test_marking_equation_net(self):250 import pm4py251 log = xes_importer.apply(os.path.join("input_data", "running-example.xes"))252 net, im, fm = pm4py.discover_petri_net_inductive(log)253 pm4py.solve_marking_equation(net, im, fm)254 255 def test_marking_equation_sync_net(self):256 import pm4py257 log = xes_importer.apply(os.path.join("input_data", "running-example.xes"))258 net, im, fm = pm4py.discover_petri_net_inductive(log)259 sync_net, sync_im, sync_fm = pm4py.construct_synchronous_product_net(log[0], net, im, fm)260 res = pm4py.solve_marking_equation(sync_net, sync_im, sync_fm)261 self.assertIsNotNone(res)262 self.assertEqual(res, 11)263 264 def test_ext_marking_equation_sync_net(self):265 import pm4py266 log = xes_importer.apply(os.path.join("input_data", "running-example.xes"))267 net, im, fm = pm4py.discover_petri_net_inductive(log)268 sync_net, sync_im, sync_fm = pm4py.construct_synchronous_product_net(log[0], net, im, fm)269 res = pm4py.solve_extended_marking_equation(log[0], sync_net, sync_im, sync_fm)270 self.assertIsNotNone(res)271 272 def test_alignments_tree_simpl_interface(self):273 import pm4py274 for legacy_obj in [True, False]:275 for diagn_df in [True, False]:276 log = pm4py.read_xes(os.path.join("input_data", "running-example.xes"), return_legacy_log_object=legacy_obj)277 tree = pm4py.read_ptml(os.path.join("input_data", "running-example.ptml"))278 res = pm4py.conformance_diagnostics_alignments(log, tree, return_diagnostics_dataframe=diagn_df)279 self.assertIsNotNone(res)280 281 def test_alignments_dfg_simpl_interface(self):282 import pm4py283 for legacy_obj in [True, False]:284 for diagn_df in [True, False]:285 log = pm4py.read_xes(os.path.join("input_data", "running-example.xes"), return_legacy_log_object=legacy_obj)286 dfg, sa, ea = pm4py.read_dfg(os.path.join("input_data", "running-example.dfg"))287 res = pm4py.conformance_diagnostics_alignments(log, dfg, sa, ea, return_diagnostics_dataframe=diagn_df)288 self.assertIsNotNone(res)289 290 def test_alignments_bpmn_simpl_interface(self):291 import pm4py292 for legacy_obj in [True, False]:293 for diagn_df in [True, False]:294 log = pm4py.read_xes(os.path.join("input_data", "running-example.xes"), return_legacy_log_object=legacy_obj)295 bpmn_graph = pm4py.read_bpmn(os.path.join("input_data", "running-example.bpmn"))296 res = pm4py.conformance_diagnostics_alignments(log, bpmn_graph, return_diagnostics_dataframe=diagn_df)297 self.assertIsNotNone(res)298 299 def test_discovery_inductive_bpmn(self):300 import pm4py301 for legacy_obj in [True, False]:302 log = pm4py.read_xes(os.path.join("input_data", "running-example.xes"), return_legacy_log_object=legacy_obj)303 bpmn_graph = pm4py.discover_bpmn_inductive(log)304 self.assertIsNotNone(bpmn_graph)305 306 def test_generation(self):307 import pm4py308 tree = pm4py.generate_process_tree()309 self.assertIsNotNone(tree)310 311 def test_play_out_tree(self):312 import pm4py313 tree = pm4py.read_ptml(os.path.join("input_data", "running-example.ptml"))314 log = pm4py.play_out(tree)315 316 def test_play_out_net(self):317 import pm4py318 net, im, fm = pm4py.read_pnml(os.path.join("input_data", "running-example.pnml"))319 log = pm4py.play_out(net, im, fm)320 321 def test_msd(self):322 import pm4py323 for legacy_obj in [True, False]:324 log = pm4py.read_xes(os.path.join("input_data", "running-example.xes"), return_legacy_log_object=legacy_obj)325 res1 = pm4py.get_minimum_self_distance_witnesses(log)326 res2 = pm4py.get_minimum_self_distances(log)327 self.assertIsNotNone(res1)328 self.assertIsNotNone(res2)329 330 def test_case_arrival(self):331 import pm4py332 for legacy_obj in [True, False]:333 log = pm4py.read_xes(os.path.join("input_data", "running-example.xes"), return_legacy_log_object=legacy_obj)334 avg = pm4py.get_case_arrival_average(log)335 self.assertIsNotNone(avg)336 337 def test_efg(self):338 for legacy_obj in [True, False]:339 log = pm4py.read_xes("input_data/running-example.xes", return_legacy_log_object=legacy_obj)340 pm4py.discover_eventually_follows_graph(log)341 342 def test_write_pnml(self):343 net, im, fm = pm4py.read_pnml("input_data/running-example.pnml")344 pm4py.write_pnml(net, im, fm, "test_output_data/running-example.pnml")345 os.remove("test_output_data/running-example.pnml")346 347 def test_write_ptml(self):348 process_tree = pm4py.read_ptml("input_data/running-example.ptml")349 pm4py.write_ptml(process_tree, "test_output_data/running-example.ptml")350 os.remove("test_output_data/running-example.ptml")351 352 def test_write_dfg(self):353 dfg, sa, ea = pm4py.read_dfg("input_data/running-example.dfg")354 pm4py.write_dfg(dfg, sa, ea, "test_output_data/running-example.dfg")355 os.remove("test_output_data/running-example.dfg")356 357 def test_write_bpmn(self):358 bpmn_graph = pm4py.read_bpmn("input_data/running-example.bpmn")359 pm4py.write_bpmn(bpmn_graph, "test_output_data/running-example.bpmn")360 os.remove("test_output_data/running-example.bpmn")361 362 def test_rebase(self):363 dataframe = pandas_utils.read_csv("input_data/running-example-transformed.csv")364 df = dataframe_utils.convert_timestamp_columns_in_df(dataframe, timest_format=constants.DEFAULT_TIMESTAMP_PARSE_FORMAT, timest_columns=["Timestamp"])365 dataframe["CaseID"] = dataframe["CaseID"].astype("string")366 367 dataframe = pm4py.rebase(dataframe, activity_key="Activity", case_id="CaseID", timestamp_key="Timestamp", timest_format=constants.DEFAULT_TIMESTAMP_PARSE_FORMAT)368 369 def test_parse_process_tree(self):370 tree = pm4py.parse_process_tree("-> ( 'a', X ( 'b', 'c' ), tau )")371 372 def test_parse_log_string(self):373 elog = pm4py.parse_event_log_string(["A,B,C", "A,B,D"])374 375 def test_project_eattr(self):376 for legacy_obj in [True, False]:377 log = pm4py.read_xes("input_data/running-example.xes", return_legacy_log_object=legacy_obj)378 lst = pm4py.project_on_event_attribute(log, "org:resource")379 380 def test_sample_cases_log(self):381 for legacy_obj in [True, False]:382 log = pm4py.read_xes("input_data/running-example.xes", return_legacy_log_object=legacy_obj)383 pm4py.sample_cases(log, 2)384 385 def test_sample_cases_df(self):386 dataframe = pandas_utils.read_csv("input_data/running-example-transformed.csv")387 dataframe = dataframe_utils.convert_timestamp_columns_in_df(dataframe, timest_format=constants.DEFAULT_TIMESTAMP_PARSE_FORMAT, timest_columns=["Timestamp"])388 dataframe["CaseID"] = dataframe["CaseID"].astype("string")389 390 pm4py.sample_cases(dataframe, 2, case_id_key="CaseID")391 392 def test_sample_events_log(self):393 for legacy_obj in [True, False]:394 log = pm4py.read_xes("input_data/running-example.xes", return_legacy_log_object=legacy_obj)395 pm4py.sample_events(log, 2)396 397 def test_sample_events_df(self):398 dataframe = pandas_utils.read_csv("input_data/running-example-transformed.csv")399 dataframe = dataframe_utils.convert_timestamp_columns_in_df(dataframe, timest_format=constants.DEFAULT_TIMESTAMP_PARSE_FORMAT, timest_columns=["Timestamp"])400 dataframe["CaseID"] = dataframe["CaseID"].astype("string")401 402 pm4py.sample_events(dataframe, 2)403 404 def test_check_soundness(self):405 net, im, fm = pm4py.read_pnml("input_data/running-example.pnml")406 self.assertTrue(pm4py.check_soundness(net, im, fm))407 408 def test_check_wfnet(self):409 net, im, fm = pm4py.read_pnml("input_data/running-example.pnml")410 self.assertTrue(pm4py.check_is_workflow_net(net))411 412 def test_artificial_start_end_log(self):413 for legacy_obj in [True, False]:414 log = pm4py.read_xes("input_data/running-example.xes", return_legacy_log_object=legacy_obj)415 pm4py.insert_artificial_start_end(log)416 417 def test_artificial_start_end_dataframe(self):418 dataframe = pandas_utils.read_csv("input_data/running-example-transformed.csv")419 dataframe = dataframe_utils.convert_timestamp_columns_in_df(dataframe, timest_format=constants.DEFAULT_TIMESTAMP_PARSE_FORMAT, timest_columns=["Timestamp"])420 dataframe["CaseID"] = dataframe["CaseID"].astype("string")421 422 pm4py.insert_artificial_start_end(dataframe, activity_key="Activity", timestamp_key="Timestamp", case_id_key="CaseID")423 424 def test_hof_filter_log(self):425 log = xes_importer.apply("input_data/running-example.xes")426 pm4py.filter_log(log, lambda x: len(x) > 5)427 428 def test_hof_filter_trace(self):429 log = xes_importer.apply("input_data/running-example.xes")430 pm4py.filter_trace(log[0], lambda x: x["concept:name"] == "decide")431 432 def test_hof_sort_log(self):433 log = xes_importer.apply("input_data/running-example.xes")434 pm4py.sort_log(log, key=lambda x: x.attributes["concept:name"])435 436 def test_hof_sort_trace(self):437 log = xes_importer.apply("input_data/running-example.xes")438 pm4py.sort_trace(log[0], key=lambda x: x["concept:name"])439 440 def test_split_train_test_log(self):441 for legacy_obj in [True, False]:442 log = pm4py.read_xes("input_data/running-example.xes", return_legacy_log_object=legacy_obj)443 pm4py.split_train_test(log, train_percentage=0.6)444 445 def test_split_train_test_df(self):446 dataframe = pandas_utils.read_csv("input_data/running-example-transformed.csv")447 dataframe = dataframe_utils.convert_timestamp_columns_in_df(dataframe, timest_format=constants.DEFAULT_TIMESTAMP_PARSE_FORMAT, timest_columns=["Timestamp"])448 dataframe["CaseID"] = dataframe["CaseID"].astype("string")449 450 pm4py.split_train_test(dataframe, train_percentage=0.6, case_id_key="CaseID")451 452 def test_get_prefixes_log(self):453 for legacy_obj in [True, False]:454 log = pm4py.read_xes("input_data/running-example.xes", return_legacy_log_object=legacy_obj)455 pm4py.get_prefixes_from_log(log, 3)456 457 def test_get_prefixes_df(self):458 dataframe = pandas_utils.read_csv("input_data/running-example-transformed.csv")459 dataframe = dataframe_utils.convert_timestamp_columns_in_df(dataframe, timest_format=constants.DEFAULT_TIMESTAMP_PARSE_FORMAT, timest_columns=["Timestamp"])460 dataframe["CaseID"] = dataframe["CaseID"].astype("string")461 462 pm4py.get_prefixes_from_log(dataframe, 3, case_id_key="CaseID")463 464 def test_convert_reachab(self):465 net, im, fm = pm4py.read_pnml("input_data/running-example.pnml")466 ts = pm4py.convert_to_reachability_graph(net, im, fm)467 468 def test_hw_log(self):469 for legacy_obj in [True, False]:470 log = pm4py.read_xes("input_data/running-example.xes", return_legacy_log_object=legacy_obj)471 pm4py.discover_handover_of_work_network(log)472 473 def test_hw_df(self):474 dataframe = pandas_utils.read_csv("input_data/running-example-transformed.csv")475 dataframe = dataframe_utils.convert_timestamp_columns_in_df(dataframe, timest_format=constants.DEFAULT_TIMESTAMP_PARSE_FORMAT, timest_columns=["Timestamp"])476 dataframe["CaseID"] = dataframe["CaseID"].astype("string")477 478 pm4py.discover_handover_of_work_network(dataframe, resource_key="Resource", case_id_key="CaseID", timestamp_key="Timestamp")479 480 def test_wt_log(self):481 for legacy_obj in [True, False]:482 log = pm4py.read_xes("input_data/running-example.xes", return_legacy_log_object=legacy_obj)483 pm4py.discover_working_together_network(log)484 485 def test_wt_df(self):486 dataframe = pandas_utils.read_csv("input_data/running-example-transformed.csv")487 dataframe = dataframe_utils.convert_timestamp_columns_in_df(dataframe, timest_format=constants.DEFAULT_TIMESTAMP_PARSE_FORMAT, timest_columns=["Timestamp"])488 dataframe["CaseID"] = dataframe["CaseID"].astype("string")489 490 pm4py.discover_working_together_network(dataframe, resource_key="Resource", case_id_key="CaseID", timestamp_key="Timestamp")491 492 def test_act_based_res_sim_log(self):493 for legacy_obj in [True, False]:494 log = pm4py.read_xes("input_data/running-example.xes", return_legacy_log_object=legacy_obj)495 pm4py.discover_activity_based_resource_similarity(log)496 497 def test_act_based_res_sim_df(self):498 dataframe = pandas_utils.read_csv("input_data/running-example-transformed.csv")499 dataframe = dataframe_utils.convert_timestamp_columns_in_df(dataframe, timest_format=constants.DEFAULT_TIMESTAMP_PARSE_FORMAT, timest_columns=["Timestamp"])500 dataframe["CaseID"] = dataframe["CaseID"].astype("string")501 502 pm4py.discover_activity_based_resource_similarity(dataframe, activity_key="Activity", resource_key="Resource", case_id_key="CaseID", timestamp_key="Timestamp")503 504 def test_subcontracting_log(self):505 for legacy_obj in [True, False]:506 log = pm4py.read_xes("input_data/running-example.xes", return_legacy_log_object=legacy_obj)507 pm4py.discover_subcontracting_network(log)508 509 def test_subcontracting_df(self):510 dataframe = pandas_utils.read_csv("input_data/running-example-transformed.csv")511 dataframe = dataframe_utils.convert_timestamp_columns_in_df(dataframe, timest_format=constants.DEFAULT_TIMESTAMP_PARSE_FORMAT, timest_columns=["Timestamp"])512 dataframe["CaseID"] = dataframe["CaseID"].astype("string")513 514 pm4py.discover_subcontracting_network(dataframe, resource_key="Resource", case_id_key="CaseID", timestamp_key="Timestamp")515 516 def test_roles_log(self):517 for legacy_obj in [True, False]:518 log = pm4py.read_xes("input_data/running-example.xes", return_legacy_log_object=legacy_obj)519 pm4py.discover_organizational_roles(log)520 521 def test_roles_df(self):522 dataframe = pandas_utils.read_csv("input_data/running-example-transformed.csv")523 dataframe = dataframe_utils.convert_timestamp_columns_in_df(dataframe, timest_format=constants.DEFAULT_TIMESTAMP_PARSE_FORMAT, timest_columns=["Timestamp"])524 dataframe["CaseID"] = dataframe["CaseID"].astype("string")525 526 pm4py.discover_organizational_roles(dataframe, activity_key="Activity", resource_key="Resource", case_id_key="CaseID", timestamp_key="Timestamp")527 528 def test_network_analysis_log(self):529 for legacy_obj in [True, False]:530 log = pm4py.read_xes("input_data/running-example.xes", return_legacy_log_object=legacy_obj)531 pm4py.discover_network_analysis(log, "case:concept:name", "case:concept:name", "org:resource", "org:resource", "concept:name")532 533 def test_network_analysis_df(self):534 dataframe = pandas_utils.read_csv("input_data/running-example-transformed.csv")535 dataframe = dataframe_utils.convert_timestamp_columns_in_df(dataframe, timest_format=constants.DEFAULT_TIMESTAMP_PARSE_FORMAT, timest_columns=["Timestamp"])536 dataframe["CaseID"] = dataframe["CaseID"].astype("string")537 538 pm4py.discover_network_analysis(dataframe, "CaseID", "CaseID", "Resource", "Resource", "Activity", sorting_column="Timestamp", timestamp_column="Timestamp")539 540 def test_discover_batches_log(self):541 for legacy_obj in [True, False]:542 log = pm4py.read_xes("input_data/running-example.xes", return_legacy_log_object=legacy_obj)543 pm4py.discover_batches(log)544 545 def test_discover_batches_df(self):546 dataframe = pandas_utils.read_csv("input_data/running-example-transformed.csv")547 dataframe = dataframe_utils.convert_timestamp_columns_in_df(dataframe, timest_format=constants.DEFAULT_TIMESTAMP_PARSE_FORMAT, timest_columns=["Timestamp"])548 dataframe["CaseID"] = dataframe["CaseID"].astype("string")549 550 pm4py.discover_batches(dataframe, activity_key="Activity", case_id_key="CaseID", timestamp_key="Timestamp", resource_key="Resource")551 552 def test_log_skeleton_log_simplified_interface(self):553 for legacy_obj in [True, False]:554 for diagn_df in [True, False]:555 log = pm4py.read_xes("input_data/running-example.xes", return_legacy_log_object=legacy_obj)556 model = pm4py.discover_log_skeleton(log)557 pm4py.conformance_log_skeleton(log, model, return_diagnostics_dataframe=diagn_df)558 559 def test_log_skeleton_df_simplified_interface(self):560 for diagn_df in [True, False]:561 dataframe = pandas_utils.read_csv("input_data/running-example-transformed.csv")562 dataframe = dataframe_utils.convert_timestamp_columns_in_df(dataframe,563 timest_format=constants.DEFAULT_TIMESTAMP_PARSE_FORMAT,564 timest_columns=["Timestamp"])565 dataframe["CaseID"] = dataframe["CaseID"].astype("string")566 567 model = pm4py.discover_log_skeleton(dataframe, activity_key="Activity", case_id_key="CaseID", timestamp_key="Timestamp")568 pm4py.conformance_log_skeleton(dataframe, model, activity_key="Activity", case_id_key="CaseID", timestamp_key="Timestamp", return_diagnostics_dataframe=diagn_df)569 570 def test_temporal_profile_log(self):571 for legacy_obj in [True, False]:572 log = pm4py.read_xes("input_data/running-example.xes", return_legacy_log_object=legacy_obj)573 model = pm4py.discover_temporal_profile(log)574 pm4py.conformance_temporal_profile(log, model)575 576 def test_temporal_profile_df(self):577 dataframe = pandas_utils.read_csv("input_data/running-example-transformed.csv")578 dataframe = dataframe_utils.convert_timestamp_columns_in_df(dataframe, timest_format=constants.DEFAULT_TIMESTAMP_PARSE_FORMAT, timest_columns=["Timestamp"])579 dataframe["CaseID"] = dataframe["CaseID"].astype("string")580 581 model = pm4py.discover_temporal_profile(dataframe, activity_key="Activity", case_id_key="CaseID", timestamp_key="Timestamp")582 pm4py.conformance_temporal_profile(dataframe, model, activity_key="Activity", case_id_key="CaseID", timestamp_key="Timestamp")583 584 def test_ocel_get_obj_types(self):585 ocel = pm4py.read_ocel("input_data/ocel/example_log.csv")586 pm4py.ocel_get_object_types(ocel)587 588 def test_ocel_get_attr_names(self):589 ocel = pm4py.read_ocel("input_data/ocel/example_log.csv")590 pm4py.ocel_get_attribute_names(ocel)591 592 def test_ocel_flattening(self):593 ocel = pm4py.read_ocel("input_data/ocel/example_log.csv")594 pm4py.ocel_flattening(ocel, "order")595 def test_stats_var_tuples_df(self):596 dataframe = pandas_utils.read_csv("input_data/running-example-transformed.csv")597 dataframe = dataframe_utils.convert_timestamp_columns_in_df(dataframe, timest_format=constants.DEFAULT_TIMESTAMP_PARSE_FORMAT, timest_columns=["Timestamp"])598 dataframe["CaseID"] = dataframe["CaseID"].astype("string")599 600 pm4py.get_variants_as_tuples(dataframe, activity_key="Activity", case_id_key="CaseID", timestamp_key="Timestamp")601 602 def test_stats_cycle_time_log(self):603 for legacy_obj in [True, False]:604 log = pm4py.read_xes("input_data/running-example.xes", return_legacy_log_object=legacy_obj)605 pm4py.get_cycle_time(log)606 607 def test_stats_cycle_time_df(self):608 dataframe = pandas_utils.read_csv("input_data/running-example-transformed.csv")609 dataframe = dataframe_utils.convert_timestamp_columns_in_df(dataframe, timest_format=constants.DEFAULT_TIMESTAMP_PARSE_FORMAT, timest_columns=["Timestamp"])610 dataframe["CaseID"] = dataframe["CaseID"].astype("string")611 612 pm4py.get_cycle_time(dataframe, activity_key="Activity", case_id_key="CaseID", timestamp_key="Timestamp")613 614 def test_stats_case_durations_log(self):615 for legacy_obj in [True, False]:616 log = pm4py.read_xes("input_data/running-example.xes", return_legacy_log_object=legacy_obj)617 pm4py.get_all_case_durations(log)618 619 def test_stats_case_durations_df(self):620 dataframe = pandas_utils.read_csv("input_data/running-example-transformed.csv")621 dataframe = dataframe_utils.convert_timestamp_columns_in_df(dataframe, timest_format=constants.DEFAULT_TIMESTAMP_PARSE_FORMAT, timest_columns=["Timestamp"])622 dataframe["CaseID"] = dataframe["CaseID"].astype("string")623 624 pm4py.get_all_case_durations(dataframe, activity_key="Activity", case_id_key="CaseID", timestamp_key="Timestamp")625 626 def test_stats_case_duration_log(self):627 for legacy_obj in [True, False]:628 log = pm4py.read_xes("input_data/running-example.xes", return_legacy_log_object=legacy_obj)629 pm4py.get_case_duration(log, "1")630 631 def test_stats_case_duration_df(self):632 dataframe = pandas_utils.read_csv("input_data/running-example-transformed.csv")633 dataframe = dataframe_utils.convert_timestamp_columns_in_df(dataframe, timest_format=constants.DEFAULT_TIMESTAMP_PARSE_FORMAT, timest_columns=["Timestamp"])634 dataframe["CaseID"] = dataframe["CaseID"].astype("string")635 636 pm4py.get_case_duration(dataframe, "1", activity_key="Activity", case_id_key="CaseID", timestamp_key="Timestamp")637 638 def test_stats_act_pos_summary_log(self):639 for legacy_obj in [True, False]:640 log = pm4py.read_xes("input_data/running-example.xes", return_legacy_log_object=legacy_obj)641 pm4py.get_activity_position_summary(log, "check ticket")642 643 def test_stats_act_pos_summary_df(self):644 dataframe = pandas_utils.read_csv("input_data/running-example-transformed.csv")645 dataframe = dataframe_utils.convert_timestamp_columns_in_df(dataframe, timest_format=constants.DEFAULT_TIMESTAMP_PARSE_FORMAT, timest_columns=["Timestamp"])646 dataframe["CaseID"] = dataframe["CaseID"].astype("string")647 648 pm4py.get_activity_position_summary(dataframe, "check ticket", activity_key="Activity", case_id_key="CaseID", timestamp_key="Timestamp")649 650 def test_filter_act_done_diff_res_log(self):651 for legacy_obj in [True, False]:652 log = pm4py.read_xes("input_data/running-example.xes", return_legacy_log_object=legacy_obj)653 pm4py.filter_activity_done_different_resources(log, "check ticket")654 655 def test_filter_act_done_diff_res_df(self):656 dataframe = pandas_utils.read_csv("input_data/running-example-transformed.csv")657 dataframe = dataframe_utils.convert_timestamp_columns_in_df(dataframe, timest_format=constants.DEFAULT_TIMESTAMP_PARSE_FORMAT, timest_columns=["Timestamp"])658 dataframe["CaseID"] = dataframe["CaseID"].astype("string")659 660 pm4py.filter_activity_done_different_resources(dataframe, "check ticket", activity_key="Activity", case_id_key="CaseID", timestamp_key="Timestamp", resource_key="Resource")661 662 def test_filter_four_eyes_principle_log(self):663 for legacy_obj in [True, False]:664 log = pm4py.read_xes("input_data/running-example.xes", return_legacy_log_object=legacy_obj)665 pm4py.filter_four_eyes_principle(log, "register request", "check ticket")666 667 def test_filter_four_eyes_principle_df(self):668 dataframe = pandas_utils.read_csv("input_data/running-example-transformed.csv")669 dataframe = dataframe_utils.convert_timestamp_columns_in_df(dataframe, timest_format=constants.DEFAULT_TIMESTAMP_PARSE_FORMAT, timest_columns=["Timestamp"])670 dataframe["CaseID"] = dataframe["CaseID"].astype("string")671 672 pm4py.filter_four_eyes_principle(dataframe, "register request", "check ticket", activity_key="Activity", case_id_key="CaseID", timestamp_key="Timestamp", resource_key="Resource")673 674 def test_filter_rel_occ_log(self):675 for legacy_obj in [True, False]:676 log = pm4py.read_xes("input_data/running-example.xes", return_legacy_log_object=legacy_obj)677 pm4py.filter_log_relative_occurrence_event_attribute(log, 0.8, level="cases")678 679 def test_filter_rel_occ_df(self):680 dataframe = pandas_utils.read_csv("input_data/running-example-transformed.csv")681 dataframe = dataframe_utils.convert_timestamp_columns_in_df(dataframe, timest_format=constants.DEFAULT_TIMESTAMP_PARSE_FORMAT, timest_columns=["Timestamp"])682 dataframe["CaseID"] = dataframe["CaseID"].astype("string")683 684 pm4py.filter_log_relative_occurrence_event_attribute(dataframe, 0.8, attribute_key="Activity", level="cases", case_id_key="CaseID", timestamp_key="Timestamp")685 686 def test_filter_start_activities_log(self):687 for legacy_obj in [True, False]:688 log = pm4py.read_xes("input_data/running-example.xes", return_legacy_log_object=legacy_obj)689 pm4py.filter_start_activities(log, ["register request"])690 691 def test_filter_start_activities_df(self):692 dataframe = pandas_utils.read_csv("input_data/running-example-transformed.csv")693 dataframe = dataframe_utils.convert_timestamp_columns_in_df(dataframe, timest_format=constants.DEFAULT_TIMESTAMP_PARSE_FORMAT, timest_columns=["Timestamp"])694 dataframe["CaseID"] = dataframe["CaseID"].astype("string")695 696 pm4py.filter_start_activities(dataframe, ["register request"], activity_key="Activity", case_id_key="CaseID", timestamp_key="Timestamp")697 698 def test_filter_end_activities_log(self):699 for legacy_obj in [True, False]:700 log = pm4py.read_xes("input_data/running-example.xes", return_legacy_log_object=legacy_obj)701 pm4py.filter_end_activities(log, ["pay compensation"])702 703 def test_filter_end_activities_df(self):704 dataframe = pandas_utils.read_csv("input_data/running-example-transformed.csv")705 dataframe = dataframe_utils.convert_timestamp_columns_in_df(dataframe, timest_format=constants.DEFAULT_TIMESTAMP_PARSE_FORMAT, timest_columns=["Timestamp"])706 dataframe["CaseID"] = dataframe["CaseID"].astype("string")707 708 pm4py.filter_end_activities(dataframe, ["pay compensation"], activity_key="Activity", case_id_key="CaseID", timestamp_key="Timestamp")709 710 def test_filter_eve_attr_values_log(self):711 for legacy_obj in [True, False]:712 log = pm4py.read_xes("input_data/running-example.xes", return_legacy_log_object=legacy_obj)713 pm4py.filter_event_attribute_values(log, "concept:name", ["register request", "pay compensation", "reject request"])714 715 def test_filter_eve_attr_values_df(self):716 dataframe = pandas_utils.read_csv("input_data/running-example-transformed.csv")717 dataframe = dataframe_utils.convert_timestamp_columns_in_df(dataframe, timest_format=constants.DEFAULT_TIMESTAMP_PARSE_FORMAT, timest_columns=["Timestamp"])718 dataframe["CaseID"] = dataframe["CaseID"].astype("string")719 720 pm4py.filter_event_attribute_values(dataframe, "Activity", ["register request", "pay compensation", "reject request"], case_id_key="CaseID")721 722 def test_filter_trace_attr_values_log(self):723 for legacy_obj in [True, False]:724 log = pm4py.read_xes("input_data/running-example.xes", return_legacy_log_object=legacy_obj)725 pm4py.filter_trace_attribute_values(log, "case:creator", ["Fluxicon"])726 727 def test_filter_variant_log(self):728 for legacy_obj in [True, False]:729 log = pm4py.read_xes("input_data/running-example.xes", return_legacy_log_object=legacy_obj)730 pm4py.filter_variants(log, [('register request', 'examine casually', 'check ticket', 'decide', 'reinitiate request', 'examine thoroughly', 'check ticket', 'decide', 'pay compensation')])731 732 def test_filter_variant_df(self):733 dataframe = pandas_utils.read_csv("input_data/running-example-transformed.csv")734 dataframe = dataframe_utils.convert_timestamp_columns_in_df(dataframe, timest_format=constants.DEFAULT_TIMESTAMP_PARSE_FORMAT, timest_columns=["Timestamp"])735 dataframe["CaseID"] = dataframe["CaseID"].astype("string")736 737 pm4py.filter_variants(dataframe, [('register request', 'examine casually', 'check ticket', 'decide', 'reinitiate request', 'examine thoroughly', 'check ticket', 'decide', 'pay compensation')], activity_key="Activity", case_id_key="CaseID", timestamp_key="Timestamp")738 739 def test_filter_dfg_log(self):740 for legacy_obj in [True, False]:741 log = pm4py.read_xes("input_data/running-example.xes", return_legacy_log_object=legacy_obj)742 pm4py.filter_directly_follows_relation(log, [("register request", "check ticket")])743 744 def test_filter_dfg_df(self):745 dataframe = pandas_utils.read_csv("input_data/running-example-transformed.csv")746 dataframe = dataframe_utils.convert_timestamp_columns_in_df(dataframe, timest_format=constants.DEFAULT_TIMESTAMP_PARSE_FORMAT, timest_columns=["Timestamp"])747 dataframe["CaseID"] = dataframe["CaseID"].astype("string")748 749 pm4py.filter_directly_follows_relation(dataframe, [("register request", "check ticket")], activity_key="Activity", case_id_key="CaseID", timestamp_key="Timestamp")750 751 def test_filter_efg_log(self):752 for legacy_obj in [True, False]:753 log = pm4py.read_xes("input_data/running-example.xes", return_legacy_log_object=legacy_obj)754 pm4py.filter_eventually_follows_relation(log, [("register request", "check ticket")])755 756 def test_filter_efg_df(self):757 dataframe = pandas_utils.read_csv("input_data/running-example-transformed.csv")758 dataframe = dataframe_utils.convert_timestamp_columns_in_df(dataframe, timest_format=constants.DEFAULT_TIMESTAMP_PARSE_FORMAT, timest_columns=["Timestamp"])759 dataframe["CaseID"] = dataframe["CaseID"].astype("string")760 761 pm4py.filter_eventually_follows_relation(dataframe, [("register request", "check ticket")], activity_key="Activity", case_id_key="CaseID", timestamp_key="Timestamp")762 763 def test_filter_time_range_log(self):764 for legacy_obj in [True, False]:765 log = pm4py.read_xes("input_data/running-example.xes", return_legacy_log_object=legacy_obj)766 pm4py.filter_time_range(log, "2009-01-01 01:00:00", "2011-01-01 01:00:00")767 768 def test_filter_time_range_df(self):769 dataframe = pandas_utils.read_csv("input_data/running-example-transformed.csv")770 dataframe = dataframe_utils.convert_timestamp_columns_in_df(dataframe, timest_format=constants.DEFAULT_TIMESTAMP_PARSE_FORMAT, timest_columns=["Timestamp"])771 dataframe["CaseID"] = dataframe["CaseID"].astype("string")772 773 pm4py.filter_time_range(dataframe, "2009-01-01 01:00:00", "2011-01-01 01:00:00", case_id_key="CaseID", timestamp_key="Timestamp")774 775 def test_filter_between_log(self):776 for legacy_obj in [True, False]:777 log = pm4py.read_xes("input_data/running-example.xes", return_legacy_log_object=legacy_obj)778 pm4py.filter_between(log, "check ticket", "decide")779 780 def test_filter_between_df(self):781 dataframe = pandas_utils.read_csv("input_data/running-example-transformed.csv")782 dataframe = dataframe_utils.convert_timestamp_columns_in_df(dataframe, timest_format=constants.DEFAULT_TIMESTAMP_PARSE_FORMAT, timest_columns=["Timestamp"])783 dataframe["CaseID"] = dataframe["CaseID"].astype("string")784 785 pm4py.filter_between(dataframe, "check ticket", "decide", activity_key="Activity", case_id_key="CaseID", timestamp_key="Timestamp")786 787 def test_filter_case_size_log(self):788 for legacy_obj in [True, False]:789 log = pm4py.read_xes("input_data/running-example.xes", return_legacy_log_object=legacy_obj)790 pm4py.filter_case_size(log, 10, 20)791 792 def test_filter_case_size_df(self):793 dataframe = pandas_utils.read_csv("input_data/running-example-transformed.csv")794 dataframe = dataframe_utils.convert_timestamp_columns_in_df(dataframe, timest_format=constants.DEFAULT_TIMESTAMP_PARSE_FORMAT, timest_columns=["Timestamp"])795 dataframe["CaseID"] = dataframe["CaseID"].astype("string")796 797 pm4py.filter_case_size(dataframe, 10, 20, case_id_key="CaseID")798 799 def test_filter_case_performance_log(self):800 for legacy_obj in [True, False]:801 log = pm4py.read_xes("input_data/running-example.xes", return_legacy_log_object=legacy_obj)802 pm4py.filter_case_performance(log, 86400, 8640000)803 804 def test_filter_case_performance_df(self):805 dataframe = pandas_utils.read_csv("input_data/running-example-transformed.csv")806 dataframe = dataframe_utils.convert_timestamp_columns_in_df(dataframe, timest_format=constants.DEFAULT_TIMESTAMP_PARSE_FORMAT, timest_columns=["Timestamp"])807 dataframe["CaseID"] = dataframe["CaseID"].astype("string")808 809 pm4py.filter_case_performance(dataframe, 86400, 8640000, case_id_key="CaseID", timestamp_key="Timestamp")810 811 def test_filter_activities_rework_log(self):812 for legacy_obj in [True, False]:813 log = pm4py.read_xes("input_data/running-example.xes", return_legacy_log_object=legacy_obj)814 pm4py.filter_activities_rework(log, "check ticket")815 816 def test_filter_act_rework_df(self):817 dataframe = pandas_utils.read_csv("input_data/running-example-transformed.csv")818 dataframe = dataframe_utils.convert_timestamp_columns_in_df(dataframe, timest_format=constants.DEFAULT_TIMESTAMP_PARSE_FORMAT, timest_columns=["Timestamp"])819 dataframe["CaseID"] = dataframe["CaseID"].astype("string")820 821 pm4py.filter_activities_rework(dataframe, "check ticket", activity_key="Activity", case_id_key="CaseID", timestamp_key="Timestamp")822 823 def test_filter_paths_perf_log(self):824 for legacy_obj in [True, False]:825 log = pm4py.read_xes("input_data/running-example.xes", return_legacy_log_object=legacy_obj)826 pm4py.filter_paths_performance(log, ("register request", "check ticket"), 86400, 864000)827 828 def test_filter_paths_perf_df(self):829 dataframe = pandas_utils.read_csv("input_data/running-example-transformed.csv")830 dataframe = dataframe_utils.convert_timestamp_columns_in_df(dataframe, timest_format=constants.DEFAULT_TIMESTAMP_PARSE_FORMAT, timest_columns=["Timestamp"])831 dataframe["CaseID"] = dataframe["CaseID"].astype("string")832 833 pm4py.filter_paths_performance(dataframe, ("register request", "check ticket"), 86400, 864000, activity_key="Activity", case_id_key="CaseID", timestamp_key="Timestamp")834 835 def test_filter_vars_top_k_log(self):836 for legacy_obj in [True, False]:837 log = pm4py.read_xes("input_data/running-example.xes", return_legacy_log_object=legacy_obj)838 pm4py.filter_variants_top_k(log, 1)839 840 def test_filter_vars_top_k_df(self):841 dataframe = pandas_utils.read_csv("input_data/running-example-transformed.csv")842 dataframe = dataframe_utils.convert_timestamp_columns_in_df(dataframe, timest_format=constants.DEFAULT_TIMESTAMP_PARSE_FORMAT, timest_columns=["Timestamp"])843 dataframe["CaseID"] = dataframe["CaseID"].astype("string")844 845 pm4py.filter_variants_top_k(dataframe, 1, activity_key="Activity", case_id_key="CaseID", timestamp_key="Timestamp")846 847 def test_filter_vars_coverage(self):848 for legacy_obj in [True, False]:849 log = pm4py.read_xes("input_data/running-example.xes", return_legacy_log_object=legacy_obj)850 pm4py.filter_variants_by_coverage_percentage(log, 0.1)851 852 def test_filter_vars_coverage(self):853 dataframe = pandas_utils.read_csv("input_data/running-example-transformed.csv")854 dataframe = dataframe_utils.convert_timestamp_columns_in_df(dataframe, timest_format=constants.DEFAULT_TIMESTAMP_PARSE_FORMAT, timest_columns=["Timestamp"])855 dataframe["CaseID"] = dataframe["CaseID"].astype("string")856 857 pm4py.filter_variants_by_coverage_percentage(dataframe, 0.1, activity_key="Activity", case_id_key="CaseID", timestamp_key="Timestamp")858 859 def test_filter_prefixes_log(self):860 for legacy_obj in [True, False]:861 log = pm4py.read_xes("input_data/running-example.xes", return_legacy_log_object=legacy_obj)862 pm4py.filter_prefixes(log, "check ticket")863 864 def test_filter_prefixes_df(self):865 dataframe = pandas_utils.read_csv("input_data/running-example-transformed.csv")866 dataframe = dataframe_utils.convert_timestamp_columns_in_df(dataframe, timest_format=constants.DEFAULT_TIMESTAMP_PARSE_FORMAT, timest_columns=["Timestamp"])867 dataframe["CaseID"] = dataframe["CaseID"].astype("string")868 869 pm4py.filter_prefixes(dataframe, "check ticket", activity_key="Activity", case_id_key="CaseID", timestamp_key="Timestamp")870 871 def test_filter_suffixes_log(self):872 for legacy_obj in [True, False]:873 log = pm4py.read_xes("input_data/running-example.xes", return_legacy_log_object=legacy_obj)874 pm4py.filter_suffixes(log, "check ticket")875 876 def test_filter_suffixes_df(self):877 dataframe = pandas_utils.read_csv("input_data/running-example-transformed.csv")878 dataframe = dataframe_utils.convert_timestamp_columns_in_df(dataframe, timest_format=constants.DEFAULT_TIMESTAMP_PARSE_FORMAT, timest_columns=["Timestamp"])879 dataframe["CaseID"] = dataframe["CaseID"].astype("string")880 881 pm4py.filter_suffixes(dataframe, "check ticket", activity_key="Activity", case_id_key="CaseID", timestamp_key="Timestamp")882 883 def test_discover_perf_dfg_log(self):884 for legacy_obj in [True, False]:885 log = pm4py.read_xes("input_data/running-example.xes", return_legacy_log_object=legacy_obj)886 pm4py.discover_performance_dfg(log)887 888 def test_discover_perf_dfg_df(self):889 dataframe = pandas_utils.read_csv("input_data/running-example-transformed.csv")890 dataframe = dataframe_utils.convert_timestamp_columns_in_df(dataframe, timest_format=constants.DEFAULT_TIMESTAMP_PARSE_FORMAT, timest_columns=["Timestamp"])891 dataframe["CaseID"] = dataframe["CaseID"].astype("string")892 893 pm4py.discover_performance_dfg(dataframe, activity_key="Activity", case_id_key="CaseID", timestamp_key="Timestamp")894 895 def test_discover_footprints_log(self):896 for legacy_obj in [True, False]:897 log = pm4py.read_xes("input_data/running-example.xes", return_legacy_log_object=legacy_obj)898 pm4py.discover_footprints(log)899 900 def test_discover_ts_log(self):901 for legacy_obj in [True, False]:902 log = pm4py.read_xes("input_data/running-example.xes", return_legacy_log_object=legacy_obj)903 pm4py.discover_transition_system(log)904 905 def test_discover_ts_df(self):906 dataframe = pandas_utils.read_csv("input_data/running-example-transformed.csv")907 dataframe = dataframe_utils.convert_timestamp_columns_in_df(dataframe, timest_format=constants.DEFAULT_TIMESTAMP_PARSE_FORMAT, timest_columns=["Timestamp"])908 dataframe["CaseID"] = dataframe["CaseID"].astype("string")909 910 pm4py.discover_transition_system(dataframe, activity_key="Activity", case_id_key="CaseID", timestamp_key="Timestamp")911 912 def test_discover_pref_tree_log(self):913 for legacy_obj in [True, False]:914 log = pm4py.read_xes("input_data/running-example.xes", return_legacy_log_object=legacy_obj)915 pm4py.discover_prefix_tree(log)916 917 def test_discover_pref_tree_df(self):918 dataframe = pandas_utils.read_csv("input_data/running-example-transformed.csv")919 dataframe = dataframe_utils.convert_timestamp_columns_in_df(dataframe, timest_format=constants.DEFAULT_TIMESTAMP_PARSE_FORMAT, timest_columns=["Timestamp"])920 dataframe["CaseID"] = dataframe["CaseID"].astype("string")921 922 pm4py.discover_prefix_tree(dataframe, activity_key="Activity", case_id_key="CaseID", timestamp_key="Timestamp")923 924 def test_discover_ocpn(self):925 ocel = pm4py.read_ocel("input_data/ocel/example_log.csv")926 pm4py.discover_oc_petri_net(ocel)927 928 def test_conformance_alignments_pn_log_simplified_interface(self):929 for legacy_obj in [True, False]:930 for diagn_df in [True, False]:931 log = pm4py.read_xes("input_data/running-example.xes", return_legacy_log_object=legacy_obj)932 net, im, fm = pm4py.discover_petri_net_inductive(log)933 pm4py.conformance_diagnostics_alignments(log, net, im, fm, return_diagnostics_dataframe=diagn_df)934 935 def test_conformance_alignments_pn_df_simplified_interface(self):936 for diagn_df in [True, False]:937 dataframe = pandas_utils.read_csv("input_data/running-example-transformed.csv")938 dataframe = dataframe_utils.convert_timestamp_columns_in_df(dataframe,939 timest_format=constants.DEFAULT_TIMESTAMP_PARSE_FORMAT,940 timest_columns=["Timestamp"])941 dataframe["CaseID"] = dataframe["CaseID"].astype("string")942 943 net, im, fm = pm4py.discover_petri_net_inductive(dataframe, activity_key="Activity", case_id_key="CaseID", timestamp_key="Timestamp")944 pm4py.conformance_diagnostics_alignments(dataframe, net, im, fm, activity_key="Activity", case_id_key="CaseID", timestamp_key="Timestamp", return_diagnostics_dataframe=diagn_df)945 946 def test_conformance_diagnostics_fp_log(self):947 for legacy_obj in [True, False]:948 log = pm4py.read_xes("input_data/running-example.xes", return_legacy_log_object=legacy_obj)949 tree = pm4py.discover_process_tree_inductive(log)950 pm4py.conformance_diagnostics_footprints(log, tree)951 952 def test_fitness_fp_log(self):953 for legacy_obj in [True, False]:954 log = pm4py.read_xes("input_data/running-example.xes", return_legacy_log_object=legacy_obj)955 tree = pm4py.discover_process_tree_inductive(log)956 pm4py.fitness_footprints(log, tree)957 958 def test_precision_fp_log(self):959 for legacy_obj in [True, False]:960 log = pm4py.read_xes("input_data/running-example.xes", return_legacy_log_object=legacy_obj)961 tree = pm4py.discover_process_tree_inductive(log)962 pm4py.precision_footprints(log, tree)963 964 def test_maximal_decomposition(self):965 net, im, fm = pm4py.read_pnml("input_data/running-example.pnml")966 pm4py.maximal_decomposition(net, im, fm)967 968 def test_fea_ext_log(self):969 for legacy_obj in [True, False]:970 log = pm4py.read_xes("input_data/running-example.xes", return_legacy_log_object=legacy_obj)971 pm4py.extract_features_dataframe(log)972 973 def test_fea_ext_df(self):974 dataframe = pandas_utils.read_csv("input_data/running-example-transformed.csv")975 dataframe = dataframe_utils.convert_timestamp_columns_in_df(dataframe, timest_format=constants.DEFAULT_TIMESTAMP_PARSE_FORMAT, timest_columns=["Timestamp"])976 dataframe["CaseID"] = dataframe["CaseID"].astype("string")977 978 pm4py.extract_features_dataframe(dataframe, activity_key="Activity", case_id_key="CaseID", timestamp_key="Timestamp", resource_key="Resource")979 980 def test_new_alpha_miner_df(self):981 dataframe = pandas_utils.read_csv("input_data/running-example-transformed.csv")982 dataframe = dataframe_utils.convert_timestamp_columns_in_df(dataframe, timest_format=constants.DEFAULT_TIMESTAMP_PARSE_FORMAT, timest_columns=["Timestamp"])983 dataframe["CaseID"] = dataframe["CaseID"].astype("string")984 pm4py.discover_petri_net_alpha(dataframe, case_id_key="CaseID", activity_key="Activity", timestamp_key="Timestamp")985 986 def test_new_heu_miner_df(self):987 dataframe = pandas_utils.read_csv("input_data/running-example-transformed.csv")988 dataframe = dataframe_utils.convert_timestamp_columns_in_df(dataframe, timest_format=constants.DEFAULT_TIMESTAMP_PARSE_FORMAT, timest_columns=["Timestamp"])989 dataframe["CaseID"] = dataframe["CaseID"].astype("string")990 pm4py.discover_petri_net_heuristics(dataframe, case_id_key="CaseID", activity_key="Activity", timestamp_key="Timestamp")991 992 def test_new_dfg_df(self):993 dataframe = pandas_utils.read_csv("input_data/running-example-transformed.csv")994 dataframe = dataframe_utils.convert_timestamp_columns_in_df(dataframe, timest_format=constants.DEFAULT_TIMESTAMP_PARSE_FORMAT, timest_columns=["Timestamp"])995 dataframe["CaseID"] = dataframe["CaseID"].astype("string")996 pm4py.discover_dfg(dataframe, case_id_key="CaseID", activity_key="Activity", timestamp_key="Timestamp")997 998 def test_new_perf_dfg_df(self):999 dataframe = pandas_utils.read_csv("input_data/running-example-transformed.csv")1000 dataframe = dataframe_utils.convert_timestamp_columns_in_df(dataframe, timest_format=constants.DEFAULT_TIMESTAMP_PARSE_FORMAT, timest_columns=["Timestamp"])1001 dataframe["CaseID"] = dataframe["CaseID"].astype("string")1002 pm4py.discover_performance_dfg(dataframe, case_id_key="CaseID", activity_key="Activity", timestamp_key="Timestamp")1003 1004 def test_new_tbr_df_simpl_interface(self):1005 for ret_df in [True, False]:1006 dataframe = pandas_utils.read_csv("input_data/running-example-transformed.csv")1007 dataframe = dataframe_utils.convert_timestamp_columns_in_df(dataframe,1008 timest_format=constants.DEFAULT_TIMESTAMP_PARSE_FORMAT,1009 timest_columns=["Timestamp"])1010 dataframe["CaseID"] = dataframe["CaseID"].astype("string")1011 net, im, fm = pm4py.discover_petri_net_inductive(dataframe, case_id_key="CaseID", activity_key="Activity", timestamp_key="Timestamp")1012 pm4py.conformance_diagnostics_token_based_replay(dataframe, net, im, fm, case_id_key="CaseID", activity_key="Activity", timestamp_key="Timestamp", return_diagnostics_dataframe=ret_df)1013 1014 def test_new_tbr_fitness_df(self):1015 dataframe = pandas_utils.read_csv("input_data/running-example-transformed.csv")1016 dataframe = dataframe_utils.convert_timestamp_columns_in_df(dataframe, timest_format=constants.DEFAULT_TIMESTAMP_PARSE_FORMAT, timest_columns=["Timestamp"])1017 dataframe["CaseID"] = dataframe["CaseID"].astype("string")1018 net, im, fm = pm4py.discover_petri_net_inductive(dataframe, case_id_key="CaseID", activity_key="Activity", timestamp_key="Timestamp")1019 pm4py.fitness_token_based_replay(dataframe, net, im, fm, case_id_key="CaseID", activity_key="Activity", timestamp_key="Timestamp")1020 1021 def test_new_tbr_precision_df(self):1022 dataframe = pandas_utils.read_csv("input_data/running-example-transformed.csv")1023 dataframe = dataframe_utils.convert_timestamp_columns_in_df(dataframe, timest_format=constants.DEFAULT_TIMESTAMP_PARSE_FORMAT, timest_columns=["Timestamp"])1024 dataframe["CaseID"] = dataframe["CaseID"].astype("string")1025 net, im, fm = pm4py.discover_petri_net_inductive(dataframe, case_id_key="CaseID", activity_key="Activity", timestamp_key="Timestamp")1026 pm4py.precision_token_based_replay(dataframe, net, im, fm, case_id_key="CaseID", activity_key="Activity", timestamp_key="Timestamp")1027 1028 def test_new_align_df_simpl_interface(self):1029 for diagn_df in [True, False]:1030 dataframe = pandas_utils.read_csv("input_data/running-example-transformed.csv")1031 dataframe = dataframe_utils.convert_timestamp_columns_in_df(dataframe,1032 timest_format=constants.DEFAULT_TIMESTAMP_PARSE_FORMAT,1033 timest_columns=["Timestamp"])1034 dataframe["CaseID"] = dataframe["CaseID"].astype("string")1035 net, im, fm = pm4py.discover_petri_net_inductive(dataframe, case_id_key="CaseID", activity_key="Activity", timestamp_key="Timestamp")1036 pm4py.conformance_diagnostics_alignments(dataframe, net, im, fm, case_id_key="CaseID", activity_key="Activity", timestamp_key="Timestamp", return_diagnostics_dataframe=diagn_df)1037 1038 def test_new_align_fitness_df(self):1039 dataframe = pandas_utils.read_csv("input_data/running-example-transformed.csv")1040 dataframe = dataframe_utils.convert_timestamp_columns_in_df(dataframe, timest_format=constants.DEFAULT_TIMESTAMP_PARSE_FORMAT, timest_columns=["Timestamp"])1041 dataframe["CaseID"] = dataframe["CaseID"].astype("string")1042 net, im, fm = pm4py.discover_petri_net_inductive(dataframe, case_id_key="CaseID", activity_key="Activity", timestamp_key="Timestamp")1043 pm4py.fitness_alignments(dataframe, net, im, fm, case_id_key="CaseID", activity_key="Activity", timestamp_key="Timestamp")1044 1045 def test_new_align_precision_df(self):1046 dataframe = pandas_utils.read_csv("input_data/running-example-transformed.csv")1047 dataframe = dataframe_utils.convert_timestamp_columns_in_df(dataframe, timest_format=constants.DEFAULT_TIMESTAMP_PARSE_FORMAT, timest_columns=["Timestamp"])1048 dataframe["CaseID"] = dataframe["CaseID"].astype("string")1049 net, im, fm = pm4py.discover_petri_net_inductive(dataframe, case_id_key="CaseID", activity_key="Activity",1050 timestamp_key="Timestamp")1051 pm4py.precision_alignments(dataframe, net, im, fm, case_id_key="CaseID", activity_key="Activity", timestamp_key="Timestamp")1052 1053 def test_vis_case_duration_df(self):1054 dataframe = pandas_utils.read_csv("input_data/running-example-transformed.csv")1055 dataframe = dataframe_utils.convert_timestamp_columns_in_df(dataframe, timest_format=constants.DEFAULT_TIMESTAMP_PARSE_FORMAT, timest_columns=["Timestamp"])1056 dataframe["CaseID"] = dataframe["CaseID"].astype("string")1057 1058 if importlib.util.find_spec("matplotlib"):1059 target = os.path.join("test_output_data", "case_duration.svg")1060 pm4py.save_vis_case_duration_graph(dataframe, target, activity_key="Activity", case_id_key="CaseID", timestamp_key="Timestamp")1061 os.remove(target)1062 1063 def test_vis_ev_distr_graph_df(self):1064 dataframe = pandas_utils.read_csv("input_data/running-example-transformed.csv")1065 dataframe = dataframe_utils.convert_timestamp_columns_in_df(dataframe, timest_format=constants.DEFAULT_TIMESTAMP_PARSE_FORMAT, timest_columns=["Timestamp"])1066 dataframe["CaseID"] = dataframe["CaseID"].astype("string")1067 target = os.path.join("test_output_data", "ev_distr_graph.svg")1068 1069 if importlib.util.find_spec("matplotlib"):1070 pm4py.save_vis_events_distribution_graph(dataframe, target, activity_key="Activity", case_id_key="CaseID", timestamp_key="Timestamp")1071 os.remove(target)1072 1073 def test_ocel_object_graph(self):1074 ocel = pm4py.read_ocel("input_data/ocel/example_log.jsonocel")1075 ev_graph = pm4py.discover_objects_graph(ocel, graph_type="object_interaction")1076 ev_graph = pm4py.discover_objects_graph(ocel, graph_type="object_descendants")1077 ev_graph = pm4py.discover_objects_graph(ocel, graph_type="object_inheritance")1078 ev_graph = pm4py.discover_objects_graph(ocel, graph_type="object_cobirth")1079 ev_graph = pm4py.discover_objects_graph(ocel, graph_type="object_codeath")1080 1081 def test_ocel_temporal_summary(self):1082 ocel = pm4py.read_ocel("input_data/ocel/example_log.jsonocel")1083 temp_summary = pm4py.ocel_temporal_summary(ocel)1084 1085 def test_ocel_objects_summary(self):1086 ocel = pm4py.read_ocel("input_data/ocel/example_log.jsonocel")1087 objects_summary = pm4py.ocel_objects_summary(ocel)1088 1089 def test_ocel_filtering_ev_ids(self):1090 ocel = pm4py.read_ocel("input_data/ocel/example_log.jsonocel")1091 filtered_ocel = pm4py.filter_ocel_events(ocel, ["e1"])1092 1093 def test_ocel_filtering_obj_ids(self):1094 ocel = pm4py.read_ocel("input_data/ocel/example_log.jsonocel")1095 filtered_ocel = pm4py.filter_ocel_objects(ocel, ["o1"], level=1)1096 filtered_ocel = pm4py.filter_ocel_objects(ocel, ["o1"], level=2)1097 1098 def test_ocel_filtering_obj_types(self):1099 ocel = pm4py.read_ocel("input_data/ocel/example_log.jsonocel")1100 filtered_ocel = pm4py.filter_ocel_object_types(ocel, ["order"])1101 1102 def test_ocel_filtering_cc(self):1103 ocel = pm4py.read_ocel("input_data/ocel/example_log.jsonocel")1104 filtered_ocel = pm4py.filter_ocel_cc_object(ocel, "o1")1105 1106 def test_ocel_drop_duplicates(self):1107 ocel = pm4py.read_ocel("input_data/ocel/example_log.jsonocel")1108 filtered_ocel = pm4py.ocel_drop_duplicates(ocel)1109 1110 def test_ocel_add_index_based_timedelta(self):1111 ocel = pm4py.read_ocel("input_data/ocel/example_log.jsonocel")1112 filtered_ocel = pm4py.ocel_add_index_based_timedelta(ocel)1113 1114 def test_ocel2_xml(self):1115 ocel = pm4py.read_ocel2("input_data/ocel/ocel20_example.xmlocel")1116 pm4py.write_ocel2(ocel, "test_output_data/ocel20_example.xmlocel")1117 os.remove("test_output_data/ocel20_example.xmlocel")1118 1119 def test_ocel2_sqlite(self):1120 ocel = pm4py.read_ocel2("input_data/ocel/ocel20_example.sqlite")1121 pm4py.write_ocel2(ocel, "test_output_data/ocel20_example.sqlite")1122 os.remove("test_output_data/ocel20_example.sqlite")1123 1124 1125if __name__ == "__main__":1126 unittest.main()1127