# AUTOGENERATED FILE! PLEASE DON'T EDIT
settings = {"defaultDelim": "\t", "defaultIndent": " ",
"oboFile": None, "strict": False, "lookupImgs": True, "useCtx": False}
def patchDefaultDelim(s:str):
"""
:param s:
- if not None, returns self
- else returns the default delimiter in settings"""
return settings["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 settings"""
return settings["defaultIndent"] if s is None else s
from typing import List, Iterator, Any, NewType, TypeVar, Generic
import k1lib.cli as cli
import itertools, copy, torch
__all__ = ["BaseCli", "serial", "oneToMany", "manyToMany", "manyToManySpecific", "Table", "T"]
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 technically have to call super().__init__() and super().__ror__(),
as those will only dereference any :class:`~k1lib.cli.ctx.Promise` attributes of
the cli object. As that feature is currently underdeveloped, and not that popular anyway,
you can omit it."""
[docs] def __init__(self):
"""Not expected to be instantiated by the end user."""
self._ind_resolved = False
self._ind_initial_scan_finished = False
[docs] def __and__(self, cli:"BaseCli") -> "oneToMany":
"""Duplicates input stream to multiple joined clis."""
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) -> "BaseCli":
"""Applies this cli to all incoming streams"""
return manyToMany(self)
[docs] def __or__(self, it) -> "serial":
"""Joins clis end-to-end"""
if isinstance(it, BaseCli): return serial(self, it)
[docs] def __ror__(self, it):
# Dereferences every :class:`~k1lib.cli.ctx.Promise`
# attributes inside this. Intended to be overridden by subclass
if settings["useCtx"]:
Promise = cli.ctx.Promise
if not self._ind_resolved: # short circuits
if not self._ind_initial_scan_finished:
self._ind_initial_scan_finished = True
# scan for promises, and set _promise_varname instead
for k in [k for k in self.__dict__ if isinstance(getattr(self, k), Promise)]:
v = getattr(self, k)
setattr(self, f"_promise_{k}", v); v = v()
a = [k for k in self.__dict__ if k.startswith("_promise_")]
if len(a) == 0: self._ind_resolved = True
for k in a: setattr(self, k[9:], getattr(self, k)())
[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):
"""Another way to do ``it | cli``"""
return self.__ror__(it)
[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__(); self.clis = clis
[docs] def __ror__(self, it:Iterator[Any]) -> Iterator[Any]:
super().__ror__(it); clis = iter(self.clis)
for cli in clis: it = cli.__ror__(it)
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"""
super().__init__(); self.clis = clis
[docs] def __ror__(self, it:Iterator[Any]) -> Iterator[Iterator[Any]]:
super().__ror__(it);
if isinstance(it, (list, tuple, torch.Tensor)):
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 "a.all()"
operator. Note that this operation will use a different copy of the
cli for each of the streams."""
super().__init__(); self.cli = cli
[docs] def __ror__(self, it:Iterator[Iterator[Any]]) -> Iterator[Iterator[Any]]:
super().__ror__(it)
if settings["useCtx"]:
for stream in it: yield copy.deepcopy(self.cli).__ror__(stream)
else:
for stream in it: yield self.cli.__ror__(stream)
[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 """
super().__init__(); self.clis = 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)