Source code for k1lib.cli.output

# AUTOGENERATED FILE! PLEASE DON'T EDIT
"""
For operations that feel like the termination
"""
from collections import defaultdict
from typing import Iterator, Any
from k1lib.cli.init import BaseCli, Table
import torch, numbers, numpy as np; from k1lib import cli
__all__ = ["stdout", "file", "pretty", "display", "headOut", "intercept"]
class stdout(BaseCli):
    """Prints out all lines. If not iterable, then print out the input raw"""
    def __ror__(self, it:Iterator[str]):
        try:
            it = iter(it)
            for line in it: print(line)
        except TypeError: print(it)
class file(BaseCli):
    def __init__(self, fileName:str, text:bool=True):
        """Opens a new file for writing.

:param text: if True, accepts Iterator[str], and prints out each string on a
    new line. Else accepts bytes and write in 1 go."""
        super().__init__(); self.fileName = fileName; self.text = text
    def __ror__(self, it:Iterator[str]) -> None:
        super().__ror__(it)
        if self.text:
            with open(self.fileName, "w") as f:
                for line in it: f.write(f"{line}\n")
        else:
            with open(self.fileName, "wb") as f: f.write(it)
class pretty(BaseCli):
    """Pretty prints a table"""
    def __ror__(self, it:Table[Any]) -> Iterator[str]:
        table = []; widths = defaultdict(lambda: 0)
        for row in it:
            _row = []
            for i, e in enumerate(row):
                e = f"{e}"; _row.append(e)
                widths[i] = max(len(e), widths[i])
            table.append(_row)
        for row in table:
            s = ""
            for w, e in zip(widths.values(), row):
                s += e.rstrip(" ").ljust(w+3)
            yield s
def display(lines:int=10):
    """Convenience method for displaying a table"""
    f = pretty() | stdout()
    if lines is None: return f
    else: return cli.head(lines) | f
def headOut(lines:int=10):
    """Convenience method for head() | stdout()"""
    if lines is None: return stdout()
    else: return cli.head(lines) | stdout()
class intercept(BaseCli):
    """Intercept flow at a particular point, analyze the object piped in, and
throw exception to stop flow. Example::

    3 | intercept()
"""
    def __ror__(self, s):
        print(type(s))
        if isinstance(s, (numbers.Number, str, bool)):
            print(s)
        elif isinstance(s, (tuple, list)):
            print(f"Length: {len(s)}")
            for e in s: print(f"- {type(e)}")
        elif isinstance(s, (np.ndarray, torch.Tensor)):
            print(f"Shape: {s.shape}")
        raise RuntimeError("intercepted")