Source code for k1lib.cli.inp

# AUTOGENERATED FILE! PLEASE DON'T EDIT
"""This module for tools that will likely start the processing stream."""
from typing import Iterator, Union
import urllib, subprocess, warnings, os
from k1lib.cli import BaseCli
import k1lib.cli as cli
__all__ = ["cat", "cats", "curl", "wget", "ls", "cmd", "requireCli", "toPIL"]
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.
Example::

    # display first 10 lines of file
    cat("file.txt") | headOut()
    # piping in also works
    "file.txt" | cat() | headOut()

:param fileName: if None, then return a :class:`~k1lib.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))

Example::

    # prints out first 10 lines of 2 files
    ["file1.txt", "file2.txt"] | cats() | headOut().all() | ignore()"""
    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. File can't be a binary blob.
Example::

    # prints out first 10 lines of the website
    curl("https://k1lib.github.io/") | headOut()"""
    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 ls(folder:str=None):
    """List every file and folder inside the specified folder.
Example::

    # returns List[str]
    ls("/home")
    # same as above
    "/home" | ls()
    # only outputs files, not folders
    ls("/home") | isFile()

See also: :meth:`~k1lib.cli.filt.isFile`"""
    if folder is None: return _ls()
    else: return folder | _ls()
class _ls(BaseCli):
    def __ror__(self, folder:str):
        return [f"{folder}/{e}" for e in os.listdir(folder)]
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.
Example::

    # return detailed list of files
    None | cmd("ls -la")
    # return list of files that ends with "ipynb"
    None | cmd("ls -la") | cmd('grep ipynb$')"""
        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"""
        super().__ror__(it)
        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.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.""")
class toPIL(BaseCli):
    def __init__(self):
        """Converts a path to a PIL image.
Example::

    ls(".") | toPIL().all() | item() # get first image"""
        import PIL; self.PIL = PIL
    def __ror__(self, path):
        return self.PIL.Image.open(path)