#!/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. Two scenarios, picked with --scenario. Both establish charging and then cut the relay; they differ only in when the cut lands. static (default), jig optional: 1. Relay control = 1, wait RELAY_SETTLE_S for the RCU to power up. 2. Jig -> NEAR_POSITION, settle. 3. RCU charging cmd = 1, wait for RCU main_state == CHARGING(5) and output current above the threshold held continuously for the dwell. 4. Relay control = 0 (this also cuts RCU power). 5. Jig -> FAR_POSITION, keep monitoring for FAR_OBSERVE_S. 6. RCU charging cmd = 0, settle, next cycle. move, jig required - the cut lands while the coupling is still changing: 1-3. As above, but parked at (0, 0, --move-from). 4. Jig -> (0, 0, --move-to), relay cut at half the travel time. 5. RCU charging cmd = 0, settle. 6. Jig back to (0, 0, --move-from), last, so the whole post-cut window is observed at a standstill. A background monitor thread samples RCU and TPU input registers every MONITOR_INTERVAL_S for the whole run and appends each sample to the CSV. Usage: python BE-RCU-ovp_test_csv.py -r COM4 -t COM10 -l COM13 -j COM9 --link-test python BE-RCU-ovp_test_csv.py -r COM4 -t COM10 -l COM13 -n 1 python BE-RCU-ovp_test_csv.py -r COM4 -t COM10 -l COM13 -j COM9 \ --scenario move --move-from 20 --move-to 0 -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 CLOSE_TIMEOUT_S = 3.0 # give up on a port that will not close # --------------------------------------------------------------------------- # 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 = 10.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 = 2.0 # time to wait for relay to settle # --- Scenario "move": cut the relay midway through a jig move -------------- MOVE_X_MM = 0 # held for the whole scenario; only Z moves MOVE_Y_MM = 0 MOVE_Z_FROM_MM = 20 # where the move starts - the charging spot MOVE_Z_TO_MM = 0 # where it ends JIG_SPEED_MM_S = 10.0 # override with --jig-speed JIG_MOVE_MAX_S = 5.0 # time allowed for one move, in or out 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): # A USB adapter unplugged mid-run wedges serial.close() forever, and # Ctrl+C cannot break a blocking Windows call. Close on a daemon # thread so a dead port can never hold up the exit. closer = threading.Thread(target=self._close_serial, daemon=True) closer.start() closer.join(CLOSE_TIMEOUT_S) if closer.is_alive(): logger.warning("%s: %s did not close in %.0fs, abandoning it", self.name, self.port, CLOSE_TIMEOUT_S) def _close_serial(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): # Probe once: a dead RCU costs 9 x SERIAL_TIMEOUT otherwise. try: main_pre_state = self.read_u16(RCU_MAIN_PRE_STATE_ADDRESS) except Exception as exc: logger.debug("RCU did not answer: %s", exc) return dict(BLANK_RCU_DATA) return { 'main_pre_state': main_pre_state, '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): """Command a move. Returns the perf_counter() of the go command.""" 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) return time.perf_counter() # --------------------------------------------------------------------------- # 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 write_event(self, phase, note): """Append one row now, telemetry blank. set_context() only paints the next periodic sample, which is too late to record an instant.""" with self._state_lock: cycle = self._cycle jig_x, jig_y, jig_z = self._jig relay_cmd, charging_cmd = self._relay_cmd, self._charging_cmd powered = self._rcu_powered blank = dict(BLANK_RCU_DATA) self.recorder.write([ datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S.%f")[:-3], round(time.time() - self.t0, 1), cycle, phase, jig_x, jig_y, jig_z, relay_cmd, charging_cmd, int(powered), blank['main_pre_state'], blank['main_state'], blank['output_current'], blank['output_voltage'], blank['dc_bus_voltage'], blank['error_code'], blank['shadow_error'], None, None, None, None, None, '', '', '', None, 0.0, note, ]) 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 _cut_relay(relay, retry=True): """Open the relay, tolerating one lost frame. Returns True if it landed.""" try: relay.set_control(False) return True except Exception as exc: logger.warning("Relay cut not acknowledged: %s", exc) if not retry: return False try: relay.set_control(False) logger.info("Relay cut acknowledged on retry.") return True except Exception as exc: logger.error("Relay cut failed twice: %s", exc) return False def run_cycle(cycle, rcu, relay, jig, monitor, cfg): """Run one OVP cycle with the jig standing still. Returns 'PASS', 'TIMEOUT' or 'CUT_FAILED'. """ 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. if not _cut_relay(relay): result = "CUT_FAILED" monitor.set_context(phase="relay_off", relay_cmd=0, rcu_powered=False, note=result) # 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 move_timing(cfg): """(distance_mm, travel_s, trigger_s). --trigger-at only pulls it earlier.""" distance = abs(cfg['move_from'] - cfg['move_to']) travel = distance / cfg['jig_speed'] trigger = travel / 2.0 if cfg.get('trigger_at') is not None: trigger = min(trigger, cfg['trigger_at']) return distance, travel, trigger def run_cycle_move(cycle, rcu, relay, jig, monitor, cfg): """One OVP cycle with the relay cut midway through a jig move. Returns 'PASS', 'TIMEOUT' or 'CUT_FAILED'. """ distance, travel_s, trigger_s = move_timing(cfg) home = (cfg['move_x'], cfg['move_y'], cfg['move_from']) target = (cfg['move_x'], cfg['move_y'], cfg['move_to']) logger.info("=" * 62) logger.info("CYCLE %d/%d (move scenario)", cycle, cfg['cycles']) # Step 1: close the relay and let the RCU boot. monitor.set_context(cycle=cycle, phase="relay_on", relay_cmd=1, note="") power_up_rcu(relay) monitor.set_context(rcu_powered=True) # Step 2: park at the charging position. monitor.set_context(phase="jig_home", jig=home) jig.move_to(*home) _wait(JIG_MOVE_MAX_S) # Step 3: charge, and wait for it to be established - the cut means # nothing until then. 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: the move, with the relay cut partway through it. logger.info("Moving %d mm (%.2fs), cutting relay at %.2fs", distance, travel_s, trigger_s) monitor.set_context(phase="jig_move", jig=target) t_go = jig.move_to(*target) # Sleep straight through, not via _wait(): its 0.2s poll is far too coarse # for a sub-second target. remaining = (t_go + trigger_s) - time.perf_counter() if remaining > 0: time.sleep(remaining) # write_coil, not set_control(): that helper logs before it writes, and a # Windows console write blocks while text is selected in the terminal - # which here would swallow the cut itself. Log afterwards. try: relay.write_coil(RELAY_CONTROL_CMD_ADDRESS, False) achieved = time.perf_counter() - t_go cut_ok = True except Exception as exc: achieved = time.perf_counter() - t_go logger.warning("Relay cut not acknowledged at %.3fs: %s", achieved, exc) cut_ok = _cut_relay(relay, retry=False) monitor.set_context(phase="relay_cut", relay_cmd=0, rcu_powered=False) note = "cut at %.3fs of %.2fs travel" % (achieved, travel_s) if not cut_ok: result = "CUT_FAILED" note = "CUT_FAILED: " + note logger.info("RELAY %s", note) monitor.write_event("relay_cut", note) monitor.set_context(note=result) _wait(max(0.0, JIG_MOVE_MAX_S - achieved)) # Step 5: stop charging. Best-effort - the RCU lost power at the cut. 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) # Step 6: back to the charging position, last thing in the cycle, so the # whole post-cut window is observed at a standstill. monitor.set_context(phase="jig_return", jig=home) jig.move_to(*home) _wait(JIG_MOVE_MAX_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() runner = run_cycle_move if cfg['scenario'] == 'move' else run_cycle for cycle in range(1, cycles + 1): results.append(runner(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: # CUT_FAILED is counted apart from TIMEOUT: the relay never opened, # so the cycle proves nothing either way about the RCU. logger.info("=" * 62) logger.info("Cycles run: %d/%d | PASS: %d | TIMEOUT: %d " "| CUT_FAILED: %d", len(results), cycles, results.count("PASS"), results.count("TIMEOUT"), results.count("CUT_FAILED")) 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)") parser.add_argument("--scenario", choices=("static", "move"), default="static", help="static: cut the relay with the jig still. " "move: cut it partway through a jig move, which " "needs a jig (default: %(default)s)") # int, not float: Device.write_holding does `value & 0xFFFF`. parser.add_argument("--move-x", type=int, default=MOVE_X_MM, metavar="X", help="Scenario move: X in mm, held for the whole " "scenario (default: %(default)s)") parser.add_argument("--move-y", type=int, default=MOVE_Y_MM, metavar="Y", help="Scenario move: Y in mm, held for the whole " "scenario (default: %(default)s)") parser.add_argument("--move-from", type=int, default=MOVE_Z_FROM_MM, metavar="Z", help="Scenario move: Z in mm where the move starts, " "also the charging position " "(default: %(default)s)") parser.add_argument("--move-to", type=int, default=MOVE_Z_TO_MM, metavar="Z", help="Scenario move: Z in mm where the move ends " "(default: %(default)s)") parser.add_argument("--trigger-at", type=float, default=None, metavar="SECONDS", help="Scenario move: seconds after the go command to " "cut the relay (default: half the travel time). " "Only pulls the cut earlier") parser.add_argument("--jig-speed", type=float, default=JIG_SPEED_MM_S, metavar="MM_S", help="Scenario move: jig speed in mm/s. Decides where " "the cut lands, and the jig cannot confirm it - " "measure it (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") if args.scenario == "move" and not args.link_test: if not args.jig: parser.error("--scenario move needs -j/--jig") if args.jig_speed <= 0: parser.error("--jig-speed must be positive") if args.trigger_at is not None and args.trigger_at <= 0: parser.error("--trigger-at must be positive") distance = abs(args.move_from - args.move_to) if distance == 0: parser.error("--move-from and --move-to are the same") travel = distance / args.jig_speed # The whole travel, not half: step 5 re-commands the jig after # JIG_MOVE_MAX_S, and a new target mid-move is undocumented behaviour. if travel > JIG_MOVE_MAX_S: parser.error("%d mm at %.1f mm/s takes %.1fs, over the %.1fs " "move window (JIG_MOVE_MAX_S)" % (distance, args.jig_speed, travel, JIG_MOVE_MAX_S)) if travel / 2.0 < 0.2: print("WARNING: only %.0f ms before the cut; Modbus latency is " "15-35 ms" % (travel / 2.0 * 1000), file=sys.stderr) 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, 'scenario': args.scenario, 'move_x': args.move_x, 'move_y': args.move_y, 'move_from': args.move_from, 'move_to': args.move_to, 'jig_speed': args.jig_speed, 'trigger_at': args.trigger_at, } if args.scenario == "move": distance, travel, trigger = move_timing(cfg) logger.info("Scenario move: (%d,%d,%d) -> (%d,%d,%d), %d mm at " "%.1f mm/s = %.2fs travel, cut at %.2fs", args.move_x, args.move_y, args.move_from, args.move_x, args.move_y, args.move_to, distance, args.jig_speed, travel, trigger) ovp_test(args.rcu, args.tpu, args.relay, args.jig, args.cycles, cfg, stamp) if __name__ == "__main__": main()