printer.py

#!/usr/bin/env python
#coding=utf8

'''
CloudControl CLI Printer module
'''

import cccli
from cccli.exception import *

import sys
import os
import termios
import fcntl
import struct
import signal

color = {
    # regular
    "black": "\033[0;30m",
    "red": "\033[0;31m",
    "green": "\033[0;32m",
    "yellow": "\033[0;33m",
    "blue": "\033[0;34m",
    "purple": "\033[0;35m",
    "cyan": "\033[0;36m",
    "white": "\033[0;37m",
    # lighted
    "lgrey": "\033[1;30m",
    "lred": "\033[1;31m",
    "lgreen": "\033[1;32m",
    "lyellow": "\033[1;33m",
    "lblue": "\033[1;34m",
    "lpurple": "\033[1;35m",
    "lcyan": "\033[1;36m",
    "lwhite": "\033[1;37m",
    # underline
    "ured": "\033[4;31m",
    "ugreen": "\033[4;32m",
    "uyellow": "\033[4;33m",
    "ublue": "\033[4;34m",
    "upurple": "\033[4;35m",
    "ucyan": "\033[4;36m",
    # others
    "light": "\033[1m",
    "reset": "\033[m",
    "[": "\001",
    "]": "\002",
    }


class Printer(object):
    '''Print relative class'''
    def __init__(self):
        self.readline = None
        self.history = History()
        self.completion = Completion()

    def set_interactive(self):
        '''Set interactive mode'''
        if self.readline is not None:
            return
        try:
            import readline
        except Exception as e:
            raise cliError("Unable to start readline")
        self.readline = readline
        # enable history
        self.history.readline = readline
        # enable completion
        self.completion.readline = readline

    def out(self, message="", fd=None, nl=os.linesep, flush=True):
        '''Print a message in fd ended by nl'''
        if fd is None:
            fd = sys.stdout
        fd.write("%s%s"%(message, nl))
        if flush:
            fd.flush()

    def err(self, message, fd=None, nl=os.linesep):
        if fd is None:
            fd = sys.stderr
        self.out(message, fd, nl)

    def fatal(self, message, quit=True, fd=None, nl=os.linesep):
        self.err("%sFatal%s: %s%s"%(color["lred"],color["red"],message, color["reset"]),
                 fd,
                 nl)
        if quit:
            os._exit(42)

    def error(self, message, fd=None, nl=os.linesep):
        self.err("%sError%s: %s%s"%(color["lred"],color["red"],message,color["reset"]),
                 fd,
                 nl)

    def warn(self, message, fd=None, nl=os.linesep):
        self.err("%sWarning%s: %s%s"%(color["lyellow"],color["yellow"],message,color["reset"]),
                 fd,
                 nl)

    def debug(self, message, fd=None, nl=os.linesep):
        if cccli.debug:
            self.err("%s%s%s"%(color["lgrey"],message,color["reset"]), fd, nl)

    def isinteractive(self):
        return self.readline is not None

    def interactive(self, message, fd=None, nl=os.linesep):
        '''Print only in interactive mode'''
        if self.readline is not None:
            self.err(message, fd, nl)

    def getline(self, prompt):
        '''Read a line from stdin'''
        try:
            s = raw_input(prompt)
        except EOFError:
            raise
        except KeyboardInterrupt:
            raise
        except Exception as e:
            raise cliError(str(e))
        return s

    def getpass(self, prompt):
        '''Ask for a password. No echo. Not in history'''
        if self.readline is None:
            raise cliError("Unable to ask a password in non-interactive mode")
        fd = sys.stdin.fileno()
        old = termios.tcgetattr(fd)
        new = termios.tcgetattr(fd)
        new[3] = new[3] & ~termios.ECHO
        h = list(self.history)
        self.history.clear()
        try:
            termios.tcsetattr(fd, termios.TCSADRAIN, new)
            passwd = self.getline(prompt)
        finally:
            termios.tcsetattr(fd, termios.TCSADRAIN, old)
            self.history.load(h)
        sys.stdout.write(os.linesep)
        sys.stdout.flush()
        return passwd

    def ask(self, prompt):
        '''Used to ask a question. Default answer not saved to history'''
        if self.readline is None:
            raise cliError("Unable to ask question in non-interactive mode")
        h = list(self.history)
        self.history.clear()
        r = ""
        try:
            r = self.getline(prompt)
        except EOFError:
            pass
        except KeyboardInterrupt:
            pass
        finally:
            self.history.load(h)
        return r

    def get_term_size(self):
        '''Return terminal size'''
        if self.readline is None:
            raise cliError("Unable to get term size in non-interactive mode")
        req = struct.pack("HHHH", 0, 0, 0, 0)
        resp = fcntl.ioctl(sys.stdin.fileno(), termios.TIOCGWINSZ, req)
        rows, cols, px_x, px_y = struct.unpack("HHHH", resp)
        return rows, cols

    def clear(self):
        '''Clear the terminal'''
        if self.readline is None:
            return
        self.out("\033[H\033[2J", nl="")


class History(object):
    '''History class'''
    def __init__(self):
        self.readline = None

    def __nonzero__(self):
        return self.readline is not None

    def __getattribute__(self, name):
         r = object.__getattribute__(self, "readline")
         if name == "readline":
             return r
         if r is None:
             return lambda *a,**k: None
         return object.__getattribute__(self, name)

    def __iter__(self):
        if self.readline is None:
            return
        for i in range(1, self.readline.get_current_history_length() + 1):
            yield self.readline.get_history_item(i)

    def __len__(self):
        if self.readline is None:
            return 0
        return self.readline.get_current_history_length()

    def load(self, items):
        '''Load history from a list'''
        self.clear()
        for l in items:
            self.readline.add_history(l)

    def read(self, path):
        '''Load history from a file'''
        self.clear()
        if path is None:
            return
        try:
            self.readline.read_history_file(path)
        except IOError:
            pass

    def write(self, path):
        '''Save history into path'''
        if path is None:
            return
        try:
            self.readline.write_history_file(path)
        except IOError:
            pass

    def maxsize(self, size=None):
        '''Set or return max history size'''
        if size is not None:
            self.readline.set_history_length(size)
        return self.readline.get_history_length()

    def removelast(self):
        '''Remove last history line'''
        self.readline.remove_history_item(self.readline.get_current_history_length() - 1)

    def clear(self):
        '''Clear history'''
        self.readline.clear_history()


class Completion(object):
    '''Handle completion functions'''
    def __init__(self):
        self.readline = None
        self.compfunc = None
        self.complist = list()

    def __getattribute__(self, name):
         r = object.__getattribute__(self, "readline")
         if name == "readline":
             return r
         if r is None:
             return lambda *a,**k: None
         return object.__getattribute__(self, name)

    def get_buf(self):
        '''Return current readline buffer'''
        return self.readline.get_line_buffer()

    def get_begin(self):
        '''Get the beginning index of the readline tab-completion scope'''
        return self.readline.get_begidx()

    def get_end(self):
        '''Get the ending index of the readline tab-completion scope'''
        return self.readline.get_begidx()

    def _completer(self, text, state):
        '''Readline real completer'''
        if state == 0:
            if self.compfunc is not None:
                self.complist = list(self.compfunc(text))
        try:
            return self.complist[state]
        except IndexError:
            self.complist = list()
            return None

    def set_completer(self, func):
        '''Set completer custom function which return a list of possibilities'''
        self.compfunc = func
        self.readline.set_completer(self._completer)
        self.readline.parse_and_bind("tab: complete")
        self.readline.set_completer_delims(" \t")