Android-pyApi/android_pyapi/android/hardware_thermal.py

"""
Android hardware.thermal 组件 — ThermalManager, CoolingDevice, Temperature 等
方法签名完全对齐 Android SDK
"""

from __future__ import annotations

import random
import threading
from typing import Any, Callable, Dict, List, Optional

from .._core import Stub, AndroidOnlyError, HAS_PSUTIL
from ..java.lang import RuntimeException


IllegalArgumentException = type("IllegalArgumentException", (RuntimeException,), {})


# ═══════════════════════════════════════════════════════════════
# Temperature
# ═══════════════════════════════════════════════════════════════

class Temperature(Stub):
    """Temperature — 温度数据"""

    TYPE_SKIN = 0
    TYPE_CPU = 1
    TYPE_GPU = 2
    TYPE_BATTERY = 3
    TYPE_NPU = 4
    TYPE_POWER_AMP = 5
    TYPE_MODEM = 6
    TYPE_AMBIENT = 7
    TYPE_USB_PORT = 8
    TYPE_DISPLAY = 9
    TYPE_BELT = 10
    TYPE_CAMERA = 11
    TYPE_TPU = 12
    TYPE_POGO = 13

    def __init__(self, type_: int = TYPE_AMBIENT, name: str = "ambient",
                 value: float = 25.0, status: int = 0):
        super().__init__()
        self._type = type_
        self._name = name
        self._value = value
        self._status = status

    def getType(self) -> int:
        return self._type

    def getName(self) -> str:
        return self._name

    def getValue(self) -> float:
        return self._value

    def getStatus(self) -> int:
        return self._status

    def toString(self) -> str:
        return f"Temperature{{type={self._type}, name={self._name}, value={self._value}, status={self._status}}}"


# ═══════════════════════════════════════════════════════════════
# CoolingDevice
# ═══════════════════════════════════════════════════════════════

class CoolingDevice(Stub):
    """CoolingDevice — 冷却设备"""

    def __init__(self, type_: int = 0, description: str = "cooling_fan",
                 current_level: int = 50):
        super().__init__()
        self._type = type_
        self._description = description
        self._current_level = current_level

    def getType(self) -> int:
        return self._type

    def getDescription(self) -> str:
        return self._description

    def getCurrentLevel(self) -> int:
        return self._current_level

    def toString(self) -> str:
        return f"CoolingDevice{{type={self._type}, description={self._description}, level={self._current_level}}}"


# ═══════════════════════════════════════════════════════════════
# ThermalStatus
# ═══════════════════════════════════════════════════════════════

class ThermalStatus(Stub):
    """ThermalStatus — 热状态"""

    def __init__(self):
        super().__init__()
        self._status: int = ThermalManager.THERMAL_STATUS_NONE
        self._temperatures: List[Temperature] = []

    def getStatus(self) -> int:
        return self._status

    def getType(self) -> int:
        return 0

    def getName(self) -> str:
        return "thermal_status"

    def getTemperatures(self) -> List[Temperature]:
        return list(self._temperatures)

    def toString(self) -> str:
        return f"ThermalStatus{{status={self._status}}}"


# ═══════════════════════════════════════════════════════════════
# ThermalManager
# ═══════════════════════════════════════════════════════════════

class ThermalManager(Stub):
    """ThermalManager — 热管理器"""

    THERMAL_STATUS_NONE = 0
    THERMAL_STATUS_LIGHT = 1
    THERMAL_STATUS_MODERATE = 2
    THERMAL_STATUS_SEVERE = 3
    THERMAL_STATUS_CRITICAL = 4
    THERMAL_STATUS_EMERGENCY = 5
    THERMAL_STATUS_SHUTDOWN = 6

    def __init__(self):
        super().__init__()
        self._lock = threading.Lock()
        self._listeners: List[Callable[[int], None]] = []
        self._thermal_status = self.THERMAL_STATUS_NONE
        self._status_threshold = self.THERMAL_STATUS_CRITICAL

        # Simulated temperatures
        self._temperatures: List[Temperature] = [
            Temperature(Temperature.TYPE_CPU, "cpu0", 45.0, self.THERMAL_STATUS_NONE),
            Temperature(Temperature.TYPE_GPU, "gpu0", 40.0, self.THERMAL_STATUS_NONE),
            Temperature(Temperature.TYPE_BATTERY, "battery", 30.0, self.THERMAL_STATUS_NONE),
            Temperature(Temperature.TYPE_AMBIENT, "ambient", 25.0, self.THERMAL_STATUS_NONE),
            Temperature(Temperature.TYPE_SKIN, "skin", 32.0, self.THERMAL_STATUS_NONE),
            Temperature(Temperature.TYPE_NPU, "npu0", 38.0, self.THERMAL_STATUS_NONE),
            Temperature(Temperature.TYPE_MODEM, "modem", 35.0, self.THERMAL_STATUS_NONE),
            Temperature(Temperature.TYPE_DISPLAY, "display", 36.0, self.THERMAL_STATUS_NONE),
        ]

        # Try to get actual CPU temps via psutil
        if HAS_PSUTIL:
            import psutil
            try:
                temps = psutil.sensors_temperatures()
                if temps:
                    for name, entries in temps.items():
                        for entry in entries:
                            temp_type = getattr(Temperature, f"TYPE_{name.upper()}", Temperature.TYPE_AMBIENT)
                            self._temperatures.append(
                                Temperature(temp_type, entry.label or name,
                                            entry.current, self.THERMAL_STATUS_NONE)
                            )
            except Exception:
                pass

    # ── Thermal headroom ──

    def getThermalHeadroom(self, forecastSeconds: int) -> float:
        """获取热余量百分比 (0.0 = 过温, 1.0 = 安全)"""
        max_temp = max(t.getValue() for t in self._temperatures) if self._temperatures else 25.0
        headroom = max(0.0, 1.0 - (max_temp - 25.0) / 75.0)
        return min(1.0, headroom)

    # ── Status ──

    def getThermalStatus(self) -> int:
        return self._thermal_status

    def getThermalStatusThreshold(self) -> int:
        return self._status_threshold

    # ── Listeners ──

    def addThermalStatusListener(self, listener: Callable[[int], None]):
        with self._lock:
            if listener not in self._listeners:
                self._listeners.append(listener)

    def removeThermalStatusListener(self, listener: Callable[[int], None]):
        with self._lock:
            if listener in self._listeners:
                self._listeners.remove(listener)

    def _notify_listeners(self, status: int):
        with self._lock:
            for listener in self._listeners:
                listener(status)

    # ── Helper ──

    def toString(self) -> str:
        return f"ThermalManager{{status={self._thermal_status}}}"


# ═══════════════════════════════════════════════════════════════
# __all__
# ═══════════════════════════════════════════════════════════════

__all__ = [
    "ThermalManager",
    "CoolingDevice",
    "ThermalStatus",
    "Temperature",
]