pm4py.streaming.algo.conformance.temporal.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.temporal.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.temporal.variants.classic.Parameters(value, names=<not given>, *values, module=None, qualname=None, type=None, start=1, boundary=None)[source]#

Bases: Enum

ACTIVITY_KEY = 'pm4py:param:activity_key'#

START_TIMESTAMP_KEY = 'pm4py:param:start_timestamp_key'#

TIMESTAMP_KEY = 'pm4py:param:timestamp_key'#

CASE_ID_KEY = 'pm4py:param:case_id_key'#

ZETA = 'zeta'#

DICT_VARIANT = 'dict_variant'#

DICT_ID = 'dict_id'#

CASE_DICT_ID = 'case_dict_id'#

DEV_DICT_ID = 'dev_dict_id'#

class pm4py.streaming.algo.conformance.temporal.variants.classic.TemporalProfileStreamingConformance(temporal_profile: Dict[Tuple[str, str], Tuple[float, float]], parameters: Dict[Any, Any] | None = None)[source]#

Bases: StreamingAlgorithm

check_conformance(event: Tuple[str, float, float, str])[source]#

Checks the conformance according to the temporal profile

Parameters#

event: Event

message_event_is_not_complete(event: Event)[source]#

Method that is called when the event does not contain the case, or the activity, or the timestamp

Parameters#

event: Incoming event

message_deviation(dev_descr: Tuple[str, str, str, float, float])[source]#

Method that is called to signal a deviation according to the temporal profile

Parameters#

dev_descr: Description of the deviation to be printed

pm4py.streaming.algo.conformance.temporal.variants.classic.apply(temporal_profile: Dict[Tuple[str, str], Tuple[float, float]], parameters: Dict[Any, Any] | None = None)[source]#

Initialize the streaming conformance checking.

Implements the approach described in: Stertz, Florian, Jürgen Mangler, and Stefanie Rinderle-Ma. “Temporal Conformance Checking at Runtime based on Time-infused Process Models.” arXiv preprint arXiv:2008.07262 (2020).

Parameters#

temporal_profile

Temporal profile

parameters

Parameters of the algorithm, including:

Parameters.ACTIVITY_KEY => the attribute to use as activity
Parameters.START_TIMESTAMP_KEY => the attribute to use as start timestamp
Parameters.TIMESTAMP_KEY => the attribute to use as timestamp
Parameters.ZETA => multiplier for the standard deviation
Parameters.CASE_ID_KEY => column to use as case identifier
Parameters.DICT_VARIANT => the variant of dictionary to use
Parameters.CASE_DICT_ID => the identifier of the case dictionary
Parameters.DEV_DICT_ID => the identifier of the deviations dictionary