|
21 | 21 | # This file will use Python libraries (psutil, GPUtil, etc.) to get hardware sensors |
22 | 22 | # For all platforms (Linux, Windows, macOS) but not all HW is supported |
23 | 23 |
|
| 24 | +import glob |
24 | 25 | import math |
| 26 | +import os |
25 | 27 | import platform |
| 28 | +import re |
26 | 29 | import sys |
| 30 | +import time |
27 | 31 | from collections import namedtuple |
28 | 32 | from enum import IntEnum, auto |
29 | 33 | from typing import Tuple |
@@ -119,6 +123,93 @@ def is_cpu_fan(label: str) -> bool: |
119 | 123 | return ("cpu" in label.lower()) or ("proc" in label.lower()) |
120 | 124 |
|
121 | 125 |
|
| 126 | +_disk_temp_paths = None |
| 127 | +_disk_temp_last = math.nan |
| 128 | + |
| 129 | + |
| 130 | +def _find_disk_temp_paths(): |
| 131 | + """hwmon temp files for the drive backing "/". |
| 132 | +
|
| 133 | + Only that drive's sensors are returned when it has any. Falling back to |
| 134 | + another drive would silently report the wrong disk's temperature, so the |
| 135 | + system-wide scan is used only when the root drive exposes no sensor at all. |
| 136 | + """ |
| 137 | + try: |
| 138 | + root_device = None |
| 139 | + for part in psutil.disk_partitions(all=False): |
| 140 | + if part.mountpoint == "/": |
| 141 | + root_device = part.device |
| 142 | + break |
| 143 | + |
| 144 | + # /dev/sde1 -> sde ; /dev/nvme0n1p2 -> nvme0n1 |
| 145 | + base = None |
| 146 | + if root_device: |
| 147 | + name = os.path.basename(root_device) |
| 148 | + m = re.match(r"^(nvme\d+n\d+)p\d+$", name) or re.match(r"^([a-zA-Z]+)\d*$", name) |
| 149 | + if m: |
| 150 | + base = m.group(1) |
| 151 | + |
| 152 | + if base: |
| 153 | + own = [os.path.join(h, "temp1_input") |
| 154 | + for h in sorted(glob.glob(f"/sys/block/{base}/device/hwmon/hwmon*"))] |
| 155 | + own = [c for c in own if os.path.exists(c)] |
| 156 | + if own: |
| 157 | + return own |
| 158 | + |
| 159 | + # Root drive has no sensor: fall back to any drive on the system. |
| 160 | + other = [] |
| 161 | + for hwmon in sorted(glob.glob("/sys/class/hwmon/hwmon*")): |
| 162 | + try: |
| 163 | + with open(os.path.join(hwmon, "name")) as f: |
| 164 | + if f.read().strip() not in ("drivetemp", "nvme"): |
| 165 | + continue |
| 166 | + except OSError: |
| 167 | + continue |
| 168 | + candidate = os.path.join(hwmon, "temp1_input") |
| 169 | + if os.path.exists(candidate): |
| 170 | + other.append(candidate) |
| 171 | + return other |
| 172 | + except Exception: |
| 173 | + return [] |
| 174 | + |
| 175 | + |
| 176 | +def _disk_temperature() -> float: |
| 177 | + """Temperature (°C) of the drive backing "/". |
| 178 | +
|
| 179 | + SATA drives need the `drivetemp` kernel module; NVMe drives expose this |
| 180 | + natively. Some SATA SSDs (e.g. Samsung 870 EVO) refuse the underlying SMART |
| 181 | + query while busy and return EIO on most reads, so retry briefly and fall |
| 182 | + back to this drive's last good value rather than reporting nothing. |
| 183 | + """ |
| 184 | + global _disk_temp_paths, _disk_temp_last |
| 185 | + |
| 186 | + if _disk_temp_paths is None: |
| 187 | + _disk_temp_paths = _find_disk_temp_paths() |
| 188 | + |
| 189 | + # Do NOT retry rapidly. These drives serialise SMART queries poorly: a burst |
| 190 | + # of reads contends with itself and drives the success rate down (measured |
| 191 | + # 8/9 with a single read per cycle, 1/9 while a second reader was polling). |
| 192 | + # One read per cycle plus the cached fallback is far more reliable. |
| 193 | + # The only exception is a cold start, where there is no cache to fall back |
| 194 | + # on yet, and even then attempts are spaced a full second apart. |
| 195 | + attempts = 1 if not math.isnan(_disk_temp_last) else 3 |
| 196 | + |
| 197 | + for attempt in range(attempts): |
| 198 | + for path in _disk_temp_paths: |
| 199 | + try: |
| 200 | + with open(path) as f: |
| 201 | + value = int(f.read().strip()) / 1000.0 |
| 202 | + _disk_temp_last = value |
| 203 | + return value |
| 204 | + except (OSError, ValueError): |
| 205 | + continue |
| 206 | + if attempt < attempts - 1: |
| 207 | + time.sleep(1.0) |
| 208 | + |
| 209 | + # Every read failed this cycle: reuse this drive's last good value. |
| 210 | + return _disk_temp_last |
| 211 | + |
| 212 | + |
122 | 213 | class Cpu(sensors.Cpu): |
123 | 214 | @staticmethod |
124 | 215 | def percentage(interval: float) -> float: |
@@ -473,6 +564,10 @@ def disk_free() -> int: # In bytes |
473 | 564 | except: |
474 | 565 | return -1 |
475 | 566 |
|
| 567 | + @staticmethod |
| 568 | + def disk_temperature() -> float: # In °C |
| 569 | + return _disk_temperature() |
| 570 | + |
476 | 571 |
|
477 | 572 | class Net(sensors.Net): |
478 | 573 | @staticmethod |
|
0 commit comments