add thermal derating group test, include Manual/Auto test mode

docs/script_rules.md

@@ -0,0 +1,28 @@
+# Regression Test Script Rules
+
+These rules apply to every regression test script in `tests/` (pytest-based
+or plain-script style alike).
+
+## After every test group / test case
+
+1. **Restore default config values.**
+   Any register changed for the test (TPU `Debug` dataset config, RPU flash
+   config such as `Voltage_setting_Q128`/`Current_setting_Q128`, etc.) must be
+   restored to its original value before the script exits. Read the actual
+   value from the unit at setup time (not a hardcoded nominal constant) and
+   write it back on `close()`/`finally`, using the device's own
+   config-writable + rewrite flow (`wait_until_config_writable()` ->
+   `write()` -> `rewrite_config()`).
+
+2. **Don't turn off the eload/AC source - just switch back to local.**
+   Leaving TPU/RPU powered lets the operator continue on the bench without a
+   fresh power-up sequence. On close, set the voltage back to the bench-safe
+   default (50V or 48V, whichever the test group uses) then call
+   `set_local()` - never `output_off()`/`disconnect()` the supply itself
+   mid-session.
+
+3. **The script must print every step of the test as it runs.**
+   Anyone reading the terminal should be able to follow along without reading
+   the code: print before every voltage/current change, every charger
+   enable/disable, every wait, and every register read/write of consequence -
+   not just the final PASS/FAIL.

drivers/modbus_device.py

@@ -94,6 +94,14 @@ class ModbusDevice:
         val = self.read_input_register(address)
         return val - 65536 if val > 32767 else val
 
+    def read_discrete_input(self, address):
+        def _do():
+            res = self.client.read_discrete_inputs(address, count=1, device_id=self.slave_id)
+            if res.isError():
+                raise ModbusReadError(f"{self.device_name}: failed to read discrete input {address}")
+            return res.bits[0]
+        return self._retry(_do)
+
     def read_input_register_float(self, address):
         def _do():
             res = self.client.read_input_registers(address, count=2, device_id=self.slave_id)
@@ -115,14 +123,14 @@ class ModbusDevice:
                 raise ModbusWriteError(f"{self.device_name}: write coil {address}={value} not verified")
         self._retry(_do)
 
-    def write_holding_register(self, address, value):
+    def write_holding_register(self, address, value, verify=True):
         value = int(value) & 0xFFFF  # two's-complement wrap so negative (signed) values fit the uint16 wire format
 
         def _do():
             res = self.client.write_register(address, value, device_id=self.slave_id)
             if res.isError():
                 raise ModbusWriteError(f"{self.device_name}: failed to write holding {address}")
-            if self.read_holding_register(address) != value:
+            if verify and self.read_holding_register(address) != value:
                 raise ModbusWriteError(f"{self.device_name}: write holding {address}={value} not verified")
         self._retry(_do)
 
@@ -155,6 +163,8 @@ class ModbusDevice:
         reg = self._register(label)
         if reg.object_type == "coil":
             return self.read_coil(reg.address)
+        if reg.object_type == "discrete":
+            return self.read_discrete_input(reg.address)
         if reg.object_type == "input":
             if reg.data_type == "float":
                 return self.read_input_register_float(reg.address)
@@ -169,16 +179,16 @@ class ModbusDevice:
             return self.read_holding_register(reg.address)
         raise ModbusDeviceError(f"{self.device_name}: unsupported object_type {reg.object_type!r} for {label!r}")
 
-    def write(self, label, value):
+    def write(self, label, value, verify=True):
         reg = self._register(label)
         if reg.object_type == "coil":
-            self.write_coil(reg.address, value)
+            self.write_coil(reg.address, value, verify=verify)
             return
         if reg.object_type == "holding":
             if reg.data_type == "float":
                 self.write_holding_register_float(reg.address, value)
             else:
-                self.write_holding_register(reg.address, value)
+                self.write_holding_register(reg.address, value, verify=verify)
             return
         raise ModbusDeviceError(f"{self.device_name}: cannot write to object_type {reg.object_type!r} for {label!r}")
 

drivers/rpu.py

@@ -1,9 +1,15 @@
 import time
 
 from common.config_loader import load_garuda_registers
-from drivers.modbus_device import ModbusDevice
+from drivers.modbus_device import ModbusDevice, ModbusDeviceError
 
-DEFAULT_DATASETS = ["Sec_normal", "sec_flash"]
+DEFAULT_DATASETS = ["Sec_normal", "sec_flash", "event_log"]
+
+# SEC_STATE values that allow modbus_to_config()/write_all_flash() to run
+# (garuda-dsp/app/app_modbus/app_modbus.c update_modbus_input()).
+# SEC_RUN (charging) is NOT in this set - config writes are silently ignored
+# while charging is active.
+CONFIG_WRITABLE_STATES = {1, 9, 238, 240, 241}  # READY, PRE_READY, ERROR, DEBUG_READY, DEBUG_FREQ_CAL
 
 
 class Rpu(ModbusDevice):
@@ -25,11 +31,17 @@ class Rpu(ModbusDevice):
     def set_charging(self, enable):
         self.write_coil(self._addr("Enable_charger"), enable)
 
+    def read_q128(self, label):
+        return self.read(label) / self.Q128_DIVISOR
+
+    def write_q128(self, label, value):
+        self.write(label, round(value * self.Q128_DIVISOR))
+
     def set_runtime_voltage(self, volts):
-        self.write_holding_register(self._addr("Runtime_voltage_setting_Q128"), round(volts * self.Q128_DIVISOR))
+        self.write_q128("Runtime_voltage_setting_Q128", volts)
 
     def set_runtime_current(self, amps):
-        self.write_holding_register(self._addr("Runtime_current_setting_Q128"), round(amps * self.Q128_DIVISOR))
+        self.write_q128("Runtime_current_setting_Q128", amps)
 
     def read_error_code(self):
         return (self.read("Error_code_hi") << 16) | self.read("Error_code_lo")
@@ -37,6 +49,14 @@ class Rpu(ModbusDevice):
     def read_shadow_error_code(self):
         return (self.read("Shadow_error_code_hi") << 16) | self.read("Shadow_error_code_lo")
 
+    def wait_until_config_writable(self, timeout_s=10, poll_interval_s=0.5):
+        deadline = time.time() + timeout_s
+        while time.time() < deadline:
+            if self.read("Main_state") in CONFIG_WRITABLE_STATES:
+                return
+            time.sleep(poll_interval_s)
+        raise ModbusDeviceError(f"RPU did not reach a config-writable state within {timeout_s}s")
+
     def rewrite_config(self):
         self.write_holding_register(self._addr("PASSWORD_LO"), self.REWRITE_PASSWORD_LO)
         self.write_holding_register(self._addr("PASSWORD_HI"), self.REWRITE_PASSWORD_HI)

tests/test_alignment.py

@@ -0,0 +1,255 @@
+import csv
+import time
+from dataclasses import dataclass
+from pathlib import Path
+
+import pytest
+
+from common.config_loader import load_jig_registers
+from drivers.eload import Eload
+from drivers.jig import Jig, Position
+from drivers.rpu import Rpu
+from drivers.tpu import Tpu
+
+RESULTS_DIR = Path(__file__).resolve().parent.parent / "results"
+
+DEFAULT_POSITION = Position(0, 0, 0)
+CLOSE_BORDER_POSITION = Position(0, 0, 15)
+FAR_BORDER_POSITION = Position(30, 10, 60)
+
+SWEEP_POSITIONS = [
+    Position(0, 0, 5), Position(0, 0, 10), Position(0, 0, 15), Position(0, 0, 20),
+    Position(30, 10, 15), Position(60, 10, 15), Position(80, 10, 15), Position(90, 10, 15),
+    Position(20, 10, 50), Position(30, 10, 50), Position(0, 0, 55), Position(10, 10, 55),
+    Position(30, 10, 55), Position(30, 10, 60), Position(30, 10, 65), Position(0, 0, 80),
+]
+
+DEFAULT_MOVE_WAIT_S = 5
+POSITION_MOVE_WAIT_S = 5
+START_CHARGING_WAIT_S = 1
+RAMP_UP_WAIT_S = 8
+RAMP_DOWN_WAIT_S = 3
+FREQ_CAL_TIMEOUT_S = 30
+FREQ_CAL_POLL_INTERVAL_S = 1
+
+TARGET_CURRENT_A = 30.0
+ALIGNMENT_RAMP_MA_PER_SEC = 20000
+SWEEP_VELOADS = [50.0, 60.0]
+
+# "BLE Conn mode" (TPU: Debug dataset addr 5, RPU: sec_flash addr 142) selects
+# which frequency-calibration handshake the firmware expects
+# (global_variables.h: typedef enum _BLE_MODE { BLE_DEFAULT, BLE_RECONN, ... }).
+BLE_MODE_DEFAULT = 0
+BLE_MODE_RECONN = 1
+
+# TPU firmware resets PASSWORD_LO/HI back to 0 as soon as it recognizes the
+# combined password (main.c: check_debug_test_mode() does
+# `*main_app.pops->pass_code = 0` right after matching) - verifying by
+# read-back would race against that reset, so these writes use verify=False.
+TPU_DEBUG_PASSWORD_LO = 0xBEEF
+TPU_DEBUG_PASSWORD_HI = 0xFEED
+
+RPU_DEBUG_PASSWORD_LO = 0xDAAD
+RPU_DEBUG_PASSWORD_HI = 0xCBAD
+RPU_NORMAL_PASSWORD_LO = 0xFEED
+RPU_NORMAL_PASSWORD_HI = 0xCEE5
+
+
+@dataclass
+class AlignmentContext:
+    jig: Jig
+    tpu: Tpu
+    rpu: Rpu
+    eload: Eload
+
+
+@pytest.fixture(scope="module")
+def alignment(jig_port, tpu_port, rpu_port, eload_resource):
+    jig_registers = load_jig_registers()
+    jig = Jig(jig_port, jig_registers)
+    tpu = Tpu(tpu_port, extra_datasets=["frequency_calibration"])
+    rpu = Rpu(rpu_port)
+    eload = Eload(eload_resource)
+
+    ctx = AlignmentContext(jig=jig, tpu=tpu, rpu=rpu, eload=eload)
+
+    rpu.set_charging(False)
+    rpu.wait_until_config_writable()
+    # Alignment sweep needs Iout to ramp up fast enough to settle within the
+    # per-position wait window - bump both ramp rates for the duration of
+    # this file, then restore the original values on teardown.
+    default_ramp_control = rpu.read("Ramp_control_mA_per_sec")
+    default_ramp_slow = rpu.read("Ramp_slow_mA_per_sec")
+    rpu.write("Ramp_control_mA_per_sec", ALIGNMENT_RAMP_MA_PER_SEC)
+    rpu.write("Ramp_slow_mA_per_sec", ALIGNMENT_RAMP_MA_PER_SEC)
+    rpu.rewrite_config()
+
+    try:
+        yield ctx
+    finally:
+        rpu.set_charging(False)
+        jig.move_to_position(DEFAULT_POSITION, wait_time=DEFAULT_MOVE_WAIT_S)
+
+        rpu.set_charging(False)
+        rpu.wait_until_config_writable()
+        rpu.write("Ramp_control_mA_per_sec", default_ramp_control)
+        rpu.write("Ramp_slow_mA_per_sec", default_ramp_slow)
+        rpu.rewrite_config()
+
+        jig.close()
+        tpu.close()
+        rpu.close()
+        eload.disconnect()
+
+
+def _wait_for_freq_cal_done(tpu, position_name, timeout_s=FREQ_CAL_TIMEOUT_S):
+    deadline = time.time() + timeout_s
+    while time.time() < deadline:
+        if tpu.read("FREQ_CAL_DONE"):
+            error_code = tpu.read("FREQ_CAL_ERROR_CODE")
+            assert error_code == 0, f"[{position_name}] frequency calibration failed, error code {error_code}"
+            return
+        time.sleep(FREQ_CAL_POLL_INTERVAL_S)
+    pytest.fail(f"[{position_name}] frequency calibration timed out after {timeout_s}s")
+
+
+def _run_freq_cal_step(tpu, cmd, position_name, ble_mode):
+    if ble_mode == BLE_MODE_DEFAULT:
+        # Method 1 (BLE mode 0): TPU and RPU acknowledge each other directly
+        # over the wireless link - no host-driven ACK handshake needed.
+        tpu.write("FREQ_CAL_CMD", cmd)
+    else:
+        # Method 2 (BLE mode 1): host must drive the FREQ_CAL_CMD_ACK
+        # handshake itself (self-clearing coil, confirmed in
+        # freq_calibration.c: "*freq_cal.cal_modbus_ack = false").
+        tpu.write("FREQ_CAL_CMD_ACK", True, verify=False)
+        time.sleep(0.1)
+        tpu.write("FREQ_CAL_CMD", cmd)
+        time.sleep(0.1)
+        tpu.write("FREQ_CAL_CMD_ACK", True, verify=False)
+    time.sleep(0.1)
+    _wait_for_freq_cal_done(tpu, position_name)
+
+
+def test_frequency_calibration(alignment):
+    jig = alignment.jig
+    tpu = alignment.tpu
+    rpu = alignment.rpu
+
+    tpu_ble_mode = tpu.read("BLE Conn mode")
+    rpu_ble_mode = rpu.read("BLE Conn mode")
+    if tpu_ble_mode != rpu_ble_mode:
+        pytest.fail(
+            f"TPU/RPU BLE Conn mode mismatch (TPU={tpu_ble_mode}, RPU={rpu_ble_mode}) - "
+            f"both must be the same mode (0=BLE_DEFAULT or 1=BLE_RECONN) to calibrate"
+        )
+    ble_mode = tpu_ble_mode
+
+    rpu.set_charging(False)
+    jig.move_to_position(DEFAULT_POSITION, wait_time=DEFAULT_MOVE_WAIT_S)
+
+    if ble_mode == BLE_MODE_RECONN:
+        # Enter RPU debug / force-frequency-calibration mode (Method 2 only).
+        rpu.write("PASSWORD_LO", RPU_DEBUG_PASSWORD_LO)
+        rpu.write("PASSWORD_HI", RPU_DEBUG_PASSWORD_HI)
+        time.sleep(0.1)
+        rpu.write("FORCE_FREQ_CAL", True)
+        time.sleep(0.1)
+
+    try:
+        jig.move_to_position(CLOSE_BORDER_POSITION, wait_time=POSITION_MOVE_WAIT_S)
+        _run_freq_cal_step(tpu, 1, "close-border", ble_mode)
+
+        jig.move_to_position(FAR_BORDER_POSITION, wait_time=POSITION_MOVE_WAIT_S)
+        _run_freq_cal_step(tpu, 2, "far-border", ble_mode)
+
+    finally:
+        if ble_mode == BLE_MODE_RECONN:
+            rpu.write("FORCE_FREQ_CAL", False)
+            time.sleep(0.1)
+            rpu.write("PASSWORD_LO", RPU_NORMAL_PASSWORD_LO)
+            rpu.write("PASSWORD_HI", RPU_NORMAL_PASSWORD_HI)
+            time.sleep(0.1)
+
+    freq_low = tpu.read("FREQ_CAL_RESULT_LOW")
+    freq_high = tpu.read("FREQ_CAL_RESULT_HIGH")
+
+    # Commit the calibrated frequencies: TPU needs its own debug-mode
+    # password first (combined 0xFEEDBEEF -> LO=0xBEEF, HI=0xFEED).
+    tpu.write("PASSWORD_LO", TPU_DEBUG_PASSWORD_LO, verify=False)
+    tpu.write("PASSWORD_HI", TPU_DEBUG_PASSWORD_HI, verify=False)
+    time.sleep(0.1)
+    tpu.write("Coupling_freq_1", freq_low)
+    tpu.write("Coupling_freq_2", freq_high)
+    time.sleep(0.1)
+    tpu.rewrite_config()
+
+    jig.move_to_position(DEFAULT_POSITION, wait_time=DEFAULT_MOVE_WAIT_S)
+
+    print(f"Frequency calibration done: f1 (close-border) = {freq_low:.2f} kHz, "
+          f"f2 (far-border) = {freq_high:.2f} kHz")
+
+
+@pytest.mark.parametrize("veload", SWEEP_VELOADS)
+def test_alignment_sweep(alignment, veload):
+    jig = alignment.jig
+    tpu = alignment.tpu
+    rpu = alignment.rpu
+    eload = alignment.eload
+
+    RESULTS_DIR.mkdir(exist_ok=True)
+    csv_path = RESULTS_DIR / f"alignment_sweep_{veload:.0f}V_{time.strftime('%Y%m%d_%H%M%S')}.csv"
+
+    rpu.set_charging(False)
+    jig.move_to_position(DEFAULT_POSITION, wait_time=DEFAULT_MOVE_WAIT_S)
+
+    eload.set_remote()
+    eload.set_mode_cv()
+    eload.set_voltage(veload)
+    eload.output_on()
+
+    try:
+        with open(csv_path, "w", newline="", encoding="utf-8") as f:
+            writer = csv.writer(f)
+            writer.writerow([
+                "position_index", "x", "y", "z", "iref",
+                "pfc_vout", "pfc_il_peak", "inv_iout_peak",
+                "coupling_phase_1", "coupling_phase_2", "rpu_pi_i_out",
+                "coil_vac_out", "coil_vac_out_est",
+                "tpu_uart_lost_rate", "tpu_ble_rssi", "tpu_shadow_error_code",
+                "rpu_output_current", "rpu_dc_batt_voltage",
+                "rpu_ble_uart_lost_rate", "rpu_ble_rssi", "rpu_shadow_error_code",
+            ])
+
+            for index, pos in enumerate(SWEEP_POSITIONS, start=1):
+                jig.move_to_position(pos, wait_time=POSITION_MOVE_WAIT_S)
+
+                rpu.set_charging(True)
+                time.sleep(START_CHARGING_WAIT_S)
+                rpu.set_runtime_current(TARGET_CURRENT_A)
+                time.sleep(RAMP_UP_WAIT_S)
+
+                actual_pos = jig.get_current_position()
+                writer.writerow([
+                    index, actual_pos.x, actual_pos.y, actual_pos.z, TARGET_CURRENT_A,
+                    tpu.read("PFC_Vout"), tpu.read("PFC_iL_peak"), tpu.read("Inv_Iout_peak"),
+                    tpu.read("coupling phase 1") / 10.0, tpu.read("coupling phase 2") / 10.0,
+                    tpu.read("RPU_PI_I_OUT") / 100.0,
+                    tpu.read("Coil_VAC_out"), tpu.read("Coil_VAC_out_est"),
+                    tpu.read("UART_Lost_rate"), tpu.read("BLE_RSSI"), hex(tpu.read_shadow_error_code()),
+                    rpu.read("Output_current_Q128") / rpu.Q128_DIVISOR,
+                    rpu.read("DC_Batt_voltage_Q128") / rpu.Q128_DIVISOR,
+                    rpu.read("BLE_UART_Lost_rate"), rpu.read("BLE_RSSI"), hex(rpu.read_shadow_error_code()),
+                ])
+                f.flush()
+
+                rpu.set_charging(False)
+                time.sleep(RAMP_DOWN_WAIT_S)
+
+    finally:
+        rpu.set_charging(False)
+        eload.output_off()
+        eload.set_local()
+        jig.move_to_position(DEFAULT_POSITION, wait_time=DEFAULT_MOVE_WAIT_S)
+
+    print(f"Saved to {csv_path}")

tests/test_cv.py

@@ -0,0 +1,373 @@
+import argparse
+import csv
+import sys
+import time
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+
+from common.error_codes import RpuErrorCode
+from drivers.eload import Eload
+from drivers.rpu import Rpu
+from drivers.tpu import Tpu
+
+RESULTS_DIR = Path(__file__).resolve().parent.parent / "results"
+
+# Vac (AC source input, 230V) is set manually by the test operator before
+# running this script - not controlled here.
+# Jig position (0,0,40) is skipped for now - no jig available; re-add
+# jig.move_to_position() once one is connected.
+VREF = 55.0
+IREF = 30.0
+
+SWEEP_ELOAD_VOLTAGES = [50.0, 52.0, 54.0, 54.5, 54.7, 54.9, 55.0]
+EXTENDED_VOLTAGE_LIMIT = 55.5
+EXTENDED_VOLTAGE_STEP = 0.1
+LOW_CURRENT_THRESHOLD_A = 1.0
+STEP_WAIT_S = 20
+HIGH_VOLTAGE_THRESHOLD = 52.0
+HIGH_VOLTAGE_STEP_WAIT_S = 30
+TOGGLE_HIGH_VOLTAGE_STEP_WAIT_S = 45
+
+START_CHARGING_WAIT_S = 1
+STOP_CHARGING_SETTLE_WAIT_S = 3
+
+# RPU is powered from the eload (battery terminal) - the eload must be
+# outputting voltage before RPU's Modbus port comes up, or connecting to it
+# fails outright.
+RPU_POWER_UP_ELOAD_VOLTAGE = 50.0
+RPU_POWER_UP_WAIT_S = 1
+
+POWER_LIMITS_W = [1500, 1000, 500]
+POWER_LIMIT_ELOAD_VOLTAGE = 50.0
+POWER_LIMIT_SETTLE_WAIT_S = 20
+POWER_LIMIT_TOLERANCE_PCT = 0.05
+
+PRECHARGE_ELOAD_VOLTAGE = 53.0
+PRECHARGE_CV_CURRENT_LIMITS = [5.0, 10.0]
+PRECHARGE_WAIT_S = 60
+PRECHARGE_CV_CURRENT_TOLERANCE_PCT = 0.05
+
+EXPECTED_RPU_CV_LOW_CURRENT = RpuErrorCode.CV_LOW_CURRENT
+
+CASE_NAMES = [
+    "cv_sweep_continuous_enable",
+    "cv_sweep_toggle_enable",
+    "power_limit_sweep",
+    "precharge_cv_current_limit",
+]
+
+
+class CvTestSystem:
+    def __init__(self, tpu_port, rpu_port, eload_resource):
+        # Eload powers RPU (via the battery terminal) - it must be on before
+        # RPU's Modbus port is reachable, so connect/enable it first.
+        print(f"Powering up RPU via eload {eload_resource} at {RPU_POWER_UP_ELOAD_VOLTAGE}V...")
+        self.eload = Eload(eload_resource)
+        self.eload.set_remote()
+        self.eload.set_mode_cv()
+        self.eload.set_voltage(RPU_POWER_UP_ELOAD_VOLTAGE)
+        self.eload.output_on()
+        time.sleep(RPU_POWER_UP_WAIT_S)
+
+        print(f"Connecting TPU ({tpu_port}) and RPU ({rpu_port})...")
+        self.tpu = Tpu(tpu_port)
+        self.rpu = Rpu(rpu_port)
+
+        RESULTS_DIR.mkdir(exist_ok=True)
+        csv_path = RESULTS_DIR / f"cv_{time.strftime('%Y%m%d_%H%M%S')}.csv"
+        self.csv_file = open(csv_path, "w", newline="", encoding="utf-8")
+        self.writer = csv.writer(self.csv_file)
+        self.writer.writerow(
+            ["phase", "setting_label", "setting_value", "rpu_output_voltage", "rpu_output_current", "note"]
+        )
+        print(f"Logging to {csv_path}")
+
+        self.rpu.set_charging(False)
+
+        print(f"Setting Voltage_setting_Q128={VREF}V, Current_setting_Q128={IREF}A for the whole test group...")
+        self.rpu.wait_until_config_writable()
+        self.default_voltage_setting_raw = self.rpu.read("Voltage_setting_Q128")
+        self.default_current_setting_raw = self.rpu.read("Current_setting_Q128")
+        self.rpu.write_q128("Voltage_setting_Q128", VREF)
+        self.rpu.write_q128("Current_setting_Q128", IREF)
+        self.rpu.rewrite_config()
+
+    def close(self):
+        print("Closing: restoring Voltage_setting_Q128/Current_setting_Q128 defaults...")
+        self.rpu.set_charging(False)
+        time.sleep(STOP_CHARGING_SETTLE_WAIT_S)
+        self.rpu.wait_until_config_writable()
+        self.rpu.write("Voltage_setting_Q128", self.default_voltage_setting_raw)
+        self.rpu.write("Current_setting_Q128", self.default_current_setting_raw)
+        self.rpu.rewrite_config()
+        self.eload.set_voltage(RPU_POWER_UP_ELOAD_VOLTAGE)
+        self.eload.set_local()
+        self.csv_file.close()
+        self.tpu.close()
+        self.rpu.close()
+        self.eload.disconnect()
+        print("Closed.")
+
+    def log_row(self, phase, setting_label, setting_value, note=""):
+        rpu_voltage = self.tpu.read("RPU_output_voltage")
+        rpu_current = self.tpu.read("RPU_output_current")
+        self.writer.writerow([phase, setting_label, setting_value, rpu_voltage, rpu_current, note])
+        self.csv_file.flush()
+        return rpu_voltage, rpu_current
+
+    def stop_charging_and_wait_config_writable(self):
+        print("Stopping charging, waiting for config-writable state...")
+        self.rpu.set_charging(False)
+        time.sleep(STOP_CHARGING_SETTLE_WAIT_S)
+        self.rpu.wait_until_config_writable()
+
+    def _sweep_step(self, phase, voltage, toggle_charger_per_step):
+        """Set eload to `voltage`, wait, record, and check for the CV_LOW_CURRENT
+        cutoff. Returns (low_current_triggered, protection_ok) for this step."""
+        rpu = self.rpu
+        eload = self.eload
+
+        print(f"[{phase}] Setting eload to {voltage}V...")
+        eload.set_voltage(voltage)
+
+        if toggle_charger_per_step:
+            print(f"[{phase}] Enabling charger...")
+            rpu.set_charging(True)
+            time.sleep(START_CHARGING_WAIT_S)
+
+        if voltage > HIGH_VOLTAGE_THRESHOLD:
+            step_wait = TOGGLE_HIGH_VOLTAGE_STEP_WAIT_S if toggle_charger_per_step else HIGH_VOLTAGE_STEP_WAIT_S
+        else:
+            step_wait = STEP_WAIT_S
+        print(f"[{phase}] Waiting {step_wait}s to settle...")
+        time.sleep(step_wait)
+        rpu_current = self.tpu.read("RPU_output_current")
+        print(f"[{phase}] Vout_load={voltage}V -> Iout={rpu_current:.2f}A")
+
+        low_current_triggered = False
+        protection_ok = False
+        note = ""
+
+        if rpu_current < LOW_CURRENT_THRESHOLD_A:
+            low_current_triggered = True
+            print(f"[{phase}] Iout below {LOW_CURRENT_THRESHOLD_A}A, checking protection...")
+
+            charging_enabled = rpu.read("Enable_charger")
+            rpu_error = rpu.read_error_code()
+            rpu_shadow_error = rpu.read_shadow_error_code()
+            protection_ok = (not charging_enabled) and (
+                rpu_error == EXPECTED_RPU_CV_LOW_CURRENT or rpu_shadow_error == EXPECTED_RPU_CV_LOW_CURRENT
+            )
+            note = (f"low-current check: charging_enabled={charging_enabled}, rpu_error=0x{rpu_error:08X}, "
+                    f"rpu_shadow_error=0x{rpu_shadow_error:08X}")
+            print(f"[{phase}] low-current check: charging_enabled={charging_enabled}, "
+                  f"rpu_error=0x{rpu_error:08X}, rpu_shadow_error=0x{rpu_shadow_error:08X}")
+
+        self.log_row(phase, "Vout_load", voltage, note=note)
+
+        if toggle_charger_per_step:
+            print(f"[{phase}] Disabling charger...")
+            rpu.set_charging(False)
+            time.sleep(STOP_CHARGING_SETTLE_WAIT_S)
+
+        return low_current_triggered, protection_ok
+
+    def _run_voltage_sweep(self, phase, toggle_charger_per_step):
+        rpu = self.rpu
+        eload = self.eload
+
+        low_current_seen = False
+        protection_ok = False
+
+        if not toggle_charger_per_step:
+            print(f"[{phase}] Setting eload to {SWEEP_ELOAD_VOLTAGES[0]}V and enabling charger for the whole sweep...")
+            eload.set_voltage(SWEEP_ELOAD_VOLTAGES[0])
+            rpu.set_charging(True)
+            time.sleep(START_CHARGING_WAIT_S)
+
+        for voltage in SWEEP_ELOAD_VOLTAGES:
+            triggered, ok = self._sweep_step(phase, voltage, toggle_charger_per_step)
+            if triggered and not low_current_seen:
+                low_current_seen = True
+                protection_ok = ok
+
+        # If current never dropped below threshold even at 55V, keep nudging
+        # the eload voltage up by 0.1V until it does or 55.5V is reached.
+        voltage = SWEEP_ELOAD_VOLTAGES[-1]
+        while not low_current_seen and voltage < EXTENDED_VOLTAGE_LIMIT - 1e-9:
+            voltage = round(voltage + EXTENDED_VOLTAGE_STEP, 1)
+            triggered, ok = self._sweep_step(phase, voltage, toggle_charger_per_step)
+            if triggered:
+                low_current_seen = True
+                protection_ok = ok
+
+        if not (toggle_charger_per_step and not low_current_seen):
+            rpu.set_charging(False)
+
+        if not low_current_seen:
+            raise AssertionError(
+                f"[{phase}] output current never dropped below {LOW_CURRENT_THRESHOLD_A}A "
+                f"even up to {EXTENDED_VOLTAGE_LIMIT}V"
+            )
+        if not protection_ok:
+            raise AssertionError(f"[{phase}] CV_LOW_CURRENT protection did not trigger correctly")
+
+    def _set_cutoff_minimum_current_for_sweep(self):
+        self.stop_charging_and_wait_config_writable()
+        default_value = self.rpu.read_q128("Cutoff_minimum_current_Q128")
+        print(f"Setting Cutoff_minimum_current_Q128={LOW_CURRENT_THRESHOLD_A}A (default was {default_value}A)...")
+        self.rpu.write_q128("Cutoff_minimum_current_Q128", LOW_CURRENT_THRESHOLD_A)
+        self.rpu.rewrite_config()
+        return default_value
+
+    def _restore_cutoff_minimum_current(self, default_value):
+        self.stop_charging_and_wait_config_writable()
+        print(f"Restoring Cutoff_minimum_current_Q128={default_value}A...")
+        self.rpu.write_q128("Cutoff_minimum_current_Q128", default_value)
+        self.rpu.rewrite_config()
+
+    def run_cv_sweep_continuous_enable(self):
+        default_cutoff_min_current = self._set_cutoff_minimum_current_for_sweep()
+        try:
+            self._run_voltage_sweep("cv_sweep_continuous_enable", toggle_charger_per_step=False)
+        finally:
+            self.rpu.set_charging(False)
+            self._restore_cutoff_minimum_current(default_cutoff_min_current)
+
+    def run_cv_sweep_toggle_enable(self):
+        default_cutoff_min_current = self._set_cutoff_minimum_current_for_sweep()
+        try:
+            self._run_voltage_sweep("cv_sweep_toggle_enable", toggle_charger_per_step=True)
+        finally:
+            self.rpu.set_charging(False)
+            self._restore_cutoff_minimum_current(default_cutoff_min_current)
+
+    def run_power_limit_sweep(self):
+        rpu = self.rpu
+        eload = self.eload
+
+        default_power_limit = rpu.read("Power_limit_ref")
+        failures = []
+
+        try:
+            print(f"Setting eload to {POWER_LIMIT_ELOAD_VOLTAGE}V and enabling charger...")
+            eload.set_voltage(POWER_LIMIT_ELOAD_VOLTAGE)
+            rpu.set_charging(True)
+            time.sleep(START_CHARGING_WAIT_S)
+
+            for power_limit in POWER_LIMITS_W:
+                print(f"Setting Power_limit_ref={power_limit}W, waiting {POWER_LIMIT_SETTLE_WAIT_S}s to settle...")
+                rpu.write("Power_limit_ref", power_limit)
+                time.sleep(POWER_LIMIT_SETTLE_WAIT_S)
+
+                rpu_voltage, rpu_current = self.log_row("power_limit_sweep", "Power_limit_ref", power_limit)
+                pout = rpu_voltage * rpu_current
+                error_pct = abs(pout - power_limit) / power_limit
+                print(f"Power_limit_ref={power_limit}W -> Vout={rpu_voltage:.2f}V, Iout={rpu_current:.2f}A, "
+                      f"Pout={pout:.1f}W ({error_pct * 100:.1f}% error)")
+                if error_pct > POWER_LIMIT_TOLERANCE_PCT:
+                    failures.append(f"Power_limit_ref={power_limit}W -> Pout={pout:.1f}W "
+                                     f"({error_pct * 100:.1f}% error, allowed {POWER_LIMIT_TOLERANCE_PCT * 100:.0f}%)")
+
+        finally:
+            print(f"Disabling charger, restoring Power_limit_ref={default_power_limit}W...")
+            rpu.set_charging(False)
+            rpu.write("Power_limit_ref", default_power_limit)
+
+        if failures:
+            raise AssertionError("power_limit_sweep: " + "; ".join(failures))
+
+    def run_precharge_cv_current_limit(self):
+        rpu = self.rpu
+        eload = self.eload
+
+        self.stop_charging_and_wait_config_writable()
+        default_precharge_enable = rpu.read("Precharge enable")
+        default_cv_current_limit = rpu.read_q128("CV current limit")
+
+        print("Enabling Precharge enable...")
+        rpu.write("Precharge enable", True)
+        rpu.rewrite_config()
+
+        failures = []
+        try:
+            print(f"Setting eload to {PRECHARGE_ELOAD_VOLTAGE}V...")
+            eload.set_voltage(PRECHARGE_ELOAD_VOLTAGE)
+
+            for cv_limit in PRECHARGE_CV_CURRENT_LIMITS:
+                self.stop_charging_and_wait_config_writable()
+                print(f"Setting CV current limit={cv_limit}A...")
+                rpu.write_q128("CV current limit", cv_limit)
+                rpu.rewrite_config()
+
+                print(f"Enabling charger, waiting {START_CHARGING_WAIT_S + PRECHARGE_WAIT_S}s for precharge...")
+                rpu.set_charging(True)
+                time.sleep(START_CHARGING_WAIT_S + PRECHARGE_WAIT_S)
+
+                charging_state = rpu.read("Charging_state")
+                _, rpu_current = self.log_row(
+                    "precharge", "CV_current_limit", cv_limit, note=f"charging_state={charging_state}"
+                )
+                error_pct = abs(rpu_current - cv_limit) / cv_limit
+                print(f"CV_current_limit={cv_limit}A -> Charging_state={charging_state}, Iout={rpu_current:.2f}A "
+                      f"({error_pct * 100:.1f}% error)")
+                if error_pct > PRECHARGE_CV_CURRENT_TOLERANCE_PCT:
+                    failures.append(f"CV_current_limit={cv_limit}A -> Iout={rpu_current:.2f}A "
+                                     f"({error_pct * 100:.1f}% error, allowed {PRECHARGE_CV_CURRENT_TOLERANCE_PCT * 100:.0f}%)")
+
+        finally:
+            self.stop_charging_and_wait_config_writable()
+            print(f"Restoring Precharge enable={default_precharge_enable}, "
+                  f"CV current limit={default_cv_current_limit}A...")
+            rpu.write("Precharge enable", default_precharge_enable)
+            rpu.write_q128("CV current limit", default_cv_current_limit)
+            rpu.rewrite_config()
+
+        if failures:
+            raise AssertionError("precharge_cv_current_limit: " + "; ".join(failures))
+
+    def run_all(self):
+        cases = [
+            ("cv_sweep_continuous_enable", self.run_cv_sweep_continuous_enable),
+            ("cv_sweep_toggle_enable", self.run_cv_sweep_toggle_enable),
+            ("power_limit_sweep", self.run_power_limit_sweep),
+            ("precharge_cv_current_limit", self.run_precharge_cv_current_limit),
+        ]
+        results = {}
+        for name, func in cases:
+            print(f"\n=== Running {name} ===")
+            try:
+                func()
+                results[name] = "PASS"
+            except AssertionError as exc:
+                results[name] = f"FAIL: {exc}"
+            except Exception as exc:
+                results[name] = f"ERROR: {exc}"
+
+        print("\n=== CV test summary ===")
+        for name, result in results.items():
+            print(f"{name}: {result}")
+        return results
+
+
+def main():
+    parser = argparse.ArgumentParser(description="CV group regression tests")
+    parser.add_argument("--tpu-port", default="COM15")
+    parser.add_argument("--rpu-port", default="COM16")
+    parser.add_argument("--eload-resource", default="ASRL5::INSTR")
+    parser.add_argument("--case", choices=CASE_NAMES + ["all"], default="all")
+    args = parser.parse_args()
+
+    system = CvTestSystem(args.tpu_port, args.rpu_port, args.eload_resource)
+    try:
+        if args.case == "all":
+            system.run_all()
+        else:
+            getattr(system, f"run_{args.case}")()
+    finally:
+        system.close()
+
+
+if __name__ == "__main__":
+    main()

tests/test_prot_pfc.py

@@ -0,0 +1,340 @@
+import csv
+import time
+from dataclasses import dataclass
+from pathlib import Path
+
+import pytest
+
+from common.config_loader import load_jig_registers
+from common.error_codes import RpuErrorCode, TpuErrorCode
+from drivers.eload import Eload
+from drivers.jig import Jig, Position
+from drivers.rpu import Rpu
+from drivers.tpu import Tpu
+
+RESULTS_DIR = Path(__file__).resolve().parent.parent / "results"
+
+# Vac_in (AC source input) is set manually by the test operator before running
+# this file - not controlled by the script.
+POSITION = Position(0, 0, 40)
+VELOAD = 48.0
+IOUT_REF = 30.0
+
+DEFAULT_PFC_CURRENT_MAX = 30.0
+DEFAULT_PFC_VDC_BUS_MIN = 350.0
+DEFAULT_PFC_VDC_BUS_MAX = 425.0
+
+FAULT_PFC_CURRENT_MAX = 8.0
+FAULT_PFC_VDC_BUS_MAX = 395.0
+FAULT_PFC_VDC_BUS_MIN = 445.0
+AC_VOLTAGE_CALI_INCREASE_PCT = 0.10
+
+EXPECTED_TPU_OUT_OVERCURRENT = TpuErrorCode.PFC_IL_OVER_CURRENT
+EXPECTED_RPU_OUT_OVERCURRENT = RpuErrorCode.TPU_PFC_FAILURE
+
+EXPECTED_TPU_DC_OVERVOLTAGE = TpuErrorCode.PFC_DC_OVER_VOLTAGE
+EXPECTED_RPU_DC_OVERVOLTAGE = RpuErrorCode.BLE_UART_LOSS
+
+EXPECTED_TPU_DC_UNDERVOLTAGE = TpuErrorCode.PFC_DC_UNDER_VOLTAGE
+EXPECTED_RPU_DC_UNDERVOLTAGE = 0
+
+EXPECTED_TPU_START_TIMEOUT = TpuErrorCode.PFC_START_TIMEOUT
+EXPECTED_RPU_START_TIMEOUT = 0
+
+START_CHARGING_WAIT_S = 1
+CHARGING_RAMP_WAIT_S = 20  # wait for Iout to ramp up to the 30A target before recording
+STOP_CHARGING_SETTLE_WAIT_S = 3
+CONFIG_SETTLE_WAIT_S = 5  # settle wait for cases that don't need charging enabled
+
+STABILITY_SAMPLE_DURATION_S = 5.0
+STABILITY_SAMPLE_INTERVAL_S = 0.2
+STABILITY_FIELDS = ["PFC_Vout", "PFC_iL_peak"]
+
+
+def _hex32(value):
+    return f"0x{value:08X}"
+
+
+def _hex16(value):
+    return f"0x{value:04X}"
+
+
+def _error_ok(actual, actual_shadow, expected):
+    return actual == expected or actual_shadow == expected
+
+
+@dataclass
+class PfcBaseline:
+    """Shared connections + helpers for every PROT-PFC test case in this file."""
+    jig: Jig
+    tpu: Tpu
+    rpu: Rpu
+    eload: Eload
+    writer: object
+    csv_file: object
+
+    def sample_min_max(self, labels, duration=STABILITY_SAMPLE_DURATION_S, interval=STABILITY_SAMPLE_INTERVAL_S):
+        """Poll `labels` every `interval` seconds for `duration` seconds and
+        return {label_min, label_max} - same pattern as vac_coil_estimation.py's
+        sample_stability()."""
+        samples = {label: [] for label in labels}
+        start = time.time()
+        while time.time() - start < duration:
+            for label in labels:
+                samples[label].append(self.tpu.read(label))
+            time.sleep(interval)
+
+        stats = {}
+        for label, vals in samples.items():
+            stats[f"{label}_min"] = min(vals)
+            stats[f"{label}_max"] = max(vals)
+        return stats
+
+    def log_row(self, phase, setting_label, setting_value, stability_fields=STABILITY_FIELDS, pass_fail=""):
+        # The 5s sampling window IS the "record" step for this row. Only the
+        # fields relevant to this test case are sampled - others stay blank.
+        stats = self.sample_min_max(stability_fields)
+        self.writer.writerow([
+            phase, setting_label, setting_value,
+            stats.get("PFC_Vout_min", ""), stats.get("PFC_Vout_max", ""),
+            stats.get("PFC_iL_peak_min", ""), stats.get("PFC_iL_peak_max", ""),
+            self.tpu.read("RPU_output_voltage"), self.tpu.read("RPU_output_current"),
+            self.tpu.read("PFC_state"), self.tpu.read("Main_state"), self.tpu.read("Main_pre_state"),
+            _hex32(self.tpu.read_error_code()), _hex32(self.tpu.read_shadow_error_code()),
+            _hex16(self.tpu.read("Issue code")),
+            _hex32(self.rpu.read_error_code()), _hex32(self.rpu.read_shadow_error_code()),
+            pass_fail,
+        ])
+        self.csv_file.flush()
+
+    def start_charging(self):
+        self.rpu.set_charging(True)
+        time.sleep(START_CHARGING_WAIT_S)
+        self.rpu.set_runtime_current(IOUT_REF)
+        time.sleep(CHARGING_RAMP_WAIT_S)
+
+    def stop_charging_and_wait_config_writable(self):
+        # PFC_Current_max (and other "Debug" dataset config) is only committed
+        # by the firmware while TPU is NOT actively charging (TPU_RUN_INV is
+        # excluded from the config-writable states).
+        self.rpu.set_charging(False)
+        time.sleep(STOP_CHARGING_SETTLE_WAIT_S)
+        self.tpu.wait_until_config_writable()
+
+    def restore_config(self, label, value):
+        self.stop_charging_and_wait_config_writable()
+        self.tpu.write(label, value)
+        self.tpu.rewrite_config()
+
+
+@pytest.fixture(scope="module")
+def pfc_baseline(jig_port, tpu_port, rpu_port, eload_resource):
+    jig_registers = load_jig_registers()
+    jig = Jig(jig_port, jig_registers)
+    tpu = Tpu(tpu_port)
+    rpu = Rpu(rpu_port)
+    eload = Eload(eload_resource)
+
+    RESULTS_DIR.mkdir(exist_ok=True)
+    csv_path = RESULTS_DIR / f"prot_pfc_{time.strftime('%Y%m%d_%H%M%S')}.csv"
+    csv_file = open(csv_path, "w", newline="", encoding="utf-8")
+    writer = csv.writer(csv_file)
+    writer.writerow([
+        "phase", "setting_label", "setting_value", "pfc_vout_min", "pfc_vout_max",
+        "pfc_il_peak_min", "pfc_il_peak_max", "rpu_output_voltage", "rpu_output_current",
+        "pfc_state", "main_state", "main_pre_state", "tpu_error_code", "tpu_shadow_error_code",
+        "tpu_issue_code", "rpu_error_code", "rpu_shadow_error_code", "pass_fail",
+    ])
+
+    baseline = PfcBaseline(jig=jig, tpu=tpu, rpu=rpu, eload=eload, writer=writer, csv_file=csv_file)
+
+    try:
+        rpu.set_charging(False)
+        jig.move_to_position(POSITION)
+
+        eload.set_remote()
+        eload.set_mode_cv()
+        eload.set_voltage(VELOAD)
+        eload.output_on()
+
+        pfc_current_max = tpu.read("PFC_Current_max")
+        pfc_vdc_bus_min = tpu.read("PFC_Vdc_bus_min")
+        pfc_vdc_bus_max = tpu.read("PFC_Vdc_bus_max")
+        assert pfc_current_max == pytest.approx(DEFAULT_PFC_CURRENT_MAX, abs=0.5), \
+            f"PFC_Current_max not at default: {pfc_current_max}"
+        assert pfc_vdc_bus_min == pytest.approx(DEFAULT_PFC_VDC_BUS_MIN, abs=1), \
+            f"PFC_Vdc_bus_min not at default: {pfc_vdc_bus_min}"
+        assert pfc_vdc_bus_max == pytest.approx(DEFAULT_PFC_VDC_BUS_MAX, abs=1), \
+            f"PFC_Vdc_bus_max not at default: {pfc_vdc_bus_max}"
+
+        # Enable charging, wait for Iout to ramp up to 30A, then record for 5s
+        # (both PFC_Vout and PFC_iL_peak are relevant as a healthy reference).
+        baseline.start_charging()
+        baseline.log_row("baseline", "PFC_Imax", pfc_current_max)
+
+        yield baseline
+
+    finally:
+        rpu.set_charging(False)
+        eload.output_off()
+        eload.set_local()
+        csv_file.close()
+        jig.close()
+        tpu.close()
+        rpu.close()
+        eload.disconnect()
+
+
+def test_pfc_out_overcurrent(pfc_baseline):
+    tpu = pfc_baseline.tpu
+    rpu = pfc_baseline.rpu
+
+    try:
+        pfc_baseline.stop_charging_and_wait_config_writable()
+        tpu.write("PFC_Current_max", FAULT_PFC_CURRENT_MAX)
+        tpu.rewrite_config()
+
+        # This case needs charging active to push real output current above
+        # the lowered threshold - wait for ramp-up same as baseline.
+        pfc_baseline.start_charging()
+
+        tpu_error = tpu.read_error_code()
+        tpu_shadow_error = tpu.read_shadow_error_code()
+        rpu_error = rpu.read_error_code()
+        rpu_shadow_error = rpu.read_shadow_error_code()
+
+        tpu_error_ok = _error_ok(tpu_error, tpu_shadow_error, EXPECTED_TPU_OUT_OVERCURRENT)
+        rpu_error_ok = _error_ok(rpu_error, rpu_shadow_error, EXPECTED_RPU_OUT_OVERCURRENT)
+        pass_fail = "PASS" if (tpu_error_ok and rpu_error_ok) else "FAIL"
+
+        # Current-protection case - only PFC_iL_peak is relevant, PFC_Vout stays blank.
+        pfc_baseline.log_row("pfc_out_overcurrent", "PFC_Imax", FAULT_PFC_CURRENT_MAX,
+                              stability_fields=["PFC_iL_peak"], pass_fail=pass_fail)
+
+        assert tpu_error_ok, (
+            f"TPU error code mismatch: got {_hex32(tpu_error)}/{_hex32(tpu_shadow_error)}, "
+            f"expected {_hex32(EXPECTED_TPU_OUT_OVERCURRENT)}"
+        )
+        assert rpu_error_ok, (
+            f"RPU error code mismatch: got {_hex32(rpu_error)}/{_hex32(rpu_shadow_error)}, "
+            f"expected {_hex32(EXPECTED_RPU_OUT_OVERCURRENT)}"
+        )
+
+    finally:
+        pfc_baseline.restore_config("PFC_Current_max", DEFAULT_PFC_CURRENT_MAX)
+
+
+def test_pfc_dc_overvoltage(pfc_baseline):
+    tpu = pfc_baseline.tpu
+    rpu = pfc_baseline.rpu
+
+    try:
+        pfc_baseline.stop_charging_and_wait_config_writable()
+        tpu.write("PFC_Vdc_bus_max", FAULT_PFC_VDC_BUS_MAX)
+        tpu.rewrite_config()
+
+        # No charging needed - the DC bus is already regulated by PFC before
+        # RPU enables charging, just settle then sample.
+        time.sleep(CONFIG_SETTLE_WAIT_S)
+
+        tpu_error = tpu.read_error_code()
+        tpu_shadow_error = tpu.read_shadow_error_code()
+        rpu_error = rpu.read_error_code()
+        rpu_shadow_error = rpu.read_shadow_error_code()
+
+        tpu_error_ok = _error_ok(tpu_error, tpu_shadow_error, EXPECTED_TPU_DC_OVERVOLTAGE)
+        rpu_error_ok = _error_ok(rpu_error, rpu_shadow_error, EXPECTED_RPU_DC_OVERVOLTAGE)
+        pass_fail = "PASS" if (tpu_error_ok and rpu_error_ok) else "FAIL"
+
+        pfc_baseline.log_row("pfc_dc_overvoltage", "Vdc_bus_max", FAULT_PFC_VDC_BUS_MAX,
+                              stability_fields=["PFC_Vout"], pass_fail=pass_fail)
+
+        assert tpu_error_ok, (
+            f"TPU error code mismatch: got {_hex32(tpu_error)}/{_hex32(tpu_shadow_error)}, "
+            f"expected {_hex32(EXPECTED_TPU_DC_OVERVOLTAGE)}"
+        )
+        assert rpu_error_ok, (
+            f"RPU error code mismatch: got {_hex32(rpu_error)}/{_hex32(rpu_shadow_error)}, "
+            f"expected {_hex32(EXPECTED_RPU_DC_OVERVOLTAGE)}"
+        )
+
+    finally:
+        pfc_baseline.restore_config("PFC_Vdc_bus_max", DEFAULT_PFC_VDC_BUS_MAX)
+
+
+def test_pfc_dc_undervoltage(pfc_baseline):
+    tpu = pfc_baseline.tpu
+    rpu = pfc_baseline.rpu
+
+    try:
+        pfc_baseline.stop_charging_and_wait_config_writable()
+        tpu.write("PFC_Vdc_bus_min", FAULT_PFC_VDC_BUS_MIN)
+        tpu.rewrite_config()
+
+        time.sleep(CONFIG_SETTLE_WAIT_S)
+
+        tpu_error = tpu.read_error_code()
+        tpu_shadow_error = tpu.read_shadow_error_code()
+        rpu_error = rpu.read_error_code()
+        rpu_shadow_error = rpu.read_shadow_error_code()
+
+        tpu_error_ok = _error_ok(tpu_error, tpu_shadow_error, EXPECTED_TPU_DC_UNDERVOLTAGE)
+        rpu_error_ok = _error_ok(rpu_error, rpu_shadow_error, EXPECTED_RPU_DC_UNDERVOLTAGE)
+        pass_fail = "PASS" if (tpu_error_ok and rpu_error_ok) else "FAIL"
+
+        pfc_baseline.log_row("pfc_dc_undervoltage", "Vdc_bus_min", FAULT_PFC_VDC_BUS_MIN,
+                              stability_fields=["PFC_Vout"], pass_fail=pass_fail)
+
+        assert tpu_error_ok, (
+            f"TPU error code mismatch: got {_hex32(tpu_error)}/{_hex32(tpu_shadow_error)}, "
+            f"expected {_hex32(EXPECTED_TPU_DC_UNDERVOLTAGE)}"
+        )
+        assert rpu_error_ok, (
+            f"RPU error code mismatch: got {_hex32(rpu_error)}/{_hex32(rpu_shadow_error)}, "
+            f"expected {_hex32(EXPECTED_RPU_DC_UNDERVOLTAGE)}"
+        )
+
+    finally:
+        pfc_baseline.restore_config("PFC_Vdc_bus_min", DEFAULT_PFC_VDC_BUS_MIN)
+
+
+def test_pfc_start_timeout(pfc_baseline):
+    tpu = pfc_baseline.tpu
+    rpu = pfc_baseline.rpu
+
+    # PFC_AC_Voltage_Cali_A default varies per unit - read the current value
+    # first (before the try block) so `finally` can always restore it exactly,
+    # even if something below fails before the fault value is written.
+    pfc_baseline.stop_charging_and_wait_config_writable()
+    default_cali_a = tpu.read("PFC_AC_Voltage_Cali_A")
+
+    try:
+        fault_cali_a = default_cali_a * (1 + AC_VOLTAGE_CALI_INCREASE_PCT)
+        tpu.write("PFC_AC_Voltage_Cali_A", fault_cali_a)
+        tpu.rewrite_config()
+
+        time.sleep(CONFIG_SETTLE_WAIT_S)
+
+        tpu_error = tpu.read_error_code()
+        tpu_shadow_error = tpu.read_shadow_error_code()
+        rpu_error = rpu.read_error_code()
+        rpu_shadow_error = rpu.read_shadow_error_code()
+
+        tpu_error_ok = _error_ok(tpu_error, tpu_shadow_error, EXPECTED_TPU_START_TIMEOUT)
+        rpu_error_ok = _error_ok(rpu_error, rpu_shadow_error, EXPECTED_RPU_START_TIMEOUT)
+        pass_fail = "PASS" if (tpu_error_ok and rpu_error_ok) else "FAIL"
+
+        pfc_baseline.log_row("pfc_start_timeout", "PFC_AC_Voltage_Cali_A", fault_cali_a,
+                              stability_fields=["PFC_Vout"], pass_fail=pass_fail)
+
+        assert tpu_error_ok, (
+            f"TPU error code mismatch: got {_hex32(tpu_error)}/{_hex32(tpu_shadow_error)}, "
+            f"expected {_hex32(EXPECTED_TPU_START_TIMEOUT)}"
+        )
+        assert rpu_error_ok, (
+            f"RPU error code mismatch: got {_hex32(rpu_error)}/{_hex32(rpu_shadow_error)}, "
+            f"expected {_hex32(EXPECTED_RPU_START_TIMEOUT)}"
+        )
+
+    finally:
+        pfc_baseline.restore_config("PFC_AC_Voltage_Cali_A", default_cali_a)

tests/test_thermal_derating.py

@@ -0,0 +1,614 @@
+import argparse
+import csv
+import sys
+import time
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+
+from common.config_loader import load_jig_registers
+from drivers.eload import Eload, EloadError
+from drivers.jig import Jig, Position
+from drivers.rpu import Rpu
+from drivers.tpu import Tpu
+
+RESULTS_DIR = Path(__file__).resolve().parent.parent / "results"
+
+VREF = 55.0
+IREF = 30.0
+VELOAD = 50.0
+# AC source (Vac_in) is set manually by the test operator - not controlled here.
+
+POSITION_MOVE_WAIT_S = 5
+RPU_POWER_UP_WAIT_S = 1
+STOP_CHARGING_SETTLE_WAIT_S = 3
+SAMPLE_INTERVAL_S = 2
+SUBTEST_TIMEOUT_S = 15 * 60
+
+# Only shown as the suggested value in the derating_1/derating_2 prompts in
+# MANUAL mode - the tester types the actual value to use for each sub-test.
+# AUTO mode always uses these two values outright. Both are a percentage of
+# rated current, not amps.
+DERATING_1_SUGGESTED = 6.0
+DERATING_2_SUGGESTED = 10.0
+
+# Only the mosfet sub-tests (INV/PFC on TPU, SR on RPU) need to wait for
+# mosfets to cool down before starting - MCU/COIL don't stress the mosfets,
+# so they skip this check entirely (see run_subtest). Polls (not logged to
+# CSV) until below this, or the operator hits Ctrl+C to cancel. RPU's is
+# higher than TPU's - RPU's SR mosfets idle closer to 44C, so 44 was
+# blocking normal starts.
+MOSFET_SUBTEST_NAMES = {"INV", "PFC", "SR"}
+SAFE_TPU_MOSFET_TEMP_C = 35.0
+SAFE_RPU_MOSFET_TEMP_C = 44.0
+SAFE_TEMP_POLL_INTERVAL_S = 2
+
+# AUTO mode's fixed thresholds for the "mosfet" sub-tests (TPU INV, RPU SR;
+# TPU PFC has its own, lower triple). COIL/MCU sub-tests instead ramp 1C
+# above the baseline reading, 1C apart per stage (see _auto_subtest_values)
+# - so they have no fixed triple here.
+AUTO_INV_THRESHOLDS = (44.0, 46.0, 48.0)
+AUTO_SR_THRESHOLDS = (48.0, 50.0, 52.0)
+AUTO_PFC_THRESHOLDS = (40.0, 44.0, 46.0)
+
+# INV/SR-only: the stage_1/stage_2 pair each device's abnormal cases (below)
+# reuse. Each device has its own dedicated pair - not shared with any other
+# device/sub-test - so changing one later can never silently change another.
+AUTO_INV_ABNORMAL_THRESHOLDS = (44.0, 46.0)
+AUTO_SR_ABNORMAL_THRESHOLDS = (50.0, 52.0)
+
+# INV/SR-only: AUTO mode runs these 5 threshold-stage combinations back to
+# back instead of a single fixed triple. Case 1 is the normal ascending
+# order (each device's own AUTO threshold above); cases 2-5 are deliberately
+# abnormal (non-increasing, or leaving one or two stages untouched), built
+# from that device's own *_ABNORMAL_THRESHOLDS pair above. None means "leave
+# that stage's threshold as-is, don't write it".
+MOSFET_TEST_CASES = {
+    "INV": [
+        AUTO_INV_THRESHOLDS,
+        (20.0, AUTO_INV_ABNORMAL_THRESHOLDS[0], AUTO_INV_ABNORMAL_THRESHOLDS[1]),
+        (20.0, 20.0, AUTO_INV_ABNORMAL_THRESHOLDS[0]),
+        (None, None, AUTO_INV_ABNORMAL_THRESHOLDS[0]),
+        (None, AUTO_INV_ABNORMAL_THRESHOLDS[0], AUTO_INV_ABNORMAL_THRESHOLDS[1]),
+    ],
+    "SR": [
+        AUTO_SR_THRESHOLDS,
+        (20.0, AUTO_SR_ABNORMAL_THRESHOLDS[0], AUTO_SR_ABNORMAL_THRESHOLDS[1]),
+        (20.0, 20.0, AUTO_SR_ABNORMAL_THRESHOLDS[0]),
+        (None, None, AUTO_SR_ABNORMAL_THRESHOLDS[0]),
+        (None, AUTO_SR_ABNORMAL_THRESHOLDS[0], AUTO_SR_ABNORMAL_THRESHOLDS[1]),
+    ],
+}
+
+CSV_HEADER = [
+    "elapsed_s", "device", "subtest", "phase",
+    "derating_state", "derating_current", "rpu_output_current", "object_temp_c",
+    "tpu_error_code", "tpu_shadow_error_code", "tpu_issue_code", "tpu_event_0",
+    "rpu_error_code", "rpu_shadow_error_code", "rpu_issue_code", "rpu_event_0",
+]
+
+# TPU thresholds/derating live in the Debug dataset, RPU's in sec_flash -
+# all plain int/uint, no Q128 scaling despite the RPU labels ending in
+# _Q128 (confirmed against real defaults: raw 6/18, not 0.046875/0.140625).
+# derating_1/derating_2 are a percentage of rated current, not amps.
+DEVICE_CONFIG = {
+    "TPU": {
+        "derating_state_label": "DERATING_STATE",
+        "derating_current_label": "DERATING_CURRENT",
+        "derating_limit_labels": ("Current_derating_1", "Current_derating_2"),
+        "subtests": {
+            "INV": {
+                "thresholds": ("INV_temp_threshold_stage_1", "INV_temp_threshold_stage_2",
+                                "INV_temp_threshold_stage_3"),
+                "temp_labels": ("MOS_INV_TPBP_TEMP", "MOS_INV_TNBN_TEMP"),
+                "position": Position(0, 0, 20),
+                "auto_thresholds": AUTO_INV_THRESHOLDS,
+            },
+            "PFC": {
+                "thresholds": ("PFC_temp_threshold_stage_1", "PFC_temp_threshold_stage_2",
+                                "PFC_temp_threshold_stage_3"),
+                "temp_labels": ("AC_RECT_temp", "MOS_PFC_temp"),
+                "position": Position(0, 0, 20),
+                "auto_thresholds": AUTO_PFC_THRESHOLDS,
+            },
+            "MCU": {
+                "thresholds": ("MCU_temp_threshold_stage_1", "MCU_temp_threshold_stage_2",
+                                "MCU_temp_threshold_stage_3"),
+                "temp_labels": ("MCU_Core_temp",),
+                "position": Position(0, 0, 20),
+                "auto_thresholds": None,
+            },
+            "COIL": {
+                "thresholds": ("Coil_temp_threshold_stage_1", "Coil_temp_threshold_stage_2",
+                                "Coil_temp_threshold_stage_3"),
+                "temp_labels": ("NTC_COIL_temp",),
+                "position": Position(0, 0, 40),
+                "auto_thresholds": None,
+            },
+        },
+    },
+    "RPU": {
+        "derating_state_label": "Derating State",
+        "derating_current_label": "Derating Current",
+        "derating_limit_labels": ("Current_derating_1_Q128", "Current_derating_2_Q128"),
+        "subtests": {
+            "SR": {
+                "thresholds": ("Dev_temp_threshold_stage_1", "Dev_temp_threshold_stage_2",
+                                "Dev_temp_threshold_stage_3"),
+                "temp_labels": ("Ana_MOS1_tp_temp", "Ana_MOS2_bp_temp", "Ana_MOS3_tn_temp", "Ana_MOS4_bn_temp"),
+                "position": Position(0, 0, 20),
+                "auto_thresholds": AUTO_SR_THRESHOLDS,
+            },
+            "MCU": {
+                "thresholds": ("MCU_temp_threshold_stage_1", "MCU_temp_threshold_stage_2",
+                                "MCU_temp_threshold_stage_3"),
+                "temp_labels": ("Ana_MCU_temp",),
+                "position": Position(0, 0, 20),
+                "auto_thresholds": None,
+            },
+            "COIL": {
+                "thresholds": ("Coil_temp_threshold_stage_1", "Coil_temp_threshold_stage_2",
+                                "Coil_temp_threshold_stage_3"),
+                "temp_labels": ("Ana_Coil_temp",),
+                "position": Position(0, 0, 40),
+                "auto_thresholds": None,
+            },
+        },
+    },
+}
+
+
+def _hex32(value):
+    return f"0x{value:08X}"
+
+
+def _hex16(value):
+    return f"0x{value:04X}"
+
+
+def _prompt_float(prompt, default=None):
+    while True:
+        raw = input(prompt).strip()
+        if not raw and default is not None:
+            return default
+        try:
+            return float(raw)
+        except ValueError:
+            print("Invalid number, try again.")
+
+
+def _prompt_float_or_keep(prompt):
+    """Returns None (keep the register's current value, don't write it) if
+    the tester presses Enter with no input."""
+    while True:
+        raw = input(prompt).strip()
+        if not raw:
+            return None
+        try:
+            return float(raw)
+        except ValueError:
+            print("Invalid number, try again.")
+
+
+class ThermalDeratingTestSystem:
+    def __init__(self, jig_port, tpu_port, rpu_port, eload_resource):
+        if jig_port:
+            self.jig = Jig(jig_port, load_jig_registers())
+        else:
+            self.jig = None
+            print("No --jig-port given - jig positioning will be manual for every sub-test.")
+
+        print(f"Powering up RPU via eload {eload_resource} at {VELOAD}V...")
+        self.eload = Eload(eload_resource)
+        self.eload.set_remote()
+        self.eload.set_mode_cv()
+        self.eload.set_voltage(VELOAD)
+        self.eload.output_on()
+        time.sleep(RPU_POWER_UP_WAIT_S)
+
+        if not self.eload.is_output_on():
+            raise EloadError(f"eload {eload_resource} output did not turn on")
+        measured_voltage = self.eload.measure_voltage()
+        print(f"Eload output confirmed ON, measured {measured_voltage:.2f}V (target {VELOAD}V).")
+
+        print(f"Connecting TPU ({tpu_port}) and RPU ({rpu_port})...")
+        self.tpu = Tpu(tpu_port)
+        self.rpu = Rpu(rpu_port)
+
+        tpu_state = self.tpu.read("Main_state")
+        print(f"TPU connection confirmed, Main_state={tpu_state}.")
+        rpu_state = self.rpu.read("Main_state")
+        print(f"RPU connection confirmed, Main_state={rpu_state}.")
+
+        self.rpu.set_charging(False)
+
+        print(f"Setting Voltage_setting_Q128={VREF}V, Current_setting_Q128={IREF}A for the whole session...")
+        self.rpu.wait_until_config_writable()
+        self.default_voltage_setting_raw = self.rpu.read("Voltage_setting_Q128")
+        self.default_current_setting_raw = self.rpu.read("Current_setting_Q128")
+        self.rpu.write_q128("Voltage_setting_Q128", VREF)
+        self.rpu.write_q128("Current_setting_Q128", IREF)
+        self.rpu.rewrite_config()
+
+        self.mode = self._prompt_mode()
+
+        self.print_tpu_temperatures()
+        self.print_rpu_temperatures()
+
+    def close(self):
+        print("Closing: restoring Voltage_setting_Q128/Current_setting_Q128 defaults...")
+        self.rpu.set_charging(False)
+        time.sleep(STOP_CHARGING_SETTLE_WAIT_S)
+        self.rpu.wait_until_config_writable()
+        self.rpu.write("Voltage_setting_Q128", self.default_voltage_setting_raw)
+        self.rpu.write("Current_setting_Q128", self.default_current_setting_raw)
+        self.rpu.rewrite_config()
+
+        print(f"Switching eload to local control at {VELOAD}V (output stays on to keep RPU powered)...")
+        self.eload.set_voltage(VELOAD)
+        self.eload.set_local()
+
+        if self.jig is not None:
+            self.jig.close()
+
+        self.tpu.close()
+        self.rpu.close()
+        self.eload.disconnect()
+        print("Closed.")
+
+    def print_tpu_temperatures(self):
+        inv_temp = max(self.tpu.read("MOS_INV_TPBP_TEMP"), self.tpu.read("MOS_INV_TNBN_TEMP"))
+        pfc_temp = max(self.tpu.read("AC_RECT_temp"), self.tpu.read("MOS_PFC_temp"))
+        coil_temp = self.tpu.read("NTC_COIL_temp")
+        mcu_temp = self.tpu.read("MCU_Core_temp")
+        print(f"TPU temperatures -> INV mosfet(max): {inv_temp}C, PFC mosfet(max): {pfc_temp}C, "
+              f"COIL: {coil_temp}C, MCU: {mcu_temp}C")
+
+    def print_rpu_temperatures(self):
+        sr_temp = max(self.rpu.read("Ana_MOS1_tp_temp"), self.rpu.read("Ana_MOS2_bp_temp"),
+                      self.rpu.read("Ana_MOS3_tn_temp"), self.rpu.read("Ana_MOS4_bn_temp"))
+        coil_temp = self.rpu.read("Ana_Coil_temp")
+        mcu_temp = self.rpu.read("Ana_MCU_temp")
+        print(f"RPU temperatures -> SR mosfet(max): {sr_temp}C, COIL: {coil_temp}C, MCU: {mcu_temp}C")
+
+    def _read_tpu_mosfet_temp(self):
+        return max(self.tpu.read("MOS_INV_TPBP_TEMP"), self.tpu.read("MOS_INV_TNBN_TEMP"),
+                   self.tpu.read("AC_RECT_temp"), self.tpu.read("MOS_PFC_temp"))
+
+    def _read_rpu_mosfet_temp(self):
+        return max(self.rpu.read("Ana_MOS1_tp_temp"), self.rpu.read("Ana_MOS2_bp_temp"),
+                   self.rpu.read("Ana_MOS3_tn_temp"), self.rpu.read("Ana_MOS4_bn_temp"))
+
+    def _wait_for_safe_mosfet_temp(self, device_name):
+        """Polls (not logged to CSV) until the sub-test's own device mosfet
+        temp is below that device's safe threshold - a TPU sub-test only
+        cares about TPU's mosfets (and TPU's threshold), an RPU one only
+        about RPU's. Returns False instead if the operator hits Ctrl+C, so
+        the caller can cancel this sub-test and go back to test selection."""
+        if device_name == "TPU":
+            read_temp, safe_temp = self._read_tpu_mosfet_temp, SAFE_TPU_MOSFET_TEMP_C
+        else:
+            read_temp, safe_temp = self._read_rpu_mosfet_temp, SAFE_RPU_MOSFET_TEMP_C
+        try:
+            while True:
+                temp = read_temp()
+                if temp < safe_temp:
+                    return True
+                print(f"{device_name} mosfet={temp}C - waiting to drop below "
+                      f"{safe_temp}C before starting (not logged; Ctrl+C to cancel "
+                      f"this sub-test)...")
+                time.sleep(SAFE_TEMP_POLL_INTERVAL_S)
+        except KeyboardInterrupt:
+            print("\nCtrl+C - cancelling this sub-test.")
+            return False
+
+    def _prompt_mode(self):
+        while True:
+            choice = input("Test mode [MANUAL/AUTO] (Ctrl+C to exit): ").strip().upper()
+            if choice in ("MANUAL", "AUTO"):
+                return choice
+            print("Invalid input - type MANUAL or AUTO.")
+
+    def _prompt_auto_scope(self):
+        while True:
+            choice = input("AUTO test which device? [TPU/RPU/ALL]: ").strip().upper()
+            if choice in ("TPU", "RPU", "ALL"):
+                break
+            print("Invalid input - type TPU, RPU, or ALL.")
+
+        devices = ["TPU", "RPU"] if choice == "ALL" else [choice]
+        plan = [(d, s) for d in devices for s in DEVICE_CONFIG[d]["subtests"]]
+
+        if choice == "ALL":
+            keys = [f"{d}-{s}" for d, s in plan]
+            print(f"Sub-tests available: {', '.join(keys)}")
+            raw = input("Exclude any sub-tests? Comma-separated DEVICE-SUBTEST "
+                        "(e.g. TPU-COIL,RPU-MCU), blank for none: ").strip()
+            exclude = {item.strip().upper() for item in raw.split(",")} if raw else set()
+            plan = [(d, s) for d, s in plan if f"{d}-{s}" not in exclude]
+        else:
+            keys = [s for _, s in plan]
+            print(f"{choice} sub-tests available: {', '.join(keys)}")
+            raw = input("Exclude any sub-tests? Comma-separated (e.g. COIL,MCU), blank for none: ").strip()
+            exclude = {item.strip().upper() for item in raw.split(",")} if raw else set()
+            plan = [(d, s) for d, s in plan if s not in exclude]
+
+        # INV/SR expand into 5 case runs each; every other sub-test is a
+        # single run (case_num=None).
+        expanded_plan = []
+        for d, s in plan:
+            if s in MOSFET_TEST_CASES:
+                for case_num in range(1, len(MOSFET_TEST_CASES[s]) + 1):
+                    expanded_plan.append((d, s, case_num))
+            else:
+                expanded_plan.append((d, s, None))
+
+        print(f"AUTO plan: {', '.join(f'{d}-{s}' + (f' case{c}' if c else '') for d, s, c in expanded_plan)}")
+        return expanded_plan
+
+    def _auto_subtest_values(self, device, subtest, subtest_name, case_num):
+        if subtest_name in MOSFET_TEST_CASES:
+            thresholds = MOSFET_TEST_CASES[subtest_name][case_num - 1]
+        elif subtest["auto_thresholds"] is not None:
+            thresholds = subtest["auto_thresholds"]
+        else:
+            baseline_temp = self._read_object_temp(device, subtest["temp_labels"])
+            thresholds = (baseline_temp + 1, baseline_temp + 2, baseline_temp + 3)
+        return thresholds, DERATING_1_SUGGESTED, DERATING_2_SUGGESTED
+
+    def _prompt_device(self):
+        print("Which device do you want to test?")
+        while True:
+            choice = input("Device [TPU/RPU] (Ctrl+C to exit): ").strip().upper()
+            if choice in DEVICE_CONFIG:
+                return choice
+            print("Invalid input - type TPU or RPU.")
+
+    def _prompt_subtest(self, device_name):
+        keywords = list(DEVICE_CONFIG[device_name]["subtests"].keys())
+        print(f"{device_name} sub-tests available: {', '.join(keywords)}")
+        while True:
+            choice = input(f"Sub-test [{'/'.join(keywords)}] (Ctrl+C to exit): ").strip().upper()
+            if choice in DEVICE_CONFIG[device_name]["subtests"]:
+                return choice
+            print(f"Invalid input - choose one of: {', '.join(keywords)}.")
+
+    def _prompt_subtest_values(self, device_name, subtest_name):
+        labels = DEVICE_CONFIG[device_name]["subtests"][subtest_name]["thresholds"]
+        print(f"Enter values for {device_name}-{subtest_name} "
+              f"(press Enter on a threshold to keep its current default value):")
+        t1 = _prompt_float_or_keep(f"  {labels[0]} (C) [Enter to keep default]: ")
+        t2 = _prompt_float_or_keep(f"  {labels[1]} (C) [Enter to keep default]: ")
+        t3 = _prompt_float_or_keep(f"  {labels[2]} (C) [Enter to keep default]: ")
+        d1 = _prompt_float(f"  derating_1 (%) [Enter for suggested {DERATING_1_SUGGESTED}]: ",
+                            default=DERATING_1_SUGGESTED)
+        d2 = _prompt_float(f"  derating_2 (%) [Enter for suggested {DERATING_2_SUGGESTED}]: ",
+                            default=DERATING_2_SUGGESTED)
+        return (t1, t2, t3), d1, d2
+
+    def _write_subtest_config(self, device_name, device, subtest, thresholds, derating1, derating2):
+        threshold_labels = subtest["thresholds"]
+        derating_labels = DEVICE_CONFIG[device_name]["derating_limit_labels"]
+
+        print(f"Stopping charger, waiting for {device_name} config-writable state...")
+        self.rpu.set_charging(False)
+        time.sleep(STOP_CHARGING_SETTLE_WAIT_S)
+        device.wait_until_config_writable()
+
+        defaults = {}
+        for label, value in zip(threshold_labels, thresholds):
+            defaults[label] = device.read(label)
+            if value is None:
+                print(f"Keeping {label} at its current value ({defaults[label]}C) - not written.")
+            else:
+                print(f"Setting {label}={value}C (default was {defaults[label]}C)...")
+                device.write(label, value)
+
+        for label, value in zip(derating_labels, (derating1, derating2)):
+            defaults[label] = device.read(label)
+            print(f"Setting {label}={value}% (default was {defaults[label]}%)...")
+            device.write(label, value)
+
+        print(f"Rewriting {device_name} config to flash, waiting {device.REWRITE_WAIT_S}s...")
+        device.rewrite_config()
+        return defaults
+
+    def _restore_subtest_config(self, device_name, device, subtest, defaults):
+        threshold_labels = subtest["thresholds"]
+        derating_labels = DEVICE_CONFIG[device_name]["derating_limit_labels"]
+
+        print(f"Restoring {device_name} thresholds/derating defaults...")
+        self.rpu.set_charging(False)
+        time.sleep(STOP_CHARGING_SETTLE_WAIT_S)
+        device.wait_until_config_writable()
+
+        # defaults[label] is always the raw register value read before the
+        # write (see _write_subtest_config) - write it back as-is, no Q128
+        # conversion needed even for the RPU's _Q128-named registers.
+        for label in threshold_labels + derating_labels:
+            print(f"Restoring {label}={defaults[label]}...")
+            device.write(label, defaults[label])
+
+        print(f"Rewriting {device_name} config to flash, waiting {device.REWRITE_WAIT_S}s...")
+        device.rewrite_config()
+
+    def _read_object_temp(self, device, temp_labels):
+        return max(device.read(label) for label in temp_labels)
+
+    def _read_derating(self, device_name, device):
+        cfg = DEVICE_CONFIG[device_name]
+        state = device.read(cfg["derating_state_label"])
+        current = device.read(cfg["derating_current_label"])
+        return state, current
+
+    def _read_errors(self):
+        tpu_error = self.tpu.read_error_code()
+        tpu_shadow_error = self.tpu.read_shadow_error_code()
+        tpu_issue = self.tpu.read("Issue code")
+        tpu_event0 = self.tpu.read("event[0]")
+
+        rpu_error = self.rpu.read_error_code()
+        rpu_shadow_error = self.rpu.read_shadow_error_code()
+        rpu_issue = self.rpu.read("Issue code")
+        rpu_event0 = self.rpu.read("event[0]")
+
+        any_error = bool(tpu_error or tpu_shadow_error or rpu_error or rpu_shadow_error)
+
+        return {
+            "tpu_error": tpu_error, "tpu_shadow_error": tpu_shadow_error,
+            "tpu_issue": tpu_issue, "tpu_event0": tpu_event0,
+            "rpu_error": rpu_error, "rpu_shadow_error": rpu_shadow_error,
+            "rpu_issue": rpu_issue, "rpu_event0": rpu_event0,
+            "any_error": any_error,
+        }
+
+    def _log_row(self, writer, csv_file, device_name, device, subtest_name, temp_labels, phase, elapsed):
+        derating_state, derating_current = self._read_derating(device_name, device)
+        rpu_output_current = self.tpu.read("RPU_output_current")
+        object_temp = self._read_object_temp(device, temp_labels)
+        errors = self._read_errors()
+
+        writer.writerow([
+            f"{elapsed:.1f}", device_name, subtest_name, phase,
+            derating_state, derating_current, rpu_output_current, object_temp,
+            _hex32(errors["tpu_error"]), _hex32(errors["tpu_shadow_error"]),
+            _hex16(errors["tpu_issue"]), _hex16(errors["tpu_event0"]),
+            _hex32(errors["rpu_error"]), _hex32(errors["rpu_shadow_error"]),
+            _hex16(errors["rpu_issue"]), _hex16(errors["rpu_event0"]),
+        ])
+        csv_file.flush()
+
+        print(f"[{device_name}/{subtest_name}/{phase}] t={elapsed:.1f}s "
+              f"derating_state={derating_state} derating_current={derating_current} "
+              f"rpu_output_current={rpu_output_current:.2f}A object_temp={object_temp}C "
+              f"tpu_error={_hex32(errors['tpu_error'])} rpu_error={_hex32(errors['rpu_error'])}")
+
+        return errors
+
+    def run_subtest(self, device_name, subtest_name, case_num=None):
+        subtest = DEVICE_CONFIG[device_name]["subtests"][subtest_name]
+        device = self.tpu if device_name == "TPU" else self.rpu
+        position = subtest["position"]
+
+        if self.jig is not None:
+            print(f"Moving jig to {position}...")
+            self.jig.move_to_position(position, wait_time=POSITION_MOVE_WAIT_S)
+        else:
+            input(f"No --jig-port given - manually position RXP relative to TXP at {position}, "
+                  f"then press Enter to continue...")
+
+        self.print_tpu_temperatures()
+        self.print_rpu_temperatures()
+
+        if self.mode == "AUTO" and subtest_name in MOSFET_SUBTEST_NAMES:
+            if not self._wait_for_safe_mosfet_temp(device_name):
+                print("Sub-test cancelled - back to test selection.")
+                raise KeyboardInterrupt
+
+        if self.mode == "AUTO":
+            thresholds, derating1, derating2 = self._auto_subtest_values(device, subtest, subtest_name, case_num)
+            case_note = f" (case {case_num})" if case_num else ""
+            print(f"AUTO mode{case_note}: thresholds={thresholds}, derating_1={derating1}%, derating_2={derating2}%")
+        else:
+            thresholds, derating1, derating2 = self._prompt_subtest_values(device_name, subtest_name)
+
+        defaults = self._write_subtest_config(device_name, device, subtest, thresholds, derating1, derating2)
+        interrupted = False
+
+        try:
+            RESULTS_DIR.mkdir(exist_ok=True)
+            case_suffix = f"_case{case_num}" if case_num else ""
+            csv_path = (RESULTS_DIR /
+                        f"thermal_derating_{device_name}_{subtest_name}{case_suffix}_"
+                        f"{time.strftime('%Y%m%d_%H%M%S')}.csv")
+            with open(csv_path, "w", newline="", encoding="utf-8") as csv_file:
+                writer = csv.writer(csv_file)
+                writer.writerow(CSV_HEADER)
+                print(f"Logging to {csv_path}")
+
+                print("Recording baseline snapshot (charger still disabled)...")
+                self._log_row(writer, csv_file, device_name, device, subtest_name,
+                              subtest["temp_labels"], "baseline", 0.0)
+
+                print("Enabling charger...")
+                self.rpu.set_charging(True)
+
+                start = time.time()
+                try:
+                    while True:
+                        time.sleep(SAMPLE_INTERVAL_S)
+                        elapsed = time.time() - start
+                        errors = self._log_row(writer, csv_file, device_name, device, subtest_name,
+                                                subtest["temp_labels"], "charging", elapsed)
+
+                        enable_charger = self.rpu.read("Enable_charger")
+                        if not enable_charger:
+                            if errors["any_error"]:
+                                print(f"Charger stopped and a fault was recorded - {device_name}/{subtest_name} "
+                                      f"thermal derating trip confirmed.")
+                            else:
+                                print("WARNING: Enable_charger read 0 but no error code was recorded - "
+                                      "unexpected stop, check the unit.")
+                            break
+
+                        if elapsed > SUBTEST_TIMEOUT_S:
+                            print(f"FAIL: {device_name}/{subtest_name} ran over {SUBTEST_TIMEOUT_S / 60:.0f} "
+                                  f"minutes without tripping - stopping.")
+                            break
+                except KeyboardInterrupt:
+                    print("\nCtrl+C - stopping charger...")
+                    interrupted = True
+
+        finally:
+            print("Disabling charger...")
+            self.rpu.set_charging(False)
+            time.sleep(STOP_CHARGING_SETTLE_WAIT_S)
+            self._restore_subtest_config(device_name, device, subtest, defaults)
+
+        if interrupted and self.mode == "AUTO":
+            raise KeyboardInterrupt
+
+    def run(self):
+        if self.mode == "AUTO":
+            plan = self._prompt_auto_scope()
+            try:
+                for device_name, subtest_name, case_num in plan:
+                    case_note = f" case{case_num}" if case_num else ""
+                    print(f"\n=== AUTO: {device_name}-{subtest_name}{case_note} ===")
+                    self.run_subtest(device_name, subtest_name, case_num=case_num)
+            except KeyboardInterrupt:
+                print("\nCtrl+C - stopping the AUTO run, skipping remaining sub-tests.")
+                return
+            print("\nAUTO run finished for all selected sub-tests.")
+            return
+
+        while True:
+            try:
+                device_name = self._prompt_device()
+                subtest_name = self._prompt_subtest(device_name)
+            except KeyboardInterrupt:
+                print("\nExiting.")
+                return
+
+            self.run_subtest(device_name, subtest_name)
+            print("\nSub-test finished. Choose another test, or press Ctrl+C to exit.")
+
+
+def main():
+    parser = argparse.ArgumentParser(description="TPU/RPU thermal derating regression test")
+    parser.add_argument("--jig-port", default=None,
+                         help="Jig COM port. Omit to skip jig control and position RXP/TXP manually instead.")
+    parser.add_argument("--tpu-port", default="COM15")
+    parser.add_argument("--rpu-port", default="COM16")
+    parser.add_argument("--eload-resource", default="ASRL5::INSTR")
+    args = parser.parse_args()
+
+    system = ThermalDeratingTestSystem(args.jig_port, args.tpu_port, args.rpu_port, args.eload_resource)
+    try:
+        system.run()
+    finally:
+        system.close()
+
+
+if __name__ == "__main__":
+    main()