Merge pull request #2941 from ariel-anieli/refactor-alert

Refactorized alert plugin

glances/plugins/alert/__init__.py

@@ -9,6 +9,7 @@
 """Alert plugin."""
 
 from datetime import datetime
+from functools import reduce
 
 from glances.events_list import glances_events
 
@@ -123,38 +124,43 @@ class PluginModel(GlancesPluginModel):
         # Set the stats to the glances_events
         self.stats = glances_events.get()
 
-    def msg_curse(self, args=None, max_width=None):
+    def build_hdr_msg(self, ret):
-        """Return the dict to display in the curse interface."""
+        def cond(elem):
-        # Init the return message
+            return elem['end'] == -1 and 'global_msg' in elem
-        ret = []
+
+        global_message = [elem['global_msg'] for elem in self.stats if cond(elem)]
+        title = global_message[0] if len(global_message) > 0 else "EVENTS history"
+
+        ret.append(self.curse_add_line(title, "TITLE"))
 
-        # Only process if display plugin enable...
-        if not self.stats or self.is_disabled():
         return ret
 
-        # Build the string message
+    def add_new_line(self, ret, alert):
-        # Header with the global message
-        global_message = [e['global_msg'] for e in self.stats if (e['end'] == -1 and 'global_msg' in e)]
-        if len(global_message) > 0:
-            ret.append(self.curse_add_line(global_message[0], "TITLE"))
-        else:
-            ret.append(self.curse_add_line("EVENTS history", "TITLE"))
-        # Loop over alerts
-        for alert in self.stats:
-            # New line
         ret.append(self.curse_new_line())
-            # Start
+
-            alert_dt = datetime.fromtimestamp(alert['begin'], tz=datetime.now().astimezone().tzinfo)
+        return ret
+
+    def add_start_time(self, ret, alert):
+        timezone = datetime.now().astimezone().tzinfo
+        alert_dt = datetime.fromtimestamp(alert['begin'], tz=timezone)
         ret.append(self.curse_add_line(alert_dt.strftime("%Y-%m-%d %H:%M:%S(%z)")))
-            # Duration
+
+        return ret
+
+    def add_duration(self, ret, alert):
         if alert['end'] > 0:
             # If finished display duration
-                msg = ' ({})'.format(datetime.fromtimestamp(alert['end']) - datetime.fromtimestamp(alert['begin']))
+            end = datetime.fromtimestamp(alert['end'])
+            begin = datetime.fromtimestamp(alert['begin'])
+            msg = f' ({end - begin})'
         else:
             msg = ' (ongoing)'
         ret.append(self.curse_add_line(msg))
         ret.append(self.curse_add_line(" - "))
-            # Infos
+
+        return ret
+
+    def add_infos(self, ret, alert):
         if alert['end'] > 0:
             # If finished do not display status
             msg = '{} on {}'.format(alert['state'], alert['type'])
@@ -162,13 +168,19 @@ class PluginModel(GlancesPluginModel):
         else:
             msg = str(alert['type'])
             ret.append(self.curse_add_line(msg, decoration=alert['state']))
-            # Min / Mean / Max
+
+        return ret
+
+    def add_min_mean_max(self, ret, alert):
         if self.approx_equal(alert['min'], alert['max'], tolerance=0.1):
             msg = ' ({:.1f})'.format(alert['avg'])
         else:
             msg = ' (Min:{:.1f} Mean:{:.1f} Max:{:.1f})'.format(alert['min'], alert['avg'], alert['max'])
         ret.append(self.curse_add_line(msg))
-            # Top processes
+
+        return ret
+
+    def add_top_proc(self, ret, alert):
         top_process = ', '.join(alert['top'])
         if top_process != '':
             msg = f': {top_process}'
@@ -176,6 +188,29 @@ class PluginModel(GlancesPluginModel):
 
         return ret
 
+    def loop_over_alert(self, init, alert):
+        steps = [
+            self.add_new_line,
+            self.add_start_time,
+            self.add_duration,
+            self.add_infos,
+            self.add_min_mean_max,
+            self.add_top_proc,
+        ]
+
+        return reduce(lambda ret, step: step(ret, alert), steps, init)
+
+    def msg_curse(self, args=None, max_width=None):
+        """Return the dict to display in the curse interface."""
+        # Init the return message
+        init = []
+
+        # Only process if display plugin enable...
+        if not self.stats or self.is_disabled():
+            return init
+
+        return reduce(self.loop_over_alert, self.stats, self.build_hdr_msg(init))
+
     def approx_equal(self, a, b, tolerance=0.0):
         """Compare a with b using the tolerance (if numerical)."""
         if str(int(a)).isdigit() and str(int(b)).isdigit():