glances.plugins.glances_sensors.Plugin.update() - Code Metrics - Inspection of "Resolve conflict with master branch" - nicolargo/glances - Measure and Improve Code Quality continuously with Scrutinizer

Test Failed

Push — master ( aea994...69b639 )

by Nicolas

created 2021-05-02 14:37 UTC

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

↳ Parent: glances.plugins.glances_sensors

Complexity

Conditions

Size

Total Lines	68
Code Lines	41

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	14
eloc	41
nop	1
dl	0
loc	68
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.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.glancesgrabsensors = 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.batpercent_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

    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.glancesgrabsensors.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.glancesgrabsensors.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:
                batpercent = self.__set_type(self.batpercent_plugin.update(),
                                             'battery')
            except Exception as e:
                logger.error("Cannot grab battery percent (%s)" % e)
            else:
                # Append Batteries %
                stats.extend(batpercent)

        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 informations
        # Alert
        for i in self.stats:
            if not i['value']:
                continue
            if i['type'] == 'battery':
                self.views[i[self.get_key()]]['value']['decoration'] = self.get_alert(100 - i['value'], header=i['type'])
            else:
                self.views[i[self.get_key()]]['value']['decoration'] = self.get_alert(i['value'], header=i['type'])

    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_disable():
            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 = '{:>13}'.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'):
                    value = to_fahrenheit(i['value'])
                    unit = 'F'
                else:
                    value = i['value']
                    unit = i['unit']
                try:
                    msg = '{:>13.0f}{}'.format(value, unit)
                    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.stemps = {}
        try:
            # psutil>=5.1.0, Linux-only
            self.stemps = 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.sfans = {}
        try:
            # psutil>=5.2.0, Linux-only
            self.sfans = psutil.sensors_fans()
        except AttributeError:
            logger.debug("Cannot grab fans speed. Platform not supported.")
        else:
            self.init_fan = True

        # !!! Disable Fan: High CPU consumption with psutil 5.2.0 or higher
        # Delete the two followings lines when corrected (https://github.com/giampaolo/psutil/issues/1199)
        # Correct and tested with PsUtil 5.6.1 (Ubuntu 18.04)
        # self.init_fan = False
        # logger.debug("Fan speed sensors disable (see https://github.com/giampaolo/psutil/issues/1199)")

        # 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.stemps
            self.stemps = psutil.sensors_temperatures()
        elif type == SENSOR_FAN_UNIT and self.init_fan:
            input_list = self.sfans
            self.sfans = psutil.sensors_fans()
        else:
            return ret
        for chipname, chip in iteritems(input_list):
            i = 1
            for feature in chip:
                sensors_current = {}
                # Sensor name
                if feature.label == '':
                    sensors_current['label'] = chipname + ' ' + str(i)
                else:
                    sensors_current['label'] = feature.label
                # Fan speed and unit
                sensors_current['value'] = int(feature.current)
                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.plugins.glances_plugin import GlancesPlugin
31
32			SENSOR_TEMP_UNIT = 'C'
33			SENSOR_FAN_UNIT = 'R'
34
35
36			class Plugin(GlancesPlugin):
37			"""Glances sensors plugin.
38
39			The stats list includes both sensors and hard disks stats, if any.
40			The sensors are already grouped by chip type and then sorted by name.
41			The hard disks are already sorted by name.
42			"""
43
44			def __init__(self, args=None, config=None):
45			"""Init the plugin."""
46			super(Plugin, self).__init__(args=args,
47			config=config,
48			stats_init_value=[])
49
50			start_duration = Counter()
51
52			# Init the sensor class
53			start_duration.reset()
54			self.glancesgrabsensors = GlancesGrabSensors()
55			logger.debug("Generic sensor plugin init duration: {} seconds".format(start_duration.get()))
56
57			# Instance for the HDDTemp Plugin in order to display the hard disks
58			# temperatures
59			start_duration.reset()
60			self.hddtemp_plugin = HddTempPlugin(args=args, config=config)
61			logger.debug("HDDTemp sensor plugin init duration: {} seconds".format(start_duration.get()))
62
63			# Instance for the BatPercent in order to display the batteries
64			# capacities
65			start_duration.reset()
66			self.batpercent_plugin = BatPercentPlugin(args=args, config=config)
67			logger.debug("Battery sensor plugin init duration: {} seconds".format(start_duration.get()))
68
69			# We want to display the stat in the curse interface
70			self.display_curse = True
71
72			def get_key(self):
73			"""Return the key of the list."""
74			return 'label'
75
76			@GlancesPlugin._check_decorator
77			@GlancesPlugin._log_result_decorator
78			def update(self):
79			"""Update sensors stats using the input method."""
80			# Init new stats
81			stats = self.get_init_value()
82
83			if self.input_method == 'local':
84			# Update stats using the dedicated lib
85			stats = []
86			# Get the temperature
87			try:
88			temperature = self.__set_type(self.glancesgrabsensors.get('temperature_core'),
89			'temperature_core')
90			except Exception as e:
91			logger.error("Cannot grab sensors temperatures (%s)" % e)
92			else:
93			# Append temperature
94			stats.extend(temperature)
95			# Get the FAN speed
96			try:
97			fan_speed = self.__set_type(self.glancesgrabsensors.get('fan_speed'),
98			'fan_speed')
99			except Exception as e:
100			logger.error("Cannot grab FAN speed (%s)" % e)
101			else:
102			# Append FAN speed
103			stats.extend(fan_speed)
104			# Update HDDtemp stats
105			try:
106			hddtemp = self.__set_type(self.hddtemp_plugin.update(),
107			'temperature_hdd')
108			except Exception as e:
109			logger.error("Cannot grab HDD temperature (%s)" % e)
110			else:
111			# Append HDD temperature
112			stats.extend(hddtemp)
113			# Update batteries stats
114			try:
115			batpercent = self.__set_type(self.batpercent_plugin.update(),
116			'battery')
117			except Exception as e:
118			logger.error("Cannot grab battery percent (%s)" % e)
119			else:
120			# Append Batteries %
121			stats.extend(batpercent)
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 informations
168			# Alert
169			for i in self.stats:
170			if not i['value']:
171			continue
172			if i['type'] == 'battery':
173			self.views[i[self.get_key()]]['value']['decoration'] = self.get_alert(100 - i['value'], header=i['type'])
174			else:
175			self.views[i[self.get_key()]]['value']['decoration'] = self.get_alert(i['value'], header=i['type'])
176
177			def msg_curse(self, args=None, max_width=None):
178			"""Return the dict to display in the curse interface."""
179			# Init the return message
180			ret = []
181
182			# Only process if stats exist and display plugin enable...
183			if not self.stats or self.is_disable():
184			return ret
185
186			# Max size for the interface name
187			name_max_width = max_width - 12
188
189			# Header
190			msg = '{:{width}}'.format('SENSORS', width=name_max_width)
191			ret.append(self.curse_add_line(msg, "TITLE"))
192
193			# Stats
194			for i in self.stats:
195			# Do not display anything if no battery are detected
196			if i['type'] == 'battery' and i['value'] == []:
197			continue
198			# New line
199			ret.append(self.curse_new_line())
200			msg = '{:{width}}'.format(i["label"][:name_max_width],
201			width=name_max_width)
202			ret.append(self.curse_add_line(msg))
203			if i['value'] in (b'ERR', b'SLP', b'UNK', b'NOS'):
204			msg = '{:>13}'.format(i['value'])
205			ret.append(self.curse_add_line(
206			msg, self.get_views(item=i[self.get_key()],
207			key='value',
208			option='decoration')))
209			else:
210			if (args.fahrenheit and i['type'] != 'battery' and
211			i['type'] != 'fan_speed'):
212			value = to_fahrenheit(i['value'])
213			unit = 'F'
214			else:
215			value = i['value']
216			unit = i['unit']
217			try:
218			msg = '{:>13.0f}{}'.format(value, unit)
219			ret.append(self.curse_add_line(
220			msg, self.get_views(item=i[self.get_key()],
221			key='value',
222			option='decoration')))
223			except (TypeError, ValueError):
224			pass
225
226			return ret
227
228
229			class GlancesGrabSensors(object):
230			"""Get sensors stats."""
231
232			def __init__(self):
233			"""Init sensors stats."""
234			# Temperatures
235			self.init_temp = False
236			self.stemps = {}
237			try:
238			# psutil>=5.1.0, Linux-only
239			self.stemps = psutil.sensors_temperatures()
240			except AttributeError:
241			logger.debug("Cannot grab temperatures. Platform not supported.")
242			else:
243			self.init_temp = True
244			# Solve an issue #1203 concerning a RunTimeError warning message displayed
245			# in the curses interface.
246			warnings.filterwarnings("ignore")
247
248			# Fans
249			self.init_fan = False
250			self.sfans = {}
251			try:
252			# psutil>=5.2.0, Linux-only
253			self.sfans = psutil.sensors_fans()
254			except AttributeError:
255			logger.debug("Cannot grab fans speed. Platform not supported.")
256			else:
257			self.init_fan = True
258
259			# !!! Disable Fan: High CPU consumption with psutil 5.2.0 or higher
260			# Delete the two followings lines when corrected (https://github.com/giampaolo/psutil/issues/1199)
261			# Correct and tested with PsUtil 5.6.1 (Ubuntu 18.04)
262			# self.init_fan = False
263			# logger.debug("Fan speed sensors disable (see https://github.com/giampaolo/psutil/issues/1199)")
264
265			# Init the stats
266			self.reset()
267
268			def reset(self):
269			"""Reset/init the stats."""
270			self.sensors_list = []
271
272			def __update__(self):
273			"""Update the stats."""
274			# Reset the list
275			self.reset()
276
277			if not self.init_temp:
278			return self.sensors_list
279
280			# Temperatures sensors
281			self.sensors_list.extend(self.build_sensors_list(SENSOR_TEMP_UNIT))
282
283			# Fans sensors
284			self.sensors_list.extend(self.build_sensors_list(SENSOR_FAN_UNIT))
285
286			return self.sensors_list
287
288			def build_sensors_list(self, type):
289			"""Build the sensors list depending of the type.
290
291			type: SENSOR_TEMP_UNIT or SENSOR_FAN_UNIT
292
293			output: a list
294			"""
295			ret = []
296			if type == SENSOR_TEMP_UNIT and self.init_temp:
297			input_list = self.stemps
298			self.stemps = psutil.sensors_temperatures()
299			elif type == SENSOR_FAN_UNIT and self.init_fan:
300			input_list = self.sfans
301			self.sfans = psutil.sensors_fans()
302			else:
303			return ret
304			for chipname, chip in iteritems(input_list):
305			i = 1
306			for feature in chip:
307			sensors_current = {}
308			# Sensor name
309			if feature.label == '':
310			sensors_current['label'] = chipname + ' ' + str(i)
311			else:
312			sensors_current['label'] = feature.label
313			# Fan speed and unit
314			sensors_current['value'] = int(feature.current)
315			sensors_current['unit'] = type
316			# Add sensor to the list
317			ret.append(sensors_current)
318			i += 1
319			return ret
320
321			def get(self, sensor_type='temperature_core'):
322			"""Get sensors list."""
323			self.__update__()
324			if sensor_type == 'temperature_core':
325			ret = [s for s in self.sensors_list if s['unit'] == SENSOR_TEMP_UNIT]
326			elif sensor_type == 'fan_speed':
327			ret = [s for s in self.sensors_list if s['unit'] == SENSOR_FAN_UNIT]
328			else:
329			# Unknown type
330			logger.debug("Unknown sensor type %s" % sensor_type)
331			ret = []
332			return ret
333

nicolargo / glances

Push — master ( aea994...69b639 )

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