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