# 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.bioinfo.cli.init import BaseCli
import k1lib.bioinfo.cli as cli
__all__ = ["cat", "cats", "curl", "wget", "ls", "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)
[docs]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.bioinfo.cli.init.BaseCli`
that accepts a file name and outputs Iterator[str]"""
if fileName is None: return _cat()
else: return _catSimple(fileName)
[docs]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()"""
[docs] def __ror__(self, fileNames:Iterator[str]) -> Iterator[Iterator[str]]:
for fileName in fileNames:
yield _catSimple(fileName)
[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/") | cli.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, dirs=True, files=True):
"""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", dirs=False)
# same as above
"/home" | ls(dirs=False)"""
if folder is None: return _ls(dirs, files)
else: return folder | _ls(dirs, files)
class _ls(BaseCli):
def __init__(self, dirs:bool, files:bool):
self.dirs = dirs; self.files = files
def __ror__(self, folder:str):
answer = (f"{folder}/{e}" for e in os.listdir(folder))
if not self.dirs: answer = (f for f in answer if not os.path.isdir(f))
if not self.files: answer = (f for f in answer if not os.path.isfile(f))
return list(answer)
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
[docs]class cmd(BaseCli):
[docs] 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
[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)
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."""
[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]def infiniteF(f):
"""Essentially just ``while True: yield f()``."""
while True: yield f()