Imrayya
0b4f739d12
- FastAPI server with audio mixer (Windows + Voicemeeter), media tracking, notifications (progress + alerts), telemetry, and system tray .exe build - Test suite covering notifications, audio mixer, media, telemetry, and API - E-Ink web client (vanilla JS, DOM API, block meters, swipe gestures) - Android WebView APK for BOOX Go 7 Color Gen II (Android 13, Kotlin) - Design decision records in .pi/docs/design/ - PyInstaller build script for server .exe
137 lines
4.6 KiB
Python
137 lines
4.6 KiB
Python
"""Hardware telemetry via HWiNFO64 Shared Memory.
|
|
|
|
Reads CPU/GPU core temperatures, VRAM usage, and other hardware metrics
|
|
from HWiNFO's shared memory interface.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import logging
|
|
from dataclasses import dataclass, field
|
|
from typing import Any
|
|
|
|
logger = logging.getLogger("paperdash.telemetry")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Data Models
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@dataclass
|
|
class SensorReading:
|
|
"""A single hardware sensor reading."""
|
|
|
|
name: str
|
|
value: float
|
|
unit: str
|
|
category: str = "" # e.g., "CPU", "GPU", "System"
|
|
|
|
|
|
@dataclass
|
|
class TelemetryState:
|
|
"""Aggregated hardware telemetry state."""
|
|
|
|
sensors: list[SensorReading] = field(default_factory=list)
|
|
cpu_temp: float = 0.0
|
|
gpu_temp: float = 0.0
|
|
cpu_usage: float = 0.0
|
|
gpu_usage: float = 0.0
|
|
vram_usage: float = 0.0
|
|
vram_total: float = 0.0
|
|
ram_usage: float = 0.0
|
|
ram_total: float = 0.0
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Telemetry Manager
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TelemetryManager:
|
|
"""Reads hardware telemetry from HWiNFO64 Shared Memory."""
|
|
|
|
def __init__(self) -> None:
|
|
self._state = TelemetryState()
|
|
self._poll_interval: float = 5.0
|
|
|
|
async def run(self) -> None:
|
|
"""Periodically poll hardware telemetry."""
|
|
while True:
|
|
try:
|
|
self._state = await self._poll_hwinfo()
|
|
except Exception as e:
|
|
logger.error(f"Telemetry poll error: {e}")
|
|
await asyncio.sleep(self._poll_interval)
|
|
|
|
async def _poll_hwinfo(self) -> TelemetryState:
|
|
"""Read hardware state from HWiNFO64 Shared Memory."""
|
|
try:
|
|
import pyhwinfo # type: ignore
|
|
|
|
data = pyhwinfo.read_all()
|
|
sensors = []
|
|
state = TelemetryState()
|
|
|
|
for sensor in data:
|
|
reading = SensorReading(
|
|
name=sensor.get("name", ""),
|
|
value=sensor.get("value", 0.0),
|
|
unit=sensor.get("unit", ""),
|
|
category=sensor.get("category", ""),
|
|
)
|
|
sensors.append(reading)
|
|
|
|
# Extract key metrics
|
|
name_lower = reading.name.lower()
|
|
if "cpu core" in name_lower and "temp" in name_lower:
|
|
state.cpu_temp = max(state.cpu_temp, reading.value)
|
|
elif "gpu" in name_lower and "temp" in name_lower:
|
|
state.gpu_temp = max(state.gpu_temp, reading.value)
|
|
elif "cpu" in name_lower and "usage" in name_lower:
|
|
state.cpu_usage = max(state.cpu_usage, reading.value)
|
|
elif "gpu" in name_lower and "usage" in name_lower:
|
|
state.gpu_usage = max(state.gpu_usage, reading.value)
|
|
elif "dedicated" in name_lower and "memory" in name_lower:
|
|
if "used" in name_lower:
|
|
state.vram_usage = reading.value
|
|
elif "total" in name_lower:
|
|
state.vram_total = reading.value
|
|
elif "memory" in name_lower and "usage" in name_lower:
|
|
state.ram_usage = reading.value
|
|
elif "memory" in name_lower and "total" in name_lower:
|
|
state.ram_total = reading.value
|
|
|
|
state.sensors = sensors
|
|
return state
|
|
|
|
except ImportError:
|
|
logger.warning("pyhwinfo not installed — telemetry unavailable")
|
|
return TelemetryState()
|
|
except Exception as e:
|
|
logger.error(f"HWiNFO read error: {e}")
|
|
return TelemetryState()
|
|
|
|
def get_state(self) -> dict[str, Any]:
|
|
"""Return current telemetry state as a serializable dict."""
|
|
return {
|
|
"cpu_temp": self._state.cpu_temp,
|
|
"gpu_temp": self._state.gpu_temp,
|
|
"cpu_usage": self._state.cpu_usage,
|
|
"gpu_usage": self._state.gpu_usage,
|
|
"vram_usage": self._state.vram_usage,
|
|
"vram_total": self._state.vram_total,
|
|
"ram_usage": self._state.ram_usage,
|
|
"ram_total": self._state.ram_total,
|
|
"sensors": [
|
|
{
|
|
"name": s.name,
|
|
"value": s.value,
|
|
"unit": s.unit,
|
|
"category": s.category,
|
|
}
|
|
for s in self._state.sensors
|
|
],
|
|
}
|