From c0ca523143f485da47dc1e227059fdf6f84b5ccf Mon Sep 17 00:00:00 2001 From: vanminhpe Date: Wed, 23 Sep 2026 10:07:00 +0800 Subject: [PATCH] init OVP test for BE --- .gitignore | 7 + BE-RCU-ovp_test_csv.py | 926 +++++++++++++++++++++++++++++++++++++++++ relay_test.py | 110 +++++ 3 files changed, 1043 insertions(+) create mode 100644 .gitignore create mode 100644 BE-RCU-ovp_test_csv.py create mode 100644 relay_test.py diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..6ebd9fc --- /dev/null +++ b/.gitignore @@ -0,0 +1,7 @@ +# Python bytecode +__pycache__/ +*.py[cod] + +# Test output +*.csv +*.log diff --git a/BE-RCU-ovp_test_csv.py b/BE-RCU-ovp_test_csv.py new file mode 100644 index 0000000..df34c43 --- /dev/null +++ b/BE-RCU-ovp_test_csv.py @@ -0,0 +1,926 @@ +#!/usr/bin/env python +""" +BE OVP (over-voltage protection) cycling test. + +Coordinates four Modbus RTU devices, each on its own COM port: + + RCU - Beyond receiver unit (a.k.a. RPU). Charging command + telemetry. + TPU - Beyond transmitter unit. Telemetry (mirrors RCU values over RF). + RELAY - Phoenix/Fenghuang-family board carrying the PFC relay used to + create the over-voltage condition (coil 12 = PFC_relay_enable). + JIG - XYZ positioner. + +The RCU is powered THROUGH the relay: with the relay open it is dead and +answers nothing. So the relay is closed and given RELAY_SETTLE_S to boot the +RCU before any RCU access anywhere in this script - the link test included - +and while the relay is open the RCU telemetry columns are left blank. + +Per cycle: + 1. Relay control = 1, then wait RELAY_SETTLE_S for the RCU to power up. + 2. Jig -> NEAR_POSITION, settle. + 3. RCU charging cmd = 1, then wait for RCU main_state == CHARGING(5) and + output current above the threshold held continuously for the dwell. + Abort the cycle on timeout. + 4. Relay control = 0 (this also cuts RCU power). + 5. Jig -> FAR_POSITION, keep monitoring for FAR_OBSERVE_S. + 6. RCU charging cmd = 0 (best-effort; the RCU is unpowered by now), + settle, next cycle. + +A background monitor thread samples RCU and TPU input registers every 5 s for +the whole run (all cycles, including the aborted ones) and appends each sample +to the CSV. + +Usage: + # check the wiring: read registers from all four ports, then exit + python BE-RCU-ovp_test_csv.py -r COM4 -t COM10 -l COM13 -j COM9 --link-test + + # run the test (the link test also runs as a pre-flight check) + python BE-RCU-ovp_test_csv.py -r COM4 -t COM10 -l COM13 -j COM9 -n 1 +""" + +import argparse +import csv +import datetime +import logging +import os +import sys +import threading +import time + +import minimalmodbus + + +# --------------------------------------------------------------------------- +# Serial / Modbus link +# --------------------------------------------------------------------------- +SLAVE_ID = 0x10 # all four devices share slave id 16 +BAUDRATE = 9600 +SERIAL_TIMEOUT = 2.0 # seconds + + +# --------------------------------------------------------------------------- +# Register map +# +# Sources (in order of authority): +# [1] be-rssi-auto-en/charging_test.py - RCU/TPU "beyond" family classes +# [2] be-rssi-auto-en/base-definition.md - BE system register spec +# [3] be-rssi-auto-en/be_corner_test.py - working BE jig/charging test +# [4] pegasus/config-fenghuang/config.json - phoenix-family board (relay) +# [5] df-code/main.cpp - firmware-side main_state enum +# +# All telemetry below is read with function code 4 (input registers); commands +# are coils (function code 5) and jig positions are holding registers (fc 6). +# --------------------------------------------------------------------------- + +# --- RCU / RPU, "beyond" family [1][2] ------------------------------------- +RCU_CHARGING_CMD_ADDRESS = 0 # coil - 1 = start charging, 0 = stop +RCU_RANGE_CHECK_CMD_ADDRESS = 2 # coil - not used by this test +RCU_SHADOW_ERROR_LO_ADDRESS = 0 # input +RCU_SHADOW_ERROR_HI_ADDRESS = 1 # input +RCU_DC_BUS_VOLTAGE_ADDRESS = 30 # input, Q128 (128 counts = 1 V) +RCU_OUTPUT_CURRENT_ADDRESS = 31 # input, Q128 (128 counts = 1 A) +RCU_MAIN_PRE_STATE_ADDRESS = 50 # input +RCU_MAIN_STATE_ADDRESS = 51 # input +RCU_ERROR_CODE_LO_ADDRESS = 52 # input +RCU_ERROR_CODE_HI_ADDRESS = 53 # input +RCU_OUTPUT_VOLTAGE_ADDRESS = 106 # input, Q128 (phoenix family uses 32) +Q128_DIVISOR = 128.0 + +# --- TPU, "beyond" family [1][2] ------------------------------------------- +TPU_PRI_ENABLE_RF_ADDRESS = 30 # coil - not used by this test +TPU_DC_BUS_VOLTAGE_ADDRESS = 38 # input, float32 (2 regs, big-endian) +TPU_OUTPUT_CURRENT_ADDRESS = 40 # input, float32 +TPU_MAIN_PRE_STATE_ADDRESS = 49 # input +TPU_MAIN_STATE_ADDRESS = 50 # input +TPU_ERROR_CODE_LO_ADDRESS = 61 # input +TPU_ERROR_CODE_HI_ADDRESS = 62 # input +TPU_SHADOW_ERROR_LO_ADDRESS = 63 # input +TPU_SHADOW_ERROR_HI_ADDRESS = 64 # input +TPU_OUTPUT_VOLTAGE_ADDRESS = 74 # input, float32 +TPU_RCU_ERROR_CODE_HI_ADDRESS = 134 # input (note: hi is the LOWER address) +TPU_RCU_ERROR_CODE_LO_ADDRESS = 135 # input + +# --- RELAY board, phoenix/fenghuang family [4] ----------------------------- +RELAY_CONTROL_CMD_ADDRESS = 12 # coil - PFC_relay_enable, 1 = on +RELAY_MAIN_PRE_STATE_ADDRESS = 60 # input +RELAY_MAIN_STATE_ADDRESS = 61 # input + +# --- JIG [1][2][3] --------------------------------------------------------- +JIG_CMD_ADDRESS = 30 # coil - 1 = move to position +JIG_X_ADDRESS = 90 # holding - mm, signed +JIG_Y_ADDRESS = 91 # holding - mm, signed +JIG_Z_ADDRESS = 92 # holding - mm, signed + +# --- RCU main_state enum [5] ----------------------------------------------- +RCU_STATE_IDLE = 1 +RCU_STATE_HANDSHAKE = 2 +RCU_STATE_PAIRED = 3 +RCU_STATE_PRE_CHARGING = 4 +RCU_STATE_CHARGING = 5 +RCU_STATE_STOP = 6 +RCU_STATE_ERROR = 238 +RCU_STATE_CONFIGURATION = 240 + + +# --------------------------------------------------------------------------- +# Test profile +# --------------------------------------------------------------------------- +NEAR_POSITION = (-30, 0, 20) # (x, y, z) mm - charging position +FAR_POSITION = (30, 0, 20) # (x, y, z) mm - move-away position + +JIG_SETTLE_S = 5.0 # after the near move, before the relay +CHARGING_TIMEOUT_S = 60.0 # max wait for the charging criteria +CHARGING_DWELL_S = 30.0 # current must stay above threshold this long +CHARGING_CURRENT_THRESHOLD_A = 3.0 +FAR_OBSERVE_S = 15.0 # keep monitoring after the move-away +CYCLE_GAP_S = 5.0 # after charging cmd = 0 +MONITOR_INTERVAL_S = 2.0 # telemetry sampling / CSV row period +RELAY_SETTLE_S = 10.0 # time to wait for relay to settle + +RESULTS_DIR = "results" + +logger = logging.getLogger("ovp") + + +# --------------------------------------------------------------------------- +# Modbus plumbing +# --------------------------------------------------------------------------- +class Device: + """One Modbus RTU device on its own serial port. + + Every access takes the per-device lock: the monitor thread reads telemetry + while the main thread writes commands, and minimalmodbus instruments are + not safe for concurrent use. + """ + + def __init__(self, port, name, slave_id=SLAVE_ID): + self.name = name + self.port = port + self._lock = threading.Lock() + self.inst = minimalmodbus.Instrument( + port=port, slaveaddress=slave_id, mode='rtu') + self.inst.serial.baudrate = BAUDRATE + self.inst.serial.timeout = SERIAL_TIMEOUT + self.inst.close_port_after_each_call = False + logger.info("%s connected on %s (slave id %d)", name, port, slave_id) + + # Registers read by the link test, as (label, address, kind). + PROBE_REGISTERS = () + + # --- raw accessors ----------------------------------------------------- + def read_u16(self, address): + with self._lock: + return self.inst.read_register(address, functioncode=4) + + def read_holding_u16(self, address): + with self._lock: + return self.inst.read_register(address, functioncode=3) + + def read_f32(self, address): + """Read a big-endian IEEE-754 float from two input registers.""" + with self._lock: + return self.inst.read_float( + address, functioncode=4, number_of_registers=2) + + def write_coil(self, address, value): + with self._lock: + self.inst.write_bit(address, 1 if value else 0, functioncode=5) + + def write_holding(self, address, value): + # Jig coordinates may be negative; Modbus registers are unsigned. + with self._lock: + self.inst.write_register(address, value & 0xFFFF, functioncode=6) + + def probe(self): + """Read PROBE_REGISTERS one at a time to prove the link is alive. + + Returns a list of (label, address, kind, value, error) where exactly + one of value/error is None. + """ + reads = [] + for label, address, kind in self.PROBE_REGISTERS: + try: + if kind == 'input': + value = self.read_u16(address) + elif kind == 'holding': + value = self.read_holding_u16(address) + elif kind == 'holding_signed': + raw = self.read_holding_u16(address) + value = raw - 0x10000 if raw > 0x7FFF else raw + elif kind == 'float': + value = round(self.read_f32(address), 3) + else: + raise ValueError("unknown probe kind: %s" % kind) + reads.append((label, address, kind, value, None)) + except Exception as exc: + reads.append((label, address, kind, None, str(exc))) + return reads + + def close(self): + try: + self.inst.serial.close() + logger.info("%s disconnected", self.name) + except Exception as exc: + logger.warning("%s: close failed: %s", self.name, exc) + + +def _safe(fn, *args): + """Run a read and return None instead of raising, so one bad register + never kills the monitor thread mid-run.""" + try: + return fn(*args) + except Exception as exc: + logger.debug("read failed: %s", exc) + return None + + +def _combine_error(lo, hi): + """Join the lo/hi register halves into one 32-bit code as 8 hex digits. + + Format matches be-rssi-auto-en/charging_test.py and the Beyond error code + map, so a value can be looked up there directly. An unreadable register + gives "" rather than 00000000, which would read as "no error". + """ + if lo is None or hi is None: + return "" + return "%08X" % ((hi << 16) | lo) + + +def _q128(raw): + return None if raw is None else round(raw / Q128_DIVISOR, 3) + + +def _round(value, digits=3): + return None if value is None else round(value, digits) + + +# --------------------------------------------------------------------------- +# Devices +# --------------------------------------------------------------------------- +# What a sample looks like when the RCU is unpowered (relay open). +BLANK_RCU_DATA = { + 'main_pre_state': None, + 'main_state': None, + 'output_current': None, + 'output_voltage': None, + 'dc_bus_voltage': None, + 'error_code': '', + 'shadow_error': '', +} + + +class RCU(Device): + PROBE_REGISTERS = ( + ('main_state', RCU_MAIN_STATE_ADDRESS, 'input'), + ('main_pre_state', RCU_MAIN_PRE_STATE_ADDRESS, 'input'), + ('output_current_q128', RCU_OUTPUT_CURRENT_ADDRESS, 'input'), + ('output_voltage_q128', RCU_OUTPUT_VOLTAGE_ADDRESS, 'input'), + ) + + def __init__(self, port): + super().__init__(port, "RCU") + + def set_charging(self, enable): + logger.info("RCU charging cmd = %d", 1 if enable else 0) + self.write_coil(RCU_CHARGING_CMD_ADDRESS, enable) + + def read_data(self): + return { + 'main_pre_state': _safe(self.read_u16, RCU_MAIN_PRE_STATE_ADDRESS), + 'main_state': _safe(self.read_u16, RCU_MAIN_STATE_ADDRESS), + 'output_current': _q128( + _safe(self.read_u16, RCU_OUTPUT_CURRENT_ADDRESS)), + 'output_voltage': _q128( + _safe(self.read_u16, RCU_OUTPUT_VOLTAGE_ADDRESS)), + 'dc_bus_voltage': _q128( + _safe(self.read_u16, RCU_DC_BUS_VOLTAGE_ADDRESS)), + 'error_code': _combine_error( + _safe(self.read_u16, RCU_ERROR_CODE_LO_ADDRESS), + _safe(self.read_u16, RCU_ERROR_CODE_HI_ADDRESS)), + 'shadow_error': _combine_error( + _safe(self.read_u16, RCU_SHADOW_ERROR_LO_ADDRESS), + _safe(self.read_u16, RCU_SHADOW_ERROR_HI_ADDRESS)), + } + + +class TPU(Device): + PROBE_REGISTERS = ( + ('main_state', TPU_MAIN_STATE_ADDRESS, 'input'), + ('main_pre_state', TPU_MAIN_PRE_STATE_ADDRESS, 'input'), + ('rcu_output_current', TPU_OUTPUT_CURRENT_ADDRESS, 'float'), + ('rcu_output_voltage', TPU_OUTPUT_VOLTAGE_ADDRESS, 'float'), + ) + + def __init__(self, port): + super().__init__(port, "TPU") + + def read_data(self): + return { + 'main_pre_state': _safe(self.read_u16, TPU_MAIN_PRE_STATE_ADDRESS), + 'main_state': _safe(self.read_u16, TPU_MAIN_STATE_ADDRESS), + 'output_current': _round( + _safe(self.read_f32, TPU_OUTPUT_CURRENT_ADDRESS)), + 'output_voltage': _round( + _safe(self.read_f32, TPU_OUTPUT_VOLTAGE_ADDRESS)), + 'dc_bus_voltage': _round( + _safe(self.read_f32, TPU_DC_BUS_VOLTAGE_ADDRESS)), + 'error_code': _combine_error( + _safe(self.read_u16, TPU_ERROR_CODE_LO_ADDRESS), + _safe(self.read_u16, TPU_ERROR_CODE_HI_ADDRESS)), + 'shadow_error': _combine_error( + _safe(self.read_u16, TPU_SHADOW_ERROR_LO_ADDRESS), + _safe(self.read_u16, TPU_SHADOW_ERROR_HI_ADDRESS)), + 'rcu_error_code': _combine_error( + _safe(self.read_u16, TPU_RCU_ERROR_CODE_LO_ADDRESS), + _safe(self.read_u16, TPU_RCU_ERROR_CODE_HI_ADDRESS)), + } + + +class Relay(Device): + PROBE_REGISTERS = ( + ('main_state', RELAY_MAIN_STATE_ADDRESS, 'input'), + ('main_pre_state', RELAY_MAIN_PRE_STATE_ADDRESS, 'input'), + ) + + def __init__(self, port): + super().__init__(port, "RELAY") + + def set_control(self, enable): + logger.info("RELAY control cmd = %d", 1 if enable else 0) + self.write_coil(RELAY_CONTROL_CMD_ADDRESS, enable) + + def read_data(self): + return {'main_state': _safe(self.read_u16, RELAY_MAIN_STATE_ADDRESS)} + + +class Jig(Device): + # The jig exposes no input registers, so the link test reads its position + # holding registers back instead - harmless, and it proves the link. + PROBE_REGISTERS = ( + ('x_position_mm', JIG_X_ADDRESS, 'holding_signed'), + ('y_position_mm', JIG_Y_ADDRESS, 'holding_signed'), + ('z_position_mm', JIG_Z_ADDRESS, 'holding_signed'), + ) + + def __init__(self, port): + super().__init__(port, "JIG") + + def move_to(self, x, y, z): + logger.info("JIG move to (%d, %d, %d) mm", x, y, z) + self.write_holding(JIG_X_ADDRESS, x) + self.write_holding(JIG_Y_ADDRESS, y) + self.write_holding(JIG_Z_ADDRESS, z) + self.write_coil(JIG_CMD_ADDRESS, True) + + +# --------------------------------------------------------------------------- +# RCU power +# --------------------------------------------------------------------------- +def power_up_rcu(relay, settle_s=None): + """Close the relay and wait for the RCU to boot. + + The RCU is fed through the relay, so it is dead - and will not answer a + single Modbus request - until the relay is on. Every RCU access, reads + included, has to come after this. + """ + # Read the constant here, not as a default, so it stays overridable. + settle_s = RELAY_SETTLE_S if settle_s is None else settle_s + relay.set_control(True) + logger.info("Relay closed - waiting %.0fs for the RCU to power up...", + settle_s) + time.sleep(settle_s) + + +# --------------------------------------------------------------------------- +# Link test +# --------------------------------------------------------------------------- +def link_test(devices): + """Read a handful of registers from every device and report per port. + + Returns (all_ok, any_dead): all_ok is False if any single register could + not be read, any_dead is True if some device answered nothing at all - + which usually means a wrong or swapped COM port. + """ + logger.info("-" * 62) + logger.info("LINK TEST") + + all_ok = True + any_dead = False + + for dev in devices: + reads = dev.probe() + good = sum(1 for r in reads if r[4] is None) + if good == len(reads): + status = "OK" + elif good: + status = "PARTIAL" + all_ok = False + else: + status = "NO REPLY" + all_ok = False + any_dead = True + + logger.info(" %-6s %-10s %-9s %d/%d registers", + dev.name, dev.port, status, good, len(reads)) + for label, address, kind, value, err in reads: + if err is None: + logger.info(" %-21s %-14s @%-4d = %s", + label, kind, address, value) + else: + logger.warning(" %-21s %-14s @%-4d ! %s", + label, kind, address, err) + + logger.info("-" * 62) + return all_ok, any_dead + + +def run_link_test(rcu_port, tpu_port, relay_port, jig_port): + """Stand-alone link test: open every port, probe it, close it. + + Returns True only if all four devices answered every probe. + """ + devices = [] + connect_ok = True + + wanted = [(RCU, rcu_port), (TPU, tpu_port), (Relay, relay_port)] + if jig_port: + wanted.append((Jig, jig_port)) + else: + logger.info("No jig port given - skipping the jig.") + + for cls, port in wanted: + try: + devices.append(cls(port)) + except Exception as exc: + logger.error("%s on %s: connect failed: %s", + cls.__name__.upper(), port, exc) + connect_ok = False + + relay = next((d for d in devices if isinstance(d, Relay)), None) + + try: + # The RCU has no power until the relay is on, so it could not answer + # a probe before this. + if relay is not None: + power_up_rcu(relay) + else: + logger.warning("No relay - the RCU is unpowered, so its probe " + "will fail regardless of wiring.") + + all_ok, _any_dead = link_test(devices) if devices else (False, True) + finally: + # Leave the rig de-energised again. + if relay is not None: + try: + relay.set_control(False) + except Exception as exc: + logger.warning("Could not switch relay off: %s", exc) + for dev in devices: + dev.close() + + result = connect_ok and all_ok + logger.info("Link test: %s", "PASS" if result else "FAIL") + return result + + +# --------------------------------------------------------------------------- +# CSV +# --------------------------------------------------------------------------- +class CSVRecorder: + HEADERS = [ + 'Timestamp', 'Elapsed_s', 'Cycle', 'Phase', + 'Jig_X_mm', 'Jig_Y_mm', 'Jig_Z_mm', + 'Relay_Cmd', 'RCU_Charging_Cmd', 'RCU_Powered', + 'RCU_Main_Pre_State', 'RCU_Main_State', + 'RCU_Output_Current_A', 'RCU_Output_Voltage_V', 'RCU_DC_Bus_Voltage_V', + 'RCU_Error_Code', 'RCU_Shadow_Error', + 'TPU_Main_Pre_State', 'TPU_Main_State', + 'TPU_Output_Current_A', 'TPU_Output_Voltage_V', 'TPU_DC_Bus_Voltage_V', + 'TPU_Error_Code', 'TPU_Shadow_Error', 'TPU_RCU_Error_Code', + 'Relay_Main_State', 'Charging_Dwell_s', 'Note', + ] + + def __init__(self, path): + self.path = path + self._lock = threading.Lock() + self._file = open(path, 'w', newline='', encoding='utf-8') + self._writer = csv.writer(self._file) + self._writer.writerow(self.HEADERS) + self._file.flush() + + def write(self, row): + with self._lock: + self._writer.writerow(row) + self._file.flush() + + def close(self): + with self._lock: + self._file.close() + + +# --------------------------------------------------------------------------- +# Monitor thread +# --------------------------------------------------------------------------- +class Monitor(threading.Thread): + """Samples RCU/TPU/relay telemetry on a fixed interval for the whole run. + + It also tracks how long the charging criteria have held continuously, so + the main thread never has to touch the serial ports while it waits. + """ + + def __init__(self, rcu, tpu, relay, recorder, t0, + interval=MONITOR_INTERVAL_S, + current_threshold=CHARGING_CURRENT_THRESHOLD_A): + super().__init__(name="monitor", daemon=True) + self.rcu = rcu + self.tpu = tpu + self.relay = relay + self.recorder = recorder + self.t0 = t0 + self.interval = interval + self.current_threshold = current_threshold + + self._stop = threading.Event() + self._state_lock = threading.Lock() + self._cycle = 0 + self._phase = "startup" + self._note = "" + self._jig = (None, None, None) + self._relay_cmd = 0 + self._charging_cmd = 0 + self._rcu_powered = False + self._ok_since = None + self._latest = {} + + # --- context set by the main thread ------------------------------------ + def set_context(self, cycle=None, phase=None, jig=None, + relay_cmd=None, charging_cmd=None, note=None, + rcu_powered=None): + with self._state_lock: + if cycle is not None: + self._cycle = cycle + if phase is not None: + self._phase = phase + if jig is not None: + self._jig = tuple(jig) + if relay_cmd is not None: + self._relay_cmd = int(bool(relay_cmd)) + if charging_cmd is not None: + self._charging_cmd = int(bool(charging_cmd)) + if note is not None: + self._note = note + if rcu_powered is not None: + self._rcu_powered = bool(rcu_powered) + + def reset_dwell(self): + with self._state_lock: + self._ok_since = None + + def dwell_seconds(self): + """Seconds the charging criteria have held continuously (0 if not).""" + with self._state_lock: + if self._ok_since is None: + return 0.0 + return time.time() - self._ok_since + + def latest(self): + with self._state_lock: + return dict(self._latest) + + def stop(self): + self._stop.set() + + # --- loop -------------------------------------------------------------- + def run(self): + while not self._stop.is_set(): + started = time.time() + self._sample(started) + # Keep the cadence steady regardless of how long the reads took. + self._stop.wait(max(0.0, self.interval - (time.time() - started))) + + def _sample(self, now): + with self._state_lock: + rcu_powered = self._rcu_powered + + # With the relay open the RCU is dead: every read would burn a full + # serial timeout and stall the sampling cadence, so skip it entirely + # and log the row with the RCU columns blank. + rcu = self.rcu.read_data() if rcu_powered else dict(BLANK_RCU_DATA) + tpu = self.tpu.read_data() + relay = self.relay.read_data() + + criteria_met = ( + rcu['main_state'] == RCU_STATE_CHARGING + and rcu['output_current'] is not None + and rcu['output_current'] > self.current_threshold + ) + + with self._state_lock: + if criteria_met: + if self._ok_since is None: + self._ok_since = now + else: + self._ok_since = None + dwell = 0.0 if self._ok_since is None else now - self._ok_since + + cycle, phase, note = self._cycle, self._phase, self._note + jig_x, jig_y, jig_z = self._jig + relay_cmd, charging_cmd = self._relay_cmd, self._charging_cmd + powered = self._rcu_powered + self._latest = {'rcu': rcu, 'tpu': tpu, 'relay': relay} + + self.recorder.write([ + datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S.%f")[:-3], + round(now - self.t0, 1), cycle, phase, + jig_x, jig_y, jig_z, relay_cmd, charging_cmd, int(powered), + rcu['main_pre_state'], rcu['main_state'], + rcu['output_current'], rcu['output_voltage'], rcu['dc_bus_voltage'], + rcu['error_code'], rcu['shadow_error'], + tpu['main_pre_state'], tpu['main_state'], + tpu['output_current'], tpu['output_voltage'], tpu['dc_bus_voltage'], + tpu['error_code'], tpu['shadow_error'], tpu['rcu_error_code'], + relay['main_state'], round(dwell, 1), note, + ]) + + logger.info( + " [%5.1fs] RCU state=%s I=%s A V=%s V | TPU state=%s I=%s A " + "| relay state=%s | dwell=%.0fs", + now - self.t0, rcu['main_state'], rcu['output_current'], + rcu['output_voltage'], tpu['main_state'], tpu['output_current'], + relay['main_state'], dwell) + + +# --------------------------------------------------------------------------- +# Test sequence +# --------------------------------------------------------------------------- +def _wait(seconds, stop_when=None, poll=0.2): + """Sleep while the monitor keeps logging. Returns True if stop_when fired.""" + deadline = time.time() + seconds + while time.time() < deadline: + if stop_when is not None and stop_when(): + return True + time.sleep(poll) + return False + + +def run_cycle(cycle, rcu, relay, jig, monitor, cfg): + """Run one OVP cycle. Returns 'PASS' or 'TIMEOUT'.""" + logger.info("=" * 62) + logger.info("CYCLE %d/%d", cycle, cfg['cycles']) + + # Step 1: close the relay and let the RCU boot. The previous cycle left it + # open, so until this lands there is nothing on the other end to talk to. + # rcu_powered is only raised once the settle is over: the monitor would + # otherwise burn a full serial timeout per register on a booting RCU. + monitor.set_context(cycle=cycle, phase="relay_on", relay_cmd=1, note="") + power_up_rcu(relay) + monitor.set_context(rcu_powered=True) + + # Step 2: move to the charging position and let it settle. Without a jig + # the pads are already fixed at the charging distance, so only wait. + if jig is not None: + monitor.set_context(phase="jig_near", jig=NEAR_POSITION) + jig.move_to(*NEAR_POSITION) + else: + monitor.set_context(phase="settle") + _wait(JIG_SETTLE_S) + + # Step 3: start charging and wait for the criteria. + monitor.reset_dwell() + monitor.set_context(phase="charging", charging_cmd=1) + rcu.set_charging(True) + + logger.info( + "Waiting for RCU main_state=%d and output current > %.1f A held for " + "%.0fs (timeout %.0fs)...", + RCU_STATE_CHARGING, cfg['current_threshold'], cfg['dwell'], + cfg['timeout']) + + met = _wait(cfg['timeout'], + stop_when=lambda: monitor.dwell_seconds() >= cfg['dwell']) + result = "PASS" if met else "TIMEOUT" + + if met: + logger.info("Cycle %d: criteria met after %.1fs of dwell.", + cycle, monitor.dwell_seconds()) + else: + latest = monitor.latest().get('rcu', {}) + logger.warning( + "Cycle %d: TIMEOUT after %.0fs (last RCU state=%s, current=%s A).", + cycle, cfg['timeout'], latest.get('main_state'), + latest.get('output_current')) + + monitor.set_context(note=result) + + # Step 4: open the relay. This also cuts the RCU's power, so from here on + # the RCU is dead and its telemetry columns stay blank. + relay.set_control(False) + monitor.set_context(phase="relay_off", relay_cmd=0, rcu_powered=False) + + # Step 5: move away, keep monitoring. Without a jig there is nothing to + # move, so this becomes a plain observation window after the relay opens. + if jig is not None: + monitor.set_context(phase="jig_far", jig=FAR_POSITION) + jig.move_to(*FAR_POSITION) + else: + monitor.set_context(phase="observe") + _wait(FAR_OBSERVE_S) + + # Step 6: stop charging and settle before the next cycle. The relay is + # already open, so this write is best-effort housekeeping - an unpowered + # RCU cannot acknowledge it, and that is not a test failure. + try: + rcu.set_charging(False) + except Exception as exc: + logger.info("Charging cmd = 0 not acknowledged, as expected with the " + "relay open and the RCU unpowered (%s).", exc) + monitor.set_context(phase="charging_off", charging_cmd=0) + _wait(CYCLE_GAP_S) + monitor.set_context(note="") + + logger.info("Cycle %d finished: %s", cycle, result) + return result + + +def ovp_test(rcu_port, tpu_port, relay_port, jig_port, cycles, cfg, stamp): + os.makedirs(RESULTS_DIR, exist_ok=True) + csv_path = os.path.join(RESULTS_DIR, "ovp_test_%s.csv" % stamp) + + rcu = tpu = relay = jig = None + monitor = None + recorder = None + results = [] + + try: + rcu = RCU(rcu_port) + tpu = TPU(tpu_port) + relay = Relay(relay_port) + jig = Jig(jig_port) if jig_port else None + if jig is None: + logger.info("No jig port given - running without the jig; keep " + "the pads fixed at the charging distance.") + + # The RCU is powered through the relay, so close it and let the RCU + # boot before anything - the pre-flight probe included - talks to it. + power_up_rcu(relay) + + # Pre-flight: prove every port talks before committing to a long run. + if cfg['link_test']: + probe_devices = [rcu, tpu, relay] + if jig is not None: + probe_devices.append(jig) + all_ok, any_dead = link_test(probe_devices) + if any_dead: + logger.error("A device did not answer at all - aborting. " + "Check the -r/-t/-l/-j port assignments.") + return + if all_ok: + logger.info("Link test: PASS") + else: + logger.warning("Link test: PARTIAL - some registers could not " + "be read; continuing anyway.") + + # Known-safe starting point before any logging begins. The relay stays + # closed: cycle 1 re-asserts it and waits out the settle again. + rcu.set_charging(False) + + recorder = CSVRecorder(csv_path) + logger.info("Logging telemetry to: %s", os.path.abspath(csv_path)) + + monitor = Monitor(rcu, tpu, relay, recorder, t0=time.time(), + interval=cfg['interval'], + current_threshold=cfg['current_threshold']) + monitor.set_context(cycle=0, phase="startup", + jig=(0, 0, 0) if jig is not None else None, + relay_cmd=1, charging_cmd=0, rcu_powered=True) + monitor.start() + + for cycle in range(1, cycles + 1): + results.append( + run_cycle(cycle, rcu, relay, jig, monitor, cfg)) + + except KeyboardInterrupt: + logger.warning("Interrupted by user.") + except minimalmodbus.ModbusException as exc: + logger.error("Modbus error: %s", exc) + except Exception as exc: + logger.exception("Unexpected error: %s", exc) + finally: + if monitor is not None: + monitor.stop() + monitor.join(timeout=cfg['interval'] + SERIAL_TIMEOUT * 4) + + # Always leave the rig de-energised. + if rcu is not None: + try: + rcu.set_charging(False) + except Exception as exc: + # Expected when we stop with the relay already open. + logger.info("Charging cmd = 0 not acknowledged (RCU likely " + "unpowered): %s", exc) + if relay is not None: + try: + relay.set_control(False) + except Exception as exc: + logger.warning("Could not switch relay off: %s", exc) + + for dev in (rcu, tpu, relay, jig): + if dev is not None: + dev.close() + if recorder is not None: + recorder.close() + + # Nothing to summarise if we bailed out before logging started. + if recorder is not None: + passed = results.count("PASS") + logger.info("=" * 62) + logger.info("Cycles run: %d/%d | PASS: %d | TIMEOUT: %d", + len(results), cycles, passed, len(results) - passed) + for i, res in enumerate(results, start=1): + logger.info(" cycle %d: %s", i, res) + logger.info("Results saved to: %s", os.path.abspath(csv_path)) + + +# --------------------------------------------------------------------------- +def main(): + parser = argparse.ArgumentParser( + description="BE OVP cycling test across RCU, TPU, relay and jig.") + parser.add_argument("-r", "--rcu", required=True, + help="COM port for the RCU (e.g. COM3)") + parser.add_argument("-t", "--tpu", required=True, + help="COM port for the TPU") + parser.add_argument("-l", "--relay", required=True, + help="COM port for the relay board") + parser.add_argument("-j", "--jig", default=None, + help="COM port for the jig. Omit to run without one: " + "the pads must then already be fixed at the " + "charging distance, and the move-away phase " + "becomes a plain observation window.") + parser.add_argument("-n", "--cycles", type=int, default=None, + help="Number of test cycles " + "(not needed with --link-test)") + parser.add_argument("--link-test", action="store_true", + help="Read a few registers from every port, report " + "per port, then exit without running any cycles") + parser.add_argument("--no-link-test", action="store_true", + help="Skip the pre-flight link test that normally " + "runs before the first cycle") + parser.add_argument("--current-threshold", type=float, + default=CHARGING_CURRENT_THRESHOLD_A, + help="Output current the RCU must exceed, in amps " + "(default: %(default)s)") + parser.add_argument("--dwell", type=float, default=CHARGING_DWELL_S, + help="Seconds the criteria must hold " + "(default: %(default)s)") + parser.add_argument("--timeout", type=float, default=CHARGING_TIMEOUT_S, + help="Seconds before a cycle is abandoned " + "(default: %(default)s)") + parser.add_argument("--interval", type=float, default=MONITOR_INTERVAL_S, + help="Telemetry sampling interval in seconds " + "(default: %(default)s)") + args = parser.parse_args() + + if not args.link_test: + if args.cycles is None: + parser.error("-n/--cycles is required unless --link-test is given") + if args.cycles < 1: + parser.error("--cycles must be at least 1") + if args.dwell > args.timeout: + parser.error("--dwell cannot exceed --timeout") + + os.makedirs(RESULTS_DIR, exist_ok=True) + + # One log per run, stamped like its CSV so the two pair up by name. A + # link test writes its own file with a _linktest suffix, so a quick probe + # never lands in the middle of a real run's log. + stamp = datetime.datetime.now().strftime("%Y%m%d_%H%M%S") + log_name = "ovp_test_%s%s.log" % (stamp, + "_linktest" if args.link_test else "") + log_path = os.path.join(RESULTS_DIR, log_name) + + logging.basicConfig( + level=logging.INFO, + format="%(asctime)s - %(levelname)s - %(message)s", + datefmt="%Y-%m-%d %H:%M:%S", + handlers=[logging.FileHandler(log_path, encoding="utf-8"), + logging.StreamHandler(sys.stdout)]) + logger.info("Logging this run to: %s", os.path.abspath(log_path)) + + if args.link_test: + ok = run_link_test(args.rcu, args.tpu, args.relay, args.jig) + sys.exit(0 if ok else 1) + + cfg = { + 'cycles': args.cycles, + 'current_threshold': args.current_threshold, + 'dwell': args.dwell, + 'timeout': args.timeout, + 'interval': args.interval, + 'link_test': not args.no_link_test, + } + ovp_test(args.rcu, args.tpu, args.relay, args.jig, args.cycles, cfg, + stamp) + + +if __name__ == "__main__": + main() diff --git a/relay_test.py b/relay_test.py new file mode 100644 index 0000000..f5cb399 --- /dev/null +++ b/relay_test.py @@ -0,0 +1,110 @@ +#!/usr/bin/env python +""" +Small coil on/off tester for the BE rig. + +Defaults to the PFC relay (coil 12, "PFC_relay_enable") on the relay board, +but any coil address works via -a. + +Usage: + python relay_test.py -p COM13 --on # close the relay + python relay_test.py -p COM13 --off # open it + python relay_test.py -p COM13 --read # read the coil back + python relay_test.py -p COM13 --toggle 5 -n 3 # 3 on/off cycles, 5s apart + python relay_test.py -p COM4 -a 0 --on # RCU charging cmd instead +""" + +import argparse +import sys +import time + +import minimalmodbus + +SLAVE_ID = 0x10 +BAUDRATE = 9600 +SERIAL_TIMEOUT = 1.0 + +RELAY_CONTROL_CMD_ADDRESS = 12 # coil - PFC_relay_enable, 1 = on + + +def connect(port, slave_id): + inst = minimalmodbus.Instrument(port=port, slaveaddress=slave_id, + mode='rtu') + inst.serial.baudrate = BAUDRATE + inst.serial.timeout = SERIAL_TIMEOUT + inst.close_port_after_each_call = False + return inst + + +def read_coil(inst, address): + """Read the coil back with function code 1. Not every board supports it.""" + try: + return inst.read_bit(address, functioncode=1) + except Exception as exc: + return "unreadable (%s)" % exc + + +def write_coil(inst, address, value): + inst.write_bit(address, 1 if value else 0, functioncode=5) + + +def main(): + parser = argparse.ArgumentParser(description="Toggle a Modbus coil.") + parser.add_argument("-p", "--port", required=True, + help="COM port (e.g. COM13)") + parser.add_argument("-a", "--address", type=int, + default=RELAY_CONTROL_CMD_ADDRESS, + help="Coil address (default: %(default)s, the PFC " + "relay)") + parser.add_argument("-s", "--slave", type=lambda v: int(v, 0), + default=SLAVE_ID, + help="Slave id (default: 16)") + action = parser.add_mutually_exclusive_group(required=True) + action.add_argument("--on", action="store_true", help="Write 1") + action.add_argument("--off", action="store_true", help="Write 0") + action.add_argument("--read", action="store_true", help="Read the coil") + action.add_argument("--toggle", type=float, metavar="SECONDS", + help="Alternate on/off, holding each state this long") + parser.add_argument("-n", "--cycles", type=int, default=1, + help="Number of on/off cycles for --toggle " + "(default: %(default)s)") + args = parser.parse_args() + + try: + inst = connect(args.port, args.slave) + except Exception as exc: + print("Connect failed on %s: %s" % (args.port, exc)) + return 1 + + print("%s slave %d coil %d" % (args.port, args.slave, args.address)) + + try: + if args.read: + print(" read = %s" % read_coil(inst, args.address)) + + elif args.on or args.off: + value = bool(args.on) + write_coil(inst, args.address, value) + print(" wrote = %d" % value) + print(" read = %s" % read_coil(inst, args.address)) + + else: + for cycle in range(1, args.cycles + 1): + for value in (1, 0): + write_coil(inst, args.address, value) + print(" cycle %d: wrote %d, read %s" + % (cycle, value, read_coil(inst, args.address))) + time.sleep(args.toggle) + + except KeyboardInterrupt: + print("Interrupted.") + except Exception as exc: + print(" FAILED: %s" % exc) + return 1 + finally: + inst.serial.close() + + return 0 + + +if __name__ == "__main__": + sys.exit(main())