l.behm/bumblebee-status
1
1
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity

[modules/apt] Update to new API

Browse source 
While at it, remove dependency on "parse" module and replace with more
standard "re" dependency.
This commit is contained in:
tobi-wan-kenobi 2020-04-13 13:25:08 +02:00
parent 964407534f
commit 82502c5651
1 changed files with 42 additions and 43 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

85
modules/contrib/apt.py
View file

@ -1,80 +1,79 @@
# pylint: disable=C0111,R0903 # pylint: disable=C0111,R0903
"""Displays APT package update information (<to upgrade>/<to remove >) """Displays APT package update information (<to upgrade>/<to remove >)
Requires the following debian packages: Requires the following packages:
* python-parse
* aptitude * aptitude
""" """
import re
import threading import threading
from parse import *
import bumblebee.util import core.module
import bumblebee.input import core.widget
import bumblebee.output import core.decorators
import bumblebee.engine
APT_CHECK_PATH = ("aptitude full-upgrade --simulate --assume-yes") import util.cli
PATTERN = "{} packages upgraded, {} newly installed, {} to remove and {} not upgraded."
PATTERN = '{} packages upgraded, {} newly installed, {} to remove and {} not upgraded.'
def parse_result(to_parse): def parse_result(to_parse):
# We want to line with the iforamtion about package upgrade # We want to line with the iforamtion about package upgrade
line_to_parse = to_parse.split("\n")[-4] line_to_parse = to_parse.split('\n')[-4]
result = re.search('(.+) packages upgraded, (.+) newly installed, (.+) to remove', line_to_parse)
result = parse(PATTERN, line_to_parse) return int(result.group(1)), int(result.group(3))
return int(result[0]), int(result[2]) def get_apt_check_info(module):
widget = module.widget()
def get_apt_check_info(widget):
try: try:
res = bumblebee.util.execute(APT_CHECK_PATH) res = util.cli.execute('aptitude full-upgrade --simulate --assume-yes')
widget.set("error", None) widget.set('error', None)
except (RuntimeError, FileNotFoundError) as e: except (RuntimeError, FileNotFoundError) as e:
widget.set("error", "unable to query APT: {}".format(e)) widget.set('error', 'unable to query APT: {}'.format(e))
return return
to_upgrade = 0 to_upgrade = 0
to_remove = 0 to_remove = 0
try: try:
to_upgrade, to_remove = parse_result(res) to_upgrade, to_remove = parse_result(res)
except e: widget.set('to_upgrade', to_upgrade)
widget.set("error", "parse error: {}".format(e)) widget.set('to_remove', to_remove)
return except Exception as e:
widget.set('error', 'parse error: {}'.format(e))
widget.set("to_upgrade", to_upgrade) core.event.trigger('update', [ module.id ], redraw_only=True)
widget.set("to_remove", to_remove)
class Module(bumblebee.engine.Module): class Module(core.module.Module):
def __init__(self, engine, config): @core.decorators.every(minutes=30)
widget = bumblebee.output.Widget(full_text=self.updates) def __init__(self, config):
super(Module, self).__init__(engine, config, widget) super().__init__(config, core.widget.Widget(self.updates))
self.interval_factor(60) self.__thread = None
self.interval(30)
def updates(self, widget): def updates(self, widget):
result = [] if widget.get('error'):
if widget.get("error"): return widget.get('error')
return widget.get("error") return '{} to upgrade, {} to remove'.format(
for t in ["to_upgrade", "to_remove"]: widget.get('to_upgrade', 0), widget.get('to_remove', 0)
result.append(str(widget.get(t, 0))) )
return "/".join(result)
def update(self, widgets): def update(self):
thread = threading.Thread(target=get_apt_check_info, args=(widgets[0],)) if self.__thread and self.__thread.isAlive(): return
thread.start()
self.__thread = threading.Thread(target=get_apt_check_info, args=(self,))
self.__thread.start()
def state(self, widget): def state(self, widget):
cnt = 0 cnt = 0
ret = "good" ret = 'good'
for t in ["to_upgrade", "to_remove"]: for t in ['to_upgrade', 'to_remove']:
cnt += widget.get(t, 0) cnt += widget.get(t, 0)
if cnt > 50: if cnt > 50:
ret = "critical" ret = 'critical'
elif cnt > 0: elif cnt > 0:
ret = "warning" ret = 'warning'
if widget.get("error"): if widget.get('error'):
ret = "critical" ret = 'critical'
return ret return ret