# AUTOGENERATED FILE! PLEASE DON'T EDIT
"""This module for tools that will likely start the processing stream."""
from typing import Iterator, Union, Any
import urllib, subprocess, warnings, os, k1lib, threading
from k1lib.cli import BaseCli; import k1lib.cli as cli
from k1lib.cli.typehint import *
__all__ = ["cat", "curl", "wget", "ls", "cmd", "requireCli"]
def _catSimple(fileName:str=None, text:bool=True, _all:bool=False) -> Iterator[Union[str, bytes]]:
fileName = os.path.expanduser(fileName)
if text:
if _all:
with open(fileName) as f:
lines = f.read().splitlines()
yield lines
else:
with open(fileName) as f:
while True:
line = f.readline()
if line == "": return
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, _all:bool):
res = _catSimple(fileName, text, _all)
return res if text and (not _all) else next(res)
class _cat(BaseCli):
def __init__(self, text, _all:bool): self.text = text; self._all = _all
def _typehint(self, ignored=None): return tIter(str)
def __ror__(self, fileName:str) -> Union[Iterator[str], bytes]:
return _catWrapper(fileName, self.text, self._all)
[docs]def cat(fileName:str=None, text:bool=True, _all=False):
"""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()
# recommended to insert a `tOpt()` in the middle and `yieldT` in the end, to do lots of optimizations
"file.txt" | tOpt() | cat() | headOut() | yieldT
# rename file
cat("img.png", False) | file("img2.png")
: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
:param _all: if True, read entire file at once, instead of reading
line-by-line. Faster, but uses more memory. Only works with text
mode, binary mode always read the entire file"""
if fileName is None: return _cat(text, _all)
else: return _catWrapper(fileName, text, _all)
[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. Also returns the file name, in case you want to pipe it
to something else.
: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)
return 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") | filt(os.path.isfile)"""
if folder is None: return _ls()
else: return folder | _ls()
class _ls(BaseCli):
def _typehint(self, ignored=None): return tList(str)
def __ror__(self, folder:str):
folder = os.path.expanduser(folder.rstrip(os.sep))
return [f"{folder}{os.sep}{e}" for e in os.listdir(folder)]
k1lib.settings.cli.add("quiet", False, "whether to mute extra outputs from clis or not")
newline = b'\n'[0]
class lazySt:
def __init__(self, st, text:bool):
"""Converts byte stream into lazy text/byte stream, with nice __repr__."""
self.st = st; self.text = text;
def __iter__(self):
f = (lambda x: x.decode("utf-8")) if self.text else (lambda x: x)
while True:
line = self.st.readline()
if len(line) == 0: break
yield f(line[:-1]) if line[-1] == newline else f(line)
def __repr__(self): self | cli.stdout(); return ""
def executeCmd(cmd:str, inp:bytes, text):
"""Runs a command, and returns stdout and stderr streams"""
p = subprocess.Popen(cmd, shell=True, stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE, cwd=k1lib.settings.wd)
if inp is not None: p.stdin.write(inp)
p.stdin.close(); return lazySt(p.stdout, text), lazySt(p.stderr, text)
def printStderr(err):
if not k1lib.settings.cli.quiet:
e, it = err | cli.peek()
if it != []: it | cli.insert("\nError encountered:\n") | cli.apply(k1lib.fmt.txt.red) | cli.stdout()
[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.""")
[docs]class cmd(BaseCli):
[docs] def __init__(self, cmd:str, mode:int=1, text=True, block=False): # 0: return (stdout, stderr). 1: return stdout, 2: return stderr
"""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 :data:`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 print out the outputs. 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.
Let's see a few more exotic examples. File ``a.sh``:
.. code-block:: bash
#!/bin/bash
echo 1; sleep 0.5
echo This message goes to stderr >&2
echo 2; sleep 0.5
echo $(</dev/stdin)
sleep 0.5; echo 3
Examples::
# returns [b'1', b'2', b'45', b'3'] and prints out the error message
"45" | cmd("./a.sh", text=False) | deref()
# returns [b'This message goes to stderr']
"45" | cmd("./a.sh", mode=2, text=False) | deref()
# returns [[b'1', b'2', b'45', b'3'], [b'This message goes to stderr']]
"45" | cmd("./a.sh", mode=0, text=False) | deref()
Performance-wise, stdout and stderr will yield values right away as soon
as the process outputs it, so you get real time feedback. However, this will
convert the entire input into a :class:`bytes` object, and not feed it bit by
bit lazily, so if you have a humongous input, it might slow you down a little.
Also, because stdout and stderr yield values right away, it means that if you
want the operation to be blocking until finished, you have to consume the output::
None | cmd("mkdir abc")
# might fail, because this might get executed before the previous line
None | cmd("echo a>abc/rg.txt")
None | cmd("mkdir abc") | ignore()
# will succeed, because this will be guaranteed to execute after the previous line
None | cmd("echo a>abc/rg.txt")
Settings:
- cli.quiet: if True, won't display errors in mode 1
:param mode: if 0, returns ``(stdout, stderr)``. If 1, returns ``stdout`` and prints
``stderr`` if there are any errors. If 2, returns ``stderr``
:param text: whether to decode the outputs into :class:`str` or return raw :class:`bytes`
:param block: whether to wait for the task to finish before returning to Python or not"""
super().__init__(); self.cmd = cmd; self.mode = mode
self.text = text; self.block = block; self.ro = k1lib.RunOnce()
def _typehint(self, ignored=None):
t = tIter(str) if self.text else tIter(bytes)
if self.mode == 0: return tCollection(t, t)
return t
[docs] def __ror__(self, it:Union[None, str, bytes, Iterator[Any]]) -> Iterator[Union[str, bytes]]:
"""Pipes in lines of input, or if there's nothing to
pass, then pass None"""
if not self.ro.done():
if it != None:
if not isinstance(it, (str, bytes)): it = it | cli.toStr() | cli.join("\n")
if not isinstance(it, bytes): it = it.encode("utf-8")
self.out, self.err = executeCmd(self.cmd, it, self.text); mode = self.mode
if self.block:
self.out = self.out | cli.deref()
self.err = self.err | cli.deref()
if mode == 0: return (self.out, self.err)
elif mode == 1:
threading.Thread(target=lambda: printStderr(self.err)).start()
return self.out
elif mode == 2: return self.err
def __gt__(self, it): return None | self | it
def __repr__(self):
return (None | self).__repr__()