Package ivs :: Package aux :: Module termtools
[hide private]
[frames] | no frames]

Source Code for Module ivs.aux.termtools

  1  """ 
  2  Tools for cursor and color control in the terminal 
  3   
  4  Example usage (this does not work in the documentation): 
  5   
  6  >>> print green('this text is green') 
  7  this text is green 
  8  >>> print blink('this text blinks')+' this text does not blink' 
  9  this text blinks this text does not blink 
 10   
 11  You can combine all the possibilities, as the functions are generated on the fly: 
 12   
 13  >>> print blink_green_bgred_bold('blinking green bold text on red background') 
 14  blinking green bold text on red background 
 15       
 16  """ 
 17  from __future__ import print_function 
 18  import functools 
 19  import inspect 
 20  import sys 
 21  import types 
 22  import subprocess 
 23  import time 
 24  RED="\[\033[0;35m\]" 
 25  YELLOW="\[\033[0;33m\]" 
 26  instructs = {'black':"\033[30m", 
 27            'red':"\033[31m", 
 28            'green':"\033[32m", 
 29            'yellow':"\033[33m", 
 30            'blue':"\033[34m", 
 31            'magenta':"\033[35m", 
 32            'cyan':"\033[36m", 
 33            'white':"\033[37m", 
 34            'bgblack':"\033[40m", 
 35            'bgred':"\033[41m", 
 36            'bggreen':"\033[42m", 
 37            'bgyellow':"\033[43m", 
 38            'bgblue':"\033[44m", 
 39            'bgmagenta':"\033[45m", 
 40            'bgcyan':"\033[46m", 
 41            'bgwhite':"\033[47m", 
 42            'blink':'\033[5m', 
 43            'underline':'\033[4m', 
 44            'bold':'\033[1m', 
 45            'reset':'\033[m'} 
 46   
47 -def overwrite_line(message):
48 """ 49 Save cursor at current position, clear current line, print the message and reset the cursor. 50 51 @param message: message to print to the screen 52 @type message: str 53 """ 54 ESC=chr(27) 55 print('{ESC}[s{ESC}[2K{message}{ESC}[u'.format(ESC=ESC,message=message),end='')
56 57
58 -def line_at_a_time(fileobj):
59 """ 60 Return one line at a time from a file-like object. 61 62 >>> #p1 = subprocess.Popen('ls',shell=True,stdout=subprocess.PIPE) 63 >>> #for line in line_at_a_time(p1.stdout): 64 ... # print 'next line:',line.strip() 65 >>> #retcode = p1.wait() 66 67 use C{os.kill(p1.pid,SIGKILL)} to with C{SIGKILL} from C{signal} standard 68 module to kill the process. 69 70 return model_number 71 Works around the iter behavior of pipe files in 72 Python 2.x, e.g., instead of "for line in file" you can 73 write "for line in line_at_a_time(file)" 74 """ 75 while True: 76 line = fileobj.readline() 77 if not line: 78 return 79 yield line
80 81
82 -def subprocess_timeout(command, time_out):
83 """ 84 85 Kill a running subprocess after a certain amount of time. 86 87 Command represents the command for the process you would give in a terminal e.g. 'ls -l' in a terminal becomes ["ls", "-l"] or 'firefox' becomes ["firefox"]'; time_out is expressed in seconds. If the process did not complete before time_out, the process is killed. 88 89 @param command: command to run 90 @type command: str 91 """ 92 93 # launching the command 94 c = subprocess.Popen(command) 95 96 # now waiting for the command to complete 97 t = 0 98 while t < time_out and c.poll() is None: 99 time.sleep(1) # (comment 1) 100 t += 1 101 102 # there are two possibilities for the while to have stopped: 103 if c.poll() is None: 104 # in the case the process did not complete, we kill it 105 c.terminate() 106 # and fill the return code with some error value 107 returncode = -1 # (comment 2) 108 109 else: 110 # in the case the process completed normally 111 returncode = c.poll() 112 113 return returncode
114
115 -class CallInstruct:
116 """ 117 Generate a callable function on-the-fly. 118 119 The function takes text as an input and will first print all the terminal 120 instructions, then the text and then reset the terminal settings to the 121 normal value. 122 """
123 - def __init__(self,instr):
124 """ 125 Remember which instructions to print. 126 """ 127 self.instr = instr
128 - def __call__(self,text):
129 """ 130 Print the instructions, the text and reset the terminal. 131 """ 132 return "".join([instructs[i] for i in self.instr.split('_')]) + text + '\033[m'
133 134 135
136 -class MemoryMonitor(object):
137
138 - def __init__(self, username=None):
139 """Create new MemoryMonitor instance.""" 140 self.username = username 141 self.time = [] 142 self.free = [] 143 self.used = []
144
145 - def usage(self):
146 """Return int containing memory used by user's processes.""" 147 #self.process = subprocess.Popen("ps -u %s -o rss | awk '{sum+=$1} END {print sum}'" % self.username, 148 # shell=True, 149 # stdout=subprocess.PIPE, 150 # ) 151 process = subprocess.Popen("free -mto",shell=True,stdout=subprocess.PIPE) 152 self.time.append(time.time()) 153 used,free = process.communicate()[0].split('\n')[1].split()[2:4] 154 self.free.append(free) 155 self.used.append(used)
156 157 158 159
160 -class Wrapper(object):
161 """ 162 Wrap the module so that the C{getattr} function can be redefined. 163 """
164 - def __init__(self, wrapped):
165 self.wrapped = wrapped
166 - def __getattr__(self, name):
167 # Perform custom logic here 168 try: 169 return getattr(self.wrapped, name) 170 except AttributeError: 171 return CallInstruct(name) # Some sensible default
172 173 #-- Wrap the module so that the C{getattr} function can be redefined. 174 sys.modules[__name__] = Wrapper(sys.modules[__name__]) 175