Skip to content

Commit

Permalink
Merge pull request #2768 from nicolargo/2765-sensors-plugin-is-buggy
Browse files Browse the repository at this point in the history 
refactor: plugin(sensors) - cleanup + typing + fixes
  • Loading branch information
@RazCrimson
RazCrimson committed May 14, 2024
2 parents 127f2e4 + 7a25f71 commit 3aa9225
Showing 1 changed file with 82 additions and 139 deletions.
221 changes: 82 additions & 139 deletions glances/plugins/sensors/__init__.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -8,30 +8,33 @@
#

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

import psutil
import warnings
import threading

from glances.logger import logger
from glances.globals import iteritems, to_fahrenheit
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

SENSOR_TEMP_TYPE = 'temperature_core'
SENSOR_TEMP_UNIT = 'C'

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

SENSOR_HDDTEMP_TYPE = 'temperature_hdd'
SENSOR_HDDTEMP_UNIT = 'C'

SENSORS_BATTERY_TYPE = 'battery'
SENSORS_BATTERY_UNIT = '%'
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
Expand Down Expand Up @@ -82,28 +85,38 @@ def __init__(self, args=None, config=None):
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()
# Hotfix! Refactor to use only one `GlancesGrabSensors` later
self.glances_grab_sensors_fan_speed = GlancesGrabSensors()
self.glances_grab_sensors_temperature = GlancesGrabSensors()
logger.debug("Generic sensor plugin init duration: {} seconds".format(start_duration.get()))
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
# Instance for the HDDTemp Plugin in order to display the hard disks temperatures
start_duration.reset()
self.hddtemp_plugin = HddTempPluginModel(args=args, config=config)
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
# Instance for the BatPercent in order to display the batteries capacities
start_duration.reset()
self.batpercent_plugin = BatPercentPluginModel(args=args, config=config)
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

Expand All @@ -115,37 +128,15 @@ def get_key(self):
"""Return the key of the list."""
return 'label'

def __get_temperature(self, stats, index):
try:
temperature = self.__set_type(self.glances_grab_sensors_temperature.get(SENSOR_TEMP_TYPE), SENSOR_TEMP_TYPE)
except Exception as e:
logger.error("Cannot grab sensors temperatures (%s)" % e)
else:
stats[index] = self.__transform_sensors(temperature)

def __get_fan_speed(self, stats, index):
def __get_sensor_data(self, sensor_type: SensorType) -> List[Dict]:
try:
fan_speed = self.__set_type(self.glances_grab_sensors_fan_speed.get(SENSOR_FAN_TYPE), SENSOR_FAN_TYPE)
data = self.sensors_grab_map[sensor_type].update()
data = self.__set_type(data, sensor_type)
except Exception as e:
logger.error("Cannot grab FAN speed (%s)" % e)
logger.error(f"Cannot grab sensors `{sensor_type}` ({e})")
return []
else:
stats[index] = self.__transform_sensors(fan_speed)

def __get_hddtemp(self, stats, index):
try:
hddtemp = self.__set_type(self.hddtemp_plugin.update(), SENSOR_HDDTEMP_TYPE)
except Exception as e:
logger.error("Cannot grab HDD temperature (%s)" % e)
else:
stats[index] = self.__transform_sensors(hddtemp)

def __get_bat_percent(self, stats, index):
try:
bat_percent = self.__set_type(self.batpercent_plugin.update(), SENSORS_BATTERY_TYPE)
except Exception as e:
logger.error("Cannot grab battery percent (%s)" % e)
else:
stats[index] = self.__transform_sensors(bat_percent)
return self.__transform_sensors(data)

def __transform_sensors(self, threads_stats):
"""Hide, alias and sort the result"""
Expand All @@ -172,22 +163,17 @@ def update(self):
stats = self.get_init_value()

if self.input_method == 'local':
threads_stats = [None] * 4
threads = [
threading.Thread(name=SENSOR_TEMP_TYPE, target=self.__get_temperature, args=(threads_stats, 0)),
threading.Thread(name=SENSOR_FAN_TYPE, target=self.__get_fan_speed, args=(threads_stats, 1)),
threading.Thread(name=SENSOR_HDDTEMP_TYPE, target=self.__get_hddtemp, args=(threads_stats, 2)),
threading.Thread(name=SENSORS_BATTERY_TYPE, target=self.__get_bat_percent, args=(threads_stats, 3)),
]
# Start threads in //
for t in threads:
t.start()
# Wait threads are finished
for t in threads:
t.join()
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 s in threads_stats:
stats.extend(s)
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:
Expand Down Expand Up @@ -220,7 +206,7 @@ def __set_type(self, stats, sensor_type):
"""
for i in stats:
# Set the sensors type
i.update({'type': sensor_type})
i.update({'type': str(sensor_type)})
# also add the key name
i.update({'key': self.get_key()})

Expand All @@ -237,10 +223,10 @@ def update_views(self):
if not i['value']:
continue
# Alert processing
if i['type'] == SENSOR_TEMP_TYPE:
if self.is_limit('critical', stat_name=SENSOR_TEMP_TYPE + '_' + i['label']):
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=SENSOR_TEMP_TYPE + '_' + i['label'])
alert = self.get_alert(current=i['value'], header=SensorType.CPU_TEMP + '_' + i['label'])
else:
# Else use the system thresholds
if i['critical'] is None:
Expand All @@ -253,7 +239,7 @@ def update_views(self):
alert = 'WARNING'
else:
alert = 'OK'
elif i['type'] == SENSORS_BATTERY_TYPE:
elif i['type'] == SensorType.BATTERY:
# Battery is in %
alert = self.get_alert(current=100 - i['value'], header=i['type'])
else:
Expand Down Expand Up @@ -297,7 +283,7 @@ def msg_curse(self, args=None, max_width=None):
# Stats
for i in self.stats:
# Do not display anything if no battery are detected
if i['type'] == SENSORS_BATTERY_TYPE and i['value'] == []:
if i['type'] == SensorType.BATTERY and i['value'] == []:
continue
# New line
ret.append(self.curse_new_line())
Expand All @@ -309,7 +295,7 @@ def msg_curse(self, args=None, max_width=None):
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_BATTERY_TYPE and i['type'] != SENSOR_FAN_TYPE:
if args.fahrenheit and i['type'] != SensorType.BATTERY and i['type'] != SensorType.FAN_SPEED:
trend = ''
value = to_fahrenheit(i['value'])
unit = 'F'
Expand All @@ -334,73 +320,42 @@ def msg_curse(self, args=None, max_width=None):
class GlancesGrabSensors(object):
"""Get sensors stats."""

def __init__(self):
def __init__(self, sensor_type: Literal[SensorType.FAN_SPEED, SensorType.CPU_TEMP]):
"""Init sensors stats."""
# Temperatures
self.init_temp = False
self.sensor_temps = {}
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:
# psutil>=5.1.0, Linux-only
self.sensor_temps = psutil.sensors_temperatures()
self.__fetch_psutil()
self.init = True
except AttributeError:
logger.debug("Cannot grab temperatures. Platform not supported.")
else:
self.init_temp = True
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")

# Fans
self.init_fan = False
self.sensor_fans = {}
try:
# psutil>=5.2.0, Linux-only
self.sensor_fans = psutil.sensors_fans()
except AttributeError:
logger.debug("Cannot grab fans speed. Platform not supported.")
else:
self.init_fan = True
# psutil>=5.1.0, Linux-only
return psutil.sensors_temperatures()

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

def reset(self):
"""Reset/init the stats."""
self.sensors_list = []
raise ValueError(f"Unsupported sensor_type: {self.sensor_type}")

def __update__(self):
def update(self) -> list[dict]:
"""Update the stats."""
# Reset the list
self.reset()

if not self.init_temp:
return self.sensors_list
if not self.init:
return []

# Temperatures sensors
self.sensors_list.extend(self.build_sensors_list(SENSOR_TEMP_UNIT))

# Fans sensors
self.sensors_list.extend(self.build_sensors_list(SENSOR_FAN_UNIT))

return self.sensors_list

def build_sensors_list(self, type):
"""Build the sensors list depending of the type.
type: SENSOR_TEMP_UNIT or SENSOR_FAN_UNIT
output: a list
"""
ret = []
if type == SENSOR_TEMP_UNIT and self.init_temp:
input_list = self.sensor_temps
self.sensor_temps = psutil.sensors_temperatures()
elif type == SENSOR_FAN_UNIT and self.init_fan:
input_list = self.sensor_fans
self.sensor_fans = psutil.sensors_fans()
else:
return ret
for chip_name, chip in iteritems(input_list):
data = self.__fetch_psutil()
for chip_name, chip in data.items():
label_index = 1
for chip_name_index, feature in enumerate(chip):
sensors_current = {}
Expand All @@ -413,25 +368,13 @@ def build_sensors_list(self, type):
else:
sensors_current['label'] = feature.label
# Sensors value, limit and unit
sensors_current['unit'] = type
sensors_current['value'] = int(getattr(feature, 'current', 0) if getattr(feature, 'current', 0) else 0)
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

def get(self, sensor_type=SENSOR_TEMP_TYPE):
"""Get sensors list."""
self.__update__()
if sensor_type == SENSOR_TEMP_TYPE:
ret = [s for s in self.sensors_list if s['unit'] == SENSOR_TEMP_UNIT]
elif sensor_type == SENSOR_FAN_TYPE:
ret = [s for s in self.sensors_list if s['unit'] == SENSOR_FAN_UNIT]
else:
# Unknown type
logger.debug("Unknown sensor type %s" % sensor_type)
ret = []
return ret

0 comments on commit 3aa9225

Please sign in to comment.