glances.plugins.glances_sensors.Plugin.update() - Code Metrics - Inspection of "Update release note for Glances 3.2.5" - nicolargo/glances - Measure and Improve Code Quality continuously with Scrutinizer

Test Failed

Push — master ( 7e7379...128504 )

by Nicolas

created 2022-04-10 14:14 UTC

glances.plugins.glances_sensors.Plugin.update() F

↳ Parent: glances.plugins.glances_sensors

Complexity

Conditions

Size

Total Lines	64
Code Lines	37

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	14
eloc	37
nop	1
dl	0
loc	64
rs	3.6
c	0
b	0
f	0

How to fix Long Method Complexity

# -*- coding: utf-8 -*-
#
# This file is part of Glances.
#
# Copyright (C) 2019 Nicolargo <[email protected]>
#
# Glances is free software; you can redistribute it and/or modify
# it under the terms of the GNU Lesser General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# Glances is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Lesser General Public License for more details.
#
# You should have received a copy of the GNU Lesser General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.

"""Sensors plugin."""

import psutil
import warnings

from glances.logger import logger
from glances.compat import iteritems, to_fahrenheit
from glances.timer import Counter
from glances.plugins.sensors.glances_batpercent import Plugin as BatPercentPlugin
from glances.plugins.sensors.glances_hddtemp import Plugin as HddTempPlugin
from glances.outputs.glances_unicode import unicode_message
from glances.plugins.glances_plugin import GlancesPlugin

SENSOR_TEMP_UNIT = 'C'
SENSOR_FAN_UNIT = 'R'


class Plugin(GlancesPlugin):
    """Glances sensors plugin.

    The stats list includes both sensors and hard disks stats, if any.
    The sensors are already grouped by chip type and then sorted by name.
    The hard disks are already sorted by name.
    """

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

        start_duration = Counter()

        # Init the sensor class
        start_duration.reset()
        self.glances_grab_sensors = GlancesGrabSensors()
        logger.debug("Generic sensor plugin init duration: {} seconds".format(start_duration.get()))

        # Instance for the HDDTemp Plugin in order to display the hard disks
        # temperatures
        start_duration.reset()
        self.hddtemp_plugin = HddTempPlugin(args=args, config=config)
        logger.debug("HDDTemp sensor plugin init duration: {} seconds".format(start_duration.get()))

        # Instance for the BatPercent in order to display the batteries
        # capacities
        start_duration.reset()
        self.bat_percent_plugin = BatPercentPlugin(args=args, config=config)
        logger.debug("Battery sensor plugin init duration: {} seconds".format(start_duration.get()))

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

        # Not necessary to refresh every refresh time
        # By default set to refresh * 2
        if self.get_refresh() == args.time:
            self.set_refresh(self.get_refresh() * 2)

    def get_key(self):
        """Return the key of the list."""
        return 'label'

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

        if self.input_method == 'local':
            # Update stats using the dedicated lib
            stats = []
            # Get the temperature
            try:
                temperature = self.__set_type(self.glances_grab_sensors.get('temperature_core'), 'temperature_core')
            except Exception as e:
                logger.error("Cannot grab sensors temperatures (%s)" % e)
            else:
                # Append temperature
                stats.extend(temperature)
            # Get the FAN speed
            try:
                fan_speed = self.__set_type(self.glances_grab_sensors.get('fan_speed'), 'fan_speed')
            except Exception as e:
                logger.error("Cannot grab FAN speed (%s)" % e)
            else:
                # Append FAN speed
                stats.extend(fan_speed)
            # Update HDDtemp stats
            try:
                hddtemp = self.__set_type(self.hddtemp_plugin.update(), 'temperature_hdd')
            except Exception as e:
                logger.error("Cannot grab HDD temperature (%s)" % e)
            else:
                # Append HDD temperature
                stats.extend(hddtemp)
            # Update batteries stats
            try:
                bat_percent = self.__set_type(self.bat_percent_plugin.update(), 'battery')
            except Exception as e:
                logger.error("Cannot grab battery percent (%s)" % e)
            else:
                # Append Batteries %
                stats.extend(bat_percent)

        elif self.input_method == 'snmp':
            # Update stats using SNMP
            # No standard:
            # http://www.net-snmp.org/wiki/index.php/Net-SNMP_and_lm-sensors_on_Ubuntu_10.04

            pass

        # Global change on stats
        self.stats = self.get_init_value()
        for stat in stats:
            # Do not take hide stat into account
            if self.is_hide(stat["label"].lower()):
                continue
            # Set the alias for each stat
            alias = self.has_alias(stat["label"].lower())
            if alias:
                stat["label"] = alias
            # Update the stats
            self.stats.append(stat)

        return self.stats

    def __set_type(self, stats, sensor_type):
        """Set the plugin type.

        4 types of stats is possible in the sensors plugin:
        - Core temperature: 'temperature_core'
        - Fan speed: 'fan_speed'
        - HDD temperature: 'temperature_hdd'
        - Battery capacity: 'battery'
        """
        for i in stats:
            # Set the sensors type
            i.update({'type': sensor_type})
            # also add the key name
            i.update({'key': self.get_key()})

        return stats

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

        # Add specifics information
        # Alert
        for i in self.stats:
            if not i['value']:
                continue
            # Alert processing
            if i['type'] == 'temperature_core' and not self.is_limit('critical', stat_name=i['type']):
                if i['critical'] is None:
                    alert = 'DEFAULT'
                elif i['value'] >= i['critical']:
                    alert = 'CRITICAL'
                elif i['warning'] is None:
                    alert = 'DEFAULT'
                elif i['value'] >= i['warning']:
                    alert = 'WARNING'
                else:
                    alert = 'OK'
            elif i['type'] == 'battery':
                alert = self.get_alert(current=100 - i['value'], header=i['type'])
            else:
                alert = self.get_alert(current=i['value'], header=i['type'])
            # Set the alert in the view
            self.views[i[self.get_key()]]['value']['decoration'] = alert

    def battery_trend(self, stats):
        """Return the trend characterr for the battery"""
        if 'status' not in stats:
            return ''
        if stats['status'].startswith('Charg'):
            return unicode_message('ARROW_UP')
        elif stats['status'].startswith('Discharg'):
            return unicode_message('ARROW_DOWN')
        elif stats['status'].startswith('Full'):
            return unicode_message('CHECK')
        return ''

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

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

        # Max size for the interface name
        name_max_width = max_width - 12

        # Header
        msg = '{:{width}}'.format('SENSORS', width=name_max_width)
        ret.append(self.curse_add_line(msg, "TITLE"))

        # Stats
        for i in self.stats:
            # Do not display anything if no battery are detected
            if i['type'] == 'battery' and i['value'] == []:
                continue
            # New line
            ret.append(self.curse_new_line())
            msg = '{:{width}}'.format(i["label"][:name_max_width], width=name_max_width)
            ret.append(self.curse_add_line(msg))
            if i['value'] in (b'ERR', b'SLP', b'UNK', b'NOS'):
                msg = '{:>14}'.format(i['value'])
                ret.append(
                    self.curse_add_line(msg, self.get_views(item=i[self.get_key()], key='value', option='decoration'))
                )
            else:
                if args.fahrenheit and i['type'] != 'battery' and i['type'] != 'fan_speed':
                    trend = ''
                    value = to_fahrenheit(i['value'])
                    unit = 'F'
                else:
                    trend = self.battery_trend(i)
                    value = i['value']
                    unit = i['unit']
                try:
                    msg = '{:.0f}{}{}'.format(value, unit, trend)
                    msg = '{:>14}'.format(msg)
                    ret.append(
                        self.curse_add_line(
                            msg, self.get_views(item=i[self.get_key()], key='value', option='decoration')
                        )
                    )
                except (TypeError, ValueError):
                    pass

        return ret


class GlancesGrabSensors(object):
    """Get sensors stats."""

    def __init__(self):
        """Init sensors stats."""
        # Temperatures
        self.init_temp = False
        self.sensor_temps = {}
        try:
            # psutil>=5.1.0, Linux-only
            self.sensor_temps = psutil.sensors_temperatures()
        except AttributeError:
            logger.debug("Cannot grab temperatures. Platform not supported.")
        else:
            self.init_temp = True
            # Solve an issue #1203 concerning a RunTimeError warning message displayed
            # in the curses interface.
            warnings.filterwarnings("ignore")

        # Fans
        self.init_fan = False
        self.sensor_fans = {}
        try:
            # psutil>=5.2.0, Linux-only
            self.sensor_fans = psutil.sensors_fans()
        except AttributeError:
            logger.debug("Cannot grab fans speed. Platform not supported.")
        else:
            self.init_fan = True

        # Init the stats
        self.reset()

    def reset(self):
        """Reset/init the stats."""
        self.sensors_list = []

    def __update__(self):
        """Update the stats."""
        # Reset the list
        self.reset()

        if not self.init_temp:
            return self.sensors_list

        # Temperatures sensors
        self.sensors_list.extend(self.build_sensors_list(SENSOR_TEMP_UNIT))

        # Fans sensors
        self.sensors_list.extend(self.build_sensors_list(SENSOR_FAN_UNIT))

        return self.sensors_list

    def build_sensors_list(self, type):
        """Build the sensors list depending of the type.

        type: SENSOR_TEMP_UNIT or SENSOR_FAN_UNIT

        output: a list
        """
        ret = []
        if type == SENSOR_TEMP_UNIT and self.init_temp:
            input_list = self.sensor_temps
            self.sensor_temps = psutil.sensors_temperatures()
        elif type == SENSOR_FAN_UNIT and self.init_fan:
            input_list = self.sensor_fans
            self.sensor_fans = psutil.sensors_fans()
        else:
            return ret
        for chip_name, chip in iteritems(input_list):
            i = 1
            for feature in chip:
                sensors_current = {}
                # Sensor name
                if feature.label == '':
                    sensors_current['label'] = chip_name + ' ' + str(i)
                else:
                    sensors_current['label'] = feature.label
                # Sensors value, limit and unit
                sensors_current['value'] = int(getattr(feature, 'current', 0) if getattr(feature, 'current', 0) else 0)
                warning = getattr(feature, 'high', None)
                sensors_current['warning'] = int(warning) if warning is not None else None
                critical = getattr(feature, 'critical', None)
                sensors_current['critical'] = int(critical) if critical is not None else None
                sensors_current['unit'] = type
                # Add sensor to the list
                ret.append(sensors_current)
                i += 1
        return ret

    def get(self, sensor_type='temperature_core'):
        """Get sensors list."""
        self.__update__()
        if sensor_type == 'temperature_core':
            ret = [s for s in self.sensors_list if s['unit'] == SENSOR_TEMP_UNIT]
        elif sensor_type == 'fan_speed':
            ret = [s for s in self.sensors_list if s['unit'] == SENSOR_FAN_UNIT]
        else:
            # Unknown type
            logger.debug("Unknown sensor type %s" % sensor_type)
            ret = []
        return ret


1			# -- coding: utf-8 --
2			#
3			# This file is part of Glances.
4			#
5			# Copyright (C) 2019 Nicolargo <[email protected]>
6			#
7			# Glances is free software; you can redistribute it and/or modify
8			# it under the terms of the GNU Lesser General Public License as published by
9			# the Free Software Foundation, either version 3 of the License, or
10			# (at your option) any later version.
11			#
12			# Glances is distributed in the hope that it will be useful,
13			# but WITHOUT ANY WARRANTY; without even the implied warranty of
14			# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
15			# GNU Lesser General Public License for more details.
16			#
17			# You should have received a copy of the GNU Lesser General Public License
18			# along with this program. If not, see <http://www.gnu.org/licenses/>.
19
20			"""Sensors plugin."""
21
22			import psutil
23			import warnings
24
25			from glances.logger import logger
26			from glances.compat import iteritems, to_fahrenheit
27			from glances.timer import Counter
28			from glances.plugins.sensors.glances_batpercent import Plugin as BatPercentPlugin
29			from glances.plugins.sensors.glances_hddtemp import Plugin as HddTempPlugin
30			from glances.outputs.glances_unicode import unicode_message
31			from glances.plugins.glances_plugin import GlancesPlugin
32
33			SENSOR_TEMP_UNIT = 'C'
34			SENSOR_FAN_UNIT = 'R'
35
36
37			class Plugin(GlancesPlugin):
38			"""Glances sensors plugin.
39
40			The stats list includes both sensors and hard disks stats, if any.
41			The sensors are already grouped by chip type and then sorted by name.
42			The hard disks are already sorted by name.
43			"""
44
45			def __init__(self, args=None, config=None):
46			"""Init the plugin."""
47			super(Plugin, self).__init__(args=args, config=config, stats_init_value=[])
48
49			start_duration = Counter()
50
51			# Init the sensor class
52			start_duration.reset()
53			self.glances_grab_sensors = GlancesGrabSensors()
54			logger.debug("Generic sensor plugin init duration: {} seconds".format(start_duration.get()))
55
56			# Instance for the HDDTemp Plugin in order to display the hard disks
57			# temperatures
58			start_duration.reset()
59			self.hddtemp_plugin = HddTempPlugin(args=args, config=config)
60			logger.debug("HDDTemp sensor plugin init duration: {} seconds".format(start_duration.get()))
61
62			# Instance for the BatPercent in order to display the batteries
63			# capacities
64			start_duration.reset()
65			self.bat_percent_plugin = BatPercentPlugin(args=args, config=config)
66			logger.debug("Battery sensor plugin init duration: {} seconds".format(start_duration.get()))
67
68			# We want to display the stat in the curse interface
69			self.display_curse = True
70
71			# Not necessary to refresh every refresh time
72			# By default set to refresh * 2
73			if self.get_refresh() == args.time:
74			self.set_refresh(self.get_refresh() * 2)
75
76			def get_key(self):
77			"""Return the key of the list."""
78			return 'label'
79
80			@GlancesPlugin._check_decorator
81			@GlancesPlugin._log_result_decorator
82			def update(self):
83			"""Update sensors stats using the input method."""
84			# Init new stats
85			stats = self.get_init_value()
86
87			if self.input_method == 'local':
88			# Update stats using the dedicated lib
89			stats = []
90			# Get the temperature
91			try:
92			temperature = self.__set_type(self.glances_grab_sensors.get('temperature_core'), 'temperature_core')
93			except Exception as e:
94			logger.error("Cannot grab sensors temperatures (%s)" % e)
95			else:
96			# Append temperature
97			stats.extend(temperature)
98			# Get the FAN speed
99			try:
100			fan_speed = self.__set_type(self.glances_grab_sensors.get('fan_speed'), 'fan_speed')
101			except Exception as e:
102			logger.error("Cannot grab FAN speed (%s)" % e)
103			else:
104			# Append FAN speed
105			stats.extend(fan_speed)
106			# Update HDDtemp stats
107			try:
108			hddtemp = self.__set_type(self.hddtemp_plugin.update(), 'temperature_hdd')
109			except Exception as e:
110			logger.error("Cannot grab HDD temperature (%s)" % e)
111			else:
112			# Append HDD temperature
113			stats.extend(hddtemp)
114			# Update batteries stats
115			try:
116			bat_percent = self.__set_type(self.bat_percent_plugin.update(), 'battery')
117			except Exception as e:
118			logger.error("Cannot grab battery percent (%s)" % e)
119			else:
120			# Append Batteries %
121			stats.extend(bat_percent)
122
123			elif self.input_method == 'snmp':
124			# Update stats using SNMP
125			# No standard:
126			# http://www.net-snmp.org/wiki/index.php/Net-SNMP_and_lm-sensors_on_Ubuntu_10.04
127
128			pass
129
130			# Global change on stats
131			self.stats = self.get_init_value()
132			for stat in stats:
133			# Do not take hide stat into account
134			if self.is_hide(stat["label"].lower()):
135			continue
136			# Set the alias for each stat
137			alias = self.has_alias(stat["label"].lower())
138			if alias:
139			stat["label"] = alias
140			# Update the stats
141			self.stats.append(stat)
142
143			return self.stats
144
145			def __set_type(self, stats, sensor_type):
146			"""Set the plugin type.
147
148			4 types of stats is possible in the sensors plugin:
149			- Core temperature: 'temperature_core'
150			- Fan speed: 'fan_speed'
151			- HDD temperature: 'temperature_hdd'
152			- Battery capacity: 'battery'
153			"""
154			for i in stats:
155			# Set the sensors type
156			i.update({'type': sensor_type})
157			# also add the key name
158			i.update({'key': self.get_key()})
159
160			return stats
161
162			def update_views(self):
163			"""Update stats views."""
164			# Call the father's method
165			super(Plugin, self).update_views()
166
167			# Add specifics information
168			# Alert
169			for i in self.stats:
170			if not i['value']:
171			continue
172			# Alert processing
173			if i['type'] == 'temperature_core' and not self.is_limit('critical', stat_name=i['type']):
174			if i['critical'] is None:
175			alert = 'DEFAULT'
176			elif i['value'] >= i['critical']:
177			alert = 'CRITICAL'
178			elif i['warning'] is None:
179			alert = 'DEFAULT'
180			elif i['value'] >= i['warning']:
181			alert = 'WARNING'
182			else:
183			alert = 'OK'
184			elif i['type'] == 'battery':
185			alert = self.get_alert(current=100 - i['value'], header=i['type'])
186			else:
187			alert = self.get_alert(current=i['value'], header=i['type'])
188			# Set the alert in the view
189			self.views[i[self.get_key()]]['value']['decoration'] = alert
190
191			def battery_trend(self, stats):
192			"""Return the trend characterr for the battery"""
193			if 'status' not in stats:
194			return ''
195			if stats['status'].startswith('Charg'):
196			return unicode_message('ARROW_UP')
197			elif stats['status'].startswith('Discharg'):
198			return unicode_message('ARROW_DOWN')
199			elif stats['status'].startswith('Full'):
200			return unicode_message('CHECK')
201			return ''
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			ret = []
207
208			# Only process if stats exist and display plugin enable...
209			if not self.stats or self.is_disabled():
210			return ret
211
212			# Max size for the interface name
213			name_max_width = max_width - 12
214
215			# Header
216			msg = '{:{width}}'.format('SENSORS', width=name_max_width)
217			ret.append(self.curse_add_line(msg, "TITLE"))
218
219			# Stats
220			for i in self.stats:
221			# Do not display anything if no battery are detected
222			if i['type'] == 'battery' and i['value'] == []:
223			continue
224			# New line
225			ret.append(self.curse_new_line())
226			msg = '{:{width}}'.format(i["label"][:name_max_width], width=name_max_width)
227			ret.append(self.curse_add_line(msg))
228			if i['value'] in (b'ERR', b'SLP', b'UNK', b'NOS'):
229			msg = '{:>14}'.format(i['value'])
230			ret.append(
231			self.curse_add_line(msg, self.get_views(item=i[self.get_key()], key='value', option='decoration'))
232			)
233			else:
234			if args.fahrenheit and i['type'] != 'battery' and i['type'] != 'fan_speed':
235			trend = ''
236			value = to_fahrenheit(i['value'])
237			unit = 'F'
238			else:
239			trend = self.battery_trend(i)
240			value = i['value']
241			unit = i['unit']
242			try:
243			msg = '{:.0f}{}{}'.format(value, unit, trend)
244			msg = '{:>14}'.format(msg)
245			ret.append(
246			self.curse_add_line(
247			msg, self.get_views(item=i[self.get_key()], key='value', option='decoration')
248			)
249			)
250			except (TypeError, ValueError):
251			pass
252
253			return ret
254
255
256			class GlancesGrabSensors(object):
257			"""Get sensors stats."""
258
259			def __init__(self):
260			"""Init sensors stats."""
261			# Temperatures
262			self.init_temp = False
263			self.sensor_temps = {}
264			try:
265			# psutil>=5.1.0, Linux-only
266			self.sensor_temps = psutil.sensors_temperatures()
267			except AttributeError:
268			logger.debug("Cannot grab temperatures. Platform not supported.")
269			else:
270			self.init_temp = True
271			# Solve an issue #1203 concerning a RunTimeError warning message displayed
272			# in the curses interface.
273			warnings.filterwarnings("ignore")
274
275			# Fans
276			self.init_fan = False
277			self.sensor_fans = {}
278			try:
279			# psutil>=5.2.0, Linux-only
280			self.sensor_fans = psutil.sensors_fans()
281			except AttributeError:
282			logger.debug("Cannot grab fans speed. Platform not supported.")
283			else:
284			self.init_fan = True
285
286			# Init the stats
287			self.reset()
288
289			def reset(self):
290			"""Reset/init the stats."""
291			self.sensors_list = []
292
293			def __update__(self):
294			"""Update the stats."""
295			# Reset the list
296			self.reset()
297
298			if not self.init_temp:
299			return self.sensors_list
300
301			# Temperatures sensors
302			self.sensors_list.extend(self.build_sensors_list(SENSOR_TEMP_UNIT))
303
304			# Fans sensors
305			self.sensors_list.extend(self.build_sensors_list(SENSOR_FAN_UNIT))
306
307			return self.sensors_list
308
309			def build_sensors_list(self, type):
310			"""Build the sensors list depending of the type.
311
312			type: SENSOR_TEMP_UNIT or SENSOR_FAN_UNIT
313
314			output: a list
315			"""
316			ret = []
317			if type == SENSOR_TEMP_UNIT and self.init_temp:
318			input_list = self.sensor_temps
319			self.sensor_temps = psutil.sensors_temperatures()
320			elif type == SENSOR_FAN_UNIT and self.init_fan:
321			input_list = self.sensor_fans
322			self.sensor_fans = psutil.sensors_fans()
323			else:
324			return ret
325			for chip_name, chip in iteritems(input_list):
326			i = 1
327			for feature in chip:
328			sensors_current = {}
329			# Sensor name
330			if feature.label == '':
331			sensors_current['label'] = chip_name + ' ' + str(i)
332			else:
333			sensors_current['label'] = feature.label
334			# Sensors value, limit and unit
335			sensors_current['value'] = int(getattr(feature, 'current', 0) if getattr(feature, 'current', 0) else 0)
336			warning = getattr(feature, 'high', None)
337			sensors_current['warning'] = int(warning) if warning is not None else None
338			critical = getattr(feature, 'critical', None)
339			sensors_current['critical'] = int(critical) if critical is not None else None
340			sensors_current['unit'] = type
341			# Add sensor to the list
342			ret.append(sensors_current)
343			i += 1
344			return ret
345
346			def get(self, sensor_type='temperature_core'):
347			"""Get sensors list."""
348			self.__update__()
349			if sensor_type == 'temperature_core':
350			ret = [s for s in self.sensors_list if s['unit'] == SENSOR_TEMP_UNIT]
351			elif sensor_type == 'fan_speed':
352			ret = [s for s in self.sensors_list if s['unit'] == SENSOR_FAN_UNIT]
353			else:
354			# Unknown type
355			logger.debug("Unknown sensor type %s" % sensor_type)
356			ret = []
357			return ret
358

nicolargo / glances

Push — master ( 7e7379...128504 )

glances.plugins.glances_sensors.Plugin.update() F

Complexity

Size

Duplication

Importance

How to fix Long Method Complexity

Long Method

Complexity

Duplication Side-by-Side

Filter issues like