Add cross-platform Python host agent with psutil and optional NVML.

Replace Windows-only WMI collector so Win10/11, Linux, and macOS can feed the same SI protocol.
This commit is contained in:
2026-10-27 20:21:08 +08:00
parent 116615818b
commit 8cc030178f
4 changed files with 494 additions and 247 deletions

View File

@@ -11,7 +11,7 @@
## 原理
ESP32 无法直接读 PC 硬件。本机跑 [`host/send_sysinfo.ps1`](host/send_sysinfo.ps1),经 **USB 串口** 按行推送:
ESP32 无法直接读 PC 硬件。主机端跑跨平台脚本,经 **USB 串口** 按行推送:
```
SI|host=DESKTOP|cpu=i7-13700K|cpu%=12.3|gpu=RTX 4070|gpu%=5|ram=18.2/31.8|disk=512/931
@@ -19,7 +19,26 @@ SI|host=DESKTOP|cpu=i7-13700K|cpu%=12.3|gpu=RTX 4070|gpu%=5|ram=18.2/31.8|disk=5
屏上 LVGL 面板解析后画进度条;超过约 5 秒无新包显示 **stale**。
GPU 占用:优先 `nvidia-smi`,否则尝试 Windows GPU Engine 计数器;都失败则显示 `n/a`(名称仍可读)。
### 通用主机方案(Win10/11 · Linux · macOS)
业界常规做法(本仓库已采用):
| 层级 | 库 / 接口 | 覆盖 |
|------|-----------|------|
| **CPU / 内存 / 磁盘** | [`psutil`](https://psutil.readthedocs.io/) | Windows / Linux / macOS / BSD… |
| **串口** | [`pyserial`](https://pyserial.readthedocs.io/) | 同上 |
| **GPU 占用(可选)** | [`nvidia-ml-py`](https://pypi.org/project/nvidia-ml-py/)(NVML)→ `nvidia-smi` → `amd-smi`/`rocm-smi` | NVIDIA 最稳;AMD 有驱动/CLI 时可用;没有则 `gpu%=-1`(名称仍尽量读) |
入口:**[`host/send_sysinfo.py`](host/send_sysinfo.py)**(推荐)。Windows 也可 `.\host\send_sysinfo.ps1`(内部仍调 Python)。
```powershell
cd host
pip install -r requirements.txt
pip install nvidia-ml-py # 有 NVIDIA 时建议装
python send_sysinfo.py # 自动选串口
python send_sysinfo.py -p COM5 # Windows
python send_sysinfo.py -p /dev/ttyUSB0 # Linux
```
## 刷写
@@ -52,13 +71,14 @@ cd firmware\cyd # 或 firmware\cyd_35r
## 推送本机信息
板子刷好并插上 USB 后,在 **PC** 上:
板子刷好并插上 USB 后,在主机上:
```powershell
cd host
.\send_sysinfo.ps1
python send_sysinfo.py
# 或
.\send_sysinfo.ps1 -Port COM5 -IntervalMs 1000
python send_sysinfo.py -p COM5 -i 1000
.\send_sysinfo.ps1 -Port COM5 # Windows 快捷入口
```
注意:串口同时只能被一个程序占用;跑推送脚本时不要开 `pio device monitor`。
@@ -68,6 +88,8 @@ cd host
```
include/pins.h, touch.h, host_info.h, ui.h, lv_conf.h
src/main.cpp, touch.cpp, host_info.cpp, ui.cpp
host/send_sysinfo.ps1
host/send_sysinfo.py # 跨平台主入口
host/requirements.txt
host/send_sysinfo.ps1 # Windows 启动器
firmware/
```

7
host/requirements.txt Normal file
View File

@@ -0,0 +1,7 @@
# Cross-platform host agent (Win10/11, Linux, macOS)
psutil>=5.9.0
pyserial>=3.5
# Optional: NVIDIA GPU utilization on Windows/Linux (recommended if you have NVIDIA)
# pip install nvidia-ml-py
# nvidia-ml-py>=12.535.0

View File

@@ -1,10 +1,11 @@
# Push this PC's CPU / GPU / RAM / disk stats to CYD over USB serial.
# Thin Windows launcher for the cross-platform Python agent.
# Prefer: python send_sysinfo.py
# This script only finds Python and forwards args.
#
# Usage:
# .\send_sysinfo.ps1
# .\send_sysinfo.ps1 -Port COM5
# .\send_sysinfo.ps1 -Port COM5 -IntervalMs 1000
#
# Serial I/O: Windows PowerShell 5 (System.IO.Ports) or Python + pyserial.
param(
[string]$Port = "",
@@ -13,261 +14,44 @@ param(
)
$ErrorActionPreference = "Stop"
function Find-CydPort {
$ports = @()
try {
$ports = [System.IO.Ports.SerialPort]::GetPortNames() | Sort-Object
} catch {}
if (-not $ports -or $ports.Count -eq 0) {
try {
$ports = Get-CimInstance Win32_SerialPort -ErrorAction SilentlyContinue |
Select-Object -ExpandProperty DeviceID | Sort-Object
} catch {}
}
if (-not $ports -or $ports.Count -eq 0) {
$key = 'HKLM:\HARDWARE\DEVICEMAP\SERIALCOMM'
if (Test-Path $key) {
$ports = (Get-ItemProperty $key).PSObject.Properties |
Where-Object { $_.Name -notmatch '^PS' } |
ForEach-Object { [string]$_.Value } |
Sort-Object
}
}
if (-not $ports -or $ports.Count -eq 0) {
throw "No serial ports found. Plug in the CYD USB cable."
}
if ($ports.Count -eq 1) { return $ports[0] }
Write-Host "Available ports: $($ports -join ', ')"
$pick = $ports[-1]
Write-Host "Using last port: $pick (override with -Port)"
return $pick
}
function Sanitize([string]$s, [int]$max = 40) {
if (-not $s) { return "" }
$t = ($s -replace '[|=\r\n]', ' ').Trim()
if ($t.Length -gt $max) { $t = $t.Substring(0, $max).Trim() }
return $t
}
function Get-CpuName {
$p = Get-CimInstance Win32_Processor -ErrorAction SilentlyContinue | Select-Object -First 1
if (-not $p) { return "CPU" }
$n = [string]$p.Name
$n = $n -replace '\(R\)|\(TM\)|CPU @.*|with Radeon.*', ''
$n = ($n -replace '\s+', ' ').Trim()
return Sanitize $n 44
}
function Get-CpuPct {
try {
$c = Get-CimInstance Win32_Processor -ErrorAction Stop
$vals = @($c | ForEach-Object { [double]$_.LoadPercentage })
if ($vals.Count -eq 0) { return 0.0 }
return [math]::Round(($vals | Measure-Object -Average).Average, 1)
} catch {
return 0.0
}
}
function Get-GpuName {
$gpus = Get-CimInstance Win32_VideoController -ErrorAction SilentlyContinue |
Where-Object {
$_.Name -and
$_.Name -notmatch 'Microsoft Basic|Remote Desktop|Oray|IddDriver|Virtual|Parsec|TeamViewer|VNC|Indirect'
}
if (-not $gpus) { return "GPU" }
$pick = $gpus | Where-Object { $_.Name -match 'NVIDIA|GeForce|RTX|GTX|Radeon|AMD' } |
Select-Object -First 1
if (-not $pick) { $pick = $gpus | Select-Object -First 1 }
$n = [string]$pick.Name
$n = $n -replace 'NVIDIA GeForce ', ''
$n = $n -replace 'AMD ', ''
return Sanitize $n 44
}
function Get-GpuPct {
$smi = Get-Command nvidia-smi -ErrorAction SilentlyContinue
if ($smi) {
try {
$out = & nvidia-smi --query-gpu=utilization.gpu --format=csv,noheader,nounits 2>$null
if ($LASTEXITCODE -eq 0 -and $out) {
$first = ([string]($out | Select-Object -First 1)).Trim()
if ($first -match '^\d+') { return [double]$first }
}
} catch {}
}
try {
$counters = Get-Counter '\GPU Engine(*)\Utilization Percentage' -ErrorAction Stop
$samples = $counters.CounterSamples |
Where-Object { $_.CookedValue -ge 0 -and $_.InstanceName -match 'engtype_3D|engtype_Graphics' }
if ($samples) {
$sum = ($samples | Measure-Object -Property CookedValue -Sum).Sum
if ($sum -gt 100) { $sum = 100 }
return [math]::Round($sum, 1)
}
} catch {}
return -1.0
}
function Get-RamGb {
$os = Get-CimInstance Win32_OperatingSystem -ErrorAction Stop
$total = [math]::Round($os.TotalVisibleMemorySize / 1MB, 1)
$free = [math]::Round($os.FreePhysicalMemory / 1MB, 1)
$used = [math]::Round($total - $free, 1)
return @{ Used = $used; Total = $total }
}
function Get-DiskGb {
$d = Get-CimInstance Win32_LogicalDisk -Filter "DeviceID='C:'" -ErrorAction SilentlyContinue
if (-not $d) {
$d = Get-CimInstance Win32_LogicalDisk -ErrorAction SilentlyContinue |
Where-Object { $_.DriveType -eq 3 -and $_.Size -gt 0 } |
Select-Object -First 1
}
if (-not $d -or -not $d.Size) {
return @{ Used = 0.0; Total = 0.0 }
}
$total = [math]::Round($d.Size / 1GB, 1)
$free = [math]::Round($d.FreeSpace / 1GB, 1)
$used = [math]::Round($total - $free, 1)
return @{ Used = $used; Total = $total }
}
function Build-Line {
$hostName = Sanitize $env:COMPUTERNAME 36
$cpu = Get-CpuName
$cpuPct = Get-CpuPct
$gpu = Get-GpuName
$gpuPct = Get-GpuPct
$ram = Get-RamGb
$disk = Get-DiskGb
$gpuField = if ($gpuPct -lt 0) { "gpu%=-1" } else { "gpu%=$gpuPct" }
return "SI|host=$hostName|cpu=$cpu|cpu%=$cpuPct|gpu=$gpu|$gpuField|ram=$($ram.Used)/$($ram.Total)|disk=$($disk.Used)/$($disk.Total)"
}
$here = Split-Path -Parent $MyInvocation.MyCommand.Path
$pyScript = Join-Path $here "send_sysinfo.py"
function Find-Python {
foreach ($c in @(
"python",
"py",
(Join-Path $env:USERPROFILE "AppData\Local\Programs\Python\Python312\python.exe"),
(Join-Path $env:USERPROFILE "AppData\Local\Programs\Python\Python311\python.exe")
)) {
if (-not $c) { continue }
if ($c -eq "py") {
$cmd = Get-Command py -ErrorAction SilentlyContinue
if ($cmd) { return @($cmd.Source, "-3") }
continue
}
$cmd = Get-Command $c -ErrorAction SilentlyContinue
if ($cmd) { return $cmd.Source }
if (Test-Path $c) { return $c }
if ($cmd) { return @($cmd.Source) }
if (Test-Path $c) { return @($c) }
}
return $null
}
function Test-DotNetSerial {
try {
[void][System.IO.Ports.SerialPort]
return $true
} catch {
try {
Add-Type -AssemblyName System.IO.Ports -ErrorAction Stop | Out-Null
return $true
} catch {
return $false
}
}
}
if (-not $Port) { $Port = Find-CydPort }
Write-Host "Sending host info -> $Port @ $Baud every ${IntervalMs}ms (Ctrl+C to stop)"
if (Test-DotNetSerial) {
$sp = [System.IO.Ports.SerialPort]::new($Port, $Baud)
$sp.ReadTimeout = 200
$sp.WriteTimeout = 2000
$sp.NewLine = "`n"
$sp.DtrEnable = $true
$sp.RtsEnable = $true
try {
$sp.Open()
Start-Sleep -Milliseconds 400
while ($true) {
$line = Build-Line
$sp.WriteLine($line)
Write-Host $line
Start-Sleep -Milliseconds $IntervalMs
}
} finally {
if ($sp.IsOpen) { $sp.Close() }
$sp.Dispose()
}
exit 0
}
$py = Find-Python
if (-not $py) {
throw "No System.IO.Ports (PS7) and no python. Use Windows PowerShell 5, or: pip install pyserial"
throw "Python not found. Install Python 3.8+ then: pip install -r requirements.txt"
}
$pyCheck = & $py -c "import serial; print('ok')" 2>$null
if ($LASTEXITCODE -ne 0 -or "$pyCheck" -notmatch 'ok') {
throw "Python found but pyserial missing. Run: `"$py`" -m pip install pyserial"
& $py[0] @($py[1..($py.Length-1)]) -c "import psutil,serial" 2>$null
if ($LASTEXITCODE -ne 0) {
Write-Host "Installing host deps..."
& $py[0] @($py[1..($py.Length-1)]) -m pip install -r (Join-Path $here "requirements.txt")
& $py[0] @($py[1..($py.Length-1)]) -m pip install nvidia-ml-py
}
$writer = Join-Path $PSScriptRoot "_serial_writer.py"
@'
import sys, serial, time
port, baud = sys.argv[1], int(sys.argv[2])
# Keep DTR/RTS stable so ESP32 is not reset each write
ser = serial.Serial()
ser.port = port
ser.baudrate = baud
ser.timeout = 0.2
ser.dtr = False
ser.rts = False
ser.open()
time.sleep(0.3)
try:
for line in sys.stdin:
line = line.rstrip("\r\n")
if not line:
continue
ser.write((line + "\n").encode("utf-8", errors="ignore"))
ser.flush()
print(line, flush=True)
finally:
ser.close()
'@ | Set-Content -Encoding ascii $writer
$argv = @($pyScript, "--baud", "$Baud", "--interval-ms", "$IntervalMs")
if ($Port) { $argv += @("--port", $Port) }
$psi = New-Object System.Diagnostics.ProcessStartInfo
$psi.FileName = $py
$psi.Arguments = "`"$writer`" $Port $Baud"
$psi.UseShellExecute = $false
$psi.RedirectStandardInput = $true
$psi.RedirectStandardOutput = $true
$psi.RedirectStandardError = $true
$psi.CreateNoWindow = $true
$proc = [System.Diagnostics.Process]::Start($psi)
try {
while (-not $proc.HasExited) {
$line = Build-Line
$proc.StandardInput.WriteLine($line)
$proc.StandardInput.Flush()
# Non-blocking-ish read of echo
if (-not $proc.StandardOutput.EndOfStream) {
$echo = $proc.StandardOutput.ReadLine()
if ($echo) { Write-Host $echo }
} else {
Write-Host $line
}
Start-Sleep -Milliseconds $IntervalMs
}
$err = $proc.StandardError.ReadToEnd()
if ($err) { Write-Host $err }
} finally {
try { $proc.StandardInput.Close() } catch {}
if (-not $proc.HasExited) {
Start-Sleep -Milliseconds 200
if (-not $proc.HasExited) { $proc.Kill() }
}
$proc.Dispose()
Remove-Item $writer -ErrorAction SilentlyContinue
}
Write-Host "Launch: $($py -join ' ') $($argv -join ' ')"
& $py[0] @($py[1..($py.Length-1)]) @argv
exit $LASTEXITCODE

434
host/send_sysinfo.py Normal file
View File

@@ -0,0 +1,434 @@
#!/usr/bin/env python3
"""
Cross-platform host agent: push CPU / GPU / RAM / disk to CYD over USB serial.
Works on Windows 10/11, Linux, and macOS (CPU/RAM/disk). GPU utilization is
best-effort via optional vendor backends (NVIDIA NVML first, then CLI fallbacks).
Protocol (one line, ASCII-safe):
SI|host=...|cpu=...|cpu%=..|gpu=...|gpu%=..|ram=used/total|disk=used/total
Deps (required): pip install -r requirements.txt
psutil, pyserial
Optional GPU: pip install nvidia-ml-py
"""
from __future__ import annotations
import argparse
import glob
import os
import platform
import re
import socket
import subprocess
import sys
import time
from typing import List, Optional, Tuple
try:
import psutil
except ImportError:
print("Missing psutil. Run: pip install -r requirements.txt", file=sys.stderr)
raise SystemExit(1)
try:
import serial
from serial.tools import list_ports
except ImportError:
print("Missing pyserial. Run: pip install -r requirements.txt", file=sys.stderr)
raise SystemExit(1)
def sanitize(s: str, max_len: int = 40) -> str:
if not s:
return ""
t = re.sub(r"[|=\r\n]+", " ", str(s)).strip()
t = re.sub(r"\s+", " ", t)
return t[:max_len].strip()
def cpu_name() -> str:
# Linux
if sys.platform.startswith("linux"):
try:
with open("/proc/cpuinfo", "r", encoding="utf-8", errors="ignore") as f:
for line in f:
if line.lower().startswith("model name") or line.lower().startswith(
"hardware"
):
name = line.split(":", 1)[1].strip()
name = re.sub(r"\(R\)|\(TM\)|CPU @.*|with Radeon.*", "", name)
return sanitize(name, 44) or "CPU"
except OSError:
pass
# macOS
if sys.platform == "darwin":
try:
out = subprocess.check_output(
["sysctl", "-n", "machdep.cpu.brand_string"],
stderr=subprocess.DEVNULL,
text=True,
timeout=2,
).strip()
if out:
return sanitize(out, 44)
except (OSError, subprocess.SubprocessError):
pass
# Windows — registry (no admin)
if sys.platform == "win32":
try:
import winreg
key = winreg.OpenKey(
winreg.HKEY_LOCAL_MACHINE,
r"HARDWARE\DESCRIPTION\System\CentralProcessor\0",
)
name, _ = winreg.QueryValueEx(key, "ProcessorNameString")
winreg.CloseKey(key)
name = re.sub(r"\(R\)|\(TM\)|CPU @.*|with Radeon.*", "", str(name))
return sanitize(name, 44) or "CPU"
except OSError:
pass
# Fallback
return sanitize(platform.processor() or "CPU", 44)
def cpu_pct() -> float:
# Non-blocking after first call; warm up once at start.
return round(float(psutil.cpu_percent(interval=None)), 1)
def ram_gb() -> Tuple[float, float]:
vm = psutil.virtual_memory()
total = round(vm.total / (1024**3), 1)
used = round(vm.used / (1024**3), 1)
return used, total
def disk_root() -> str:
if sys.platform == "win32":
return os.environ.get("SystemDrive", "C:") + "\\"
return "/"
def disk_gb() -> Tuple[float, float]:
root = disk_root()
try:
u = psutil.disk_usage(root)
except OSError:
# Last resort: first mount with size
for p in psutil.disk_partitions(all=False):
try:
u = psutil.disk_usage(p.mountpoint)
if u.total > 0:
break
except OSError:
continue
else:
return 0.0, 0.0
total = round(u.total / (1024**3), 1)
used = round(u.used / (1024**3), 1)
return used, total
_VIRTUAL_GPU_RE = re.compile(
r"Microsoft Basic|Remote Desktop|Oray|IddDriver|Virtual|Parsec|"
r"TeamViewer|VNC|Indirect|QXL|VMware|VirtualBox|Hyper-V|llvmpipe|SVGA",
re.I,
)
def _is_real_gpu_name(name: str) -> bool:
return bool(name) and not _VIRTUAL_GPU_RE.search(name)
# --- GPU: optional layered backends ---
def _gpu_nvidia_nvml() -> Optional[Tuple[str, float]]:
try:
import pynvml # provided by nvidia-ml-py
except ImportError:
return None
try:
pynvml.nvmlInit()
try:
count = pynvml.nvmlDeviceGetCount()
if count < 1:
return None
# Prefer highest utilization discrete card
best = None
for i in range(count):
h = pynvml.nvmlDeviceGetHandleByIndex(i)
raw = pynvml.nvmlDeviceGetName(h)
name = raw.decode("utf-8", "ignore") if isinstance(raw, bytes) else str(raw)
name = name.replace("NVIDIA GeForce ", "").replace("NVIDIA ", "")
util = pynvml.nvmlDeviceGetUtilizationRates(h).gpu
cand = (sanitize(name, 44), float(util))
if best is None or cand[1] > best[1]:
best = cand
return best
finally:
try:
pynvml.nvmlShutdown()
except Exception:
pass
except Exception:
return None
def _gpu_nvidia_smi() -> Optional[Tuple[str, float]]:
try:
out = subprocess.check_output(
[
"nvidia-smi",
"--query-gpu=name,utilization.gpu",
"--format=csv,noheader,nounits",
],
stderr=subprocess.DEVNULL,
text=True,
timeout=3,
)
except (OSError, subprocess.SubprocessError):
return None
best = None
for line in out.splitlines():
parts = [p.strip() for p in line.split(",")]
if len(parts) < 2:
continue
name = parts[0].replace("NVIDIA GeForce ", "").replace("NVIDIA ", "")
try:
util = float(parts[1])
except ValueError:
continue
cand = (sanitize(name, 44), util)
if best is None or cand[1] > best[1]:
best = cand
return best
def _gpu_amd_cli() -> Optional[Tuple[str, float]]:
# Linux ROCm / Windows amd-smi when present
for cmd in (
["amd-smi", "metric", "--json"],
["rocm-smi", "--showuse", "--showproductname", "--csv"],
["rocm-smi", "-u", "--showproductname"],
):
try:
out = subprocess.check_output(
cmd, stderr=subprocess.DEVNULL, text=True, timeout=4
)
except (OSError, subprocess.SubprocessError):
continue
# Heuristic parse — keep resilient across CLI churn
name = "AMD GPU"
util = -1.0
m = re.search(r"Card series:\s*(.+)", out, re.I)
if m:
name = m.group(1).strip()
m = re.search(r"GPU\[0\].*?=\s*(\d+)", out)
if not m:
m = re.search(r"GPU use \(%\):\s*(\d+)", out, re.I)
if not m:
m = re.search(r"\"gpu_activity\"\s*:\s*\{\s*\"value\"\s*:\s*(\d+)", out)
if m:
util = float(m.group(1))
if util >= 0:
return sanitize(name, 44), util
return None
def _gpu_name_windows() -> str:
try:
import winreg
base = r"SYSTEM\CurrentControlSet\Control\Class\{4d36e968-e325-11ce-bfc1-08002be10318}"
root = winreg.OpenKey(winreg.HKEY_LOCAL_MACHINE, base)
best = ""
i = 0
while True:
try:
sub = winreg.EnumKey(root, i)
except OSError:
break
i += 1
if not re.fullmatch(r"\d{4}", sub):
continue
try:
k = winreg.OpenKey(root, sub)
name, _ = winreg.QueryValueEx(k, "DriverDesc")
winreg.CloseKey(k)
except OSError:
continue
name = str(name)
if not _is_real_gpu_name(name):
continue
if re.search(r"NVIDIA|GeForce|RTX|GTX|Radeon|AMD|Intel Arc", name, re.I):
best = name
break
if not best:
best = name
winreg.CloseKey(root)
best = best.replace("NVIDIA GeForce ", "").replace("NVIDIA ", "").replace("AMD ", "")
return sanitize(best, 44)
except OSError:
return ""
def _gpu_name_linux() -> str:
# lspci is widely available
try:
out = subprocess.check_output(
["lspci"], stderr=subprocess.DEVNULL, text=True, timeout=3
)
except (OSError, subprocess.SubprocessError):
# DRM card names
for path in sorted(glob.glob("/sys/class/drm/card*/device/vendor")):
try:
card = path.rsplit("/device/", 1)[0]
# uevent MODALIAS / device product via lspci preferred
_ = card
except OSError:
pass
return ""
for line in out.splitlines():
if "VGA" in line or "3D" in line or "Display" in line:
name = line.split(":", 2)[-1].strip() if ":" in line else line
name = re.sub(r"^.*?(VGA|3D|Display) compatible controller:\s*", "", line)
if _is_real_gpu_name(name):
name = name.replace("NVIDIA Corporation ", "").replace(
"Advanced Micro Devices, Inc. [AMD/ATI] ", ""
)
return sanitize(name, 44)
return ""
def _gpu_name_macos() -> str:
try:
out = subprocess.check_output(
["system_profiler", "SPDisplaysDataType"],
stderr=subprocess.DEVNULL,
text=True,
timeout=5,
)
except (OSError, subprocess.SubprocessError):
return "Apple GPU"
m = re.search(r"Chipset Model:\s*(.+)", out)
if m:
return sanitize(m.group(1).strip(), 44)
return "Apple GPU"
def gpu_info() -> Tuple[str, float]:
"""Return (name, utilization%) — util -1 if unknown."""
for fn in (_gpu_nvidia_nvml, _gpu_nvidia_smi, _gpu_amd_cli):
got = fn()
if got:
return got
# Name-only fallbacks
name = ""
if sys.platform == "win32":
name = _gpu_name_windows()
elif sys.platform.startswith("linux"):
name = _gpu_name_linux()
elif sys.platform == "darwin":
name = _gpu_name_macos()
return (name or "GPU", -1.0)
def build_line() -> str:
host = sanitize(socket.gethostname(), 36)
cpu = cpu_name()
c_pct = cpu_pct()
gpu, g_pct = gpu_info()
ram_u, ram_t = ram_gb()
disk_u, disk_t = disk_gb()
gpu_field = f"gpu%={g_pct}" if g_pct >= 0 else "gpu%=-1"
return (
f"SI|host={host}|cpu={cpu}|cpu%={c_pct}|gpu={gpu}|"
f"{gpu_field}|ram={ram_u}/{ram_t}|disk={disk_u}/{disk_t}"
)
def find_ports() -> List[str]:
ports = [p.device for p in list_ports.comports()]
if ports:
return sorted(ports)
# Extra Linux globs
if sys.platform.startswith("linux"):
ports = sorted(
glob.glob("/dev/ttyUSB*")
+ glob.glob("/dev/ttyACM*")
+ glob.glob("/dev/tty.usb*")
)
return ports
def pick_port(explicit: str) -> str:
if explicit:
return explicit
ports = find_ports()
if not ports:
raise SystemExit("No serial ports found. Plug in the CYD USB cable.")
if len(ports) == 1:
return ports[0]
print("Available ports:", ", ".join(ports))
pick = ports[-1]
print(f"Using last port: {pick} (override with --port)")
return pick
def open_serial(port: str, baud: int):
ser = serial.Serial()
ser.port = port
ser.baudrate = baud
ser.timeout = 0.2
ser.write_timeout = 2
# Avoid toggling DTR/RTS reset on open when possible
ser.dtr = False
ser.rts = False
ser.open()
return ser
def main() -> int:
ap = argparse.ArgumentParser(description="Push host sysinfo to CYD over USB serial")
ap.add_argument("--port", "-p", default="", help="Serial port (COMx / /dev/ttyUSB0)")
ap.add_argument("--baud", "-b", type=int, default=460800)
ap.add_argument("--interval-ms", "-i", type=int, default=1000)
ap.add_argument("--once", action="store_true", help="Send one line and exit")
args = ap.parse_args()
# Warm CPU percent baseline
psutil.cpu_percent(interval=None)
port = pick_port(args.port)
print(
f"Sending host info -> {port} @ {args.baud} every {args.interval_ms}ms "
f"({platform.system()} {platform.release()}; Ctrl+C to stop)"
)
ser = open_serial(port, args.baud)
time.sleep(0.35)
try:
while True:
line = build_line()
ser.write((line + "\n").encode("utf-8", errors="ignore"))
ser.flush()
print(line, flush=True)
if args.once:
break
time.sleep(max(0.2, args.interval_ms / 1000.0))
except KeyboardInterrupt:
print("\nStopped.")
finally:
ser.close()
return 0
if __name__ == "__main__":
raise SystemExit(main())