vanminh/ovp-test
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
c0ca523143f485da47dc1e227059fdf6f84b5ccf
ovp-test/BE-RCU-ovp_test_csv.py
T
vanminh c0ca523143 init OVP test for BE
2026-09-23 10:07:00 +08:00

927 lines
36 KiB
Python
Raw Blame History

#!/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()
Reference in New Issue View Git Blame Copy Permalink