summaryrefslogtreecommitdiffstats
diff options
context:
space:
mode:
Diffstat
-rwxr-xr-xpy3status/__init__.py580
1 files changed, 321 insertions, 259 deletions
diff --git a/py3status/__init__.py b/py3status/__init__.py
index 3e794e9..a021356 100755
--- a/py3status/__init__.py
+++ b/py3status/__init__.py
@@ -30,6 +30,7 @@ import imp
import sys
import argparse
import threading
+from threading import Thread
from json import loads
from json import dumps
@@ -52,301 +53,362 @@ from syslog import LOG_ERR
from syslog import LOG_INFO
try:
- # python3
- from queue import Queue
- from queue import Empty
+ # python3
+ from queue import Queue
+ from queue import Empty
except ImportError:
- # python2
- from Queue import Queue
- from Queue import Empty
+ # python2
+ from Queue import Queue
+ from Queue import Empty
+
+# module globals and defaults
+################################################################################
+CACHE_TIMEOUT = 60
+DISABLE_TRANSFORM = False
+I3STATUS_CONFIG = '/etc/i3status.conf'
+INCLUDE_PATH = '.i3/py3status'
+INTERVAL = 1
+USER_CACHE = {}
+USER_CLASSES = {}
# functions
################################################################################
def print_line(message):
- """
- Non-buffered printing to stdout
- """
- sys.stdout.write(message + '\n')
- sys.stdout.flush()
+ """
+ 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()
+ """
+ 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(config_file):
+ """
+ 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(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 key in config:
+ 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 'time_' + key in config:
+ config['time_' + key] = eval(value)
+ return config
-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 key in config:
- 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 'time_' + key in config:
- config['time_' + key] = eval(value)
- return config
+class I3status(Thread):
+ """
+ Run i3status in a thread and send its output to a Queue
+ """
+ def __init__(self, config_file):
+ """
+ set our useful properties
+ """
+ Thread.__init__(self)
+ self.config_file = config_file
+ self.kill = False
+ self.queue = Queue()
+ self.init = True
+ self.started = False
-def i3status(message_queue):
- """
- Execute i3status in a thread and send its output to a Queue to py3status
- """
- i3status_pipe = Popen(['i3status', '-c', I3STATUS_CONFIG_FILE], stdout=PIPE, stderr=PIPE)
- message_queue.put(i3status_pipe.stdout.readline())
- message_queue.put(i3status_pipe.stdout.readline())
- while True:
- line = i3status_pipe.stdout.readline()
- if len(line) > 0:
- message_queue.put(line)
- else:
- break
+ def stop(self):
+ """
+ break the i3status loop
+ """
+ self.kill = True
+
+ def run(self):
+ """
+ run i3status and queue its output
+ """
+ i3status_pipe = Popen(
+ ['i3status', '-c', self.config_file],
+ stdout=PIPE,
+ stderr=PIPE,
+ )
+ self.queue.put(i3status_pipe.stdout.readline())
+ self.queue.put(i3status_pipe.stdout.readline())
+ while not self.kill:
+ line = i3status_pipe.stdout.readline()
+ if len(line) > 0:
+ self.queue.put(line)
+ try:
+ line = line.decode()
+ except Exception:
+ pass
+ if len(line) > 1 and line.startswith('['):
+ self.init = False
+ elif line.startswith(',['):
+ self.started = True
+ 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 = ','
+ """
+ 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)
+ # integrated transformations
+ if not DISABLE_TRANSFORM:
+ j = transform(loads(line), **kwargs)
+ else:
+ j = loads(line)
- # user-based injection and transformation
- j = inject(j)
+ # user-based injection and transformation
+ j = inject(j)
- print_line(prefix+dumps(j))
+ output = prefix+dumps(j)
+ print_line(output)
+ return output
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 fn in sorted( USER_CLASSES.keys() ):
- my_class, my_methods = USER_CLASSES[fn]
- for my_method in my_methods:
- 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
- try:
- meth = getattr(my_class, my_method)
- index, result = meth(j, I3STATUS_CONFIG)
- except Exception:
- err = sys.exc_info()[1]
- syslog(LOG_ERR, "user method %s failed (%s)" % (my_method, str(err)))
- index, result = (0, {'name': '', 'full_text': ''})
+ """
+ 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 class_name in sorted( USER_CLASSES.keys() ):
+ my_class, my_methods = USER_CLASSES[class_name]
+ for my_method in my_methods:
+ 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
+ try:
+ meth = getattr(my_class, my_method)
+ index, result = meth(j, I3STATUS_CONFIG)
+ except Exception:
+ err = sys.exc_info()[1]
+ syslog(LOG_ERR, "user method %s failed (%s)" \
+ % (my_method, str(err)))
+ index, result = (0, {'name': '', 'full_text': ''})
- # respect user-defined cache timeout for this module
- if 'cached_until' not in result:
- result['cached_until'] = time() + CACHE_TIMEOUT
+ # respect user-defined cache timeout for this module
+ if 'cached_until' not in result:
+ result['cached_until'] = time() + CACHE_TIMEOUT
- # validate the response
- assert isinstance(result, dict), "user method didn't return a dict"
- assert 'full_text' in result, "missing 'full_text' key"
- assert 'name' in result, "missing 'name' key"
- finally:
- USER_CACHE[my_method] = (index, result)
- j.insert(index, result)
- except Exception:
- err = sys.exc_info()[1]
- syslog(LOG_ERR, "injection failed (%s)" % str(err))
- return j
+ # validate the response
+ assert isinstance(result, dict), "user should return a dict"
+ assert 'full_text' in result, "missing 'full_text' key"
+ assert 'name' in result, "missing 'name' key"
+ finally:
+ USER_CACHE[my_method] = (index, result)
+ j.insert(index, result)
+ except Exception:
+ err = sys.exc_info()[1]
+ 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
- """
- try:
- for item in j:
- # time modification
- if item['name'] in [ 'time', 'tztime' ]:
- date = datetime.strptime(item['full_text'], I3STATUS_CONFIG['time_format']) + timedelta(seconds=kwargs['delta'])
- item['full_text'] = date.strftime(I3STATUS_CONFIG['time_format'])
- except Exception:
- err = sys.exc_info()[1]
- syslog(LOG_ERR, "transformation failed (%s)" % (str(err)))
- return j
+ """
+ Integrated transformations:
+ - update the 'time' object so that it's updated at INTERVAL seconds
+ """
+ try:
+ for item in j:
+ # time modification
+ if item['name'] in [ 'time', 'tztime' ]:
+ time_format = I3STATUS_CONFIG['time_format']
+ date = datetime.strptime( item['full_text'], time_format ) \
+ + timedelta(seconds=kwargs['delta'])
+ item['full_text'] = date.strftime(time_format)
+ if kwargs['delta'] > 0:
+ item['transformed'] = True
+ except Exception:
+ err = sys.exc_info()[1]
+ syslog(LOG_ERR, "transformation failed (%s)" % (str(err)))
+ 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)
+ """
+ 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
################################################################################
def main():
- try:
- # global definition
- global CACHE_TIMEOUT, DISABLE_TRANSFORM, I3STATUS_CONFIG_FILE
- global I3STATUS_CONFIG, INCLUDE_PATH, INTERVAL, USER_CACHE, USER_CLASSES
+ """
+ Main logic function
+ """
+ try:
+ # global definition
+ global CACHE_TIMEOUT, DISABLE_TRANSFORM
+ global I3STATUS_CONFIG, INCLUDE_PATH, INTERVAL, USER_CACHE, USER_CLASSES
+
+ # command line options
+ parser = argparse.ArgumentParser(
+ description='The agile, python-powered, i3status wrapper')
+ parser = argparse.ArgumentParser(add_help=True)
+ parser.add_argument('-c', action="store",
+ dest="i3status_conf", type=str,
+ default=I3STATUS_CONFIG, 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=INCLUDE_PATH, help="user-based class include directory")
+ parser.add_argument('-n', action="store",
+ dest="interval", type=int,
+ default=INTERVAL, help="update interval in seconds (default 1 sec)")
+ parser.add_argument('-t', action="store",
+ dest="cache_timeout", type=int, default=CACHE_TIMEOUT,
+ help="default injection cache timeout in seconds (default 60 sec)")
+ options = parser.parse_args()
- # command line options
- PARSER = argparse.ArgumentParser(description='The agile, python-powered, i3status wrapper')
- PARSER = argparse.ArgumentParser(add_help=True)
- 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()
+ # configuration and helper variables
+ CACHE_TIMEOUT = options.cache_timeout
+ DISABLE_TRANSFORM = True if options.disable_transform else False
+ I3STATUS_CONFIG = i3status_config_reader(options.i3status_conf)
+ INCLUDE_PATH = os.path.abspath( options.include_path ) + '/'
+ INTERVAL = options.interval
- # configuration and helper variables
- CACHE_TIMEOUT = OPTS.cache_timeout
- DELTA = 0
- DISABLE_TRANSFORM = True if OPTS.disable_transform else False
- FORCED = False
- I3STATUS_CONFIG_FILE = OPTS.i3status_conf
- I3STATUS_CONFIG = i3status_config_reader()
- INCLUDE_PATH = os.path.abspath( OPTS.include_path ) + '/'
- INTERVAL = OPTS.interval
- STARTED = False
+ # py3status uses only the i3bar protocol
+ assert I3STATUS_CONFIG['output_format'] == 'i3bar', \
+ 'i3status output_format should be set to "i3bar"'
+ except Exception:
+ err = sys.exc_info()[1]
+ syslog(LOG_ERR, "py3status init error (%s)" % str(err))
+ sys.exit(1)
- # py3status uses only the i3bar protocol
- assert I3STATUS_CONFIG['output_format'] == 'i3bar', 'unsupported output_format'
+ try:
+ USER_CACHE = {}
+ USER_CLASSES = {}
+ # read user-written Py3status class files for dynamic inclusion
+ if INCLUDE_PATH and os.path.isdir(INCLUDE_PATH):
+ for file_name in os.listdir(INCLUDE_PATH):
+ module, class_inst = load_from_file(INCLUDE_PATH + file_name)
+ if module and class_inst:
+ USER_CLASSES[file_name] = (class_inst, [])
+ for method in dir(class_inst):
+ if not method.startswith('__'):
+ USER_CLASSES[file_name][1].append(method)
- # read user-written Py3status class files for 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[fn] = (class_inst, [])
- for method in dir(class_inst):
- if not method.startswith('__'):
- USER_CLASSES[fn][1].append(method)
+ # spawn a i3status process on a separate thread
+ # we will receive its output on a Queue which we will poll for messages
+ i3status_thread = I3status(options.i3status_conf)
+ i3status_thread.start()
- # spawn a i3status process on a separate thread
- # we will receive its output on a Queue which we will poll for messages
- MESSAGE_QUEUE = Queue()
- I3STATUS_THREAD = threading.Thread(target=i3status, name='i3status', args=(MESSAGE_QUEUE, ))
- I3STATUS_THREAD.start()
+ # SIGUSR1 forces a refresh of the bar both for py3status and i3status,
+ # this mimics the USR1 signal handling of i3status (see man i3status)
+ def sig_handler(signum, frame):
+ """
+ Raise a Warning level exception when a user sends a SIGUSR1 signal
+ """
+ raise UserWarning("received USR1, forcing refresh")
+ signal(SIGUSR1, sig_handler)
- # SIGUSR1 forces a refresh of the bar both for py3status and i3status,
- # this mimics the USR1 signal handling of i3status (see man i3status)
- def sig_handler(signum, frame):
- """
- Raise a Warning level exception when a user sends a SIGUSR1 signal
- """
- raise UserWarning("received USR1, forcing refresh")
- signal(SIGUSR1, sig_handler)
+ # control variables
+ forced = False
- # main loop
- while True:
- try:
- # get a timestamp of now
- TS = time()
+ # main loop
+ while True:
+ try:
+ # get a timestamp of now
+ tst = time()
- # try to read i3status' output for at most INTERVAL seconds else
- # raise the Empty exception
- LINE = MESSAGE_QUEUE.get(timeout=INTERVAL)
- try:
- # python3 compatibility code
- LINE = LINE.decode()
- except UnicodeDecodeError:
- pass
+ # try to read i3status' output for at most INTERVAL seconds else
+ # raise the Empty exception
+ line = i3status_thread.queue.get(timeout=INTERVAL)
+ try:
+ # python3 compatibility code
+ line = line.decode()
+ except UnicodeDecodeError:
+ pass
- # i3status first output lines should be processed asap as only
- # the following lines will be processed by py3status
- if LINE.startswith(',['):
- if not FORCED and I3STATUS_CONFIG['interval'] > INTERVAL:
- # add a calculated sleep honoring py3status refresh
- # time of the bar every INTERVAL seconds
- sleep( INTERVAL - float( '{:.2}'.format( time()-TS ) ) )
- else:
- # reset the SIGUSR1 flag forcing the refresh of the bar
- FORCED = False
- # flag i3status as started
- STARTED = True
+ # i3status first output lines should be processed asap as only
+ # the following lines will be processed by py3status
+ if i3status_thread.started:
+ if not forced and I3STATUS_CONFIG['interval'] > INTERVAL:
+ # add a calculated sleep honoring py3status refresh
+ # time of the bar every INTERVAL seconds
+ sleep(INTERVAL - float( '{:.2}'.format( time()-tst ) ))
+ else:
+ # reset the SIGUSR1 flag forcing the refresh of the bar
+ forced = False
- # process the output line and reset the time DELTA
- process_line(LINE, delta=0)
- DELTA = 0
- except Empty:
- # make sure i3status has started before modifying its output
- if STARTED:
- # the DELTA helps us adjust the clock without a syscall
- DELTA += INTERVAL
- process_line(LINE, delta=DELTA)
- if threading.active_count() < 2:
- # i3status died ? oups
- break
- else:
- syslog(LOG_INFO, "waiting for i3status")
- except UserWarning:
- # SIGUSR1 was received, we also force i3status to refresh by
- # sending it a SIGUSR1 as well then we refresh the bar asap
- msg = sys.exc_info()[1]
- syslog(LOG_INFO, str(msg))
- call(["killall", "-s", "USR1", "i3status"])
- USER_CACHE = {}
- FORCED = True
- except KeyboardInterrupt:
- break
+ # process the output line straight from i3status
+ process_line(line, delta=0)
+ except Empty:
+ # make sure i3status has started before modifying its output
+ if i3status_thread.started or not i3status_thread.init:
+ line = process_line(line, delta=INTERVAL)
+ if threading.active_count() < 2:
+ # i3status died ? oups
+ break
+ else:
+ syslog(LOG_INFO, "waiting for i3status")
+ except UserWarning:
+ # SIGUSR1 was received, we also force i3status to refresh by
+ # sending it a SIGUSR1 as well then we refresh the bar asap
+ msg = sys.exc_info()[1]
+ syslog(LOG_INFO, str(msg))
+ call(["killall", "-s", "USR1", "i3status"])
+ USER_CACHE = {}
+ forced = True
+ except KeyboardInterrupt:
+ break
- I3STATUS_THREAD.join()
- except Exception:
- err = sys.exc_info()[1]
- syslog(LOG_ERR, "py3status error (%s)" % str(err))
- sys.exit(1)
+ i3status_thread.stop()
+ except Exception:
+ err = sys.exc_info()[1]
+ syslog(LOG_ERR, "py3status error (%s)" % str(err))
+ sys.exit(2)
if __name__ == '__main__':
- main()
+ main()

Want to contribute? Host your changes somewhere and send me the link: jan (at) sydanviita.fi. Also, check out my website.


forge.jan.systems — powered by cgit