|
| 1 | +#!/usr/bin/env python3 |
| 2 | +""" |
| 3 | +USB COM Port Monitor |
| 4 | +Small always-on-top desktop widget. |
| 5 | +
|
| 6 | +Usage: python monitor.py |
| 7 | +Needs: pip install pyserial pywin32 |
| 8 | +""" |
| 9 | + |
| 10 | +import tkinter as tk |
| 11 | +import math |
| 12 | +import time |
| 13 | + |
| 14 | +try: |
| 15 | + import serial.tools.list_ports |
| 16 | +except ImportError: |
| 17 | + print("We need pyserial library (and optionaly pywin32)."); |
| 18 | + |
| 19 | +try: |
| 20 | + import win32file, pywintypes |
| 21 | + HAS_WIN32 = True |
| 22 | +except ImportError: |
| 23 | + HAS_WIN32 = False |
| 24 | + |
| 25 | +# ── tunables ────────────────────────────────────────────────────────────────── |
| 26 | +REFRESH_MS = 500 |
| 27 | +NEW_DOT_S = 8 # seconds the dot/age text stays yellow |
| 28 | + |
| 29 | +C_ROW_FLASH = "#8a6218" # amber peak colour for new-port flash |
| 30 | +FLASH_ATTACK_S = 0.3 # seconds to reach peak brightness |
| 31 | +FLASH_DECAY_K = 0.50 # controls decay speed (higher = faster early drop) |
| 32 | +# brightness(t) = exp(-K * sqrt(t - attack)) ← ~0 by 60 s |
| 33 | + |
| 34 | +BOLD_THRESHOLD = 0.1 # keep bold text while brightness is above this |
| 35 | + |
| 36 | +# ── palette ─────────────────────────────────────────────────────────────────── |
| 37 | +C_BG = "#1a1a1a" |
| 38 | +C_HDR = "#252525" |
| 39 | +C_PORT = "#61dafb" |
| 40 | +C_VIDPID = "#f0e68c" |
| 41 | +C_FREE = "#7ec87e" |
| 42 | +C_OPEN = "#e06c75" |
| 43 | +C_NEW = "#ffcc44" |
| 44 | +C_AGE = "#666666" |
| 45 | +C_DESC = "#999999" |
| 46 | +C_HEAD = "#444444" |
| 47 | +C_SER = "#b0a0d0" |
| 48 | + |
| 49 | +FONT = ("Consolas", 10) |
| 50 | +FONT_BOLD = ("Consolas", 10, "bold") |
| 51 | +FONT_HDR = ("Consolas", 9, "bold") |
| 52 | + |
| 53 | +ALPHA_OPAQUE = 0.96 |
| 54 | +ALPHA_DIM = 0.35 |
| 55 | + |
| 56 | +COLS = [ |
| 57 | + ("Port", "w", False), |
| 58 | + ("VID:PID", "w", False), |
| 59 | + ("Age", "e", False), |
| 60 | + ("", "c", False), |
| 61 | + ("Status", "w", False), |
| 62 | + ("Serial / Loc", "w", False), |
| 63 | + ("Description", "w", True), |
| 64 | +] |
| 65 | + |
| 66 | + |
| 67 | +# ── flash curve ───────────────────────────────────────────────────────────── |
| 68 | +def _flash_brightness(t: float) -> float: |
| 69 | + """ |
| 70 | + 0..1 flash intensity at t seconds after port appeared. |
| 71 | +
|
| 72 | + Shape: linear attack up to FLASH_ATTACK_S, then |
| 73 | + exp(-K * sqrt(t - attack)) — fast initial drop, very slow tail. |
| 74 | + Reaches ≈0 around 60 s for default K=0.65. |
| 75 | + """ |
| 76 | + if t <= 0: |
| 77 | + return 0.0 |
| 78 | + if t < FLASH_ATTACK_S: |
| 79 | + return t / FLASH_ATTACK_S # quick linear ramp to 1.0 |
| 80 | + return math.exp(-FLASH_DECAY_K * math.sqrt(t - FLASH_ATTACK_S)) |
| 81 | + |
| 82 | + |
| 83 | +def _blend(c1: str, c2: str, t: float) -> str: |
| 84 | + """Interpolate two #rrggbb colours: t=0 → c1, t=1 → c2.""" |
| 85 | + t = max(0.0, min(1.0, t)) |
| 86 | + r = int(int(c1[1:3], 16) + (int(c2[1:3], 16) - int(c1[1:3], 16)) * t) |
| 87 | + g = int(int(c1[3:5], 16) + (int(c2[3:5], 16) - int(c1[3:5], 16)) * t) |
| 88 | + b = int(int(c1[5:7], 16) + (int(c2[5:7], 16) - int(c1[5:7], 16)) * t) |
| 89 | + return f"#{r:02x}{g:02x}{b:02x}" |
| 90 | + |
| 91 | + |
| 92 | +# ── port-open detection ─────────────────────────────────────────────────────── |
| 93 | +def _is_open_win32(device: str) -> bool: |
| 94 | + path = r"\\.\ "[:-1] + device |
| 95 | + try: |
| 96 | + h = win32file.CreateFile( |
| 97 | + path, |
| 98 | + win32file.GENERIC_READ | win32file.GENERIC_WRITE, |
| 99 | + 0, None, win32file.OPEN_EXISTING, 0, None, |
| 100 | + ) |
| 101 | + win32file.CloseHandle(h) |
| 102 | + return False |
| 103 | + except pywintypes.error as e: |
| 104 | + if e.winerror in (5, 32): |
| 105 | + return True |
| 106 | + return False |
| 107 | + |
| 108 | + |
| 109 | +def _is_open_fallback(device: str) -> bool: |
| 110 | + import serial |
| 111 | + try: |
| 112 | + s = serial.Serial(device, timeout=0) |
| 113 | + s.close() |
| 114 | + return False |
| 115 | + except Exception: |
| 116 | + return True |
| 117 | + |
| 118 | + |
| 119 | +def is_open(device: str) -> bool: |
| 120 | + return _is_open_win32(device) if HAS_WIN32 else _is_open_fallback(device) |
| 121 | + |
| 122 | + |
| 123 | +# ── main window ─────────────────────────────────────────────────────────────── |
| 124 | +class ComMonitor(tk.Tk): |
| 125 | + def __init__(self): |
| 126 | + super().__init__() |
| 127 | + self.overrideredirect(True) |
| 128 | + self.attributes("-topmost", True) |
| 129 | + self.attributes("-alpha", ALPHA_OPAQUE) |
| 130 | + self.configure(bg=C_BG) |
| 131 | + |
| 132 | + self._first_seen: dict[str, float] = {} # device → first-seen time |
| 133 | + self._flash_start: dict[str, float] = {} # device → flash start time (new arrivals only) |
| 134 | + self._initialized = False # skip flash for ports present at startup |
| 135 | + self._row_widgets: list[list[tk.Widget]] = [] |
| 136 | + self._drag_ox = self._drag_oy = 0 |
| 137 | + self._dimmed = False |
| 138 | + |
| 139 | + self._build_ui() |
| 140 | + self._refresh() |
| 141 | + |
| 142 | + # ── layout ──────────────────────────────────────────────────────────────── |
| 143 | + def _build_ui(self): |
| 144 | + bar = tk.Frame(self, bg=C_HDR, cursor="fleur") |
| 145 | + bar.pack(fill=tk.X) |
| 146 | + bar.bind("<ButtonPress-1>", self._drag_start) |
| 147 | + bar.bind("<B1-Motion>", self._drag_move) |
| 148 | + |
| 149 | + tk.Label(bar, text=" USB COM Monitor", |
| 150 | + bg=C_HDR, fg="#666666", font=("Segoe UI", 8), |
| 151 | + pady=4).pack(side=tk.LEFT) |
| 152 | + tk.Button(bar, text=" ✕ ", bg=C_HDR, fg="#666666", bd=0, relief="flat", |
| 153 | + font=("Segoe UI", 8), |
| 154 | + activebackground="#c0392b", activeforeground="white", |
| 155 | + command=self.destroy).pack(side=tk.RIGHT) |
| 156 | + |
| 157 | + self._dim_btn = tk.Button( |
| 158 | + bar, text=" ◑ ", bg=C_HDR, fg="#666666", bd=0, relief="flat", |
| 159 | + font=("Segoe UI", 8), |
| 160 | + activebackground=C_HDR, activeforeground="#aaaaaa", |
| 161 | + command=self._toggle_dim) |
| 162 | + self._dim_btn.pack(side=tk.RIGHT) |
| 163 | + |
| 164 | + self._grid = tk.Frame(self, bg=C_BG) |
| 165 | + self._grid.pack(fill=tk.BOTH, expand=True, padx=6, pady=(3, 6)) |
| 166 | + |
| 167 | + for c, (name, anchor, stretch) in enumerate(COLS): |
| 168 | + tk.Label(self._grid, text=name, bg=C_BG, fg=C_HEAD, |
| 169 | + font=FONT_HDR, anchor=anchor, padx=3 |
| 170 | + ).grid(row=0, column=c, sticky="ew") |
| 171 | + if stretch: |
| 172 | + self._grid.columnconfigure(c, weight=1) |
| 173 | + |
| 174 | + tk.Frame(self._grid, bg="#333333", height=1).grid( |
| 175 | + row=1, column=0, columnspan=len(COLS), sticky="ew", pady=(1, 3)) |
| 176 | + |
| 177 | + self._empty_lbl = tk.Label( |
| 178 | + self._grid, text="(no COM ports detected)", |
| 179 | + bg=C_BG, fg="#3a3a3a", font=FONT, anchor="w", padx=3) |
| 180 | + |
| 181 | + self.bind("<Escape>", lambda _: self.destroy()) |
| 182 | + self.bind("<Control-q>", lambda _: self.destroy()) |
| 183 | + |
| 184 | + # ── transparency toggle ─────────────────────────────────────────────────── |
| 185 | + def _toggle_dim(self): |
| 186 | + self._dimmed = not self._dimmed |
| 187 | + self.attributes("-alpha", ALPHA_DIM if self._dimmed else ALPHA_OPAQUE) |
| 188 | + self._dim_btn.config(fg="#aaaaaa" if self._dimmed else "#666666") |
| 189 | + |
| 190 | + # ── drag ───────────────────────────────────────────────────────────────── |
| 191 | + def _drag_start(self, e): |
| 192 | + self._drag_ox = e.x_root - self.winfo_x() |
| 193 | + self._drag_oy = e.y_root - self.winfo_y() |
| 194 | + |
| 195 | + def _drag_move(self, e): |
| 196 | + self.geometry(f"+{e.x_root - self._drag_ox}+{e.y_root - self._drag_oy}") |
| 197 | + |
| 198 | + # ── helpers ─────────────────────────────────────────────────────────────── |
| 199 | + @staticmethod |
| 200 | + def _age_str(s: float) -> str: |
| 201 | + s = int(s) |
| 202 | + if s < 60: return f"{s}s" |
| 203 | + m, s = divmod(s, 60) |
| 204 | + if m < 60: return f"{m}m{s:02d}s" |
| 205 | + h, m = divmod(m, 60) |
| 206 | + return f"{h}h{m:02d}m" |
| 207 | + |
| 208 | + # ── refresh ─────────────────────────────────────────────────────────────── |
| 209 | + def _refresh(self): |
| 210 | + now = time.time() |
| 211 | + ports = sorted(serial.tools.list_ports.comports(), key=lambda p: p.device) |
| 212 | + |
| 213 | + # update registries |
| 214 | + current = {p.device for p in ports} |
| 215 | + for dev in list(self._first_seen): |
| 216 | + if dev not in current: |
| 217 | + del self._first_seen[dev] |
| 218 | + self._flash_start.pop(dev, None) |
| 219 | + for p in ports: |
| 220 | + if p.device not in self._first_seen: |
| 221 | + self._first_seen[p.device] = now |
| 222 | + if self._initialized: # don't flash ports seen at startup |
| 223 | + # shift back so first render lands at peak brightness, not t=0 |
| 224 | + self._flash_start[p.device] = now - FLASH_ATTACK_S |
| 225 | + self._initialized = True |
| 226 | + |
| 227 | + # rebuild rows |
| 228 | + for row in self._row_widgets: |
| 229 | + for w in row: |
| 230 | + w.destroy() |
| 231 | + self._row_widgets.clear() |
| 232 | + self._empty_lbl.grid_forget() |
| 233 | + |
| 234 | + if not ports: |
| 235 | + self._empty_lbl.grid(row=2, column=0, columnspan=len(COLS), |
| 236 | + sticky="w", pady=6) |
| 237 | + else: |
| 238 | + for i, p in enumerate(ports): |
| 239 | + age_s = now - self._first_seen[p.device] |
| 240 | + occupied = is_open(p.device) |
| 241 | + |
| 242 | + vid = f"{p.vid:04X}" if p.vid is not None else "----" |
| 243 | + pid = f"{p.pid:04X}" if p.pid is not None else "----" |
| 244 | + |
| 245 | + serial_no = p.serial_number or "" |
| 246 | + location = p.location or "" |
| 247 | + ser_loc = " / ".join(filter(None, [serial_no, location])) or "—" |
| 248 | + |
| 249 | + desc = p.description or "" |
| 250 | + if desc == p.device: |
| 251 | + desc = "" |
| 252 | + |
| 253 | + # flash envelope |
| 254 | + flash_t = now - self._flash_start[p.device] if p.device in self._flash_start else None |
| 255 | + if flash_t is not None: |
| 256 | + brightness = _flash_brightness(flash_t) |
| 257 | + row_bg = _blend(C_ROW_FLASH, C_BG, 1.0 - brightness) |
| 258 | + row_font = FONT_BOLD if brightness > BOLD_THRESHOLD else FONT |
| 259 | + else: |
| 260 | + row_bg = C_BG |
| 261 | + row_font = FONT |
| 262 | + |
| 263 | + fresh = age_s < NEW_DOT_S |
| 264 | + dot_c = C_NEW if fresh else (C_OPEN if occupied else C_FREE) |
| 265 | + age_c = C_NEW if fresh else C_AGE |
| 266 | + stat_t = "OPEN" if occupied else "free" |
| 267 | + stat_c = C_OPEN if occupied else C_FREE |
| 268 | + |
| 269 | + cells = [ |
| 270 | + dict(text=p.device, fg=C_PORT, anchor="w"), |
| 271 | + dict(text=f"{vid}:{pid}", fg=C_VIDPID, anchor="w"), |
| 272 | + dict(text=self._age_str(age_s), fg=age_c, anchor="e"), |
| 273 | + dict(text="●", fg=dot_c, anchor="center"), |
| 274 | + dict(text=stat_t, fg=stat_c, anchor="w"), |
| 275 | + dict(text=ser_loc, fg=C_SER, anchor="w"), |
| 276 | + dict(text=desc, fg=C_DESC, anchor="w"), |
| 277 | + ] |
| 278 | + row_w = [] |
| 279 | + for c, kw in enumerate(cells): |
| 280 | + lbl = tk.Label(self._grid, bg=row_bg, font=row_font, |
| 281 | + padx=3, pady=1, **kw) |
| 282 | + lbl.grid(row=i + 2, column=c, sticky="ew") |
| 283 | + row_w.append(lbl) |
| 284 | + self._row_widgets.append(row_w) |
| 285 | + |
| 286 | + self.after(REFRESH_MS, self._refresh) |
| 287 | + |
| 288 | + |
| 289 | +if __name__ == "__main__": |
| 290 | + ComMonitor().mainloop() |
0 commit comments