glances.plugins.sensors.GlancesGrabSensors.__init__() - Code Metrics - nicolargo/glances - Measure and Improve Code Quality continuously with Scrutinizer

GlancesGrabSensors.init() A
last analyzed 2025-03-22 16:55 UTC

↳ Parent: glances.plugins.sensors

Complexity

Conditions

Size

Total Lines	11
Code Lines	9

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	2
eloc	9
nop	2
dl	0
loc	11
rs	9.95
c	0
b	0
f	0

#
# This file is part of Glances.
#
# SPDX-FileCopyrightText: 2024 Nicolas Hennion <[email protected]>
#
# SPDX-License-Identifier: LGPL-3.0-only
#

"""Sensors plugin."""

import warnings
from concurrent.futures import ThreadPoolExecutor
from typing import Any

import psutil

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

# Define all kind of sensors available in Glances
sensors_definition = {
    'cpu_temp': {'type': 'temperature_core', 'unit': 'C'},
    'fan_speed': {'type': 'fan_speed', 'unit': 'R'},
    'hdd_temp': {'type': 'temperature_hdd', 'unit': 'C'},
    'battery': {'type': '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().__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(sensors_definition.get('cpu_temp'))
        logger.debug(f"CPU Temp sensor plugin init duration: {start_duration.get()} seconds")

        start_duration.reset()
        glances_grab_sensors_fan_speed = GlancesGrabSensors(sensors_definition.get('fan_speed'))
        logger.debug(f"Fan speed sensor plugin init duration: {start_duration.get()} seconds")

        # 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(f"HDDTemp sensor plugin init duration: {start_duration.get()} seconds")

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

        self.sensors_grab_map = {}

        if glances_grab_sensors_cpu_temp.init:
            self.sensors_grab_map[sensors_definition.get('cpu_temp').get('type')] = glances_grab_sensors_cpu_temp

        if glances_grab_sensors_fan_speed.init:
            self.sensors_grab_map[sensors_definition.get('fan_speed').get('type')] = glances_grab_sensors_fan_speed

        self.sensors_grab_map[sensors_definition.get('hdd_temp').get('type')] = hddtemp_plugin
        self.sensors_grab_map[sensors_definition.get('battery').get('type')] = 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: str) -> 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
        return sorted(stats_transformed, key=lambda d: d['label'])

    @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())
        if self.has_alias("{}_{}".format(stats["label"], stats["type"]).lower()):
            return self.has_alias("{}_{}".format(stats["label"], stats["type"]).lower())
        return stats["label"]

    def __set_type(self, stats: list[dict[str, Any]], sensor_type: str) -> list[dict[str, Any]]:
        """Set the plugin type.

        4 types of stats is possible in the sensors plugin, see sensors_definition variable
        """
        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 __get_system_thresholds(self, sensor):
        """Return the alert level thanks to the system thresholds.
        Note: Only Warning (aka High) and Critical thresholds are used.
        """
        alert = 'OK'
        if sensor['critical'] is None:
            alert = 'DEFAULT'
        elif sensor['value'] >= sensor['critical']:
            alert = 'CRITICAL'
        elif sensor['warning'] is None:
            alert = 'DEFAULT'
        elif sensor['value'] >= sensor['warning']:
            alert = 'WARNING'
        return alert

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

        # Add specifics information
        # Alert
        for i in self.stats:
            if not i['value']:
                continue
            # Alert processing
            if i['type'] == sensors_definition.get('cpu_temp').get('type'):
                if self.is_limit('critical', stat_name=i['type'] + '_' + i['label']):
                    # Get thresholds for the specific sensor in the glances.conf file (see #2058)
                    alert = self.get_alert(current=i['value'], header=i['type'] + '_' + i['label'])
                elif self.is_limit('critical', stat_name=i['type']):
                    # Get thresholds for the sensor type in the glances.conf file (see #3049)
                    alert = self.get_alert(current=i['value'], header=i['type'])
                else:
                    # Else use the system thresholds
                    alert = self.__get_system_thresholds(i)
            elif i['type'] == sensors_definition.get('battery').get('type'):
                # 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')
        if stats['status'].startswith('Discharg'):
            return unicode_message('ARROW_DOWN')
        if 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 empty curse message
            logger.debug(f"No max_width defined for the {self.plugin_name} plugin, it will not be displayed.")
            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'] == sensors_definition.get('battery').get('type') 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'] != sensors_definition.get('battery').get('type')
                    and i['type'] != sensors_definition.get('fan_speed').get('type')
                ):
                    trend = ''
                    value = to_fahrenheit(i['value'])
                    unit = 'F'
                else:
                    trend = self.battery_trend(i)
                    value = i['value']
                    unit = i['unit']
                try:
                    msg = f'{value:.0f}{unit}{trend}'
                    msg = f'{msg:>14}'
                    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:
    """Get sensors stats."""

    def __init__(self, sensor_def: dict):
        """Init sensors stats."""
        self.sensor_type = sensor_def.get('type')
        self.sensor_unit = sensor_def.get('unit')

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

    def __fetch_data(self) -> dict[str, list]:
        if self.sensor_type == sensors_definition.get('cpu_temp').get('type'):
            # 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 == sensors_definition.get('fan_speed').get('type'):
            # psutil>=5.2.0, Linux-only
            return psutil.sensors_fans()

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

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

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

nicolargo / glances

GlancesGrabSensors.__init__() A last analyzed 2025-03-22 16:55 UTC

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like

GlancesGrabSensors.init() A
last analyzed 2025-03-22 16:55 UTC