Skip to content

Type annotations for adafruit_itertools #24

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 1 commit into from
Apr 25, 2024
Merged

Type annotations for adafruit_itertools #24

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
47 changes: 32 additions & 15 deletions adafruit_itertools/__init__.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -53,7 +53,10 @@
pass


def accumulate(iterable, func=lambda x, y: x + y):
def accumulate(
iterable: Iterable[_T],
func: Callable[[_T, _T], _T] = lambda x, y: x + y, # type: ignore[operator]
) -> Iterator[_T]:
"""Make an iterator that returns accumulated sums, or accumulated
results of other binary functions (specified via the optional func
argument). If func is supplied, it should be a function of two
Expand Down Expand Up @@ -200,7 +203,7 @@ def count(start: _N = 0, step: _N = 1) -> Iterator[_N]:
start += step


def cycle(p):
def cycle(p: Iterable[_T]) -> Iterator[_T]:
"""Make an iterator returning elements from the iterable and saving a copy
of each. When the iterable is exhausted, return elements from the saved
copy. Repeats indefinitely.
Expand All @@ -209,7 +212,7 @@ def cycle(p):

"""
try:
len(p)
len(p) # type: ignore[arg-type]
except TypeError:
# len() is not defined for this type. Assume it is
# a finite iterable so we must cache the elements.
Expand Down Expand Up @@ -242,7 +245,9 @@ def dropwhile(predicate: _Predicate[_T], iterable: Iterable[_T]) -> Iterator[_T]
yield x


def filterfalse(predicate: _Predicate[_T], iterable: Iterable[_T]) -> Iterator[_T]:
def filterfalse(
predicate: Optional[_Predicate[_T]], iterable: Iterable[_T]
) -> Iterator[_T]:
"""Make an iterator that filters elements from iterable returning only those
for which the predicate is False. If predicate is None, return the items
that are false.
Expand Down Expand Up @@ -288,23 +293,29 @@ class groupby:
# [k for k, g in groupby('AAAABBBCCDAABBB')] --> A B C D A B
# [list(g) for k, g in groupby('AAAABBBCCD')] --> AAAA BBB CC D

def __init__(self, iterable, key=None):
def __init__(
self,
iterable: Iterable[_T],
key: Optional[Callable[[_T], Any]] = None,
):
self.keyfunc = key if key is not None else lambda x: x
self.it = iter(iterable)
self.tgtkey = self.currkey = self.currvalue = object()
# Sentinel values, not actually returned during iteration.
self.currvalue: _T = object() # type: ignore[assignment]
self.tgtkey = self.currkey = self.currvalue

def __iter__(self):
def __iter__(self) -> Iterator[Tuple[Any, Iterator[_T]]]:
return self

def __next__(self):
def __next__(self) -> Tuple[Any, Iterator[_T]]:
self.id = object()
while self.currkey == self.tgtkey:
self.currvalue = next(self.it) # Exit on StopIteration
self.currkey = self.keyfunc(self.currvalue)
self.tgtkey = self.currkey
return (self.currkey, self._grouper(self.tgtkey, self.id))

def _grouper(self, tgtkey, id):
def _grouper(self, tgtkey: Any, id: object) -> Iterator[_T]:
while self.id is id and self.currkey == tgtkey:
yield self.currvalue
try:
Expand All @@ -314,7 +325,12 @@ def _grouper(self, tgtkey, id):
self.currkey = self.keyfunc(self.currvalue)


def islice(p, start, stop=(), step=1):
def islice(
p: Iterable[_T],
start: int,
stop: Optional[int] = (), # type: ignore[assignment]
step: int = 1,
) -> Iterator[_T]:
"""Make an iterator that returns selected elements from the
iterable. If start is non-zero and stop is unspecified, then the
value for start is used as end, and start is taken to be 0. Thus the
Expand Down Expand Up @@ -420,7 +436,8 @@ def permutations(
return


def product(*args: Iterable[_T], r: int = 1) -> Iterator[Tuple[_T, ...]]:
# def product(*args: Iterable[_T], r: int = 1) -> Iterator[Tuple[_T, ...]]:
def product(*args: Iterable[Any], r: int = 1) -> Iterator[Tuple[Any, ...]]:
"""Cartesian product of input iterables.

Roughly equivalent to nested for-loops in a generator expression. For
Expand All @@ -444,7 +461,7 @@ def product(*args: Iterable[_T], r: int = 1) -> Iterator[Tuple[_T, ...]]:
# product('ABCD', 'xy') --> Ax Ay Bx By Cx Cy Dx Dy
# product(range(2), repeat=3) --> 000 001 010 011 100 101 110 111
pools = [tuple(pool) for pool in args] * r
result: List[List[_T]] = [[]]
result: List[List[Any]] = [[]]
for pool in pools:
result = [x + [y] for x in result for y in pool]
for prod in result:
Expand Down Expand Up @@ -513,8 +530,8 @@ def tee(iterable: Iterable[_T], n: int = 2) -> Sequence[Iterator[_T]]:


def zip_longest(
*args: Iterable[_T], fillvalue: _OptionalFill = None
) -> Iterator[Tuple[Union[_T, _OptionalFill], ...]]:
*args: Iterable[Any], fillvalue: _OptionalFill = None
) -> Iterator[Tuple[Any, ...]]:
"""Make an iterator that aggregates elements from each of the
iterables. If the iterables are of uneven length, missing values are
filled-in with fillvalue. Iteration continues until the longest
Expand All @@ -524,7 +541,7 @@ def zip_longest(
:param fillvalue: value to fill in those missing from shorter iterables
"""
# zip_longest('ABCD', 'xy', fillvalue='-') --> Ax By C- D-
iterators: List[Iterator[Union[_T, _OptionalFill]]] = [iter(it) for it in args]
iterators: List[Iterator[Any]] = [iter(it) for it in args]
num_active = len(iterators)
if not num_active:
return
Expand Down
15 changes: 14 additions & 1 deletion tests/README.rst
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -8,7 +8,8 @@ Itertools Tests

These tests run under CPython, and are intended to verify that the
Adafruit library functions return the same outputs compared to ones in
the standard `itertools` module.
the standard `itertools` module, and also to exercise some type
annotations.

These tests run automatically from the standard `circuitpython github
workflow <wf_>`_. To run them manually, first install these packages
Expand All @@ -21,4 +22,16 @@ the following command::

$ python -m pytest

Type annotation tests don't run automatically at this point. But to
verify type-related issues manually, first install these packages if
necessary::

$ pip3 install mypy

Then ensure you're in the *root* directory of the repository and run
the following command::

$ mypy --warn-unused-ignores --disallow-untyped-defs tests


.. _wf: https://github.com/adafruit/workflows-circuitpython-libs/blob/6e1562eaabced4db1bd91173b698b1cc1dfd35ab/build/action.yml#L78-L84
Loading