glances.plugins.sensors.PluginModel.update_views() - Code Metrics - Inspection of "version 4.3.0" - nicolargo/glances - Measure and Improve Code Quality continuously with Scrutinizer

Test Failed

Push — master ( ce0fc3...e09530 )

by Nicolas

created 2024-12-27 14:16 UTC

glances.plugins.sensors.PluginModel.update_views() B

↳ Parent: glances.plugins.sensors

Complexity

Conditions

Size

Total Lines	30
Code Lines	17

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	7
eloc	17
nop	1
dl	0
loc	30
rs	8
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 enum import Enum
from typing import Any, Literal

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


class SensorType(str, Enum):
    # Switch to `enum.StrEnum` when we only support py311+
    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().__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(f"CPU Temp sensor plugin init duration: {start_duration.get()} seconds")

        start_duration.reset()
        glances_grab_sensors_fan_speed = GlancesGrabSensors(SensorType.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: 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

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

nicolargo / glances

Push — master ( ce0fc3...e09530 )

glances.plugins.sensors.PluginModel.update_views() B

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like