feat: initial project scaffold — server, client, Android APK
- 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
This commit is contained in:
@@ -0,0 +1,136 @@
|
||||
"""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
|
||||
],
|
||||
}
|
||||
Reference in New Issue
Block a user