coretemp_shared.py

"""
coretemp_shared.py - lettore shared memory di CoreTemp (Windows).

CoreTemp espone un oggetto di shared memory chiamato `CoreTempMappingObjectEx`
con la struttura documentata sul sito ufficiale (alcpu.com/CoreTemp).
Fornisce temperatura per core e frequenza CPU istantanea.

Uso:
    r = read()
    if r:
        print(r["cpu_temp_c"], r["cpu_speed_mhz"])
"""

from __future__ import annotations

import ctypes
import mmap
from typing import Optional


class _CoreTempSharedData(ctypes.Structure):
    _pack_ = 1
    _fields_ = [
        ("uiLoad",         ctypes.c_uint   * 256),  # 1024
        ("uiTjMax",        ctypes.c_uint   * 128),  # 512
        ("uiCoreCnt",      ctypes.c_uint),          # 4
        ("uiCPUCnt",       ctypes.c_uint),          # 4
        ("fTemp",          ctypes.c_float  * 256),  # 1024
        ("fVID",           ctypes.c_float),         # 4
        ("fCPUSpeed",      ctypes.c_float),         # 4   (MHz, real-time)
        ("fFSBSpeed",      ctypes.c_float),         # 4
        ("fMultiplier",    ctypes.c_float),         # 4
        ("sCPUName",       ctypes.c_char   * 100),  # 100
        ("ucFahrenheit",   ctypes.c_ubyte),         # 1
        ("ucDeltaToTjMax", ctypes.c_ubyte),         # 1
    ]


_SIZE = ctypes.sizeof(_CoreTempSharedData)
_NAME = "CoreTempMappingObjectEx"
_mm: Optional[mmap.mmap] = None


def _open() -> bool:
    global _mm
    try:
        _mm = mmap.mmap(-1, _SIZE, _NAME, access=mmap.ACCESS_READ)
        return True
    except (OSError, ValueError):
        _mm = None
        return False


def close() -> None:
    global _mm
    if _mm is not None:
        try: _mm.close()
        except Exception: pass
        _mm = None


def read() -> Optional[dict]:
    """Ritorna {"cpu_temp_c": int, "cpu_speed_mhz": int} o None se CoreTemp non gira."""
    global _mm
    if _mm is None and not _open():
        return None
    try:
        _mm.seek(0)
        buf = _mm.read(_SIZE)
        d = _CoreTempSharedData()
        ctypes.memmove(ctypes.addressof(d), buf, len(buf))

        if d.uiCoreCnt == 0 or d.uiCoreCnt > 256:
            return None

        n = min(int(d.uiCoreCnt) * max(int(d.uiCPUCnt), 1), 256)
        if n <= 0:
            return None

        if d.ucDeltaToTjMax:
            tjmax = d.uiTjMax[0] if d.uiTjMax[0] else 100
            temps = [tjmax - d.fTemp[i] for i in range(n)]
        else:
            temps = [d.fTemp[i] for i in range(n)]

        max_t = max(temps)
        if d.ucFahrenheit:
            max_t = (max_t - 32.0) * 5.0 / 9.0

        speed = float(d.fCPUSpeed)
        if speed <= 0:
            speed = float(d.fFSBSpeed) * float(d.fMultiplier)

        return {
            "cpu_temp_c":   int(round(max_t)),
            "cpu_speed_mhz": int(round(speed)),
        }
    except Exception:
        # mappa potrebbe essere stata chiusa (CoreTemp uscito)
        close()
        return None