glances.plugins.glances_sensors.GlancesGrabSensors.__update__() - Code Metrics - Inspection of "version 3.2.0" - nicolargo/glances - Measure and Improve Code Quality continuously with Scrutinizer

Test Failed

Push — master ( 69b639...e7fa0a )

by Nicolas

created 2021-07-03 08:40 UTC

GlancesGrabSensors.update() A

↳ Parent: glances.plugins.glances_sensors

Complexity

Conditions

Size

Total Lines	15
Code Lines	7

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	2
eloc	7
nop	1
dl	0
loc	15
rs	10
c	0
b	0
f	0

# -*- 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

        # 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.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
            # 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 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

        # 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
                # 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.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			# Not necessary to refresh every refresh time
73			# By default set to refresh * 2
74			if self.get_refresh() == args.time:
75			self.set_refresh(self.get_refresh() * 2)
76
77			def get_key(self):
78			"""Return the key of the list."""
79			return 'label'
80
81			@GlancesPlugin._check_decorator
82			@GlancesPlugin._log_result_decorator
83			def update(self):
84			"""Update sensors stats using the input method."""
85			# Init new stats
86			stats = self.get_init_value()
87
88			if self.input_method == 'local':
89			# Update stats using the dedicated lib
90			stats = []
91			# Get the temperature
92			try:
93			temperature = self.__set_type(self.glancesgrabsensors.get('temperature_core'),
94			'temperature_core')
95			except Exception as e:
96			logger.error("Cannot grab sensors temperatures (%s)" % e)
97			else:
98			# Append temperature
99			stats.extend(temperature)
100			# Get the FAN speed
101			try:
102			fan_speed = self.__set_type(self.glancesgrabsensors.get('fan_speed'),
103			'fan_speed')
104			except Exception as e:
105			logger.error("Cannot grab FAN speed (%s)" % e)
106			else:
107			# Append FAN speed
108			stats.extend(fan_speed)
109			# Update HDDtemp stats
110			try:
111			hddtemp = self.__set_type(self.hddtemp_plugin.update(),
112			'temperature_hdd')
113			except Exception as e:
114			logger.error("Cannot grab HDD temperature (%s)" % e)
115			else:
116			# Append HDD temperature
117			stats.extend(hddtemp)
118			# Update batteries stats
119			try:
120			batpercent = self.__set_type(self.batpercent_plugin.update(),
121			'battery')
122			except Exception as e:
123			logger.error("Cannot grab battery percent (%s)" % e)
124			else:
125			# Append Batteries %
126			stats.extend(batpercent)
127
128			elif self.input_method == 'snmp':
129			# Update stats using SNMP
130			# No standard:
131			# http://www.net-snmp.org/wiki/index.php/Net-SNMP_and_lm-sensors_on_Ubuntu_10.04
132
133			pass
134
135			# Global change on stats
136			self.stats = self.get_init_value()
137			for stat in stats:
138			# Do not take hide stat into account
139			if self.is_hide(stat["label"].lower()):
140			continue
141			# Set the alias for each stat
142			alias = self.has_alias(stat["label"].lower())
143			if alias:
144			stat["label"] = alias
145			# Update the stats
146			self.stats.append(stat)
147
148			return self.stats
149
150			def __set_type(self, stats, sensor_type):
151			"""Set the plugin type.
152
153			4 types of stats is possible in the sensors plugin:
154			- Core temperature: 'temperature_core'
155			- Fan speed: 'fan_speed'
156			- HDD temperature: 'temperature_hdd'
157			- Battery capacity: 'battery'
158			"""
159			for i in stats:
160			# Set the sensors type
161			i.update({'type': sensor_type})
162			# also add the key name
163			i.update({'key': self.get_key()})
164
165			return stats
166
167			def update_views(self):
168			"""Update stats views."""
169			# Call the father's method
170			super(Plugin, self).update_views()
171
172			# Add specifics informations
173			# Alert
174			for i in self.stats:
175			if not i['value']:
176			continue
177			# Alert processing
178			if i['type'] == 'temperature_core' and not self.is_limit('critical', stat_name=i['type']):
179			if i['critical'] is None:
180			alert = 'DEFAULT'
181			elif i['value'] >= i['critical']:
182			alert = 'CRITICAL'
183			elif i['warning'] is None:
184			alert = 'DEFAULT'
185			elif i['value'] >= i['warning']:
186			alert = 'WARNING'
187			else:
188			alert = 'OK'
189			elif i['type'] == 'battery':
190			alert = self.get_alert(current=100 - i['value'],
191			header=i['type'])
192			else:
193			alert = self.get_alert(current=i['value'],
194			header=i['type'])
195			# Set the alert in the view
196			self.views[i[self.get_key()]]['value']['decoration'] = alert
197
198			def msg_curse(self, args=None, max_width=None):
199			"""Return the dict to display in the curse interface."""
200			# Init the return message
201			ret = []
202
203			# Only process if stats exist and display plugin enable...
204			if not self.stats or self.is_disable():
205			return ret
206
207			# Max size for the interface name
208			name_max_width = max_width - 12
209
210			# Header
211			msg = '{:{width}}'.format('SENSORS', width=name_max_width)
212			ret.append(self.curse_add_line(msg, "TITLE"))
213
214			# Stats
215			for i in self.stats:
216			# Do not display anything if no battery are detected
217			if i['type'] == 'battery' and i['value'] == []:
218			continue
219			# New line
220			ret.append(self.curse_new_line())
221			msg = '{:{width}}'.format(i["label"][:name_max_width],
222			width=name_max_width)
223			ret.append(self.curse_add_line(msg))
224			if i['value'] in (b'ERR', b'SLP', b'UNK', b'NOS'):
225			msg = '{:>13}'.format(i['value'])
226			ret.append(self.curse_add_line(
227			msg, self.get_views(item=i[self.get_key()],
228			key='value',
229			option='decoration')))
230			else:
231			if (args.fahrenheit and i['type'] != 'battery' and
232			i['type'] != 'fan_speed'):
233			value = to_fahrenheit(i['value'])
234			unit = 'F'
235			else:
236			value = i['value']
237			unit = i['unit']
238			try:
239			msg = '{:>13.0f}{}'.format(value, unit)
240			ret.append(self.curse_add_line(
241			msg, self.get_views(item=i[self.get_key()],
242			key='value',
243			option='decoration')))
244			except (TypeError, ValueError):
245			pass
246
247			return ret
248
249
250			class GlancesGrabSensors(object):
251			"""Get sensors stats."""
252
253			def __init__(self):
254			"""Init sensors stats."""
255			# Temperatures
256			self.init_temp = False
257			self.stemps = {}
258			try:
259			# psutil>=5.1.0, Linux-only
260			self.stemps = psutil.sensors_temperatures()
261			except AttributeError:
262			logger.debug("Cannot grab temperatures. Platform not supported.")
263			else:
264			self.init_temp = True
265			# Solve an issue #1203 concerning a RunTimeError warning message displayed
266			# in the curses interface.
267			warnings.filterwarnings("ignore")
268
269			# Fans
270			self.init_fan = False
271			self.sfans = {}
272			try:
273			# psutil>=5.2.0, Linux-only
274			self.sfans = psutil.sensors_fans()
275			except AttributeError:
276			logger.debug("Cannot grab fans speed. Platform not supported.")
277			else:
278			self.init_fan = True
279
280			# Init the stats
281			self.reset()
282
283			def reset(self):
284			"""Reset/init the stats."""
285			self.sensors_list = []
286
287			def __update__(self):
288			"""Update the stats."""
289			# Reset the list
290			self.reset()
291
292			if not self.init_temp:
293			return self.sensors_list
294
295			# Temperatures sensors
296			self.sensors_list.extend(self.build_sensors_list(SENSOR_TEMP_UNIT))
297
298			# Fans sensors
299			self.sensors_list.extend(self.build_sensors_list(SENSOR_FAN_UNIT))
300
301			return self.sensors_list
302
303			def build_sensors_list(self, type):
304			"""Build the sensors list depending of the type.
305
306			type: SENSOR_TEMP_UNIT or SENSOR_FAN_UNIT
307
308			output: a list
309			"""
310			ret = []
311			if type == SENSOR_TEMP_UNIT and self.init_temp:
312			input_list = self.stemps
313			self.stemps = psutil.sensors_temperatures()
314			elif type == SENSOR_FAN_UNIT and self.init_fan:
315			input_list = self.sfans
316			self.sfans = psutil.sensors_fans()
317			else:
318			return ret
319			for chipname, chip in iteritems(input_list):
320			i = 1
321			for feature in chip:
322			sensors_current = {}
323			# Sensor name
324			if feature.label == '':
325			sensors_current['label'] = chipname + ' ' + str(i)
326			else:
327			sensors_current['label'] = feature.label
328			# Sensors value, limit and unit
329			sensors_current['value'] = int(getattr(feature, 'current', 0) if getattr(feature, 'current', 0) else 0)
330			warning = getattr(feature, 'high', None)
331			sensors_current['warning'] = int(warning) if warning is not None else None
332			critical = getattr(feature, 'critical', None)
333			sensors_current['critical'] = int(critical) if critical is not None else None
334			sensors_current['unit'] = type
335			# Add sensor to the list
336			ret.append(sensors_current)
337			i += 1
338			return ret
339
340			def get(self, sensor_type='temperature_core'):
341			"""Get sensors list."""
342			self.__update__()
343			if sensor_type == 'temperature_core':
344			ret = [s for s in self.sensors_list if s['unit'] == SENSOR_TEMP_UNIT]
345			elif sensor_type == 'fan_speed':
346			ret = [s for s in self.sensors_list if s['unit'] == SENSOR_FAN_UNIT]
347			else:
348			# Unknown type
349			logger.debug("Unknown sensor type %s" % sensor_type)
350			ret = []
351			return ret
352

nicolargo / glances

Push — master ( 69b639...e7fa0a )

GlancesGrabSensors.__update__() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like

GlancesGrabSensors.update() A