glances.plugins.quicklook.PluginModel.update() - Code Metrics - Inspection of "Glances version 4.0.0" - nicolargo/glances - Measure and Improve Code Quality continuously with Scrutinizer

Test Failed

Push — master ( ee826a...d9056e )

by Nicolas

created 2024-05-12 08:45 UTC

glances.plugins.quicklook.PluginModel.update() B

↳ Parent: glances.plugins.quicklook

Complexity

Conditions

Size

Total Lines	48
Code Lines	27

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	7
eloc	27
nop	1
dl	0
loc	48
rs	7.8319
c	0
b	0
f	0

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

"""Quicklook plugin."""

from glances.logger import logger
from glances.cpu_percent import cpu_percent
from glances.plugins.load import get_load_average, get_nb_log_core, get_nb_phys_core
from glances.outputs.glances_bars import Bar
from glances.outputs.glances_sparklines import Sparkline
from glances.plugins.plugin.model import GlancesPluginModel

import psutil

# 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 = {
    'cpu': {
        'description': 'CPU percent usage',
        'unit': 'percent',
    },
    'mem': {
        'description': 'MEM percent usage',
        'unit': 'percent',
    },
    'swap': {
        'description': 'SWAP percent usage',
        'unit': 'percent',
    },
    'load': {
        'description': 'LOAD percent usage',
        'unit': 'percent',
    },
    'cpu_log_core': {
        'description': 'Number of logical CPU core',
        'unit': 'number',
    },
    'cpu_phys_core': {
        'description': 'Number of physical CPU core',
        'unit': 'number',
    },
    'cpu_name': {
        'description': 'CPU name',
    },
    'cpu_hz_current': {
        'description': 'CPU current frequency',
        'unit': 'hertz',
    },
    'cpu_hz': {
        'description': 'CPU max frequency',
        'unit': 'hertz',
    },
}

# Define the history items list
# All items in this list will be historised if the --enable-history tag is set
items_history_list = [
    {'name': 'cpu', 'description': 'CPU percent usage', 'y_unit': '%'},
    {'name': 'percpu', 'description': 'PERCPU percent usage', 'y_unit': '%'},
    {'name': 'mem', 'description': 'MEM percent usage', 'y_unit': '%'},
    {'name': 'swap', 'description': 'SWAP percent usage', 'y_unit': '%'},
    {'name': 'load', 'description': 'LOAD percent usage', 'y_unit': '%'},
]


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

    'stats' is a dictionary.
    """

    AVAILABLE_STATS_LIST = ['cpu', 'mem', 'swap', 'load']
    DEFAULT_STATS_LIST = ['cpu', 'mem', 'load']

    def __init__(self, args=None, config=None):
        """Init the quicklook plugin."""
        super(PluginModel, self).__init__(
            args=args, config=config, items_history_list=items_history_list, fields_description=fields_description
        )

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

        # Manage the maximum number of CPU to display (related to enhancement request #2734)
        self.max_cpu_display = config.get_int_value('percpu', 'max_cpu_display', 4) if config else 4

        # Define the stats list
        self.stats_list = self.get_conf_value('list', default=self.DEFAULT_STATS_LIST)
        if not set(self.stats_list).issubset(self.AVAILABLE_STATS_LIST):
            logger.warning('Quicklook plugin: Invalid stats list: {}'.format(self.stats_list))
            self.stats_list = self.AVAILABLE_STATS_LIST

    @GlancesPluginModel._check_decorator
    @GlancesPluginModel._log_result_decorator
    def update(self):
        """Update quicklook stats using the input method."""
        # Init new stats
        stats = self.get_init_value()

        # Grab quicklook stats: CPU, MEM and SWAP
        if self.input_method == 'local':
            # Get system information
            cpu_info = cpu_percent.get_info()
            stats['cpu_name'] = cpu_info['cpu_name']
            stats['cpu_hz_current'] = (
                self._mhz_to_hz(cpu_info['cpu_hz_current']) if cpu_info['cpu_hz_current'] is not None else None
            )
            stats['cpu_hz'] = self._mhz_to_hz(cpu_info['cpu_hz']) if cpu_info['cpu_hz'] is not None else None

            # Get the CPU percent value (global and per core)
            # Stats is shared across all plugins
            stats['cpu'] = cpu_percent.get()
            stats['percpu'] = cpu_percent.get(percpu=True)

            # Get the virtual and swap memory
            stats['mem'] = psutil.virtual_memory().percent
            try:
                stats['swap'] = psutil.swap_memory().percent
            except RuntimeError:
                # Correct issue in Illumos OS (see #1767)
                stats['swap'] = None

            # Get load
            stats['cpu_log_core'] = get_nb_log_core()
            stats['cpu_phys_core'] = get_nb_phys_core()
            try:
                # Load average is a tuple (1 min, 5 min, 15 min)
                # Process only the 15 min value (index 2)
                stats['load'] = get_load_average(percent=True)[2]
            except (TypeError, IndexError):
                stats['load'] = None

        elif self.input_method == 'snmp':
            # Not available
            pass

        # Update the stats
        self.stats = stats

        return self.stats

    def update_views(self):
        """Update stats views."""
        # Call the father's method
        super(PluginModel, self).update_views()

        # Alert for CPU, MEM and SWAP
        for key in self.stats_list:
            if key in self.stats:
                self.views[key]['decoration'] = self.get_alert(self.stats[key], header=key)

        # Alert for LOAD
        self.views['load']['decoration'] = self.get_alert(self.stats['load'], header='load')

        # Define the list of stats to display
        self.views['list'] = self.stats_list

    def msg_curse(self, args=None, max_width=10):
        """Return the list to display in the UI."""
        # Init the return message
        ret = []

        # Only process if stats exist...
        if not self.stats or self.is_disabled():
            return ret

        if not max_width:
            # No max_width defined, return an empty message
            logger.debug("No max_width defined for the {} plugin, it will not be displayed.".format(self.plugin_name))
            return ret

        # Define the data: Bar (default behavior) or Sparkline
        data = dict()
        for key in self.stats_list:
            if self.args.sparkline and self.history_enable() and not self.args.client:
                data[key] = Sparkline(max_width)
            else:
                # Fallback to bar if Sparkline module is not installed
                data[key] = Bar(max_width, bar_char=self.get_conf_value('bar_char', default=['|'])[0])

        # Build the string message
        ##########################

        # System information
        if 'cpu_hz_current' in self.stats and 'cpu_hz' in self.stats:
            msg_freq = ' {:.2f}/{:.2f}GHz'.format(
                self._hz_to_ghz(self.stats['cpu_hz_current']), self._hz_to_ghz(self.stats['cpu_hz'])
            )
        else:
            msg_freq = ''

        if 'cpu_name' in self.stats and (max_width - len(msg_freq) + 7) > 0:
            msg_name = '{:{width}}'.format(self.stats['cpu_name'], width=max_width - len(msg_freq) + 7)
        else:
            msg_name = '' if msg_freq == '' else 'Frequency'

        ret.append(self.curse_add_line(msg_name))
        ret.append(self.curse_add_line(msg_freq))
        ret.append(self.curse_new_line())

        # Loop over CPU, MEM and LOAD
        for key in self.stats_list:
            if key == 'cpu' and args.percpu:
                ret.extend(self._msg_per_cpu(data, key, max_width))
            else:
                if type(data[key]).__name__ == 'Sparkline':
                    # Sparkline display an history
                    data[key].percents = [i[1] for i in self.get_raw_history(item=key, nb=data[key].size)]
                    # A simple padding in order to align metrics to the right
                    data[key].percents += [None] * (data[key].size - len(data[key].percents))
                else:
                    # Bar only the last value
                    data[key].percent = self.stats[key]
                msg = '{:4} '.format(key.upper())
                ret.extend(self._msg_create_line(msg, data[key], key))
                ret.append(self.curse_new_line())

        # Remove the last new line
        ret.pop()

        # Return the message with decoration
        return ret

    def _msg_per_cpu(self, data, key, max_width):
        """Create per-cpu view"""
        ret = []

        # Get history (only used with the sparkline option)
        if type(data[key]).__name__ == 'Sparkline':
            raw_cpu = self.get_raw_history(item='percpu', nb=data[key].size)

        # Manage the maximum number of CPU to display (related to enhancement request #2734)
        if len(self.stats['percpu']) > self.max_cpu_display:
            # If the number of CPU is > max_cpu_display then sort and display top 'n'
            percpu_list = sorted(self.stats['percpu'], key=lambda x: x['total'], reverse=True)
        else:
            percpu_list = self.stats['percpu']

        # Display the first max_cpu_display CPU
        for cpu in percpu_list[0 : self.max_cpu_display]:
            cpu_id = cpu[cpu['key']]
            if type(data[key]).__name__ == 'Sparkline':
                # Sparkline display an history
                data[key].percents = [i[1][cpu_id]['total'] for i in raw_cpu]

                # A simple padding in order to align metrics to the right
                data[key].percents += [None] * (data[key].size - len(data[key].percents))
            else:
                # Bar will only display the last value
                data[key].percent = cpu['total']
            if cpu_id < 10:
                msg = '{:3}{} '.format(key.upper(), cpu_id)
            else:
                msg = '{:4} '.format(cpu_id)
            ret.extend(self._msg_create_line(msg, data[key], key))
            ret.append(self.curse_new_line())

        # Add a new line with sum of all others CPU
        if len(self.stats['percpu']) > self.max_cpu_display:
            if type(data[key]).__name__ == 'Sparkline':
                sum_other = Sparkline(max_width)
                # Sparkline display an history
                sum_other.percents = [
                    sum([i['total'] for i in r[1] if i[i['key']] >= self.max_cpu_display])
                    / len([i['total'] for i in r[1] if i[i['key']] >= self.max_cpu_display])
                    for r in raw_cpu
                ]
                # A simple padding in order to align metrics to the right
                sum_other.percents += [None] * (sum_other.size - len(sum_other.percents))
            else:
                # Bar will only display the last value
                sum_other = Bar(max_width, bar_char=self.get_conf_value('bar_char', default=['|'])[0])
                sum_other.percent = sum([i['total'] for i in percpu_list[self.max_cpu_display :]]) / len(
                    percpu_list[self.max_cpu_display :]
                )
            msg = msg = '{:3}* '.format(key.upper())
            ret.extend(self._msg_create_line(msg, sum_other, key))
            ret.append(self.curse_new_line())

        return ret

    def _msg_create_line(self, msg, data, key):
        """Create a new line to the Quick view."""
        return [
            self.curse_add_line(msg),
            self.curse_add_line(data.pre_char, decoration='BOLD'),
            self.curse_add_line(data.get(), self.get_views(key=key, option='decoration')),
            self.curse_add_line(data.post_char, decoration='BOLD'),
            self.curse_add_line('  '),
        ]

    def _hz_to_ghz(self, hz):
        """Convert Hz to Ghz."""
        return hz / 1000000000.0

    def _mhz_to_hz(self, hz):
        """Convert Mhz to Hz."""
        return hz * 1000000.0


1			# -- coding: utf-8 --
2			#
3			# This file is part of Glances.
4			#
5			# SPDX-FileCopyrightText: 2024 Nicolas Hennion <[email protected]>
6			#
7			# SPDX-License-Identifier: LGPL-3.0-only
8			#
9
10			"""Quicklook plugin."""
11
12			from glances.logger import logger
13			from glances.cpu_percent import cpu_percent
14			from glances.plugins.load import get_load_average, get_nb_log_core, get_nb_phys_core
15			from glances.outputs.glances_bars import Bar
16			from glances.outputs.glances_sparklines import Sparkline
17			from glances.plugins.plugin.model import GlancesPluginModel
18
19			import psutil
20
21			# Fields description
22			# description: human readable description
23			# short_name: shortname to use un UI
24			# unit: unit type
25			# rate: is it a rate ? If yes, // by time_since_update when displayed,
26			# min_symbol: Auto unit should be used if value > than 1 'X' (K, M, G)...
27			fields_description = {
28			'cpu': {
29			'description': 'CPU percent usage',
30			'unit': 'percent',
31			},
32			'mem': {
33			'description': 'MEM percent usage',
34			'unit': 'percent',
35			},
36			'swap': {
37			'description': 'SWAP percent usage',
38			'unit': 'percent',
39			},
40			'load': {
41			'description': 'LOAD percent usage',
42			'unit': 'percent',
43			},
44			'cpu_log_core': {
45			'description': 'Number of logical CPU core',
46			'unit': 'number',
47			},
48			'cpu_phys_core': {
49			'description': 'Number of physical CPU core',
50			'unit': 'number',
51			},
52			'cpu_name': {
53			'description': 'CPU name',
54			},
55			'cpu_hz_current': {
56			'description': 'CPU current frequency',
57			'unit': 'hertz',
58			},
59			'cpu_hz': {
60			'description': 'CPU max frequency',
61			'unit': 'hertz',
62			},
63			}
64
65			# Define the history items list
66			# All items in this list will be historised if the --enable-history tag is set
67			items_history_list = [
68			{'name': 'cpu', 'description': 'CPU percent usage', 'y_unit': '%'},
69			{'name': 'percpu', 'description': 'PERCPU percent usage', 'y_unit': '%'},
70			{'name': 'mem', 'description': 'MEM percent usage', 'y_unit': '%'},
71			{'name': 'swap', 'description': 'SWAP percent usage', 'y_unit': '%'},
72			{'name': 'load', 'description': 'LOAD percent usage', 'y_unit': '%'},
73			]
74
75
76			class PluginModel(GlancesPluginModel):
77			"""Glances quicklook plugin.
78
79			'stats' is a dictionary.
80			"""
81
82			AVAILABLE_STATS_LIST = ['cpu', 'mem', 'swap', 'load']
83			DEFAULT_STATS_LIST = ['cpu', 'mem', 'load']
84
85			def __init__(self, args=None, config=None):
86			"""Init the quicklook plugin."""
87			super(PluginModel, self).__init__(
88			args=args, config=config, items_history_list=items_history_list, fields_description=fields_description
89			)
90
91			# We want to display the stat in the curse interface
92			self.display_curse = True
93
94			# Manage the maximum number of CPU to display (related to enhancement request #2734)
95			self.max_cpu_display = config.get_int_value('percpu', 'max_cpu_display', 4) if config else 4
96
97			# Define the stats list
98			self.stats_list = self.get_conf_value('list', default=self.DEFAULT_STATS_LIST)
99			if not set(self.stats_list).issubset(self.AVAILABLE_STATS_LIST):
100			logger.warning('Quicklook plugin: Invalid stats list: {}'.format(self.stats_list))
101			self.stats_list = self.AVAILABLE_STATS_LIST
102
103			@GlancesPluginModel._check_decorator
104			@GlancesPluginModel._log_result_decorator
105			def update(self):
106			"""Update quicklook stats using the input method."""
107			# Init new stats
108			stats = self.get_init_value()
109
110			# Grab quicklook stats: CPU, MEM and SWAP
111			if self.input_method == 'local':
112			# Get system information
113			cpu_info = cpu_percent.get_info()
114			stats['cpu_name'] = cpu_info['cpu_name']
115			stats['cpu_hz_current'] = (
116			self._mhz_to_hz(cpu_info['cpu_hz_current']) if cpu_info['cpu_hz_current'] is not None else None
117			)
118			stats['cpu_hz'] = self._mhz_to_hz(cpu_info['cpu_hz']) if cpu_info['cpu_hz'] is not None else None
119
120			# Get the CPU percent value (global and per core)
121			# Stats is shared across all plugins
122			stats['cpu'] = cpu_percent.get()
123			stats['percpu'] = cpu_percent.get(percpu=True)
124
125			# Get the virtual and swap memory
126			stats['mem'] = psutil.virtual_memory().percent
127			try:
128			stats['swap'] = psutil.swap_memory().percent
129			except RuntimeError:
130			# Correct issue in Illumos OS (see #1767)
131			stats['swap'] = None
132
133			# Get load
134			stats['cpu_log_core'] = get_nb_log_core()
135			stats['cpu_phys_core'] = get_nb_phys_core()
136			try:
137			# Load average is a tuple (1 min, 5 min, 15 min)
138			# Process only the 15 min value (index 2)
139			stats['load'] = get_load_average(percent=True)[2]
140			except (TypeError, IndexError):
141			stats['load'] = None
142
143			elif self.input_method == 'snmp':
144			# Not available
145			pass
146
147			# Update the stats
148			self.stats = stats
149
150			return self.stats
151
152			def update_views(self):
153			"""Update stats views."""
154			# Call the father's method
155			super(PluginModel, self).update_views()
156
157			# Alert for CPU, MEM and SWAP
158			for key in self.stats_list:
159			if key in self.stats:
160			self.views[key]['decoration'] = self.get_alert(self.stats[key], header=key)
161
162			# Alert for LOAD
163			self.views['load']['decoration'] = self.get_alert(self.stats['load'], header='load')
164
165			# Define the list of stats to display
166			self.views['list'] = self.stats_list
167
168			def msg_curse(self, args=None, max_width=10):
169			"""Return the list to display in the UI."""
170			# Init the return message
171			ret = []
172
173			# Only process if stats exist...
174			if not self.stats or self.is_disabled():
175			return ret
176
177			if not max_width:
178			# No max_width defined, return an empty message
179			logger.debug("No max_width defined for the {} plugin, it will not be displayed.".format(self.plugin_name))
180			return ret
181
182			# Define the data: Bar (default behavior) or Sparkline
183			data = dict()
184			for key in self.stats_list:
185			if self.args.sparkline and self.history_enable() and not self.args.client:
186			data[key] = Sparkline(max_width)
187			else:
188			# Fallback to bar if Sparkline module is not installed
189			data[key] = Bar(max_width, bar_char=self.get_conf_value('bar_char', default=['\|'])[0])
190
191			# Build the string message
192			##########################
193
194			# System information
195			if 'cpu_hz_current' in self.stats and 'cpu_hz' in self.stats:
196			msg_freq = ' {:.2f}/{:.2f}GHz'.format(
197			self._hz_to_ghz(self.stats['cpu_hz_current']), self._hz_to_ghz(self.stats['cpu_hz'])
198			)
199			else:
200			msg_freq = ''
201
202			if 'cpu_name' in self.stats and (max_width - len(msg_freq) + 7) > 0:
203			msg_name = '{:{width}}'.format(self.stats['cpu_name'], width=max_width - len(msg_freq) + 7)
204			else:
205			msg_name = '' if msg_freq == '' else 'Frequency'
206
207			ret.append(self.curse_add_line(msg_name))
208			ret.append(self.curse_add_line(msg_freq))
209			ret.append(self.curse_new_line())
210
211			# Loop over CPU, MEM and LOAD
212			for key in self.stats_list:
213			if key == 'cpu' and args.percpu:
214			ret.extend(self._msg_per_cpu(data, key, max_width))
215			else:
216			if type(data[key]).__name__ == 'Sparkline':
217			# Sparkline display an history
218			data[key].percents = [i[1] for i in self.get_raw_history(item=key, nb=data[key].size)]
219			# A simple padding in order to align metrics to the right
220			data[key].percents += [None] * (data[key].size - len(data[key].percents))
221			else:
222			# Bar only the last value
223			data[key].percent = self.stats[key]
224			msg = '{:4} '.format(key.upper())
225			ret.extend(self._msg_create_line(msg, data[key], key))
226			ret.append(self.curse_new_line())
227
228			# Remove the last new line
229			ret.pop()
230
231			# Return the message with decoration
232			return ret
233
234			def _msg_per_cpu(self, data, key, max_width):
235			"""Create per-cpu view"""
236			ret = []
237
238			# Get history (only used with the sparkline option)
239			if type(data[key]).__name__ == 'Sparkline':
240			raw_cpu = self.get_raw_history(item='percpu', nb=data[key].size)
241
242			# Manage the maximum number of CPU to display (related to enhancement request #2734)
243			if len(self.stats['percpu']) > self.max_cpu_display:
244			# If the number of CPU is > max_cpu_display then sort and display top 'n'
245			percpu_list = sorted(self.stats['percpu'], key=lambda x: x['total'], reverse=True)
246			else:
247			percpu_list = self.stats['percpu']
248
249			# Display the first max_cpu_display CPU
250			for cpu in percpu_list[0 : self.max_cpu_display]:
251			cpu_id = cpu[cpu['key']]
252			if type(data[key]).__name__ == 'Sparkline':
253			# Sparkline display an history
254			data[key].percents = [i[1][cpu_id]['total'] for i in raw_cpu]
			0 ignored issues – show introduced 2024-04-26 06:17 UTC by Report Bug Copy Issue Report The variable `raw_cpu` does not seem to be defined in case `type(SubscriptNode).__name__ == 'Sparkline'` on line `239` is `False`. Are you sure this can never be the case? Loading history...
255			# A simple padding in order to align metrics to the right
256			data[key].percents += [None] * (data[key].size - len(data[key].percents))
257			else:
258			# Bar will only display the last value
259			data[key].percent = cpu['total']
260			if cpu_id < 10:
261			msg = '{:3}{} '.format(key.upper(), cpu_id)
262			else:
263			msg = '{:4} '.format(cpu_id)
264			ret.extend(self._msg_create_line(msg, data[key], key))
265			ret.append(self.curse_new_line())
266
267			# Add a new line with sum of all others CPU
268			if len(self.stats['percpu']) > self.max_cpu_display:
269			if type(data[key]).__name__ == 'Sparkline':
270			sum_other = Sparkline(max_width)
271			# Sparkline display an history
272			sum_other.percents = [
273			sum([i['total'] for i in r[1] if i[i['key']] >= self.max_cpu_display])
274			/ len([i['total'] for i in r[1] if i[i['key']] >= self.max_cpu_display])
275			for r in raw_cpu
276			]
277			# A simple padding in order to align metrics to the right
278			sum_other.percents += [None] * (sum_other.size - len(sum_other.percents))
279			else:
280			# Bar will only display the last value
281			sum_other = Bar(max_width, bar_char=self.get_conf_value('bar_char', default=['\|'])[0])
282			sum_other.percent = sum([i['total'] for i in percpu_list[self.max_cpu_display :]]) / len(
283			percpu_list[self.max_cpu_display :]
284			)
285			msg = msg = '{:3}* '.format(key.upper())
286			ret.extend(self._msg_create_line(msg, sum_other, key))
287			ret.append(self.curse_new_line())
288
289			return ret
290
291			def _msg_create_line(self, msg, data, key):
292			"""Create a new line to the Quick view."""
293			return [
294			self.curse_add_line(msg),
295			self.curse_add_line(data.pre_char, decoration='BOLD'),
296			self.curse_add_line(data.get(), self.get_views(key=key, option='decoration')),
297			self.curse_add_line(data.post_char, decoration='BOLD'),
298			self.curse_add_line(' '),
299			]
300
301			def _hz_to_ghz(self, hz):
302			"""Convert Hz to Ghz."""
303			return hz / 1000000000.0
304
305			def _mhz_to_hz(self, hz):
306			"""Convert Mhz to Hz."""
307			return hz * 1000000.0
308

nicolargo / glances

Push — master ( ee826a...d9056e )

glances.plugins.quicklook.PluginModel.update() B

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like