Source code for k1lib.bioinfo.cli.init

# 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
import itertools
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)
class serial(BaseCli):
    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
    def __ror__(self, it:Iterator[Any]) -> Iterator[Any]:
        for cli in self.clis: it = cli.__ror__(it)
        return it
class oneToMany(BaseCli):
    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
    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)
class manyToMany(BaseCli):
    def __init__(self, cli):
        """Applies multiple streams to a single cli. Used in the "a.all()"
operator."""
        self.cli = cli
    def __ror__(self, it:Iterator[Iterator[Any]]) -> Iterator[Iterator[Any]]:
        for stream in it: yield self.cli.__ror__(stream)
class manyToManySpecific(BaseCli):
    def __init__(self, *clis:List[BaseCli]):
        """Applies multiple streams to multiple clis independently. Used in
the "a + b" joining operator """
        self.clis = clis
    def __ror__(self, its:Iterator[Any]) -> Iterator[Any]:
        for cli, it in zip(self.clis, its):
            yield cli.__ror__(it)