#!/usr/bin/env python3
# $Id: hdd_led.py,v 1.4 2026/10/14 21:21:28 jdeifik Exp $
# Copyright Jeff turbo Deifik 2026 All rights reserved.
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, version 2 of the License.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see .
"""
hdd_led - per-volume disk activity indicators for the Windows desktop.
* One small always-on-top "LED" per disk volume (C:, D:, ...).
* Green when idle, red when the volume is doing I/O.
* Hover an LED to see read speed, write speed, IOPS, total bytes read and
written (4 significant digits), label, free space and the programs that are
(most likely) responsible for the activity.
* Ctrl-C in the console window (or a normal exit) saves settings and quits
cleanly.
* Network drives (SMB) are measured too; drives that cannot be measured get a
grey LED, and right-click > Diagnostics shows what Windows reports.
* Drag LEDs anywhere; positions are remembered.
* Right-click an LED for settings, start-with-Windows, hide volume, exit.
Requirements: Windows 10/11, Python 3.12+ (standard library only).
Data source: Windows Performance Data Helper (PDH) "LogicalDisk" counters,
a documented, stable interface (queried with the English counter names, so
it works on any display language).
Run: pythonw hdd_led.py
Build: pip install pyinstaller
pyinstaller --noconsole --onefile hdd_led.py
"""
from __future__ import annotations
import ctypes
import json
import os
import re
import shutil
import signal
import struct
import threading
from collections import deque
import string
import sys
import time
import tkinter as tk
from datetime import datetime
from ctypes import byref, c_double, c_size_t, c_ulong, c_void_p, c_wchar_p, POINTER, Structure
from dataclasses import asdict, dataclass, field, fields
from pathlib import Path
from tkinter import colorchooser, messagebox, ttk
APP_NAME = "hdd_led"
TRANSPARENT = "#010203" # colour keyed out so the LED can be round
RESCAN_MS = 2000 # how often to look for added/removed volumes
ERROR_ALREADY_EXISTS = 183
RW_DOMINANCE = 0.10 # a direction below 10% of the other counts as background
# --------------------------------------------------------------------------
# Configuration
# --------------------------------------------------------------------------
def config_path() -> Path:
base = Path(os.environ.get("APPDATA", Path.home()))
folder = base / APP_NAME
folder.mkdir(parents=True, exist_ok=True)
return folder / "config.json"
LAST_ERROR = ""
def log_error(text: str) -> None:
"""Append an error to %APPDATA%\\\\error.log (there is no console under pythonw)."""
global LAST_ERROR
lines = [ln for ln in text.strip().splitlines() if ln.strip()]
LAST_ERROR = lines[-1].strip() if lines else ""
try:
path = config_path().with_name("error.log")
if path.exists() and path.stat().st_size > 200_000:
path.write_text("", encoding="utf-8")
with open(path, "a", encoding="utf-8") as fh:
fh.write(f"[{datetime.now():%Y-%m-%d %H:%M:%S}] {text.rstrip()}\n\n")
except OSError:
pass
@dataclass
class Config:
size: int = 28 # LED diameter in pixels
opacity: float = 0.90 # 0.2 .. 1.0
idle_color: str = "#22c55e" # green
active_color: str = "#ef4444" # red
read_color: str = "#f59e0b" # used when separate_rw is on
write_color: str = "#3b82f6" # blue; used when separate_rw is on
separate_rw: bool = False # different colour for read-only / write-only
shape: str = "circle" # "circle" or "square"
show_letters: bool = True
always_on_top: bool = True
locked: bool = False # disable dragging
remember_positions: bool = True # restore LED positions after restart
poll_ms: int = 150 # sampling interval
hold_ms: int = 80 # keep "active" colour at least this long
threshold_kbps: float = 0.0 # ignore I/O below this rate (KB/s)
use_physical_disk: bool = True # also count raw activity of the disk behind a lone volume
show_unavailable: bool = True # grey LED for drives that cannot be measured
unavailable_color: str = "#9ca3af"
show_totals: bool = True # total bytes read/written in the tooltip
show_processes: bool = True # list responsible programs in the tooltip
process_count: int = 3 # max programs listed per drive
hidden: list[str] = field(default_factory=list)
positions: dict[str, list[int]] = field(default_factory=dict)
@classmethod
def load(cls) -> "Config":
cfg = cls()
try:
data = json.loads(config_path().read_text(encoding="utf-8"))
known = {f.name for f in fields(cls)}
for key, value in data.items():
if key in known:
setattr(cfg, key, value)
except (OSError, ValueError):
pass
return cfg
def save(self) -> None:
try:
config_path().write_text(json.dumps(asdict(self), indent=2), encoding="utf-8")
except OSError:
pass
# --------------------------------------------------------------------------
# Helpers
# --------------------------------------------------------------------------
def format_rate(bytes_per_sec: float) -> str:
value = float(bytes_per_sec)
for unit in ("B/s", "KB/s", "MB/s", "GB/s"):
if value < 1024 or unit == "GB/s":
return f"{value:,.0f} {unit}" if unit == "B/s" else f"{value:,.1f} {unit}"
value /= 1024
return f"{value:.1f} GB/s"
def format_size(num: float) -> str:
value = float(num)
for unit in ("B", "KB", "MB", "GB", "TB"):
if value < 1024 or unit == "TB":
return f"{value:,.1f} {unit}" if unit != "B" else f"{value:,.0f} B"
value /= 1024
return f"{value:.1f} TB"
def format_total(num_bytes: float) -> str:
"""Byte count with exactly 4 significant digits, e.g. 1.234 GB, 56.78 MB, 123.4 KB."""
value = float(num_bytes)
units = ("B", "KB", "MB", "GB", "TB", "PB")
i = 0
while value >= 1024 and i < len(units) - 1:
value /= 1024
i += 1
if i == 0:
return f"{value:.0f} B" # whole bytes (at most 4 digits here)
digits = len(str(int(value))) if value >= 1 else 1
decimals = max(0, 4 - digits)
text = f"{value:.{decimals}f}"
if len(text.split(".")[0]) > digits: # rounding carried over (9.9996 -> 10.000)
decimals = max(0, decimals - 1)
text = f"{value:.{decimals}f}"
return f"{text} {units[i]}"
def console(message: str) -> None:
"""Print to the console if there is one (there is none under pythonw)."""
stream = sys.stderr or sys.stdout
if stream is not None:
try:
print(message, file=stream, flush=True)
except (OSError, ValueError):
pass
EXIT_REQUESTED = threading.Event() # set by Ctrl-C / Ctrl-Break / terminate / console close
FINISHED = threading.Event() # set when the program has finished shutting down
_APP: "App | None" = None
def request_exit(reason: str) -> None:
"""Ask the program to exit. Safe to call from any thread.
The GUI thread normally does the shutdown. If it does not finish within a few
seconds (for example because it is stuck in a call to an unresponsive drive),
a watchdog saves the settings and ends the process anyway. A second request
ends it immediately."""
if EXIT_REQUESTED.is_set():
console(f"\n{APP_NAME}: forcing exit now.")
_save_config_quietly()
os._exit(130)
EXIT_REQUESTED.set()
console(f"\n{APP_NAME}: {reason} received - saving settings and exiting cleanly...")
threading.Thread(target=_watchdog, args=(3.0,), daemon=True, name="exit-watchdog").start()
def _save_config_quietly() -> None:
try:
if _APP is not None:
_APP.cfg.save()
except Exception:
pass
def _watchdog(seconds: float) -> None:
if not FINISHED.wait(seconds):
console(f"{APP_NAME}: the program did not respond in time - forcing exit.")
_save_config_quietly()
os._exit(130)
def install_console_handler():
"""Handle Ctrl-C, Ctrl-Break and console-close on their own thread, so that they
work even when the GUI thread is busy or blocked. Returns the callback, which the
caller must keep referenced."""
kernel32 = ctypes.windll.kernel32
handler_type = ctypes.WINFUNCTYPE(ctypes.c_int, ctypes.c_ulong)
names = {0: "Ctrl-C", 1: "Ctrl-Break", 2: "console window close", 5: "logoff", 6: "shutdown"}
def handler(ctrl_type: int) -> int:
if ctrl_type not in names:
return 0
request_exit(names[ctrl_type])
if ctrl_type >= 2: # Windows ends the process when we return
FINISHED.wait(4.5)
return 1
callback = handler_type(handler)
kernel32.SetConsoleCtrlHandler.argtypes = [handler_type, ctypes.c_int]
kernel32.SetConsoleCtrlHandler(callback, 1)
return callback
def text_color_for(bg: str) -> str:
try:
r, g, b = (int(bg[i:i + 2], 16) for i in (1, 3, 5))
except (ValueError, IndexError):
return "#000000"
return "#000000" if (0.299 * r + 0.587 * g + 0.114 * b) > 150 else "#ffffff"
DRIVE_TYPE_NAMES = {0: "unknown", 1: "no root", 2: "removable", 3: "fixed",
4: "network", 5: "optical", 6: "RAM disk"}
def drive_type(vol: str) -> int:
return ctypes.windll.kernel32.GetDriveTypeW(f"{vol}\\")
def has_media(vol: str) -> bool:
"""True if the volume is mounted/readable (False for an empty card reader or DVD drive)."""
return bool(ctypes.windll.kernel32.GetVolumeInformationW(
f"{vol}\\", None, 0, None, None, None, None, 0))
def unc_for_drive(vol: str) -> str:
"""\\\\server\\share behind a mapped network drive letter, or '' if none."""
try:
buf = ctypes.create_unicode_buffer(1024)
size = ctypes.c_ulong(1024)
if ctypes.windll.mpr.WNetGetConnectionW(vol, buf, byref(size)) == 0:
return buf.value
except (OSError, AttributeError):
pass
return ""
def volume_label(vol: str) -> str:
if drive_type(vol) == 4: # avoid slow calls on network drives
return unc_for_drive(vol)
buf = ctypes.create_unicode_buffer(261)
ok = ctypes.windll.kernel32.GetVolumeInformationW(
f"{vol}\\", buf, 261, None, None, None, None, 0)
return buf.value if ok else ""
def volume_disk_numbers(vol: str) -> list[int]:
"""Physical disk number(s) that hold a volume (IOCTL_VOLUME_GET_VOLUME_DISK_EXTENTS)."""
try:
k32 = ctypes.WinDLL("kernel32", use_last_error=True)
k32.CreateFileW.restype = ctypes.c_ssize_t
k32.CreateFileW.argtypes = [c_wchar_p, c_ulong, c_ulong, c_void_p, c_ulong, c_ulong, c_void_p]
k32.DeviceIoControl.argtypes = [ctypes.c_ssize_t, c_ulong, c_void_p, c_ulong,
c_void_p, c_ulong, POINTER(c_ulong), c_void_p]
k32.CloseHandle.argtypes = [ctypes.c_ssize_t]
handle = k32.CreateFileW(f"\\\\.\\{vol}", 0, 3, None, 3, 0, None) # open, no access rights
if handle in (-1, 0):
return []
try:
buf = ctypes.create_string_buffer(4096)
returned = c_ulong(0)
if not k32.DeviceIoControl(handle, 0x560000, None, 0, buf, 4096, byref(returned), None):
return []
count = struct.unpack_from(" str | None:
"""Find a reasonably large file (>= 1 MB, preferably >= 64 MB) on a volume."""
skip = {"$recycle.bin", "system volume information", "windows", "$windows.~bt",
"recovery", "programdata", "appdata", "winsxs"}
deadline = time.monotonic() + seconds
queue = deque([vol + "\\"])
best, best_size = None, 0
while queue and time.monotonic() < deadline:
folder = queue.popleft()
try:
with os.scandir(folder) as entries:
for entry in entries:
try:
if entry.is_dir(follow_symlinks=False):
if entry.name.lower() not in skip:
queue.append(entry.path)
elif entry.is_file(follow_symlinks=False):
size = entry.stat(follow_symlinks=False).st_size
if size > best_size and size >= 1_000_000:
best, best_size = entry.path, size
if size >= 64 * 1024 * 1024:
return best
except OSError:
continue
except OSError:
continue
return best
def read_test(path: str, limit: int = 100 * 1024 * 1024) -> tuple[int, float, str]:
"""Read up to `limit` bytes of a file bypassing the Windows file cache.
Returns (bytes read, seconds, error text)."""
k32 = ctypes.WinDLL("kernel32", use_last_error=True)
k32.CreateFileW.restype = ctypes.c_ssize_t
k32.CreateFileW.argtypes = [c_wchar_p, c_ulong, c_ulong, c_void_p, c_ulong, c_ulong, c_void_p]
k32.VirtualAlloc.restype = c_void_p
k32.VirtualAlloc.argtypes = [c_void_p, c_size_t, c_ulong, c_ulong]
k32.VirtualFree.argtypes = [c_void_p, c_size_t, c_ulong]
k32.ReadFile.argtypes = [ctypes.c_ssize_t, c_void_p, c_ulong, POINTER(c_ulong), c_void_p]
k32.CloseHandle.argtypes = [ctypes.c_ssize_t]
chunk = 1024 * 1024 # multiple of any sector size
# GENERIC_READ, share all, OPEN_EXISTING, NO_BUFFERING | SEQUENTIAL_SCAN
handle = k32.CreateFileW(path, 0x80000000, 7, None, 3, 0x20000000 | 0x08000000, None)
if handle in (-1, 0):
return 0, 0.0, f"cannot open file (Windows error {ctypes.get_last_error()})"
buf = k32.VirtualAlloc(None, chunk, 0x3000, 0x04) # page-aligned buffer
if not buf:
k32.CloseHandle(handle)
return 0, 0.0, "cannot allocate read buffer"
total, started = 0, time.monotonic()
try:
got = c_ulong(0)
while total < limit:
if not k32.ReadFile(handle, buf, chunk, byref(got), None) or got.value == 0:
break
total += got.value
finally:
k32.VirtualFree(buf, 0, 0x8000)
k32.CloseHandle(handle)
return total, time.monotonic() - started, ""
class VolumeCatalog(threading.Thread):
"""Finds the drives and their details (label, physical disk, free space) on a
background thread.
Several of these Windows calls can block for many seconds on a drive that is
asleep, disconnected or faulty. Doing them here keeps the windows responsive;
the GUI only ever reads the latest finished snapshot in `snap`."""
def __init__(self) -> None:
super().__init__(name="volume-catalog", daemon=True)
self.snap: dict = {"volumes": [], "kind": {}, "label": {}, "disks": {}, "usage": {}}
self.stop_event = threading.Event()
self.step = "starting" # what the thread is doing right now
self.step_since = time.monotonic()
self.scans = 0
self._disk_cache: dict[str, tuple[float, list[int]]] = {}
def _at(self, step: str) -> None:
self.step = step
self.step_since = time.monotonic()
def _publish(self, **changes) -> None:
snap = dict(self.snap)
snap.update(changes)
self.snap = snap # replaced as a whole: readers never see a half-update
def run(self) -> None:
while not self.stop_event.is_set():
try:
self.scan()
except Exception:
import traceback
log_error("in the volume scanner\n" + traceback.format_exc())
self._at("idle")
self.stop_event.wait(RESCAN_MS / 1000)
def scan(self) -> None:
k32 = ctypes.windll.kernel32
self._at("listing drive letters")
mask = k32.GetLogicalDrives()
kinds: dict[str, int] = {}
for i, letter in enumerate(string.ascii_uppercase):
if letter not in "AB" and (mask >> i) & 1:
kinds[f"{letter}:"] = k32.GetDriveTypeW(f"{letter}:\\")
if not self.snap["volumes"]: # first scan: show the obvious drives immediately
self._publish(volumes=[v for v, k in kinds.items() if k in (3, 4, 6)], kind=dict(kinds))
volumes = []
for vol, kind in kinds.items():
if kind in (3, 4, 6):
volumes.append(vol)
elif kind in (2, 5): # card readers / optical drives only if media is present
self._at(f"checking for media in {vol}")
if has_media(vol):
volumes.append(vol)
self._publish(volumes=volumes, kind=dict(kinds))
for vol in list(self._disk_cache):
if vol not in volumes:
del self._disk_cache[vol]
for vol in volumes:
self._at(f"reading the label of {vol}")
label = volume_label(vol)
cached = self._disk_cache.get(vol)
if cached is None or time.monotonic() - cached[0] > 60:
if kinds.get(vol) in (2, 3): # only local disks have a physical disk; opening
self._at(f"finding the physical disk of {vol}") # a network drive can hang
numbers = volume_disk_numbers(vol)
else:
numbers = []
cached = (time.monotonic(), numbers)
self._disk_cache[vol] = cached
self._publish(label={**self.snap["label"], vol: label},
disks={**self.snap["disks"], vol: cached[1]})
for vol in volumes:
self._at(f"reading the free space of {vol}")
try:
u = shutil.disk_usage(vol + "\\")
usage = (u.used, u.total, u.free)
except OSError:
usage = None
self._publish(usage={**self.snap["usage"], vol: usage})
keep = set(volumes)
self._publish(**{key: {v: x for v, x in self.snap[key].items() if v in keep}
for key in ("label", "disks", "usage")})
self.scans += 1
# --- start with Windows ----------------------------------------------------
RUN_KEY = r"Software\Microsoft\Windows\CurrentVersion\Run"
def startup_command() -> str:
if getattr(sys, "frozen", False):
return f'"{sys.executable}"'
exe = sys.executable
if exe.lower().endswith("python.exe"):
exe = exe[:-len("python.exe")] + "pythonw.exe"
return f'"{exe}" "{Path(__file__).resolve()}"'
def startup_enabled() -> bool:
import winreg
try:
with winreg.OpenKey(winreg.HKEY_CURRENT_USER, RUN_KEY) as key:
winreg.QueryValueEx(key, APP_NAME)
return True
except OSError:
return False
def set_startup(enabled: bool) -> None:
import winreg
with winreg.OpenKey(winreg.HKEY_CURRENT_USER, RUN_KEY, 0, winreg.KEY_SET_VALUE) as key:
if enabled:
winreg.SetValueEx(key, APP_NAME, 0, winreg.REG_SZ, startup_command())
else:
try:
winreg.DeleteValue(key, APP_NAME)
except FileNotFoundError:
pass
# --------------------------------------------------------------------------
# PDH (Performance Data Helper) access
# --------------------------------------------------------------------------
PDH_FMT_DOUBLE = 0x00000200
PDH_MORE_DATA = 0x800007D2
class PDH_FMT_COUNTERVALUE(Structure):
_fields_ = [("CStatus", c_ulong), ("doubleValue", c_double)]
class Pdh:
def __init__(self) -> None:
dll = ctypes.WinDLL("pdh")
dll.PdhOpenQueryW.argtypes = [ctypes.c_wchar_p, c_size_t, POINTER(c_void_p)]
dll.PdhAddEnglishCounterW.argtypes = [c_void_p, ctypes.c_wchar_p, c_size_t, POINTER(c_void_p)]
dll.PdhCollectQueryData.argtypes = [c_void_p]
dll.PdhGetFormattedCounterValue.argtypes = [
c_void_p, c_ulong, POINTER(c_ulong), POINTER(PDH_FMT_COUNTERVALUE)]
dll.PdhCloseQuery.argtypes = [c_void_p]
dll.PdhGetFormattedCounterArrayW.argtypes = [
c_void_p, c_ulong, POINTER(c_ulong), POINTER(c_ulong), c_void_p]
for name in ("PdhOpenQueryW", "PdhAddEnglishCounterW", "PdhCollectQueryData",
"PdhGetFormattedCounterValue", "PdhCloseQuery",
"PdhGetFormattedCounterArrayW"):
getattr(dll, name).restype = c_ulong
self.dll = dll
class VolumeMonitor:
"""Reads/writes/IOPS for one volume through three LogicalDisk counters."""
COUNTERS = ("Disk Read Bytes/sec", "Disk Write Bytes/sec", "Disk Transfers/sec")
source = "LogicalDisk performance counters"
def __init__(self, pdh: Pdh, volume: str) -> None:
self.pdh = pdh
self.volume = volume
self.query = c_void_p()
self.counters: list[c_void_p] = []
self.valid = False
self.read_bps = 0.0
self.write_bps = 0.0
self.iops = 0.0
self.status = "starting"
self.bad_since: float | None = None # when counters last stopped returning data
self.last_pdh_status = 0
dll = pdh.dll
status = dll.PdhOpenQueryW(None, 0, byref(self.query))
if status != 0:
self.status = f"could not open a performance query (PDH 0x{status:08X})"
return
for name in self.COUNTERS:
handle = c_void_p()
path = f"\\LogicalDisk({volume})\\{name}"
status = dll.PdhAddEnglishCounterW(self.query, path, 0, byref(handle))
if status != 0:
dll.PdhCloseQuery(self.query)
self.status = f"no LogicalDisk({volume}) counters in Windows (PDH 0x{status:08X})"
return
self.counters.append(handle)
dll.PdhCollectQueryData(self.query) # prime the rate counters
self.valid = True
self.status = "OK"
@property
def stale(self) -> bool:
"""True when the counters have returned no data for a few seconds.
This happens when the drive was ejected/re-inserted or Windows re-created
its performance instance: the old counter handle then stays dead forever,
so the monitor must be rebuilt."""
return self.valid and self.bad_since is not None and time.monotonic() - self.bad_since > 3.0
def _value(self, handle: c_void_p) -> float | None:
out = PDH_FMT_COUNTERVALUE()
status = self.pdh.dll.PdhGetFormattedCounterValue(handle, PDH_FMT_DOUBLE, None, byref(out))
if status == 0 and out.CStatus in (0, 1): # valid data / new data
return max(0.0, out.doubleValue)
self.last_pdh_status = status or out.CStatus
return None
def _note(self, ok: bool) -> None:
if ok:
self.bad_since = None
self.status = "OK"
else:
if self.bad_since is None:
self.bad_since = time.monotonic()
self.status = f"counters returning no data (PDH 0x{self.last_pdh_status:08X})"
def sample(self) -> None:
if not self.valid:
return
status = self.pdh.dll.PdhCollectQueryData(self.query)
if status != 0:
self.last_pdh_status = status
self._note(False)
self.read_bps = self.write_bps = self.iops = 0.0
return
values = [self._value(h) for h in self.counters]
self._note(any(v is not None for v in values))
self.read_bps, self.write_bps, self.iops = (v or 0.0 for v in values)
def close(self) -> None:
if self.valid:
self.pdh.dll.PdhCloseQuery(self.query)
self.valid = False
class NullMonitor:
"""Stand-in for a drive whose activity cannot be measured; shown as a grey LED."""
source = "none"
valid = False
stale = False
read_bps = write_bps = iops = 0.0
def __init__(self, volume: str, reason: str) -> None:
self.volume = volume
self.status = reason
def sample(self) -> None:
pass
def close(self) -> None:
pass
class PDH_FMT_COUNTERVALUE_ITEM(Structure):
_fields_ = [("szName", c_wchar_p), ("FmtValue", PDH_FMT_COUNTERVALUE)]
class ProcessIoMonitor:
"""System-wide per-program I/O rates (bytes/sec) from the Process counters.
These counters are per process, not per volume, and also include non-disk
I/O (network, pipes). App uses them together with the per-volume rates to
estimate which programs are behind each drive's activity.
"""
PATHS = ("\\Process(*)\\IO Read Bytes/sec", "\\Process(*)\\IO Write Bytes/sec")
SKIP = {"_total", "idle"}
def __init__(self, pdh: Pdh) -> None:
self.pdh = pdh
self.query = c_void_p()
self.counters: list[c_void_p] = []
self.valid = False
dll = pdh.dll
if dll.PdhOpenQueryW(None, 0, byref(self.query)) != 0:
return
for path in self.PATHS:
handle = c_void_p()
if dll.PdhAddEnglishCounterW(self.query, path, 0, byref(handle)) != 0:
dll.PdhCloseQuery(self.query)
return
self.counters.append(handle)
dll.PdhCollectQueryData(self.query)
self.valid = True
def _array(self, handle: c_void_p) -> dict[str, float]:
dll = self.pdh.dll
size, count = c_ulong(0), c_ulong(0)
status = dll.PdhGetFormattedCounterArrayW(handle, PDH_FMT_DOUBLE, byref(size), byref(count), None)
if status != PDH_MORE_DATA or size.value == 0:
return {}
buf = ctypes.create_string_buffer(size.value)
status = dll.PdhGetFormattedCounterArrayW(handle, PDH_FMT_DOUBLE, byref(size), byref(count), buf)
if status != 0:
return {}
items = (PDH_FMT_COUNTERVALUE_ITEM * count.value).from_buffer(buf)
out: dict[str, float] = {}
for item in items:
name = item.szName
if not name or name.lower() in self.SKIP or item.FmtValue.CStatus not in (0, 1):
continue
base = re.sub(r"#\d+$", "", name) # "chrome#3" -> "chrome"
out[base] = out.get(base, 0.0) + max(0.0, item.FmtValue.doubleValue)
return out
def sample(self) -> dict[str, float] | None:
"""Return {program name: read+write bytes/sec} or None if unavailable."""
if not self.valid or self.pdh.dll.PdhCollectQueryData(self.query) != 0:
return None
reads = self._array(self.counters[0])
writes = self._array(self.counters[1])
total = dict(reads)
for name, value in writes.items():
total[name] = total.get(name, 0.0) + value
return total
def close(self) -> None:
if self.valid:
self.pdh.dll.PdhCloseQuery(self.query)
self.valid = False
def counter_array(pdh: Pdh, handle: c_void_p) -> dict[str, float]:
"""Values of a wildcard counter, keyed by instance name (valid values only)."""
dll = pdh.dll
size, count = c_ulong(0), c_ulong(0)
status = dll.PdhGetFormattedCounterArrayW(handle, PDH_FMT_DOUBLE, byref(size), byref(count), None)
if status != PDH_MORE_DATA or size.value == 0:
return {}
buf = ctypes.create_string_buffer(size.value)
status = dll.PdhGetFormattedCounterArrayW(handle, PDH_FMT_DOUBLE, byref(size), byref(count), buf)
if status != 0:
return {}
out: dict[str, float] = {}
for item in (PDH_FMT_COUNTERVALUE_ITEM * count.value).from_buffer(buf):
if item.szName and item.FmtValue.CStatus in (0, 1):
out[item.szName] = max(0.0, item.FmtValue.doubleValue)
return out
class SmbMonitor:
"""Activity of a mapped network drive, from the 'SMB Client Shares' counters."""
source = "SMB client share counters"
NAMES = ("Read Bytes/sec", "Write Bytes/sec", "Read Requests/sec", "Write Requests/sec")
stale = False
def __init__(self, pdh: Pdh, volume: str) -> None:
self.pdh = pdh
self.volume = volume
self.query = c_void_p()
self.counters: list[c_void_p] = []
self.valid = False
self.read_bps = self.write_bps = self.iops = 0.0
self.unc = unc_for_drive(volume)
self.status = "not a mapped network drive"
if not self.unc:
return
dll = pdh.dll
status = dll.PdhOpenQueryW(None, 0, byref(self.query))
if status != 0:
self.status = f"could not open a performance query (PDH 0x{status:08X})"
return
for name in self.NAMES:
handle = c_void_p()
status = dll.PdhAddEnglishCounterW(self.query, f"\\SMB Client Shares(*)\\{name}", 0, byref(handle))
if status != 0:
dll.PdhCloseQuery(self.query)
self.status = f"SMB client counters not available (PDH 0x{status:08X})"
return
self.counters.append(handle)
dll.PdhCollectQueryData(self.query)
self.valid = True
self.status = f"waiting for {self.unc} to appear in the SMB counters"
def sample(self) -> None:
if not self.valid:
return
if self.pdh.dll.PdhCollectQueryData(self.query) != 0:
self.read_bps = self.write_bps = self.iops = 0.0
return
want = self.unc.lower().rstrip("\\")
values = []
for handle in self.counters:
found = None
for name, value in counter_array(self.pdh, handle).items():
if name.lower().rstrip("\\") == want:
found = value
break
values.append(found)
if all(v is None for v in values):
self.status = f"{self.unc} is not listed in the SMB counters (no recent activity?)"
self.read_bps = self.write_bps = self.iops = 0.0
return
self.status = "OK"
read, write, read_req, write_req = (v or 0.0 for v in values)
self.read_bps, self.write_bps, self.iops = read, write, read_req + write_req
def close(self) -> None:
if self.valid:
self.pdh.dll.PdhCloseQuery(self.query)
self.valid = False
class PhysicalDiskMonitor:
"""Read/write rate of whole physical disks (used by the Diagnostics window)."""
PATHS = ("\\PhysicalDisk(*)\\Disk Read Bytes/sec", "\\PhysicalDisk(*)\\Disk Write Bytes/sec",
"\\PhysicalDisk(*)\\Disk Transfers/sec")
def __init__(self, pdh: Pdh) -> None:
self.pdh = pdh
self.query = c_void_p()
self.counters: list[c_void_p] = []
self.valid = False
dll = pdh.dll
if dll.PdhOpenQueryW(None, 0, byref(self.query)) != 0:
return
for path in self.PATHS:
handle = c_void_p()
if dll.PdhAddEnglishCounterW(self.query, path, 0, byref(handle)) != 0:
dll.PdhCloseQuery(self.query)
return
self.counters.append(handle)
dll.PdhCollectQueryData(self.query)
self.valid = True
def sample(self) -> dict[int, tuple[float, float, float]] | None:
if not self.valid or self.pdh.dll.PdhCollectQueryData(self.query) != 0:
return None
result: dict[int, list[float]] = {} # disk number -> [read B/s, write B/s, IOPS]
for idx, handle in enumerate(self.counters):
for name, value in counter_array(self.pdh, handle).items():
match = re.match(r"(\d+)(\s|$)", name) # instance names look like "1 E: F:"
if match:
result.setdefault(int(match.group(1)), [0.0, 0.0, 0.0])[idx] += value
return {k: (v[0], v[1], v[2]) for k, v in result.items()}
def close(self) -> None:
if self.valid:
self.pdh.dll.PdhCloseQuery(self.query)
self.valid = False
# --------------------------------------------------------------------------
# Tooltip
# --------------------------------------------------------------------------
class Tooltip:
def __init__(self, root: tk.Tk) -> None:
self.root = root
self.win: tk.Toplevel | None = None
self.label: tk.Label | None = None
def show(self, text: str, x: int, y: int) -> None:
if self.win is None:
self.win = tk.Toplevel(self.root)
self.win.overrideredirect(True)
self.win.attributes("-topmost", True)
self.label = tk.Label(self.win, text=text, justify="left", bg="#fffbe6",
fg="#222222", relief="solid", borderwidth=1,
font=("Consolas", 9), padx=6, pady=4)
self.label.pack()
else:
self.label.config(text=text)
self.win.update_idletasks()
w, h = self.win.winfo_reqwidth(), self.win.winfo_reqheight()
sw, sh = self.root.winfo_screenwidth(), self.root.winfo_screenheight()
px = min(max(0, x), sw - w)
py = y - h - 8
if py < 0:
py = y + 40
self.win.geometry(f"+{px}+{min(py, sh - h)}")
self.win.lift()
def hide(self) -> None:
if self.win is not None:
self.win.destroy()
self.win = None
self.label = None
# --------------------------------------------------------------------------
# One LED window per volume
# --------------------------------------------------------------------------
class Indicator:
def __init__(self, app: "App", volume: str, monitor: VolumeMonitor, index: int) -> None:
self.app = app
self.volume = volume
self.monitor = monitor
self.active_until = 0.0
self.current_color = ""
self.io_acc = 0.0 # bytes moved since last process sample
self.attrib: dict[str, float] = {} # program -> smoothed attributed bytes/sec
self._drag: tuple[int, int] | None = None
self._hovering = False
self._tip_job: str | None = None
self.win = tk.Toplevel(app.root)
self.win.overrideredirect(True)
self.win.configure(bg=TRANSPARENT)
self.win.attributes("-transparentcolor", TRANSPARENT)
self.canvas = tk.Canvas(self.win, bg=TRANSPARENT, highlightthickness=0, bd=0)
self.canvas.pack()
self.shape_id = self.canvas.create_oval(0, 0, 1, 1)
self.text_id = self.canvas.create_text(0, 0, text=volume[0], font=("Segoe UI", 9, "bold"))
self.apply_config()
self._place(index)
for widget in (self.canvas,):
widget.bind("", self._press)
widget.bind("", self._motion)
widget.bind("", self._release)
widget.bind("", self._context_menu)
widget.bind("", self._enter)
widget.bind("", self._leave)
# -- layout ------------------------------------------------------------
def _place(self, index: int) -> None:
cfg = self.app.cfg
sw, sh = self.app.root.winfo_screenwidth(), self.app.root.winfo_screenheight()
pos = cfg.positions.get(self.volume) if cfg.remember_positions else None
if pos and len(pos) == 2:
x, y = int(pos[0]), int(pos[1])
else:
# default: top of the screen, 30% from the left edge, LEDs side by side
gap = 6
x = int(sw * 0.30) + index * (cfg.size + gap)
y = 0
x = min(max(0, x), max(0, sw - cfg.size))
y = min(max(0, y), max(0, sh - cfg.size))
self.win.geometry(f"{cfg.size}x{cfg.size}+{x}+{y}")
def apply_config(self) -> None:
cfg = self.app.cfg
s = cfg.size
self.canvas.config(width=s, height=s)
self.canvas.delete(self.shape_id)
draw = self.canvas.create_oval if cfg.shape == "circle" else self.canvas.create_rectangle
self.shape_id = draw(1, 1, s - 1, s - 1, outline="#202020", width=1)
self.canvas.tag_lower(self.shape_id)
self.canvas.coords(self.text_id, s / 2, s / 2)
self.canvas.itemconfig(self.text_id, font=("Segoe UI", max(6, int(s * 0.36)), "bold"),
state="normal" if cfg.show_letters else "hidden")
self.win.attributes("-alpha", min(1.0, max(0.2, cfg.opacity)))
self.win.attributes("-topmost", cfg.always_on_top)
x, y = self.win.winfo_x(), self.win.winfo_y()
self.win.geometry(f"{s}x{s}+{x}+{y}")
self.current_color = "" # force repaint
self.refresh()
@property
def label(self) -> str:
return self.app.catalog.snap["label"].get(self.volume, "")
# -- state -------------------------------------------------------------
def refresh(self) -> None:
cfg = self.app.cfg
m = self.monitor
if not m.valid: # cannot be measured: grey LED
if self.current_color != cfg.unavailable_color:
self.current_color = cfg.unavailable_color
self.canvas.itemconfig(self.shape_id, fill=cfg.unavailable_color)
self.canvas.itemconfig(self.text_id, fill=text_color_for(cfg.unavailable_color))
return
threshold = cfg.threshold_kbps * 1024
reading = m.read_bps > threshold
writing = m.write_bps > threshold
now = time.monotonic()
if reading or writing or (threshold == 0 and m.iops > 0):
self.active_until = now + cfg.hold_ms / 1000.0
# Classify by the dominant direction: a trickle of housekeeping writes during a
# big read (or the reverse) must not turn a read-only LED into "both".
if reading and writing:
if m.write_bps < m.read_bps * RW_DOMINANCE:
writing = False
elif m.read_bps < m.write_bps * RW_DOMINANCE:
reading = False
self._last_rw = (reading, writing)
active = now <= self.active_until # '<=': a 0 ms hold must still show this sample
if not active:
color = cfg.idle_color
elif cfg.separate_rw:
r, w = getattr(self, "_last_rw", (True, True))
if r and w or not (r or w): # both directions, or I/O seen but no bytes measured
color = cfg.active_color
else:
color = cfg.read_color if r else cfg.write_color
else:
color = cfg.active_color
if color != self.current_color:
self.current_color = color
self.canvas.itemconfig(self.shape_id, fill=color)
self.canvas.itemconfig(self.text_id, fill=text_color_for(color))
def update_attribution(self, procs: dict[str, float], vol_rate: float,
total_vol_rate: float, total_proc_rate: float) -> None:
"""Split this volume's I/O among programs, weighted by their activity."""
smooth = 0.6 if vol_rate > 1.0 else 0.25 # fade out quickly once the drive is idle
share = vol_rate / total_vol_rate if total_vol_rate > 0 else 0.0
for name in set(procs) | set(self.attrib):
rate = procs.get(name, 0.0)
if vol_rate > 1.0 and total_proc_rate > 0 and rate > 0:
new = min(rate * share, vol_rate * rate / total_proc_rate)
else:
new = 0.0
value = smooth * self.attrib.get(name, 0.0) + (1 - smooth) * new
if value < 512:
self.attrib.pop(name, None)
else:
self.attrib[name] = value
def top_programs(self) -> list[tuple[str, float]]:
if not self.attrib or time.monotonic() > self.active_until + 1.5:
return []
ranked = sorted(self.attrib.items(), key=lambda kv: kv[1], reverse=True)
cutoff = ranked[0][1] * 0.10 # skip programs far behind the leader
return [kv for kv in ranked if kv[1] >= cutoff][:max(1, self.app.cfg.process_count)]
def tooltip_text(self) -> str:
m = self.monitor
title = f"{self.volume} {self.label}".rstrip()
if not m.valid:
kind = DRIVE_TYPE_NAMES.get(self.app.catalog.snap["kind"].get(self.volume, 0), "unknown")
return "\n".join([title, "", "Activity cannot be measured for this drive:",
f" {m.status}", f" Drive type: {kind}", "",
"(right-click > Diagnostics for details)"])
lines = [title,
f"Read: {format_rate(m.read_bps)}",
f"Write: {format_rate(m.write_bps)}",
f"IOPS: {m.iops:,.0f}"]
phys_n = getattr(m, "phys_disk", None)
if phys_n is not None:
lines.append(f"Source: physical disk {phys_n} (raw access)")
elif not isinstance(m, VolumeMonitor):
lines.append(f"Source: {m.source}")
if self.app.cfg.show_totals:
total_read, total_written = self.app.totals.get(self.volume, (0.0, 0.0))
lines.append("")
lines.append(f"Total read: {format_total(total_read)}")
lines.append(f"Total written: {format_total(total_written)}")
lines.append(f"(since {self.app.totals_since:%Y-%m-%d %H:%M})")
usage = self.app.catalog.snap["usage"].get(self.volume)
if usage:
used, total, free = usage
pct = used / total * 100 if total else 0
lines.append("")
lines.append(f"Used: {format_size(used)} of {format_size(total)} ({pct:.0f}%)")
lines.append(f"Free: {format_size(free)}")
# Programs go last, in a block of fixed height, so that the lines above it
# never move and the tooltip does not change size as programs come and go.
if self.app.cfg.show_processes and self.app.procmon is not None:
slots = max(1, self.app.cfg.process_count)
top = self.top_programs()
lines.append("")
lines.append("Active programs (estimated):")
rows = [f" {name[:24]:<24} {format_rate(rate):>10}" for name, rate in top]
if not rows:
rows = [" (none)"]
rows += ["\u00a0"] * (slots - len(rows))
lines.extend(rows)
return "\n".join(lines)
# -- tooltip -----------------------------------------------------------
def _enter(self, _event: tk.Event) -> None:
self._hovering = True
self._tip_job = self.win.after(250, self.update_tooltip)
def _leave(self, _event: tk.Event) -> None:
self._hovering = False
if self._tip_job:
self.win.after_cancel(self._tip_job)
self._tip_job = None
self.app.tooltip.hide()
def update_tooltip(self) -> None:
if self._hovering and self._drag is None:
self.app.tooltip.show(self.tooltip_text(), self.win.winfo_rootx(), self.win.winfo_rooty())
# -- dragging ----------------------------------------------------------
def _press(self, event: tk.Event) -> None:
if not self.app.cfg.locked:
self._drag = (event.x_root - self.win.winfo_x(), event.y_root - self.win.winfo_y())
self.app.tooltip.hide()
def _motion(self, event: tk.Event) -> None:
if self._drag:
self.win.geometry(f"+{event.x_root - self._drag[0]}+{event.y_root - self._drag[1]}")
def _release(self, _event: tk.Event) -> None:
if self._drag:
self._drag = None
if self.app.cfg.remember_positions:
self.app.save_positions()
# -- context menu ------------------------------------------------------
def _context_menu(self, event: tk.Event) -> None:
self.app.tooltip.hide()
menu = tk.Menu(self.win, tearoff=0)
menu.add_command(label=f"{self.volume} - Open in Explorer",
command=lambda: os.startfile(self.volume + "\\"))
menu.add_separator()
menu.add_command(label="Settings...", command=self.app.open_settings)
menu.add_command(label=f"Hide {self.volume}", command=lambda: self.app.hide_volume(self.volume))
menu.add_command(label="Reset positions", command=self.app.reset_positions)
menu.add_command(label="Reset totals", command=self.app.reset_totals)
menu.add_command(label="Diagnostics...", command=self.app.open_diagnostics)
menu.add_separator()
menu.add_command(label="Exit", command=self.app.quit)
try:
menu.tk_popup(event.x_root, event.y_root)
finally:
menu.grab_release()
def destroy(self) -> None:
self.app.tooltip.hide()
self.win.destroy()
# --------------------------------------------------------------------------
# Settings dialog
# --------------------------------------------------------------------------
class SettingsDialog:
def __init__(self, app: "App") -> None:
self.app = app
cfg = app.cfg
self.win = tk.Toplevel(app.root)
self.win.title(f"{APP_NAME} settings")
self.win.resizable(False, False)
self.win.attributes("-topmost", True)
self.size = tk.IntVar(value=cfg.size)
self.opacity = tk.IntVar(value=int(cfg.opacity * 100))
self.poll = tk.IntVar(value=cfg.poll_ms)
self.hold = tk.IntVar(value=cfg.hold_ms)
self.threshold = tk.DoubleVar(value=cfg.threshold_kbps)
self.shape = tk.StringVar(value=cfg.shape)
self.letters = tk.BooleanVar(value=cfg.show_letters)
self.topmost = tk.BooleanVar(value=cfg.always_on_top)
self.locked = tk.BooleanVar(value=cfg.locked)
self.remember = tk.BooleanVar(value=cfg.remember_positions)
self.show_unavail = tk.BooleanVar(value=cfg.show_unavailable)
self.use_phys = tk.BooleanVar(value=cfg.use_physical_disk)
self.show_totals = tk.BooleanVar(value=cfg.show_totals)
self.show_procs = tk.BooleanVar(value=cfg.show_processes)
self.proc_count = tk.IntVar(value=cfg.process_count)
self.separate = tk.BooleanVar(value=cfg.separate_rw)
self.autostart = tk.BooleanVar(value=startup_enabled())
self.colors = {"idle_color": cfg.idle_color, "active_color": cfg.active_color,
"read_color": cfg.read_color, "write_color": cfg.write_color,
"unavailable_color": cfg.unavailable_color}
self.color_buttons: dict[str, tk.Button] = {}
pad = {"padx": 8, "pady": 3}
frame = ttk.Frame(self.win, padding=10)
frame.grid()
row = 0
ttk.Label(frame, text="Appearance", font=("Segoe UI", 9, "bold")).grid(row=row, column=0, sticky="w", **pad)
row += 1
for text, var, lo, hi in (("Size (px)", self.size, 12, 96),
("Opacity (%)", self.opacity, 20, 100)):
ttk.Label(frame, text=text).grid(row=row, column=0, sticky="w", **pad)
ttk.Spinbox(frame, from_=lo, to=hi, textvariable=var, width=7).grid(row=row, column=1, sticky="w", **pad)
row += 1
ttk.Label(frame, text="Shape").grid(row=row, column=0, sticky="w", **pad)
ttk.Combobox(frame, textvariable=self.shape, values=("circle", "square"),
state="readonly", width=8).grid(row=row, column=1, sticky="w", **pad)
row += 1
for key, text in (("idle_color", "Idle colour"), ("active_color", "Active colour"),
("read_color", "Read-only colour"), ("write_color", "Write-only colour"),
("unavailable_color", "Unavailable colour")):
ttk.Label(frame, text=text).grid(row=row, column=0, sticky="w", **pad)
btn = tk.Button(frame, width=8, bg=self.colors[key], activebackground=self.colors[key],
command=lambda k=key: self.pick(k))
btn.grid(row=row, column=1, sticky="w", **pad)
self.color_buttons[key] = btn
row += 1
ttk.Checkbutton(frame, text="Use separate colours for read-only / write-only",
variable=self.separate).grid(row=row, column=0, columnspan=2, sticky="w", **pad)
row += 1
ttk.Separator(frame).grid(row=row, column=0, columnspan=2, sticky="ew", pady=6)
row += 1
ttk.Label(frame, text="Behaviour", font=("Segoe UI", 9, "bold")).grid(row=row, column=0, sticky="w", **pad)
row += 1
for text, var, lo, hi, inc in (("Update interval (ms)", self.poll, 50, 2000, 50),
("Minimum red time (ms)", self.hold, 0, 1000, 10),
("Ignore I/O below (KB/s)", self.threshold, 0, 100000, 1),
("Max programs in tooltip", self.proc_count, 1, 8, 1)):
ttk.Label(frame, text=text).grid(row=row, column=0, sticky="w", **pad)
ttk.Spinbox(frame, from_=lo, to=hi, increment=inc, textvariable=var, width=7).grid(
row=row, column=1, sticky="w", **pad)
row += 1
for text, var in (("Also count raw disk activity (drive alone on its disk)", self.use_phys),
("Show drives that cannot be measured (grey LED)", self.show_unavail),
("Show total bytes read/written in tooltip", self.show_totals),
("Show responsible programs in tooltip", self.show_procs),
("Show drive letters on LEDs", self.letters),
("Always on top (uncheck to let other windows cover the LEDs)", self.topmost),
("Remember indicator positions", self.remember),
("Lock position (disable dragging)", self.locked),
("Start with Windows", self.autostart)):
ttk.Checkbutton(frame, text=text, variable=var).grid(row=row, column=0, columnspan=2, sticky="w", **pad)
row += 1
pos_buttons = ttk.Frame(frame)
pos_buttons.grid(row=row, column=0, columnspan=2, sticky="w", **pad)
ttk.Button(pos_buttons, text="Save positions now",
command=self.app.save_positions).pack(side="left", padx=(0, 6))
ttk.Button(pos_buttons, text="Reset to default",
command=self.app.reset_positions).pack(side="left")
row += 1
ttk.Separator(frame).grid(row=row, column=0, columnspan=2, sticky="ew", pady=6)
row += 1
ttk.Label(frame, text="Volumes to show", font=("Segoe UI", 9, "bold")).grid(row=row, column=0, sticky="w", **pad)
row += 1
self.vol_vars: dict[str, tk.BooleanVar] = {}
for vol in app.catalog.snap["volumes"]:
var = tk.BooleanVar(value=vol not in cfg.hidden)
self.vol_vars[vol] = var
label = app.catalog.snap["label"].get(vol, "")
ttk.Checkbutton(frame, text=f"{vol} {label}".rstrip(), variable=var).grid(
row=row, column=0, columnspan=2, sticky="w", **pad)
row += 1
buttons = ttk.Frame(frame)
buttons.grid(row=row, column=0, columnspan=2, sticky="e", pady=(10, 0))
ttk.Button(buttons, text="OK", command=self.ok).pack(side="left", padx=4)
ttk.Button(buttons, text="Apply", command=self.apply).pack(side="left", padx=4)
ttk.Button(buttons, text="Cancel", command=self.win.destroy).pack(side="left", padx=4)
def pick(self, key: str) -> None:
_, hex_color = colorchooser.askcolor(color=self.colors[key], parent=self.win)
if hex_color:
self.colors[key] = hex_color
self.color_buttons[key].config(bg=hex_color, activebackground=hex_color)
def apply(self) -> None:
cfg = self.app.cfg
try:
cfg.size = min(96, max(12, int(self.size.get())))
cfg.opacity = min(100, max(20, int(self.opacity.get()))) / 100
cfg.poll_ms = min(2000, max(50, int(self.poll.get())))
cfg.hold_ms = min(1000, max(0, int(self.hold.get())))
cfg.threshold_kbps = max(0.0, float(self.threshold.get()))
cfg.process_count = min(8, max(1, int(self.proc_count.get())))
except (tk.TclError, ValueError):
messagebox.showerror(APP_NAME, "Please enter valid numbers.", parent=self.win)
return
cfg.shape = self.shape.get()
cfg.show_letters = self.letters.get()
cfg.always_on_top = self.topmost.get()
cfg.locked = self.locked.get()
cfg.remember_positions = self.remember.get()
cfg.show_processes = self.show_procs.get()
cfg.show_totals = self.show_totals.get()
cfg.show_unavailable = self.show_unavail.get()
cfg.use_physical_disk = self.use_phys.get()
cfg.separate_rw = self.separate.get()
for key, value in self.colors.items():
setattr(cfg, key, value)
cfg.hidden = [v for v, var in self.vol_vars.items() if not var.get()]
try:
set_startup(self.autostart.get())
except OSError as exc:
messagebox.showwarning(APP_NAME, f"Could not change startup setting:\n{exc}", parent=self.win)
cfg.save()
self.app.apply_config()
if cfg.remember_positions:
self.app.save_positions()
def ok(self) -> None:
self.apply()
self.win.destroy()
# --------------------------------------------------------------------------
# Diagnostics window
# --------------------------------------------------------------------------
class DiagnosticsDialog:
def __init__(self, app: "App") -> None:
self.app = app
self.win = tk.Toplevel(app.root)
self.win.title(f"{APP_NAME} diagnostics")
self.win.attributes("-topmost", True)
self.text = tk.Text(self.win, width=88, height=30, font=("Consolas", 9), wrap="none")
self.text.pack(fill="both", expand=True, padx=6, pady=(6, 0))
bar = ttk.Frame(self.win)
bar.pack(fill="x", padx=6, pady=6)
ttk.Button(bar, text="Copy to clipboard", command=self.copy).pack(side="left")
ttk.Label(bar, text=" Read self-test on:").pack(side="left")
volumes = list(app.catalog.snap["volumes"])
self.test_vol = tk.StringVar(value=volumes[0] if volumes else "")
ttk.Combobox(bar, textvariable=self.test_vol, values=volumes, width=5,
state="readonly").pack(side="left", padx=4)
self.test_btn = ttk.Button(bar, text="Run", command=self.run_test)
self.test_btn.pack(side="left")
self.test: dict | None = None
ttk.Button(bar, text="Close", command=self.close).pack(side="right")
self.phys = PhysicalDiskMonitor(app.pdh)
self.disks: dict[str, list[int]] = {}
self.count = 0
self.job: str | None = None
self.win.protocol("WM_DELETE_WINDOW", self.close)
self.refresh()
def refresh(self) -> None:
self.disks = self.app.catalog.snap["disks"]
self.count += 1
phys = self.phys.sample() if self.phys.valid else None
top = self.text.yview()[0]
self.text.config(state="normal")
self.text.delete("1.0", "end")
self.text.insert("1.0", self.test_text() + self.app.diagnostics_text(phys, self.disks))
self.text.config(state="disabled")
self.text.yview_moveto(top)
self.job = self.win.after(500, self.refresh)
# -- read self-test ------------------------------------------------------
def run_test(self) -> None:
vol = self.test_vol.get()
if not vol or (self.test and not self.test.get("final")):
return
app = self.app
disks = app.disk_map.get(vol) or []
disk = disks[0] if len(disks) == 1 else None
test = {"vol": vol, "disk": disk, "final": None, "done_at": None, "res": {},
"vol0": app.totals.get(vol, [0.0, 0.0])[0],
"log0": app.logical_totals.get(vol, [0.0, 0.0])[0],
"ph0": app.phys_totals.get(disk, [0.0, 0.0])[0] if disk is not None else None}
def work() -> None:
res = test["res"]
try:
path = find_test_file(vol)
if not path:
res["error"] = "no file of at least 1 MB found (copy a large file to the drive first)"
else:
res["path"] = path
res["bytes"], res["secs"], res["error"] = read_test(path)
except Exception as exc: # never let the thread die silently
res["error"] = f"{type(exc).__name__}: {exc}"
test["done_at"] = time.monotonic()
self.test = test
self.test_btn.config(state="disabled")
threading.Thread(target=work, daemon=True).start()
def test_text(self) -> str:
t = self.test
if not t:
return ""
if t["final"]:
return t["final"]
if t["done_at"] is None:
return f"Self-test on {t['vol']}: searching for a file and reading it ...\n\n"
if time.monotonic() - t["done_at"] < 1.5: # let the counters catch up
return f"Self-test on {t['vol']}: finished reading, collecting counters ...\n\n"
app, res, vol = self.app, t["res"], t["vol"]
if res.get("error") or not res.get("bytes"):
t["final"] = f"Self-test on {vol}: could not run - {res.get('error') or 'nothing was read'}\n\n"
else:
read = res["bytes"]
counted = app.totals.get(vol, [0.0, 0.0])[0] - t["vol0"]
logical = app.logical_totals.get(vol, [0.0, 0.0])[0] - t["log0"]
lines = [f"Self-test on {vol}: read {format_total(read)} in {res['secs']:.1f} s "
f"(uncached) from {res['path']}",
f" volume counters (LogicalDisk) saw : {format_total(max(logical, 0))}"
f" ({logical / read * 100:.0f}% of what was read)",
f" total shown in tooltip changed by : {format_total(max(counted, 0))}"]
if t["ph0"] is not None:
phys = app.phys_totals.get(t["disk"], [0.0, 0.0])[0] - t["ph0"]
lines.append(f" physical disk {t['disk']} counters saw : {format_total(max(phys, 0))}"
f" ({phys / read * 100:.0f}%)")
else:
phys = None
if logical >= 0.5 * read or counted >= 0.5 * read:
verdict = ("PASS - the program does see reads on this drive. If your own program's reads are\n"
" not counted, they are probably served from the Windows file cache (no disk\n"
" access), or they go to a different drive than you expect.")
elif phys is not None and phys >= 0.5 * read:
verdict = ("PARTIAL - only the physical-disk counters saw the reads; the volume counters\n"
" did not (Windows does not attribute this drive's reads to the volume).")
else:
verdict = ("FAIL - no counter saw the reads. Windows is not providing read counters for\n"
" this drive (see its status line below), or the counters are broken: try\n"
" 'lodctr /R' as administrator.")
lines.append(verdict)
lines.append("(Other disk activity during the test is included in these figures.)")
t["final"] = "\n".join(lines) + "\n\n"
self.test_btn.config(state="normal")
return t["final"]
def copy(self) -> None:
self.win.clipboard_clear()
self.win.clipboard_append(self.text.get("1.0", "end"))
def close(self) -> None:
if self.job:
self.win.after_cancel(self.job)
self.phys.close()
self.win.destroy()
# --------------------------------------------------------------------------
# Application
# --------------------------------------------------------------------------
class App:
def __init__(self) -> None:
self.cfg = Config.load()
self.catalog = VolumeCatalog()
self.catalog.start() # drive discovery runs off the GUI thread
self.root = tk.Tk()
self.root.withdraw()
self.root.title(APP_NAME)
self.tooltip = Tooltip(self.root)
self.pdh = Pdh()
self.procmon: ProcessIoMonitor | None = ProcessIoMonitor(self.pdh)
if not self.procmon.valid:
self.procmon = None
self.physmon: PhysicalDiskMonitor | None = PhysicalDiskMonitor(self.pdh)
if not self.physmon.valid:
self.physmon = None
self.tick_count = 0
self.logical_totals: dict[str, list[float]] = {} # volume counters only (before raw-disk fallback)
self.phys_totals: dict[int, list[float]] = {} # physical disk -> [read, written] bytes
self.error_count = 0
self._closed = False
self.started = time.monotonic()
self.totals: dict[str, list[float]] = {} # volume -> [bytes read, bytes written]
self.totals_since = datetime.now()
self.root.report_callback_exception = self._callback_exception
self.last_tick = time.monotonic()
self.last_proc = self.last_tick
self.indicators: dict[str, Indicator] = {}
self.settings: SettingsDialog | None = None
self.rescan()
self.root.after(self.cfg.poll_ms, self.tick)
self.root.after(300, self.rescan_loop)
self.root.after(100, self._check_exit_request)
@property
def disk_map(self) -> dict[str, list[int]]:
"""volume -> physical disk number(s), from the background scanner."""
return self.catalog.snap["disks"]
# -- volumes -------------------------------------------------------------
def make_monitor(self, vol: str):
"""Best available activity source for a volume (never raises)."""
monitor = VolumeMonitor(self.pdh, vol)
if monitor.valid:
return monitor
reason = monitor.status
if self.catalog.snap["kind"].get(vol) == 4: # mapped network drive
smb = SmbMonitor(self.pdh, vol)
if smb.valid:
return smb
reason = f"{reason}; {smb.status}"
return NullMonitor(vol, reason)
def rescan(self) -> None:
present = [v for v in self.catalog.snap["volumes"] if v not in self.cfg.hidden]
for vol in list(self.indicators):
if vol not in present:
ind = self.indicators.pop(vol)
ind.monitor.close()
ind.destroy()
for vol in present:
ind = self.indicators.get(vol)
if ind is None:
monitor = self.make_monitor(vol)
if not monitor.valid and not self.cfg.show_unavailable:
continue
self.indicators[vol] = Indicator(self, vol, monitor, present.index(vol))
elif ind.monitor.stale or not ind.monitor.valid:
# Counters went dead (drive re-inserted) or were never available: try again.
ind.monitor.close()
ind.monitor = self.make_monitor(vol)
ind.current_color = ""
if not ind.monitor.valid and not self.cfg.show_unavailable:
self.indicators.pop(vol).destroy()
def _report(self, where: str) -> None:
"""Log an unexpected error without ever stopping the sampling loop."""
import traceback
self.error_count += 1
text = f"in {where}\n{traceback.format_exc()}"
if self.error_count <= 20 or self.error_count % 100 == 0:
log_error(text)
else:
log_error(text.splitlines()[-1])
console(f"{APP_NAME}: error {text}")
def solo_disk(self, vol: str) -> int | None:
"""Physical disk number if this volume is the only lettered volume on exactly one disk."""
numbers = self.disk_map.get(vol) or []
if len(numbers) != 1:
return None
n = numbers[0]
sharers = [v for v, d in self.disk_map.items() if n in d]
return n if len(sharers) == 1 else None
def rescan_loop(self) -> None:
try:
self.rescan()
for ind in self.indicators.values(): # Windows can drop topmost; re-assert
if self.cfg.always_on_top:
ind.win.attributes("-topmost", True)
ind.win.lift()
except Exception:
self._report("volume rescan")
finally:
if not self._closed:
# look often during the first seconds so the LEDs appear quickly
fast = time.monotonic() - self.started < 10
self.root.after(300 if fast else RESCAN_MS, self.rescan_loop)
def tick(self) -> None:
try:
self._tick()
except Exception:
self._report("sampling loop")
finally:
if not self._closed:
self.root.after(self.cfg.poll_ms, self.tick)
def _tick(self) -> None:
now = time.monotonic()
tick_dt = now - self.last_tick
self.last_tick = now
self.tick_count += 1
phys = None
if self.physmon is not None and self.cfg.use_physical_disk:
try:
phys = self.physmon.sample()
except Exception:
self._report("raw-disk counters (feature switched off)")
self.physmon = None
if phys is not None:
for disk, (pr, pw, _pi) in phys.items():
t = self.phys_totals.setdefault(disk, [0.0, 0.0])
t[0] += pr * tick_dt
t[1] += pw * tick_dt
for ind in list(self.indicators.values()):
try:
m = ind.monitor
m.sample()
lt = self.logical_totals.setdefault(ind.volume, [0.0, 0.0])
lt[0] += m.read_bps * tick_dt
lt[1] += m.write_bps * tick_dt
m.phys_disk = None
if phys is not None and m.valid:
n = self.solo_disk(ind.volume)
if n in phys:
# The volume counters can miss raw/sector-level access that the
# disk counters see; for a drive alone on its disk, use the larger.
pr, pw, pi = phys[n]
if pr > m.read_bps * 1.5 + 4096 or pw > m.write_bps * 1.5 + 4096:
m.read_bps = max(m.read_bps, pr)
m.write_bps = max(m.write_bps, pw)
m.iops = max(m.iops, pi)
m.phys_disk = n
ind.io_acc += (m.read_bps + m.write_bps) * tick_dt
total = self.totals.setdefault(ind.volume, [0.0, 0.0])
total[0] += m.read_bps * tick_dt
total[1] += m.write_bps * tick_dt
ind.refresh()
except Exception:
self._report(f"sampling {ind.volume}")
try:
self.sample_processes(now)
except Exception:
self._report("program attribution (feature switched off)")
self.procmon = None
for ind in list(self.indicators.values()):
if ind._hovering:
try:
ind.update_tooltip()
except Exception:
self._report("tooltip")
def sample_processes(self, now: float) -> None:
"""Every ~0.5 s (only while some drive is busy) update program attribution."""
if self.procmon is None or not self.cfg.show_processes:
return
dt = now - self.last_proc
if dt < 0.5:
return
busy = any(now < ind.active_until + 3.0 for ind in self.indicators.values())
if not busy:
for ind in self.indicators.values():
ind.io_acc = 0.0
ind.attrib.clear()
self.last_proc = now
return
procs = self.procmon.sample()
stale = dt > 1.5 # long gap: counter average is diluted, skip this one
if procs is not None and not stale:
vol_rates = {v: ind.io_acc / dt for v, ind in self.indicators.items()}
total_vol = sum(vol_rates.values())
total_proc = sum(procs.values())
for vol, ind in self.indicators.items():
ind.update_attribution(procs, vol_rates[vol], total_vol, total_proc)
for ind in self.indicators.values():
ind.io_acc = 0.0
self.last_proc = now
# -- actions -------------------------------------------------------------
def apply_config(self) -> None:
self.rescan()
for ind in self.indicators.values():
ind.apply_config()
def hide_volume(self, vol: str) -> None:
if vol not in self.cfg.hidden:
self.cfg.hidden.append(vol)
self.cfg.save()
self.rescan()
def reset_totals(self) -> None:
self.totals.clear()
self.totals_since = datetime.now()
def request_exit(self, signum: int, _frame=None) -> None:
"""Signal handler (non-console cases, e.g. SIGTERM)."""
request_exit({signal.SIGINT: "Ctrl-C"}.get(signum, f"signal {signum}"))
def _check_exit_request(self) -> None:
if EXIT_REQUESTED.is_set():
self.quit()
elif not self._closed:
self.root.after(100, self._check_exit_request)
def _callback_exception(self, exc, val, tb) -> None:
if isinstance(val, KeyboardInterrupt):
self.request_exit(signal.SIGINT)
else:
import traceback
text = "".join(traceback.format_exception(exc, val, tb))
self.error_count += 1
log_error("in a Tk callback\n" + text)
console(text)
def print_summary(self) -> None:
if self.totals:
console(f"{APP_NAME}: totals since {self.totals_since:%Y-%m-%d %H:%M}")
for vol, (r, w) in sorted(self.totals.items()):
console(f" {vol} read {format_total(r)} written {format_total(w)}")
def save_positions(self) -> None:
for vol, ind in self.indicators.items():
self.cfg.positions[vol] = [ind.win.winfo_x(), ind.win.winfo_y()]
self.cfg.save()
def reset_positions(self) -> None:
self.cfg.positions.clear()
self.cfg.save()
for ind in self.indicators.values():
ind.destroy()
self.indicators.clear()
self.rescan()
def diagnostics_text(self, phys: dict[int, tuple[float, float, float]] | None,
disks: dict[str, list[int]]) -> str:
age = time.monotonic() - self.last_tick
out = [f"{APP_NAME} diagnostics {datetime.now():%Y-%m-%d %H:%M:%S}",
f"Python {sys.version.split()[0]} update interval {self.cfg.poll_ms} ms "
f"min red time {self.cfg.hold_ms} ms ignore below {self.cfg.threshold_kbps:g} KB/s",
f"sampler: {self.tick_count} ticks, last {age:.1f} s ago "
f"errors: {self.error_count} raw-disk counters: "
f"{'on' if self.physmon is not None and self.cfg.use_physical_disk else 'off'} "
f"program counters: {'on' if self.procmon is not None else 'off'}"]
cat = self.catalog
out.append(f"volume scanner: {cat.scans} scans, now: {cat.step} "
f"({time.monotonic() - cat.step_since:.0f} s) "
"(a long time here means that drive is not responding)")
if LAST_ERROR:
out.append(f"last error: {LAST_ERROR} (details: {config_path().with_name('error.log')})")
out.append("")
snap = self.catalog.snap
for vol in snap["volumes"]:
kind = DRIVE_TYPE_NAMES.get(snap["kind"].get(vol, 0), "unknown")
out.append(f"{vol} [{kind}] {snap['label'].get(vol, '')}".rstrip())
ind = self.indicators.get(vol)
if vol in self.cfg.hidden:
out.append(" hidden (Settings > Volumes to show)")
elif ind is None:
out.append(" no indicator (drive not measurable and grey LEDs are turned off)")
else:
m = ind.monitor
out.append(f" source : {m.source}")
out.append(f" status : {m.status}")
out.append(f" volume : read {format_rate(m.read_bps)} write {format_rate(m.write_bps)}"
f" IOPS {m.iops:,.0f}")
total_read, total_written = self.totals.get(vol, (0.0, 0.0))
out.append(f" totals : read {format_total(total_read)} written {format_total(total_written)}")
numbers = disks.get(vol, [])
if ind is not None and getattr(ind.monitor, "phys_disk", None) is not None:
out.append(f" raw-disk assist: ACTIVE (physical disk {ind.monitor.phys_disk} "
"showed more activity than the volume counters)")
elif ind is not None and ind.monitor.valid:
solo = self.solo_disk(vol)
out.append(" raw-disk assist: " + (f"ready (alone on physical disk {solo})" if solo is not None
else "not applicable (disk shared with another lettered volume or unknown)"))
if numbers and phys is not None:
for n in numbers:
r, w, _iops = phys.get(n, (0.0, 0.0, 0.0))
out.append(f" physical disk {n}: read {format_rate(r)} write {format_rate(w)}"
" (all volumes on it + raw access)")
elif numbers:
out.append(f" physical disk(s): {numbers}")
out.append("")
out.append("If a drive shows reads on its physical disk but 0 for the volume, the reads bypass")
out.append("the file system (raw/sector access) or come from another volume on the same disk.")
return "\n".join(out)
def open_diagnostics(self) -> None:
if getattr(self, "diagnostics", None) and self.diagnostics.win.winfo_exists():
self.diagnostics.win.lift()
return
self.diagnostics = DiagnosticsDialog(self)
def open_settings(self) -> None:
if self.settings and self.settings.win.winfo_exists():
self.settings.win.lift()
return
self.settings = SettingsDialog(self)
def quit(self) -> None:
if self._closed:
return
self._closed = True
self.catalog.stop_event.set()
if self.cfg.remember_positions:
self.save_positions()
self.cfg.save()
for ind in self.indicators.values():
ind.monitor.close()
if self.procmon is not None:
self.procmon.close()
self.root.destroy()
def run(self) -> None:
self.root.mainloop()
def main() -> None:
global _APP
if sys.platform != "win32":
sys.exit(f"{APP_NAME} only runs on Windows.")
# single instance
kernel32 = ctypes.windll.kernel32
mutex = kernel32.CreateMutexW(None, False, f"Local\\{APP_NAME}_single_instance")
if kernel32.GetLastError() == ERROR_ALREADY_EXISTS:
sys.exit(0)
keep_alive = install_console_handler() # Ctrl-C works even if the GUI thread is stuck
kernel32.SetErrorMode(1) # SEM_FAILCRITICALERRORS: no 'insert disk' dialogs
try: # crisp rendering on high-DPI displays
ctypes.windll.shcore.SetProcessDpiAwareness(1)
except (AttributeError, OSError):
pass
try:
app = App()
except Exception:
import traceback
log_error("while starting\n" + traceback.format_exc())
console(f"{APP_NAME}: could not start:\n{traceback.format_exc()}")
FINISHED.set()
sys.exit(1)
_APP = app
for name in ("SIGINT", "SIGBREAK", "SIGTERM"): # fallback for non-console situations
sig = getattr(signal, name, None)
if sig is not None:
signal.signal(sig, app.request_exit)
try:
app.run()
except KeyboardInterrupt: # fallback if it arrives outside a callback
request_exit("Ctrl-C")
app.quit()
app.quit() # no-op if already closed
app.print_summary()
console(f"{APP_NAME}: exited cleanly.")
FINISHED.set()
kernel32.CloseHandle(mutex)
del keep_alive
if __name__ == "__main__":
main()