Merge remote-tracking branch 'upstream/hotfixes' into release

Author: fit-alessandro-berti

pm4py/algo/discovery/genetic/util.py

@@ -25,22 +25,21 @@
 import itertools
 
 # typing
-from typing import Self, Union
-from collections.abc import Iterable
-TransitionMap = dict[str,list[frozenset]]
+from typing import Dict, FrozenSet, Iterable, List, Set, Tuple, Union
+TransitionMap = Dict[str, List[FrozenSet[str]]]
 InputMap = OutputMap = TransitionMap
-Individual = tuple[InputMap,OutputMap]
+Individual = Tuple[InputMap, OutputMap]
 
 class iset(frozenset):
     "Indexable frozenset printing as set, i.e. without `frozenset(…)`"
     def __repr__(self):
         return "{" + repr(sorted(self))[1:-1] + "}"
 
     @staticmethod
-    def flat(item: Iterable) -> Self:
+    def flat(item: Iterable) -> "iset":
         return iset(itertools.chain(*item))
 
-def rand_partition(pool: Iterable) -> list[Union[set,frozenset]]:
+def rand_partition(pool: Iterable) -> List[Union[Set[str], FrozenSet[str]]]:
     pool = set(pool)
     #     also ensures no activity in two partitions
     #     s. 4. Causal Matrix, Def. 4; https://doi.org/10.1007/11494744_5
@@ -70,7 +69,7 @@ def remove_transition_from_partitions(tmap: TransitionMap, key: str, t: str):
                 del partitions[i]
             return
 
-def get_src_sink_sets_for_wfnet(I: InputMap, O: OutputMap, T: list[str]) -> tuple[list[str],list[str]]:
+def get_src_sink_sets_for_wfnet(I: InputMap, O: OutputMap, T: List[str]) -> Tuple[List[str], List[str]]:
     """Determines input set and output set, which need to be connected by a place to create a WF-net"""
     def add2graphs(graphs, t, nextT):
         # find graph

pm4py/algo/discovery/genetic/variants/classic.py

@@ -47,7 +47,7 @@ def tqdm(iterable=None, *args, **kwargs):
 from pm4py.objects.genetic_matrix.obj import GeneticMatrix
 
 # typing
-from typing import Any, Iterable, Optional, Union
+from typing import Any, Dict, Iterable, List, Optional, Tuple, Union
 from pandas.core.frame import DataFrame
 from pm4py.objects.log.obj import EventLog, EventStream
 from pm4py.objects.petri_net.obj import PetriNet, Marking
@@ -56,8 +56,8 @@ def tqdm(iterable=None, *args, **kwargs):
 
 def apply(
     log: Union[EventLog, EventStream, DataFrame],
-    parameters: Optional[dict[Union[str, Parameters], Any]] = None,
-) -> tuple[PetriNet, Marking, Marking]:
+    parameters: Optional[Dict[Union[str, Parameters], Any]] = None,
+) -> Tuple[PetriNet, Marking, Marking]:
     """
     Discovers a Petri net using Genetic Miner
 
@@ -178,7 +178,7 @@ def apply(
             log_csv.writerow([datetime.now(), len(history)] + fitness)
     return matrix2petrinet(GeneticMatrix(*population[0], T))
 
-def individuals(log: Union[DataFrame, EventLog], sample_size=1, T=None, keys: dict[str,str] = {"activity_key":xes.DEFAULT_NAME_KEY, "timestamp_key":xes.DEFAULT_TIMESTAMP_KEY, "case_id_key":constants.CASE_CONCEPT_NAME}) -> list[Individual]:
+def individuals(log: Union[DataFrame, EventLog], sample_size=1, T=None, keys: Dict[str, str] = {"activity_key":xes.DEFAULT_NAME_KEY, "timestamp_key":xes.DEFAULT_TIMESTAMP_KEY, "case_id_key":constants.CASE_CONCEPT_NAME}) -> List[Individual]:
     if not T:
         T = tuple(log[keys['activity_key']].unique())
     # polyfill for itertools.pairwise (requires Python 3.10+)
@@ -217,14 +217,14 @@ def pairwise(iterable):
     return samples
 
 def tournament(
-    population: list[Individual],
+    population: List[Individual],
     log: Union[DataFrame, EventLog],
     T,
     sort=True,
     activity_key: str = xes.DEFAULT_NAME_KEY,
     timestamp_key: str = xes.DEFAULT_TIMESTAMP_KEY,
     case_id_key: str = constants.CASE_CONCEPT_NAME,
-) -> tuple[list[Individual], list[float]]:
+) -> Tuple[List[Individual], List[float]]:
     """sort=True: sort descending by fitness (i.e. best first)"""
     # @src 6.2. Fitness Calculation; https://doi.org/10.1007/11494744_5
     fitness = []
@@ -260,7 +260,7 @@ def tournament(
         population, fitness = list(population), list(fitness)
     return (population, fitness)
 
-def sample_parents(population: list[Individual], fitness: list[float], elitism_min_sample: int):
+def sample_parents(population: List[Individual], fitness: List[float], elitism_min_sample: int):
     population_fitness = tuple(zip(population, fitness))
     parent1 = sorted(
         random.sample(population_fitness, k=elitism_min_sample),
@@ -277,7 +277,7 @@ def sample_parents(population: list[Individual], fitness: list[float], elitism_m
     )[0][0]
     return (parent1, parent2)
 
-def crossover(parent1: Individual, parent2: Individual, T: list[str]) -> tuple[Individual,Individual]:
+def crossover(parent1: Individual, parent2: Individual, T: List[str]) -> Tuple[Individual, Individual]:
     # @src 6.3. Genetic Operations: Crossover; https://doi.org/10.1007/11494744_5
     # 1. cross-over point
     t = random.choice(T)
@@ -344,7 +344,7 @@ def mutate(individual: Individual, rate: float = 0.01) -> Individual:
             O[t] = rand_partition(itertools.chain(*O[t]))
     return (I,O)
 
-def repair(I: InputMap, O: OutputMap, C: numpy.ndarray, T: list[str]) -> Individual:
+def repair(I: InputMap, O: OutputMap, C: numpy.ndarray, T: List[str]) -> Individual:
     """
     Merging partitions until petri-net is (coherent) workflow net
     see last requirement in 4. Causal Matrix, Def. 4; https://doi.org/10.1007/11494744_5

pm4py/objects/genetic_matrix/obj.py

@@ -26,9 +26,9 @@
 class GeneticMatrix:
     def __init__(
         self,
-        input_map: dict[str,list[frozenset]],
-        output_map: dict[str,list[frozenset]],
-        transitions: list[str],
+        input_map: Dict[str, List[FrozenSet[str]]],
+        output_map: Dict[str, List[FrozenSet[str]]],
+        transitions: List[str],
     ):
         """
         Initialize a Genetic matrix