0
# Conformance Checking and Fitness
1
2
Comprehensive conformance checking methods for measuring how well process models align with event logs. PM4PY provides token-based replay, alignments-based methods, and specialized techniques for different model types and conformance dimensions.
3
4
## Capabilities
5
6
### Token-Based Replay Methods
7
8
Token-based replay simulates the execution of event log traces on process models to measure fitness and detect deviations.
9
10
```python { .api }
11
def conformance_diagnostics_token_based_replay(log, petri_net, initial_marking, final_marking, activity_key='concept:name', timestamp_key='time:timestamp', case_id_key='case:concept:name', return_diagnostics_dataframe=False, opt_parameters=None):
12
"""
13
Full token-based replay diagnostics with detailed trace analysis.
14
15
Parameters:
16
- log (Union[EventLog, pd.DataFrame]): Event log data
17
- petri_net (PetriNet): Petri net model
18
- initial_marking (Marking): Initial marking
19
- final_marking (Marking): Final marking
20
- activity_key (str): Activity attribute name
21
- timestamp_key (str): Timestamp attribute name
22
- case_id_key (str): Case ID attribute name
23
- return_diagnostics_dataframe (bool): Return results as DataFrame
24
- opt_parameters (Optional[Dict]): Additional parameters
25
26
Returns:
27
List[Dict[str, Any]]: Detailed diagnostics per trace
28
"""
29
30
def fitness_token_based_replay(log, petri_net, initial_marking, final_marking, activity_key='concept:name', timestamp_key='time:timestamp', case_id_key='case:concept:name'):
31
"""
32
Calculate fitness using token-based replay.
33
Fast method for measuring model-log fitness.
34
35
Parameters:
36
- log (Union[EventLog, pd.DataFrame]): Event log data
37
- petri_net (PetriNet): Petri net model
38
- initial_marking (Marking): Initial marking
39
- final_marking (Marking): Final marking
40
- activity_key (str): Activity attribute name
41
- timestamp_key (str): Timestamp attribute name
42
- case_id_key (str): Case ID attribute name
43
44
Returns:
45
Dict[str, float]: Fitness metrics including log_fitness, average_trace_fitness
46
"""
47
48
def precision_token_based_replay(log, petri_net, initial_marking, final_marking, activity_key='concept:name', timestamp_key='time:timestamp', case_id_key='case:concept:name'):
49
"""
50
Calculate precision using token-based replay.
51
52
Parameters:
53
- log (Union[EventLog, pd.DataFrame]): Event log data
54
- petri_net (PetriNet): Petri net model
55
- initial_marking (Marking): Initial marking
56
- final_marking (Marking): Final marking
57
- activity_key (str): Activity attribute name
58
- timestamp_key (str): Timestamp attribute name
59
- case_id_key (str): Case ID attribute name
60
61
Returns:
62
float: Precision value between 0 and 1
63
"""
64
```
65
66
### Alignments-Based Methods
67
68
Alignments provide optimal mappings between log traces and model executions, offering detailed diagnostics and accurate conformance measurements.
69
70
```python { .api }
71
def conformance_diagnostics_alignments(log, *args, multi_processing=True, activity_key='concept:name', timestamp_key='time:timestamp', case_id_key='case:concept:name', variant_str=None, return_diagnostics_dataframe=False, **kwargs):
72
"""
73
Full alignment diagnostics with optimal trace-model mappings.
74
Most accurate method for conformance analysis.
75
76
Parameters:
77
- log (Union[EventLog, pd.DataFrame]): Event log data
78
- *args: Model components (petri_net, initial_marking, final_marking)
79
- multi_processing (bool): Enable parallel processing
80
- activity_key (str): Activity attribute name
81
- timestamp_key (str): Timestamp attribute name
82
- case_id_key (str): Case ID attribute name
83
- variant_str (Optional[str]): Algorithm variant
84
- return_diagnostics_dataframe (bool): Return results as DataFrame
85
- **kwargs: Additional parameters
86
87
Returns:
88
List[Dict[str, Any]]: Detailed alignment diagnostics per trace
89
"""
90
91
def fitness_alignments(log, petri_net, initial_marking, final_marking, multi_processing=True, activity_key='concept:name', timestamp_key='time:timestamp', case_id_key='case:concept:name', variant_str=None):
92
"""
93
Calculate fitness using optimal alignments.
94
Gold standard for fitness measurement.
95
96
Parameters:
97
- log (Union[EventLog, pd.DataFrame]): Event log data
98
- petri_net (PetriNet): Petri net model
99
- initial_marking (Marking): Initial marking
100
- final_marking (Marking): Final marking
101
- multi_processing (bool): Enable parallel processing
102
- activity_key (str): Activity attribute name
103
- timestamp_key (str): Timestamp attribute name
104
- case_id_key (str): Case ID attribute name
105
- variant_str (Optional[str]): Algorithm variant
106
107
Returns:
108
Dict[str, float]: Fitness metrics with detailed breakdown
109
"""
110
111
def precision_alignments(log, petri_net, initial_marking, final_marking, multi_processing=True, activity_key='concept:name', timestamp_key='time:timestamp', case_id_key='case:concept:name'):
112
"""
113
Calculate precision using alignments.
114
115
Parameters:
116
- log (Union[EventLog, pd.DataFrame]): Event log data
117
- petri_net (PetriNet): Petri net model
118
- initial_marking (Marking): Initial marking
119
- final_marking (Marking): Final marking
120
- multi_processing (bool): Enable parallel processing
121
- activity_key (str): Activity attribute name
122
- timestamp_key (str): Timestamp attribute name
123
- case_id_key (str): Case ID attribute name
124
125
Returns:
126
float: Precision value between 0 and 1
127
"""
128
```
129
130
### Footprints-Based Methods (Deprecated)
131
132
Legacy footprints-based conformance checking methods (deprecated in version 2.3.0).
133
134
```python { .api }
135
def conformance_diagnostics_footprints(*args):
136
"""
137
Footprints-based conformance diagnostics (deprecated in 2.3.0).
138
139
Parameters:
140
- *args: Log and model arguments
141
142
Returns:
143
Union[List[Dict[str, Any]], Dict[str, Any]]: Conformance results
144
"""
145
146
def fitness_footprints(*args):
147
"""
148
Footprints-based fitness calculation (deprecated in 2.3.0).
149
150
Parameters:
151
- *args: Log and model arguments
152
153
Returns:
154
Dict[str, float]: Fitness metrics
155
"""
156
157
def precision_footprints(*args):
158
"""
159
Footprints-based precision calculation (deprecated in 2.3.0).
160
161
Parameters:
162
- *args: Log and model arguments
163
164
Returns:
165
float: Precision value
166
"""
167
```
168
169
### Advanced Conformance Methods
170
171
Specialized conformance checking methods for different model types and quality dimensions.
172
173
```python { .api }
174
def generalization_tbr(log, petri_net, initial_marking, final_marking, activity_key='concept:name', timestamp_key='time:timestamp', case_id_key='case:concept:name'):
175
"""
176
Calculate generalization using token-based replay.
177
Measures model's ability to handle unseen behavior.
178
179
Parameters:
180
- log (Union[EventLog, pd.DataFrame]): Event log data
181
- petri_net (PetriNet): Petri net model
182
- initial_marking (Marking): Initial marking
183
- final_marking (Marking): Final marking
184
- activity_key (str): Activity attribute name
185
- timestamp_key (str): Timestamp attribute name
186
- case_id_key (str): Case ID attribute name
187
188
Returns:
189
float: Generalization value between 0 and 1
190
"""
191
192
def replay_prefix_tbr(prefix, net, im, fm, activity_key='concept:name'):
193
"""
194
Replay activity prefix using token-based replay.
195
Useful for predictive process monitoring.
196
197
Parameters:
198
- prefix (List[str]): Activity sequence prefix
199
- net (PetriNet): Petri net model
200
- im (Marking): Initial marking
201
- fm (Marking): Final marking
202
- activity_key (str): Activity attribute name
203
204
Returns:
205
Marking: Resulting marking after prefix replay
206
"""
207
```
208
209
### Temporal Conformance
210
211
Conformance checking for temporal constraints and time-aware models.
212
213
```python { .api }
214
def conformance_temporal_profile(log, temporal_profile, zeta=1.0, activity_key='concept:name', timestamp_key='time:timestamp', case_id_key='case:concept:name', return_diagnostics_dataframe=False):
215
"""
216
Temporal profile conformance checking for time constraints.
217
218
Parameters:
219
- log (Union[EventLog, pd.DataFrame]): Event log data
220
- temporal_profile (Dict): Temporal constraints between activities
221
- zeta (float): Deviation factor for time constraints
222
- activity_key (str): Activity attribute name
223
- timestamp_key (str): Timestamp attribute name
224
- case_id_key (str): Case ID attribute name
225
- return_diagnostics_dataframe (bool): Return results as DataFrame
226
227
Returns:
228
List[List[Tuple[float, float, float, float]]]: Temporal conformance results per trace
229
"""
230
```
231
232
### Declarative Conformance
233
234
Conformance checking for declarative models with temporal logic constraints.
235
236
```python { .api }
237
def conformance_declare(log, declare_model, activity_key='concept:name', timestamp_key='time:timestamp', case_id_key='case:concept:name', return_diagnostics_dataframe=False):
238
"""
239
DECLARE model conformance checking for temporal logic constraints.
240
241
Parameters:
242
- log (Union[EventLog, pd.DataFrame]): Event log data
243
- declare_model (Dict): DECLARE model with constraints
244
- activity_key (str): Activity attribute name
245
- timestamp_key (str): Timestamp attribute name
246
- case_id_key (str): Case ID attribute name
247
- return_diagnostics_dataframe (bool): Return results as DataFrame
248
249
Returns:
250
List[Dict[str, Any]]: Constraint violation diagnostics per trace
251
"""
252
253
def conformance_log_skeleton(log, log_skeleton, activity_key='concept:name', timestamp_key='time:timestamp', case_id_key='case:concept:name', return_diagnostics_dataframe=False):
254
"""
255
Log skeleton conformance checking for behavioral constraints.
256
257
Parameters:
258
- log (Union[EventLog, pd.DataFrame]): Event log data
259
- log_skeleton (Dict): Log skeleton constraints
260
- activity_key (str): Activity attribute name
261
- timestamp_key (str): Timestamp attribute name
262
- case_id_key (str): Case ID attribute name
263
- return_diagnostics_dataframe (bool): Return results as DataFrame
264
265
Returns:
266
List[Set[Any]]: Constraint violations per trace
267
"""
268
```
269
270
### Utility Functions
271
272
Helper functions for conformance checking operations.
273
274
```python { .api }
275
def check_is_fitting(*args, activity_key='concept:name'):
276
"""
277
Check if trace fits model (deprecated in 2.3.0).
278
Simple boolean fitness check.
279
280
Parameters:
281
- *args: Trace and model arguments
282
- activity_key (str): Activity attribute name
283
284
Returns:
285
bool: True if trace fits the model
286
"""
287
```
288
289
## Usage Examples
290
291
### Basic Conformance Checking
292
293
```python
294
import pm4py
295
296
# Load event log and discover model
297
log = pm4py.read_xes('event_log.xes')
298
net, initial_marking, final_marking = pm4py.discover_petri_net_inductive(log)
299
300
# Calculate fitness using alignments (recommended)
301
fitness = pm4py.fitness_alignments(log, net, initial_marking, final_marking)
302
print(f"Log fitness: {fitness['log_fitness']}")
303
print(f"Average trace fitness: {fitness['average_trace_fitness']}")
304
305
# Calculate precision
306
precision = pm4py.precision_alignments(log, net, initial_marking, final_marking)
307
print(f"Precision: {precision}")
308
309
# Calculate generalization
310
generalization = pm4py.generalization_tbr(log, net, initial_marking, final_marking)
311
print(f"Generalization: {generalization}")
312
```
313
314
### Detailed Conformance Diagnostics
315
316
```python
317
import pm4py
318
319
# Get detailed alignment diagnostics
320
diagnostics = pm4py.conformance_diagnostics_alignments(
321
log, net, initial_marking, final_marking,
322
multi_processing=True,
323
return_diagnostics_dataframe=True
324
)
325
326
# Analyze diagnostics
327
for trace_diagnostic in diagnostics:
328
case_id = trace_diagnostic['case_id']
329
fitness = trace_diagnostic['fitness']
330
cost = trace_diagnostic['cost']
331
alignment = trace_diagnostic['alignment']
332
333
print(f"Case {case_id}: fitness={fitness}, cost={cost}")
334
335
# Analyze alignment moves
336
for move in alignment:
337
move_type = move['type'] # 'sync', 'log', 'model'
338
if move_type == 'log':
339
print(f" Missing in model: {move['activity']}")
340
elif move_type == 'model':
341
print(f" Extra in model: {move['activity']}")
342
```
343
344
### Fast Conformance with Token-Based Replay
345
346
```python
347
import pm4py
348
349
# Use token-based replay for faster computation
350
fitness_tbr = pm4py.fitness_token_based_replay(log, net, initial_marking, final_marking)
351
precision_tbr = pm4py.precision_token_based_replay(log, net, initial_marking, final_marking)
352
353
print(f"TBR Fitness: {fitness_tbr['log_fitness']}")
354
print(f"TBR Precision: {precision_tbr}")
355
356
# Get detailed token-based diagnostics
357
tbr_diagnostics = pm4py.conformance_diagnostics_token_based_replay(
358
log, net, initial_marking, final_marking
359
)
360
361
for diag in tbr_diagnostics:
362
print(f"Case: {diag['case_id']}, Fitness: {diag['trace_fitness']}")
363
print(f" Missing tokens: {diag['missing_tokens']}")
364
print(f" Remaining tokens: {diag['remaining_tokens']}")
365
```
366
367
### Temporal Conformance Checking
368
369
```python
370
import pm4py
371
372
# Discover temporal profile
373
temporal_profile = pm4py.discover_temporal_profile(log)
374
375
# Check temporal conformance
376
temporal_conformance = pm4py.conformance_temporal_profile(
377
log, temporal_profile,
378
zeta=1.5 # Allow 50% deviation from expected times
379
)
380
381
# Analyze temporal violations
382
for trace_idx, trace_result in enumerate(temporal_conformance):
383
print(f"Trace {trace_idx}:")
384
for constraint_result in trace_result:
385
expected_min, expected_max, actual_time, deviation = constraint_result
386
if deviation > 0:
387
print(f" Time violation: expected [{expected_min}, {expected_max}], actual {actual_time}")
388
```
389
390
### Declarative Conformance Checking
391
392
```python
393
import pm4py
394
395
# Discover DECLARE constraints
396
declare_model = pm4py.discover_declare(log, min_support_ratio=0.8)
397
398
# Check DECLARE conformance
399
declare_conformance = pm4py.conformance_declare(log, declare_model)
400
401
# Analyze constraint violations
402
for trace_result in declare_conformance:
403
case_id = trace_result['case_id']
404
violations = trace_result['violations']
405
406
if violations:
407
print(f"Case {case_id} violations:")
408
for constraint, violation_info in violations.items():
409
print(f" {constraint}: {violation_info}")
410
```
411
412
### Quality Metrics Dashboard
413
414
```python
415
import pm4py
416
417
def compute_quality_metrics(log, net, initial_marking, final_marking):
418
"""Compute comprehensive quality metrics for a process model."""
419
420
# Fitness (how well the model explains the log)
421
fitness = pm4py.fitness_alignments(log, net, initial_marking, final_marking)
422
423
# Precision (how much extra behavior the model allows)
424
precision = pm4py.precision_alignments(log, net, initial_marking, final_marking)
425
426
# Generalization (model's ability to handle unseen behavior)
427
generalization = pm4py.generalization_tbr(log, net, initial_marking, final_marking)
428
429
# Simplicity (model complexity)
430
simplicity = pm4py.simplicity_petri_net(net)
431
432
return {
433
'fitness': fitness['log_fitness'],
434
'precision': precision,
435
'generalization': generalization,
436
'simplicity': simplicity
437
}
438
439
# Evaluate model quality
440
metrics = compute_quality_metrics(log, net, initial_marking, final_marking)
441
print("Model Quality Metrics:")
442
for metric, value in metrics.items():
443
print(f" {metric.capitalize()}: {value:.3f}")
444
```