pm4py.streaming.algo.conformance.tbr.variants package#

PM4Py – A Process Mining Library for Python

This program is free software: you can redistribute it and/or modify it under the terms of the GNU Affero General Public License as published by the Free Software Foundation, either version 3 of the License, or any later version.

This program 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 Affero General Public License for more details.

You should have received a copy of the GNU Affero General Public License along with this program. If not, see this software project’s root or visit <https://www.gnu.org/licenses/>.

Website: https://processintelligence.solutions Contact: info@processintelligence.solutions

Submodules#

pm4py.streaming.algo.conformance.tbr.variants.classic module#

PM4Py – A Process Mining Library for Python

You should have received a copy of the GNU Affero General Public License along with this program. If not, see this software project’s root or visit <https://www.gnu.org/licenses/>.

Website: https://processintelligence.solutions Contact: info@processintelligence.solutions

class pm4py.streaming.algo.conformance.tbr.variants.classic.Parameters[source]#

Bases: object

DICT_VARIANT = 'dict_variant'#

DICT_ID = 'dict_id'#

CASE_DICT_ID = 'case_dict_id'#

MISSING_DICT_ID = 'missing_dict_id'#

REMAINING_DICT_ID = 'remaining_dict_id'#

CASE_ID_KEY = 'pm4py:param:case_id_key'#

ACTIVITY_KEY = 'pm4py:param:activity_key'#

MAXIMUM_ITERATIONS_INVISIBLES = 'maximum_iterations_invisibles'#

class pm4py.streaming.algo.conformance.tbr.variants.classic.TbrStreamingConformance(net, im, fm, parameters=None)[source]#

Bases: StreamingAlgorithm

build_dictionaries(parameters)[source]#

Builds the dictionaries needed to store the information during the replay

Parameters#

parameters

Parameters:

Parameters.DICT_VARIANT: type of dictionary to use
Parameters.CASE_DICT_ID: identifier of the dictionary hosting the markings (0)
Parameters.MISSING_DICT_ID: identifier of the dictionary hosting the missing tokens (1)
Parameters.REMAINING_DICT_ID: identifier of the dictionary hosting the remaining tokens (2)

get_paths_net()[source]#

Gets the dictionary of shortest paths using invisibles transitions

Returns#

dictio_spaths: Dictionary of shortest paths

encode_str(stru)[source]#: Encodes a string for storage in generic dictionaries

encode_marking(mark)[source]#: Encodes a marking for storage in generic dictionaries

decode_marking(ems)[source]#: Decodes a marking from a generic dictionary to a Marking object

verify_tbr(case, activity)[source]#

Verifies an activity happening in a case

Parameters#

case: Case
activity: Activity

enable_trans_with_invisibles(marking, activity)[source]#

Enables a visible transition (that is not enabled) through invisible transitions

Parameters#

marking: Marking
activity: Activity to enable

Returns#

new_marking: New marking (where the transition CAN be enabled)

get_status(case)[source]#

Gets the status of an open case

Parameters#

case: Case

terminate(case)[source]#

Terminate a case, checking if the final marking is reached

Parameters#

case: Case ID

Returns#

dictio: Dictionary containing: the marking, the count of missing and remaining tokens

terminate_all()[source]#: Terminate all open cases

reach_fm_with_invisibles(marking)[source]#

Reaches the final marking using invisible transitions

Parameters#

marking: Marking

Returns#

new_marking: New marking (hopely equal to the final marking)

message_case_or_activity_not_in_event(event)[source]#: Sends a message if the case or the activity are not there in the event

message_activity_not_possible(activity, case)[source]#

Sends a message if the activity is not possible according to the model

Parameters#

activity: Activity
case: Case

message_missing_tokens(activity, case)[source]#

Sends a message if the insertion of missing tokens occur

Parameters#

activity: Activity
case: Case

message_case_not_in_dictionary(case)[source]#

Sends a message if the provided case is not in the dictionary

Parameters#

activity: Activity
case: Case

message_final_marking_not_reached(case, marking)[source]#

Sends a message if the final marking could not be reached for the current case

Parameters#

case: Case

pm4py.streaming.algo.conformance.tbr.variants.classic.apply(net, im, fm, parameters=None)[source]#

Method that creates the TbrStreamingConformance object

Parameters#

net: Petri net
im: Initial marking
fm: Final marking
parameters: Parameters of the algorithm

Returns#

conf_stream_obj: Conformance streaming object