ebca1ac1eb5ece03125f1d7b3703547f078a1671
   1#
   2#   temperature.py
   3#   
   4#   Find current temperature of various system components (CPU, motherboard,
   5#   hard drives, ambient). Detection of motherboard-based temperatures (CPU
   6#   etc) uses the pysensors library, and produces a similar output to
   7#   lmsensors. HDD temperatures are obtained from the hddtemp daemon
   8#   (http://www.guzu.net/linux/hddtemp.php) which was orphaned since 2007. For
   9#   hddtemp to work, it must be started in daemon mode, either manually or with
  10#   a unit file. Manually, it would be started like this:
  11#
  12#       sudo hddtemp -d /dev/sda /dev/sdb ... /dev/sdX
  13#
  14
  15import re
  16import sensors
  17import socket, sys
  18from telnetlib import Telnet
  19from typing import List, Dict, NamedTuple
  20
  21from logparse.formatting import *
  22from logparse.util import readlog
  23from logparse import config
  24
  25import logging
  26logger = logging.getLogger(__name__)
  27
  28class Drive(NamedTuple):
  29    path: str
  30    model: str
  31    temperature: int
  32    units: str
  33
  34class HddtempClient:
  35
  36    def __init__(self, host: str='127.0.0.1', port: int=7634, timeout: int=10, sep: str='|') -> None:
  37        self.host = host
  38        self.port = port
  39        self.timeout = timeout
  40        self.sep = sep
  41
  42    def _parse_drive(self, drive: str) -> Drive:
  43        try:
  44            drive_data = drive.split(self.sep)
  45            return Drive(drive_data[0], drive_data[1], int(drive_data[2]), drive_data[3])
  46        except Exception as e:
  47            logger.warning("Error processing drive: {0}".format(str(drive_data)))
  48            return None
  49
  50    def _parse(self, data: str) -> List[Drive]:
  51        line = data.lstrip(self.sep).rstrip(self.sep)   # Remove first/last
  52        drives = line.split(self.sep * 2)
  53        parsed_drives = []
  54        for drive in drives:
  55            parsed_drive = self._parse_drive(drive)
  56            if parsed_drive != None:
  57                parsed_drives.append(parsed_drive)
  58
  59#        return [self._parse_drive(drive) for drive in drives if drive != None]
  60#        return list(filter(lambda drive: self._parse_drive(drive), drives))
  61        return parsed_drives
  62
  63    def get_drives(self) -> List[Drive]:    # Obtain data from telnet server
  64        try:
  65            with Telnet(self.host, self.port, timeout=self.timeout) as tn:
  66                raw_data = tn.read_all()
  67            return self._parse(raw_data.decode('ascii'))    # Return parsed data
  68        except Exception as e:
  69            logger.warning("Couldn't read data from {0}:{1} - {2}".format(self.host, self.port, str(e)))
  70            return 1
  71
  72
  73def parse_log():
  74
  75    logger.debug("Starting temp section")
  76    section = Section("temperatures")
  77
  78    sensors.init()
  79
  80    systemp = Data("Sys", [])
  81    coretemp = Data("Cores", [])
  82    pkgtemp = Data("Processor", [])
  83
  84    try:
  85        for chip in sensors.iter_detected_chips():
  86            for feature in chip:
  87                if "Core" in feature.label:
  88                    coretemp.items.append([feature.label, float(feature.get_value())])
  89                    continue
  90                if "CPUTIN" in feature.label:
  91                    pkgtemp.items.append([feature.label, float(feature.get_value())])
  92                    continue
  93                if "SYS" in feature.label:
  94                    systemp.items.append([feature.label, float(feature.get_value())])
  95                    continue
  96
  97        logger.debug("Core data is {0}".format(str(coretemp.items)))
  98        logger.debug("Sys data is {0}".format(str(systemp.items)))
  99        logger.debug("Pkg data is {0}".format(str(pkgtemp.items)))
 100        for temp_data in [systemp, coretemp, pkgtemp]:
 101            logger.debug("Looking at temp data {0}".format(str(temp_data.items)))
 102            if len(temp_data.items) > 1:
 103                avg = float(sum(feature[1] for feature in temp_data.items)) / len(temp_data.items)
 104                logger.debug("Avg temp for {0} is {1} {2}{3}".format(temp_data.subtitle, str(avg), DEG, CEL))
 105                temp_data.subtitle += " (avg {0}{1}{2})".format(str(avg), DEG, CEL)
 106                temp_data.items = ["{0}: {1}{2}{3}".format(feature[0], str(feature[1]), DEG, CEL) for feature in temp_data.items]
 107            else:
 108                temp_data.items = [str(temp_data.items[0][1]) + DEG + CEL]
 109            section.append_data(temp_data)
 110
 111    finally:
 112        logger.debug("Finished reading onboard temperatures")
 113        sensors.cleanup()
 114
 115
 116    # drive temp
 117
 118    # For this to work, `hddtemp` must be running in daemon mode.
 119    # Start it like this (bash):   sudo hddtemp -d /dev/sda /dev/sdX...
 120    
 121    received = ''
 122    sumtemp = 0.0 
 123    data = ""
 124    hddtemp_data = Data("Disks")
 125    
 126    client = HddtempClient(
 127        host=config.prefs.get("temperatures", "host"),
 128        port=config.prefs.getint("temperatures", "port"),
 129        sep=config.prefs.get("temperatures", "separator"),
 130        timeout=int(config.prefs.get("temperatures", "timeout")))
 131    drives = client.get_drives()
 132    logger.debug("Received drive info: " + str(drives))
 133
 134    for drive in sorted(drives, key=lambda x: x.path):
 135        if drive.path in config.prefs.get("temperatures", "drives").split():
 136            sumtemp += drive.temperature
 137            hddtemp_data.items.append(("{0} ({1})".format(drive.path, drive.model) if config.prefs.getboolean("temperatures", "show-model") else drive.path) + ": {0}{1}{2}".format(drive.temperature, DEG, drive.units))
 138        else:
 139            drives.remove(drive)
 140            logger.debug("Ignoring drive {0} ({1}) due to config".format(drive.path, drive.model))
 141    logger.debug("Sorted drive info: " + str(drives))
 142
 143    hddavg = '{0:.1f}{1}{2}'.format(sumtemp/len(drives), DEG, drives[0].units) # use units of first drive
 144    logger.debug("Sum of temperatures: {}; Number of drives: {}; => Avg disk temp is {}".format(str(sumtemp), str(len(drives)), hddavg)) 
 145    hddtemp_data.subtitle += " (avg {0}{1}{2})".format(str(hddavg), DEG, CEL)
 146
 147    logger.debug("Finished processing drive temperatures")
 148    logger.info("Finished temp section")
 149
 150    return section