glances.plugins.alert.PluginModel.msg_curse() - Code Metrics - Inspection of "version 4.2.0" - nicolargo/glances - Measure and Improve Code Quality continuously with Scrutinizer

Test Failed

Push — master ( cc9054...a8608f )

by Nicolas

created 2024-10-20 16:02 UTC

glances.plugins.alert.PluginModel.msg_curse() A

↳ Parent: glances.plugins.alert

Complexity

Conditions

Size

Total Lines	10
Code Lines	5

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	3
eloc	5
nop	3
dl	0
loc	10
rs	10
c	0
b	0
f	0

#
# This file is part of Glances.
#
# SPDX-FileCopyrightText: 2024 Nicolas Hennion <[email protected]>
#
# SPDX-License-Identifier: LGPL-3.0-only
#

"""Alert plugin."""

from datetime import datetime
from functools import reduce

from glances.events_list import glances_events

# from glances.logger import logger
from glances.plugins.plugin.model import GlancesPluginModel

# {
#     "begin": "begin",
#     "end": "end",
#     "state": "WARNING|CRITICAL",
#     "type": "CPU|LOAD|MEM",
#     "max": MAX,
#     "avg": AVG,
#     "min": MIN,
#     "sum": SUM,
#     "count": COUNT,
#     "top": [top3 process list],
#     "desc": "Processes description",
#     "sort": "top sort key"
#     "global": "global alert message"
# }
# Fields description
# description: human readable description
# short_name: shortname to use un UI
# unit: unit type
# rate: is it a rate ? If yes, // by time_since_update when displayed,
# min_symbol: Auto unit should be used if value > than 1 'X' (K, M, G)...
fields_description = {
    'begin': {
        'description': 'Begin timestamp of the event',
        'unit': 'timestamp',
    },
    'end': {
        'description': 'End timestamp of the event (or -1 if ongoing)',
        'unit': 'timestamp',
    },
    'state': {
        'description': 'State of the event (WARNING|CRITICAL)',
        'unit': 'string',
    },
    'type': {
        'description': 'Type of the event (CPU|LOAD|MEM)',
        'unit': 'string',
    },
    'max': {
        'description': 'Maximum value during the event period',
        'unit': 'float',
    },
    'avg': {
        'description': 'Average value during the event period',
        'unit': 'float',
    },
    'min': {
        'description': 'Minimum value during the event period',
        'unit': 'float',
    },
    'sum': {
        'description': 'Sum of the values during the event period',
        'unit': 'float',
    },
    'count': {
        'description': 'Number of values during the event period',
        'unit': 'int',
    },
    'top': {
        'description': 'Top 3 processes name during the event period',
        'unit': 'list',
    },
    'desc': {
        'description': 'Description of the event',
        'unit': 'string',
    },
    'sort': {
        'description': 'Sort key of the top processes',
        'unit': 'string',
    },
    'global_msg': {
        'description': 'Global alert message',
        'unit': 'string',
    },
}


class PluginModel(GlancesPluginModel):
    """Glances alert plugin.

    Only for display.
    """

    def __init__(self, args=None, config=None):
        """Init the plugin."""
        super().__init__(args=args, config=config, stats_init_value=[], fields_description=fields_description)

        # We want to display the stat in the curse interface
        self.display_curse = True

        # Set the message position
        self.align = 'bottom'

        # Set the maximum number of events to display
        if config is not None and (config.has_section('alert') or config.has_section('alerts')):
            glances_events.set_max_events(config.get_int_value('alert', 'max_events', default=10))
            glances_events.set_min_duration(config.get_int_value('alert', 'min_duration', default=6))
            glances_events.set_min_interval(config.get_int_value('alert', 'min_interval', default=6))
        else:
            glances_events.set_max_events(10)
            glances_events.set_min_duration(6)
            glances_events.set_min_interval(6)

    def update(self):
        """Nothing to do here. Just return the global glances_log."""
        # Set the stats to the glances_events
        self.stats = glances_events.get()

    def build_hdr_msg(self, ret):
        def cond(elem):
            return elem['end'] == -1 and 'global_msg' in elem

        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"))

        return ret

    def add_new_line(self, ret, alert):
        ret.append(self.curse_new_line())

        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)")))

        return ret

    def add_duration(self, ret, alert):
        if alert['end'] > 0:
            # If finished display duration
            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(" - "))

        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'])
            ret.append(self.curse_add_line(msg))
        else:
            msg = str(alert['type'])
            ret.append(self.curse_add_line(msg, decoration=alert['state']))

        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))

        return ret

    def add_top_proc(self, ret, alert):
        top_process = ', '.join(alert['top'])
        if top_process != '':
            msg = f': {top_process}'
            ret.append(self.curse_add_line(msg))

        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():
            return abs(a - b) <= max(abs(a), abs(b)) * tolerance
        return a == b


1			#
2			# This file is part of Glances.
3			#
4			# SPDX-FileCopyrightText: 2024 Nicolas Hennion <[email protected]>
5			#
6			# SPDX-License-Identifier: LGPL-3.0-only
7			#
8
9			"""Alert plugin."""
10
11			from datetime import datetime
12			from functools import reduce
13
14			from glances.events_list import glances_events
15
16			# from glances.logger import logger
17			from glances.plugins.plugin.model import GlancesPluginModel
18
19			# {
20			# "begin": "begin",
21			# "end": "end",
22			# "state": "WARNING\|CRITICAL",
23			# "type": "CPU\|LOAD\|MEM",
24			# "max": MAX,
25			# "avg": AVG,
26			# "min": MIN,
27			# "sum": SUM,
28			# "count": COUNT,
29			# "top": [top3 process list],
30			# "desc": "Processes description",
31			# "sort": "top sort key"
32			# "global": "global alert message"
33			# }
34			# Fields description
35			# description: human readable description
36			# short_name: shortname to use un UI
37			# unit: unit type
38			# rate: is it a rate ? If yes, // by time_since_update when displayed,
39			# min_symbol: Auto unit should be used if value > than 1 'X' (K, M, G)...
40			fields_description = {
41			'begin': {
42			'description': 'Begin timestamp of the event',
43			'unit': 'timestamp',
44			},
45			'end': {
46			'description': 'End timestamp of the event (or -1 if ongoing)',
47			'unit': 'timestamp',
48			},
49			'state': {
50			'description': 'State of the event (WARNING\|CRITICAL)',
51			'unit': 'string',
52			},
53			'type': {
54			'description': 'Type of the event (CPU\|LOAD\|MEM)',
55			'unit': 'string',
56			},
57			'max': {
58			'description': 'Maximum value during the event period',
59			'unit': 'float',
60			},
61			'avg': {
62			'description': 'Average value during the event period',
63			'unit': 'float',
64			},
65			'min': {
66			'description': 'Minimum value during the event period',
67			'unit': 'float',
68			},
69			'sum': {
70			'description': 'Sum of the values during the event period',
71			'unit': 'float',
72			},
73			'count': {
74			'description': 'Number of values during the event period',
75			'unit': 'int',
76			},
77			'top': {
78			'description': 'Top 3 processes name during the event period',
79			'unit': 'list',
80			},
81			'desc': {
82			'description': 'Description of the event',
83			'unit': 'string',
84			},
85			'sort': {
86			'description': 'Sort key of the top processes',
87			'unit': 'string',
88			},
89			'global_msg': {
90			'description': 'Global alert message',
91			'unit': 'string',
92			},
93			}
94
95
96			class PluginModel(GlancesPluginModel):
97			"""Glances alert plugin.
98
99			Only for display.
100			"""
101
102			def __init__(self, args=None, config=None):
103			"""Init the plugin."""
104			super().__init__(args=args, config=config, stats_init_value=[], fields_description=fields_description)
105
106			# We want to display the stat in the curse interface
107			self.display_curse = True
108
109			# Set the message position
110			self.align = 'bottom'
111
112			# Set the maximum number of events to display
113			if config is not None and (config.has_section('alert') or config.has_section('alerts')):
114			glances_events.set_max_events(config.get_int_value('alert', 'max_events', default=10))
115			glances_events.set_min_duration(config.get_int_value('alert', 'min_duration', default=6))
116			glances_events.set_min_interval(config.get_int_value('alert', 'min_interval', default=6))
117			else:
118			glances_events.set_max_events(10)
119			glances_events.set_min_duration(6)
120			glances_events.set_min_interval(6)
121
122			def update(self):
123			"""Nothing to do here. Just return the global glances_log."""
124			# Set the stats to the glances_events
125			self.stats = glances_events.get()
126
127			def build_hdr_msg(self, ret):
128			def cond(elem):
129			return elem['end'] == -1 and 'global_msg' in elem
130
131			global_message = [elem['global_msg'] for elem in self.stats if cond(elem)]
132			title = global_message[0] if len(global_message) > 0 else "EVENTS history"
133
134			ret.append(self.curse_add_line(title, "TITLE"))
135
136			return ret
137
138			def add_new_line(self, ret, alert):
139			ret.append(self.curse_new_line())
140
141			return ret
142
143			def add_start_time(self, ret, alert):
144			timezone = datetime.now().astimezone().tzinfo
145			alert_dt = datetime.fromtimestamp(alert['begin'], tz=timezone)
146			ret.append(self.curse_add_line(alert_dt.strftime("%Y-%m-%d %H:%M:%S(%z)")))
147
148			return ret
149
150			def add_duration(self, ret, alert):
151			if alert['end'] > 0:
152			# If finished display duration
153			end = datetime.fromtimestamp(alert['end'])
154			begin = datetime.fromtimestamp(alert['begin'])
155			msg = f' ({end - begin})'
156			else:
157			msg = ' (ongoing)'
158			ret.append(self.curse_add_line(msg))
159			ret.append(self.curse_add_line(" - "))
160
161			return ret
162
163			def add_infos(self, ret, alert):
164			if alert['end'] > 0:
165			# If finished do not display status
166			msg = '{} on {}'.format(alert['state'], alert['type'])
167			ret.append(self.curse_add_line(msg))
168			else:
169			msg = str(alert['type'])
170			ret.append(self.curse_add_line(msg, decoration=alert['state']))
171
172			return ret
173
174			def add_min_mean_max(self, ret, alert):
175			if self.approx_equal(alert['min'], alert['max'], tolerance=0.1):
176			msg = ' ({:.1f})'.format(alert['avg'])
177			else:
178			msg = ' (Min:{:.1f} Mean:{:.1f} Max:{:.1f})'.format(alert['min'], alert['avg'], alert['max'])
179			ret.append(self.curse_add_line(msg))
180
181			return ret
182
183			def add_top_proc(self, ret, alert):
184			top_process = ', '.join(alert['top'])
185			if top_process != '':
186			msg = f': {top_process}'
187			ret.append(self.curse_add_line(msg))
188
189			return ret
190
191			def loop_over_alert(self, init, alert):
192			steps = [
193			self.add_new_line,
194			self.add_start_time,
195			self.add_duration,
196			self.add_infos,
197			self.add_min_mean_max,
198			self.add_top_proc,
199			]
200
201			return reduce(lambda ret, step: step(ret, alert), steps, init)
202
203			def msg_curse(self, args=None, max_width=None):
204			"""Return the dict to display in the curse interface."""
205			# Init the return message
206			init = []
207
208			# Only process if display plugin enable...
209			if not self.stats or self.is_disabled():
210			return init
211
212			return reduce(self.loop_over_alert, self.stats, self.build_hdr_msg(init))
213
214			def approx_equal(self, a, b, tolerance=0.0):
215			"""Compare a with b using the tolerance (if numerical)."""
216			if str(int(a)).isdigit() and str(int(b)).isdigit():
217			return abs(a - b) <= max(abs(a), abs(b)) * tolerance
218			return a == b
219

nicolargo / glances

Push — master ( cc9054...a8608f )

glances.plugins.alert.PluginModel.msg_curse() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like