diff options
Diffstat (limited to 'py3status/__init__.py')
| -rwxr-xr-x | py3status/__init__.py | 548 |
1 files changed, 418 insertions, 130 deletions
diff --git a/py3status/__init__.py b/py3status/__init__.py index 4182b7c..49bd97e 100755 --- a/py3status/__init__.py +++ b/py3status/__init__.py @@ -14,6 +14,7 @@ from signal import SIGUSR1 from subprocess import Popen from subprocess import PIPE from subprocess import call +from tempfile import NamedTemporaryFile from threading import Event, Thread from time import sleep, time from syslog import syslog, LOG_ERR, LOG_INFO, LOG_WARNING @@ -81,69 +82,147 @@ class I3status(Thread): Our output will be read asynchronously from 'last_output'. """ Thread.__init__(self) - self.i3status_config_path = i3status_config_path - self.standalone = standalone - self.config = self.i3status_config_reader() self.error = None + self.i3status_module_names = [ + 'battery', + 'cpu_temperature', + 'cpu_usage', + 'ddate', + 'disk', + 'ethernet', + 'ipv6', + 'load', + 'path_exists', + 'run_watch', + 'time', + 'tztime', + 'volume', + 'wireless' + ] + self.json_list = None + self.json_list_ts = None self.last_output = None self.last_output_ts = None self.last_prefix = None self.lock = lock - self.json_list = None - self.json_list_ts = None + self.standalone = standalone + self.time_format = '%Y-%m-%d %H:%M:%S' + # + self.config = self.i3status_config_reader(i3status_config_path) - def i3status_config_reader(self): + def valid_config_param(self, param_name): + """ + Check if a given section name is a valid parameter for i3status. + """ + valid_config_params = self.i3status_module_names + ['general', 'order'] + return param_name.split(' ')[0] in valid_config_params + + def i3status_config_reader(self, i3status_config_path): """ Parse i3status.conf so we can adapt our code to the i3status config. """ - in_time = False - in_general = False config = { - 'colors': False, - 'color_good': '#00FF00', - 'color_bad': '#FF0000', - 'color_degraded': '#FFFF00', - 'color_separator': '#333333', - 'interval': 5, - 'output_format': None, - 'time_format': '%Y-%m-%d %H:%M:%S' + 'general': { + 'color_bad': '#FF0000', + 'color_degraded': '#FFFF00', + 'color_good': '#00FF00', + 'color_separator': '#333333', + 'colors': False, + 'interval': 5, + 'output_format': 'i3bar' + }, + 'i3s_modules': [], + 'on_click': {}, + 'order': [], + 'py3_modules': [] } # some ugly parsing - if os.path.isfile(self.i3status_config_path): - for line in open(self.i3status_config_path, 'r'): - line = line.strip(' \t\n\r') - if line.startswith('general'): - in_general = True - elif line.startswith('time') or line.startswith('tztime'): - in_time = True - elif line.startswith('}'): - in_general = False - in_time = False - if in_general and '=' in line: - key = line.split('=')[0].strip() - value = line.split('=')[1].strip() - if key in config: - if value in ['true', 'false']: - value = 'True' if value == 'true' else 'False' + in_section = False + section_name = '' + + for line in open(i3status_config_path, 'r'): + line = line.strip(' \t\n\r') + + if not line or line.startswith('#'): + continue + + if line.startswith('order'): + in_section = True + section_name = 'order' + + if not in_section: + section_name = line.split('{')[0].strip() + if section_name not in config: + config[section_name] = {} + + if '{' in line: + in_section = True + + if section_name and '=' in line: + line = line.split('}')[0].strip() + + key = line.split('=')[0].strip() + value = line.split('=')[1].strip() + try: + e_value = eval(value) + if isinstance(e_value, str) or isinstance(e_value, int): + value = e_value + else: + raise ValueError() + except NameError: + pass + except ValueError: + pass + + if section_name == 'order': + config[section_name].append(value) + line = '}' + + # detect internal modules to be loaded dynamically + if not self.valid_config_param(value): + config['py3_modules'].append(value) + else: + config['i3s_modules'].append(value) + else: + if not key.startswith('on_click'): + config[section_name][key] = value + else: + # on_click special parameters try: - config[key] = eval(value) - except NameError: - config[key] = value - if in_time and '=' in line: - key = line.split('=')[0].strip() - value = line.split('=')[1].strip() - if 'time_' + key in config: - config['time_' + key] = eval(value) + button = int(key.split()[1]) + if button not in range(1, 6): + raise ValueError( + 'should be 1, 2, 3, 4 or 5' + ) + except IndexError as e: + raise IndexError( + 'missing "button id" for "on_click" ' + 'parameter in section {}'.format(section_name) + ) + except ValueError as e: + raise ValueError( + 'invalid "button id" ' + 'for "on_click" parameter ' + 'in section {} ({})'.format(section_name, e) + ) + on_c = config['on_click'] + on_c[section_name] = on_c.get(section_name, {}) + on_c[section_name][button] = value - # force output format on standalone mode - if self.standalone: - config['output_format'] = 'i3bar' + # override time format + if section_name in ['time', 'tztime'] and key == 'format': + self.time_format = value - # py3status uses only the i3bar protocol - assert config['output_format'] == 'i3bar', \ - 'i3status output_format should be set to "i3bar" on {}'.format( - self.i3status_config_path + if '}' in line: + in_section = False + section_name = '' + + # py3status only uses the i3bar protocol because it needs JSON output + if config['general']['output_format'] != 'i3bar': + raise RuntimeError( + 'i3status output_format should ' + 'be set to "i3bar" on {}'.format(i3status_config_path) ) return config @@ -154,7 +233,7 @@ class I3status(Thread): """ json_list = deepcopy(self.json_list) try: - time_format = self.config['time_format'] + time_format = self.time_format for item in json_list: if 'name' in item and item['name'] in ['time', 'tztime']: i3status_time = item['full_text'].encode('UTF-8', 'replace') @@ -172,9 +251,7 @@ class I3status(Thread): ) date = datetime.strptime(i3status_time, time_format) date += delta - item['full_text'] = date.strftime( - self.config['time_format'] - ) + item['full_text'] = date.strftime(self.time_format) item['transformed'] = True except Exception: err = sys.exc_info()[1] @@ -191,60 +268,106 @@ class I3status(Thread): self.json_list = deepcopy(self.last_output) self.json_list_ts = deepcopy(self.last_output_ts) + def get_modules_output(self, json_list, py3_modules): + """ + Return the final json list to be displayed on the i3bar by taking + into account every py3status configured module and i3status'. + Simply put, this method honors the initial 'order' configured by + the user in his i3status.conf. + """ + ordered = [] + for module in self.config['order']: + if module in py3_modules: + for method in py3_modules[module].methods.values(): + ordered.append(method['last_output']) + else: + for m, j in zip(self.config['i3s_modules'], json_list): + if m == module: + ordered.append(j) + return ordered + + def write_tmp_i3status_config(self, tmpfile): + """ + Given a temporary file descriptor, write a valid i3status config file + based on the parsed one from 'i3status_config_path'. + """ + for section_name, conf in self.config.items(): + if section_name in ['i3s_modules', 'py3_modules']: + continue + elif section_name == 'order': + for module_name in conf: + if self.valid_config_param(module_name): + tmpfile.write('order += "%s"\n' % module_name) + tmpfile.write('\n') + elif self.valid_config_param(section_name): + tmpfile.write('%s {\n' % section_name) + for key, value in conf.items(): + tmpfile.write(' %s = "%s"\n' % (key, value)) + tmpfile.write('}\n\n') + tmpfile.flush() + def run(self): """ - Spawn i3status and poll its output. + Spawn i3status using a self generated config file and poll its output. """ - i3status_pipe = Popen( - ['i3status', '-c', self.i3status_config_path], - stdout=PIPE, - stderr=PIPE, - ) - self.poller_inp = IOPoller(i3status_pipe.stdout) - self.poller_err = IOPoller(i3status_pipe.stderr) + with NamedTemporaryFile(prefix='py3status_') as tmpfile: + self.write_tmp_i3status_config(tmpfile) + syslog( + LOG_INFO, + 'i3status spawned using config file {}'.format(tmpfile.name) + ) - try: - # at first, poll very quickly to avoid delay in first i3bar display - timeout = 0.001 + i3status_pipe = Popen( + ['i3status', '-c', tmpfile.name], + stdout=PIPE, + stderr=PIPE, + ) + self.poller_inp = IOPoller(i3status_pipe.stdout) + self.poller_err = IOPoller(i3status_pipe.stderr) - # loop on i3status output - while self.lock.is_set(): - line = self.poller_inp.readline(timeout) - if line: - if line.startswith('[{'): - with jsonify(line) as (prefix, json_list): - self.last_output = json_list - self.last_output_ts = datetime.utcnow() - self.last_prefix = ',' - self.update_json_list() - print_line(line) - elif not line.startswith(','): - if 'version' in line: - header = loads(line) - header.update({'click_events': True}) - line = dumps(header) - print_line(line) - else: - timeout = 0.5 - with jsonify(line) as (prefix, json_list): - self.last_output = json_list - self.last_output_ts = datetime.utcnow() - self.last_prefix = prefix - else: - err = self.poller_err.readline(timeout) - code = i3status_pipe.poll() - if code is not None: - if err: - msg = 'i3status died and said: {}'.format(err) + try: + # at first, poll very quickly + # to avoid delay in first i3bar display + timeout = 0.001 + + # loop on i3status output + while self.lock.is_set(): + line = self.poller_inp.readline(timeout) + if line: + if line.startswith('[{'): + with jsonify(line) as (prefix, json_list): + self.last_output = json_list + self.last_output_ts = datetime.utcnow() + self.last_prefix = ',' + self.update_json_list() + print_line(line) + elif not line.startswith(','): + if 'version' in line: + header = loads(line) + header.update({'click_events': True}) + line = dumps(header) + print_line(line) else: - msg = 'i3status died with code {}'.format(code) - raise IOError(msg) + timeout = 0.5 + with jsonify(line) as (prefix, json_list): + self.last_output = json_list + self.last_output_ts = datetime.utcnow() + self.last_prefix = prefix else: - # poll is CPU intensive, breath a bit - sleep(timeout) - except IOError: - err = sys.exc_info()[1] - self.error = err + err = self.poller_err.readline(timeout) + code = i3status_pipe.poll() + if code is not None: + if err: + msg = 'i3status died and said: {}'.format(err) + else: + msg = 'i3status died with code {}'.format(code) + raise IOError(msg) + else: + # poll is CPU intensive, breath a bit + sleep(timeout) + except IOError: + err = sys.exc_info()[1] + self.error = err def mock(self): """ @@ -273,7 +396,7 @@ class Events(Thread): """ This class is responsible for dispatching event JSONs sent by the i3bar. """ - def __init__(self, lock, config, modules): + def __init__(self, lock, config, modules, on_click): """ We need to poll stdin to receive i3bar messages. """ @@ -282,6 +405,7 @@ class Events(Thread): self.lock = lock self.modules = modules self.poller_inp = IOPoller(sys.stdin) + self.on_click = on_click def dispatch(self, module, obj, event): """ @@ -306,7 +430,7 @@ class Events(Thread): it will dispatch i3status click events to this module so you can catch them and trigger any function call based on the event. """ - for module in self.modules: + for module in self.modules.values(): if not module.click_events: continue if module.module_name == 'i3bar_click_events.py': @@ -314,6 +438,57 @@ class Events(Thread): else: return False + def refresh(self, module_name): + """ + Force a cache expiration for all the methods of the given module. + """ + module = self.modules.get(module_name) + if module is not None: + if self.config['debug']: + syslog(LOG_INFO, 'refresh module {}'.format(module_name)) + for obj in module.methods.values(): + obj['cached_until'] = time() + + @staticmethod + def refresh_all(module_name): + """ + Force a full refresh of py3status and i3status modules by sending + a SIGUSR1 signal to py3status. + """ + Popen(['killall', '-USR1', 'py3status']) + + def on_click_dispatcher(self, module_name, command): + """ + Dispatch on_click config parameters to either: + - Our own methods for special py3status commands (listed below) + - The i3-msg program which is part of i3wm + """ + py3_commands = ['refresh', 'refresh_all'] + if command is None: + return + elif command in py3_commands: + # this is a py3status command handled by this class + method = getattr(self, command) + method(module_name) + else: + # this is a i3 message + self.i3_msg(module_name, command) + + @staticmethod + def i3_msg(module_name, command): + """ + Execute the given i3 message and log its output. + """ + i3_msg_pipe = Popen(['i3-msg', command], stdout=PIPE) + syslog( + LOG_INFO, + 'i3-msg module={} command="{}" stdout={}'.format( + module_name, + command, + i3_msg_pipe.stdout.read() + ) + ) + def run(self): """ Wait for an i3bar JSON event, then find the right module to dispatch @@ -336,13 +511,28 @@ class Events(Thread): if self.config['debug']: syslog(LOG_INFO, 'received event {}'.format(event)) - # setup default action on button 2 press + # usage variables + button = event.get('button', 0) default_event = False dispatched = False - if 'button' in event and event['button'] == 2: + instance = event.get('instance', '') + name = event.get('name', '') + + # guess the module config name + module_name = '{} {}'.format(name, instance).strip() + + # execute any configured i3-msg command + if self.on_click.get(module_name, {}).get(button): + self.on_click_dispatcher( + module_name, + self.on_click[module_name].get(button) + ) + dispatched = True + # otherwise setup default action on button 2 press + elif button == 2: default_event = True - for module in self.modules: + for module in self.modules.values(): # skip modules not supporting click_events # unless we have a default_event set if not module.click_events and not default_event: @@ -350,9 +540,9 @@ class Events(Thread): # check for the method name/instance for obj in module.methods.values(): - if event['name'] == obj['name']: - if 'instance' in event: - if event['instance'] == obj['instance']: + if name == obj['name']: + if instance: + if instance == obj['instance']: self.dispatch(module, obj, event) dispatched = True break @@ -413,6 +603,20 @@ class Module(Thread): inst = py_mod.Py3status() return (mod_name, inst) + @staticmethod + def load_from_namespace(module_name): + """ + Load a py3status bundled module. + """ + inst = None + name = 'py3status.modules.{}'.format(module_name.split(' ')[0]) + py_mod = __import__(name) + components = name.split('.') + for comp in components[1:]: + py_mod = getattr(py_mod, comp) + inst = py_mod.Py3status() + return (module_name, inst) + def clear_cache(self): """ Reset the cache for all methods of this module. @@ -431,9 +635,20 @@ class Module(Thread): - 'on_click' methods as they'll be called upon a click_event - 'kill' methods as they'll be called upon this thread's exit """ - module, class_inst = self.load_from_file(include_path + f_name) + if include_path is not None: + module, class_inst = self.load_from_file(include_path + f_name) + else: + module, class_inst = self.load_from_namespace(f_name) + if module and class_inst: self.module_class = class_inst + + # apply module configuration from i3status config + mod_config = self.i3status_thread.config.get(self.module_name, {}) + for config, value in mod_config.items(): + setattr(self.module_class, config, value) + + # get the available methods for execution for method in sorted(dir(class_inst)): if method.startswith('_'): continue @@ -480,7 +695,7 @@ class Module(Thread): click_method = getattr(self.module_class, 'on_click') click_method( self.i3status_thread.json_list, - self.i3status_thread.config, + self.i3status_thread.config['general'], event ) except Exception: @@ -511,18 +726,31 @@ class Module(Thread): try: # execute method and get its output method = getattr(self.module_class, meth) - position, result = method( + response = method( self.i3status_thread.json_list, - self.i3status_thread.config + self.i3status_thread.config['general'] ) - # validate the result - assert isinstance(result, dict), "should return a dict" - assert 'full_text' in result, "missing 'full_text' key" - assert 'name' in result, "missing 'name' key" + if isinstance(response, dict): + # this is a shiny new module giving a dict response + position, result = None, response + result['name'] = self.module_name.split(' ')[0] + result['instance'] = ''.join( + self.module_name.split(' ')[1:] + ) + else: + # this is an old school module reporting its position + position, result = response + if not isinstance(position, int): + raise TypeError('position is not an int') + if not isinstance(result, dict): + raise TypeError('response should be a dict') + if 'name' not in result: + raise KeyError('missing "name" key in response') - # validate the position - assert isinstance(position, int), "position is not an int" + # validate the response + if not 'full_text' in result: + raise KeyError('missing "full_text" key in response') # initialize method object if my_method['name'] is None: @@ -568,7 +796,7 @@ class Module(Thread): kill_method = getattr(self.module_class, 'kill') kill_method( self.i3status_thread.json_list, - self.i3status_thread.config + self.i3status_thread.config['general'] ) except Exception: # this would be stupid to die on exit @@ -583,8 +811,9 @@ class Py3statusWrapper(): """ Useful variables we'll need. """ - self.modules = [] + self.modules = {} self.lock = Event() + self.py3_modules = [] def get_config(self): """ @@ -666,14 +895,27 @@ class Py3statusWrapper(): def list_modules(self): """ - Search import directories and files through given include paths. + Search import directories and files through given include paths with + respect to i3status.conf configured py3status modules. + + User provided modules take precedence over py3status generic modules + but if none has been configured then we'll load every file present + as this is the legacy behavior. + This method is a generator and loves to yield. """ for include_path in sorted(self.config['include_paths']): include_path = os.path.abspath(include_path) + '/' - if os.path.isdir(include_path): - for f_name in sorted(os.listdir(include_path)): - if f_name.endswith('.py'): + if not os.path.isdir(include_path): + continue + + for f_name in sorted(os.listdir(include_path)): + if f_name.endswith('.py'): + if self.py3_modules: + mod_name = f_name.rstrip('.py') + if mod_name in self.py3_modules: + yield (include_path, f_name) + else: yield (include_path, f_name) def setup(self): @@ -718,7 +960,12 @@ class Py3statusWrapper(): ) # setup input events thread - self.events_thread = Events(self.lock, self.config, self.modules) + self.events_thread = Events( + self.lock, + self.config, + self.modules, + self.i3status_thread.config['on_click'] + ) self.events_thread.start() if self.config['debug']: syslog(LOG_INFO, 'events thread started') @@ -728,8 +975,13 @@ class Py3statusWrapper(): sys.stdout = open('/dev/null', 'w') sys.stderr = open('/dev/null', 'w') - # load and spawn modules threads + # get the list of py3status configured modules + self.py3_modules = self.i3status_thread.config['py3_modules'] + + # load and spawn external modules threads + # based on inclusion folder for include_path, f_name in self.list_modules(): + module_name = f_name.rstrip('.py') try: my_m = Module( self.lock, @@ -741,7 +993,7 @@ class Py3statusWrapper(): # only start and handle modules with available methods if my_m.methods: my_m.start() - self.modules.append(my_m) + self.modules[module_name] = my_m elif self.config['debug']: syslog( LOG_INFO, @@ -752,6 +1004,33 @@ class Py3statusWrapper(): msg = 'loading {} failed ({})'.format(f_name, err) self.i3_nagbar(msg, level='warning') + # load and spawn i3status.conf configured modules threads + for module_name in self.py3_modules: + # ignore if the user already provided this module + if module_name in self.modules: + continue + try: + my_m = Module( + self.lock, + self.config, + None, + module_name, + self.i3status_thread + ) + # only start and handle modules with available methods + if my_m.methods: + my_m.start() + self.modules[module_name] = my_m + elif self.config['debug']: + syslog( + LOG_INFO, + 'ignoring {} (no methods found)'.format(module_name) + ) + except Exception: + err = sys.exc_info()[1] + msg = 'loading {} failed ({})'.format(module_name, err) + self.i3_nagbar(msg, level='warning') + def i3_nagbar(self, msg, level='error'): """ Make use of i3-nagbar to display errors and warnings to the user. @@ -791,7 +1070,7 @@ class Py3statusWrapper(): """ For every module, reset the 'cached_until' of all its methods. """ - for module in self.modules: + for module in self.modules.values(): module.clear_cache() def get_modules_output(self, json_list): @@ -804,7 +1083,8 @@ class Py3statusWrapper(): # prepopulate the list so that every usable index exists, thx @Lujeni m_list = [ '' for value in range( - sum([len(x.methods) for x in self.modules]) + len(json_list) + sum([len(x.methods) for x in self.modules.values()]) + + len(json_list) ) ] @@ -817,7 +1097,7 @@ class Py3statusWrapper(): # run through modules/methods output and insert them in reverse order debug_msg = '' - for m in reversed(self.modules): + for m in reversed(self.modules.values()): for meth in m.methods: position = m.methods[meth]['position'] last_output = m.methods[meth]['last_output'] @@ -925,7 +1205,7 @@ class Py3statusWrapper(): json_list = self.i3status_thread.adjust_time(delta) # check that every module thread is alive - for module in self.modules: + for module in self.modules.values(): if not module.is_alive(): # don't spam the user with i3-nagbar warnings if not hasattr(module, 'i3_nagbar'): @@ -937,7 +1217,15 @@ class Py3statusWrapper(): # construct the global output, modules first if self.modules: - json_list = self.get_modules_output(json_list) + if self.py3_modules: + # new style i3status configured ordering + json_list = self.i3status_thread.get_modules_output( + json_list, + self.modules + ) + else: + # old style ordering + json_list = self.get_modules_output(json_list) # dump the line to stdout print_line('{}{}'.format(prefix, dumps(json_list))) |