# AUTOGENERATED FILE! PLEASE DON'T EDIT
cliSettings = {"defaultDelim": "\t", "defaultIndent": " ",
"oboFile": None, "strict": False, "lookupImgs": True,
"svgScale": 0.7, "inf": float("inf")}
def patchDefaultDelim(s:str):
"""
:param s:
- if not None, returns self
- else returns the default delimiter in cliSettings"""
return cliSettings["defaultDelim"] if s is None else s
def patchDefaultIndent(s:str):
"""
:param s:
- if not None, returns self
- else returns the default indent character in cliSettings"""
return cliSettings["defaultIndent"] if s is None else s
from typing import List, Iterator, Any, NewType, TypeVar, Generic
import k1lib.cli as cli; from numbers import Number
import itertools, copy, torch; import numpy as np
__all__ = ["cliSettings", "BaseCli", "Table", "T",
"serial", "oneToMany", "manyToMany", "manyToManySpecific"]
from contextlib import contextmanager
@contextmanager
def settingsContext():
"""Context to preserve whatever is in cliSettings.
Example::
with cliSettings["context"]():
cliSettings["inf"] = 21
# automatically restores old value"""
d = dict(cliSettings); yield; cliSettings.clear(); cliSettings.update(d)
cliSettings["context"] = settingsContext
T = TypeVar("T")
"""Generic type variable"""
class _MetaType(type):
def __getitem__(self, generic):
d = {"__args__": generic, "_n": self._n, "__doc__": self.__doc__}
return _MetaType(self._n, (), d)
def __repr__(self):
def main(self):
def trueName(o):
if isinstance(o, _MetaType): return main(o)
try: return o.__name__
except: return f"{o}"
if hasattr(self, "__args__"):
if isinstance(self.__args__, tuple):
return f"{self._n}[{', '.join([trueName(e) for e in self.__args__])}]"
else: return f"{self._n}[{trueName(self.__args__)}]"
return self._n
return main(self)
def newTypeHint(name, docs=""):
"""Creates a new type hint that can be sliced and yet still looks fine
in sphinx. Crudely written by my poorly understood idea of Python's
metaclasses. Seriously, this shit is bonkers, read over it https://stackoverflow.com/questions/100003/what-are-metaclasses-in-python
Example::
Table = newTypeHint("Table", "some docs")
Table[int] # prints out as "Table[int]", and sphinx fell for it too
Table[Table[str], float] # prints out as "Table[Table[str], float]"
"""
return _MetaType(name, (), {"_n": name, "__doc__": docs})
#Table = newTypeHint("Table", """Essentially just Iterator[List[T]]. This class is just here so that I can generate the docs with nicely formatted types like "Table[str]".""")
#Table = NewType("Table", List)
class Table(Generic[T]):
"""Essentially just Iterator[List[T]]. This class is just here so that I can generate the docs with nicely formatted types like "Table[str]"."""
pass
Table._name = "Table"
#Table.__module__ = "cli"
class Row(list):
"""Not really used currently. Just here for potential future feature"""
pass
[docs]class BaseCli:
"""A base class for all the cli stuff. You can definitely create new cli tools that
have the same feel without extending from this class, but advanced stream operations
(like ``+``, ``&``, ``.all()``, ``|``) won't work.
At the moment, you don't have to call super().__init__() and super().__ror__(),
as __init__'s only job right now is to solidify any :class:`~k1lib.cli.modifier.op`
passed to it, and __ror__ does nothing."""
[docs] def __init__(self, fs=[]):
"""Not expected to be instantiated by the end user.
:param fs: if functions inside here is actually a :class:`~k1lib.cli.modifier.op`,
then solidifies it (make it not absorb __call__ anymore)"""
[f.op_solidify() for f in fs if isinstance(f, cli.op)]
[docs] def __and__(self, cli:"BaseCli") -> "oneToMany":
"""Duplicates input stream to multiple joined clis.
Example::
# returns [[5], [0, 1, 2, 3, 4]]
range(5) | (shape() & iden()) | deref()
Kinda like :class:`~k1lib.cli.modifier.apply`. There're just multiple ways of doing
this. This I think, is more intuitive, and :class:`~k1lib.cli.modifier.apply` is more
for lambdas and columns mode. Performances are pretty much identical."""
if isinstance(self, oneToMany):
self.clis.append(cli); return self
if isinstance(cli, oneToMany):
cli.clis.append(self); return cli
return oneToMany(self, cli)
[docs] def __add__(self, cli:"BaseCli") -> "manyToManySpecific":
"""Parallel pass multiple streams to multiple clis."""
if isinstance(self, manyToManySpecific):
self.clis.append(cli); return self
if isinstance(cli, manyToManySpecific):
cli.clis.append(self); return cli
return manyToManySpecific(self, cli)
[docs] def all(self, n:int=1) -> "BaseCli":
"""Applies this cli to all incoming streams.
:param n: how many times should I chain ``.all()``?"""
s = self
for i in range(n): s = manyToMany(s)
return s
[docs] def __or__(self, cli) -> "serial":
"""Joins clis end-to-end"""
if isinstance(self, serial):
self.clis.append(cli); return self
if isinstance(cli, serial):
cli.clis = [self] + cli.clis; return cli
return serial(self, cli)
[docs] def __ror__(self, it): return NotImplemented
[docs] def f(self) -> Table[Table[int]]:
"""Creates a normal function :math:`f(x)` which is equivalent to
``x | self``."""
return lambda it: self.__ror__(it)
[docs] def __lt__(self, it):
"""Default backup join symbol `>`, in case `it` implements __ror__()"""
return self.__ror__(it)
[docs] def __call__(self, it, *args):
"""Another way to do ``it | cli``. If multiple arguments are fed, then the
argument list is passed to cli instead of just the first element. Example::
@applyS
def f(it):
return it
f(2) # returns 2
f(2, 3) # returns [2, 3]"""
if len(args) == 0: return self.__ror__(it)
else: return self.__ror__([it, *args])
def fastF(c):
"""Tries to figure out what's going on, is it a normal function, or an applyS, or a
BaseCli, etc., and return a really fast function for execution."""
if isinstance(c, cli.op):
l = len(c._ab_steps)
if l == 1: return c._ab_steps[0][1]
elif l == 0: return lambda x: x
elif l == 2:
a, b = c._ab_steps[0][1], c._ab_steps[1][1]
return lambda x: b(a(x))
return c.ab_operate
if isinstance(c, cli.applyS): return fastF(c.f)
if isinstance(c, BaseCli): return c.__ror__
return c
[docs]class serial(BaseCli):
[docs] def __init__(self, *clis:List[BaseCli]):
"""Merges clis into 1, feeding end to end. Used in chaining clis
together without a prime iterator. Meaning, without this, stuff like this
fails to run::
[1, 2] | a() | b() # runs
c = a() | b(); [1, 2] | c # doesn't run if this class doesn't exist"""
super().__init__(fs=clis); self.clis = list(clis)
[docs] def __ror__(self, it:Iterator[Any]) -> Iterator[Any]:
super().__ror__(it); clis = self.clis#iter(self.clis)
for cli in clis: it = it | cli
return it
[docs]class oneToMany(BaseCli):
[docs] def __init__(self, *clis:List[BaseCli]):
"""Duplicates 1 stream into multiple streams, each for a cli in the
list. Used in the "a & b" joining operator. See also: :meth:`BaseCli.__and__`"""
super().__init__(fs=clis); self.clis = clis
[docs] def __ror__(self, it:Iterator[Any]) -> Iterator[Iterator[Any]]:
super().__ror__(it); notIterable = False
try: iter(it)
except: notIterable = True
if notIterable or isinstance(it, (list, tuple, torch.Tensor, str, Number, np.number, bool)):
for cli in self.clis: yield cli.__ror__(it)
else:
its = itertools.tee(it, len(self.clis))
for cli, it in zip(self.clis, its): yield cli.__ror__(it)
[docs]class manyToMany(BaseCli):
[docs] def __init__(self, cli:BaseCli):
"""Applies multiple streams to a single cli. Used in the :meth:`BaseCli.all`
operator."""
super().__init__(fs=[cli]); self.cli = cli
[docs] def __ror__(self, it:Iterator[Iterator[Any]]) -> Iterator[Iterator[Any]]:
super().__ror__(it); f = fastF(self.cli)
return (f(s) for s in it)
[docs]class manyToManySpecific(BaseCli):
[docs] def __init__(self, *clis:List[BaseCli]):
"""Applies multiple streams to multiple clis independently. Used in
the "a + b" joining operator. See also: :meth:`BaseCli.__add__`"""
super().__init__(fs=clis); self.clis = list(clis)
[docs] def __ror__(self, its:Iterator[Any]) -> Iterator[Any]:
super().__ror__(its)
for cli, it in zip(self.clis, its): yield cli.__ror__(it)