ocpa.algo.enhancement.token_replay_based_performance package

Subpackages

Submodules

ocpa.algo.enhancement.token_replay_based_performance.algorithm module

ocpa.algo.enhancement.token_replay_based_performance.algorithm.apply(ocpn: ocpa.objects.oc_petri_net.obj.ObjectCentricPetriNet, ocel: ocpa.objects.log.ocel.OCEL, variant='opera', parameters=None)

Computing object-centric performance measures using object-centric event logs. It is model-based approach where we 1) discover an object-centric Petri net, 2) replay an object-centric event log, and 3) compute object-centric performance measures based on the replay history. It provides object-centric performance measures such as lagging, pooling, synchronization, and flow times, along with traditional performance measures such as waiting, service, and sojourn times.

Parameters

  • ocpn (OCPN) – Object-centric Petri net

  • ocel (OCEL) – Object-centric event log

  • measures (List) – A list of performance measures to analyze. ‘act_freq’: Activity Frequency, ‘arc_freq’: Arc Frequency, ‘object_count’: Object Count, ‘waiting_time’: Waiting Time, ‘service_time’: Service Time, ‘sojourn_time’: Sojourn Time, ‘synchronization_time’: Synchronization Time, ‘pooling_time’: Pooling Time, ‘lagging_time’: Lagging Time, ‘flow_time’: Flow Time.

  • agg (List) – A list of aggregations. ‘mean’: Mean, ‘median’: Median, ‘stdev’: Standard Deviation, ‘sum’: Summation, ‘min’: Minimum, ‘max’: Maximum

Returns

performance measures per activity, e.g., {‘Act1’: {‘Measure1’: value, ‘Measure2: value, …}, ‘Act2’: {…}, …}. Value has different formats depending on the performance measure, e.g., the one for ‘act_freq’ is Integer, the one for ‘object_count’, ‘pooling_time’, ‘lagging_time’, and ‘synchronization_time’ is a nested dictionary (e.g., {‘ObjectType1’: {‘mean’: Real, ‘median’: Real, …}, ‘ObjectType2’: {…}, …}), the one for ‘waiting_time’, ‘service_time’, ‘sojourn_time’, and ‘flow_time’ is a dictionary (e.g., {‘mean’: Real, ‘median’: Real, …}).

Return type

Dict

ocpa.algo.enhancement.token_replay_based_performance.util module

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/>.

class ocpa.algo.enhancement.token_replay_based_performance.util.ApplyTraceTokenReplay(trace, net, initial_marking, final_marking, trans_map, enable_pltr_fitness, place_fitness, transition_fitness, notexisting_activities_in_model, places_shortest_path_by_hidden, consider_remaining_in_fitness, activity_key='concept:name', reach_mark_through_hidden=True, stop_immediately_when_unfit=False, walk_through_hidden_trans=True, post_fix_caching=None, marking_to_activity_caching=None, is_reduction=False, thread_maximum_ex_time=10, cleaning_token_flood=False, s_components=None, trace_occurrences=1, consider_activities_not_in_model_in_fitness=False)

Bases: object

run()

Runs the thread and stores the results

class ocpa.algo.enhancement.token_replay_based_performance.util.DebugConst

Bases: object

REACH_ITF1 = -1
REACH_ITF2 = -1
REACH_MRH = -1
class ocpa.algo.enhancement.token_replay_based_performance.util.MarkingToActivityCaching

Bases: object

Marking to activity caching

exception ocpa.algo.enhancement.token_replay_based_performance.util.NoConceptNameException(message)

Bases: Exception

class ocpa.algo.enhancement.token_replay_based_performance.util.Parameters(value)

Bases: enum.Enum

An enumeration.

ACTIVITY_KEY = 'pm4py:param:activity_key'
CASE_ID_KEY = 'pm4py:param:case_id_key'
CLEANING_TOKEN_FLOOD = 'cleaning_token_flood'
CONSIDER_ACTIVITIES_NOT_IN_MODEL_IN_FITNESS = 'consider_activities_not_in_model_in_fitness'
CONSIDER_REMAINING_IN_FITNESS = 'consider_remaining_in_fitness'
DISABLE_VARIANTS = 'disable_variants'
ENABLE_PLTR_FITNESS = 'enable_pltr_fitness'
IS_REDUCTION = 'is_reduction'
PARAMETER_VARIANT_DELIMITER = 'variant_delimiter'
PLACES_SHORTEST_PATH_BY_HIDDEN = 'places_shortest_path_by_hidden'
RETURN_NAMES = 'return_names'
SHOW_PROGRESS_BAR = 'show_progress_bar'
STOP_IMMEDIATELY_UNFIT = 'stop_immediately_unfit'
THREAD_MAX_EX_TIME = 'thread_maximum_ex_time'
TRY_TO_REACH_FINAL_MARKING_THROUGH_HIDDEN = 'try_to_reach_final_marking_through_hidden'
VARIANTS = 'variants'
WALK_THROUGH_HIDDEN_TRANS = 'walk_through_hidden_trans'
class ocpa.algo.enhancement.token_replay_based_performance.util.PostFixCaching

Bases: object

Post fix caching object

class ocpa.algo.enhancement.token_replay_based_performance.util.TechnicalParameters(value)

Bases: enum.Enum

An enumeration.

ENABLE_MARKTOACT_CACHE = False
ENABLE_POSTFIX_CACHE = False
MAX_DEF_THR_EX_TIME = 10
MAX_IT_FINAL1 = 5
MAX_IT_FINAL2 = 5
MAX_NO_THREADS = 1024
MAX_POSTFIX_SUFFIX_LENGTH = 20
MAX_REC_DEPTH = 50
MAX_REC_DEPTH_HIDTRANSENABL = 2
ocpa.algo.enhancement.token_replay_based_performance.util.add_missing_tokens(t, marking)

Adds missing tokens needed to activate a transition

Parameters

  • t – Transition that should be enabled

  • marking – Current marking

ocpa.algo.enhancement.token_replay_based_performance.util.apply_hidden_trans(t, net, marking, places_shortest_paths_by_hidden, act_tr, rec_depth, visit_trans, vis_mark, trace_id, timestamp)

Apply hidden transitions in order to enable a given transition

Parameters

  • t – Transition to eventually enable

  • net – Petri net

  • marking – Marking

  • places_shortest_paths_by_hidden – Shortest paths between places connected by hidden transitions

  • act_tr – All activated transitions

  • rec_depth – Current recursion depth

  • visit_trans – All visited transitions by hiddenTrans method

  • vis_mark – All visited markings

ocpa.algo.enhancement.token_replay_based_performance.util.apply_trace(trace, net, initial_marking, final_marking, trans_map, enable_pltr_fitness, place_fitness, transition_fitness, notexisting_activities_in_model, places_shortest_path_by_hidden, consider_remaining_in_fitness, activity_key='concept:name', try_to_reach_final_marking_through_hidden=True, stop_immediately_unfit=False, walk_through_hidden_trans=True, post_fix_caching=None, marking_to_activity_caching=None, is_reduction=False, thread_maximum_ex_time=10, enable_postfix_cache=False, enable_marktoact_cache=False, cleaning_token_flood=False, s_components=None, trace_occurrences=1, consider_activities_not_in_model_in_fitness=False)

Apply the token replaying algorithm to a trace

Parameters

  • trace – Trace in the event log

  • net – Petri net

  • initial_marking – Initial marking

  • final_marking – Final marking

  • trans_map – Map between transitions labels and transitions

  • enable_pltr_fitness – Enable fitness retrieval at place/transition level

  • place_fitness – Current dictionary of places associated with unfit traces

  • transition_fitness – Current dictionary of transitions associated with unfit traces

  • notexisting_activities_in_model – Map that stores the notexisting activities in the model

  • places_shortest_path_by_hidden – Shortest paths between places by hidden transitions

  • consider_remaining_in_fitness – Boolean value telling if the remaining tokens should be considered in fitness evaluation

  • activity_key – Name of the attribute that contains the activity

  • try_to_reach_final_marking_through_hidden – Boolean value that decides if we shall try to reach the final marking through hidden transitions

  • stop_immediately_unfit – Boolean value that decides if we shall stop immediately when a non-conformance is detected

  • walk_through_hidden_trans – Boolean value that decides if we shall walk through hidden transitions in order to enable visible transitions

  • post_fix_caching – Stores the post fix caching object

  • marking_to_activity_caching – Stores the marking-to-activity cache

  • is_reduction – Expresses if the token-based replay is called in a reduction attempt

  • thread_maximum_ex_time – Alignment threads maximum allowed execution time

  • enable_postfix_cache – Enables postfix cache

  • enable_marktoact_cache – Enables marking to activity cache

  • cleaning_token_flood – Decides if a cleaning of the token flood shall be operated

  • s_components – S-components of the Petri net (if workflow net)

ocpa.algo.enhancement.token_replay_based_performance.util.apply_traces(log, net, initial_marking, final_marking, enable_pltr_fitness=False, consider_remaining_in_fitness=False, activity_key='concept:name', reach_mark_through_hidden=True, stop_immediately_unfit=False, walk_through_hidden_trans=True, places_shortest_path_by_hidden=None, is_reduction=False, thread_maximum_ex_time=10, cleaning_token_flood=False, show_progress_bar=True, consider_activities_not_in_model_in_fitness=False)

Apply token-based replay to a log

Parameters

  • log – Trace log

  • net – Petri net

  • initial_marking – Initial marking

  • final_marking – Final marking

  • enable_pltr_fitness – Enable fitness retrieval at place level

  • consider_remaining_in_fitness – Boolean value telling if the remaining tokens should be considered in fitness evaluation

  • activity_key – Name of the attribute that contains the activity

  • reach_mark_through_hidden – Boolean value that decides if we shall try to reach the final marking through hidden transitions

  • stop_immediately_unfit – Boolean value that decides if we shall stop immediately when a non-conformance is detected

  • walk_through_hidden_trans – Boolean value that decides if we shall walk through hidden transitions in order to enable visible transitions

  • places_shortest_path_by_hidden – Shortest paths between places by hidden transitions

  • variants – List of variants contained in the event log

  • is_reduction – Expresses if the token-based replay is called in a reduction attempt

  • thread_maximum_ex_time – Alignment threads maximum allowed execution time

  • cleaning_token_flood – Decides if a cleaning of the token flood shall be operated

  • disable_variants – Disable variants grouping

  • return_object_names – Decides whether names instead of object pointers shall be returned

ocpa.algo.enhancement.token_replay_based_performance.util.apply_variants_dictionary(variants, net, initial_marking, final_marking, parameters=None)
ocpa.algo.enhancement.token_replay_based_performance.util.apply_variants_list(variants_list, net, initial_marking, final_marking, parameters=None)
ocpa.algo.enhancement.token_replay_based_performance.util.apply_variants_list_petri_string(variants_list, petri_string, parameters=None)
ocpa.algo.enhancement.token_replay_based_performance.util.apply_variants_list_petri_string_multiprocessing(output, variants_list, petri_string, parameters=None)
ocpa.algo.enhancement.token_replay_based_performance.util.break_condition_final_marking(marking, final_marking)

Verify break condition for final marking

Parameters

  • marking – Current marking

  • final_marking – Target final marking

ocpa.algo.enhancement.token_replay_based_performance.util.enable_hidden_transitions(net, marking, activated_transitions, visited_transitions, all_visited_markings, hidden_transitions_to_enable, t, trace_id, timestamp)

Actually enable hidden transitions on the Petri net

Parameters

  • net – Petri net

  • marking – Current marking

  • activated_transitions – All activated transitions during the replay

  • visited_transitions – All visited transitions by the recursion

  • all_visited_markings – All visited markings

  • hidden_transitions_to_enable – List of hidden transition to enable

  • t – Transition against we should check if they are enabled

ocpa.algo.enhancement.token_replay_based_performance.util.execute_and_record(t, pn, m, token_visits, trace_id, prev_timestamp, timestamp)
ocpa.algo.enhancement.token_replay_based_performance.util.get_consumed_tokens(t)

Get tokens consumed firing a transition

Parameters

t – Transition that should be enabled

ocpa.algo.enhancement.token_replay_based_performance.util.get_diagnostics_dataframe(log: pm4py.objects.log.obj.EventLog, tbr_output: List[Dict[str, Any]], parameters: Optional[Dict[Union[str, ocpa.algo.enhancement.token_replay_based_performance.util.Parameters], Any]] = None) → pandas.DataFrame

Gets the results of token-based replay in a dataframe

Parameters

  • log – Event log

  • tbr_output – Output of the token-based replay technique

Returns

Diagnostics dataframe

Return type

dataframe

ocpa.algo.enhancement.token_replay_based_performance.util.get_hidden_transitions_to_enable(marking, places_with_missing, places_shortest_path_by_hidden)

Calculate an ordered list of transitions to visit in order to enable a given transition

Parameters

  • marking – Current marking

  • places_with_missing – List of places with missing tokens

  • places_shortest_path_by_hidden – Minimal connection between places by hidden transitions

ocpa.algo.enhancement.token_replay_based_performance.util.get_places_with_missing_tokens(t, marking)

Get places with missing tokens

Parameters

  • t – Transition to enable

  • marking – Current marking

ocpa.algo.enhancement.token_replay_based_performance.util.get_produced_tokens(t)

Get tokens produced firing a transition

Parameters

t – Transition that should be enabled

ocpa.algo.enhancement.token_replay_based_performance.util.get_req_transitions_for_final_marking(marking, final_marking, places_shortest_path_by_hidden)

Gets required transitions for final marking

Parameters

  • marking – Current marking

  • final_marking – Final marking assigned to the Petri net

  • places_shortest_path_by_hidden – Minimal connection between places by hidden transitions

ocpa.algo.enhancement.token_replay_based_performance.util.get_variant_from_trace(trace, activity_key, disable_variants=False)

Gets the variant from the trace (allow disabling)

Parameters

  • trace – Trace

  • activity_key – Attribute that is the activity

  • disable_variants – Boolean value that disable variants

Returns

Variant describing the trace

Return type

variant

ocpa.algo.enhancement.token_replay_based_performance.util.get_variants_from_log(log, activity_key, disable_variants=False)

Gets the variants from the log (allow disabling by giving each trace a different variant)

Parameters

  • log – Trace log

  • activity_key – Attribute that is the activity

  • disable_variants – Boolean value that disable variants

Returns

Variants contained in the log

Return type

variants

ocpa.algo.enhancement.token_replay_based_performance.util.merge_dicts(x, y)

Merge two dictionaries keeping the least value

Parameters

  • x – First map (string, integer)

  • y – Second map (string, integer)

ocpa.algo.enhancement.token_replay_based_performance.util.run_timed_replay(log: pm4py.objects.log.obj.EventLog, net: pm4py.objects.petri_net.obj.PetriNet, initial_marking: pm4py.objects.petri_net.obj.Marking, final_marking: pm4py.objects.petri_net.obj.Marking, parameters: Optional[Dict[Union[str, ocpa.algo.enhancement.token_replay_based_performance.util.Parameters], Any]] = None) → List[Dict[str, Any]]

Method to apply token-based replay

Parameters

  • log – Log

  • net – Petri net

  • initial_marking – Initial marking

  • final_marking – Final marking

  • parameters – Parameters of the algorithm

ocpa.algo.enhancement.token_replay_based_performance.util.single_element_statistics(log, net, initial_marking, replay_results, activity_key='concept:name', ht_perf_method='last', parameters=None)

Get single Petrinet element statistics

Parameters

  • log – Log

  • net – Petri net

  • initial_marking – Initial marking

  • aligned_traces – Result of the token-based replay

  • variants_idx – Variants along with indexes of belonging traces

  • activity_key – Activity key (must be specified if different from concept:name)

  • timestamp_key – Timestamp key (must be specified if different from time:timestamp)

  • ht_perf_method – Method to use in order to annotate hidden transitions (performance value could be put on the last possible point (last) or in the first possible point (first)

  • parameters – Possible parameters of the algorithm

Returns

Petri net element statistics (frequency, unaggregated performance)

Return type

statistics

Module contents