11'''
2- PM4Py – A Process Mining Library for Python
3- Copyright (C) 2024 Process Intelligence Solutions UG (haftungsbeschränkt )
2+ PM4Py – A Process Mining Library for Python
3+ Copyright (C) 2026 Process Intelligence Solutions UG (haftungsbeschränkt )
44
55This program is free software: you can redistribute it and/or modify
66it under the terms of the GNU Affero General Public License as
2525import itertools
2626
2727# typing
28+ from typing import Self , Union
2829from collections .abc import Iterable
29- from typing import Dict , FrozenSet , List , Tuple
30- InputMap = Dict [str , List [FrozenSet ]]
31- OutputMap = Dict [str , List [FrozenSet ]]
32- Individual = Tuple [InputMap , OutputMap ]
30+ TransitionMap = dict [str ,list [frozenset ]]
31+ InputMap = OutputMap = TransitionMap
32+ Individual = tuple [InputMap ,OutputMap ]
3333
3434class iset (frozenset ):
3535 "Indexable frozenset printing as set, i.e. without `frozenset(…)`"
3636 def __repr__ (self ):
3737 return "{" + repr (sorted (self ))[1 :- 1 ] + "}"
3838
3939 @staticmethod
40- def flat (item : Iterable ) -> "iset" :
40+ def flat (item : Iterable ) -> Self :
4141 return iset (itertools .chain (* item ))
4242
43- def rand_partition (pool : Iterable ) -> List [ iset ]:
43+ def rand_partition (pool : Iterable ) -> list [ Union [ set , frozenset ] ]:
4444 pool = set (pool )
4545 # also ensures no activity in two partitions
4646 # s. 4. Causal Matrix, Def. 4; https://doi.org/10.1007/11494744_5
@@ -54,7 +54,23 @@ def rand_partition(pool: Iterable) -> List[iset]:
5454 pool -= draw
5555 return partition
5656
57- def get_src_sink_sets_for_wfnet (I : InputMap , O : OutputMap , T : List [str ]) -> Tuple [List [str ], List [str ]]:
57+ def add_singleton_partition (tmap : TransitionMap , key : str , t : str ):
58+ partitions = tmap .setdefault (key , []) # tmap[key] or []
59+ if not any (t in partition for partition in partitions ):
60+ partitions .append (iset ({t }))
61+
62+ def remove_transition_from_partitions (tmap : TransitionMap , key : str , t : str ):
63+ partitions = tmap .setdefault (key , []) # tmap[key] or []
64+ for i ,partition in enumerate (partitions ):
65+ if t in partition :
66+ updated = iset (partition - {t })
67+ if updated :
68+ partitions [i ] = updated
69+ else :
70+ del partitions [i ]
71+ return
72+
73+ def get_src_sink_sets_for_wfnet (I : InputMap , O : OutputMap , T : list [str ]) -> tuple [list [str ],list [str ]]:
5874 """Determines input set and output set, which need to be connected by a place to create a WF-net"""
5975 def add2graphs (graphs , t , nextT ):
6076 # find graph
@@ -70,8 +86,8 @@ def add2graphs(graphs, t, nextT):
7086 else :
7187 successors .extend (S )
7288 graph += successors
73- # merge if end = start
74- for tn in successors :
89+ # merge if path end = start
90+ for tn in successors : # tn = end
7591 for g2 in graphs [:]: # [:] = copy
7692 if g2 [0 ] == tn and g2 != graph :
7793 graph += g2
@@ -82,6 +98,6 @@ def add2graphs(graphs, t, nextT):
8298 for t in T :
8399 graphsI = add2graphs (graphsI , t , I )
84100 graphsO = add2graphs (graphsO , t , O )
85- first = [g [0 ] for g in graphsO ] # ⋃first = reachable via O[∀t ]
86- last = [g [0 ] for g in graphsI ] # ⋃first = reachable via I[∀t ]
87- return first , last
101+ sources = [g [0 ] for g in graphsO ] # ⋃sources = ∀t reachable from O[∀sources ]
102+ sinks = [g [0 ] for g in graphsI ] # ⋃sinks = ∀t reachable from I[∀sinks ]
103+ return sources , sinks
0 commit comments