Source code

Revision control

Copy as Markdown

Other Tools

"""List of functions which are useful in many places."""
import sys
import typing
T = typing.TypeVar("T")
U = typing.TypeVar("U")
def keep_until(
iterable: typing.Iterable[T],
pred: typing.Callable[[T], bool]
) -> typing.Iterable[T]:
"""Filter an iterable generator or list and keep all elements until the first
time the predicate becomes true, including the element where the predicate
is true. All elements after are skipped."""
for e in iterable:
yield e
if pred(e):
return
def split(
iterable: typing.Iterable[T],
pred: typing.Callable[[T], bool]
) -> typing.Tuple[typing.List[T], typing.List[T]]:
"""Given a predicate, split the input iterator into a tuple of 2 list. A first
list which satisfy the predicate and a second list which does not."""
yes = []
no = []
for e in iterable:
if pred(e):
yes.append(e)
else:
no.append(e)
return yes, no
def consume(iterator: typing.Iterable[T], progress: bool) -> None:
"""Drain the iterator. If progress is true, print dots on stdout."""
i = 0
to_feed = str(i)
try:
for _ in iterator:
if progress:
if len(to_feed) > 0:
sys.stdout.write(to_feed[0])
to_feed = to_feed[1:]
else:
sys.stdout.write(".")
i += 1
if i % 100 == 0:
sys.stdout.write("\n")
to_feed = str(i)
sys.stdout.flush()
finally:
if progress and i != 0:
sys.stdout.write("\n")
sys.stdout.flush()
class default_id_dict(dict, typing.Mapping[T, T]):
def __missing__(self, key) -> T:
return key
class default_fwd_dict(dict, typing.Mapping[T, U]):
def __init__(self, fwd: typing.Mapping[T, U]):
super().__init__()
self.fwd = fwd
def __missing__(self, key: T) -> U:
return self.fwd[key]