glances.plugins.glances_sensors.Plugin.update()   D
last analyzed

Complexity

Conditions 13

Size

Total Lines 64
Code Lines 35

Duplication

Lines 0
Ratio 0 %

Importance

Changes 0
Metric Value
cc 13
eloc 35
nop 1
dl 0
loc 64
rs 4.2
c 0
b 0
f 0

How to fix   Long Method    Complexity   

Long Method

Small methods make your code easier to understand, in particular if combined with a good name. Besides, if your method is small, finding a good name is usually much easier.

For example, if you find yourself adding comments to a method's body, this is usually a good sign to extract the commented part to a new method, and use the comment as a starting point when coming up with a good name for this new method.

Commonly applied refactorings include:

Complexity

Complex classes like glances.plugins.glances_sensors.Plugin.update() often do a lot of different things. To break such a class down, we need to identify a cohesive component within that class. A common approach to find such a component is to look for fields/methods that share the same prefixes, or suffixes.

Once you have determined the fields that belong together, you can apply the Extract Class refactoring. If the component makes sense as a sub-class, Extract Subclass is also a candidate, and is often faster.

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