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() if self.dummy is not None: # Begin by parsing dummy data to create all lemonbar elements parsed = self.parse(self.dummy) common.parsing_queue.put([parsed]) def run(self): common.logger.info('Started module {}'.format(self.name)) common.health_logger.info('Module {} up'.format(self.name)) while True: line = self.status_handle.readline() if not line: common.logger.info('Reached end of module {}'.format(self.name)) common.health_logger.info('Module {} down'.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 def format_load(data, module, alert): # Helper function to format network modules # Changes colors scheme to inactive when interfaces are down, or to alert when # alert level is reached # Returns tuple (down, up) if data[0] == 'down': # wlan module.alt_scheme = parser.COLOR_SCHEME.INA return ('x', 'x') else: (d_v, u_v) = (data[0],data[1]) if max(float(d_v), float(u_v)) > float(alert): module.alt_scheme = parser.COLOR_SCHEME.NET_ALERT else: # Reset to default module.alt_scheme = None return (d_v, u_v) class ConkyFastModule(LemonModule): def __init__(self): self.prefix = 'CNK_FAST' self.dummy = 'CNK_FAST Fri 23 Aug 00:00:00 MUTE down down down down' 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 self.wlan_load = parser.LemonUnit('wlan_load', action='wlan') self.eth_load = parser.LemonUnit('eth_load', action='eth') self.volume = parser.LemonUnit('volume', action='pavu') self.date = parser.LemonUnit('date', action='date') self.time = parser.LemonUnit('time', action='toggle_secs' , alt_scheme=parser.COLOR_SCHEME.SPECIAL) parser.g_parser.register_unit(self.wlan_load) parser.g_parser.register_unit(self.eth_load) parser.g_parser.register_unit(self.volume) parser.g_parser.register_unit(self.date) parser.g_parser.register_unit(self.time) def _stop_module(self): if self.p_handle is not None: self.p_handle.terminate() parser.g_parser.remove_unit(self.wlan_load) parser.g_parser.remove_unit(self.eth_load) parser.g_parser.remove_unit(self.volume) parser.g_parser.remove_unit(self.date) parser.g_parser.remove_unit(self.time) def _parse_data(self, data): # wlan and eth (wland_v, wlanu_v) = format_load(data[5:7], self.wlan_load, config.net_alert) self.wlan_load.items = [(config.icon_wlan + config.icon_dl, wland_v) ,(config.icon_ul, wlanu_v)] (ethd_v, ethu_v) = format_load(data[7:9], self.eth_load, config.net_alert) self.eth_load.items = [(config.icon_eth + config.icon_dl, ethd_v) ,(config.icon_ul, ethu_v)] # Volume 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 self.volume.items = [(icon_v, vols)] # Date and time self.date.items = [(config.icon_clock, ' '.join(data[0:3]))] self.time.items = [('', data[3] if common.show_secs else data[3][:-3])] class ConkySlowModule(LemonModule): def __init__(self): self.prefix = 'CNK_SLOW' self.dummy = 'CNK_SLOW 11 2.24G 91 74 F100 100.00 pl' 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 self.sys_load = parser.LemonUnit('sys_load', action='load') self.disk = parser.LemonUnit('disk') self.brightness = parser.LemonUnit('brightness', action='adj_br' , external={'BRIGHT': self.parse_brightness, 'BAJS': self.parse_brightness}) self.battery = parser.LemonUnit('battery', action='dpms') self.language = parser.LemonUnit('language', action='lang' , external={'LANG': self.parse_language}) parser.g_parser.register_unit(self.sys_load) parser.g_parser.register_unit(self.disk) parser.g_parser.register_unit(self.brightness) parser.g_parser.register_unit(self.battery) parser.g_parser.register_unit(self.language) def _stop_module(self): parser.g_parser.remove_unit(self.sys_load) parser.g_parser.remove_unit(self.disk) parser.g_parser.remove_unit(self.brightness) parser.g_parser.remove_unit(self.battery) parser.g_parser.remove_unit(self.language) if self.p_handle is not None: self.p_handle.terminate() def _parse_data(self, data): self.parse_sys_load (data[0:2]) # System load self.parse_disk (data[2:4]) # Disk usage self.parse_battery (data[4:5]) # Battery self.parse_brightness (data[5:6]) # Screen brightness self.parse_language (data[6:7]) # Language def parse_sys_load(self, data): if int(data[0]) > int(config.cpu_alert): self.sys_load.alt_scheme = parser.COLOR_SCHEME.CPU_ALERT else: self.sys_load.alt_scheme = None self.sys_load.items = [(config.icon_cpu, data[0] + '%') ,(config.icon_ul, data[1])] def parse_disk(self, data): self.disk.items = [(config.icon_hd , data[0] + '%') ,(config.icon_home, data[1] + '%')] def parse_battery(self, data): (batt_stat, batt) = (data[0][0], data[0][1:]) batt_i = int(batt) icon_batt = config.icon_charging if batt_stat == 'C' else \ config.icon_charged if batt_stat == 'F' else \ config.icon_batt_0 if batt_i < 20 else \ config.icon_batt_1 if batt_i < 40 else \ config.icon_batt_2 if batt_i < 60 else \ config.icon_batt_3 if batt_i < 80 else \ config.icon_batt_4 self.battery.items = [(icon_batt, batt+'%')] def parse_brightness(self, data): brtxt = str(int(float(data[0]))) self.brightness.items = [(config.icon_bright, brtxt+'%')] def parse_language(self, data): self.language.items = [(config.icon_lang, data[0])] def start_all(): global m_conky_fast, m_conky_slow m_conky_slow = ConkySlowModule() m_conky_fast = ConkyFastModule() m_conky_slow.start() m_conky_fast.start() def stop_all(): global m_conky_fast, m_conky_slow m_conky_slow.stop() m_conky_fast.stop()