diff options
| author | Ultrabug <ultrabug@gentoo.org> | 2013-02-23 18:43:08 +0100 |
|---|---|---|
| committer | Ultrabug <ultrabug@gentoo.org> | 2013-02-23 18:43:08 +0100 |
| commit | 85a0f4e42d4a9333c34326121e01d1b8fe306b9d (patch) | |
| tree | a136eb152fee0aa8f59f3600479347825fde299c /py3status/py3status.py | |
| parent | 3eaeaf2082d163c22a72f824641e2d1c06c4a9d8 (diff) | |
move to a proper setuptools installation, update the doc
Diffstat (limited to 'py3status/py3status.py')
| -rwxr-xr-x | py3status/py3status.py | 256 |
1 files changed, 0 insertions, 256 deletions
diff --git a/py3status/py3status.py b/py3status/py3status.py deleted file mode 100755 index 8c22fff..0000000 --- a/py3status/py3status.py +++ /dev/null @@ -1,256 +0,0 @@ -#!/usr/bin/python - -""" py3status wraps the mighty i3status for enhanced i3bar customization. """ - -# includes -################################################################################ -import os -import imp -import sys -import argparse -import threading - -from json import loads -from json import dumps - -from time import time -from time import sleep - -from datetime import datetime -from datetime import timedelta - -from Queue import Queue -from Queue import Empty - -from syslog import syslog -from syslog import LOG_ERR - -# functions -################################################################################ -def print_line(message): - """ - Non-buffered printing to stdout - """ - sys.stdout.write(message + '\n') - sys.stdout.flush() - -def read_line(): - """ - Interrupted respecting reader for stdin - """ - try: - line = sys.stdin.readline().strip() - # i3status sends EOF, or an empty line - if not line: - sys.exit(3) - return line - except KeyboardInterrupt: - sys.exit() - -def i3status_config_reader(): - """ - i3status.conf reader so we can adapt our code to the i3status config - """ - in_time = False - in_general = False - config = { - 'colors': False, - 'color_good': None, - 'color_bad' : None, - 'color_degraded' : None, - 'color_separator': None, - 'interval': 5, - 'output_format': None, - 'time_format': '%Y-%m-%d %H:%M:%S', - } - for line in open(I3STATUS_CONFIG_FILE, 'r'): - line = line.strip(' \t\n\r') - if line.startswith('general'): - in_general = True - elif line.startswith('time'): - in_time = True - elif line.startswith('}'): - in_general = False - in_time = False - if in_general and '=' in line: - key, value = line.split('=')[0].strip(), line.split('=')[1].strip() - if config.has_key(key): - if value in ['true', 'false']: - value = 'True' if value == 'true' else 'False' - config[key] = eval(value) - if in_time and '=' in line: - key, value = line.split('=')[0].strip(), line.split('=')[1].strip() - if config.has_key('time_' + key): - config['time_' + key] = eval(value) - return config - -def i3status(message_queue, stop_thread): - """ - Execute i3status in a thread and send its output to a Queue to py3status - """ - from subprocess import Popen, PIPE - i3status_pipe = Popen(['i3status', '-c', I3STATUS_CONFIG_FILE], stdout=PIPE) - message_queue.put(i3status_pipe.stdout.readline()) - message_queue.put(i3status_pipe.stdout.readline()) - while not stop_thread: - line = i3status_pipe.stdout.readline() - if line != '': - message_queue.put(line) - else: - break - -def process_line(line, **kwargs): - """ - Main line processor logic - """ - if line.startswith('{') and 'version' in line: - print_line(line.strip('\n')) - elif line == '[\n': - print_line(line.strip('\n')) - else: - prefix = '' - if line.startswith(','): - line, prefix = line[1:], ',' - elif kwargs['delta'] > 0: - prefix = ',' - - # integrated transformations - if not DISABLE_TRANSFORM: - j = transform(loads(line), **kwargs) - else: - j = loads(line) - - # user-based injection and transformation - j = inject(j) - - print_line(prefix+dumps(j)) - -def inject(j): - """ - Run on every user class included and execute every method on the json, - then inject the result at the start of the json - """ - # inject our own functions' results - for my_class in USER_CLASSES.keys(): - for my_method in USER_CLASSES[my_class]: - try: - # handle a cache on user class methods results - try: - index, result = USER_CACHE[my_method] - if time() > result['cached_until']: - raise KeyError, 'cache timeout' - except KeyError: - # execute the method - meth = getattr(my_class, my_method) - index, result = meth(j) - - # respect user-defined cache timeout for this module - if not result.has_key('cached_until'): - result['cached_until'] = time() + CACHE_TIMEOUT - - # validate the response - assert isinstance(result, dict), "user method didn't return a dict" - assert result.has_key('full_text'), "missing 'full_text' key" - assert result.has_key('name'), "missing 'name' key" - finally: - USER_CACHE[my_method] = (index, result) - j.insert(index, result) - except Exception, err: - syslog(LOG_ERR, "injection failed (%s)" % str(err)) - return j - -def transform(j, **kwargs): - """ - Integrated transformations: - - update the 'time' object so that it's updated at INTERVAL seconds - - update the 'run_watch' objects so that we rely on the color instead of useless 'yes' or 'no' status - """ - for item in j: - # time modification - if item['name'] == 'time': - date = datetime.strptime(item['full_text'], I3STATUS_CONFIG['time_format']) + timedelta(seconds=kwargs['delta']) - item['full_text'] = date.strftime(I3STATUS_CONFIG['time_format']) - # run_watch modification, only when color is enabled on i3status - if item['name'] == 'run_watch' and I3STATUS_CONFIG['colors']: - item['full_text'] = item['full_text'].split(':')[0] - return j - -def load_from_file(filepath): - """ - Load Py3status user class for later injection - """ - inst = None - expected_class = 'Py3status' - mod_name, file_ext = os.path.splitext(os.path.split(filepath)[-1]) - if file_ext.lower() == '.py': - py_mod = imp.load_source(mod_name, filepath) - if hasattr(py_mod, expected_class): - inst = py_mod.Py3status() - return (mod_name, inst) - -# main stuff -################################################################################ -if __name__ == '__main__': - try: - # command line options - PARSER = argparse.ArgumentParser(description='The agile, python-powered, i3status wrapper') - PARSER = argparse.ArgumentParser(add_help=True) - PARSER = argparse.ArgumentParser(version='0.1') - PARSER.add_argument('-c', action="store", dest="i3status_conf", type=str, default="/etc/i3status.conf", help="path to i3status config file") - PARSER.add_argument('-d', action="store_true", dest="disable_transform", help="disable integrated transformations") - PARSER.add_argument('-i', action="store", dest="include_path", type=str, default='.i3/py3status', help="user-based class include directory") - PARSER.add_argument('-n', action="store", dest="interval", type=int, default=1, help="polling interval in seconds (default 1 sec)") - PARSER.add_argument('-t', action="store", dest="cache_timeout", type=int, default=60, help="injection cache timeout in seconds (default 60 sec)") - OPTS = PARSER.parse_args() - - # globals - CACHE_TIMEOUT = OPTS.cache_timeout - DISABLE_TRANSFORM = True if OPTS.disable_transform else False - I3STATUS_CONFIG_FILE = OPTS.i3status_conf - I3STATUS_CONFIG = i3status_config_reader() - INCLUDE_PATH = os.path.abspath( OPTS.include_path ) + '/' - INTERVAL = OPTS.interval - - # py3status uses only the i3bar protocol - assert I3STATUS_CONFIG['output_format'] == 'i3bar', 'unsupported output_format' - - # dynamic inclusion - USER_CACHE = {} - USER_CLASSES = {} - if INCLUDE_PATH and os.path.isdir(INCLUDE_PATH): - for fn in os.listdir(INCLUDE_PATH): - module, class_inst = load_from_file(INCLUDE_PATH + fn) - if module and class_inst: - USER_CLASSES[class_inst] = [] - for method in dir(class_inst): - if not method.startswith('__'): - USER_CLASSES[class_inst].append(method) - - # run threaded i3status - STOP_THREAD = False - MESSAGE_QUEUE = Queue() - I3STATUS_THREAD = threading.Thread(target=i3status, name='i3status', args=(MESSAGE_QUEUE, STOP_THREAD, )) - I3STATUS_THREAD.start() - - # main loop - while True: - try: - TS = time() - LINE = MESSAGE_QUEUE.get(timeout=INTERVAL) - if LINE.startswith(',['): - sleep( INTERVAL - float( '{:.2}'.format( time()-TS ) ) ) - process_line(LINE, delta=0) - DELTA = 0 - except Empty: - DELTA += INTERVAL - process_line(LINE, delta=DELTA) - if threading.active_count() < 2: - break - except KeyboardInterrupt: - break - - STOP_THREAD = True - I3STATUS_THREAD.join() - except Exception, err: - syslog(LOG_ERR, "py3status error (%s)" % str(err)) - sys.exit(1) |
