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/rpu.py

@@ -3,7 +3,7 @@ import time
 from common.config_loader import load_garuda_registers
 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()).

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()