# AUTOGENERATED FILE! PLEASE DON'T EDIT
from typing import List, Iterator, Any, NewType, TypeVar, Generic
import k1lib.cli as cli; from numbers import Number
import k1lib, itertools, copy, torch; import numpy as np
__all__ = ["BaseCli", "Table", "T", "fastF",
"serial", "oneToMany", "manyToMany", "mtmS"]
settings = k1lib.Settings()
atomic = k1lib.Settings()
settings.add("atomic", atomic, "classes/types that are considered atomic and specified cli tools should never try to iterate over them")
settings.add("defaultDelim", "\t", "default delimiter used in-between columns when creating tables. Defaulted to tab character.")
settings.add("defaultIndent", " ", "default indent used for displaying nested structures")
settings.add("strict", False, "turning it on can help you debug stuff, but could also be a pain to work with")
settings.add("inf", float("inf"), "infinity definition for many clis. Here because you might want to temporarily not loop things infinitely")
k1lib.settings.add("cli", settings, "from k1lib.cli module")
def patchDefaultDelim(st:str):
"""
:param s:
- if not None, returns self
- else returns the default delimiter in :attr:`~k1lib.settings`"""
return settings.defaultDelim if st is None else st
def patchDefaultIndent(st:str):
"""
:param s:
- if not None, returns self
- else returns the default indent character in :attr:`~k1lib.settings`"""
return settings.defaultIndent if st is None else st
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): return self._after(cli)
if isinstance(cli, oneToMany): return cli._before(self)
return oneToMany(self, cli)
[docs] def __add__(self, cli:"BaseCli") -> "mtmS":
"""Parallel pass multiple streams to multiple clis."""
if isinstance(self, mtmS): return self._after(cli)
if isinstance(cli, mtmS): return cli._before(self)
return mtmS(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): return self._after(cli)
if isinstance(cli, serial): return cli._before(self)
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])
[docs]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. Example::
# both returns 16, fastF returns "lambda x: x**2", so it's really fast
fastF(op()**2)(4)
fastF(applyS(lambda x: x**2))(4)"""
if isinstance(c, cli.op): return c.ab_fastF()
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); self._cache()
def _cache(self):
self._hasTrace = any(isinstance(c, cli.trace) for c in self.clis)
self._cliCs = [fastF(c) for c in self.clis]; return self
[docs] def __ror__(self, it:Iterator[Any]) -> Iterator[Any]:
if self._hasTrace: # slower, but tracable
for cli in self.clis: it = it | cli
else: # faster, but not tracable
for cli in self._cliCs: it = cli(it)
return it
def _before(self, c): self.clis = [c] + self.clis; return self._cache()
def _after(self, c): self.clis = self.clis + [c]; return self._cache()
atomic.add("baseAnd", (Number, np.number, str, dict, bool, bytes, list, tuple, torch.Tensor), "used by BaseCli.__and__")
def _iterable(it):
try: iter(it); return True
except: return False
[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; self._cache()
[docs] def __ror__(self, it:Iterator[Any]) -> Iterator[Iterator[Any]]:
if isinstance(it, atomic.baseAnd) or not _iterable(it):
for cli in self._cliCs: yield cli(it)
else:
its = itertools.tee(it, len(self.clis))
for cli, it in zip(self._cliCs, its): yield cli(it)
def _cache(self): self._cliCs = [fastF(c) for c in self.clis]; return self
def _before(self, c): self.clis = [c] + self.clis; return self._cache()
def _after(self, c): self.clis = self.clis + [c]; return self._cache()
[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; self._cliC = fastF(cli)
[docs] def __ror__(self, it:Iterator[Iterator[Any]]) -> Iterator[Iterator[Any]]:
f = self._cliC; return (f(s) for s in it)
[docs]class mtmS(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__`.
Weird name is actually a shorthand for "many to many specific"."""
super().__init__(fs=clis); self.clis = list(clis); self._cache()
def _cache(self): self._cliCs = [fastF(c) for c in self.clis]; return self
def _before(self, c): self.clis = [c] + self.clis; return self._cache()
def _after(self, c): self.clis = self.clis + [c]; return self._cache()
[docs] def __ror__(self, its:Iterator[Any]) -> Iterator[Any]:
for cli, it in zip(self._cliCs, its): yield cli(it)
[docs] @staticmethod
def f(f, i:int, n:int=100):
"""Convenience method, so
that this::
mtmS(iden(), op()**2, iden(), iden(), iden())
# also the same as this btw:
(iden() + op()**2 + iden() + iden() + iden())
is the same as this::
mtmS.single(op()**2, 1, 5)
Example::
# returns [5, 36, 7, 8, 9]
range(5, 10) | mtmS.single(op()**2, 1, 5) | deref()
:param i: where should I put the function?
:param n: how many clis in total? Defaulted to 100"""
return mtmS(*([cli.iden()]*i + [f] + [cli.iden()]*(n-i-1)))