Source code for k1lib.bioinfo.cli.inp

# AUTOGENERATED FILE! PLEASE DON'T EDIT
from typing import Iterator, Union
import urllib, subprocess, warnings
from k1lib.bioinfo.cli.init import BaseCli
import k1lib.bioinfo.cli as cli
__all__ = ["cat", "cats", "curl", "wget", "cmd", "requireCli", "infiniteF"]
def _catSimple(fileName:str=None) -> Iterator[str]:
    with open(fileName) as f:
        for line in f.readlines():
            if line[-1] == "\n": yield line[:-1]
            else: yield line
class _cat(BaseCli):
    def __ror__(self, fileName:str) -> Iterator[str]:
        return _catSimple(fileName)
def cat(fileName:str=None):
    """Reads a file line by line.

:param fileName: if None, then return a :class:`~k1lib.bioinfo.cli.init.BaseCli`
    that accepts a file name and outputs Iterator[str]"""
    if fileName is None: return _cat()
    else: return _catSimple(fileName)
class cats(BaseCli):
    """Like :meth:`cat`, but opens multiple files at once, returning
streams. Looks something like this::

    apply(lambda s: cat(s))"""
    def __ror__(self, fileNames:Iterator[str]) -> Iterator[Iterator[str]]:
        for fileName in fileNames:
            yield _catSimple(fileName)
def curl(url:str) -> Iterator[str]:
    """Gets file from url"""
    for line in urllib.request.urlopen(url):
        line = line.decode()
        if line[-1] == "\n": yield line[:-1]
        else: yield line
def wget(url:str, fileName:str=None):
    """Downloads a file

:param url: The url of the file
:param fileName: if None, then tries to infer it from the url"""
    if fileName is None: fileName = url.split("/")[-1]
    urllib.request.urlretrieve(url, fileName)
def executeCmd(cmd:str, inp:str=None):
    """Runs a command, and returns the output line by line"""
    if inp is None:
        process = subprocess.Popen(cmd, shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
        out, err = process.communicate()
    else:
        process = subprocess.Popen(cmd, shell=True, stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
        out, err = process.communicate(input=inp.encode("utf-8"))
    return out.decode().split("\n"), err
class cmd(BaseCli):
    def __init__(self, cmd:str):
        """Runs a command, and returns the output line by line."""
        super().__init__(); self.cmd = cmd; self.err = b''
    @property
    def err(self) -> bytes:
        """Error from the last command"""
        return self._err
    @err.setter
    def err(self, value): self._err = value
    def __ror__(self, it:Union[Iterator[str], None]) -> Iterator[str]:
        """Pipes in lines of input, or if there's nothing to
pass, then pass None"""
        out, err = executeCmd(self.cmd) if it is None else executeCmd(self.cmd, it | cli.to1Str("\n") | cli.item())
        if err: warnings.warn(f"Error encountered:\n\n{err.decode()}")
        self._err = err; return out
    def __repr__(self):
        return """`k1lib.bioinfo.cli.cmd` instance. Pipes
input generator or None in to execute."""
def requireCli(cliTool:str):
    """Searches for a particular cli tool (eg. "ls"), throws ImportError
if not found, else do nothing"""
    a = cmd(cliTool); None | a;
    if len(a.err) > 0:
        raise ImportError(f"""Can't find cli tool {cliTool}. Please install
it first.""")
def infiniteF(f):
    """Essentially just ``while True: yield f()``."""
    while True: yield f()