# AUTOGENERATED FILE! PLEASE DON'T EDIT
settings = {"defaultDelim": "\t", "defaultIndent": " ",
"oboFile": None, "strict": False, "lookupImgs": True}
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
import itertools
__all__ = ["serial", "oneToMany", "manyToMany", "manyToManySpecific"]
class _MetaType(type):
def __getitem__(self, generic):
d = {"__args__": generic, "_n": self._n, "__doc__": self.__doc__}
return _MetaType("MetaObj", (), 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]".""")
class Row(list):
"""Not really used currently. Just here for potential future feature"""
pass
T = TypeVar("T")
class BaseCli:
def __and__(self, cli:"BaseCli"):
if isinstance(self, oneToMany):
self.clis.append(cli); return self
if isinstance(cli, oneToMany):
cli.clis.append(self); return cli
return oneToMany(self, cli)
def __add__(self, cli):
if isinstance(self, manyToManySpecific):
self.clis.append(cli); return self
if isinstance(cli, manyToManySpecific):
cli.clis.append(self); return cli
return manyToManySpecific(self, cli)
def all(self) -> "BaseCli":
"""Applies this BaseCli to all incoming streams"""
return manyToMany(self)
def __or__(self, it):
if isinstance(it, BaseCli):
return serial(self, it)
def f(self):
"""Creates a normal function :math:`f(x)` which is equivalent to
``x | self``."""
return lambda it: self.__ror__(it)
def __lt__(self, it):
"""Default backup join symbol `>`, in case `it` implements __ror__()"""
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
"""
self.clis = clis
[docs] def __ror__(self, it:Iterator[Any]) -> Iterator[Any]:
for cli in self.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"""
self.clis = clis
[docs] def __ror__(self, it:Iterator[Any]) -> Iterator[Iterator[Any]]:
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):
"""Applies multiple streams to a single cli. Used in the "a.all()"
operator."""
self.cli = cli
[docs] def __ror__(self, it:Iterator[Iterator[Any]]) -> Iterator[Iterator[Any]]:
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 """
self.clis = clis
[docs] def __ror__(self, its:Iterator[Any]) -> Iterator[Any]:
for cli, it in zip(self.clis, its):
yield cli.__ror__(it)