dev4py.utils.collectors

  1"""The `collectors` module provides a set of collectors inspired by java (java.util.stream.Collectors)"""
  2
  3# Copyright 2022 the original author or authors (i.e.: St4rG00se for Dev4py).
  4#
  5# Licensed under the Apache License, Version 2.0 (the "License");
  6# you may not use this file except in compliance with the License.
  7# You may obtain a copy of the License at
  8#
  9#      https://www.apache.org/licenses/LICENSE-2.0
 10#
 11# Unless required by applicable law or agreed to in writing, software
 12# distributed under the License is distributed on an "AS IS" BASIS,
 13# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 14# See the License for the specific language governing permissions and
 15# limitations under the License.
 16
 17from dataclasses import dataclass
 18from functools import partial
 19from typing import Generic, Any, Final, Optional, cast
 20
 21from dev4py.utils import lists, dicts, objects, tuples
 22from dev4py.utils.objects import require_non_none, is_none, to_self
 23from dev4py.utils.types import BiConsumer, BiFunction, Supplier, T, R, Function, K, V
 24
 25
 26##############################
 27#  MODULE CLASSES/FUNCTIONS  #
 28##############################
 29@dataclass(frozen=True)
 30class Collector(Generic[T, R]):
 31    """
 32    Collector class: A generic dataclass to describe a collector[T,R]
 33
 34    Args:
 35        supplier (Supplier[R]): A function that creates and returns a new result container
 36        accumulator (BiFunction[R, T, R]): A function that folds a value into a result container
 37        combiner (BiFunction[R, R, R]): A function that accepts two partial results and merges them
 38    """
 39    supplier: Supplier[R]
 40    accumulator: BiFunction[R, T, R]
 41    combiner: BiFunction[R, R, R]
 42
 43    def __post_init__(self):
 44        """
 45        Post init values checker
 46        Raises:
 47            TypeError: Raises a TypeError at least one parameter is None
 48        """
 49        require_non_none(self.supplier)
 50        require_non_none(self.accumulator)
 51        require_non_none(self.combiner)
 52
 53
 54def of(supplier: Supplier[R], accumulator: BiFunction[R, T, R], combiner: BiFunction[R, R, R]) -> Collector[T, R]:
 55    """
 56    Returns a Collector by using the given parameters
 57
 58    Args:
 59        supplier (Supplier[R]): A function that creates and returns a new result container
 60        accumulator (BiFunction[R, T, R]): A function that folds a value into a result container
 61        combiner (BiFunction[R, R, R]): A function that accepts two partial results and merges them
 62
 63    Returns:
 64        collector: a collector[T,R] built with given parameters
 65
 66    Raises:
 67        TypeError: Raises a TypeError at least one parameter is None
 68    """
 69    return Collector(
 70        supplier=require_non_none(supplier),
 71        accumulator=require_non_none(accumulator),
 72        combiner=require_non_none(combiner)
 73    )
 74
 75
 76def of_biconsumers(supplier: Supplier[R], accumulator: BiConsumer[R, T], combiner: BiConsumer[R, R]) -> Collector[T, R]:
 77    """
 78    Returns a Collector by using the given parameters (biconsumer instead of bifunction)
 79
 80    Args:
 81        supplier (Supplier[R]): A function that creates and returns a new result container
 82        accumulator (BiConsumer[R, T]): A function that folds a value into a result container
 83        combiner (BiConsumer[R, R]): A function that accepts two partial results and merges them
 84
 85    Returns:
 86        collector: a collector[T,R] built with given parameters
 87
 88    Raises:
 89        TypeError: Raises a TypeError at least one parameter is None
 90    """
 91    return of(
 92        supplier=require_non_none(supplier),
 93        accumulator=_to_bifunction(accumulator),
 94        combiner=_to_bifunction(combiner)
 95    )
 96
 97
 98def to_list() -> Collector[T, list[T]]:
 99    """
100    Returns a Collector that accumulates the input elements into a new list
101
102    Returns:
103        Collector[T, list[T]]: a Collector that accumulates the input elements into a new list
104    """
105    return of(
106        supplier=lists.empty_list,
107        accumulator=lists.append,
108        combiner=lists.extend
109    )
110
111
112def to_dict(key_mapper: Function[T, K], value_mapper: Function[T, V]) -> Collector[T, dict[K, V]]:
113    """
114    Returns a Collector that accumulates elements into a dict whose keys and values are the result of applying the
115    provided mapping functions to the input elements
116
117    Args:
118        key_mapper: a mapping function to produce keys
119        value_mapper: a mapping function to produce values
120
121    Returns:
122         Collector[T, dict[K, V]]: a Collector which collects elements into a dict whose keys and values are the result
123            of applying mapping functions to the input elements
124    """
125    require_non_none(key_mapper)
126    require_non_none(value_mapper)
127    return of_biconsumers(
128        supplier=dicts.empty_dict,
129        accumulator=partial(_to_dict_accumulator, key_mapper=key_mapper, value_mapper=value_mapper),
130        combiner=_to_dict_combiner
131    )
132
133
134def to_none() -> Collector[T, None]:
135    """
136    Returns a Collector that always returns None
137
138    Returns:
139        Collector[T, None]: A collector that always returns None
140    """
141    return of(
142        supplier=objects.to_none,
143        accumulator=objects.to_none,
144        combiner=objects.to_none
145    )
146
147
148def to_tuple() -> Collector[T, tuple[T, ...]]:
149    """
150    Returns a Collector that accumulates the input elements into a new tuple
151
152    Returns:
153        Collector[T, tuple[T, ...]]: a Collector that accumulates the input elements into a new tuple
154    """
155    return of(
156        supplier=tuples.empty_tuple,
157        accumulator=tuples.append,
158        combiner=tuples.extend
159    )
160
161
162def to_counter() -> Collector[T, int]:
163    """
164    Returns a Collector accepting elements of type T that counts the number of input elements
165
166    Returns:
167        Collector[T, int]: a Collector accepting elements of type T that counts the number of input elements
168    """
169    return of(
170        supplier=_to_counter_supplier,
171        accumulator=_to_counter_accumulator,
172        combiner=_to_counter_combiner
173    )
174
175
176def grouping_by(
177        key_mapper: Function[T, K], value_mapper: Function[T, V] = to_self  # type: ignore
178) -> Collector[T, dict[K, list[V]]]:
179    """
180    Returns a collector that groups elements into a dictionary based on the provided key mapper and value mapper
181    functions.
182
183    Args:
184        key_mapper (Function[T, K]): A function that maps an element to its key.
185        value_mapper (Function[T, V], optional): A function that maps an element to its value. Defaults to to_self.
186
187    Returns:
188        Collector[T, dict[K, list[V]]]: A collector that groups elements into a dictionary.
189    """
190    require_non_none(key_mapper)
191    require_non_none(value_mapper)
192    return of_biconsumers(
193        supplier=dicts.empty_dict,
194        accumulator=partial(_grouping_by_accumulator, key_mapper=key_mapper, value_mapper=value_mapper),
195        combiner=_grouping_by_combiner
196    )
197
198
199##############################
200#  PRIVATE MODULE FUNCTIONS  #
201##############################
202def _bifunction_from_biconsumer(t: T, r: R, biconsumer: BiConsumer[T, R]) -> T:
203    """
204    private function to create a BiFunction from a BiConsumer by using `partial` function
205    Note: lambda or inner function are not used in order to be compatible with multiprocessing (lambda are not
206    serializable)
207    """
208    require_non_none(biconsumer)(t, r)
209    return t
210
211
212def _to_bifunction(biconsumer: BiConsumer[T, R]) -> BiFunction[T, R, T]:
213    """
214    private function to create a BiFunction from a BiConsumer
215    Note: lambda are not used in order to be compatible with multiprocessing (lambda are not serializable)
216    """
217    return partial(_bifunction_from_biconsumer, biconsumer=require_non_none(biconsumer))
218
219
220def _to_dict_accumulator(d: dict[K, V], value: T, key_mapper: Function[T, K], value_mapper: Function[T, V]) -> None:
221    """
222    private function to represent a dict accumulator
223    Note: lambda are not used in order to be compatible with multiprocessing (lambda are not serializable)
224    """
225    dicts.put_value(d, key_mapper(value), value_mapper(value))
226
227
228def _to_dict_combiner(d1: dict[K, V], d2: dict[K, V]) -> None:
229    """
230    private function to represent a dict combiner
231    Note: lambda are not used in order to be compatible with multiprocessing (lambda are not serializable)
232    """
233    dicts.update(d1, d2)
234
235
236def _to_counter_supplier() -> int:
237    """
238    private function to represent a counter supplier
239    Note: lambda are not used in order to be compatible with multiprocessing (lambda are not serializable)
240    """
241    return 0
242
243
244def _to_counter_accumulator(i: int, value: Any) -> int:  # pylint: disable=W0613
245    """
246    private function to represent a counter accumulator
247    Note: lambda are not used in order to be compatible with multiprocessing (lambda are not serializable)
248    """
249    return i + 1
250
251
252def _to_counter_combiner(i1: int, i2: int) -> int:
253    """
254    private function to represent a counter combiner
255    Note: lambda are not used in order to be compatible with multiprocessing (lambda are not serializable)
256    """
257    return i1 + i2
258
259
260def _grouping_by_accumulator(
261        dictionary: dict[K, list[V]], value: T, key_mapper: Function[T, K], value_mapper: Function[T, V]
262) -> None:
263    """
264    private function to represent a grouping by accumulator
265    Note: lambda are not used in order to be compatible with multiprocessing (lambda are not serializable)
266    """
267    key: Final[K] = key_mapper(value)
268    val: Final[V] = value_mapper(value)
269    values: Final[Optional[list[V]]] = dictionary.get(key)
270    if is_none(values):
271        dictionary[key] = [val]
272    else:
273        cast(list[V], values).append(val)
274
275
276def _grouping_by_combiner(dictionary_1: dict[K, list[V]], dictionary_2: dict[K, list[V]]) -> None:
277    """
278    private function to represent a grouping by combiner
279    Note: lambda are not used in order to be compatible with multiprocessing (lambda are not serializable)
280    """
281    require_non_none(dictionary_1)
282    for key, dictionary_2_values in require_non_none(dictionary_2).items():
283        dictionary_1_values: Optional[list[V]] = dictionary_1.get(key)
284        if is_none(dictionary_1_values):
285            dictionary_1[key] = dictionary_2_values  # no need new list ref because internal function
286        else:
287            cast(list[V], dictionary_1_values).extend(dictionary_2_values)