kuba/dotfiles
1
0
Fork 0
Code Issues Pull Requests Projects Releases Activity

lemonbar: Implement modular approach for conky slow and conky fast

Browse Source
This commit is contained in:
kuben
2019-08-19 11:57:40 +02:00
parent 4aba84ced0
commit a88a619702
5 changed files with 178 additions and 95 deletions
Download Patch File Download Diff File Expand all files Collapse all files

11
.i3/lemonbar/i3_lemonbar_common.py
View File

@@ -1,4 +1,5 @@
import subprocess, os, getpass
import re # regexp
from enum import Enum
import i3_lemonbar_config as config
@@ -97,8 +98,16 @@ commands_dict = {'toggle_secs': toggle_secs
,'bluetooth': bluetooth
}
# Helper functions
ansi_escape = re.compile(r'(\x9B|\x1B\[)[0-?]*[ -/]*[@-~]')
def strip_ansi_unicode(s):
# ANSI escape sequences are for colors in terminal and similar
strip_ansi = ansi_escape.sub('', s)
strip_unicode = (strip_ansi.encode('ascii', 'ignore')).decode('utf-8')
return strip_unicode
# Callbacks for workspaces
def add_callbacks(i3ws):
def add_callbacks(i3ws): #TODO this should be done by module
i3ws.change_callbacks.append(set_bg)
i3ws.focus_callbacks.append(set_keymap)
i3ws.focus_callbacks.append(kill_floating_windows)

31
.i3/lemonbar/i3_lemonbar_launcher.py
View File

@@ -5,15 +5,14 @@ import subprocess
import contextlib
from threading import Thread
import argparse
import re # regexp
import i3_lemonbar_config as config
import i3_lemonbar_common as common
import i3_lemonbar_modules as modules
import i3_lemonbar_parser as lemonparser
import i3_workspaces as wspaces
p_conky_slow = None
p_conky_fast = None
p_lemonbar = None
i3_ws_obj = None
@@ -90,22 +89,15 @@ def clean_up():
nice_delete(config.pid_file)
nice_delete(config.fifo_file_status)
nice_term(p_conky_slow)
nice_term(p_conky_fast)
modules.stop_all()
if i3_ws_obj is not None:
i3_ws_obj.quit()
sys.exit(0)
def write_sys_status():
global p_conky_slow, p_conky_fast, i3_ws_obj
# Buffering = 1 to write entire lines
with open(config.fifo_file_status, 'w', buffering=1) as fifo:
p_conky_slow = subprocess.Popen(['conky', '-c', config.path+'conky_slow'], # Use communicate
stdout=fifo, stderr=fifo)
common.logger.debug('Started conky slow')
p_conky_fast = subprocess.Popen(['conky', '-c', config.path+'conky_fast'],
stdout=fifo, stderr=fifo)
common.logger.debug('Started conky fast')
i3_ws_obj = wspaces.i3ws(fifo_file=config.fifo_file_status)
global i3_ws_obj
i3_ws_obj = wspaces.i3ws(logger=common.logger, fifo_file=config.fifo_file_status)
i3_ws_obj.work()
def queue_parse_job(job):
@@ -145,6 +137,10 @@ def parse_status():
common.health_logger.info('Queue closed')
break
# Ugly temporary solution
if (isinstance(data, list)):
psd = data[0]
else:
psd = lemonparser.parse_line(data)
p_lemonbar.stdin.write(psd + '\n')
p_lemonbar.stdin.flush()
@@ -181,12 +177,6 @@ def exec_commands():
break
common.logger.debug('Read: "{0}"'.format(data.strip('\n')))
ansi_escape = re.compile(r'(\x9B|\x1B\[)[0-?]*[ -/]*[@-~]')
def strip_ansi_unicode(s):
# ANSI escape sequences are for colors in terminal and similar
strip_ansi = ansi_escape.sub('', s)
strip_unicode = (strip_ansi.encode('ascii', 'ignore')).decode('utf-8')
return strip_unicode
def user_screen():
# Create screen session, in UTF-8 mode, logging to fifo
@@ -196,7 +186,7 @@ def user_screen():
while empty_count < 3:
try:
line = screen_read.readline().strip('\n')
line = strip_ansi_unicode(line)
line = common.strip_ansi_unicode(line)
common.logger.debug('Screen read line {}'.format(line))
if (not line.startswith('[CHG]')
and not line.isspace()):
@@ -276,6 +266,7 @@ if __name__ == "__main__":
i3_thread(target = put_fifo_in_queue, desc='')
i3_thread(target = write_sys_status, desc='Write sys status thread')
i3_thread(target = user_screen, desc='Screen thread for user commands')
modules.start_all()
common.logger.debug('Threads started')
i3_thread.join_threads()

129
.i3/lemonbar/i3_lemonbar_modules.py Normal file
View File

@@ -0,0 +1,129 @@
import threading
import subprocess
import i3_lemonbar_config as config
import i3_lemonbar_common as common
import i3_lemonbar_parser as parser
class LemonModule(threading.Thread):
# Module generates some status message, parses it, and handles on click actions
# Every module runs in its own thread
def __init__(self):
super().__init__()
self._start_module()
# TODO if exists dummy data send it
def run(self):
while True:
line = self.status_handle.readline()
if not line:
common.logger.info('Reached end of module {}'.format(self.name))
common.health_logger.info('Reached end of module {}'.format(self.name))
break
parsed = self.parse(line)
common.parsing_queue.put([parsed]) # TODO Wrapped in list, temporary solution
def stop(self):
self._stop_module()
def parse(self, line):
# TODO get rid of prefix
data = line[len(self.prefix):].split()
self._parse_data(data) # Update correct field
formatted_line = parser.format_line() # Construct entire line
return formatted_line
class ConkyFastModule(LemonModule):
def __init__(self):
self.prefix = 'CNK_FAST'
super().__init__()
def _start_module(self):
self.p_handle = subprocess.Popen(['conky', '-c', config.path+'conky_fast'],
stdout=subprocess.PIPE, text=True)
self.status_handle = self.p_handle.stdout
common.logger.debug('Started conky fast module')
def _stop_module(self):
if self.p_handle is not None:
self.p_handle.terminate()
def _parse_data(self, data):
mute = data[4] == 'MUTE' or data[4] == 'NONE'
(vol,vols) = (-1,'×') if mute else (int(data[4]), data[4]+'%')
icon_v = config.icon_vol_mute if vol == 0 else \
config.icon_vol_low if vol < 50 else config.icon_vol
parser.volume = parser.single_sect(icon=icon_v, click='pavu', text=vols
, alt=parser.COLOR_SCHEME.A2)
tme = data[3] if common.show_secs else data[3][:-3]
parser.date = parser.single_sect(icon=config.icon_clock, click='date'
, text=' '.join(data[0:3]), alt=parser.COLOR_SCHEME.A1)
parser.time = parser.single_sect(click='toggle_secs', text=tme, alt=parser.COLOR_SCHEME.SPECIAL)
parser.update_net(data[5:9])
class ConkySlowModule(LemonModule):
def __init__(self):
self.prefix = 'CNK_SLOW'
super().__init__()
def _start_module(self):
self.p_handle = subprocess.Popen(['conky', '-c', config.path+'conky_slow'],
stdout=subprocess.PIPE, text=True)
self.status_handle = self.p_handle.stdout
common.logger.debug('Started conky slow module')
def _stop_module(self):
if self.p_handle is not None:
self.p_handle.terminate()
def _parse_data(self, data):
# System load
if int(data[0]) > int(config.cpu_alert):
cpu_alt = parser.COLOR_SCHEME.CPU_ALERT
else:
cpu_alt = parser.COLOR_SCHEME.A2
parser.sys_load = ''.join([
parser.double_sect(text1 = (data[0] + '%'), text2 = data[1], icon1 = config.icon_cpu
, icon2 = config.icon_mem, alt=cpu_alt , click = 'load')
, parser.double_sect(text1 = (data[2] + '%'), text2 = (data[3] + '%')
, icon1 = config.icon_hd, icon2 = config.icon_home, alt=parser.COLOR_SCHEME.A1)
]) # cpu mem disk_r disk_home
#sec_color = config.color_sec_b1 if (b == 1) else config.color_sec_b2
(batt_stat, batt) = (data[4][0], int(data[4][1:]))
icon_batt = config.icon_charging if batt_stat == 'C' else \
config.icon_charged if batt_stat == 'F' else \
config.icon_batt_0 if batt < 20 else \
config.icon_batt_1 if batt < 40 else \
config.icon_batt_2 if batt < 60 else \
config.icon_batt_3 if batt < 80 else \
config.icon_batt_4
battery = parser.single_sect(icon=icon_batt, click='dpms', text=(str(batt)+'%')
, alt=parser.COLOR_SCHEME.A2)
parser.update_bright([data[5]])
parser.update_lang([data[6]])
def start_all():
global m_conky_fast, m_conky_slow
m_conky_fast = ConkyFastModule()
m_conky_slow = ConkySlowModule()
m_conky_fast.start()
m_conky_slow.start()
def stop_all():
global m_conky_fast, m_conky_slow
m_conky_fast.stop()
m_conky_slow.stop()

64
.i3/lemonbar/i3_lemonbar_parser.py
View File

@@ -14,7 +14,7 @@ battery = ''
date = ''
language = ''
time = ''
response = '%{r}'
response = ''
blank = ' '
@@ -87,7 +87,7 @@ def control_sect(sec_color=None, head='', buttons=[], actions=[]):
# Constants
power_opts = ''.join([block(fg=conf.color_fore, bg=conf.color_poweropts)
, ' Abort (Esc) | System (l) lock, (e) logout, (s) suspend, (h) hibernate'
, ', (r) reboot, (Shift+s) shutdown %{r}'])
, ', (r) reboot, (Shift+s) shutdown'])
controls = ' '.join([block(fg=conf.color_head, bg=conf.color_sec_b2)
, conf.sep_right, block(fg=conf.color_head, bg=conf.color_sec_b2, click='mode cycle')
, conf.icon_prog
@@ -102,7 +102,7 @@ def update_response(data):
resp = ' '.join(data)
#common.logger.debug('Got response {}'.format(resp))
response= ''.join(['%{r}', single_sect(text=resp)])
response= single_sect(text=resp)
def update_displays(data):
global displays
@@ -146,36 +146,6 @@ def update_workspaces(data):
workspaces = ''.join([prefix, ' '.join(wspces)])
def update_conky_slow(data):
global sys_load, battery
# System load
if int(data[0]) > int(conf.cpu_alert):
cpu_alt = COLOR_SCHEME.CPU_ALERT
else:
cpu_alt = COLOR_SCHEME.A2
sys_load = ''.join(['%{r}'
, double_sect(text1 = (data[0] + '%'), text2 = data[1], icon1 = conf.icon_cpu
, icon2 = conf.icon_mem, alt=cpu_alt , click = 'load')
, double_sect(text1 = (data[2] + '%'), text2 = (data[3] + '%')
, icon1 = conf.icon_hd, icon2 = conf.icon_home, alt=COLOR_SCHEME.A1)
]) # cpu mem disk_r disk_home
#sec_color = conf.color_sec_b1 if (b == 1) else conf.color_sec_b2
(batt_stat, batt) = (data[4][0], int(data[4][1:]))
icon_batt = conf.icon_charging if batt_stat == 'C' else \
conf.icon_charged if batt_stat == 'F' else \
conf.icon_batt_0 if batt < 20 else \
conf.icon_batt_1 if batt < 40 else \
conf.icon_batt_2 if batt < 60 else \
conf.icon_batt_3 if batt < 80 else \
conf.icon_batt_4
battery = single_sect(icon=icon_batt, click='dpms', text=(str(batt)+'%')
, alt=COLOR_SCHEME.A2)
update_bright([data[5]])
update_lang([data[6]])
def update_lang(data):
global language
language = single_sect(icon=conf.icon_lang, click='lang', text=data[0]
@@ -187,22 +157,6 @@ def update_bright(data):
brightness = single_sect(icon=conf.icon_bright, click='adj_br'
, text=(brtxt+'%'), alt=COLOR_SCHEME.A1)
def update_conky_fast(data):
global date, time, volume
mute = data[4] == 'MUTE' or data[4] == 'NONE'
(vol,vols) = (-1,'×') if mute else (int(data[4]), data[4]+'%')
icon_v = conf.icon_vol_mute if vol == 0 else \
conf.icon_vol_low if vol < 50 else conf.icon_vol
volume = single_sect(icon=icon_v, click='pavu', text=vols
, alt=COLOR_SCHEME.A2)
tme = data[3] if common.show_secs else data[3][:-3]
date = single_sect(icon=conf.icon_clock, click='date'
, text=' '.join(data[0:3]), alt=COLOR_SCHEME.A1)
time = single_sect(click='toggle_secs', text=tme, alt=COLOR_SCHEME.SPECIAL)
update_net(data[5:9])
def update_net(data):
global net_load
@@ -245,19 +199,17 @@ def update_title(data):
def format_line():
# Need to end with a reset block, otherwise the color fills the %{r} spacer
if common.mode == common.bar_mode.normal:
return ''.join(['%{l}', reset, displays, workspaces, win_title, sys_load, net_load
return ''.join(['%{l}', reset, displays, workspaces, win_title, '%{r}', sys_load, net_load
, volume, brightness, battery, date, language, time, reset])
elif common.mode == common.bar_mode.power:
return ''.join(['%{l}', reset, power_opts, sys_load, net_load
return ''.join(['%{l}', reset, power_opts, '%{r}', sys_load, net_load
, volume, brightness, battery, date, language, time, reset_power])
elif common.mode == common.bar_mode.control:
return ''.join(['%{l}', reset, displays, workspaces, controls, response, time, reset])
return ''.join(['%{l}', reset, displays, workspaces, controls, '%{r}', response, time, reset])
else:
return ''.join(['%{l}', reset, displays, workspaces, 'Not normal' , time, reset])
return ''.join(['%{l}', reset, displays, workspaces, 'Not normal', '%{r}', time, reset])
parsers_dict = { 'CNK_FAST':update_conky_fast
,'CNK_SLOW':update_conky_slow
,'WSP':update_workspaces
parsers_dict = { 'WSP':update_workspaces
,'LANG':update_lang
,'BRIGHT':update_bright
,'WIN':update_title

26
.i3/lemonbar/i3_workspaces.py
View File

@@ -20,7 +20,6 @@ import time
from subprocess import call
import i3ipc
import i3_lemonbar_config as config
import i3_lemonbar_common as common
class State(object):
@@ -59,20 +58,21 @@ class i3ws(object):
focus_callbacks = []
def __init__(self, state=None, fifo_file=None):
def __init__(self, logger, state=None, fifo_file=None):
if state:
self.state = state
if fifo_file:
self.fifo = open(fifo_file, 'w')
self.logger = logger
common.add_callbacks(self)
def work(self):
# While loop to restart connection as long as there is an i3 ipc socket
while self.running:
self.resetConn()
common.logger.debug('Started i3 workspaces manager')
self.logger.debug('Started i3 workspaces manager')
self.enterMain()
common.logger.debug('Finished i3 workspaces manager')
self.logger.debug('Finished i3 workspaces manager')
time.sleep(0.5) # TODO max number of attempts + catch exception instead
def resetConn(self):
@@ -93,12 +93,12 @@ class i3ws(object):
# Locking call
self.conn.main()
except BrokenPipeError:
common.logger.debug('Broken pipe in i3 workspaces thread, exiting')
self.logger.debug('Broken pipe in i3 workspaces thread, exiting')
except:
common.logger.debug('Unknown exception in i3 workspaces thread, exiting')
self.logger.debug('Unknown exception in i3 workspaces thread, exiting')
def shutdown(self, i3, e):
common.logger.debug('Shut down i3ipc')
self.logger.debug('Shut down i3ipc')
def change(self, i3, e):
# Receives event and workspace data
@@ -154,7 +154,7 @@ class i3ws(object):
sys.stdout.flush()
def quit(self):
common.logger.debug('Quitting i3 workspace script')
self.logger.debug('Quitting i3 workspace script')
self.running = False
self.conn.main_quit()
if self.fifo is not None:
@@ -170,18 +170,20 @@ def handle_exit(signal, frame):
sys.exit(1)
if __name__ == '__main__':
# Run as stand-alone
# Setup logger to stdout with debug log level
formatter = logging.Formatter('%(asctime)s %(levelname)s %(message)s')
handler = logging.StreamHandler(sys.stdout)
handler.setFormatter(formatter)
common.logger = logging.getLogger('Normal logger')
common.logger.setLevel(logging.DEBUG)
common.logger.addHandler(handler)
logger = logging.getLogger('Normal logger')
logger.setLevel(logging.DEBUG)
logger.addHandler(handler)
# Capture Keyboard Interrupt (i3ipc captures this internally)
signal.signal(signal.SIGINT, handle_exit)
# Start main
ws = i3ws()
ws = i3ws(logger = logger)
ws.work()