# 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, k1lib
from k1lib.cli import BaseCli; import k1lib.cli as cli
__all__ = ["cat", "curl", "wget", "ls", "cmd", "requireCli"]
def _catSimple(fileName:str=None, text:bool=True) -> Iterator[Union[str, bytes]]:
fileName = os.path.expanduser(fileName)
if text:
with open(fileName) as f:
for line in f.readlines():
if line[-1] == "\n": yield line[:-1]
else: yield line
else:
with open(fileName, "rb") as f: yield f.read()
def _catWrapper(fileName:str, text:bool):
res = _catSimple(fileName, text)
return res if text else next(res)
class _cat(BaseCli):
def __init__(self, text): self.text = text
def __ror__(self, fileName:str) -> Union[Iterator[str], bytes]:
return _catWrapper(fileName, self.text)
[docs]def cat(fileName:str=None, text:bool=True):
"""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()
# rename file
cat("img.png", False) | file("img2.png", False)
:param fileName: if None, then return a :class:`~k1lib.cli.init.BaseCli`
that accepts a file name and outputs Iterator[str]
:param text: if True, read text file, else read binary file"""
if fileName is None: return _cat(text)
else: return _catWrapper(fileName, text)
[docs]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
[docs]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)
[docs]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):
folder = os.path.expanduser(folder)
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, cwd=k1lib.settings.wd)
out, err = process.communicate()
else:
process = subprocess.Popen(cmd, shell=True, stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE, cwd=k1lib.settings.wd)
out, err = process.communicate(input=inp.encode("utf-8"))
return out.decode().split("\n"), err
[docs]class cmd(BaseCli):
[docs] def __init__(self, cmd:str):
"""Runs a command, and returns the output line by line. Can pipe in some
inputs. If no inputs then have to pipe in :data:`None`. 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$')
It might be tiresome to pipe in None all the time. So, you can use ">" operator
to yield values right away::
# prints out first 10 lines of list of files
cmd("ls -la") > headOut()
If you're using Jupyter notebook/lab, then if you were to display a :class:`cmd`
object, it will show old outputs (or executes and show new outputs if it has not
executed yet). So, a single command ``cmd("mkdir")`` displayed at the end of a
cell is enough to trigger creating the directory.
Reminder that ">" operator in here sort of has a different meaning to that of
:class:`~k1lib.cli.init.BaseCli`. So you kinda have to becareful about this::
# returns a serial cli, cmd not executed
cmd("ls -la") | deref()
# executes cmd with no input stream and pipes output to deref
cmd("ls -la") > deref()
# returns a serial cli
cmd("ls -la") > grep("txt") > headOut()
# executes pipeline
cmd("ls -la") > grep("txt") | headOut()
General advice is, right ater a :class:`cmd`, use ">", and use "|" everywhere else."""
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
[docs] 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)
self.out, err = executeCmd(self.cmd) if it is None else executeCmd(self.cmd, it | cli.join("\n") | cli.item())
if err: print(f"Error encountered:\n\n{err.decode()}")
self._err = err; return self.out
def _out(self):
if not hasattr(self, "out"): self.__ror__(None)
return self.out
def __gt__(self, it): return self._out() | it
def __repr__(self): return "\n".join(self._out())
[docs]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.""")