Compare commits
10
Commits
1f11c0b832
...
364a7215b5
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
364a7215b5 | ||
|
|
9db1f93927 | ||
|
|
e12ba2dda1 | ||
|
|
44393ccc44 | ||
|
|
833ae17c6e | ||
|
|
87959b20e6 | ||
|
|
6666e852b2 | ||
|
|
9d7c350c21 | ||
|
|
162b3a612b | ||
|
|
1b6db67805 |
Binary file not shown.
@@ -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}")
|
||||
|
||||
|
||||
+23
-3
@@ -1,10 +1,16 @@
|
||||
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"]
|
||||
|
||||
# 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):
|
||||
"""Covers every register in Sec_normal + sec_flash by label via read()/write()
|
||||
@@ -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)
|
||||
|
||||
@@ -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}")
|
||||
@@ -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()
|
||||
@@ -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)
|
||||
Reference in New Issue
Block a user