diff --git a/examples/timestamp_granularity.py b/examples/timestamp_granularity.py index a6dcc7f74..3fb139add 100644 --- a/examples/timestamp_granularity.py +++ b/examples/timestamp_granularity.py @@ -12,17 +12,15 @@ def execute_script(): # Here are some common options that you can use as a granularity: # - # 'D': Day - # 'H': Hour - # 'T' or 'min': Minute - # 'S': Second - # 'L' or 'ms': Millisecond - # 'U': Microsecond - # 'N': Nanosecond + # 'h': Hour + # 'min': Minute + # 's': Second + # 'ms': Millisecond + # 'ns': Nanosecond st = time.time_ns() # cast on the minute - dataframe["time:timestamp"] = dataframe["time:timestamp"].dt.floor('T') + dataframe["time:timestamp"] = dataframe["time:timestamp"].dt.floor(freq='min') ct = time.time_ns() print("required time for the timestamp casting: %.2f seconds" % ((ct-st)/10**9)) diff --git a/tests/dec_tree_test.py b/tests/dec_tree_test.py index 537587767..d733344ae 100644 --- a/tests/dec_tree_test.py +++ b/tests/dec_tree_test.py @@ -1,5 +1,6 @@ import os import unittest +import importlib.util from pm4py.objects.log.importer.xes import importer as xes_importer from pm4py.objects.log.util import get_class_representation @@ -8,43 +9,45 @@ class DecisionTreeTest(unittest.TestCase): def test_decisiontree_evattrvalue(self): - from pm4py.util import ml_utils - from pm4py.visualization.decisiontree import visualizer as dt_vis - - # to avoid static method warnings in tests, - # that by construction of the unittest package have to be expressed in such way - self.dummy_variable = "dummy_value" - log_path = os.path.join("input_data", "roadtraffic50traces.xes") - log = xes_importer.apply(log_path) - data, feature_names = log_to_features.apply(log, variant=log_to_features.Variants.TRACE_BASED, - parameters={"str_tr_attr": [], "str_ev_attr": ["concept:name"], - "num_tr_attr": [], "num_ev_attr": ["amount"]}) - target, classes = get_class_representation.get_class_representation_by_str_ev_attr_value_value(log, - "concept:name") - clf = ml_utils.DecisionTreeClassifier(max_depth=7) - clf.fit(data, target) - gviz = dt_vis.apply(clf, feature_names, classes, - parameters={dt_vis.Variants.CLASSIC.value.Parameters.FORMAT: "svg"}) - del gviz + if importlib.util.find_spec("sklearn"): + from pm4py.util import ml_utils + from pm4py.visualization.decisiontree import visualizer as dt_vis + + # to avoid static method warnings in tests, + # that by construction of the unittest package have to be expressed in such way + self.dummy_variable = "dummy_value" + log_path = os.path.join("input_data", "roadtraffic50traces.xes") + log = xes_importer.apply(log_path) + data, feature_names = log_to_features.apply(log, variant=log_to_features.Variants.TRACE_BASED, + parameters={"str_tr_attr": [], "str_ev_attr": ["concept:name"], + "num_tr_attr": [], "num_ev_attr": ["amount"]}) + target, classes = get_class_representation.get_class_representation_by_str_ev_attr_value_value(log, + "concept:name") + clf = ml_utils.DecisionTreeClassifier(max_depth=7) + clf.fit(data, target) + gviz = dt_vis.apply(clf, feature_names, classes, + parameters={dt_vis.Variants.CLASSIC.value.Parameters.FORMAT: "svg"}) + del gviz def test_decisiontree_traceduration(self): - from pm4py.util import ml_utils - from pm4py.visualization.decisiontree import visualizer as dt_vis - - # to avoid static method warnings in tests, - # that by construction of the unittest package have to be expressed in such way - self.dummy_variable = "dummy_value" - log_path = os.path.join("input_data", "roadtraffic50traces.xes") - log = xes_importer.apply(log_path) - data, feature_names = log_to_features.apply(log, variant=log_to_features.Variants.TRACE_BASED, - parameters={"str_tr_attr": [], "str_ev_attr": ["concept:name"], - "num_tr_attr": [], "num_ev_attr": ["amount"]}) - target, classes = get_class_representation.get_class_representation_by_trace_duration(log, 2 * 8640000) - clf = ml_utils.DecisionTreeClassifier(max_depth=7) - clf.fit(data, target) - gviz = dt_vis.apply(clf, feature_names, classes, - parameters={dt_vis.Variants.CLASSIC.value.Parameters.FORMAT: "svg"}) - del gviz + if importlib.util.find_spec("sklearn"): + from pm4py.util import ml_utils + from pm4py.visualization.decisiontree import visualizer as dt_vis + + # to avoid static method warnings in tests, + # that by construction of the unittest package have to be expressed in such way + self.dummy_variable = "dummy_value" + log_path = os.path.join("input_data", "roadtraffic50traces.xes") + log = xes_importer.apply(log_path) + data, feature_names = log_to_features.apply(log, variant=log_to_features.Variants.TRACE_BASED, + parameters={"str_tr_attr": [], "str_ev_attr": ["concept:name"], + "num_tr_attr": [], "num_ev_attr": ["amount"]}) + target, classes = get_class_representation.get_class_representation_by_trace_duration(log, 2 * 8640000) + clf = ml_utils.DecisionTreeClassifier(max_depth=7) + clf.fit(data, target) + gviz = dt_vis.apply(clf, feature_names, classes, + parameters={dt_vis.Variants.CLASSIC.value.Parameters.FORMAT: "svg"}) + del gviz if __name__ == "__main__": diff --git a/tests/doc_tests.py b/tests/doc_tests.py index 2ad7a6367..03a762d81 100644 --- a/tests/doc_tests.py +++ b/tests/doc_tests.py @@ -1,5 +1,6 @@ from pm4py.objects.log.util import dataframe_utils import unittest +import importlib.util import os from pm4py.util import constants, pandas_utils from pm4py.objects.conversion.process_tree import converter as process_tree_converter @@ -601,51 +602,53 @@ def test_59(self): gviz = pt_visualizer.apply(tree, parameters={pt_visualizer.Variants.WO_DECORATION.value.Parameters.FORMAT: "png"}) def test_60(self): - import os - from pm4py.objects.log.importer.xes import importer as xes_importer - log = xes_importer.apply(os.path.join("input_data", "roadtraffic50traces.xes")) - - from pm4py.algo.transformation.log_to_features.variants import trace_based - str_trace_attributes = [] - str_event_attributes = ["concept:name"] - num_trace_attributes = [] - num_event_attributes = ["amount"] - data, feature_names = trace_based.apply(log) - data, feature_names = trace_based.apply(log, parameters={"str_tr_attr": str_trace_attributes, "str_ev_attr": str_event_attributes, "num_tr_attr": num_trace_attributes, "num_ev_attr": num_event_attributes}) - - from pm4py.objects.log.util import get_class_representation - target, classes = get_class_representation.get_class_representation_by_str_ev_attr_value_value(log, - "concept:name") - - from pm4py.util import ml_utils - clf = ml_utils.DecisionTreeClassifier() - clf.fit(data, target) - - from pm4py.visualization.decisiontree import visualizer as dectree_visualizer - gviz = dectree_visualizer.apply(clf, feature_names, classes) + if importlib.util.find_spec("sklearn"): + import os + from pm4py.objects.log.importer.xes import importer as xes_importer + log = xes_importer.apply(os.path.join("input_data", "roadtraffic50traces.xes")) + + from pm4py.algo.transformation.log_to_features.variants import trace_based + str_trace_attributes = [] + str_event_attributes = ["concept:name"] + num_trace_attributes = [] + num_event_attributes = ["amount"] + data, feature_names = trace_based.apply(log) + data, feature_names = trace_based.apply(log, parameters={"str_tr_attr": str_trace_attributes, "str_ev_attr": str_event_attributes, "num_tr_attr": num_trace_attributes, "num_ev_attr": num_event_attributes}) + + from pm4py.objects.log.util import get_class_representation + target, classes = get_class_representation.get_class_representation_by_str_ev_attr_value_value(log, + "concept:name") + + from pm4py.util import ml_utils + clf = ml_utils.DecisionTreeClassifier() + clf.fit(data, target) + + from pm4py.visualization.decisiontree import visualizer as dectree_visualizer + gviz = dectree_visualizer.apply(clf, feature_names, classes) def test_61(self): - import os - from pm4py.objects.log.importer.xes import importer as xes_importer - log = xes_importer.apply(os.path.join("input_data", "roadtraffic50traces.xes")) + if importlib.util.find_spec("sklearn"): + import os + from pm4py.objects.log.importer.xes import importer as xes_importer + log = xes_importer.apply(os.path.join("input_data", "roadtraffic50traces.xes")) - from pm4py.algo.transformation.log_to_features.variants import trace_based - str_trace_attributes = [] - str_event_attributes = ["concept:name"] - num_trace_attributes = [] - num_event_attributes = ["amount"] - data, feature_names = trace_based.apply(log) - data, feature_names = trace_based.apply(log, parameters={"str_tr_attr": str_trace_attributes, "str_ev_attr": str_event_attributes, "num_tr_attr": num_trace_attributes, "num_ev_attr": num_event_attributes}) + from pm4py.algo.transformation.log_to_features.variants import trace_based + str_trace_attributes = [] + str_event_attributes = ["concept:name"] + num_trace_attributes = [] + num_event_attributes = ["amount"] + data, feature_names = trace_based.apply(log) + data, feature_names = trace_based.apply(log, parameters={"str_tr_attr": str_trace_attributes, "str_ev_attr": str_event_attributes, "num_tr_attr": num_trace_attributes, "num_ev_attr": num_event_attributes}) - from pm4py.objects.log.util import get_class_representation - target, classes = get_class_representation.get_class_representation_by_trace_duration(log, 2 * 8640000) + from pm4py.objects.log.util import get_class_representation + target, classes = get_class_representation.get_class_representation_by_trace_duration(log, 2 * 8640000) - from pm4py.util import ml_utils - clf = ml_utils.DecisionTreeClassifier() - clf.fit(data, target) + from pm4py.util import ml_utils + clf = ml_utils.DecisionTreeClassifier() + clf.fit(data, target) - from pm4py.visualization.decisiontree import visualizer as dectree_visualizer - gviz = dectree_visualizer.apply(clf, feature_names, classes) + from pm4py.visualization.decisiontree import visualizer as dectree_visualizer + gviz = dectree_visualizer.apply(clf, feature_names, classes) def test_62(self): from pm4py.objects.log.importer.xes import importer as xes_importer @@ -718,56 +721,57 @@ def test_tbr_diagn_1(self): pass def test_tbr_diagn_2(self): - import os - from pm4py.objects.log.importer.xes import importer as xes_importer - log = xes_importer.apply(os.path.join("input_data", "receipt.xes")) - - from pm4py.algo.filtering.log.variants import variants_filter - filtered_log = variants_filter.filter_log_variants_percentage(log, 0.2) - - from pm4py.algo.discovery.inductive import algorithm as inductive_miner - process_tree = inductive_miner.apply(log) - net, initial_marking, final_marking = process_tree_converter.apply(process_tree) - - # build decision trees - string_attributes = ["org:group"] - numeric_attributes = [] - parameters = {"string_attributes": string_attributes, "numeric_attributes": numeric_attributes} - - from pm4py.algo.conformance.tokenreplay import algorithm as token_based_replay - parameters_tbr = {token_based_replay.Variants.TOKEN_REPLAY.value.Parameters.DISABLE_VARIANTS: True, - token_based_replay.Variants.TOKEN_REPLAY.value.Parameters.ENABLE_PLTR_FITNESS: True} - replayed_traces, place_fitness, trans_fitness, unwanted_activities = token_based_replay.apply(log, net, - initial_marking, - final_marking, - parameters=parameters_tbr) - - from pm4py.algo.conformance.tokenreplay.diagnostics import root_cause_analysis - trans_root_cause = root_cause_analysis.diagnose_from_trans_fitness(log, trans_fitness, parameters=parameters) - - - from pm4py.visualization.decisiontree import visualizer as dt_vis - for trans in trans_root_cause: - clf = trans_root_cause[trans]["clf"] - feature_names = trans_root_cause[trans]["feature_names"] - classes = trans_root_cause[trans]["classes"] - - # visualization could be called - gviz = dt_vis.apply(clf, feature_names, classes) - break - - from pm4py.algo.conformance.tokenreplay.diagnostics import root_cause_analysis - act_root_cause = root_cause_analysis.diagnose_from_notexisting_activities(log, unwanted_activities, - parameters=parameters) - - from pm4py.visualization.decisiontree import visualizer as dt_vis - for act in act_root_cause: - clf = act_root_cause[act]["clf"] - feature_names = act_root_cause[act]["feature_names"] - classes = act_root_cause[act]["classes"] - # visualization could be called - gviz = dt_vis.apply(clf, feature_names, classes) - break + if importlib.util.find_spec("sklearn"): + import os + from pm4py.objects.log.importer.xes import importer as xes_importer + log = xes_importer.apply(os.path.join("input_data", "receipt.xes")) + + from pm4py.algo.filtering.log.variants import variants_filter + filtered_log = variants_filter.filter_log_variants_percentage(log, 0.2) + + from pm4py.algo.discovery.inductive import algorithm as inductive_miner + process_tree = inductive_miner.apply(log) + net, initial_marking, final_marking = process_tree_converter.apply(process_tree) + + # build decision trees + string_attributes = ["org:group"] + numeric_attributes = [] + parameters = {"string_attributes": string_attributes, "numeric_attributes": numeric_attributes} + + from pm4py.algo.conformance.tokenreplay import algorithm as token_based_replay + parameters_tbr = {token_based_replay.Variants.TOKEN_REPLAY.value.Parameters.DISABLE_VARIANTS: True, + token_based_replay.Variants.TOKEN_REPLAY.value.Parameters.ENABLE_PLTR_FITNESS: True} + replayed_traces, place_fitness, trans_fitness, unwanted_activities = token_based_replay.apply(log, net, + initial_marking, + final_marking, + parameters=parameters_tbr) + + from pm4py.algo.conformance.tokenreplay.diagnostics import root_cause_analysis + trans_root_cause = root_cause_analysis.diagnose_from_trans_fitness(log, trans_fitness, parameters=parameters) + + + from pm4py.visualization.decisiontree import visualizer as dt_vis + for trans in trans_root_cause: + clf = trans_root_cause[trans]["clf"] + feature_names = trans_root_cause[trans]["feature_names"] + classes = trans_root_cause[trans]["classes"] + + # visualization could be called + gviz = dt_vis.apply(clf, feature_names, classes) + break + + from pm4py.algo.conformance.tokenreplay.diagnostics import root_cause_analysis + act_root_cause = root_cause_analysis.diagnose_from_notexisting_activities(log, unwanted_activities, + parameters=parameters) + + from pm4py.visualization.decisiontree import visualizer as dt_vis + for act in act_root_cause: + clf = act_root_cause[act]["clf"] + feature_names = act_root_cause[act]["feature_names"] + classes = act_root_cause[act]["classes"] + # visualization could be called + gviz = dt_vis.apply(clf, feature_names, classes) + break def test_max_decomp(self): import os @@ -1015,37 +1019,38 @@ def test_evaluation(self): simp = simplicity_evaluator.apply(net) def test_sna(self): - import os - from pm4py.objects.log.importer.xes import importer as xes_importer + if importlib.util.find_spec("pyvis"): + import os + from pm4py.objects.log.importer.xes import importer as xes_importer - log = xes_importer.apply(os.path.join("input_data", "running-example.xes")) + log = xes_importer.apply(os.path.join("input_data", "running-example.xes")) - from pm4py.algo.organizational_mining.sna import algorithm as sna - hw_values = sna.apply(log, variant=sna.Variants.HANDOVER_LOG) + from pm4py.algo.organizational_mining.sna import algorithm as sna + hw_values = sna.apply(log, variant=sna.Variants.HANDOVER_LOG) - from pm4py.visualization.sna import visualizer as sna_visualizer - gviz_hw_py = sna_visualizer.apply(hw_values, variant=sna_visualizer.Variants.PYVIS) + from pm4py.visualization.sna import visualizer as sna_visualizer + gviz_hw_py = sna_visualizer.apply(hw_values, variant=sna_visualizer.Variants.PYVIS) - from pm4py.algo.organizational_mining.sna import algorithm as sna - sub_values = sna.apply(log, variant=sna.Variants.SUBCONTRACTING_LOG) + from pm4py.algo.organizational_mining.sna import algorithm as sna + sub_values = sna.apply(log, variant=sna.Variants.SUBCONTRACTING_LOG) - from pm4py.visualization.sna import visualizer as sna_visualizer - gviz_sub_py = sna_visualizer.apply(sub_values, variant=sna_visualizer.Variants.PYVIS) + from pm4py.visualization.sna import visualizer as sna_visualizer + gviz_sub_py = sna_visualizer.apply(sub_values, variant=sna_visualizer.Variants.PYVIS) - from pm4py.algo.organizational_mining.sna import algorithm as sna - wt_values = sna.apply(log, variant=sna.Variants.WORKING_TOGETHER_LOG) + from pm4py.algo.organizational_mining.sna import algorithm as sna + wt_values = sna.apply(log, variant=sna.Variants.WORKING_TOGETHER_LOG) - from pm4py.visualization.sna import visualizer as sna_visualizer - gviz_wt_py = sna_visualizer.apply(wt_values, variant=sna_visualizer.Variants.PYVIS) + from pm4py.visualization.sna import visualizer as sna_visualizer + gviz_wt_py = sna_visualizer.apply(wt_values, variant=sna_visualizer.Variants.PYVIS) - from pm4py.algo.organizational_mining.sna import algorithm as sna - ja_values = sna.apply(log, variant=sna.Variants.JOINTACTIVITIES_LOG) + from pm4py.algo.organizational_mining.sna import algorithm as sna + ja_values = sna.apply(log, variant=sna.Variants.JOINTACTIVITIES_LOG) - from pm4py.visualization.sna import visualizer as sna_visualizer - gviz_ja_py = sna_visualizer.apply(ja_values, variant=sna_visualizer.Variants.PYVIS) + from pm4py.visualization.sna import visualizer as sna_visualizer + gviz_ja_py = sna_visualizer.apply(ja_values, variant=sna_visualizer.Variants.PYVIS) - from pm4py.algo.organizational_mining.roles import algorithm as roles_discovery - roles = roles_discovery.apply(log) + from pm4py.algo.organizational_mining.roles import algorithm as roles_discovery + roles = roles_discovery.apply(log) def test_playout(self): import os diff --git a/tests/execute_tests.py b/tests/execute_tests.py index 509064700..40a35f397 100644 --- a/tests/execute_tests.py +++ b/tests/execute_tests.py @@ -8,7 +8,12 @@ parent_dir = os.path.dirname(current_dir) sys.path.insert(0, parent_dir) +EXECUTE_TESTS = True + import pm4py +import numpy +import pandas +import networkx pm4py.util.constants.SHOW_PROGRESS_BAR = False pm4py.util.constants.SHOW_EVENT_LOG_DEPRECATION = False @@ -324,8 +329,31 @@ def main(): - runner = unittest.TextTestRunner() - runner.run(suite) + if EXECUTE_TESTS: + runner = unittest.TextTestRunner() + runner.run(suite) + + print("numpy version: "+str(numpy.__version__)) + print("pandas version: "+str(pandas.__version__)) + print("networkx version: "+str(networkx.__version__)) + + if importlib.util.find_spec("scipy"): + import scipy + print("scipy version: "+str(scipy.__version__)) + + if importlib.util.find_spec("lxml"): + import lxml + print("lxml version: "+str(lxml.__version__)) + + if importlib.util.find_spec("matplotlib"): + import matplotlib + print("matplotlib version: "+str(matplotlib.__version__)) + + if importlib.util.find_spec("sklearn"): + import sklearn + print("sklearn version: "+str(sklearn.__version__)) + + print("pm4py version: "+str(pm4py.__version__)) if __name__ == "__main__": diff --git a/tests/other_tests.py b/tests/other_tests.py index 4d1b4e5fe..4310d2cb6 100644 --- a/tests/other_tests.py +++ b/tests/other_tests.py @@ -1,5 +1,6 @@ import os import unittest +import importlib.util from pm4py.algo.discovery.log_skeleton import algorithm as lsk_alg from pm4py.algo.conformance.log_skeleton import algorithm as lsk_conf_alg from pm4py.objects.process_tree.importer import importer as ptree_importer @@ -22,23 +23,25 @@ class OtherPartsTests(unittest.TestCase): def test_emd_1(self): - from pm4py.algo.evaluation.earth_mover_distance import algorithm as earth_mover_distance - M = {("a", "b", "d", "e"): 0.49, ("a", "d", "b", "e"): 0.49, ("a", "c", "d", "e"): 0.01, - ("a", "d", "c", "e"): 0.01} - L1 = {("a", "b", "d", "e"): 0.49, ("a", "d", "b", "e"): 0.49, ("a", "c", "d", "e"): 0.01, - ("a", "d", "c", "e"): 0.01} - earth_mover_distance.apply(M, L1) + if importlib.util.find_spec("pyemd"): + from pm4py.algo.evaluation.earth_mover_distance import algorithm as earth_mover_distance + M = {("a", "b", "d", "e"): 0.49, ("a", "d", "b", "e"): 0.49, ("a", "c", "d", "e"): 0.01, + ("a", "d", "c", "e"): 0.01} + L1 = {("a", "b", "d", "e"): 0.49, ("a", "d", "b", "e"): 0.49, ("a", "c", "d", "e"): 0.01, + ("a", "d", "c", "e"): 0.01} + earth_mover_distance.apply(M, L1) def test_emd_2(self): - from pm4py.algo.evaluation.earth_mover_distance import algorithm as earth_mover_distance - log = xes_importer.apply(os.path.join("input_data", "running-example.xes")) - lang_log = variants_get.get_language(log) - process_tree = inductive_miner.apply(log) - net1, im1, fm1 = process_tree_converter.apply(process_tree) - lang_model1 = variants_get.get_language( - algorithm.apply(net1, im1, fm1, variant=algorithm.Variants.STOCHASTIC_PLAYOUT, - parameters={algorithm.Variants.STOCHASTIC_PLAYOUT.value.Parameters.LOG: log})) - emd = earth_mover_distance.apply(lang_model1, lang_log) + if importlib.util.find_spec("pyemd"): + from pm4py.algo.evaluation.earth_mover_distance import algorithm as earth_mover_distance + log = xes_importer.apply(os.path.join("input_data", "running-example.xes")) + lang_log = variants_get.get_language(log) + process_tree = inductive_miner.apply(log) + net1, im1, fm1 = process_tree_converter.apply(process_tree) + lang_model1 = variants_get.get_language( + algorithm.apply(net1, im1, fm1, variant=algorithm.Variants.STOCHASTIC_PLAYOUT, + parameters={algorithm.Variants.STOCHASTIC_PLAYOUT.value.Parameters.LOG: log})) + emd = earth_mover_distance.apply(lang_model1, lang_log) def test_importing_dfg(self): dfg, sa, ea = dfg_importer.apply(os.path.join("input_data", "running-example.dfg")) diff --git a/tests/simplified_interface.py b/tests/simplified_interface.py index 6b003454d..5c0064f05 100644 --- a/tests/simplified_interface.py +++ b/tests/simplified_interface.py @@ -1,3 +1,4 @@ +import importlib.util import os import unittest @@ -200,16 +201,18 @@ def test_new_statistics_df(self): pm4py.get_case_arrival_average(df, case_id_key="CaseID", activity_key="Activity", timestamp_key="Timestamp") def test_serialization_log(self): - for legacy_obj in [True, False]: - log = pm4py.read_xes("input_data/running-example.xes", return_legacy_log_object=legacy_obj) - ser = pm4py.serialize(log) - log2 = pm4py.deserialize(ser) + if importlib.util.find_spec("pyarrow"): + for legacy_obj in [True, False]: + log = pm4py.read_xes("input_data/running-example.xes", return_legacy_log_object=legacy_obj) + ser = pm4py.serialize(log) + log2 = pm4py.deserialize(ser) def test_serialization_dataframe(self): - df = pandas_utils.read_csv("input_data/running-example.csv") - df = dataframe_utils.convert_timestamp_columns_in_df(df, timest_format=constants.DEFAULT_TIMESTAMP_PARSE_FORMAT, timest_columns=["time:timestamp"]) - ser = pm4py.serialize(df) - df2 = pm4py.deserialize(ser) + if importlib.util.find_spec("pyarrow"): + df = pandas_utils.read_csv("input_data/running-example.csv") + df = dataframe_utils.convert_timestamp_columns_in_df(df, timest_format=constants.DEFAULT_TIMESTAMP_PARSE_FORMAT, timest_columns=["time:timestamp"]) + ser = pm4py.serialize(df) + df2 = pm4py.deserialize(ser) def test_serialization_petri_net(self): net, im, fm = pm4py.read_pnml("input_data/running-example.pnml") diff --git a/tests/simplified_interface_2.py b/tests/simplified_interface_2.py index 55624c9d0..9cd52e61f 100644 --- a/tests/simplified_interface_2.py +++ b/tests/simplified_interface_2.py @@ -1,6 +1,7 @@ import pm4py import os from pm4py.util import constants, pandas_utils +import importlib.util import unittest @@ -68,11 +69,12 @@ def test_conversion_petri_to_nx(self): nx_digraph = pm4py.convert_petri_net_to_networkx(net, im, fm) def test_stochastic_language(self): - log1 = pm4py.read_xes("input_data/running-example.xes") - log2 = pm4py.read_xes("input_data/running-example.xes", return_legacy_log_object=True) - lang1 = pm4py.get_stochastic_language(log1) - lang2 = pm4py.get_stochastic_language(log2) - pm4py.compute_emd(lang1, lang2) + if importlib.util.find_spec("pyemd"): + log1 = pm4py.read_xes("input_data/running-example.xes") + log2 = pm4py.read_xes("input_data/running-example.xes", return_legacy_log_object=True) + lang1 = pm4py.get_stochastic_language(log1) + lang2 = pm4py.get_stochastic_language(log2) + pm4py.compute_emd(lang1, lang2) def test_hybrid_ilp_miner(self): for legacy_obj in [True, False]: diff --git a/tests/sna_test.py b/tests/sna_test.py index 1fd6d4210..96e12006c 100644 --- a/tests/sna_test.py +++ b/tests/sna_test.py @@ -1,5 +1,6 @@ import os import unittest +import importlib.util from pm4py.algo.organizational_mining.sna import algorithm as sna_alg, util as sna_util, util from pm4py.objects.log.importer.xes import importer as xes_importer @@ -38,12 +39,13 @@ def test_log_orgmining_local_attr(self): algorithm.apply_from_group_attribute(log) def test_log_orgmining_local_clustering(self): - from pm4py.algo.organizational_mining.local_diagnostics import algorithm - from pm4py.algo.organizational_mining.sna.variants.log import jointactivities - log = xes_importer.apply(os.path.join("input_data", "receipt.xes")) - ja = jointactivities.apply(log) - clustering = util.cluster_affinity_propagation(ja) - algorithm.apply_from_clustering_or_roles(log, clustering) + if importlib.util.find_spec("sklearn"): + from pm4py.algo.organizational_mining.local_diagnostics import algorithm + from pm4py.algo.organizational_mining.sna.variants.log import jointactivities + log = xes_importer.apply(os.path.join("input_data", "receipt.xes")) + ja = jointactivities.apply(log) + clustering = util.cluster_affinity_propagation(ja) + algorithm.apply_from_clustering_or_roles(log, clustering) def test_log_orgmining_local_roles(self): from pm4py.algo.organizational_mining.local_diagnostics import algorithm @@ -53,9 +55,10 @@ def test_log_orgmining_local_roles(self): algorithm.apply_from_clustering_or_roles(log, roles) def test_sna_clustering(self): - log = xes_importer.apply(os.path.join("..", "tests", "input_data", "running-example.xes")) - hw_values = sna_alg.apply(log, variant=sna_alg.Variants.HANDOVER_LOG) - clusters = sna_util.cluster_affinity_propagation(hw_values) + if importlib.util.find_spec("sklearn"): + log = xes_importer.apply(os.path.join("..", "tests", "input_data", "running-example.xes")) + hw_values = sna_alg.apply(log, variant=sna_alg.Variants.HANDOVER_LOG) + clusters = sna_util.cluster_affinity_propagation(hw_values) def test_res_profiles_log(self): from pm4py.algo.organizational_mining.resource_profiles import algorithm diff --git a/tests/xes_impexp_test.py b/tests/xes_impexp_test.py index df08db2b8..64785e6c6 100644 --- a/tests/xes_impexp_test.py +++ b/tests/xes_impexp_test.py @@ -3,6 +3,7 @@ from tests.constants import INPUT_DATA_DIR, OUTPUT_DATA_DIR, PROBLEMATIC_XES_DIR, COMPRESSED_INPUT_DATA import logging import unittest +import importlib.util import os @@ -52,12 +53,14 @@ def test_importXESfromGZIP_imp2(self): del log def test_rustxes_xes_import(self): - log = xes_importer.apply(os.path.join(INPUT_DATA_DIR, "receipt.xes"), variant=xes_importer.Variants.RUSTXES) - self.assertEqual(len(log), 1434) + if importlib.util.find_spec("rustxes"): + log = xes_importer.apply(os.path.join(INPUT_DATA_DIR, "receipt.xes"), variant=xes_importer.Variants.RUSTXES) + self.assertEqual(len(log), 1434) def test_rustxes_xesgz_import(self): - log = xes_importer.apply(os.path.join(INPUT_DATA_DIR, "bpic2012.xes.gz"), variant=xes_importer.Variants.RUSTXES) - self.assertEqual(len(log), 13087) + if importlib.util.find_spec("rustxes"): + log = xes_importer.apply(os.path.join(INPUT_DATA_DIR, "bpic2012.xes.gz"), variant=xes_importer.Variants.RUSTXES) + self.assertEqual(len(log), 13087) if __name__ == "__main__":