File: collections.py

package info (click to toggle)
python-xsdata 24.1-2
  • links: PTS, VCS
  • area: main
  • in suites: sid
  • size: 2,936 kB
  • sloc: python: 29,257; xml: 404; makefile: 27; sh: 6
file content (116 lines) | stat: -rw-r--r-- 2,919 bytes parent folder | download
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
from collections import defaultdict
from typing import (
    Any,
    Callable,
    Dict,
    Iterable,
    Iterator,
    List,
    Optional,
    Sequence,
    Set,
    TypeVar,
)

T = TypeVar("T")


def is_array(value: Any) -> bool:
    if isinstance(value, tuple):
        return not hasattr(value, "_fields")

    return isinstance(value, (list, set, frozenset))


def unique_sequence(items: Iterable[T], key: Optional[str] = None) -> List[T]:
    """
    Return a new list with the unique values from an iterable.

    Optionally you can also provide a lambda to generate the unique key
    of each item in the iterable object.
    """
    seen = set()

    def is_new(val: Any) -> bool:
        if key:
            val = getattr(val, key)

        if val in seen:
            return False

        seen.add(val)
        return True

    return [item for item in items if is_new(item)]


def remove(items: Iterable[T], predicate: Callable) -> List[T]:
    """Return a new list without the items that match the predicate."""
    return [x for x in items if not predicate(x)]


def group_by(items: Iterable[T], key: Callable) -> Dict[Any, List[T]]:
    """Group the items of an iterable object by the result of the callable."""
    result = defaultdict(list)
    for item in items:
        result[key(item)].append(item)
    return result


def apply(items: Iterable, func: Callable):
    """Apply the given function to each item of the iterable object."""
    for item in items:
        func(item)


def find(items: Sequence, value: Any) -> int:
    """Return the index of the value in the given sequence without raising
    exception in case of failure."""
    try:
        return items.index(value)
    except ValueError:
        return -1


def first(items: Iterator[T]) -> Optional[T]:
    """Return the first item of the iterator."""
    return next(items, None)


def prepend(target: List, *args: Any):
    """Prepend items to the target list."""
    target[:0] = args


def connected_components(lists: List[List[Any]]) -> Iterator[List[Any]]:
    """
    Merge lists of lists that share common elements.

    https://stackoverflow.com/questions/4842613/merge-lists-that-share-
    common-elements
    """
    neighbors = defaultdict(set)
    for each in lists:
        for item in each:
            neighbors[item].update(each)

    def component(node: Any, neigh: Dict[Any, Set], see: Set[Any]):
        nodes = {node}
        while nodes:
            next_node = nodes.pop()
            see.add(next_node)
            nodes |= neigh[next_node] - see
            yield next_node

    seen: Set[Any] = set()
    for item in neighbors:
        if item not in seen:
            yield sorted(component(item, neighbors, seen))


def find_connected_component(groups: List[List[Any]], value: Any) -> int:
    for index, group in enumerate(groups):
        if value in group:
            return index

    return -1