|
| 1 | +""" |
| 2 | +Play out and align a reset/inhibitor Petri net. |
| 3 | +
|
| 4 | +This example uses the hospital-discharge model stored in the test data. It |
| 5 | +demonstrates the two places where custom Petri-net semantics must be supplied: |
| 6 | +
|
| 7 | +1. basic playout, so inhibitor arcs affect enablement and reset arcs affect |
| 8 | + firing; |
| 9 | +2. the semantics-aware Dijkstra alignment variant, so its state-space search |
| 10 | + follows exactly the same rules as the model simulation. |
| 11 | +
|
| 12 | +Run this file from any working directory: |
| 13 | +
|
| 14 | + python examples/inhibitor_reset_playout_and_alignment.py |
| 15 | +""" |
| 16 | + |
| 17 | +import sys |
| 18 | +from pathlib import Path |
| 19 | + |
| 20 | +# A directly executed file puts examples/ (not the repository root) first on |
| 21 | +# sys.path. Add the checkout root before importing PM4Py so this example uses |
| 22 | +# the local implementation, including VERSION_DIJKSTRA_SEMANTICS. |
| 23 | +REPOSITORY_ROOT = Path(__file__).resolve().parent.parent |
| 24 | +sys.path.insert(0, str(REPOSITORY_ROOT)) |
| 25 | + |
| 26 | +import pm4py |
| 27 | +from pm4py.algo.conformance.alignments.petri_net import ( |
| 28 | + algorithm as alignments, |
| 29 | +) |
| 30 | +from pm4py.algo.simulation.playout.petri_net import algorithm as playout |
| 31 | +from pm4py.objects.log.obj import Event, EventLog, Trace |
| 32 | +from pm4py.objects.petri_net.inhibitor_reset.semantics import ( |
| 33 | + InhibitorResetSemantics, |
| 34 | +) |
| 35 | +from pm4py.objects.petri_net.obj import InhibitorNet, ResetNet |
| 36 | + |
| 37 | + |
| 38 | +# Resolve input files relative to this source file. Using __file__ instead of |
| 39 | +# the current working directory makes the example work both from the repository |
| 40 | +# root and from the examples directory. |
| 41 | +DATA_DIR = REPOSITORY_ROOT / "tests" / "input_data" / "inh_res_nets" |
| 42 | +MODEL_PATH = DATA_DIR / "hospital_discharge.pnml" |
| 43 | +LOG_PATH = DATA_DIR / "logs" / "hospital_discharge.xes" |
| 44 | + |
| 45 | + |
| 46 | +def activity_sequence(trace: Trace): |
| 47 | + """Return only the readable activity names from a PM4Py trace.""" |
| 48 | + return [event["concept:name"] for event in trace] |
| 49 | + |
| 50 | + |
| 51 | +def execute_script(): |
| 52 | + # PNML preserves the specialized reset and inhibitor arc types. The PM4Py |
| 53 | + # importer therefore reconstructs a ResetInhibitorNet automatically. |
| 54 | + net, initial_marking, final_marking = pm4py.read_pnml(str(MODEL_PATH)) |
| 55 | + |
| 56 | + # The companion XES file was generated by basic playout. Requesting the |
| 57 | + # legacy EventLog representation gives us Trace/Event objects, which can be |
| 58 | + # passed directly to the alignment API used below. |
| 59 | + stored_log = pm4py.read_xes( |
| 60 | + str(LOG_PATH), return_legacy_log_object=True |
| 61 | + ) |
| 62 | + |
| 63 | + # A reset arc consumes every token from its source place when its transition |
| 64 | + # fires. An inhibitor arc does not consume tokens; instead, it enables its |
| 65 | + # transition only while the source place is empty. Checking the imported |
| 66 | + # objects is a useful sanity check that PNML round-tripping retained both. |
| 67 | + reset_arcs = [ |
| 68 | + arc for arc in net.arcs if isinstance(arc, ResetNet.ResetArc) |
| 69 | + ] |
| 70 | + inhibitor_arcs = [ |
| 71 | + arc for arc in net.arcs if isinstance(arc, InhibitorNet.InhibitorArc) |
| 72 | + ] |
| 73 | + assert reset_arcs, "the imported model should contain a reset arc" |
| 74 | + assert inhibitor_arcs, "the imported model should contain an inhibitor arc" |
| 75 | + |
| 76 | + print("Model:", net.name) |
| 77 | + print("Visible activities:", len([t for t in net.transitions if t.label])) |
| 78 | + print("Stored traces:", len(stored_log)) |
| 79 | + print("First stored trace:", activity_sequence(stored_log[0])) |
| 80 | + |
| 81 | + # ClassicSemantics remains PM4Py's default because it is right for ordinary |
| 82 | + # Petri nets. This model is not ordinary, so the same specialized semantics |
| 83 | + # object is deliberately passed to both playout and alignment. |
| 84 | + semantics = InhibitorResetSemantics() |
| 85 | + |
| 86 | + # Generate a few fresh traces. "add_only_if_fm_is_reached" excludes partial |
| 87 | + # runs stopped by the length limit, while maxTraceLength protects us from an |
| 88 | + # unlucky sequence that repeatedly chooses one of the model's optional |
| 89 | + # loops. |
| 90 | + simulated_log = playout.apply( |
| 91 | + net, |
| 92 | + initial_marking, |
| 93 | + final_marking, |
| 94 | + variant=playout.Variants.BASIC_PLAYOUT, |
| 95 | + parameters={ |
| 96 | + "petri_semantics": semantics, |
| 97 | + "noTraces": 5, |
| 98 | + "maxTraceLength": 40, |
| 99 | + "add_only_if_fm_is_reached": True, |
| 100 | + }, |
| 101 | + ) |
| 102 | + |
| 103 | + print("\nFreshly simulated traces:") |
| 104 | + for index, trace in enumerate(simulated_log): |
| 105 | + print(f" {index}: {activity_sequence(trace)}") |
| 106 | + |
| 107 | + # VERSION_DIJKSTRA_SEMANTICS searches states of the form |
| 108 | + # (position in trace, model marking). For each model or synchronous move it |
| 109 | + # calls the supplied semantics to find enabled transitions and fire them. |
| 110 | + # This avoids flattening the model into a classic synchronous-product net, |
| 111 | + # which would lose the special behavior of reset and inhibitor arcs. |
| 112 | + alignment_results = alignments.apply( |
| 113 | + simulated_log, |
| 114 | + net, |
| 115 | + initial_marking, |
| 116 | + final_marking, |
| 117 | + variant=alignments.Variants.VERSION_DIJKSTRA_SEMANTICS, |
| 118 | + parameters={ |
| 119 | + "petri_semantics": semantics, |
| 120 | + "show_progress_bar": False, |
| 121 | + }, |
| 122 | + ) |
| 123 | + |
| 124 | + # Every trace came from this exact model, so every event should synchronize |
| 125 | + # with a model transition and every optimal alignment should cost zero. |
| 126 | + assert all(result["cost"] == 0 for result in alignment_results) |
| 127 | + print("\nAlignment costs for simulated traces:") |
| 128 | + print([result["cost"] for result in alignment_results]) |
| 129 | + |
| 130 | + # Finally, insert an event that the model does not know. Dijkstra represents |
| 131 | + # it as a log move: ("Unexpected Manual Override", ">>"). With PM4Py's |
| 132 | + # standard costs that deviation contributes 10,000 to the alignment. |
| 133 | + original = simulated_log[0] |
| 134 | + deviating_trace = Trace( |
| 135 | + [Event(dict(event)) for event in original], |
| 136 | + attributes=dict(original.attributes), |
| 137 | + ) |
| 138 | + deviating_trace.insert( |
| 139 | + 1, Event({"concept:name": "Unexpected Manual Override"}) |
| 140 | + ) |
| 141 | + deviation = alignments.apply( |
| 142 | + deviating_trace, |
| 143 | + net, |
| 144 | + initial_marking, |
| 145 | + final_marking, |
| 146 | + variant=alignments.Variants.VERSION_DIJKSTRA_SEMANTICS, |
| 147 | + parameters={"petri_semantics": semantics}, |
| 148 | + ) |
| 149 | + |
| 150 | + assert deviation["cost"] == 10000 |
| 151 | + assert ("Unexpected Manual Override", ">>") in deviation["alignment"] |
| 152 | + print("\nAlignment of a deliberately deviating trace:") |
| 153 | + print(" cost:", deviation["cost"]) |
| 154 | + print(" moves:", deviation["alignment"]) |
| 155 | + |
| 156 | + # EventLog is imported above to make clear that both persisted and simulated |
| 157 | + # logs use PM4Py's normal log type; this assertion also documents the return |
| 158 | + # type of basic Petri-net playout. |
| 159 | + assert isinstance(simulated_log, EventLog) |
| 160 | + |
| 161 | + |
| 162 | +if __name__ == "__main__": |
| 163 | + execute_script() |
0 commit comments