diff --git a/tests/test_cv.py b/tests/test_cv.py new file mode 100644 index 0000000..1ed6083 --- /dev/null +++ b/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()