# 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 k1lib.bioinfo.cli as cli
import itertools, copy
__all__ = ["BaseCli", "serial", "oneToMany", "manyToMany", "manyToManySpecific"]
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]".""")
class Row(list):
"""Not really used currently. Just here for potential future feature"""
pass
T = TypeVar("T")
[docs]class BaseCli:
[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.bioinfo.cli.ctx.Promise`
# attributes inside this. Intended to be overridden by subclass
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)]:
setattr(self, f"_promise_{k}", v := getattr(self, k)); 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):
"""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)
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"""
super().__init__(); self.clis = clis
[docs] def __ror__(self, it:Iterator[Any]) -> Iterator[Iterator[Any]]:
super().__ror__(it); 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)
for stream in it: yield copy.deepcopy(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)