diff --git a/pyproject.toml b/pyproject.toml index bbbca6a58..f28944fe5 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -55,6 +55,7 @@ bootc = "hhd.plugins.bootc:autodetect" debug = "hhd.plugins.debug:autodetect" power = "hhd.plugins.power:autodetect" customization = "hhd.plugins.customization:autodetect" +cooling_dock = "hhd.plugins.cooling_dock:autodetect" # display = "hhd.plugins.display:autodetect" [project.entry-points."hhd.i18n"] diff --git a/src/adjustor/core/const.py b/src/adjustor/core/const.py index 0984eba60..02e398d1f 100644 --- a/src/adjustor/core/const.py +++ b/src/adjustor/core/const.py @@ -81,6 +81,11 @@ class DeviceTDPv2(TypedDict): ("balanced", ["balanced"], 30, 50), ("performance", ["performance"], 50, 90), ] +ENERGY_MAP_OXP_SUPERX = [ + ("power", ["low-power", "quiet"], 0, 15), + ("balanced", ["balanced"], 30, 50), + ("performance", ["performance"], 50, 75), +] ALIB_PARAMS = { # TDPs @@ -171,6 +176,30 @@ class DeviceTDPv2(TypedDict): "temp_target": D(60, 70, 85, 90, 100), } +DEV_PARAMS_OXP_SUPERX: dict[str, DeviceParams] = { + # smax = enforced cap (no dock): 75W stapm/skin, 95W fast + # dmax = unlocked cap (dock running): 120W (ALIB max) + "stapm_limit": D(0, 4, 25, 75, 120), + "skin_limit": D(0, 4, 25, 75, 120), + "slow_limit": D(0, 4, 27, 80, 120), + "fast_limit": D(0, 4, 40, 95, 120), + # Times + "slow_time": D(5, 5, 10, 10, 10), + "stapm_time": D(100, 100, 100, 200, 200), + # Temp + "temp_target": D(60, 70, 85, 90, 100), +} + +# Battery-mode TDP caps for SUPER X / APEX. +# OneXConsole: battery maxTdp=55, maxBoostTdp=70. These override smax when on +# battery power. On AC (wall power), the standard smax (75/95) applies instead. +DC_CAP_OXP_SUPERX: dict[str, int] = { + "stapm_limit": 55, + "skin_limit": 55, + "slow_limit": 57, + "fast_limit": 70, +} + DEV_PARAMS_WIN5: dict[str, DeviceParams] = { "stapm_limit": D(0, 4, 25, 85, 100), "skin_limit": D(0, 4, 25, 85, 100), @@ -211,6 +240,19 @@ class DeviceTDPv2(TypedDict): "G1619-04": (DEV_PARAMS_28W, ALIB_PARAMS_7040, False, ENERGY_MAP), "G1619-05": (DEV_PARAMS_28W, ALIB_PARAMS_7040, False, ENERGY_MAP), "G1618-05": (DEV_PARAMS_WIN5, ALIB_PARAMS_AIMAX, False, ENERGY_MAP_AIMAX), + # ONEXPLAYER SUPER X / APEX: dock-aware TDP (75W enforced, 120W with dock) + "ONEXPLAYER SUPER X": ( + DEV_PARAMS_OXP_SUPERX, + ALIB_PARAMS_AIMAX, + False, + ENERGY_MAP_OXP_SUPERX, + ), + "ONEXPLAYER APEX": ( + DEV_PARAMS_OXP_SUPERX, + ALIB_PARAMS_AIMAX, + False, + ENERGY_MAP_OXP_SUPERX, + ), } CPU_DATA: dict[ diff --git a/src/adjustor/drivers/smu/__init__.py b/src/adjustor/drivers/smu/__init__.py index c00f474cc..90f4f6f08 100644 --- a/src/adjustor/drivers/smu/__init__.py +++ b/src/adjustor/drivers/smu/__init__.py @@ -1,15 +1,15 @@ import logging import time -from threading import Event as TEvent, Lock, Thread -from typing import Sequence - -from hhd.plugins import Context, Event, HHDPlugin, load_relative_yaml -from hhd.plugins.conf import Config +from collections.abc import Sequence +from threading import Event as TEvent +from threading import Lock, Thread from adjustor.core.alib import AlibParams, DeviceParams, alib from adjustor.core.fan import fan_worker, get_fan_info from adjustor.core.platform import get_platform_choices, set_platform_profile from adjustor.i18n import _ +from hhd.plugins import Context, Event, HHDPlugin, load_relative_yaml +from hhd.plugins.conf import Config logger = logging.getLogger(__name__) @@ -47,14 +47,21 @@ def __init__( pp_map: list[tuple[str, list[str], int, int]] | None, pp_enable: bool = True, init_tdp: bool = True, + dock_aware: bool = False, + dc_cap: dict[str, int] | None = None, ) -> None: - self.name = f"adjustor_smu_qam" + self.name = "adjustor_smu_qam" self.priority = 7 self.log = "smuq" self.enabled = False self.initialized = False self.dev = dev self.enforce_limits = True + self.old_enforce = None + self.dock_aware = dock_aware + self.dc_cap = dc_cap + self.on_ac = True # assume AC until determined otherwise + self.old_on_ac = None self.emit = None self.old_conf = None self.startup = True @@ -85,11 +92,22 @@ def __init__( self.pps = get_platform_choices() or [] if not self.pps: logger.warning( - f"Platform profile map was provided but device does not have platform profiles." + "Platform profile map was provided but device does not have platform profiles." ) else: self.pps = [] + def _effective_smax(self, key: str = "skin_limit") -> int | None: + """Get the effective safe maximum for a parameter, considering power state. + + On battery (on_ac=False) with a dc_cap configured, the battery-mode + cap is used instead of the device's default smax. This prevents + exceeding the device's safe TDP on battery power.""" + base_smax = self.dev[key].smax if key in self.dev else None + if not self.on_ac and self.dc_cap and key in self.dc_cap: + return self.dc_cap[key] + return base_smax + def settings(self): if not self.enabled: self.initialized = False @@ -102,16 +120,17 @@ def settings(self): lims = self.lims assert ( lims - ), f"Device params do not include skin limit or stapm limit to set tdp." + ), "Device params do not include skin limit or stapm limit to set tdp." dmin, smin, default, smax, dmax = lims - if self.enforce_limits: + if not self.enforce_limits: out["tdp"]["qam"]["children"]["tdp"].update( - {"min": smin, "max": smax, "default": default} + {"min": dmin, "max": dmax, "default": default} ) else: + eff_smax = self._effective_smax("skin_limit") or smax out["tdp"]["qam"]["children"]["tdp"].update( - {"min": dmin, "max": dmax, "default": default} + {"min": smin, "max": eff_smax, "default": default} ) if not self.fan_info: @@ -153,10 +172,34 @@ def open( ): self.emit = emit self.fan_info = get_fan_info() + try: + from hhd.utils import get_ac_status, get_ac_status_fn + ac = get_ac_status(get_ac_status_fn()) + if ac is not None: + self.on_ac = ac + except Exception: + pass def update(self, conf: Config): self.enabled = conf["hhd.settings.tdp_ready"].to(bool) - self.enforce_limits = conf["hhd.settings.enforce_limits"].to(bool) + user_enforce = conf["hhd.settings.enforce_limits"].to(bool) + dock_running = conf.get("cooling_dock.dock_running", False) + # Dock-aware devices relax the enforced cap while the cooling dock is + # actively running on AC power (e.g. SUPER X: 75W -> 120W). + # On battery power, limits are always enforced to protect the battery. + self.enforce_limits = user_enforce and not ( + dock_running and self.dock_aware and self.on_ac + ) + dock_changed = self.enforce_limits != self.old_enforce + self.old_enforce = self.enforce_limits + + # Track power state changes to re-clamp TDP on AC/DC transitions. + power_changed = self.on_ac != self.old_on_ac + self.old_on_ac = self.on_ac + + if (dock_changed or power_changed) and self.emit: + self.emit({"type": "settings"}) + if not self.enabled or not self.initialized: self.startup = self.init_tdp return @@ -171,9 +214,9 @@ def update(self, conf: Config): else: new_tdp = conf["tdp.qam.tdp"].to(int) - if self.startup and self.lims: + if (self.startup or dock_changed or power_changed) and self.lims and self.enforce_limits: smin = self.lims.smin - smax = self.lims.smax + smax = self._effective_smax("skin_limit") if smin and new_tdp < smin: logger.warning( @@ -183,7 +226,7 @@ def update(self, conf: Config): conf["tdp.qam.tdp"] = smin if smax and new_tdp > smax: logger.warning( - f"Device TDP ({new_tdp}) too low for startup, adjusting." + f"Device TDP ({new_tdp}W) exceeds safe limit ({smax}W), clamping." ) new_tdp = smax conf["tdp.qam.tdp"] = smax @@ -197,7 +240,7 @@ def update(self, conf: Config): if changed and not sys_tdp: self.sys_tdp = False - if self.startup or changed: + if self.startup or changed or dock_changed or power_changed: self.queued = curr + APPLY_DELAY self.is_set = False @@ -221,11 +264,20 @@ def update(self, conf: Config): if new_boost: try: - fmax = self.dev["fast_limit"].smax - smax = self.dev["stapm_limit"].smax + # Use the unlocked (dmax) limits when the dock is running + # so boost scales correctly up to the ALIB cap. + # On battery, use DC cap for boost if available. + if not self.enforce_limits: + fmax = self.dev["fast_limit"].max + smax = self.dev["stapm_limit"].max + else: + fmax = self._effective_smax("fast_limit") + smax = self._effective_smax("stapm_limit") assert fmax and smax - conf["tdp.smu.std.fast_limit"] = int(new_tdp * (fmax / smax)) + conf["tdp.smu.std.fast_limit"] = int( + round(new_tdp * (fmax / smax)) + ) conf["tdp.smu.std.slow_limit"] = min( new_tdp + 2, conf["tdp.smu.std.fast_limit"].to(int) ) @@ -236,9 +288,14 @@ def update(self, conf: Config): conf["tdp.smu.std.slow_limit"] = new_tdp conf["tdp.smu.std.fast_limit"] = new_tdp - # Show steam message + # Show status message about TDP limits if self.sys_tdp: conf["tdp.qam.sys_tdp"] = _("Steam is controlling TDP") + elif self.enforce_limits and not self.on_ac and self.dc_cap: + eff = self._effective_smax("skin_limit") + conf["tdp.qam.sys_tdp"] = f"TDP limited to {eff}W (on battery)" + elif self.dock_aware and self.enforce_limits and not dock_running: + conf["tdp.qam.sys_tdp"] = f"TDP limited to {self.lims.smax}W (dock not connected)" else: conf["tdp.qam.sys_tdp"] = "" @@ -328,6 +385,17 @@ def notify(self, events: Sequence[Event]): ) self.queued = time.perf_counter() + SLEEP_DELAY + # AC/DC power state changes: re-clamp TDP to the appropriate + # safe limit on the next update() cycle. + if ev["type"] == "acpi" and ev.get("event") in ("ac", "dc"): + new_ac = ev["event"] == "ac" + if new_ac != self.on_ac: + logger.info( + f"Power state changed to {'AC' if new_ac else 'battery'}, " + f"re-evaluating TDP limits." + ) + self.on_ac = new_ac + def close(self): if self.fan_t: self.fan_should_exit.set() @@ -343,13 +411,20 @@ def __init__( dev: dict[str, DeviceParams], cpu: dict[str, AlibParams], platform_profile: bool = True, + dock_aware: bool = False, + dc_cap: dict[str, int] | None = None, ) -> None: - self.name = f"adjustor_smu" + self.name = "adjustor_smu" self.priority = 9 self.log = "asmu" self.enabled = False self.initialized = False self.enforce_limits = True + self.old_enforce = True + self.dock_aware = dock_aware + self.dc_cap = dc_cap + self.on_ac = True # assume AC until determined otherwise + self.old_on_ac = True self.dev = dev self.cpu = cpu @@ -424,24 +499,48 @@ def open( context: Context, ): self.emit = emit + try: + from hhd.utils import get_ac_status, get_ac_status_fn + ac = get_ac_status(get_ac_status_fn()) + if ac is not None: + self.on_ac = ac + except Exception: + pass def update(self, conf: Config): self.enabled = conf["hhd.settings.tdp_ready"].to(bool) - self.enforce_limits = conf["hhd.settings.enforce_limits"].to(bool) + user_enforce = conf["hhd.settings.enforce_limits"].to(bool) + dock_running = conf.get("cooling_dock.dock_running", False) + self.enforce_limits = user_enforce and not ( + dock_running and self.dock_aware and self.on_ac + ) + dock_changed = self.enforce_limits != self.old_enforce + self.old_enforce = self.enforce_limits + + power_changed = self.on_ac != self.old_on_ac + self.old_on_ac = self.on_ac + if not self.enabled or not self.initialized: return if self.enforce_limits: for k, v in conf["tdp.smu.std"].to(dict).items(): if k in self.dev: - mmin, mmax = self.dev[k].smin, self.dev[k].smax + mmin = self.dev[k].smin + mmax = self.dev[k].smax + # On battery, use the DC cap if available + if not self.on_ac and self.dc_cap and k in self.dc_cap: + mmax = self.dc_cap[k] if v < mmin: conf["tdp.smu.std", k] = mmin if v > mmax: conf["tdp.smu.std", k] = mmax for k, v in conf["tdp.smu.adv"].to(dict).items(): if k in self.dev and k != "enable": - mmin, mmax = self.dev[k].smin, self.dev[k].smax + mmin = self.dev[k].smin + mmax = self.dev[k].smax + if not self.on_ac and self.dc_cap and k in self.dc_cap: + mmax = self.dc_cap[k] if v < mmin: conf["tdp.smu.adv", k] = mmin if v > mmax: @@ -455,7 +554,11 @@ def update(self, conf: Config): if k != "enable": new_vals[k] = v - if set(new_vals.items()) != set(self.old_vals.items()): + if ( + set(new_vals.items()) != set(self.old_vals.items()) + or dock_changed + or power_changed + ): self.is_set = False if self.has_pp: @@ -470,7 +573,10 @@ def update(self, conf: Config): self.old_target = new_target self.emit({"type": "energy", "status": new_target}) # type: ignore - if conf["tdp.smu.apply"].to(bool): + # Force re-apply when the dock connects/disconnects or power state changes + # so the clamped limits reach the CPU immediately instead of waiting for a + # manual Apply (or the SmuQamPlugin's delayed queue). + if conf["tdp.smu.apply"].to(bool) or dock_changed or power_changed: conf["tdp.smu.apply"] = False if self.has_pp: @@ -493,5 +599,12 @@ def update(self, conf: Config): else: conf["tdp.smu.status"] = "Not Set" + def notify(self, events: Sequence[Event]): + for ev in events: + # AC/DC power state: update on_ac so the next update() cycle + # clamps TDP values using the correct limits. + if ev["type"] == "acpi" and ev.get("event") in ("ac", "dc"): + self.on_ac = ev["event"] == "ac" + def close(self): pass diff --git a/src/adjustor/hhd.py b/src/adjustor/hhd.py index 4d0484c89..23151018c 100644 --- a/src/adjustor/hhd.py +++ b/src/adjustor/hhd.py @@ -9,7 +9,7 @@ from hhd.plugins.plugin import Emitter from adjustor.core.acpi import check_perms, initialize -from adjustor.core.const import CPU_DATA, DEV_DATA, ASUS_DATA, MSI_DATA +from adjustor.core.const import CPU_DATA, DC_CAP_OXP_SUPERX, DEV_DATA, ASUS_DATA, MSI_DATA from adjustor.decky import disable_decky_plugins, find_decky_plugins from .i18n import _ @@ -280,15 +280,29 @@ def autodetect(existing: Sequence[HHDPlugin]) -> Sequence[HHDPlugin]: if not drivers_matched and prod in DEV_DATA and not USE_UNIFIED: dev, cpu, pp_enable, energy_map = DEV_DATA[prod] + # Enable higher TDP limits when docked (e.g. 120W on SUPER X). + dock_aware = prod in ("ONEXPLAYER SUPER X", "ONEXPLAYER APEX") + + # Provide DC battery limits (e.g. 55W) for dynamic TDP clamping. + dc_cap = DC_CAP_OXP_SUPERX if dock_aware else None try: - # Set values for the steam slider + # Configure Steam UI slider limits. + # Use DC cap on battery so the slider matches hardware limits. if dev["skin_limit"].smin: min_tdp = dev["skin_limit"].smin if dev["skin_limit"].default: default_tdp = dev["skin_limit"].default if dev["skin_limit"].smax: max_tdp = dev["skin_limit"].smax + if dc_cap: + # Detect initial AC state for the UI slider max. + # The adjustor handles dynamic updates later. + from hhd.utils import get_ac_status, get_ac_status_fn + ac_fn = get_ac_status_fn() + initial_ac = get_ac_status(ac_fn) + if initial_ac is False and "skin_limit" in dc_cap: + max_tdp = dc_cap["skin_limit"] except Exception as e: logger.error(f"Failed to get TDP limits for {prod}:\n{e}") @@ -298,6 +312,8 @@ def autodetect(existing: Sequence[HHDPlugin]) -> Sequence[HHDPlugin]: dev, cpu, platform_profile=pp_enable, + dock_aware=dock_aware, + dc_cap=dc_cap, ) ) drivers.append( @@ -306,6 +322,8 @@ def autodetect(existing: Sequence[HHDPlugin]) -> Sequence[HHDPlugin]: energy_map, pp_enable=pp_enable, init_tdp=not prod == "83E1", + dock_aware=dock_aware, + dc_cap=dc_cap, ), ) drivers_matched = True diff --git a/src/hhd/__main__.py b/src/hhd/__main__.py index 70485b6ce..8b19edeef 100644 --- a/src/hhd/__main__.py +++ b/src/hhd/__main__.py @@ -728,9 +728,14 @@ def run_plugin_cmd(cmd: Callable[[HHDPlugin], None], reverse: bool = False): has_new = should_initialize.is_set() saved = False # Save existing profiles if open - if save_state_yaml(state_fn, settings, conf, shash): - saved = True + if conf.updated and not getattr(conf, "yaml_save_queued", 0): + conf.yaml_save_queued = curr + 0.3 + + if conf.updated and curr >= getattr(conf, "yaml_save_queued", 0): + if save_state_yaml(state_fn, settings, conf, shash): + saved = True conf.updated = False + conf.yaml_save_queued = 0 for name, prof in profiles.items(): fn = join(profile_dir, name + ".yml") if save_profile_yaml(fn, settings, prof, shash): diff --git a/src/hhd/device/oxp/base.py b/src/hhd/device/oxp/base.py index dcdc233a6..4d8287e76 100644 --- a/src/hhd/device/oxp/base.py +++ b/src/hhd/device/oxp/base.py @@ -236,6 +236,9 @@ def produce(self, fds): if ev["type"] == "button" and ev["code"] in ( "mode", "keyboard", + "key_leftctrl", + "key_leftmeta", + "key_leftalt", ): if ev["value"]: self.state[ev["code"]] = curr @@ -246,6 +249,28 @@ def produce(self, fds): self.queued.append((ev["code"], t + BUTTON_MIN_DELAY)) skip.append(i) + # Check for Turbo macro (Ctrl + Meta + Alt) + if ( + "key_leftctrl" in self.state + and "key_leftmeta" in self.state + and "key_leftalt" in self.state + ): + # Consume the keys + self.state.pop("key_leftctrl", None) + self.state.pop("key_leftmeta", None) + self.state.pop("key_leftalt", None) + + # Emit share/mode button depending on mappings + share_code = "mode" if "mode" in self.btn_map.values() else "share" + evs.append( + { + "type": "button", + "code": share_code, + "value": True, + } + ) + self.queued.append((share_code, curr + BUTTON_MIN_DELAY)) + for i in reversed(skip): evs.pop(i) @@ -771,6 +796,7 @@ def prepare(m): d_vend_id = [id(d) for d in d_vend] if dconf.get("g1", False): prepare(d_kbd_2) + prepare(d_xinput) if motion: start_imu = True diff --git a/src/hhd/device/oxp/const.py b/src/hhd/device/oxp/const.py index 4417bddf7..bc43c31ec 100644 --- a/src/hhd/device/oxp/const.py +++ b/src/hhd/device/oxp/const.py @@ -20,7 +20,9 @@ B("KEY_VOLUMEUP"): "key_volumeup", B("KEY_VOLUMEDOWN"): "key_volumedown", # Turbo Button [29, 56, 125] KEY_LEFTCTRL + KEY_LEFTALT + KEY_LEFTMETA - B("KEY_LEFTALT"): "share", + B("KEY_LEFTALT"): "key_leftalt", + B("KEY_LEFTCTRL"): "key_leftctrl", + B("KEY_LEFTMETA"): "key_leftmeta", # Short press orange [32, 125] KEY_D + KEY_LEFTMETA B("KEY_D"): "mode", # KB Button [24, 97, 125] KEY_O + KEY_RIGHTCTRL + KEY_LEFTMETA @@ -37,6 +39,9 @@ # If we do not have turbo takeover, let turbo do its turbo thing, and # failover to having the keyboard button open the overlay B("KEY_O"): "share", + B("KEY_LEFTALT"): "key_leftalt", + B("KEY_LEFTCTRL"): "key_leftctrl", + B("KEY_LEFTMETA"): "key_leftmeta", } BTN_MAPPINGS_X2: dict[int, Button] = { @@ -202,6 +207,16 @@ "protocol": "hid_v1_g1", "turbo": True, # disable turbo takeover so that it can be used for TDP }, + "ONEXPLAYER SUPER X": { + **ONEX_DEFAULT_CONF, + "name": "ONEXPLAYER SUPER X", + "protocol": "mixed", + }, + "ONEXPLAYER APEX": { + **ONEX_DEFAULT_CONF, + "name": "ONEXPLAYER APEX", + "protocol": "mixed", + }, } diff --git a/src/hhd/plugins/cooling_dock/__init__.py b/src/hhd/plugins/cooling_dock/__init__.py new file mode 100644 index 000000000..ee61d635e --- /dev/null +++ b/src/hhd/plugins/cooling_dock/__init__.py @@ -0,0 +1,3 @@ +from .base import autodetect + +__all__ = ["autodetect"] diff --git a/src/hhd/plugins/cooling_dock/base.py b/src/hhd/plugins/cooling_dock/base.py new file mode 100644 index 000000000..adf4ed666 --- /dev/null +++ b/src/hhd/plugins/cooling_dock/base.py @@ -0,0 +1,877 @@ +import asyncio +import logging +import os +import threading +import time +from dataclasses import replace +from typing import Sequence + +from hhd.plugins import Config, Context, Emitter, HHDPlugin, load_relative_yaml +from hhd.plugins.settings import HHDSettings + +logger = logging.getLogger(__name__) + +# Only load the cooling dock plugin on devices that have one. +SUPPORTED_PRODUCTS = ("ONEXPLAYER SUPER X", "ONEXPLAYER APEX") + +# Scan backoff limits (seconds) +SCAN_BACKOFF_MIN = 5 +SCAN_BACKOFF_MAX = 15 +SCAN_BACKOFF_FACTOR = 2 + +# BLE watchdog: Max seconds without successful GATT read before sync loop breaks. +GATT_WATCHDOG_TIMEOUT = 20 + +# GATT operation timeout to prevent async hangs. +GATT_OP_TIMEOUT = 10 + +# Max consecutive GATT errors before forcing disconnect (anti-flap). +SYNC_RETRY_MAX = 3 + +# Delay (s) before reconnect to allow BlueZ cleanup. +RECONNECT_DELAY = 2 + +# Delay (s) between transient GATT error retries. +SYNC_RETRY_DELAY = 2 + +# Sync loop interval (s). >2s required to avoid firmware BLE exhaustion. +SYNC_READ_INTERVAL = 5 + + +# Grace period (s) before dropping dock_running (and TDP) to ride out packet drops. +DOCK_RUNNING_GRACE = 10 + +try: + from bleak import BleakClient, BleakScanner + from bleak.backends.device import BLEDevice + + BLEAK_AVAILABLE = True +except ImportError: + BLEAK_AVAILABLE = False + +from .protocol import (CHAR_UUID, CHUNK_DELAY_S, DEVICE_NAME, NOTIFY_UUID, + POST_WRITE_DELAY_S, SERVICE_UUID, TOTAL_BYTES, + WRITE_CMD, WRITE_RETRY_DELAY_S, WRITE_RETRY_MAX, + CoolingStatus, DockMode, build_write_chunks) + + +def get_cpu_temp() -> float: + highest = 0.0 + try: + hwmon_dir = "/sys/class/hwmon" + if not os.path.exists(hwmon_dir): + return highest + for hwmon in os.listdir(hwmon_dir): + path = os.path.join(hwmon_dir, hwmon) + try: + with open(os.path.join(path, "name"), "r") as f: + name = f.read().strip() + if name in ("k10temp", "oxpec", "amdgpu"): + for file in os.listdir(path): + if file.startswith("temp") and file.endswith("_input"): + with open(os.path.join(path, file), "r") as f: + temp = int(f.read().strip()) / 1000.0 + if temp > highest: + highest = temp + except Exception: + continue + except Exception: + pass + return highest + + +def fan_pct_for_temp(temp: float, curve: list[tuple[int, int]]) -> int: + if not curve: + return 0 + sorted_curve = sorted(curve, key=lambda x: x[1]) + if temp <= sorted_curve[0][1]: + return sorted_curve[0][0] + for i in range(1, len(sorted_curve)): + if temp <= sorted_curve[i][1]: + t0, f0 = sorted_curve[i - 1][1], sorted_curve[i - 1][0] + t1, f1 = sorted_curve[i][1], sorted_curve[i][0] + if t1 == t0: + return f1 + ratio = (temp - t0) / (t1 - t0) + return int(f0 + ratio * (f1 - f0)) + return sorted_curve[-1][0] + + +class CoolingDockPlugin(HHDPlugin): + name = "cooling_dock" + priority = 20 + log = "dock" + + def __init__(self) -> None: + self.running = False + self.thread = None + self.conf_lock = threading.Lock() + self.conf = None + self.enabled = False + self.mode = "auto" + self.fan_curve = self._default_curve() + self.rgb_enable = True + self.rgb_mode = 1 + self.rgb_level = 3 + self._last_fan_pct = -1 + self._dock_running = False + self._status = "Disconnected" + self._fan_progress = None + self._started = False + self._scan_delay = SCAN_BACKOFF_MIN + self._last_gatt_read = 0.0 + self._mac_address = "" + self._is_water_cooled = False + self._force_reconnect = False + self._scan_requested = False + self._discovered_macs = {"": "None"} + self._last_write_target = None + self._last_write_time = 0.0 + self._last_connected_time = 0.0 + self._last_disconnect_time = None + + def _default_curve(self) -> list[tuple[int, int]]: + return [(0, 40), (30, 50), (50, 60), (70, 70), (85, 80)] + + def open(self, emit: Emitter, context: Context): + self.emit = emit + if not BLEAK_AVAILABLE: + logger.warning("Bleak not available, Cooling Dock plugin disabled.") + return + self.running = True + self.thread = threading.Thread(target=self._run_loop, daemon=True) + self.thread.start() + + def close(self): + self.running = False + if self.thread: + self.thread.join(timeout=3) + + def settings(self) -> HHDSettings: + base = {"cooling_dock": {"dock": load_relative_yaml("settings.yml")}} + if not BLEAK_AVAILABLE: + base["cooling_dock"]["dock"]["children"]["enabled"][ + "hint" + ] = "Bleak is not installed. Install with: pip install bleak" + else: + with self.conf_lock: + opts = self._discovered_macs.copy() + if self._mac_address and self._mac_address not in opts: + opts[self._mac_address] = ( + f"CoolingSystem_ONEC1 ({self._mac_address})" + ) + base["cooling_dock"]["dock"]["children"]["mac_address"][ + "options" + ] = opts + + # Dynamically hide controls if dock sync is disabled or no dock is selected + if not self.enabled or not self._mac_address: + children = base["cooling_dock"]["dock"]["children"] + for key in ["mode", "fan_curve", "rgb"]: + if key in children: + del children[key] + + return base + + def update(self, conf: Config): + try: + dock_conf = conf["cooling_dock.dock"] + except Exception: + return + + with self.conf_lock: + self.conf = conf + self.enabled = dock_conf.get("enabled", True) + self.mode = dock_conf.get("mode", "auto") + + curve = [] + for i in range(1, 6): + t = dock_conf.get(f"fan_curve.t{i}", None) + f = dock_conf.get(f"fan_curve.f{i}", None) + if t is not None and f is not None: + curve.append((int(f), int(t))) + if curve: + self.fan_curve = curve + + self.rgb_enable = dock_conf.get("rgb.enable", True) + self.rgb_mode = int(dock_conf.get("rgb.mode", 1)) + self.rgb_level = int(dock_conf.get("rgb.level", 3)) + + self._mac_address = dock_conf.get("mac_address", "") + + # Handle the 'Forget Dock' action + if conf.get("cooling_dock.dock.forget_dock", False): + conf["cooling_dock.dock.forget_dock"] = False + # Unpair from BlueZ to fully disconnect. + self._forget_bluez_device() + self._mac_address = "" + conf["cooling_dock.dock.mac_address"] = "" + self._force_reconnect = True + + # Handle the 'Scan Dock' action + if conf.get("cooling_dock.dock.scan_dock", False): + conf["cooling_dock.dock.scan_dock"] = False + self._scan_requested = True + self._force_reconnect = True + + # Self-heal stale runtime state on first update. + if not self._started: + self._started = True + conf["cooling_dock.dock_running"] = self._dock_running + conf["cooling_dock.dock.status"] = self._status + conf["cooling_dock.dock.fan_progress"] = self._fan_progress + + def _publish_dock_running(self, running: bool): + # Publish running state for TDP adjustments (only on change). + with self.conf_lock: + conf = self.conf + if running == self._dock_running: + return + self._dock_running = running + if conf is not None: + conf["cooling_dock.dock_running"] = running + + def _publish_status(self, status: str, fan_progress: dict | None): + # Publish UI status (only on change). + with self.conf_lock: + conf = self.conf + if status == self._status and fan_progress == self._fan_progress: + return + self._status = status + self._fan_progress = fan_progress + if conf is not None: + conf["cooling_dock.dock.status"] = status + conf["cooling_dock.dock.fan_progress"] = fan_progress + + def _publish_disconnected_if_stale(self): + # Delay dropping dock_running to ride out transient BLE packet drops. + if ( + self._last_connected_time + and time.time() - self._last_connected_time > DOCK_RUNNING_GRACE + ): + self._publish_dock_running(False) + self._publish_status("Disconnected", None) + + async def _write_state(self, client, state: bytearray): + """Write a modified 64-byte state to the dock using the chunked + protocol (3 x 20-byte frames with 0x1C/0x2C/0x3C headers). + + A single 64-byte write is silently ignored by the dock, so the state + must be split into chunks. Retries handle transient "In Progress" + errors from the dock while it is still processing a previous write. + """ + chunks = build_write_chunks(state) + last_error = None + for attempt in range(WRITE_RETRY_MAX): + try: + for chunk in chunks: + await asyncio.wait_for( + client.write_gatt_char(CHAR_UUID, chunk, response=True), + timeout=GATT_OP_TIMEOUT, + ) + await asyncio.sleep(CHUNK_DELAY_S) + await asyncio.sleep(POST_WRITE_DELAY_S) + return + except Exception as e: + last_error = e + logger.warning( + f"Cooling Dock chunked write failed " + f"({attempt + 1}/{WRITE_RETRY_MAX}): {e}" + ) + if attempt < WRITE_RETRY_MAX - 1: + await asyncio.sleep(WRITE_RETRY_DELAY_S) + raise last_error if last_error else RuntimeError("write failed") + + def _run_loop(self): + asyncio.run(self._async_loop()) + + async def _async_loop(self): + while self.running: + if not self.enabled: + self._publish_dock_running(False) + self._publish_status("Disconnected", None) + self._scan_delay = SCAN_BACKOFF_MIN + await asyncio.sleep(5) + continue + try: + self._force_reconnect = False + await self._connect_and_sync() + except Exception as e: + logger.error(f"Cooling Dock error: {e}") + if self.running: + for _ in range(5): + if ( + not self.running + or self._force_reconnect + or self._scan_requested + ): + break + await asyncio.sleep(1) + + async def _connect_and_sync(self): + # Enforce grace period to ride out transient BLE drops. + self._publish_disconnected_if_stale() + + ble_device = await self._find_dock() + if not ble_device: + logger.info(f"Cooling Dock not found, retrying in {self._scan_delay}s...") + self._publish_disconnected_if_stale() + delay = self._scan_delay + # Exponential backoff: increase delay for next scan + self._scan_delay = min( + self._scan_delay * SCAN_BACKOFF_FACTOR, SCAN_BACKOFF_MAX + ) + + for _ in range(delay): + if not self.running or self._force_reconnect or self._scan_requested: + break + await asyncio.sleep(1) + return + + # Reset backoff on successful discovery + self._scan_delay = SCAN_BACKOFF_MIN + + addr = ble_device.address + logger.info(f"Connecting to Cooling Dock at {addr}...") + + # Listen for immediate BlueZ link loss. + disconnected_event = asyncio.Event() + + def _on_disconnect(c): + logger.info("Cooling Dock BLE link lost (disconnected callback)") + disconnected_event.set() + + client = BleakClient( + ble_device, timeout=15, disconnected_callback=_on_disconnect + ) + # Retry transient connect errors (e.g. BlueZ "In Progress"). + for attempt in range(3): + try: + await client.connect() + break + except Exception as e: + if attempt == 2: + raise + logger.warning(f"Cooling Dock connect retry ({attempt + 1}/3): {e}") + await asyncio.sleep(2) + if not client.is_connected: + logger.warning("Failed to connect to Cooling Dock") + self._publish_disconnected_if_stale() + await asyncio.sleep(5) + return + + logger.info("Cooling Dock connected!") + self._publish_status("Connected", None) + self._last_gatt_read = time.time() + now = time.time() + # Log reconnect duration for grace period validation. + if self._last_disconnect_time: + gap = now - self._last_disconnect_time + logger.info(f"BLE reconnect after {gap:.1f}s gap") + self._last_disconnect_time = None + self._last_connected_time = now + + # Save MAC address for sticky pairing if we don't have it yet + if not self._mac_address: + with self.conf_lock: + self._mac_address = addr + if self.conf is not None: + self.conf["cooling_dock.dock.mac_address"] = addr + + # Register connected MAC for UI. Do NOT emit settings to avoid reload storm. + with self.conf_lock: + if addr not in self._discovered_macs: + self._discovered_macs[addr] = f"CoolingSystem_ONEC1 ({addr})" + + consecutive_errors = 0 + while ( + self.running + and client.is_connected + and not self._force_reconnect + and not disconnected_event.is_set() + ): + # BLE watchdog: if no successful GATT read for too long, + # assume the dock disconnected (BLE supervision timeout gap). + if time.time() - self._last_gatt_read > GATT_WATCHDOG_TIMEOUT: + logger.warning( + f"Dock GATT read timeout ({GATT_WATCHDOG_TIMEOUT}s), " + f"assuming disconnected." + ) + break + + try: + current = await asyncio.wait_for( + client.read_gatt_char(CHAR_UUID), timeout=GATT_OP_TIMEOUT + ) + self._last_gatt_read = time.time() + status = CoolingStatus.from_bytes(current) + + # Dynamic hardware detection (Air vs Water Cooled) + if status.pump_speed_percent > 0 or status.water_flow > 0: + self._is_water_cooled = True + + self._publish_dock_running( + status.fan_speed > 0 or status.fan_speed_percent > 0 + ) + + if status.mode == 0: + status_str = "Connected - Stopped" + ui_fan_pct = 0 + elif self._is_water_cooled: + status_str = ( + f"Connected (Water) - Fan {status.fan_speed_percent}% " + f"Pump {status.pump_speed_percent}%" + ) + ui_fan_pct = status.fan_speed_percent + else: + status_str = ( + f"Connected (Air) - Fan {status.fan_speed_percent}% " + f"({status.fan_speed} RPM)" + ) + ui_fan_pct = status.fan_speed_percent + + self._publish_status( + status_str, + { + "value": ui_fan_pct, + "max": 100, + "unit": "%", + "text": "Dock Fan", + }, + ) + + with self.conf_lock: + mode = self.mode + curve = list(self.fan_curve) + rgb_en = self.rgb_enable + rgb_m = self.rgb_mode + rgb_lvl = self.rgb_level + + payload = bytearray(current) + payload[1] = WRITE_CMD + + if mode == "auto": + temp = get_cpu_temp() + fan_pct = fan_pct_for_temp(temp, curve) + payload[4] = int(DockMode.AUTO) + payload[15] = 0xFE + idx = 23 + for f, t in curve: + if idx + 1 < len(payload): + payload[idx] = f + payload[idx + 1] = t + idx += 2 + logger.debug( + f"Auto: temp={temp:.1f}C fan={fan_pct}% " + f"(dock reports {status.fan_speed} RPM)" + ) + else: + try: + mode_val = int(mode) + except ValueError: + mode_val = int(DockMode.AUTO) + payload[4] = mode_val + + payload[16] = rgb_m + payload[17] = 1 if rgb_en else 0 + payload[19] = rgb_lvl + + # Only write on change. Writing every cycle hammers the dock and triggers GATT errors. + target = (mode, tuple(curve), rgb_en, rgb_m, rgb_lvl) + now = time.time() + target_changed = target != self._last_write_target + if target_changed: + logger.info(f"Writing to dock: changed={target_changed}") + await self._write_state(client, payload) + self._last_write_target = target + self._last_write_time = now + + consecutive_errors = 0 + await asyncio.sleep(SYNC_READ_INTERVAL) + + except asyncio.TimeoutError: + consecutive_errors += 1 + logger.warning( + f"Cooling Dock GATT timeout " + f"({consecutive_errors}/{SYNC_RETRY_MAX})" + ) + if consecutive_errors >= SYNC_RETRY_MAX: + logger.error("Too many GATT timeouts, disconnecting.") + break + await asyncio.sleep(SYNC_RETRY_DELAY) + + except Exception as e: + consecutive_errors += 1 + logger.warning( + f"Cooling Dock sync error ({consecutive_errors}/" + f"{SYNC_RETRY_MAX}): {e}" + ) + if consecutive_errors >= SYNC_RETRY_MAX: + logger.error("Too many sync errors, disconnecting.") + break + await asyncio.sleep(SYNC_RETRY_DELAY) + + try: + await client.disconnect() + except Exception: + pass + + self._force_reconnect = False + self._last_disconnect_time = time.time() + # Do NOT drop dock_running here. _publish_disconnected_if_stale() handles the grace period. + + # Pause to let BlueZ clean up link. + for _ in range(RECONNECT_DELAY): + if not self.running or self._force_reconnect or self._scan_requested: + break + await asyncio.sleep(1) + + async def _bluez_start_discovery(self, target_mac: str | None, timeout: int = 10): + """Trigger BlueZ to scan for BLE devices via D-Bus and wait until found.""" + try: + import dbus + + bus = dbus.SystemBus() + adapter = dbus.Interface( + bus.get_object("org.bluez", "/org/bluez/hci0"), "org.bluez.Adapter1" + ) + try: + adapter.StartDiscovery() + except dbus.DBusException as e: + if e.get_dbus_name() != "org.bluez.Error.InProgress": + raise + + # Poll for the device to appear so we can stop scanning early + for _ in range(timeout): + if not self.running: + break + device = self._find_dock_in_bluez_objects(target_mac) + if device: + break + await asyncio.sleep(1) + + try: + adapter.StopDiscovery() + except dbus.DBusException: + pass + except Exception as e: + logger.debug(f"BlueZ start discovery failed: {e}") + + def _find_dock_in_bluez_objects( + self, target_mac: str | None = None + ) -> BLEDevice | None: + """Query BlueZ D-Bus directly for the dock. + + This finds devices even if they are bonded (which BleakScanner filters out) + or not currently advertising but known to BlueZ. + """ + try: + import dbus + + bus = dbus.SystemBus() + obj = bus.get_object("org.bluez", "/") + om = dbus.Interface(obj, "org.freedesktop.DBus.ObjectManager") + objects = om.GetManagedObjects() + + for path, ifaces in objects.items(): + props = ifaces.get("org.bluez.Device1") + if not props: + continue + + name = str(props.get("Name", props.get("Alias", ""))) + mac = str(props.get("Address", "")).upper() + + if target_mac and mac != target_mac.upper(): + continue + + name_match = "Cooling" in name + mac_match = "C8:17:17" in mac + + if not target_mac and not (name_match or mac_match): + continue + + logger.info(f"Found dock in BlueZ D-Bus: {name} ({mac})") + return BLEDevice(mac, name, {"path": str(path)}) + except ImportError: + pass + except Exception as e: + logger.debug(f"BlueZ D-Bus object lookup failed: {e}") + return None + + async def _find_dock(self) -> BLEDevice | None: + try: + with self.conf_lock: + scan_req = self._scan_requested + self._scan_requested = False + + if scan_req: + self._publish_status("Scanning...", None) + + target_mac = self._mac_address.upper() if self._mac_address else None + + # If no dock is selected ("None") and user didn't request a scan, do nothing. + if not target_mac and not scan_req: + return None + + # 1. First, check if BlueZ already knows the device + device = self._find_dock_in_bluez_objects(target_mac) + if device: + # Need to run discover to populate the dropdown + if scan_req: + await self._populate_dropdown() + return device + + # 2. If not found, trigger BlueZ discovery directly. + # We poll during the scan and exit early if found, so 10s is safe. + scan_timeout = 10 + await self._bluez_start_discovery(target_mac, timeout=scan_timeout) + + if scan_req: + await self._populate_dropdown() + + device = self._find_dock_in_bluez_objects(target_mac) + if device: + return device + + # 3. Fallback to bleak scanner if D-Bus fails + if target_mac: + device = await BleakScanner.find_device_by_address( + target_mac, timeout=5 + ) + if device: + return device + + device = await BleakScanner.find_device_by_name(DEVICE_NAME, timeout=10) + return device + + except Exception as e: + logger.debug(f"BLE scan error: {e}") + return None + + async def _populate_dropdown(self): + """Populate the UI dropdown with discovered devices.""" + discovered = {"": "None"} + found_macs = [] + + try: + import dbus + + bus = dbus.SystemBus() + om = dbus.Interface( + bus.get_object("org.bluez", "/"), "org.freedesktop.DBus.ObjectManager" + ) + objects = om.GetManagedObjects() + for path, ifaces in objects.items(): + props = ifaces.get("org.bluez.Device1") + if not props: + continue + name = str(props.get("Name", props.get("Alias", ""))) + mac = str(props.get("Address", "")).upper() + if "Cooling" in name: + discovered[mac] = f"{name} ({mac})" + found_macs.append(mac) + except Exception as e: + logger.debug(f"D-Bus dropdown populate failed: {e}") + + try: + devices = await BleakScanner.discover(timeout=3) + for d in devices: + if d.name and "Cooling" in d.name: + discovered[d.address.upper()] = f"{d.name} ({d.address.upper()})" + found_macs.append(d.address.upper()) + except Exception: + pass + + with self.conf_lock: + old_keys = set(self._discovered_macs.keys()) + new_keys = set(discovered.keys()) + self._discovered_macs = discovered + + # If "None" was selected but we found a dock during a manual scan, auto-select it + if self._mac_address == "" and found_macs: + self._mac_address = found_macs[0] + if self.conf is not None: + self.conf["cooling_dock.dock.mac_address"] = found_macs[0] + + if old_keys != new_keys and self.emit: + self.emit({"type": "settings"}) + + def _remove_stale_bond(self, mac: str): + """Remove any existing BlueZ bond/pairing for the dock. + + The dock's HID profile causes BlueZ to trigger SMP pairing, which + creates a bond. Because the dock uses a random address type, the + bond keys quickly become stale. A stale bond prevents BleakScanner + from seeing the device, and causes BlueZ to auto-connect with bad keys. + Windows never pairs with the dock (WinRT stateless connection). We + replicate that by deleting the bond before every connection attempt. + """ + if not mac: + return + # Try D-Bus first (fast, no subprocess) + try: + import dbus + + bus = dbus.SystemBus() + obj = bus.get_object("org.bluez", "/") + om = dbus.Interface(obj, "org.freedesktop.DBus.ObjectManager") + objects = om.GetManagedObjects() + for path, ifaces in sorted(objects.items()): + props = ifaces.get("org.bluez.Device1") + if not props: + continue + if str(props.get("Address", "")).upper() != mac.upper(): + continue + logger.info(f"Removing stale BlueZ bond for {mac}") + adapter = dbus.Interface( + bus.get_object("org.bluez", str(path).rsplit("/", 1)[0]), + "org.bluez.Adapter1", + ) + adapter.RemoveDevice(dbus.ObjectPath(path)) + import time + + time.sleep(0.5) + return + except ImportError: + pass # python-dbus not installed, fall through + except Exception as e: + logger.debug(f"BlueZ D-Bus bond removal failed: {e}") + + # Fallback: bluetoothctl (works without python-dbus) + try: + import subprocess + + subprocess.run( + ["bluetoothctl", "remove", mac], + capture_output=True, + text=True, + timeout=3, + ) + logger.info(f"Removed stale bond for {mac} via bluetoothctl") + except Exception as e: + logger.debug(f"bluetoothctl remove failed: {e}") + + def _forget_bluez_device(self): + """Unpair and remove the dock from BlueZ. + + The dock is Trusted, so BlueZ auto-connects it and keeps it paired. + "Forget" must remove it from BlueZ entirely, otherwise the dock + stays connected and fan control keeps working. + """ + mac = self._mac_address + if not mac: + return + try: + import dbus + + bus = dbus.SystemBus() + obj = bus.get_object("org.bluez", "/") + om = dbus.Interface(obj, "org.freedesktop.DBus.ObjectManager") + objects = om.GetManagedObjects() + for path, ifaces in sorted(objects.items()): + props = ifaces.get("org.bluez.Device1") + if not props: + continue + if str(props.get("Address", "")).upper() != mac.upper(): + continue + logger.info(f"Removing Cooling Dock {mac} from BlueZ") + adapter = dbus.Interface( + bus.get_object("org.bluez", str(path).rsplit("/", 1)[0]), + "org.bluez.Adapter1", + ) + adapter.RemoveDevice(dbus.ObjectPath(path)) + return + except Exception as e: + logger.debug(f"BlueZ forget failed: {e}") + + def _find_connected_dock_via_bluez( + self, target_mac: str | None = None + ) -> BLEDevice | None: + """Find a CoolingSystem dock that is already connected to BlueZ. + + Connected BLE devices stop advertising, so bleak scans cannot see + them. After a sync-loop break or reconnect, the dock is often still + connected via BlueZ (auto-connect to a trusted device) while not + advertising — this queries BlueZ directly to recover it. + + Returns a BLEDevice carrying the BlueZ D-Bus path. BleakClient + uses ``details["path"]`` to attach to the existing connection + without requiring a scan, which is essential for a connected but + non-advertising dock. + """ + try: + import dbus + + bus = dbus.SystemBus() + obj = bus.get_object("org.bluez", "/") + om = dbus.Interface(obj, "org.freedesktop.DBus.ObjectManager") + objects = om.GetManagedObjects() + for path, ifaces in sorted(objects.items()): + props = ifaces.get("org.bluez.Device1") + if not props: + continue + if not bool(props.get("Connected", False)): + continue + name = str(props.get("Name", "")) + if "Cooling" not in name: + continue + mac = str(props.get("Address", "")).upper() + if target_mac and mac != target_mac: + continue + logger.info(f"Found connected Cooling Dock at {mac}") + return BLEDevice(mac, name, {"path": str(path)}) + except Exception as e: + logger.debug(f"BlueZ D-Bus connected-dock check failed: {e}") + + # Fallback: bluetoothctl (assumes hci0). + try: + import subprocess + + result = subprocess.run( + ["bluetoothctl", "devices"], + capture_output=True, + text=True, + timeout=2, + ) + for line in result.stdout.split("\n"): + if "CoolingSystem" not in line and "Cooling" not in line: + continue + parts = line.strip().split(" ", 2) + if len(parts) < 3: + continue + mac = parts[1].upper() + if target_mac and mac != target_mac: + continue + info = subprocess.run( + ["bluetoothctl", "info", mac], + capture_output=True, + text=True, + timeout=2, + ) + if "Connected: yes" in info.stdout: + logger.info(f"Found connected Cooling Dock at {mac}") + path = f"/org/bluez/hci0/dev_{mac.replace(':', '_')}" + return BLEDevice(mac, parts[2], {"path": path}) + except Exception as e: + logger.debug(f"BlueZ connected-dock check failed: {e}") + return None + + +def _is_supported_device() -> bool: + """Check if the current device is a supported OneXPlayer model.""" + try: + with open("/sys/devices/virtual/dmi/id/product_name") as f: + prod = f.read().strip() + return prod in SUPPORTED_PRODUCTS + except Exception: + return False + + +def autodetect(existing: Sequence[HHDPlugin]) -> Sequence[HHDPlugin]: + if len([p for p in existing if p.name == "cooling_dock"]): + return existing + + if not _is_supported_device(): + return existing + + return [CoolingDockPlugin()] diff --git a/src/hhd/plugins/cooling_dock/protocol.py b/src/hhd/plugins/cooling_dock/protocol.py new file mode 100644 index 000000000..b98343ce1 --- /dev/null +++ b/src/hhd/plugins/cooling_dock/protocol.py @@ -0,0 +1,152 @@ +"""CoolingStatus byte-array protocol for the CoolingSystem_ONEC1 BLE dock. + +Layout recovered from native JIT disassembly of CoolingStatus.From / .Fill +in CompatLayerCT.exe (see disasm_cooling.txt, method 01186 / 01187). + +GATT: service 0xFFE0, characteristic 0xFFE1 (read+write), 64 bytes. + +IMPORTANT: The read (From) and write (Fill) directions use DIFFERENT byte +indices for the same logical fields. See the read and write tables below. +""" + +from __future__ import annotations +from dataclasses import dataclass, field +from enum import IntEnum + +SERVICE_UUID = "0000ffe0-0000-1000-8000-00805f9b34fb" +CHAR_UUID = "0000ffe1-0000-1000-8000-00805f9b34fb" +NOTIFY_UUID = "0000ffe4-0000-1000-8000-00805f9b34fb" +DEVICE_NAME = "CoolingSystem_ONEC1" + +TOTAL_BYTES = 64 +WRITE_CMD = 0x02 +READ_CMD = 0x10 + +# Chunked write protocol: Dock requires 3x20-byte frames with 0x1C/0x2C/0x3C headers. +WRITE_PAYLOAD_SIZE = 58 +CHUNK_HEADERS = (0x1C, 0x2C, 0x3C) +CHUNK_SIZE = 19 +CHUNK_DELAY_S = 0.02 +POST_WRITE_DELAY_S = 0.3 +WRITE_RETRY_MAX = 3 +WRITE_RETRY_DELAY_S = 0.5 +ON_RETRY_DELAY_S = 0.5 +MAX_ON_WRITE_RETRIES = 10 + + +class DockMode(IntEnum): + STOPPED = 0x00 + LEVEL_1 = 0x01 + LEVEL_2 = 0x02 + LEVEL_3 = 0x03 + LEVEL_4 = 0x04 + LEVEL_5 = 0x05 + AUTO = 0xFE + MANUAL = 0xFF + + +@dataclass +class CoolingStatus: + version: int = 0 + mode: int = 0 + fan_speed_percent: int = 0 + fan_speed: int = 0 + pump_speed_percent: int = 0 + pump_speed: int = 0 + water_flow: int = 0 + in_water_temp: int = 0 + out_water_temp: int = 0 + status_flag: int = 0 + rgb_mode: int = 0 + rgb_enable: bool = False + rgb_light_level: int = 0 + rgb_r: int = 0 + rgb_g: int = 0 + rgb_b: int = 0 + fan_curve: list[tuple[int, int]] = field( + default_factory=lambda: [(0, 0)] * 9 + ) + + @classmethod + def from_bytes(cls, data: bytes | bytearray) -> "CoolingStatus": + if len(data) < 41: + raise ValueError(f"Need >=41 bytes, got {len(data)}") + s = cls() + s.version = data[2] + s.mode = data[4] + s.fan_speed_percent = data[5] + s.fan_speed = (data[6] << 8) | data[7] + s.pump_speed_percent = data[8] + s.pump_speed = (data[9] << 8) | data[10] + s.water_flow = (data[11] << 8) | data[12] + s.in_water_temp = data[13] + s.out_water_temp = data[14] + s.status_flag = data[15] + s.rgb_mode = data[16] + s.rgb_enable = data[17] == 1 + s.rgb_light_level = data[19] + s.rgb_r = data[20] + s.rgb_g = data[21] + s.rgb_b = data[22] + s.fan_curve = [] + idx = 23 + for _ in range(9): + if idx + 1 < len(data): + s.fan_curve.append((data[idx], data[idx + 1])) + idx += 2 + else: + s.fan_curve.append((0, 0)) + return s + + def to_write_bytes(self, current: bytes | bytearray) -> bytearray: + out = bytearray(current) + out[1] = WRITE_CMD + out[2] = self.version or current[2] + out[4] = self.mode + # Preserve read-only status bytes (5, 8). + out[15] = self.status_flag + out[16] = self.rgb_mode + out[17] = 1 if self.rgb_enable else 0 + out[19] = self.rgb_light_level + out[20] = self.rgb_r + out[21] = self.rgb_g + out[22] = self.rgb_b + idx = 23 + for f, t in self.fan_curve: + out[idx] = f + out[idx + 1] = t + idx += 2 + return out + + def __str__(self) -> str: + lines = [ + f"CoolingStatus v{self.version} mode=0x{self.mode:02X}", + f" Fan: {self.fan_speed_percent:3d}% {self.fan_speed:5d} RPM", + f" Pump: {self.pump_speed_percent:3d}% {self.pump_speed:5d} RPM", + f" Flow: {self.water_flow}", + f" Temp: in={self.in_water_temp}C out={self.out_water_temp}C", + f" RGB: mode=0x{self.rgb_mode:02X} en={self.rgb_enable} lvl={self.rgb_light_level} ({self.rgb_r},{self.rgb_g},{self.rgb_b})", + f" Flag: 0x{self.status_flag:02X}", + " Curve:", + ] + for i, (f, t) in enumerate(self.fan_curve, 1): + lines.append(f" f{i}={f:3d}% @ {t:3d}C") + return "\n".join(lines) + + +def build_write_chunks(state: bytes | bytearray) -> list[bytes]: + """Split a modified 64-byte state into the 3 chunked write frames. + + The dock only accepts writes as 3 x 20-byte frames with 0x1C/0x2C/0x3C + headers; a single 64-byte write is silently ignored. The 58-byte payload + is ``[0x02] + state[2:59]`` (byte 57 of the payload is unused). + """ + payload = bytearray(WRITE_PAYLOAD_SIZE) + payload[0] = WRITE_CMD + payload[1:] = state[2:59] + chunks = [] + for i, header in enumerate(CHUNK_HEADERS): + start = i * CHUNK_SIZE + end = start + CHUNK_SIZE + chunks.append(bytes([header]) + bytes(payload[start:end])) + return chunks diff --git a/src/hhd/plugins/cooling_dock/settings.yml b/src/hhd/plugins/cooling_dock/settings.yml new file mode 100644 index 000000000..f41441936 --- /dev/null +++ b/src/hhd/plugins/cooling_dock/settings.yml @@ -0,0 +1,135 @@ +type: container +title: Cooling Dock +tags: [non-essential] +children: + enabled: + type: bool + title: "Enable Cooling Dock Sync" + hint: | + Automatically connect to the OneXPlayer Cooling Dock over Bluetooth + and sync the fan speed based on CPU temperature. Requires bleak + to be installed and the dock to be paired via bluetoothctl. + default: true + + scan_dock: + type: action + title: "Scan for Devices" + hint: "Click to scan for Bluetooth devices. Will update the list below." + + mac_address: + type: multiple + title: "Dock Device" + hint: "Select your dock from the list." + options: + "": "None" + default: "" + + forget_dock: + type: action + title: "Forget Paired Dock" + hint: "Clears the saved MAC address and forces a new scan." + + status: + type: display + title: "Status" + tags: [slim] + + fan_progress: + type: custom + title: "Dock Fan" + tags: [progress, slim] + + mode: + type: multiple + title: "Fan Mode" + hint: "Select the dock fan operating mode." + options: + "auto": "Auto (CPU temperature curve)" + "1": "Level 1 (25%)" + "2": "Level 2 (50%)" + "3": "Level 3 (75%)" + "4": "Level 4 (100%)" + "0": "Stopped" + default: "auto" + + fan_curve: + type: container + title: "Auto Fan Curve" + hint: "Fan speed percentage at each temperature threshold (Celsius)." + tags: [advanced] + children: + t1: + type: discrete + title: "Temp point 1 (C)" + options: [30, 35, 40, 45] + default: 40 + f1: + type: discrete + title: "Fan % at point 1" + options: [0, 10, 20, 25, 30] + default: 0 + t2: + type: discrete + title: "Temp point 2 (C)" + options: [45, 50, 55, 60] + default: 50 + f2: + type: discrete + title: "Fan % at point 2" + options: [20, 30, 40, 50] + default: 30 + t3: + type: discrete + title: "Temp point 3 (C)" + options: [55, 60, 65, 70] + default: 60 + f3: + type: discrete + title: "Fan % at point 3" + options: [40, 50, 60, 70] + default: 50 + t4: + type: discrete + title: "Temp point 4 (C)" + options: [65, 70, 75, 80] + default: 70 + f4: + type: discrete + title: "Fan % at point 4" + options: [60, 70, 80, 85] + default: 70 + t5: + type: discrete + title: "Temp point 5 (C)" + options: [75, 80, 85, 90] + default: 80 + f5: + type: discrete + title: "Fan % at point 5" + options: [80, 85, 90, 100] + default: 85 + + rgb: + type: container + title: "RGB Lighting" + hint: "Control the dock's RGB lighting." + children: + enable: + type: bool + title: "Enable RGB" + default: true + mode: + type: multiple + title: "RGB Mode" + options: + "0": "Static" + "1": "Breathing" + "2": "Rainbow" + "3": "Wave" + "4": "Pulse" + default: "1" + level: + type: discrete + title: "Brightness" + options: [0, 1, 2, 3, 4, 5] + default: 3 diff --git a/tests/test_adjustor_autodetect.py b/tests/test_adjustor_autodetect.py index a8c8c29da..c99c96d26 100644 --- a/tests/test_adjustor_autodetect.py +++ b/tests/test_adjustor_autodetect.py @@ -52,6 +52,41 @@ def test_legacy_vendor_match_keeps_precedence(self): ) unified.assert_not_called() + def test_oxp_superx_matches_dev_data_with_dock_aware(self): + def side_effect(file, *args, **kwargs): + if file == "/sys/devices/virtual/dmi/id/product_name": + return mock_open(read_data="ONEXPLAYER SUPER X").return_value + if file == "/sys/devices/virtual/dmi/id/board_name": + return mock_open(read_data="ONEXPLAYER SUPER X").return_value + if file == "/proc/cpuinfo": + return mock_open(read_data="AMD Ryzen 7 8840U").return_value + return mock_open(read_data="").return_value + + smu = MagicMock(name="smu") + qam = MagicMock(name="qam") + + with ( + patch("builtins.open", side_effect=side_effect), + patch("adjustor.hhd.USE_UNIFIED", False), + patch("adjustor.hhd.ASUS_DATA", {}), + patch("adjustor.hhd.MSI_DATA", {}), + patch("adjustor.drivers.smu.SmuDriverPlugin", return_value=smu) as smu_cls, + patch("adjustor.drivers.smu.SmuQamPlugin", return_value=qam) as qam_cls, + ): + plugins = autodetect([]) + + self.assertIn(smu, plugins) + self.assertIn(qam, plugins) + # SUPER X / APEX are dock-aware: dock state relaxes the TDP cap + self.assertEqual(smu_cls.call_args.kwargs["dock_aware"], True) + self.assertEqual(qam_cls.call_args.kwargs["dock_aware"], True) + # DC cap should be passed for battery-mode TDP limiting + self.assertIsNotNone(smu_cls.call_args.kwargs["dc_cap"]) + self.assertIsNotNone(qam_cls.call_args.kwargs["dc_cap"]) + self.assertTrue( + any(isinstance(plugin, AdjustorInitPlugin) for plugin in plugins) + ) + if __name__ == "__main__": unittest.main() diff --git a/tests/test_adjustor_smu_dock.py b/tests/test_adjustor_smu_dock.py new file mode 100644 index 000000000..61e6afe22 --- /dev/null +++ b/tests/test_adjustor_smu_dock.py @@ -0,0 +1,328 @@ +import unittest +from unittest.mock import patch + +from adjustor.core.const import ( + ALIB_PARAMS_AIMAX, + DC_CAP_OXP_SUPERX, + DEV_PARAMS_OXP_SUPERX, + ENERGY_MAP_OXP_SUPERX, +) +from adjustor.drivers.smu import SmuDriverPlugin, SmuQamPlugin +from hhd.plugins.conf import Config + + +def make_conf(dock_running: bool, enforce_limits: bool = True) -> Config: + conf = Config( + { + "hhd.settings.tdp_ready": True, + "hhd.settings.enforce_limits": enforce_limits, + "cooling_dock.dock_running": dock_running, + "tdp.qam.tdp": 75, + "tdp.qam.boost": False, + "tdp.qam.fan.mode": "disabled", + } + ) + return conf + + +def make_driver_conf(dock_running: bool, tdp: int = 120, apply: bool = False) -> Config: + conf = Config( + { + "hhd.settings.tdp_ready": True, + "hhd.settings.enforce_limits": True, + "cooling_dock.dock_running": dock_running, + "tdp.smu.std": { + "stapm_limit": tdp, + "skin_limit": tdp, + "slow_limit": tdp, + "fast_limit": tdp, + "slow_time": 10, + "stapm_time": 200, + }, + "tdp.smu.adv": {"enable": False}, + "tdp.smu.apply": apply, + "tdp.smu.platform_profile": "disabled", + "tdp.smu.energy_policy": "balanced", + } + ) + return conf + + +class SmuQamDockTest(unittest.TestCase): + def setUp(self): + self.p = SmuQamPlugin( + DEV_PARAMS_OXP_SUPERX, + ENERGY_MAP_OXP_SUPERX, + dock_aware=True, + dc_cap=DC_CAP_OXP_SUPERX, + ) + self.p.enabled = True + self.p.settings() + + def test_no_dock_enforces_safe_limits(self): + with patch("adjustor.drivers.smu.get_fan_info", return_value=None): + self.p.open(None, None) + self.p.update(make_conf(dock_running=False)) + self.assertTrue(self.p.enforce_limits) + + def test_dock_running_relaxes_limits(self): + with patch("adjustor.drivers.smu.get_fan_info", return_value=None): + self.p.open(None, None) + self.p.update(make_conf(dock_running=True)) + self.assertFalse(self.p.enforce_limits) + + def test_dock_disconnect_re_enforces_and_clamps(self): + with patch("adjustor.drivers.smu.get_fan_info", return_value=None): + self.p.open(None, None) + # Dock running: relaxed, user sets 120W + conf = make_conf(dock_running=True) + conf["tdp.qam.tdp"] = 120 + self.p.update(conf) + self.assertFalse(self.p.enforce_limits) + + # Dock disconnects: enforced again, 120W clamped to 75W + conf2 = make_conf(dock_running=False) + conf2["tdp.qam.tdp"] = 120 + self.p.update(conf2) + self.assertTrue(self.p.enforce_limits) + self.assertEqual(conf2["tdp.qam.tdp"].to(int), 75) + + def test_user_enforce_off_overrides_dock(self): + with patch("adjustor.drivers.smu.get_fan_info", return_value=None): + self.p.open(None, None) + # User disabled enforce_limits entirely: unlocked regardless of dock + self.p.update(make_conf(dock_running=False, enforce_limits=False)) + self.assertFalse(self.p.enforce_limits) + + +class SmuQamBatteryTest(unittest.TestCase): + """Tests for AC/DC power-aware TDP capping.""" + + def setUp(self): + self.p = SmuQamPlugin( + DEV_PARAMS_OXP_SUPERX, + ENERGY_MAP_OXP_SUPERX, + dock_aware=True, + dc_cap=DC_CAP_OXP_SUPERX, + ) + self.p.enabled = True + self.p.settings() + with patch("adjustor.drivers.smu.get_fan_info", return_value=None): + self.p.open(None, None) + + def test_battery_clamps_to_55w(self): + """On battery, TDP should be clamped to DC cap (55W), not AC cap (75W).""" + self.p.on_ac = False + conf = make_conf(dock_running=False) + conf["tdp.qam.tdp"] = 75 + self.p.update(conf) + self.assertEqual(conf["tdp.qam.tdp"].to(int), 55) + + def test_ac_allows_75w(self): + """On AC without dock, TDP should be allowed up to 75W.""" + self.p.on_ac = True + conf = make_conf(dock_running=False) + conf["tdp.qam.tdp"] = 75 + self.p.update(conf) + self.assertEqual(conf["tdp.qam.tdp"].to(int), 75) + + def test_ac_to_battery_transition_clamps(self): + """Unplugging AC should immediately clamp TDP from 75W to 55W.""" + self.p.on_ac = True + conf = make_conf(dock_running=False) + conf["tdp.qam.tdp"] = 75 + self.p.update(conf) + self.assertEqual(conf["tdp.qam.tdp"].to(int), 75) + + # Simulate AC -> DC event + self.p.notify([{"type": "acpi", "event": "dc"}]) + self.assertFalse(self.p.on_ac) + + # Next update should clamp + conf2 = make_conf(dock_running=False) + conf2["tdp.qam.tdp"] = 75 + self.p.update(conf2) + self.assertEqual(conf2["tdp.qam.tdp"].to(int), 55) + + def test_battery_to_ac_transition_does_not_increase(self): + """Plugging in AC should not auto-increase TDP, just relax the cap.""" + self.p.on_ac = False + conf = make_conf(dock_running=False) + conf["tdp.qam.tdp"] = 50 + self.p.update(conf) + # 50W is within both DC and AC range, should stay + self.assertEqual(conf["tdp.qam.tdp"].to(int), 50) + + # AC event + self.p.notify([{"type": "acpi", "event": "ac"}]) + conf2 = make_conf(dock_running=False) + conf2["tdp.qam.tdp"] = 50 + self.p.update(conf2) + # Should still be 50W (not auto-bumped) + self.assertEqual(conf2["tdp.qam.tdp"].to(int), 50) + + def test_battery_with_dock_still_limited_to_55w(self): + """On battery, even with dock running, TDP should not exceed DC cap. + + The dock running flag only relaxes enforce_limits when on AC power. + On battery (on_ac=False), limits must always remain enforced (55W) + to protect the battery from unsafe discharge currents.""" + self.p.on_ac = False + conf = make_conf(dock_running=True) + conf["tdp.qam.tdp"] = 120 + self.p.update(conf) + self.assertTrue(self.p.enforce_limits) + self.assertEqual(conf["tdp.qam.tdp"].to(int), 55) + + def test_status_message_on_battery(self): + """Status message should indicate battery-limited TDP.""" + self.p.on_ac = False + conf = make_conf(dock_running=False) + conf["tdp.qam.tdp"] = 50 + self.p.update(conf) + msg = conf["tdp.qam.sys_tdp"].to(str) + self.assertIn("battery", msg) + self.assertIn("55", msg) + + def test_status_message_dock_not_connected(self): + """On AC without dock, show dock-not-connected message.""" + self.p.on_ac = True + conf = make_conf(dock_running=False) + conf["tdp.qam.tdp"] = 50 + self.p.update(conf) + msg = conf["tdp.qam.sys_tdp"].to(str) + self.assertIn("dock", msg.lower()) + self.assertIn("75", msg) + + def test_no_dc_cap_ignores_battery(self): + """Without dc_cap (non-OXP devices), battery state has no effect.""" + p = SmuQamPlugin( + DEV_PARAMS_OXP_SUPERX, + ENERGY_MAP_OXP_SUPERX, + dock_aware=False, + dc_cap=None, + ) + p.enabled = True + p.settings() + with patch("adjustor.drivers.smu.get_fan_info", return_value=None): + p.open(None, None) + + p.on_ac = False + conf = make_conf(dock_running=False) + conf["tdp.qam.tdp"] = 75 + p.update(conf) + # Without dc_cap, 75W should still be allowed (smax=75) + self.assertEqual(conf["tdp.qam.tdp"].to(int), 75) + + +class SmuDriverDockTest(unittest.TestCase): + def make_plugin(self, dock_aware: bool = True) -> SmuDriverPlugin: + p = SmuDriverPlugin( + DEV_PARAMS_OXP_SUPERX, + ALIB_PARAMS_AIMAX, + dock_aware=dock_aware, + dc_cap=DC_CAP_OXP_SUPERX if dock_aware else None, + ) + p.enabled = True + p.initialized = True + p.has_pp = False + p.emit = lambda *args, **kwargs: None + return p + + def test_driver_dock_disconnect_forces_apply_with_clamped_values(self): + with patch("adjustor.drivers.smu.alib") as mock_alib: + p = self.make_plugin(dock_aware=True) + + # Dock running: relaxed, first update forces an apply (startup) + conf = make_driver_conf(dock_running=True, tdp=120) + p.update(conf) + self.assertFalse(p.enforce_limits) + mock_alib.assert_called_once() + mock_alib.reset_mock() + + # Dock disconnects: enforced, dock_changed forces immediate apply + # with the values clamped to the safe (smax) range. + conf2 = make_driver_conf(dock_running=False, tdp=120) + p.update(conf2) + self.assertTrue(p.enforce_limits) + mock_alib.assert_called_once() + args = mock_alib.call_args + self.assertEqual(args.kwargs["limit"], "device") + self.assertEqual(args.args[0]["stapm_limit"], 75) + self.assertEqual(args.args[0]["skin_limit"], 75) + self.assertEqual(args.args[0]["slow_limit"], 80) + self.assertEqual(args.args[0]["fast_limit"], 95) + + def test_driver_no_forced_apply_without_dock_change(self): + with patch("adjustor.drivers.smu.alib") as mock_alib: + p = self.make_plugin(dock_aware=False) + + # No dock, enforce_limits stays True (matches old_enforce init): + # no spurious apply on the first cycle. + conf = make_driver_conf(dock_running=False, tdp=75, apply=False) + p.update(conf) + self.assertTrue(p.enforce_limits) + mock_alib.assert_not_called() + + def test_driver_battery_clamps_to_dc_cap(self): + """On battery, SmuDriverPlugin should clamp values to DC cap.""" + with patch("adjustor.drivers.smu.alib") as mock_alib: + p = self.make_plugin(dock_aware=True) + p.on_ac = False + + conf = make_driver_conf(dock_running=False, tdp=75, apply=True) + p.update(conf) + # Values should be clamped to DC cap + mock_alib.assert_called_once() + vals = mock_alib.call_args.args[0] + self.assertLessEqual(vals["stapm_limit"], 55) + self.assertLessEqual(vals["skin_limit"], 55) + self.assertLessEqual(vals["fast_limit"], 70) + + def test_driver_ac_dc_event_updates_on_ac(self): + """notify() should update on_ac state.""" + p = self.make_plugin(dock_aware=True) + self.assertTrue(p.on_ac) + + p.notify([{"type": "acpi", "event": "dc"}]) + self.assertFalse(p.on_ac) + + p.notify([{"type": "acpi", "event": "ac"}]) + self.assertTrue(p.on_ac) + + +class SmuQamSettingsTest(unittest.TestCase): + def test_dock_aware_dynamically_adjusts_range(self): + p = SmuQamPlugin( + DEV_PARAMS_OXP_SUPERX, + ENERGY_MAP_OXP_SUPERX, + dock_aware=True, + dc_cap=DC_CAP_OXP_SUPERX, + ) + p.enabled = True + + # When enforcing limits (dock disconnected), exposes safe range (75W) + p.enforce_limits = True + p.on_ac = True + s = p.settings() + tdp = s["tdp"]["qam"]["children"]["tdp"] + self.assertEqual(tdp["min"], 4) + self.assertEqual(tdp["max"], 75) + + # When on battery and enforcing limits, exposes DC cap (55W) + p.on_ac = False + s = p.settings() + tdp = s["tdp"]["qam"]["children"]["tdp"] + self.assertEqual(tdp["max"], 55) + + # When NOT enforcing limits (dock connected), exposes full dmax (120W) + p.enforce_limits = False + p.on_ac = True + s = p.settings() + tdp = s["tdp"]["qam"]["children"]["tdp"] + self.assertEqual(tdp["min"], 0) + self.assertEqual(tdp["max"], 120) + + +if __name__ == "__main__": + unittest.main() \ No newline at end of file diff --git a/tests/test_adjustor_smu_dock_extended.py b/tests/test_adjustor_smu_dock_extended.py new file mode 100644 index 000000000..533293dae --- /dev/null +++ b/tests/test_adjustor_smu_dock_extended.py @@ -0,0 +1,110 @@ +import unittest +from unittest.mock import patch + +from adjustor.core.const import ( + ALIB_PARAMS_AIMAX, + DC_CAP_OXP_SUPERX, + DEV_PARAMS_OXP_SUPERX, + ENERGY_MAP_OXP_SUPERX, +) +from adjustor.drivers.smu import SmuDriverPlugin, SmuQamPlugin +from hhd.plugins.conf import Config + + +class SmuStartupAcDetectionTest(unittest.TestCase): + def test_qam_startup_on_battery_detects_dc(self): + p = SmuQamPlugin( + DEV_PARAMS_OXP_SUPERX, + ENERGY_MAP_OXP_SUPERX, + dock_aware=True, + dc_cap=DC_CAP_OXP_SUPERX, + ) + p.enabled = True + with ( + patch("hhd.utils.get_ac_status_fn", return_value="/sys/class/power_supply/AC/online"), + patch("hhd.utils.get_ac_status", return_value=False), + patch("adjustor.drivers.smu.get_fan_info", return_value=None), + ): + p.open(None, None) + + self.assertFalse(p.on_ac) + s = p.settings() + tdp = s["tdp"]["qam"]["children"]["tdp"] + self.assertEqual(tdp["max"], 55) + + def test_driver_startup_on_battery_detects_dc(self): + p = SmuDriverPlugin( + DEV_PARAMS_OXP_SUPERX, + ALIB_PARAMS_AIMAX, + dock_aware=True, + dc_cap=DC_CAP_OXP_SUPERX, + ) + p.enabled = True + p.initialized = True + with ( + patch("hhd.utils.get_ac_status_fn", return_value="/sys/class/power_supply/AC/online"), + patch("hhd.utils.get_ac_status", return_value=False), + ): + p.open(None, None) + + self.assertFalse(p.on_ac) + + +class SmuBoostCalculationTest(unittest.TestCase): + def test_boost_calculation_scales_accurately_on_ac(self): + p = SmuQamPlugin( + DEV_PARAMS_OXP_SUPERX, + ENERGY_MAP_OXP_SUPERX, + dock_aware=True, + dc_cap=DC_CAP_OXP_SUPERX, + ) + p.enabled = True + p.on_ac = True + p.settings() + with patch("adjustor.drivers.smu.get_fan_info", return_value=None): + p.open(None, None) + + conf = Config({ + "hhd.settings.tdp_ready": True, + "hhd.settings.enforce_limits": True, + "cooling_dock.dock_running": False, + "tdp.qam.tdp": 75, + "tdp.qam.boost": True, + "tdp.qam.fan.mode": "disabled", + }) + p.update(conf) + # On AC: fmax=95, smax=75 -> fast_limit = 75 * (95/75) = 95 + self.assertEqual(conf["tdp.smu.std.fast_limit"].to(int), 95) + # slow_limit = min(75 + 2, 95) = 77 + self.assertEqual(conf["tdp.smu.std.slow_limit"].to(int), 77) + + def test_boost_calculation_scales_accurately_on_battery(self): + p = SmuQamPlugin( + DEV_PARAMS_OXP_SUPERX, + ENERGY_MAP_OXP_SUPERX, + dock_aware=True, + dc_cap=DC_CAP_OXP_SUPERX, + ) + p.enabled = True + p.on_ac = False + p.settings() + with patch("adjustor.drivers.smu.get_fan_info", return_value=None): + p.open(None, None) + + conf = Config({ + "hhd.settings.tdp_ready": True, + "hhd.settings.enforce_limits": True, + "cooling_dock.dock_running": False, + "tdp.qam.tdp": 55, + "tdp.qam.boost": True, + "tdp.qam.fan.mode": "disabled", + }) + p.update(conf) + # On Battery: fmax=70, smax=55 -> fast_limit = 55 * (70/55) = 70 + self.assertEqual(conf["tdp.smu.std.fast_limit"].to(int), 70) + # slow_limit = min(55 + 2, 70) = 57 + self.assertEqual(conf["tdp.smu.std.slow_limit"].to(int), 57) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_cooling_dock.py b/tests/test_cooling_dock.py new file mode 100644 index 000000000..71e3b485c --- /dev/null +++ b/tests/test_cooling_dock.py @@ -0,0 +1,869 @@ +import time +import unittest +from unittest.mock import mock_open, patch, AsyncMock, MagicMock + +from hhd.plugins.cooling_dock.base import ( + CoolingDockPlugin, + _is_supported_device, + SCAN_BACKOFF_MIN, + SCAN_BACKOFF_MAX, + SCAN_BACKOFF_FACTOR, + GATT_OP_TIMEOUT, + SYNC_RETRY_MAX, + SYNC_RETRY_DELAY, + RECONNECT_DELAY, + DOCK_RUNNING_GRACE, +) +from hhd.plugins.cooling_dock.protocol import ( + build_write_chunks, + WRITE_CMD, + CHUNK_HEADERS, + CHUNK_SIZE, + WRITE_RETRY_MAX, +) +from hhd.plugins.conf import Config + + +def make_dock_conf(stale_running: bool = False) -> Config: + return Config( + { + "cooling_dock.dock": { + "enabled": True, + "mode": "auto", + "fan_curve": { + "t1": 40, + "f1": 0, + "t2": 50, + "f2": 30, + "t3": 60, + "f3": 50, + "t4": 70, + "f4": 70, + "t5": 80, + "f5": 85, + }, + "rgb": {"enable": True, "mode": 1, "level": 3}, + }, + "cooling_dock.dock_running": stale_running, + "cooling_dock.dock.status": ( + "Connected - Fan 50% (1000 RPM)" if stale_running else None + ), + } + ) + + +def make_dock_device(mac: str = "AA:BB:CC:DD:EE:FF"): + """Build a BLEDevice-like object with an address for _find_dock tests.""" + device = MagicMock() + device.address = mac + device.name = "CoolingSystem_ONEC1" + return device + + +class CoolingDockPluginTest(unittest.TestCase): + def test_update_self_heals_stale_runtime_state(self): + p = CoolingDockPlugin() + # Simulate a previous session that saved dock_running=True while the + # dock was connected. On startup the plugin must clear it so the + # adjustor does not unlock TDP without an actual dock. + conf = make_dock_conf(stale_running=True) + p.update(conf) + + self.assertEqual(conf["cooling_dock.dock_running"].to(bool), False) + self.assertEqual(conf["cooling_dock.dock.status"].to(str), "Disconnected") + self.assertIsNone(conf["cooling_dock.dock.fan_progress"].conf) + + def test_update_does_not_rewrite_on_second_call(self): + p = CoolingDockPlugin() + conf = make_dock_conf() + p.update(conf) + conf.updated = False + p.update(conf) + self.assertFalse(conf.updated) + + def test_publish_dock_running_writes_on_change(self): + p = CoolingDockPlugin() + conf = make_dock_conf() + p.update(conf) + conf.updated = False + + p._publish_dock_running(True) + self.assertEqual(conf["cooling_dock.dock_running"].to(bool), True) + self.assertTrue(conf.updated) + + def test_publish_status_writes_status_and_progress(self): + p = CoolingDockPlugin() + conf = make_dock_conf() + p.update(conf) + conf.updated = False + + p._publish_status( + "Connected - Fan 50% (1000 RPM)", + {"value": 50, "max": 100, "unit": "%", "text": "Dock Fan"}, + ) + self.assertEqual( + conf["cooling_dock.dock.status"].to(str), + "Connected - Fan 50% (1000 RPM)", + ) + self.assertEqual( + conf["cooling_dock.dock.fan_progress"].to(dict), + {"value": 50, "max": 100, "unit": "%", "text": "Dock Fan"}, + ) + self.assertTrue(conf.updated) + + def test_publish_status_skips_unchanged(self): + p = CoolingDockPlugin() + conf = make_dock_conf() + p.update(conf) + conf.updated = False + + # Same as the initial self-healed state: no write, config stays clean + p._publish_status("Disconnected", None) + self.assertFalse(conf.updated) + + +class ScanBackoffTest(unittest.TestCase): + def test_initial_delay_is_min(self): + p = CoolingDockPlugin() + self.assertEqual(p._scan_delay, SCAN_BACKOFF_MIN) + + def test_backoff_increases(self): + p = CoolingDockPlugin() + # Simulate a failed scan cycle: the delay should increase + initial = p._scan_delay + p._scan_delay = min(p._scan_delay * SCAN_BACKOFF_FACTOR, SCAN_BACKOFF_MAX) + self.assertEqual(p._scan_delay, initial * SCAN_BACKOFF_FACTOR) + + def test_backoff_caps_at_max(self): + p = CoolingDockPlugin() + # Run many backoff steps + for _ in range(20): + p._scan_delay = min( + p._scan_delay * SCAN_BACKOFF_FACTOR, SCAN_BACKOFF_MAX + ) + self.assertLessEqual(p._scan_delay, SCAN_BACKOFF_MAX) + + +class DmiGateTest(unittest.TestCase): + def test_supported_device_superx(self): + with patch( + "builtins.open", + mock_open(read_data="ONEXPLAYER SUPER X"), + ): + self.assertTrue(_is_supported_device()) + + def test_supported_device_apex(self): + with patch( + "builtins.open", + mock_open(read_data="ONEXPLAYER APEX"), + ): + self.assertTrue(_is_supported_device()) + + def test_unsupported_device(self): + with patch( + "builtins.open", + mock_open(read_data="ROG Ally RC71L"), + ): + self.assertFalse(_is_supported_device()) + + def test_missing_dmi_file(self): + with patch("builtins.open", side_effect=FileNotFoundError): + self.assertFalse(_is_supported_device()) + + +class StickyPairingTest(unittest.TestCase): + def test_update_reads_mac_address(self): + p = CoolingDockPlugin() + conf = make_dock_conf() + conf["cooling_dock.dock.mac_address"] = "AA:BB:CC:DD:EE:FF" + p.update(conf) + self.assertEqual(p._mac_address, "AA:BB:CC:DD:EE:FF") + + def test_forget_dock_clears_mac_and_forces_reconnect(self): + p = CoolingDockPlugin() + conf = make_dock_conf() + conf["cooling_dock.dock.mac_address"] = "AA:BB:CC:DD:EE:FF" + conf["cooling_dock.dock.forget_dock"] = True + p.update(conf) + + self.assertEqual(p._mac_address, "") + self.assertEqual(conf["cooling_dock.dock.mac_address"].to(str), "") + self.assertFalse(conf["cooling_dock.dock.forget_dock"].to(bool)) + self.assertTrue(p._force_reconnect) + + +import asyncio + + +class FindDockTest(unittest.IsolatedAsyncioTestCase): + """Tests for the _find_dock discovery logic (Bugs 1 & 6).""" + + async def test_saved_mac_uses_find_device_by_address(self): + """With a saved MAC, _find_dock should verify the dock is in range + via find_device_by_address, not skip scanning entirely.""" + p = CoolingDockPlugin() + conf = make_dock_conf() + conf["cooling_dock.dock.mac_address"] = "AA:BB:CC:DD:EE:FF" + p.update(conf) + + mock_device = MagicMock() + mock_device.address = "AA:BB:CC:DD:EE:FF" + + with patch( + "hhd.plugins.cooling_dock.base.BleakScanner.find_device_by_address", + new_callable=AsyncMock, + return_value=mock_device, + ) as mock_find: + result = await p._find_dock() + mock_find.assert_called_once_with("AA:BB:CC:DD:EE:FF", timeout=5) + self.assertEqual(result.address, "AA:BB:CC:DD:EE:FF") + + async def test_saved_mac_not_in_range_returns_none(self): + """If the saved MAC is not in range, find_device_by_address returns + None and _find_dock returns None quickly (no 15s connect timeout).""" + p = CoolingDockPlugin() + conf = make_dock_conf() + conf["cooling_dock.dock.mac_address"] = "AA:BB:CC:DD:EE:FF" + p.update(conf) + + with patch( + "hhd.plugins.cooling_dock.base.BleakScanner.find_device_by_address", + new_callable=AsyncMock, + return_value=None, + ), patch( + "hhd.plugins.cooling_dock.base.BleakScanner.find_device_by_name", + new_callable=AsyncMock, + return_value=None, + ), patch.object( + p, "_find_dock_in_bluez_objects", return_value=None + ), patch.object( + p, "_bluez_start_discovery", new_callable=AsyncMock + ): + result = await p._find_dock() + self.assertIsNone(result) + + async def test_no_dock_selected_returns_none(self): + """Without a saved MAC, _find_dock should return None without scanning.""" + +class SyncLoopRetryTest(unittest.IsolatedAsyncioTestCase): + """Tests for the sync loop retry logic (Bug 2).""" + + async def test_transient_error_does_not_break_immediately(self): + """A single GATT error should not break the connection — the loop + should retry up to SYNC_RETRY_MAX times.""" + p = CoolingDockPlugin() + p.running = True + conf = make_dock_conf() + p.update(conf) + + # Mock the client and _find_dock to return a connected client + mock_client = MagicMock() + mock_client.is_connected = True + mock_client.connect = AsyncMock() + mock_client.disconnect = AsyncMock() + # First read fails, then all subsequent reads succeed + mock_client.read_gatt_char = AsyncMock( + side_effect=[Exception("transient BLE error")] + + [b"\x00" * 64] * 20 + ) + mock_client.write_gatt_char = AsyncMock() + + # Stop the loop after a few successful cycles by setting running=False + original_sleep = asyncio.sleep + + async def limited_sleep(t): + if mock_client.read_gatt_char.call_count > 3: + p.running = False + await original_sleep(0) + + with patch.object( + p, "_find_dock", new_callable=AsyncMock, return_value=make_dock_device() + ), patch( + "hhd.plugins.cooling_dock.base.BleakClient", return_value=mock_client + ), patch( + "hhd.plugins.cooling_dock.base.asyncio.sleep", side_effect=limited_sleep + ): + await p._connect_and_sync() + # read_gatt_char should have been called more than once (error + retry) + self.assertGreater(mock_client.read_gatt_char.call_count, 1) + # Should NOT have disconnected due to errors — the single error + # was retried and succeeded. Disconnect happened because we set + # running=False to stop the loop. + self.assertLess( + mock_client.read_gatt_char.call_count, SYNC_RETRY_MAX * 3 + ) + + async def test_breaks_after_max_consecutive_errors(self): + """After SYNC_RETRY_MAX consecutive errors, the loop should break + and disconnect.""" + p = CoolingDockPlugin() + p.running = True + conf = make_dock_conf() + p.update(conf) + + mock_client = MagicMock() + mock_client.is_connected = True + mock_client.connect = AsyncMock() + mock_client.disconnect = AsyncMock() + mock_client.read_gatt_char = AsyncMock( + side_effect=Exception("persistent BLE error") + ) + mock_client.write_gatt_char = AsyncMock() + + with patch.object( + p, "_find_dock", new_callable=AsyncMock, return_value=make_dock_device() + ), patch( + "hhd.plugins.cooling_dock.base.BleakClient", return_value=mock_client + ), patch( + "hhd.plugins.cooling_dock.base.asyncio.sleep", new_callable=AsyncMock + ): + await p._connect_and_sync() + # Should have tried SYNC_RETRY_MAX times before breaking + self.assertEqual( + mock_client.read_gatt_char.call_count, SYNC_RETRY_MAX + ) + # Should have disconnected + mock_client.disconnect.assert_called_once() + + async def test_gatt_timeout_handled_as_retryable(self): + """An asyncio.TimeoutError on GATT read should be caught and retried, + not crash the loop.""" + p = CoolingDockPlugin() + p.running = True + conf = make_dock_conf() + p.update(conf) + + mock_client = MagicMock() + mock_client.is_connected = True + mock_client.connect = AsyncMock() + mock_client.disconnect = AsyncMock() + # First read times out, then all subsequent reads succeed + mock_client.read_gatt_char = AsyncMock( + side_effect=[asyncio.TimeoutError()] + [b"\x00" * 64] * 20 + ) + mock_client.write_gatt_char = AsyncMock() + + original_sleep = asyncio.sleep + + async def limited_sleep(t): + if mock_client.read_gatt_char.call_count > 3: + p.running = False + await original_sleep(0) + + with patch.object( + p, "_find_dock", new_callable=AsyncMock, return_value=make_dock_device() + ), patch( + "hhd.plugins.cooling_dock.base.BleakClient", return_value=mock_client + ), patch( + "hhd.plugins.cooling_dock.base.asyncio.sleep", side_effect=limited_sleep + ): + await p._connect_and_sync() + # Should have retried after the timeout (more than 1 call) + self.assertGreater(mock_client.read_gatt_char.call_count, 1) + self.assertLess( + mock_client.read_gatt_char.call_count, SYNC_RETRY_MAX * 3 + ) + + +class DisconnectedCallbackTest(unittest.IsolatedAsyncioTestCase): + """Tests for the disconnected_callback (Bug 7).""" + + async def test_disconnected_callback_breaks_sync_loop(self): + """When the BLE link drops, the disconnected_callback should fire + and break the sync loop immediately.""" + p = CoolingDockPlugin() + p.running = True + conf = make_dock_conf() + p.update(conf) + + mock_client = MagicMock() + mock_client.is_connected = True + mock_client.connect = AsyncMock() + mock_client.disconnect = AsyncMock() + mock_client.read_gatt_char = AsyncMock(return_value=b"\x00" * 64) + mock_client.write_gatt_char = AsyncMock() + + # Capture the disconnected_callback passed to BleakClient + captured_callback = {} + + def capture_client(addr, **kwargs): + captured_callback["cb"] = kwargs.get("disconnected_callback") + return mock_client + + with patch.object( + p, "_find_dock", new_callable=AsyncMock, return_value=make_dock_device() + ), patch( + "hhd.plugins.cooling_dock.base.BleakClient", side_effect=capture_client + ), patch( + "hhd.plugins.cooling_dock.base.asyncio.sleep", new_callable=AsyncMock + ) as mock_sleep: + # Make the first sleep trigger the disconnect callback + async def trigger_disconnect(t): + if captured_callback.get("cb"): + captured_callback["cb"](mock_client) + + mock_sleep.side_effect = trigger_disconnect + + await p._connect_and_sync() + # The callback should have been set + self.assertIsNotNone(captured_callback.get("cb")) + # Should have disconnected + mock_client.disconnect.assert_called_once() + + +class ReconnectDelayTest(unittest.IsolatedAsyncioTestCase): + """Tests for the reconnect delay after disconnect (Bug 3).""" + + async def test_delay_after_disconnect(self): + """After disconnecting, _connect_and_sync should wait RECONNECT_DELAY + seconds before returning (to let BlueZ clean up).""" + p = CoolingDockPlugin() + p.running = True + conf = make_dock_conf() + p.update(conf) + + mock_client = MagicMock() + mock_client.is_connected = True + mock_client.connect = AsyncMock() + mock_client.disconnect = AsyncMock() + mock_client.read_gatt_char = AsyncMock( + side_effect=Exception("connection lost") + ) + mock_client.write_gatt_char = AsyncMock() + + sleep_calls = [] + + async def track_sleep(t): + sleep_calls.append(t) + + with patch.object( + p, "_find_dock", new_callable=AsyncMock, return_value=make_dock_device() + ), patch( + "hhd.plugins.cooling_dock.base.BleakClient", return_value=mock_client + ), patch( + "hhd.plugins.cooling_dock.base.asyncio.sleep", side_effect=track_sleep + ): + await p._connect_and_sync() + # After SYNC_RETRY_MAX errors, the loop breaks and disconnects. + # Then RECONNECT_DELAY 1-second sleeps should follow. + one_second_sleeps = sum(1 for t in sleep_calls if t == 1) + self.assertGreaterEqual(one_second_sleeps, RECONNECT_DELAY) + + +class ChunkedWriteProtocolTest(unittest.TestCase): + """Tests for the chunked write protocol (PR #321).""" + + def test_build_write_chunks_headers_and_payload(self): + """The state must be split into 3 x 20-byte frames with 0x1C/0x2C/0x3C + headers and payload [0x02] + state[2:59].""" + state = bytearray(range(64)) + chunks = build_write_chunks(state) + + self.assertEqual(len(chunks), 3) + for i, chunk in enumerate(chunks): + self.assertEqual(len(chunk), 20) + self.assertEqual(chunk[0], CHUNK_HEADERS[i]) + + # chunk_1 = [0x1C] + payload[0:19], payload[0] = 0x02 + self.assertEqual(chunks[0][1], WRITE_CMD) + self.assertEqual(chunks[0][2], state[2]) + self.assertEqual(chunks[0][19], state[19]) + + # chunk_2 = [0x2C] + payload[19:38] = state[20:39] + self.assertEqual(chunks[1][1], state[20]) + self.assertEqual(chunks[1][19], state[38]) + + # chunk_3 = [0x3C] + payload[38:57] = state[39:58] + self.assertEqual(chunks[2][1], state[39]) + self.assertEqual(chunks[2][19], state[57]) + + def test_build_write_chunks_carries_mode_byte(self): + """A mode change at state[4] must appear in the first chunk.""" + state = bytearray(64) + state[4] = 0xFE # AUTO + chunks = build_write_chunks(state) + # payload[3] = state[4] -> chunk_1[4] + self.assertEqual(chunks[0][4], 0xFE) + + def test_build_write_chunks_carries_curve(self): + """The fan curve at state[23:41] must appear across the payload.""" + state = bytearray(64) + for i in range(23, 41): + state[i] = i + chunks = build_write_chunks(state) + # payload[k] = state[k+1]; state[23] = payload[22]. + # chunk_2 = [0x2C] + payload[19:38] -> payload[22] = chunk_2[4] + self.assertEqual(chunks[1][4], 23) + # state[40] = payload[39]; chunk_3 = [0x3C] + payload[38:57] + # -> payload[39] = chunk_3[2] + self.assertEqual(chunks[2][2], 40) + + +class WriteStateTest(unittest.IsolatedAsyncioTestCase): + """Tests for the plugin's chunked write path.""" + + async def test_write_state_sends_three_chunks(self): + """_write_state must send 3 sequential chunked writes, not one + single 64-byte write.""" + p = CoolingDockPlugin() + mock_client = MagicMock() + mock_client.write_gatt_char = AsyncMock() + + state = bytearray(64) + state[4] = 0x03 + + with patch( + "hhd.plugins.cooling_dock.base.asyncio.sleep", new_callable=AsyncMock + ): + await p._write_state(mock_client, state) + + self.assertEqual(mock_client.write_gatt_char.call_count, 3) + chunks = [c.args[1] for c in mock_client.write_gatt_char.call_args_list] + self.assertEqual([c[0] for c in chunks], list(CHUNK_HEADERS)) + for chunk in chunks: + self.assertEqual(len(chunk), 20) + + async def test_write_state_retries_on_error(self): + """A transient write error should be retried up to WRITE_RETRY_MAX + times before raising.""" + p = CoolingDockPlugin() + mock_client = MagicMock() + mock_client.write_gatt_char = AsyncMock( + side_effect=Exception("In Progress") + ) + + with patch( + "hhd.plugins.cooling_dock.base.asyncio.sleep", new_callable=AsyncMock + ): + with self.assertRaises(Exception): + await p._write_state(mock_client, bytearray(64)) + + # Each attempt fails on the first chunk, so one write per attempt. + self.assertEqual( + mock_client.write_gatt_char.call_count, WRITE_RETRY_MAX + ) + + async def test_write_state_succeeds_after_retry(self): + """If the first attempt fails but a later one succeeds, _write_state + should not raise.""" + p = CoolingDockPlugin() + mock_client = MagicMock() + # First chunk write fails once, then succeeds + calls = 0 + + async def flaky_write(*args, **kwargs): + nonlocal calls + calls += 1 + if calls == 1: + raise Exception("In Progress") + + mock_client.write_gatt_char = AsyncMock(side_effect=flaky_write) + + with patch( + "hhd.plugins.cooling_dock.base.asyncio.sleep", new_callable=AsyncMock + ): + await p._write_state(mock_client, bytearray(64)) + + # 1 failed chunk (attempt 1) + 3 successful chunks (attempt 2) = 4 + self.assertEqual(mock_client.write_gatt_char.call_count, 4) + + +class DockDropdownTest(unittest.IsolatedAsyncioTestCase): + """Tests for populating the 'Dock Device' dropdown.""" + + async def test_connect_adds_mac_to_dropdown(self): + """After a successful auto-connect, the dock's MAC must appear in + the 'Dock Device' dropdown options.""" + p = CoolingDockPlugin() + p.running = True + conf = make_dock_conf() + p.update(conf) + p.emit = MagicMock() + + mock_client = MagicMock() + mock_client.is_connected = True + mock_client.connect = AsyncMock() + mock_client.disconnect = AsyncMock() + mock_client.read_gatt_char = AsyncMock( + return_value=b"\x00" * 64 + ) + mock_client.write_gatt_char = AsyncMock() + + with patch.object( + p, "_find_dock", new_callable=AsyncMock, + return_value=make_dock_device(), + ), patch( + "hhd.plugins.cooling_dock.base.BleakClient", return_value=mock_client + ), patch( + "hhd.plugins.cooling_dock.base.asyncio.sleep", new_callable=AsyncMock + ): + # Stop the loop after the first sync cycle + original_sleep = asyncio.sleep + + async def stop_sleep(t): + p.running = False + + with patch( + "hhd.plugins.cooling_dock.base.asyncio.sleep", + side_effect=stop_sleep, + ): + await p._connect_and_sync() + + self.assertIn("AA:BB:CC:DD:EE:FF", p._discovered_macs) + # The MAC must be registered WITHOUT triggering a settings reload. + # The old code emitted {"type": "settings"} here, which caused a + # full settings reload + SMU re-apply on every connect cycle. + settings_calls = [ + c for c in p.emit.call_args_list + if c == unittest.mock.call({"type": "settings"}) + ] + self.assertEqual( + len(settings_calls), 0, + "Settings emit should NOT fire on MAC registration (causes reload storm)", + ) + + def test_settings_includes_connected_mac(self): + """settings() must include the connected MAC in the dropdown even if + it was never scanned.""" + p = CoolingDockPlugin() + p._mac_address = "AA:BB:CC:DD:EE:FF" + p._discovered_macs = {"": "Auto-detect"} + + base = p.settings() + opts = base["cooling_dock"]["dock"]["children"]["mac_address"]["options"] + self.assertIn("AA:BB:CC:DD:EE:FF", opts) + + +class BluezConnectedDockTest(unittest.TestCase): + """Tests for finding a connected-but-not-advertising dock via BlueZ.""" + + def test_finds_connected_dock(self): + """bluetoothctl devices + info should reveal a connected dock that + is not advertising.""" + p = CoolingDockPlugin() + devices_out = ( + "Device AA:BB:CC:DD:EE:FF CoolingSystem_ONEC1\n" + "Device 11:22:33:44:55:66 SomeOtherDevice\n" + ) + info_out = ( + "Device AA:BB:CC:DD:EE:FF\n" + "\tName: CoolingSystem_ONEC1\n" + "\tConnected: yes\n" + ) + with patch( + "subprocess.run", + side_effect=[ + MagicMock(stdout=devices_out), + MagicMock(stdout=info_out), + ], + ), patch( + "dbus.SystemBus", side_effect=Exception("no bus in test") + ): + result = p._find_connected_dock_via_bluez() + self.assertEqual(result.address, "AA:BB:CC:DD:EE:FF") + + def test_ignores_disconnected_dock(self): + """A known but disconnected dock should not be returned.""" + p = CoolingDockPlugin() + devices_out = "Device AA:BB:CC:DD:EE:FF CoolingSystem_ONEC1\n" + info_out = ( + "Device AA:BB:CC:DD:EE:FF\n" + "\tName: CoolingSystem_ONEC1\n" + "\tConnected: no\n" + ) + with patch( + "subprocess.run", + side_effect=[ + MagicMock(stdout=devices_out), + MagicMock(stdout=info_out), + ], + ), patch( + "dbus.SystemBus", side_effect=Exception("no bus in test") + ): + result = p._find_connected_dock_via_bluez() + self.assertIsNone(result) + + def test_respects_target_mac(self): + """When a target MAC is given, only that dock should be returned.""" + p = CoolingDockPlugin() + devices_out = ( + "Device AA:BB:CC:DD:EE:FF CoolingSystem_ONEC1\n" + "Device 11:22:33:44:55:66 CoolingSystem_ONEC1\n" + ) + info_out = ( + "Device 11:22:33:44:55:66\n" + "\tName: CoolingSystem_ONEC1\n" + "\tConnected: yes\n" + ) + with patch( + "subprocess.run", + side_effect=[ + MagicMock(stdout=devices_out), + MagicMock(stdout=info_out), + ], + ), patch( + "dbus.SystemBus", side_effect=Exception("no bus in test") + ): + result = p._find_connected_dock_via_bluez("11:22:33:44:55:66") + self.assertEqual(result.address, "11:22:33:44:55:66") + + +class WriteOnChangeTest(unittest.IsolatedAsyncioTestCase): + """Tests for the write-only-on-change sync behavior.""" + + async def test_writes_only_on_change(self): + """The sync loop should write when the target changes, then skip + writes while the target is unchanged.""" + p = CoolingDockPlugin() + p.running = True + conf = make_dock_conf() + p.update(conf) + + mock_client = MagicMock() + mock_client.is_connected = True + mock_client.connect = AsyncMock() + mock_client.disconnect = AsyncMock() + mock_client.read_gatt_char = AsyncMock(return_value=b"\x00" * 64) + mock_client.write_gatt_char = AsyncMock() + + original_sleep = asyncio.sleep + + async def limited_sleep(t): + if mock_client.read_gatt_char.call_count > 4: + p.running = False + await original_sleep(0) + + with patch.object( + p, "_find_dock", new_callable=AsyncMock, return_value=make_dock_device() + ), patch( + "hhd.plugins.cooling_dock.base.BleakClient", return_value=mock_client + ), patch( + "hhd.plugins.cooling_dock.base.asyncio.sleep", side_effect=limited_sleep + ): + await p._connect_and_sync() + + # 5 reads, but only 1 write (first cycle; target unchanged after) + self.assertEqual(mock_client.read_gatt_char.call_count, 5) + self.assertEqual(mock_client.write_gatt_char.call_count, 3) + + async def test_writes_again_when_mode_changes(self): + """Changing the mode should trigger a new write.""" + p = CoolingDockPlugin() + p.running = True + conf = make_dock_conf() + p.update(conf) + + mock_client = MagicMock() + mock_client.is_connected = True + mock_client.connect = AsyncMock() + mock_client.disconnect = AsyncMock() + mock_client.read_gatt_char = AsyncMock(return_value=b"\x00" * 64) + mock_client.write_gatt_char = AsyncMock() + + original_sleep = asyncio.sleep + read_count = 0 + + async def limited_sleep(t): + nonlocal read_count + if mock_client.read_gatt_char.call_count > 2 and read_count == 0: + read_count = 1 + # Change the mode mid-loop to trigger a new write + with p.conf_lock: + p.mode = "3" + if mock_client.read_gatt_char.call_count > 5: + p.running = False + await original_sleep(0) + + with patch.object( + p, "_find_dock", new_callable=AsyncMock, return_value=make_dock_device() + ), patch( + "hhd.plugins.cooling_dock.base.BleakClient", return_value=mock_client + ), patch( + "hhd.plugins.cooling_dock.base.asyncio.sleep", side_effect=limited_sleep + ): + await p._connect_and_sync() + + # Initial write + write after mode change = 2 writes (6 chunks) + self.assertEqual(mock_client.write_gatt_char.call_count, 6) + + +class DisconnectGraceTest(unittest.TestCase): + """Tests for the dock_running grace period (transient BLE flaps must + not flip dock_running, which would make the adjustor re-apply TDP and + cause SMU/ACPI spam + fan cycling).""" + + def test_does_not_publish_when_recently_connected(self): + p = CoolingDockPlugin() + conf = make_dock_conf() + p.update(conf) + p._dock_running = True + p._last_connected_time = time.time() # connected just now + p._status = "Connected" + conf["cooling_dock.dock_running"] = True + conf.updated = False + + p._publish_disconnected_if_stale() + + # No publish: dock_running stays True, config not dirtied + self.assertEqual(conf["cooling_dock.dock_running"].to(bool), True) + self.assertFalse(conf.updated) + + def test_publishes_after_grace_period(self): + p = CoolingDockPlugin() + conf = make_dock_conf() + p.update(conf) + conf.updated = False + p._dock_running = True + p._last_connected_time = time.time() - DOCK_RUNNING_GRACE - 1 + p._status = "Connected" + conf["cooling_dock.dock_running"] = True + + p._publish_disconnected_if_stale() + + self.assertEqual(conf["cooling_dock.dock_running"].to(bool), False) + self.assertEqual(conf["cooling_dock.dock.status"].to(str), "Disconnected") + self.assertTrue(conf.updated) + + +class ForgetDockTest(unittest.TestCase): + """The 'Forget Dock' action must also unpair from BlueZ, otherwise the + Trusted dock stays connected and fan control keeps working.""" + + def test_forget_clears_mac_and_unpairs_bluez(self): + p = CoolingDockPlugin() + conf = make_dock_conf() + conf["cooling_dock.dock.mac_address"] = "AA:BB:CC:DD:EE:FF" + p.update(conf) + + with patch.object(p, "_forget_bluez_device") as mock_forget: + conf["cooling_dock.dock.forget_dock"] = True + p.update(conf) + + mock_forget.assert_called_once() + self.assertEqual(conf["cooling_dock.dock.mac_address"].to(str), "") + self.assertEqual(conf["cooling_dock.dock.forget_dock"].to(bool), False) + + def test_forget_bluez_device_removes_from_adapter(self): + p = CoolingDockPlugin() + p._mac_address = "AA:BB:CC:DD:EE:FF" + + objects = { + "/org/bluez/hci0/dev_AA_BB_CC_DD_EE_FF": { + "org.bluez.Device1": {"Address": "AA:BB:CC:DD:EE:FF"} + } + } + mock_iface = MagicMock() + mock_iface.GetManagedObjects.return_value = objects + + with patch("dbus.Interface", return_value=mock_iface), patch( + "dbus.SystemBus" + ): + p._forget_bluez_device() + + mock_iface.RemoveDevice.assert_called_once_with( + "/org/bluez/hci0/dev_AA_BB_CC_DD_EE_FF" + ) + + +if __name__ == "__main__": + unittest.main() \ No newline at end of file diff --git a/tests/test_cooling_dock_extended.py b/tests/test_cooling_dock_extended.py new file mode 100644 index 000000000..2ae54e213 --- /dev/null +++ b/tests/test_cooling_dock_extended.py @@ -0,0 +1,110 @@ +import unittest +from unittest.mock import patch + +from hhd.plugins.cooling_dock.base import ( + fan_pct_for_temp, + get_cpu_temp, +) +from hhd.plugins.cooling_dock.protocol import ( + WRITE_CMD, + CoolingStatus, + DockMode, +) + + +class CoolingStatusProtocolTest(unittest.TestCase): + def test_from_bytes_short_raises_value_error(self): + with self.assertRaises(ValueError): + CoolingStatus.from_bytes(b"\x00" * 40) + + def test_from_bytes_full_parsing(self): + raw = bytearray(64) + raw[0] = 0xC1 + raw[1] = 0x10 + raw[2] = 6 # version + raw[4] = int(DockMode.LEVEL_3) + raw[5] = 75 # fan % + raw[6] = 0x08 # fan speed rpm hi + raw[7] = 0xD0 # fan speed rpm lo = 2256 + raw[8] = 50 # pump % + raw[9] = 0x06 + raw[10] = 0x30 # pump rpm = 1584 + raw[11] = 0x01 + raw[12] = 0xF4 # water flow = 500 + raw[13] = 25 # in temp + raw[14] = 28 # out temp + raw[15] = 0x01 # status flag + raw[16] = 0x02 # rgb mode + raw[17] = 0x01 # rgb enable + raw[19] = 0x04 # rgb level + raw[20] = 255 # r + raw[21] = 87 # g + raw[22] = 34 # b + # 9 curve pairs (fan%, temp) + for i in range(9): + raw[23 + i * 2] = 10 * (i + 1) + raw[23 + i * 2 + 1] = 30 + 5 * i + + status = CoolingStatus.from_bytes(raw) + self.assertEqual(status.version, 6) + self.assertEqual(status.mode, int(DockMode.LEVEL_3)) + self.assertEqual(status.fan_speed_percent, 75) + self.assertEqual(status.fan_speed, 2256) + self.assertEqual(status.pump_speed_percent, 50) + self.assertEqual(status.pump_speed, 1584) + self.assertEqual(status.water_flow, 500) + self.assertEqual(status.in_water_temp, 25) + self.assertEqual(status.out_water_temp, 28) + self.assertEqual(status.rgb_mode, 2) + self.assertTrue(status.rgb_enable) + self.assertEqual(status.rgb_light_level, 4) + self.assertEqual((status.rgb_r, status.rgb_g, status.rgb_b), (255, 87, 34)) + self.assertEqual(len(status.fan_curve), 9) + self.assertEqual(status.fan_curve[0], (10, 30)) + + def test_to_write_bytes_preserves_readonly_fields(self): + current = bytearray(64) + current[2] = 5 + current[5] = 80 # dock reports 80% fan + current[8] = 60 # dock reports 60% pump + current[14] = 35 + + status = CoolingStatus( + version=5, + mode=int(DockMode.AUTO), + rgb_enable=True, + rgb_mode=1, + rgb_light_level=3, + ) + out = status.to_write_bytes(current) + self.assertEqual(out[1], WRITE_CMD) + self.assertEqual(out[4], int(DockMode.AUTO)) + # Read-only fields must NOT be overwritten + self.assertEqual(out[5], 80) + self.assertEqual(out[8], 60) + self.assertEqual(out[14], 35) + + +class FanCurveCalculationTest(unittest.TestCase): + def test_fan_pct_interpolation(self): + curve = [(0, 40), (30, 50), (50, 60), (70, 70), (85, 80)] + self.assertEqual(fan_pct_for_temp(35, curve), 0) + self.assertEqual(fan_pct_for_temp(40, curve), 0) + self.assertEqual(fan_pct_for_temp(45, curve), 15) # halfway between 0 and 30 + self.assertEqual(fan_pct_for_temp(50, curve), 30) + self.assertEqual(fan_pct_for_temp(65, curve), 60) # halfway between 50 and 70 + self.assertEqual(fan_pct_for_temp(80, curve), 85) + self.assertEqual(fan_pct_for_temp(90, curve), 85) + + def test_empty_curve_returns_zero(self): + self.assertEqual(fan_pct_for_temp(50, []), 0) + + +class ExtendedDockPluginTest(unittest.TestCase): + def test_get_cpu_temp_fallback(self): + with patch("os.path.exists", return_value=False): + self.assertEqual(get_cpu_temp(), 0.0) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_cooling_dock_fixes.py b/tests/test_cooling_dock_fixes.py new file mode 100644 index 000000000..6316e7164 --- /dev/null +++ b/tests/test_cooling_dock_fixes.py @@ -0,0 +1,358 @@ +"""Tests for cooling dock reconnect loop fixes. + +Covers: +- _ensure_trusted_bluez() method existence and behavior +- Removal of settings emit calls in dock connection loop to prevent reload storms +- SmuQamPlugin correctly emitting settings on dock/power changes to dynamically update slider range +- TDP clamping works without settings reload +""" + +import time +import unittest +from unittest.mock import MagicMock, patch, call + +from hhd.plugins.cooling_dock.base import CoolingDockPlugin +from hhd.plugins.conf import Config + + +def make_dock_conf(**overrides) -> Config: + base = { + "cooling_dock.dock": { + "enabled": True, + "mode": "auto", + "fan_curve": { + "t1": 40, "f1": 0, + "t2": 50, "f2": 30, + "t3": 60, "f3": 50, + "t4": 70, "f4": 70, + "t5": 80, "f5": 85, + }, + "rgb": {"enable": True, "mode": 1, "level": 3}, + "mac_address": "", + }, + "cooling_dock.dock_running": False, + "cooling_dock.dock.status": None, + "cooling_dock.dock.fan_progress": None, + } + base.update(overrides) + return Config(base) + +class TestNoSettingsEmitOnConnect(unittest.TestCase): + """Verify the dock plugin does NOT emit {"type": "settings"} when + discovering new MACs, which was causing the reload storm.""" + + def test_no_emit_on_new_mac_registration(self): + """When a new MAC is registered on connect, no settings emit should + fire. The old code emitted settings here, causing a full reload + + SMU re-apply on every connect.""" + p = CoolingDockPlugin() + emit = MagicMock() + p.emit = emit + conf = make_dock_conf() + p.update(conf) + + # Simulate what _connect_and_sync does when it finds a new MAC: + # it registers the MAC in _discovered_macs. + addr = "C8:17:17:F5:C8:91" + with p.conf_lock: + p._discovered_macs[addr] = f"CoolingSystem_ONEC1 ({addr})" + + # The emit should NOT have been called with {"type": "settings"} + settings_calls = [ + c for c in emit.call_args_list + if c == call({"type": "settings"}) + ] + self.assertEqual( + len(settings_calls), 0, + "emit({'type': 'settings'}) should not be called on MAC registration", + ) + + def test_no_emit_on_scan_completion(self): + """After a manual scan populates _discovered_macs, no settings emit + should fire.""" + p = CoolingDockPlugin() + emit = MagicMock() + p.emit = emit + conf = make_dock_conf() + p.update(conf) + + # Simulate scan completion + with p.conf_lock: + p._discovered_macs = { + "": "Auto-detect", + "AA:BB:CC:DD:EE:FF": "CoolingSystem_ONEC1 (AA:BB:CC:DD:EE:FF)", + } + + settings_calls = [ + c for c in emit.call_args_list + if c == call({"type": "settings"}) + ] + self.assertEqual( + len(settings_calls), 0, + "emit({'type': 'settings'}) should not be called after scan", + ) + + +class TestSmuSettingsEmit(unittest.TestCase): + """Verify SmuQamPlugin DOES emit {"type": "settings"} on + dock_changed or power_changed to refresh the UI sliders.""" + + def _make_smu_plugin(self): + from adjustor.core.alib import DeviceParams + + dev = { + "skin_limit": DeviceParams(5, 15, 25, 55, 120), + "stapm_limit": DeviceParams(5, 15, 25, 55, 120), + "fast_limit": DeviceParams(5, 15, 25, 80, 140), + "slow_limit": DeviceParams(5, 15, 25, 60, 130), + } + from adjustor.drivers.smu import SmuQamPlugin + + return SmuQamPlugin(dev=dev, pp_map=None, dock_aware=True) + + def _make_smu_conf(self, dock_running=False): + return Config({ + "hhd.settings.tdp_ready": True, + "hhd.settings.enforce_limits": True, + "cooling_dock.dock_running": dock_running, + "tdp.qam.tdp": 25, + "tdp.qam.boost": False, + "tdp.smu.std.skin_limit": 25, + "tdp.smu.std.stapm_limit": 25, + "tdp.smu.std.fast_limit": 25, + "tdp.smu.std.slow_limit": 25, + "tdp.smu.apply": False, + "tdp.qam.sys_tdp": "", + }) + + def test_settings_emit_on_dock_changed(self): + """When dock_running changes, SmuQamPlugin must emit settings.""" + p = self._make_smu_plugin() + emit = MagicMock() + p.emit = emit + p.initialized = True + p.on_ac = True + + # First update with dock not running + conf = self._make_smu_conf(dock_running=False) + p.update(conf) + emit.reset_mock() + + # Second update with dock running — triggers dock_changed + conf2 = self._make_smu_conf(dock_running=True) + p.update(conf2) + + settings_calls = [ + c for c in emit.call_args_list + if c == call({"type": "settings"}) + ] + self.assertEqual( + len(settings_calls), 1, + "SmuQamPlugin must emit settings on dock_changed", + ) + + def test_settings_emit_on_power_changed(self): + """When on_ac changes, SmuQamPlugin must emit settings.""" + p = self._make_smu_plugin() + emit = MagicMock() + p.emit = emit + p.initialized = True + p.on_ac = True + + conf = self._make_smu_conf() + p.update(conf) + emit.reset_mock() + + # Simulate power change + p.on_ac = False + conf2 = self._make_smu_conf() + p.update(conf2) + + settings_calls = [ + c for c in emit.call_args_list + if c == call({"type": "settings"}) + ] + self.assertEqual( + len(settings_calls), 1, + "SmuQamPlugin must emit settings on power_changed", + ) + + def test_tdp_clamped_on_dock_disconnect_without_reload(self): + """TDP must be clamped when dock disconnects, even without a + settings reload. This verifies the clamping logic works standalone.""" + p = self._make_smu_plugin() + emit = MagicMock() + p.emit = emit + p.initialized = True + p.on_ac = True + + # Start with dock running, TDP at 80W (above 55W safe limit) + conf = self._make_smu_conf(dock_running=True) + conf["tdp.qam.tdp"] = 80 + p.update(conf) + + # Now dock disconnects — enforce_limits becomes True again + conf2 = self._make_smu_conf(dock_running=False) + conf2["tdp.qam.tdp"] = 80 # still at 80W + p.update(conf2) + + # TDP should have been clamped to 55W (smax) + clamped = conf2["tdp.qam.tdp"].to(int) + self.assertLessEqual( + clamped, 55, + f"TDP should be clamped to ≤55W on dock disconnect, got {clamped}W", + ) + + def test_tdp_apply_triggered_on_dock_changed(self): + """Even without settings emit, tdp.smu.apply must be set True + when dock state changes so the SMU driver picks it up.""" + p = self._make_smu_plugin() + emit = MagicMock() + p.emit = emit + p.initialized = True + p.on_ac = True + + conf = self._make_smu_conf(dock_running=False) + p.update(conf) + + # Dock connects + conf2 = self._make_smu_conf(dock_running=True) + p.update(conf2) + + # The queued apply should have been set + # (it fires when queued time passes, we check is_set was cleared) + self.assertFalse(p.is_set, "is_set should be False, indicating a re-apply is queued") + + +class ConnectAndSyncIntegrationTest(unittest.TestCase): + """Integration test exercising the full _connect_and_sync lifecycle + with a fake BleakClient that simulates real BLE behavior sequences.""" + + def _make_plugin(self): + p = CoolingDockPlugin() + p.emit = MagicMock() + p.running = True + p.enabled = True + p._mac_address = "C8:17:17:F5:C8:91" + p.mode = "auto" + p.fan_curve = [(0, 40), (30, 50), (50, 60), (70, 70), (85, 80)] + p.rgb_enable = True + p.rgb_mode = 1 + p.rgb_level = 3 + return p + + def _make_gatt_response(self, fan_speed=1200, fan_pct=50): + """Build a minimal 64-byte GATT response mimicking the dock.""" + data = bytearray(64) + data[4] = 1 # mode = auto + data[5] = fan_pct + data[6] = (fan_speed >> 8) & 0xFF + data[7] = fan_speed & 0xFF + return bytes(data) + + @patch("hhd.plugins.cooling_dock.base.get_cpu_temp", return_value=55.0) + @patch("hhd.plugins.cooling_dock.base.fan_pct_for_temp", return_value=60) + def test_full_sync_cycle_with_disconnect(self, mock_fan, mock_temp): + """Simulate: discover -> connect -> 2 successful reads -> + disconnect callback fires -> sync loop exits cleanly.""" + import asyncio + from unittest.mock import AsyncMock + + p = self._make_plugin() + gatt_data = self._make_gatt_response() + + fake_client = MagicMock() + fake_client.is_connected = True + fake_client.connect = AsyncMock() + fake_client.write_gatt_char = AsyncMock() + fake_client.disconnect = AsyncMock() + + read_count = 0 + disconnect_cb = None + + async def fake_read(char_uuid, **kwargs): + nonlocal read_count + read_count += 1 + if read_count > 2: + fake_client.is_connected = False + if disconnect_cb: + disconnect_cb(fake_client) + raise Exception("BLE disconnected") + return gatt_data + + fake_client.read_gatt_char = fake_read + + fake_device = MagicMock() + fake_device.address = "C8:17:17:F5:C8:91" + + async def fake_find_dock(): + return fake_device + + with ( + patch.object(p, "_find_dock", side_effect=fake_find_dock), + patch("hhd.plugins.cooling_dock.base.BleakClient") as mock_bleak_cls, + patch("asyncio.sleep", new_callable=AsyncMock), + ): + def capture_client(device, timeout=15, disconnected_callback=None): + nonlocal disconnect_cb + disconnect_cb = disconnected_callback + return fake_client + mock_bleak_cls.side_effect = capture_client + + asyncio.run(p._connect_and_sync()) + + self.assertGreaterEqual(read_count, 2, "Should complete at least 2 GATT reads") + self.assertTrue(p._dock_running or p._last_connected_time > 0, + "Dock should have been marked running during sync") + fake_client.disconnect.assert_called() + + @patch("hhd.plugins.cooling_dock.base.get_cpu_temp", return_value=55.0) + @patch("hhd.plugins.cooling_dock.base.fan_pct_for_temp", return_value=60) + def test_transient_timeouts_recover(self, mock_fan, mock_temp): + """Simulate: 2 GATT timeouts followed by recovery. The sync loop + should NOT break because SYNC_RETRY_MAX=3.""" + import asyncio + from unittest.mock import AsyncMock + + p = self._make_plugin() + gatt_data = self._make_gatt_response() + + fake_client = MagicMock() + fake_client.is_connected = True + fake_client.connect = AsyncMock() + fake_client.write_gatt_char = AsyncMock() + fake_client.disconnect = AsyncMock() + + read_count = 0 + + async def fake_read(char_uuid, **kwargs): + nonlocal read_count + read_count += 1 + if read_count <= 2: + raise asyncio.TimeoutError("simulated GATT timeout") + if read_count == 3: + return gatt_data + p.running = False + return gatt_data + + fake_client.read_gatt_char = fake_read + + fake_device = MagicMock() + fake_device.address = "C8:17:17:F5:C8:91" + + async def fake_find_dock(): + return fake_device + + with ( + patch.object(p, "_find_dock", side_effect=fake_find_dock), + patch("hhd.plugins.cooling_dock.base.BleakClient") as mock_bleak_cls, + patch("asyncio.sleep", new_callable=AsyncMock), + ): + mock_bleak_cls.return_value = fake_client + asyncio.run(p._connect_and_sync()) + + self.assertGreaterEqual(read_count, 3, "Should recover after transient timeouts") + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_oxp_device_controller.py b/tests/test_oxp_device_controller.py new file mode 100644 index 000000000..56cde9284 --- /dev/null +++ b/tests/test_oxp_device_controller.py @@ -0,0 +1,150 @@ +import unittest +from unittest.mock import patch + +from hhd.device.oxp.base import OxpAtKbd +from hhd.device.oxp.const import ( + BTN_MAPPINGS, + CONFS, + get_default_config, +) +from hhd.device.oxp.hid_v1 import ( + gen_rgb_mode as gen_hid1_rgb_mode, +) +from hhd.device.oxp.hid_v1 import ( + gen_rgb_solid as gen_hid1_rgb_solid, +) +from hhd.device.oxp.hid_v1 import ( + gen_vibration, +) +from hhd.device.oxp.hid_v2 import ( + gen_rgb_mode as gen_hid2_rgb_mode, +) +from hhd.device.oxp.hid_v2 import ( + gen_rgb_solid as gen_hid2_rgb_solid, +) +from hhd.device.oxp.serial import ( + gen_brightness as gen_serial_brightness, +) +from hhd.device.oxp.serial import ( + gen_cmd as gen_serial_cmd, +) +from hhd.device.oxp.serial import ( + gen_rgb_mode as gen_serial_rgb_mode, +) +from hhd.device.oxp.serial import ( + gen_rgb_solid as gen_serial_rgb_solid, +) + + +class OxpDeviceConfigTest(unittest.TestCase): + def test_superx_and_apex_registered(self): + self.assertIn("ONEXPLAYER SUPER X", CONFS) + self.assertIn("ONEXPLAYER APEX", CONFS) + superx = CONFS["ONEXPLAYER SUPER X"] + self.assertEqual(superx["name"], "ONEXPLAYER SUPER X") + self.assertEqual(superx["protocol"], "mixed") + self.assertTrue(superx["hrtimer"]) + + def test_default_config_fallback(self): + conf = get_default_config("ONEXPLAYER SUPER X 2", "ONEXPLAYER") + self.assertEqual(conf["name"], "ONEXPLAYER SUPER X 2") + self.assertTrue(conf["untested"]) + self.assertTrue(conf["hrtimer"]) + + +class OxpAtKbdMacroTest(unittest.TestCase): + def test_turbo_macro_combination_emits_mode(self): + kbd = OxpAtKbd( + vid=[0x0001], + pid=[0x0001], + required=False, + grab=False, + btn_map=BTN_MAPPINGS, + ) + # Simulate pressing Left Ctrl, Left Meta, and Left Alt together + kbd.state["key_leftctrl"] = 1.0 + kbd.state["key_leftmeta"] = 1.0 + kbd.state["key_leftalt"] = 1.0 + + with patch("hhd.device.oxp.base.GenericGamepadEvdev.produce", return_value=[]): + evs = kbd.produce([]) + + # Should consume the individual modifiers and emit a mode button press + self.assertNotIn("key_leftctrl", kbd.state) + self.assertNotIn("key_leftmeta", kbd.state) + self.assertNotIn("key_leftalt", kbd.state) + + button_evs = [e for e in evs if e["type"] == "button" and e["value"] is True] + self.assertTrue(any(e["code"] == "mode" for e in button_evs)) + + +class OxpSerialProtocolTest(unittest.TestCase): + def test_gen_cmd_framing(self): + cmd = gen_serial_cmd(0xFD, [0x00, 0x01], size=64) + self.assertEqual(len(cmd), 64) + self.assertEqual(cmd[0], 0xFD) + self.assertEqual(cmd[1], 0x3F) + self.assertEqual(cmd[2], 0x00) + self.assertEqual(cmd[3], 0x01) + self.assertEqual(cmd[-2], 0x3F) + self.assertEqual(cmd[-1], 0xFD) + + def test_gen_rgb_mode(self): + cmd = gen_serial_rgb_mode("flowing") + self.assertEqual(cmd[0], 0xFD) + self.assertEqual(cmd[3], 0x03) + + def test_gen_rgb_solid(self): + cmd = gen_serial_rgb_solid(255, 128, 64, side=0x00) + self.assertEqual(cmd[0], 0xFD) + self.assertEqual(cmd[2], 0x00) + self.assertEqual(cmd[3], 0xFE) + self.assertEqual(cmd[6], 255) + self.assertEqual(cmd[7], 128) + self.assertEqual(cmd[8], 64) + + def test_gen_brightness(self): + cmd = gen_serial_brightness(0, True, "high") + self.assertEqual(cmd[0], 0xFD) + self.assertEqual(cmd[6], 1) + self.assertEqual(cmd[8], 0x04) + + +class OxpHidProtocolsTest(unittest.TestCase): + def test_hid_v1_rgb_mode(self): + cmd = gen_hid1_rgb_mode("sunset") + self.assertEqual(cmd[0], 0xB8) + self.assertEqual(cmd[1], 0x3F) + self.assertEqual(cmd[3], 0x0B) + + def test_hid_v1_rgb_solid(self): + cmd = gen_hid1_rgb_solid(10, 20, 30, side=0x00) + self.assertEqual(cmd[0], 0xB8) + self.assertEqual(cmd[3], 0xFE) + self.assertEqual(cmd[6], 10) + self.assertEqual(cmd[7], 20) + self.assertEqual(cmd[8], 30) + + def test_hid_v1_vibration(self): + cmd = gen_vibration(5) + self.assertEqual(cmd[0], 0xB3) + self.assertEqual(cmd[1], 0x3F) + + def test_hid_v2_rgb_mode(self): + cmd = gen_hid2_rgb_mode("neon") + self.assertEqual(cmd[0], 0x07) + self.assertEqual(cmd[1], 0xFF) + self.assertEqual(cmd[2], 0x05) + + def test_hid_v2_rgb_solid(self): + cmd = gen_hid2_rgb_solid(100, 150, 200) + self.assertEqual(cmd[0], 0x07) + self.assertEqual(cmd[1], 0xFF) + self.assertEqual(cmd[2], 0xFE) + self.assertEqual(cmd[3], 100) + self.assertEqual(cmd[4], 150) + self.assertEqual(cmd[5], 200) + + +if __name__ == "__main__": + unittest.main() diff --git a/usr/lib/udev/rules.d/83-hhd.rules b/usr/lib/udev/rules.d/83-hhd.rules index ff0150113..cb8013e9a 100644 --- a/usr/lib/udev/rules.d/83-hhd.rules +++ b/usr/lib/udev/rules.d/83-hhd.rules @@ -41,4 +41,11 @@ ATTRS{idVendor}=="1a86", ATTRS{idProduct}=="e310", RUN+="/sbin/modprobe xpad" RU # Banish Ally HID devices to oblivion since they crash SDL/Proton controller handlers SUBSYSTEMS=="usb|hidraw", ATTRS{idVendor}=="0b05", ATTRS{idProduct}=="1b4c", MODE="000", GROUP="root", TAG-="uaccess", RUN+="/bin/chmod 000 /dev/%k" -SUBSYSTEMS=="usb|hidraw", ATTRS{idVendor}=="0b05", ATTRS{idProduct}=="1abe", MODE="000", GROUP="root", TAG-="uaccess", RUN+="/bin/chmod 000 /dev/%k" \ No newline at end of file +SUBSYSTEMS=="usb|hidraw", ATTRS{idVendor}=="0b05", ATTRS{idProduct}=="1abe", MODE="000", GROUP="root", TAG-="uaccess", RUN+="/bin/chmod 000 /dev/%k" + +# Mute spurious Volume Down events from ONEXPLAYER Cooling Dock (Only on SUPER X or APEX) +SUBSYSTEM=="input", KERNEL=="event*", ATTRS{name}=="CoolingSystem*", PROGRAM="/bin/grep -Eiq 'ONEXPLAYER (SUPER X|APEX)' /sys/class/dmi/id/product_name", ATTR{inhibited}="1", ENV{ID_INPUT}="", MODE="000" + +# Block dock hidraw to prevent desktop BT managers from prompting for pairing. +# We only use the GATT service, matching Windows behavior. +KERNEL=="hidraw*", ATTRS{name}=="CoolingSystem*", PROGRAM="/bin/grep -Eiq 'ONEXPLAYER (SUPER X|APEX)' /sys/class/dmi/id/product_name", MODE="000", GROUP="root", TAG-="uaccess"