glances.plugins.sensors.GlancesGrabSensors.__fetch_psutil() - Code Metrics - Inspection of "Merge branch 'refs/heads/support/v4.0.x'" - nicolargo/glances - Measure and Improve Code Quality continuously with Scrutinizer

Test Failed

Push — master ( eb83f6...fecd23 )

unknown

created 2024-05-16 11:48 UTC

GlancesGrabSensors.__fetch_psutil() A

↳ Parent: glances.plugins.sensors

Complexity

Conditions

Size

Total Lines	14
Code Lines	7

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	3
eloc	7
nop	1
dl	0
loc	14
rs	10
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
#

"""Sensors plugin."""
from enum import Enum
from concurrent.futures import ThreadPoolExecutor
from typing import List, Dict, Literal, Any

import psutil
import warnings

from glances.logger import logger
from glances.globals import to_fahrenheit
from glances.timer import Counter
from glances.plugins.sensors.sensor.glances_batpercent import PluginModel as BatPercentPluginModel
from glances.plugins.sensors.sensor.glances_hddtemp import PluginModel as HddTempPluginModel
from glances.outputs.glances_unicode import unicode_message
from glances.plugins.plugin.model import GlancesPluginModel


class SensorType(str, Enum):
    CPU_TEMP = 'temperature_core'
    FAN_SPEED = 'fan_speed'
    HDD_TEMP = 'temperature_hdd'
    BATTERY = 'battery'


CPU_TEMP_UNIT = 'C'
FAN_SPEED_UNIT = 'R'
HDD_TEMP_UNIT = 'C'
BATTERY_UNIT = '%'

# Define the default refresh multiplicator
# Default value is 3 * Glances refresh time
# Can be overwritten by the refresh option in the sensors section of the glances.conf file
DEFAULT_REFRESH = 3

# 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 = {
    'label': {
        'description': 'Sensor label',
    },
    'unit': {
        'description': 'Sensor unit',
    },
    'value': {
        'description': 'Sensor value',
        'unit': 'number',
    },
    'warning': {
        'description': 'Warning threshold',
        'unit': 'number',
    },
    'critical': {
        'description': 'Critical threshold',
        'unit': 'number',
    },
    'type': {
        'description': 'Sensor type (one of battery, temperature_core, fan_speed)',
    },
}


class PluginModel(GlancesPluginModel):
    """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 label.
    The hard disks are already sorted by label.
    """

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

        # Init the sensor class
        start_duration.reset()
        glances_grab_sensors_cpu_temp = GlancesGrabSensors(SensorType.CPU_TEMP)
        logger.debug("CPU Temp sensor plugin init duration: {} seconds".format(start_duration.get()))

        start_duration.reset()
        glances_grab_sensors_fan_speed = GlancesGrabSensors(SensorType.FAN_SPEED)
        logger.debug("Fan speed 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()
        hddtemp_plugin = HddTempPluginModel(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()
        batpercent_plugin = BatPercentPluginModel(args=args, config=config)
        logger.debug("Battery sensor plugin init duration: {} seconds".format(start_duration.get()))

        self.sensors_grab_map: Dict[SensorType, Any] = {}

        if glances_grab_sensors_cpu_temp.init:
            self.sensors_grab_map[SensorType.CPU_TEMP] = glances_grab_sensors_cpu_temp

        if glances_grab_sensors_fan_speed.init:
            self.sensors_grab_map[SensorType.FAN_SPEED] = glances_grab_sensors_fan_speed

        self.sensors_grab_map[SensorType.HDD_TEMP] = hddtemp_plugin
        self.sensors_grab_map[SensorType.BATTERY] = batpercent_plugin

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

        # Not necessary to refresh every refresh time
        if args and self.get_refresh() == args.time:
            self.set_refresh(self.get_refresh() * DEFAULT_REFRESH)

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

    def __get_sensor_data(self, sensor_type: SensorType) -> List[Dict]:
        try:
            data = self.sensors_grab_map[sensor_type].update()
            data = self.__set_type(data, sensor_type)
        except Exception as e:
            logger.error(f"Cannot grab sensors `{sensor_type}` ({e})")
            return []
        else:
            return self.__transform_sensors(data)

    def __transform_sensors(self, threads_stats):
        """Hide, alias and sort the result"""
        stats_transformed = []
        for stat in threads_stats:
            # Hide sensors configured in the hide ou show configuration key
            if not self.is_display(stat["label"].lower()):
                continue
            # Set alias for sensors
            stat["label"] = self.__get_alias(stat)
            # Add the stat to the stats_transformed list
            stats_transformed.append(stat)
        # Remove duplicates thanks to https://stackoverflow.com/a/9427216/1919431
        stats_transformed = [dict(t) for t in {tuple(d.items()) for d in stats_transformed}]
        # Sort by label
        stats_transformed = sorted(stats_transformed, key=lambda d: d['label'])
        return stats_transformed

    @GlancesPluginModel._check_decorator
    @GlancesPluginModel._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':
            with ThreadPoolExecutor(max_workers=len(self.sensors_grab_map)) as executor:
                logger.debug(f"Sensors enabled sub plugins: {list(self.sensors_grab_map.keys())}")
                futures = {t: executor.submit(self.__get_sensor_data, t) for t in self.sensors_grab_map.keys()}

            # Merge the results
            for sensor_type, future in futures.items():
                try:
                    stats.extend(future.result())
                except Exception as e:
                    logger.error(f"Cannot parse sensors data for `{sensor_type}` ({e})")

        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

        # Update the stats
        self.stats = stats

        return self.stats

    def __get_alias(self, stats):
        """Return the alias of the sensor."""
        # Get the alias for each stat
        if self.has_alias(stats["label"].lower()):
            return self.has_alias(stats["label"].lower())
        elif self.has_alias("{}_{}".format(stats["label"], stats["type"]).lower()):
            return self.has_alias("{}_{}".format(stats["label"], stats["type"]).lower())
        else:
            return stats["label"]

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

        4 types of stats is possible in the sensors plugin:
        - Core temperature: SENSOR_TEMP_TYPE
        - Fan speed: SENSOR_FAN_TYPE
        - HDD temperature: 'temperature_hdd'
        - Battery capacity: 'battery'
        """
        for i in stats:
            # Set the sensors type
            i.update({'type': str(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(PluginModel, self).update_views()

        # Add specifics information
        # Alert
        for i in self.stats:
            if not i['value']:
                continue
            # Alert processing
            if i['type'] == SensorType.CPU_TEMP:
                if self.is_limit('critical', stat_name=SensorType.CPU_TEMP + '_' + i['label']):
                    # By default use the thresholds configured in the glances.conf file (see #2058)
                    alert = self.get_alert(current=i['value'], header=SensorType.CPU_TEMP + '_' + i['label'])
                else:
                    # Else use the system thresholds
                    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'] == SensorType.BATTERY:
                # Battery is in %
                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 character 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
        if max_width:
            name_max_width = max_width - 12
        else:
            # No max_width defined, return an emptu curse message
            logger.debug("No max_width defined for the {} plugin, it will not be displayed.".format(self.plugin_name))
            return ret

        # 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'] == SensorType.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'] != SensorType.BATTERY and i['type'] != SensorType.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, sensor_type: Literal[SensorType.FAN_SPEED, SensorType.CPU_TEMP]):
        """Init sensors stats."""
        self.sensor_type = sensor_type
        self.sensor_unit = CPU_TEMP_UNIT if self.sensor_type == SensorType.CPU_TEMP else FAN_SPEED_UNIT

        self.init = False
        try:
            self.__fetch_psutil()
            self.init = True
        except AttributeError:
            logger.debug(f"Cannot grab {sensor_type}. Platform not supported.")

    def __fetch_psutil(self) -> Dict[str, list]:
        if self.sensor_type == SensorType.CPU_TEMP:
            # Solve an issue #1203 concerning a RunTimeError warning message displayed
            # in the curses interface.
            warnings.filterwarnings("ignore")

            # psutil>=5.1.0, Linux-only
            return psutil.sensors_temperatures()

        if self.sensor_type == SensorType.FAN_SPEED:
            # psutil>=5.2.0, Linux-only
            return psutil.sensors_fans()

        raise ValueError(f"Unsupported sensor_type: {self.sensor_type}")

    def update(self) -> List[dict]:
        """Update the stats."""
        if not self.init:
            return []

        # Temperatures sensors
        ret = []
        data = self.__fetch_psutil()
        for chip_name, chip in data.items():
            label_index = 1
            for chip_name_index, feature in enumerate(chip):
                sensors_current = {}
                # Sensor name
                if feature.label == '':
                    sensors_current['label'] = chip_name + ' ' + str(chip_name_index)
                elif feature.label in [i['label'] for i in ret]:
                    sensors_current['label'] = feature.label + ' ' + str(label_index)
                    label_index += 1
                else:
                    sensors_current['label'] = feature.label
                # Sensors value, limit and unit
                sensors_current['unit'] = self.sensor_unit
                sensors_current['value'] = int(
                    getattr(feature, 'current', 0) if getattr(feature, 'current', 0) else 0)
                system_warning = getattr(feature, 'high', None)
                system_critical = getattr(feature, 'critical', None)
                sensors_current['warning'] = int(system_warning) if system_warning is not None else None
                sensors_current['critical'] = int(system_critical) if system_critical is not None else None
                # Add sensor to the list
                ret.append(sensors_current)
        return ret


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			"""Sensors plugin."""
11			from enum import Enum
12			from concurrent.futures import ThreadPoolExecutor
13			from typing import List, Dict, Literal, Any
14
15			import psutil
16			import warnings
17
18			from glances.logger import logger
19			from glances.globals import to_fahrenheit
20			from glances.timer import Counter
21			from glances.plugins.sensors.sensor.glances_batpercent import PluginModel as BatPercentPluginModel
22			from glances.plugins.sensors.sensor.glances_hddtemp import PluginModel as HddTempPluginModel
23			from glances.outputs.glances_unicode import unicode_message
24			from glances.plugins.plugin.model import GlancesPluginModel
25
26
27			class SensorType(str, Enum):
28			CPU_TEMP = 'temperature_core'
29			FAN_SPEED = 'fan_speed'
30			HDD_TEMP = 'temperature_hdd'
31			BATTERY = 'battery'
32
33
34			CPU_TEMP_UNIT = 'C'
35			FAN_SPEED_UNIT = 'R'
36			HDD_TEMP_UNIT = 'C'
37			BATTERY_UNIT = '%'
38
39			# Define the default refresh multiplicator
40			# Default value is 3 * Glances refresh time
41			# Can be overwritten by the refresh option in the sensors section of the glances.conf file
42			DEFAULT_REFRESH = 3
43
44			# Fields description
45			# description: human readable description
46			# short_name: shortname to use un UI
47			# unit: unit type
48			# rate: is it a rate ? If yes, // by time_since_update when displayed,
49			# min_symbol: Auto unit should be used if value > than 1 'X' (K, M, G)...
50			fields_description = {
51			'label': {
52			'description': 'Sensor label',
53			},
54			'unit': {
55			'description': 'Sensor unit',
56			},
57			'value': {
58			'description': 'Sensor value',
59			'unit': 'number',
60			},
61			'warning': {
62			'description': 'Warning threshold',
63			'unit': 'number',
64			},
65			'critical': {
66			'description': 'Critical threshold',
67			'unit': 'number',
68			},
69			'type': {
70			'description': 'Sensor type (one of battery, temperature_core, fan_speed)',
71			},
72			}
73
74
75			class PluginModel(GlancesPluginModel):
76			"""Glances sensors plugin.
77
78			The stats list includes both sensors and hard disks stats, if any.
79			The sensors are already grouped by chip type and then sorted by label.
80			The hard disks are already sorted by label.
81			"""
82
83			def __init__(self, args=None, config=None):
84			"""Init the plugin."""
85			super(PluginModel, self).__init__(
86			args=args, config=config, stats_init_value=[], fields_description=fields_description
87			)
88			start_duration = Counter()
89
90			# Init the sensor class
91			start_duration.reset()
92			glances_grab_sensors_cpu_temp = GlancesGrabSensors(SensorType.CPU_TEMP)
93			logger.debug("CPU Temp sensor plugin init duration: {} seconds".format(start_duration.get()))
94
95			start_duration.reset()
96			glances_grab_sensors_fan_speed = GlancesGrabSensors(SensorType.FAN_SPEED)
97			logger.debug("Fan speed sensor plugin init duration: {} seconds".format(start_duration.get()))
98
99			# Instance for the HDDTemp Plugin in order to display the hard disks temperatures
100			start_duration.reset()
101			hddtemp_plugin = HddTempPluginModel(args=args, config=config)
102			logger.debug("HDDTemp sensor plugin init duration: {} seconds".format(start_duration.get()))
103
104			# Instance for the BatPercent in order to display the batteries capacities
105			start_duration.reset()
106			batpercent_plugin = BatPercentPluginModel(args=args, config=config)
107			logger.debug("Battery sensor plugin init duration: {} seconds".format(start_duration.get()))
108
109			self.sensors_grab_map: Dict[SensorType, Any] = {}
110
111			if glances_grab_sensors_cpu_temp.init:
112			self.sensors_grab_map[SensorType.CPU_TEMP] = glances_grab_sensors_cpu_temp
113
114			if glances_grab_sensors_fan_speed.init:
115			self.sensors_grab_map[SensorType.FAN_SPEED] = glances_grab_sensors_fan_speed
116
117			self.sensors_grab_map[SensorType.HDD_TEMP] = hddtemp_plugin
118			self.sensors_grab_map[SensorType.BATTERY] = batpercent_plugin
119
120			# We want to display the stat in the curse interface
121			self.display_curse = True
122
123			# Not necessary to refresh every refresh time
124			if args and self.get_refresh() == args.time:
125			self.set_refresh(self.get_refresh() * DEFAULT_REFRESH)
126
127			def get_key(self):
128			"""Return the key of the list."""
129			return 'label'
130
131			def __get_sensor_data(self, sensor_type: SensorType) -> List[Dict]:
132			try:
133			data = self.sensors_grab_map[sensor_type].update()
134			data = self.__set_type(data, sensor_type)
135			except Exception as e:
136			logger.error(f"Cannot grab sensors `{sensor_type}` ({e})")
137			return []
138			else:
139			return self.__transform_sensors(data)
140
141			def __transform_sensors(self, threads_stats):
142			"""Hide, alias and sort the result"""
143			stats_transformed = []
144			for stat in threads_stats:
145			# Hide sensors configured in the hide ou show configuration key
146			if not self.is_display(stat["label"].lower()):
147			continue
148			# Set alias for sensors
149			stat["label"] = self.__get_alias(stat)
150			# Add the stat to the stats_transformed list
151			stats_transformed.append(stat)
152			# Remove duplicates thanks to https://stackoverflow.com/a/9427216/1919431
153			stats_transformed = [dict(t) for t in {tuple(d.items()) for d in stats_transformed}]
154			# Sort by label
155			stats_transformed = sorted(stats_transformed, key=lambda d: d['label'])
156			return stats_transformed
157
158			@GlancesPluginModel._check_decorator
159			@GlancesPluginModel._log_result_decorator
160			def update(self):
161			"""Update sensors stats using the input method."""
162			# Init new stats
163			stats = self.get_init_value()
164
165			if self.input_method == 'local':
166			with ThreadPoolExecutor(max_workers=len(self.sensors_grab_map)) as executor:
167			logger.debug(f"Sensors enabled sub plugins: {list(self.sensors_grab_map.keys())}")
168			futures = {t: executor.submit(self.__get_sensor_data, t) for t in self.sensors_grab_map.keys()}
169
170			# Merge the results
171			for sensor_type, future in futures.items():
172			try:
173			stats.extend(future.result())
174			except Exception as e:
175			logger.error(f"Cannot parse sensors data for `{sensor_type}` ({e})")
176
177			elif self.input_method == 'snmp':
178			# Update stats using SNMP
179			# No standard:
180			# http://www.net-snmp.org/wiki/index.php/Net-SNMP_and_lm-sensors_on_Ubuntu_10.04
181			pass
182
183			# Update the stats
184			self.stats = stats
185
186			return self.stats
187
188			def __get_alias(self, stats):
189			"""Return the alias of the sensor."""
190			# Get the alias for each stat
191			if self.has_alias(stats["label"].lower()):
192			return self.has_alias(stats["label"].lower())
193			elif self.has_alias("{}_{}".format(stats["label"], stats["type"]).lower()):
194			return self.has_alias("{}_{}".format(stats["label"], stats["type"]).lower())
195			else:
196			return stats["label"]
197
198			def __set_type(self, stats, sensor_type):
199			"""Set the plugin type.
200
201			4 types of stats is possible in the sensors plugin:
202			- Core temperature: SENSOR_TEMP_TYPE
203			- Fan speed: SENSOR_FAN_TYPE
204			- HDD temperature: 'temperature_hdd'
205			- Battery capacity: 'battery'
206			"""
207			for i in stats:
208			# Set the sensors type
209			i.update({'type': str(sensor_type)})
210			# also add the key name
211			i.update({'key': self.get_key()})
212
213			return stats
214
215			def update_views(self):
216			"""Update stats views."""
217			# Call the father's method
218			super(PluginModel, self).update_views()
219
220			# Add specifics information
221			# Alert
222			for i in self.stats:
223			if not i['value']:
224			continue
225			# Alert processing
226			if i['type'] == SensorType.CPU_TEMP:
227			if self.is_limit('critical', stat_name=SensorType.CPU_TEMP + '_' + i['label']):
228			# By default use the thresholds configured in the glances.conf file (see #2058)
229			alert = self.get_alert(current=i['value'], header=SensorType.CPU_TEMP + '_' + i['label'])
230			else:
231			# Else use the system thresholds
232			if i['critical'] is None:
233			alert = 'DEFAULT'
234			elif i['value'] >= i['critical']:
235			alert = 'CRITICAL'
236			elif i['warning'] is None:
237			alert = 'DEFAULT'
238			elif i['value'] >= i['warning']:
239			alert = 'WARNING'
240			else:
241			alert = 'OK'
242			elif i['type'] == SensorType.BATTERY:
243			# Battery is in %
244			alert = self.get_alert(current=100 - i['value'], header=i['type'])
245			else:
246			alert = self.get_alert(current=i['value'], header=i['type'])
247			# Set the alert in the view
248			self.views[i[self.get_key()]]['value']['decoration'] = alert
249
250			def battery_trend(self, stats):
251			"""Return the trend character for the battery"""
252			if 'status' not in stats:
253			return ''
254			if stats['status'].startswith('Charg'):
255			return unicode_message('ARROW_UP')
256			elif stats['status'].startswith('Discharg'):
257			return unicode_message('ARROW_DOWN')
258			elif stats['status'].startswith('Full'):
259			return unicode_message('CHECK')
260			return ''
261
262			def msg_curse(self, args=None, max_width=None):
263			"""Return the dict to display in the curse interface."""
264			# Init the return message
265			ret = []
266
267			# Only process if stats exist and display plugin enable...
268			if not self.stats or self.is_disabled():
269			return ret
270
271			# Max size for the interface name
272			if max_width:
273			name_max_width = max_width - 12
274			else:
275			# No max_width defined, return an emptu curse message
276			logger.debug("No max_width defined for the {} plugin, it will not be displayed.".format(self.plugin_name))
277			return ret
278
279			# Header
280			msg = '{:{width}}'.format('SENSORS', width=name_max_width)
281			ret.append(self.curse_add_line(msg, "TITLE"))
282
283			# Stats
284			for i in self.stats:
285			# Do not display anything if no battery are detected
286			if i['type'] == SensorType.BATTERY and i['value'] == []:
287			continue
288			# New line
289			ret.append(self.curse_new_line())
290			msg = '{:{width}}'.format(i["label"][:name_max_width], width=name_max_width)
291			ret.append(self.curse_add_line(msg))
292			if i['value'] in (b'ERR', b'SLP', b'UNK', b'NOS'):
293			msg = '{:>14}'.format(i['value'])
294			ret.append(
295			self.curse_add_line(msg, self.get_views(item=i[self.get_key()], key='value', option='decoration'))
296			)
297			else:
298			if args.fahrenheit and i['type'] != SensorType.BATTERY and i['type'] != SensorType.FAN_SPEED:
299			trend = ''
300			value = to_fahrenheit(i['value'])
301			unit = 'F'
302			else:
303			trend = self.battery_trend(i)
304			value = i['value']
305			unit = i['unit']
306			try:
307			msg = '{:.0f}{}{}'.format(value, unit, trend)
308			msg = '{:>14}'.format(msg)
309			ret.append(
310			self.curse_add_line(
311			msg, self.get_views(item=i[self.get_key()], key='value', option='decoration')
312			)
313			)
314			except (TypeError, ValueError):
315			pass
316
317			return ret
318
319
320			class GlancesGrabSensors(object):
321			"""Get sensors stats."""
322
323			def __init__(self, sensor_type: Literal[SensorType.FAN_SPEED, SensorType.CPU_TEMP]):
324			"""Init sensors stats."""
325			self.sensor_type = sensor_type
326			self.sensor_unit = CPU_TEMP_UNIT if self.sensor_type == SensorType.CPU_TEMP else FAN_SPEED_UNIT
327
328			self.init = False
329			try:
330			self.__fetch_psutil()
331			self.init = True
332			except AttributeError:
333			logger.debug(f"Cannot grab {sensor_type}. Platform not supported.")
334
335			def __fetch_psutil(self) -> Dict[str, list]:
336			if self.sensor_type == SensorType.CPU_TEMP:
337			# Solve an issue #1203 concerning a RunTimeError warning message displayed
338			# in the curses interface.
339			warnings.filterwarnings("ignore")
340
341			# psutil>=5.1.0, Linux-only
342			return psutil.sensors_temperatures()
343
344			if self.sensor_type == SensorType.FAN_SPEED:
345			# psutil>=5.2.0, Linux-only
346			return psutil.sensors_fans()
347
348			raise ValueError(f"Unsupported sensor_type: {self.sensor_type}")
349
350			def update(self) -> List[dict]:
351			"""Update the stats."""
352			if not self.init:
353			return []
354
355			# Temperatures sensors
356			ret = []
357			data = self.__fetch_psutil()
358			for chip_name, chip in data.items():
359			label_index = 1
360			for chip_name_index, feature in enumerate(chip):
361			sensors_current = {}
362			# Sensor name
363			if feature.label == '':
364			sensors_current['label'] = chip_name + ' ' + str(chip_name_index)
365			elif feature.label in [i['label'] for i in ret]:
366			sensors_current['label'] = feature.label + ' ' + str(label_index)
367			label_index += 1
368			else:
369			sensors_current['label'] = feature.label
370			# Sensors value, limit and unit
371			sensors_current['unit'] = self.sensor_unit
372			sensors_current['value'] = int(
373			getattr(feature, 'current', 0) if getattr(feature, 'current', 0) else 0)
374			system_warning = getattr(feature, 'high', None)
375			system_critical = getattr(feature, 'critical', None)
376			sensors_current['warning'] = int(system_warning) if system_warning is not None else None
377			sensors_current['critical'] = int(system_critical) if system_critical is not None else None
378			# Add sensor to the list
379			ret.append(sensors_current)
380			return ret
381

nicolargo / glances

Push — master ( eb83f6...fecd23 )

GlancesGrabSensors.__fetch_psutil() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like