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

Begin implementing screen thread in lemonbar

Browse Source
This commit is contained in:
kuben
2019-07-28 11:56:39 +02:00
parent aa1283a006
commit bfe90a1c45
3 changed files with 32 additions and 22 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
.i3/lemonbar/i3_lemonbar_common.py
View File

@@ -3,7 +3,7 @@ from enum import Enum
import i3_lemonbar_config as config import i3_lemonbar_config as config
p_bluetooth = None p_screen = None
kill_on_unfocus = [] kill_on_unfocus = []
class bar_mode(Enum): class bar_mode(Enum):
@@ -61,8 +61,8 @@ def bluetooth(args):
btcargs = [a.replace('pxc550', '00:16:94:22:29:0E') for a in btcargs] btcargs = [a.replace('pxc550', '00:16:94:22:29:0E') for a in btcargs]
btcargs.append('\n') btcargs.append('\n')
inp = ' '.join(btcargs) inp = ' '.join(btcargs)
p_bluetooth.stdin.write(inp) p_screen.stdin.write(inp)
p_bluetooth.stdin.flush() p_screen.stdin.flush()
show_secs = False show_secs = False
def toggle_secs(args): def toggle_secs(args):

40
.i3/lemonbar/i3_lemonbar_launcher.py
View File

@@ -110,7 +110,7 @@ def clean_up():
nice_delete(config.fifo_file_executor) nice_delete(config.fifo_file_executor)
nice_term(p_conky_slow) nice_term(p_conky_slow)
nice_term(p_conky_fast) nice_term(p_conky_fast)
nice_term(common.p_bluetooth) nice_term(common.p_screen)
if i3_ws_obj is not None: if i3_ws_obj is not None:
i3_ws_obj.quit() i3_ws_obj.quit()
sys.exit(0) sys.exit(0)
@@ -134,7 +134,7 @@ def parse_status():
with open(config.fifo_file_status, 'r', buffering=1) as fifo_read: with open(config.fifo_file_status, 'r', buffering=1) as fifo_read:
# Let parser read from fifo # Let parser read from fifo
logger.debug("FIFO {} opened for reading".format(config.fifo_file_status)) logger.debug("FIFO {} opened for reading".format(config.fifo_file_status))
lemonparser.parse_line('WSPINA1___main INA2___web FOC5___terms INA6___stats ') lemonparser.parse_line('WSPINA1___main INA2___web FOC5___terms INA6___stats ') # TODO modular
while True: while True:
try: try:
data = fifo_read.readline() # Blocking read data = fifo_read.readline() # Blocking read
@@ -148,9 +148,11 @@ def parse_status():
#logger.debug('Parsed "{0}"'.format(psd)) #logger.debug('Parsed "{0}"'.format(psd))
except BrokenPipeError: except BrokenPipeError:
logger.debug('Broken pipe in parse status thread, exiting') logger.debug('Broken pipe in parse status thread, exiting')
health_logger.info('Broken pipe in parse status thread, exiting')
clean_up() clean_up()
except: except:
logger.debug('Unknown exception in parse status thread, exiting') logger.debug('Unknown exception in parse status thread, exiting')
health_logger.info('Unknown exception in parse status thread, exiting')
clean_up() clean_up()
def exec_commands(): def exec_commands():
@@ -175,9 +177,10 @@ def exec_commands():
logger.debug('Read: "{0}"'.format(data.strip('\n'))) logger.debug('Read: "{0}"'.format(data.strip('\n')))
except BrokenPipeError: except BrokenPipeError:
logger.debug('Broken pipe in exec commands thread, exiting') logger.debug('Broken pipe in exec commands thread, exiting')
health_logger.info('Broken pipe in exec commands thread, exiting')
clean_up() clean_up()
except: except:
logger.debug('Unknown exception in exec commands thread, exiting') health_logger.info('Unknown exception in exec commands thread, exiting')
raise raise
clean_up() clean_up()
def strip_ansi_unicode(s): def strip_ansi_unicode(s):
@@ -187,18 +190,19 @@ def strip_ansi_unicode(s):
strip_unicode = (strip_ansi.encode('ascii', 'ignore')).decode('utf-8') strip_unicode = (strip_ansi.encode('ascii', 'ignore')).decode('utf-8')
return strip_unicode return strip_unicode
def handle_bluetooth(): def user_screen():
common.p_bluetooth = subprocess.Popen('bluetoothctl' # Attach to screen session, creating if needed, in UTF-8 mode
common.p_screen = subprocess.Popen('screen -D -R -U -S lemonbar' # TODO lemonbar_user
,stdin=subprocess.PIPE, stdout=subprocess.PIPE, text=True) ,stdin=subprocess.PIPE, stdout=subprocess.PIPE, text=True)
for line in common.p_bluetooth.stdout: for line in common.p_screen.stdout:
#try: #try:
logger.debug('Bluetooth read line {}'.format(line.strip('\n'))) logger.debug('Screen read line {}'.format(line.strip('\n')))
with open(config.fifo_file_status, 'w') as fifo: # with open(config.fifo_file_status, 'w') as fifo:
clean_line = strip_ansi_unicode(line) # clean_line = strip_ansi_unicode(line)
if (not clean_line.startswith('[bluetooth]') # if (not clean_line.startswith('[bluetooth]')
and not clean_line.startswith('[CHG]') # and not clean_line.startswith('[CHG]')
and not clean_line.isspace()): # and not clean_line.isspace()):
fifo.write('RESP [bluetoothctl] ' + clean_line) # fifo.write('RESP [bluetoothctl] ' + clean_line)
#except: #except:
# raise # raise
@@ -207,12 +211,18 @@ class i3_thread:
all_threads = [] # Static, contains all started threads all_threads = [] # Static, contains all started threads
def __init__(self, target, desc): def __init__(self, target, desc):
self.thread = Thread(target = target) self.thread = Thread(target = self.__class__.run, args=(target,desc))
self.desc = desc self.desc = desc
self.all_threads.append(self) self.all_threads.append(self)
health_logger.info('"%s" starting', desc)
self.thread.start() self.thread.start()
# Wrapper around the target
def run(target, desc):
target()
health_logger.info('"%s" reached end', desc)
def join_threads(): def join_threads():
for i3_th in i3_thread.all_threads: for i3_th in i3_thread.all_threads:
i3_th.thread.join() i3_th.thread.join()
@@ -254,7 +264,7 @@ if __name__ == "__main__":
i3_thread(target = exec_commands, desc='Exec commands thread') i3_thread(target = exec_commands, desc='Exec commands thread')
i3_thread(target = parse_status, desc='Parse status thread') i3_thread(target = parse_status, desc='Parse status thread')
i3_thread(target = write_sys_status, desc='Write sys status thread') i3_thread(target = write_sys_status, desc='Write sys status thread')
i3_thread(target = handle_bluetooth, desc='Bluetooth thread') i3_thread(target = user_screen, desc='Screen thread for user commands')
logger.debug('Threads started') logger.debug('Threads started')
i3_thread.join_threads() i3_thread.join_threads()