von/vdstatus
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

yay, found out about Thread.daemon lol, also more oop and less code duplication

Browse source
This commit is contained in:
Von Random 2016-11-19 18:44:40 +03:00
parent 240fff91ee
commit bb9658d053
8 changed files with 56 additions and 153 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

25
plugins/batt.py
View file

@ -1,20 +1,12 @@
import threading
import time
import plugins.common
BATTERY_DIR = '/sys/class/power_supply/BAT0/'
class PluginThread(threading.Thread):
class PluginThread(plugins.common.PluginThreadCommon):
def __init__(self, section, config, thread_id):
threading.Thread.__init__(self)
self.threadID = thread_id
self.status = dict()
if config.has_option(section, 'color'):
self.status['color'] = config.get(section, 'color')
self.freq = config.getint(section, 'freq', fallback=1)
self.hide = False
self.should_stop = False
super(PluginThread, self).__init__(section, config)
def main(self):
with open(BATTERY_DIR + 'capacity', 'r') as capacity, \
@ -34,14 +26,3 @@ class PluginThread(threading.Thread):
batt = 'BAT: ' + batt_capacity + '% ' + batt_stat
self.status['full_text'] = batt
def stop(self):
self.should_stop = True
def run(self):
while True:
if self.should_stop is False:
self.main()
time.sleep(self.freq)
else:
break

25
plugins/common.py Normal file
View file

@ -0,0 +1,25 @@
import threading
import time
class PluginThreadCommon:
def __init__(self, section, config):
self.status = dict()
self.hide = False
self.thread = threading.Thread(target=self.run)
self.thread.daemon = True
self.freq = config.getint(section, 'freq', fallback=1)
self.problem_value = config.getint(section, 'problem', fallback=70)
if config.has_option(section, 'color'):
self.status['color'] = config.get(section, 'color')
def start(self):
self.thread.start()
def main(self):
self.status['full_text'] = 'placeholder'
def run(self):
while True:
self.main()
time.sleep(self.freq)

24
plugins/date.py
View file

@ -1,29 +1,11 @@
import time
import threading
import plugins.common
class PluginThread(threading.Thread):
class PluginThread(plugins.common.PluginThreadCommon):
def __init__(self, section, config, thread_id):
threading.Thread.__init__(self)
self.threadID = thread_id
super(PluginThread, self).__init__(section, config)
self.date_format = config.get(section, 'format', fallback='%c')
self.status = dict()
if config.has_option(section, 'color'):
self.status['color'] = config.get(section, 'color')
self.freq = config.getint(section, 'freq', fallback=1)
self.hide = False
self.should_stop = False
def main(self):
self.status['full_text'] = time.strftime(self.date_format)
def stop(self):
self.should_stop = True
def run(self):
while True:
if self.should_stop is False:
self.main()
time.sleep(self.freq)
else:
break

26
plugins/disk.py
View file

@ -1,20 +1,11 @@
import plugins.common
import psutil
import threading
import time
class PluginThread(threading.Thread):
class PluginThread(plugins.common.PluginThreadCommon):
def __init__(self, section, config, thread_id):
threading.Thread.__init__(self)
self.threadID = thread_id
self.status = dict()
super(PluginThread, self).__init__(section, config)
self.part = config.get(section, 'part')
if config.has_option(section, 'color'):
self.status['color'] = config.get(section, 'color')
self.freq = config.getint(section, 'freq', fallback=30)
self.hide = False
self.should_stop = False
self.problem_value = config.getint(section, 'problem', fallback=70)
def main(self):
du_stat = psutil.disk_usage(self.part)
@ -27,14 +18,3 @@ class PluginThread(threading.Thread):
du_free = str(round(du_stat.free / 2**30, 2))
du = self.part + ': ' + du_free + 'G'
self.status['full_text'] = du
def stop(self):
self.should_stop = True
def run(self):
while True:
if self.should_stop is False:
self.main()
time.sleep(self.freq)
else:
break

27
plugins/load.py
View file

@ -1,21 +1,11 @@
import os
import time
import threading
import plugins.common
class PluginThread(threading.Thread):
class PluginThread(plugins.common.PluginThreadCommon):
def __init__(self, section, config, thread_id):
threading.Thread.__init__(self)
self.threadID = thread_id
self.date_format = config.get(section, 'format', fallback='%c')
self.status = dict()
if config.has_option(section, 'color'):
self.status['color'] = config.get(section, 'color')
self.freq = config.getint(section, 'freq', fallback=10)
super(PluginThread, self).__init__(section, config)
self.hide_ok = config.getboolean(section, 'hide_ok', fallback=False)
self.hide = False
self.should_stop = False
self.problem_value = config.getint(section, 'problem', fallback=100)
def main(self):
loads = os.getloadavg()
@ -27,14 +17,3 @@ class PluginThread(threading.Thread):
self.status['urgent'] = False
loads = [str(i) for i in loads]
self.status['full_text'] = 'LA: ' + ' '.join(loads)
def stop(self):
self.should_stop = True
def run(self):
while True:
if self.should_stop is False:
self.main()
time.sleep(self.freq)
else:
break

25
plugins/mem.py
View file

@ -1,32 +1,13 @@
import psutil
import threading
import time
import plugins.common
class PluginThread(threading.Thread):
class PluginThread(plugins.common.PluginThreadCommon):
def __init__(self, section, config, thread_id):
threading.Thread.__init__(self)
self.threadID = thread_id
self.status = dict()
if config.has_option(section, 'color'):
self.status['color'] = config.get(section, 'color')
self.freq = config.getint(section, 'freq', fallback=1)
self.hide = False
self.should_stop = False
super(PluginThread, self).__init__(section, config)
def main(self):
mem_stat = psutil.virtual_memory()
mem_available = str(round(mem_stat.available / 2**30, 2))
mem = 'RAM: ' + mem_available + 'G'
self.status['full_text'] = mem
def stop(self):
self.should_stop = True
def run(self):
while True:
if self.should_stop is False:
self.main()
time.sleep(self.freq)
else:
break

30
plugins/ping.py
View file

@ -1,21 +1,15 @@
import os
import random
import time
import threading
import plugins.common
class PluginThread(threading.Thread):
class PluginThread(plugins.common.PluginThreadCommon):
def __init__(self, section, config, thread_id):
threading.Thread.__init__(self)
self.threadID = thread_id
super(PluginThread, self).__init__(section, config)
self.hosts = config.get(section, 'hosts').split(',')
self.title = config.get(section, 'title')
self.timeout = config.get(section, 'timeout', fallback='150')
self.status = dict()
self.freq = config.getint(section, 'freq', fallback=5)
self.format_status('n/a')
self.hide = False
self.should_stop = False
def format_status(self, state):
self.status['full_text'] = self.title + ': ' + state
@ -34,21 +28,3 @@ class PluginThread(threading.Thread):
self.format_status('on')
break
self.format_status('off')
def sleep(self):
seconds = 0
while seconds < self.freq:
time.sleep(1)
seconds += 1
del seconds
def stop(self):
self.should_stop = True
def run(self):
while True:
if self.should_stop is False:
self.main()
self.sleep()
else:
break

27
vdstatus
View file

@ -1,6 +1,5 @@
#!/usr/bin/python3
# TODO: handle SIGINT properly
# TODO: remove code duplication in plugins, probably use a common class?
# TODO: add documentation / comments
# TODO: add a dummy plugin to use as a starting point
# TODO: interactivity support
@ -51,17 +50,18 @@ class PluginRunner:
def query(self):
outputs = list()
try:
for plugin in self.plugins_loaded:
if not plugin.hide:
outputs.append(plugin.status)
print(self.format_output(outputs), flush=True)
time.sleep(self.output_freq)
except (KeyboardInterrupt, SystemExit):
for plugin in self.plugins_loaded:
plugin.stop()
sys.exit('stopping threads...')
del outputs
for plugin in self.plugins_loaded:
if not plugin.hide:
outputs.append(plugin.status)
print(self.format_output(outputs), flush=True)
def run(self):
while True:
try:
self.query()
time.sleep(self.output_freq)
except (KeyboardInterrupt, SystemExit):
sys.exit()
@staticmethod
def format_i3wm(inputs):
@ -80,5 +80,4 @@ if __name__ == '__main__':
plugin_runner = PluginRunner(args.conf)
plugin_runner.start()
time.sleep(0.1)
while True:
plugin_runner.query()
plugin_runner.run()