Source code for pm4py.evaluation.replay_fitness.variants.token_replay

'''
    This file is part of PM4Py (More Info: https://pm4py.fit.fraunhofer.de).

    PM4Py is free software: you can redistribute it and/or modify
    it under the terms of the GNU General Public License as published by
    the Free Software Foundation, either version 3 of the License, or
    (at your option) any later version.

    PM4Py is distributed in the hope that it will be useful,
    but WITHOUT ANY WARRANTY; without even the implied warranty of
    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
    GNU General Public License for more details.

    You should have received a copy of the GNU General Public License
    along with PM4Py.  If not, see <https://www.gnu.org/licenses/>.
'''
from pm4py.algo.conformance.tokenreplay import algorithm as executor
from pm4py.algo.conformance.tokenreplay.variants import token_replay
from pm4py.evaluation.replay_fitness.parameters import Parameters
from pm4py.util import exec_utils
from pm4py.util.xes_constants import DEFAULT_NAME_KEY


[docs]def evaluate(aligned_traces, parameters=None): """ Gets a dictionary expressing fitness in a synthetic way from the list of boolean values saying if a trace in the log is fit, and the float values of fitness associated to each trace Parameters ------------ aligned_traces Result of the token-based replayer parameters Possible parameters of the evaluation Returns ----------- dictionary Containing two keys (percFitTraces and averageFitness) """ if parameters is None: parameters = {} no_traces = len(aligned_traces) fit_traces = len([x for x in aligned_traces if x["trace_is_fit"]]) sum_of_fitness = sum([x["trace_fitness"] for x in aligned_traces]) perc_fit_traces = 0.0 average_fitness = 0.0 log_fitness = 0 total_m = sum([x["missing_tokens"] for x in aligned_traces]) total_c = sum([x["consumed_tokens"] for x in aligned_traces]) total_r = sum([x["remaining_tokens"] for x in aligned_traces]) total_p = sum([x["produced_tokens"] for x in aligned_traces]) if no_traces > 0: perc_fit_traces = float(100.0 * fit_traces) / float(no_traces) average_fitness = float(sum_of_fitness) / float(no_traces) if total_c > 0 and total_p > 0: log_fitness = 0.5 * (1 - total_m / total_c) + 0.5 * (1 - total_r / total_p) return {"perc_fit_traces": perc_fit_traces, "average_trace_fitness": average_fitness, "log_fitness": log_fitness, "percentage_of_fitting_traces": perc_fit_traces }
[docs]def apply(log, petri_net, initial_marking, final_marking, parameters=None): """ Apply token replay fitness evaluation Parameters ----------- log Trace log petri_net Petri net initial_marking Initial marking final_marking Final marking parameters Parameters Returns ----------- dictionary Containing two keys (percFitTraces and averageFitness) """ if parameters is None: parameters = {} activity_key = exec_utils.get_param_value(Parameters.ACTIVITY_KEY, parameters, DEFAULT_NAME_KEY) token_replay_variant = exec_utils.get_param_value(Parameters.TOKEN_REPLAY_VARIANT, parameters, executor.Variants.TOKEN_REPLAY) cleaning_token_flood = exec_utils.get_param_value(Parameters.CLEANING_TOKEN_FLOOD, parameters, False) remaining_in_fitness = exec_utils.get_param_value(token_replay.Parameters.CONSIDER_REMAINING_IN_FITNESS, parameters, True) parameters_tr = {token_replay.Parameters.ACTIVITY_KEY: activity_key, token_replay.Parameters.CONSIDER_REMAINING_IN_FITNESS: remaining_in_fitness, token_replay.Parameters.CLEANING_TOKEN_FLOOD: cleaning_token_flood} aligned_traces = executor.apply(log, petri_net, initial_marking, final_marking, variant=token_replay_variant, parameters=parameters_tr) return evaluate(aligned_traces)