Commander ST fw 2.x: liquidctl workaround

README.md

Commander ST fw 2.x (tested on Corsair iCUE H170i ELITE CAPELLIX AIO) — single-shot apply workaround

A standalone Python script for users hitting either of these symptoms on a Corsair Commander Core / Commander ST running firmware 2.x:

1. liquidctl set <ch> speed T P T P … fails with IndexError: index out of range — _read_data hard-codes three HID read chunks (~181 bytes) and the device's full curve table for 7 channels at 7 points each exceeds that.
2. liquidctl status (or any set) times out after ~5-25 invocations, only recovers with usbreset — the device's firmware silently stops responding to HID after a burst of operations, while the USB stack and kernel hidraw layer remain healthy (dmesg is silent during the failure; lsusb -v still works).

Tracking issue: liquidctl/liquidctl#753.

What this is

A workaround, not a fix. It bypasses the upstream set_* CLI flow entirely and writes the device's three hardware-profile tables (SPEED_MODE, FIXED_PERCENT, CURVE_PERCENT) directly in one wake context, ~12 HID ops total — well under the device's firmware-lockup threshold. Because it doesn't pre-read any of those tables, it also dodges the _read_data 181-byte truncation.

Net effect: a single boot-time invocation applies your full fan/pump config and exits. The Commander ST then drives the fans autonomously from its onboard water-temp sensor; no daemon polls the device afterwards.

Trade-offs (read before using)

• Pokes private API. It imports _CMD_*, _send_command, etc. from liquidctl.driver.commander_core. Upstream doesn't guarantee that surface — a future refactor could break this script.
• Commander Core/ST-only. Hard-codes 7 channels (1 pump + 6 fan ports). Other Commander Core variants might work but are untested.
• Read-write only. It never reads back what's on the device. If anything else (iCUE on Windows, other tooling) has written settings the script then overwrites them — including channels you didn't list, which default to FIXED 0%.
• Not a fix for upstream. If the _read_data chunking and the firmware lockup both get addressed upstream, this script becomes redundant.

Files

• apply.py — the script
• apply.json.example — example config for an iCUE H170i ELITE CAPELLIX (Commander ST internal)

Usage

# adjust paths in apply.json to your setup; copy to /etc/ or wherever
sudo /usr/bin/python3 apply.py /path/to/apply.json

Or wire as a systemd oneshot at boot — see the liquidctl-curves.service template at the end of this README.

Config schema

{
  "channels": {
    "pump": {"fixed": 80},
    "fan4": {"curve": [[25, 30], [35, 50], [45, 75], [55, 100]]},
    "fan5": {"curve": [[25, 30], [35, 50], [45, 75], [55, 100]]},
    "fan6": {"curve": [[25, 30], [35, 50], [45, 75], [55, 100]]}
  }
}

Each channel takes either {"fixed": int_0_100} or {"curve": [[temp_c, duty_pct], ...]} (2–7 points). Omitted channels default to FIXED 0%. Channel names are pump, fan1..fan6.

Optional systemd unit

[Unit]
Description=Apply Corsair Commander ST fan and pump curves
After=multi-user.target
ConditionPathExists=/dev/hidraw0

[Service]
Type=oneshot
RemainAfterExit=yes
ExecStartPre=/usr/bin/usbreset 1b1c:0c32
ExecStartPre=/bin/sleep 3
ExecStart=/usr/bin/python3 /usr/local/bin/apply.py /etc/liquidctl-apply.json

[Install]
WantedBy=multi-user.target

Adjust the 1b1c:0c32 USB ID to your device (lsusb -d 1b1c:). The usbreset step recovers the device if its firmware locked up before this boot — without it, the script could time out on a stale lockup carried over from the previous session.

apply.json.example

{
  "channels": {
    "pump": {"fixed": 80},
    "fan4": {"curve": [[25, 30], [35, 50], [45, 75], [55, 100]]},
    "fan5": {"curve": [[25, 30], [35, 50], [45, 75], [55, 100]]},
    "fan6": {"curve": [[25, 30], [35, 50], [45, 75], [55, 100]]}
  }
}

apply.py

#!/usr/bin/env python3
"""
One-shot apply of fan curves + pump fixed speed to a Corsair Commander
Core / Commander ST.  Bypasses the upstream `liquidctl set …` CLI flow
to dodge two latent issues on firmware 2.x devices:

  1. `_read_data` hard-codes three HID chunks (~181 bytes) so the
     curve-table read in set_speed_profile silently truncates and the
     parser raises IndexError when 7 channels × 7 points exceed that.

  2. The device's firmware silently stops responding to HID after a
     burst of operations (~5-25 ops in a tight window) — see
     liquidctl/liquidctl#753.  A multi-invocation flow (initialize +
     N × set …) blows past that budget every boot.

This script opens the device once, drains pending reports, writes
SPEED_MODE + FIXED_PERCENT + CURVE_PERCENT directly (no read-back),
and exits.  Total cost: ~12 HID ops per boot.  Channels not listed
in the config default to FIXED 0% (no-op for unused ports).

USAGE:  sudo python3 apply.py [path/to/config.json]
        (default config path: /etc/liquidctl-apply.json)

CONFIG: {
          "channels": {
            "pump":  {"fixed": 80},
            "fan4":  {"curve": [[25, 30], [35, 50], [45, 75], [55, 100]]},
            ...
          }
        }

CAVEAT: pokes `_CMD_*` and `_send_command` from liquidctl's commander_core
        driver — private API, may break on a future liquidctl refactor.
"""
import json
import sys
from pathlib import Path

from liquidctl.driver import find_liquidctl_devices
from liquidctl.driver.commander_core import (
    CommanderCore,
    _CMD_OPEN_ENDPOINT, _CMD_CLOSE_ENDPOINT,
    _CMD_WRITE, _CMD_WRITE_MORE,
    _MODE_HW_SPEED_MODE, _DATA_TYPE_HW_SPEED_MODE,
    _MODE_HW_FIXED_PERCENT, _DATA_TYPE_HW_FIXED_PERCENT,
    _MODE_HW_CURVE_PERCENT, _DATA_TYPE_HW_CURVE_PERCENT,
    _FAN_MODE_FIXED_PERCENT, _FAN_MODE_CURVE_PERCENT,
    _REPORT_LENGTH,
)
from liquidctl.util import clamp

DEFAULT_CONFIG_PATH = "/etc/liquidctl-apply.json"
N_CHANNELS = 7                 # pump (0) + fan1..fan6 (1..6)
TEMP_SENSOR = 0x00             # AIO water-temp sensor
CHANNEL_ORDER = ("pump", "fan1", "fan2", "fan3", "fan4", "fan5", "fan6")
DUMMY_CURVE = [(25, 30), (60, 100)]
READ_DATA_LIMIT = 181          # upstream _read_data ceiling (3 chunks)


def write_endpoint_unsafe(dev, mode, data_type, payload):
    """Mirror of CommanderCore._write_data minus the read-back precheck
    that triggers the IndexError on oversized curve tables."""
    dev._send_command(_CMD_OPEN_ENDPOINT, mode)
    data_len = len(payload)
    i = 0
    while i < data_len:
        if i == 0:
            packet_len = min(_REPORT_LENGTH - 9, data_len)
            buf = bytearray(2 + 2 + len(data_type) + packet_len)
            buf[0:2] = (data_len + len(data_type)).to_bytes(2, "little")
            buf[4:4 + len(data_type)] = data_type
            buf[4 + len(data_type):] = payload[0:packet_len]
            dev._send_command(_CMD_WRITE, buf)
            i += packet_len
        else:
            packet_len = min(_REPORT_LENGTH - 3, data_len - i)
            dev._send_command(_CMD_WRITE_MORE, payload[i:i + packet_len])
            i += packet_len
    dev._send_command(_CMD_CLOSE_ENDPOINT)


def encode_curve(points):
    """Channel entry: [temp_sensor (1)] [num_points (1)] [(temp_le, duty_le) × N]."""
    buf = bytearray([TEMP_SENSOR, len(points)])
    for temp, duty in points:
        buf += int(round(temp * 10)).to_bytes(2, "little")
        buf += int(clamp(duty, 0, 100)).to_bytes(2, "little")
    return bytes(buf)


def resolve_channels(channels_cfg):
    resolved = []
    for idx, name in enumerate(CHANNEL_ORDER):
        cfg = channels_cfg.get(name, {})
        if "curve" in cfg and "fixed" in cfg:
            sys.exit(f"channel {name}: pick either 'curve' or 'fixed', not both")
        if "curve" in cfg:
            curve = [tuple(pt) for pt in cfg["curve"]]
            if not (2 <= len(curve) <= 7):
                sys.exit(f"channel {name}: curve must have 2..7 points (got {len(curve)})")
            resolved.append((idx, name, _FAN_MODE_CURVE_PERCENT, 0, curve))
        elif "fixed" in cfg:
            resolved.append((idx, name, _FAN_MODE_FIXED_PERCENT, int(cfg["fixed"]), DUMMY_CURVE))
        else:
            resolved.append((idx, name, _FAN_MODE_FIXED_PERCENT, 0, DUMMY_CURVE))
    return resolved


def build_payloads(resolved):
    speed_mode = bytearray([N_CHANNELS])
    fixed = bytearray([N_CHANNELS])
    curves = bytearray([N_CHANNELS])
    for _, _, mode, fixed_pct, curve in resolved:
        speed_mode.append(mode)
        fixed += int(clamp(fixed_pct, 0, 100)).to_bytes(2, "little")
        curves += encode_curve(curve)
    return bytes(speed_mode), bytes(fixed), bytes(curves)


def main():
    config_path = sys.argv[1] if len(sys.argv) > 1 else DEFAULT_CONFIG_PATH
    config = json.loads(Path(config_path).read_text())
    resolved = resolve_channels(config.get("channels", {}))
    speed_mode_payload, fixed_payload, curve_payload = build_payloads(resolved)

    if len(curve_payload) > READ_DATA_LIMIT:
        sys.exit(
            f"refusing — curve payload would be {len(curve_payload)} bytes "
            f"(>{READ_DATA_LIMIT}); future `liquidctl set <ch> speed …` calls "
            "would hit IndexError on the read-back step"
        )

    print(f"config:    {config_path}")
    for _, name, mode, fixed_pct, curve in resolved:
        if mode == _FAN_MODE_CURVE_PERCENT:
            pts = ", ".join(f"{t}°→{d}%" for t, d in curve)
            print(f"  {name:>5}  CURVE  {pts}")
        else:
            print(f"  {name:>5}  FIXED  {fixed_pct}%")
    print(f"payloads:  speed_mode={len(speed_mode_payload)}B  "
          f"fixed={len(fixed_payload)}B  curve={len(curve_payload)}B")

    devs = [d for d in find_liquidctl_devices() if isinstance(d, CommanderCore)]
    if not devs:
        sys.exit("no Commander Core / ST device found")
    dev = devs[0]
    print(f"device:    {dev.description}")

    dev.connect()
    try:
        with dev._wake_device_context():
            write_endpoint_unsafe(dev, _MODE_HW_SPEED_MODE,
                                  _DATA_TYPE_HW_SPEED_MODE, speed_mode_payload)
            write_endpoint_unsafe(dev, _MODE_HW_FIXED_PERCENT,
                                  _DATA_TYPE_HW_FIXED_PERCENT, fixed_payload)
            write_endpoint_unsafe(dev, _MODE_HW_CURVE_PERCENT,
                                  _DATA_TYPE_HW_CURVE_PERCENT, curve_payload)
        print("OK — settings applied")
    finally:
        dev.disconnect()


if __name__ == "__main__":
    main()