vanminh/ovp-test
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
9832b85525ea47c7bb2016cb58679de0b2a35948
ovp-test/BE-RCU-ovp_test_csv.py
T

1984 lines
81 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.
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 --observe seconds.
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)
RCU_VF_STATUS_ADDRESS = 41 # input - 0 = RUN_PI_1, 1 = RUN_PI_3, 2 = RUN_VF_2
RCU_VF_FREQ_ADDRESS = 42 # input - TX switching frequency
RCU_RUNTIME_CURRENT_ADDRESS = 9 # holding, Q128 - the current actually in use.
# Only writable once charging is running: before
# that the app overwrites it from current_setting
# (reg 14), so an early write is lost.
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 = 2.0 # keep monitoring after the cut
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
RAMP_TIMEOUT_S = 30.0 # max wait for the current to reach the cut point
RAMP_TRIGGER_PCT = 50.0 # how far between the two setpoints the cut lands
# --- 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 -----------------------------------------------------
# The rig switches relays and transfers power right next to the RS485
# pair, so a corrupted frame now and then is expected: a single flipped
# bit shows up as a CRC error. Retrying is safe because every transaction
# here is idempotent - a read, or a write of an absolute value. Without
# this one bit of line noise ends a 100-cycle run.
IO_RETRIES = 4
IO_RETRY_DELAY = 0.15
def _io(self, fn, *args, **kwargs):
# retries=1 for a transaction whose failure is the expected outcome:
# retrying a device known to be unpowered only burns serial timeouts.
tries = kwargs.pop('retries', None) or self.IO_RETRIES
last = None
for attempt in range(tries):
try:
return fn(*args, **kwargs)
except minimalmodbus.ModbusException as exc:
last = exc
if attempt + 1 < tries:
logger.debug("%s retry %d/%d: %s", self.name,
attempt + 1, tries - 1, exc)
time.sleep(self.IO_RETRY_DELAY)
raise last
def read_u16(self, address):
with self._lock:
return self._io(self.inst.read_register, address, functioncode=4)
def read_holding_u16(self, address):
with self._lock:
return self._io(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._io(self.inst.read_float, address,
functioncode=4, number_of_registers=2)
def write_coil(self, address, value, retries=None):
with self._lock:
self._io(self.inst.write_bit, address,
1 if value else 0, functioncode=5, retries=retries)
def write_holding(self, address, value):
# Jig coordinates may be negative; Modbus registers are unsigned.
with self._lock:
self._io(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': '',
'vf_status': None,
'vf_freq': None,
}
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, retries=None):
logger.info("RCU charging cmd = %d", 1 if enable else 0)
self.write_coil(RCU_CHARGING_CMD_ADDRESS, enable, retries=retries)
def set_current(self, amps):
"""Runtime charging current, amps. Call only once charging is running."""
logger.info("RCU runtime current = %.1f A", amps)
self.write_holding(RCU_RUNTIME_CURRENT_ADDRESS, int(round(amps * 128)))
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)),
# Which power zone the TX is in. The SR swaps its duty and its phase
# compensation on this, so a trip in VF is not the same event as a
# trip in PI - without this column the run cannot be interpreted.
'vf_status': _safe(self.read_u16, RCU_VF_STATUS_ADDRESS),
'vf_freq': _safe(self.read_u16, RCU_VF_FREQ_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
# ---------------------------------------------------------------------------
# ---------------------------------------------------------------------------
# Scope
#
# One capture per cycle, taken at the relay cut: armed single-shot just before
# the cut and read back once it has triggered.
#
# The gap between SR2H turning off and SR2L turning on is computed here from
# the raw waveform, not from a scope parameter. A parameter reports whichever
# edge pair it finds and that is almost always an ordinary switching cycle
# rather than the trip. After the trip both gates stay latched, so the LAST
# transition of each is unambiguously the trip.
#
# Channels: C1 = SR2H, C2 = SR2L, C3 = SR1H, C4 = SR1L.
# A negative gap means the two gates of leg 2 overlap.
# ---------------------------------------------------------------------------
SCOPE_LAN_RESOURCE = "TCPIP0::%s::inst0::INSTR" # --scope-ip fills in the host
# One entry per instrument on the bench. Everything here is a property of that
# particular scope and of the PC driving it - the VISA address, the folder it
# writes captures to, the share it can copy them to - so moving the test to
# another scope is a matter of picking a profile, not of editing constants.
SCOPE_PROFILES = {
# HDO8108A over USBTMC, on the PC it has always lived on.
"hdo": {
"resource": "USB0::0x05FF::0x1023::3808N60334::INSTR",
"dir": r"D:\vanminh\gm ovp",
"copy_dir": r"Z:\vanminh\be semicon\GM OVP\auto test",
},
# WaveRunner 8038HD over USBTMC. USB rather than LAN on purpose: the
# instrument has one Ethernet port, and a cable straight to the PC takes it
# away from the network, which is how it reaches the share below. For LAN
# instead, pass --scope-ip; the captures then stay on the scope's own disk.
"wr8038": {
"resource": "USB0::0x05FF::0x1023::5002N60075::INSTR",
"dir": r"D:\vanminh\gm ovp",
"copy_dir": r"Z:\vanminh\be semicon\GM OVP\auto test",
},
}
DEFAULT_SCOPE_PROFILE = "hdo"
SCOPE_RESOURCE = SCOPE_PROFILES[DEFAULT_SCOPE_PROFILE]["resource"]
SCOPE_DIR = SCOPE_PROFILES[DEFAULT_SCOPE_PROFILE]["dir"]
SCOPE_COPY_DIR = SCOPE_PROFILES[DEFAULT_SCOPE_PROFILE]["copy_dir"]
SCOPE_PARAMS = list(range(1, 10)) # P1..P9
SCOPE_OPEN_RETRIES = 4 # a just-closed VXI-11 session blocks the next
SCOPE_OPEN_DELAY_S = 3.0
WAVEFORM_TAIL_S = 50e-6 # of the record, enough for several periods
SCOPE_ARM_TIMEOUT_S = 20.0
SCOPE_IMAGE_TIMEOUT_S = 20.0
SAVE_WF_BELOW_NS = 150.0 # keep the raw traces when the gap is this tight
# Gate thresholds, volts. The gate lines ring by several volts after the trip
# latches, so a single-threshold edge search locks onto the ringing instead.
GATE_V_TH = 6.0
GATE_V_HI = 9.0
GATE_V_LO = 3.0
_TRUE = ("-1", "1", "true")
class Scope:
"""Teledyne LeCroy over VISA. Best-effort throughout: a scope problem must
not abort a run that is otherwise producing good telemetry."""
def __init__(self, resource, image_dir, copy_dir, params):
import pyvisa
self.rm = pyvisa.ResourceManager()
# The scope serves one remote session at a time and does not always
# release it the instant the previous one closes, so an open refused
# seconds after a finished run usually succeeds on the next try.
self.ins = None
last = None
for attempt in range(SCOPE_OPEN_RETRIES):
try:
self.ins = self.rm.open_resource(resource)
break
except Exception as exc:
last = exc
if attempt + 1 < SCOPE_OPEN_RETRIES:
logger.info("scope open failed (%s) - retrying in %.0fs "
"(%d/%d)", exc, SCOPE_OPEN_DELAY_S,
attempt + 1, SCOPE_OPEN_RETRIES - 1)
time.sleep(SCOPE_OPEN_DELAY_S)
if self.ins is None:
raise last
self.ins.timeout = 20000
# A session that was killed mid-transaction leaves unread data on the
# scope, and the next open then fails with VI_ERROR_INP_PROT_VIOL -
# which looks exactly like the scope having vanished. Clear it first.
try:
self.ins.clear()
time.sleep(0.3)
except Exception as exc:
logger.debug("device clear: %s", exc)
self.ins.write("COMM_HEADER OFF")
self.ins.write("COMM_FORMAT DEF9,WORD,BIN")
self.idn = self.ins.query("*IDN?").strip()
self.image_dir = image_dir.rstrip("\\")
self.copy_dir = copy_dir.rstrip("\\") if copy_dir else None
self.param_ids = list(params)
self.images = []
self.ext = self._hardcopy_ext()
self.labels = []
for i in self.param_ids:
eng = self._vbs("app.Measure.P%d.ParamEngine" % i) or ("P%d" % i)
src = self._vbs("app.Measure.P%d.Source1" % i) or ""
self.labels.append("P%d_%s_%s" % (i, eng, src))
if not self.folder_exists(self.image_dir):
raise IOError("scope cannot see %s" % self.image_dir)
if self.copy_dir and not self.folder_exists(self.copy_dir):
logger.warning("scope cannot see %s - images stay on its disk",
self.copy_dir)
self.copy_dir = None
# --- low level ---
def _vbs(self, expr):
try:
return self.ins.query("VBS? 'return=%s'" % expr).strip()
except Exception:
return ""
def _fso(self, expr):
return self._vbs('CreateObject("Scripting.FileSystemObject").%s' % expr)
def folder_exists(self, path):
return self._fso('FolderExists("%s")' % path).strip().lower() in _TRUE
def file_size(self, path):
if self._fso('FileExists("%s")' % path).strip().lower() not in _TRUE:
return None
try:
return float(self._fso('GetFile("%s").Size' % path))
except ValueError:
return None
def _hardcopy_ext(self):
"""Follow the format the operator set; do not assume jpg."""
try:
setup = self.ins.query("HARDCOPY_SETUP?").strip().upper()
except Exception:
return "png"
for dev, ext in (("JPEG", "jpg"), ("JPG", "jpg"), ("PNG", "png"),
("BMP", "bmp"), ("TIFF", "tif")):
if "DEV," + dev in setup:
return ext
return "png"
# --- acquisition ---
def run(self):
"""Free-run. The previous cycle left the scope stopped on its capture;
leaving it there means flying blind through the whole charge."""
self.ins.write("TRMD AUTO")
self.ins.query("*OPC?")
def arm(self):
self.ins.write("TRMD SINGLE")
self.ins.query("*OPC?")
def wait_triggered(self, timeout_s=SCOPE_ARM_TIMEOUT_S):
deadline = time.time() + timeout_s
while time.time() < deadline:
try:
if self.ins.query("TRMD?").strip().upper().startswith("STOP"):
return True
except Exception:
return False
time.sleep(0.05)
return False
def stop(self):
self.ins.write("STOP")
self.ins.query("*OPC?")
def read_params(self):
out = []
for i in self.param_ids:
raw = self._vbs("app.Measure.P%d.Out.Result.Value" % i)
try:
out.append(float(raw))
except ValueError:
out.append("")
return out
# --- waveform ---
def _wave(self, ch):
"""Raw samples of the tail of the record, with the scaling to read them.
Only the tail is fetched: _fall and _rise both scan backwards from the
end, so everything before the last few periods is transferred and then
thrown away. On a deep-memory scope over LAN that is the difference
between seconds and half a minute per cycle, and it leaves the scope's
own record untouched for the LabNotebook.
"""
def insp(name):
text = self.ins.query('%s:INSPECT? "%s"' % (ch, name))
return float(text.split(":")[-1].strip().rstrip('"'))
# WAVE_ARRAY_COUNT reports what the current setup would transfer, so
# the full length has to be read with the setup reset to "everything".
self.ins.write("WAVEFORM_SETUP SP,0,NP,0,FP,0,SN,0")
dt = insp("HORIZ_INTERVAL")
total = int(insp("WAVE_ARRAY_COUNT"))
points = total
if dt > 0:
points = min(total, max(1000, int(WAVEFORM_TAIL_S / dt)))
if points < total:
self.ins.write("WAVEFORM_SETUP SP,0,NP,%d,FP,%d,SN,0"
% (points, total - points))
raw = self.ins.query_binary_values("%s:WAVEFORM? DAT1" % ch,
datatype="h", is_big_endian=False,
container=list)
return raw, insp("VERTICAL_GAIN"), insp("VERTICAL_OFFSET"), dt
@staticmethod
def _code(volts, gain, offset):
return (volts + offset) / gain
def _fall(self, raw, gain, offset):
hi = self._code(GATE_V_HI, gain, offset)
th = self._code(GATE_V_TH, gain, offset)
k = len(raw) - 1
while k >= 0 and raw[k] <= hi:
k -= 1
if k < 0:
return None
i = k
while i < len(raw) and raw[i] >= th:
i += 1
return i if i < len(raw) else None
def _rise(self, raw, gain, offset):
lo = self._code(GATE_V_LO, gain, offset)
th = self._code(GATE_V_TH, gain, offset)
k = len(raw) - 1
while k >= 0 and raw[k] >= lo:
k -= 1
if k < 0:
return None
i = k
while i < len(raw) and raw[i] <= th:
i += 1
return i if i < len(raw) else None
def trip_gap_ns(self):
"""SR2H off -> SR2L on at the trip, in ns. None when the record holds
no latched trip, which is what a trip arriving while SR2L was already
on looks like."""
try:
r1, g1, o1, dt = self._wave("C1")
r2, g2, o2, _ = self._wave("C2")
except Exception as exc:
logger.warning("scope waveform read failed: %s", exc)
return None
if (r1[-1] * g1 - o1) > GATE_V_TH or (r2[-1] * g2 - o2) < GATE_V_TH:
return None
i1 = self._fall(r1, g1, o1)
i2 = self._rise(r2, g2, o2)
if i1 is None or i2 is None:
return None
return round((i2 - i1) * dt * 1e9, 1)
# --- image ---
def capture(self, name):
filename = name + "." + self.ext
path = self.image_dir + "\\" + filename
self.ins.write('HARDCOPY_SETUP DEST,FILE,DIR,"%s\\",FILE,"%s"'
% (self.image_dir, name))
self.ins.write("VBS 'app.Hardcopy.Print'")
deadline = time.time() + SCOPE_IMAGE_TIMEOUT_S
last = None
while time.time() < deadline:
size = self.file_size(path)
if size and size == last:
self.images.append(filename)
return filename
last = size
time.sleep(0.2)
logger.warning("scope image %s not written", path)
return ""
def save_labnotebook(self, name):
"""Write a LabNotebook beside the capture: same folder, same base name.
One .lnb holds the screen image, the setup and a trace per displayed
waveform, and Flashback restores the instrument from it - which is what
makes a tight cycle debuggable later, where the image alone is not.
Requires "Annotate Screen Image Before Saving" to be OFF in the scope's
File > Save LabNotebook dialog. With it on, Save opens the drawing
toolbar and blocks until someone presses Done on the instrument, which
hangs the remote session.
"""
target = "%s\\%s.lnb" % (self.image_dir, name)
try:
self.ins.write('VBS \'app.LabNotebook.SaveFileName = "%s"\'' % target)
self.ins.write("VBS 'app.LabNotebook.Save'")
self.ins.query("*OPC?")
return True
except Exception as exc:
logger.warning("saving LabNotebook for %s failed: %s", name, exc)
return False
def copy_images(self):
if not (self.copy_dir and self.images):
return
logger.info("copying %d scope images to %s", len(self.images),
self.copy_dir)
copied = 0
fso = 'CreateObject("Scripting.FileSystemObject")'
for name in self.images:
src = self.image_dir + "\\" + name
dst = self.copy_dir + "\\" + name
try:
self.ins.write("VBS '%s.CopyFile \"%s\", \"%s\", False'"
% (fso, src, dst))
self.ins.query("*OPC?")
copied += 1
except Exception as exc:
logger.warning(" %s: %s", name, exc)
logger.info("copied %d/%d images", copied, len(self.images))
def close(self):
try:
self.ins.close()
except Exception:
pass
class ScopeCSV:
"""One row per cycle: the gap, the operating point at the cut, P1..P9."""
# P1..P9 sit between the operating point and the two columns the run is
# read by, so Image and Trip_Gap_ns stay together at the far right.
HEADERS = ['Cycle', 'Timestamp', 'Result',
'RCU_VF_Status', 'RCU_VF_Freq', 'RCU_Output_Current_A',
'RCU_Output_Voltage_V', 'Current_Setpoint_A']
TAIL = ['Image', 'Trip_Gap_ns']
def __init__(self, path, labels):
self.path = path
self._file = open(path, 'w', newline='', encoding='utf-8')
self._writer = csv.writer(self._file)
self._writer.writerow(self.HEADERS + list(labels) + self.TAIL)
self._file.flush()
def write(self, row):
self._writer.writerow(row)
self._file.flush()
def close(self):
try:
self._file.close()
except Exception:
pass
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',
'RCU_VF_Status', 'RCU_VF_Freq',
'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, extra=()):
self.path = path
self.headers = list(self.HEADERS) + list(extra)
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):
row = list(row)
if len(row) < len(self.headers):
row += [''] * (len(self.headers) - len(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, extra=None):
"""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'],
blank['vf_status'], blank['vf_freq'],
None, None, None, None, None, '', '', '',
None, 0.0, note,
] + list(extra or []))
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'],
rcu['vf_status'], rcu['vf_freq'],
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 _has_current(monitor):
"""True once the RCU reports any output current at all."""
value = monitor.latest().get('rcu', {}).get('output_current')
return value is not None and value > 0.0
def _apply_runtime_current(rcu, monitor, cfg):
"""Set the runtime current (reg 9) now that charging has started.
It cannot be set any earlier: while the charger is off the app keeps
overwriting reg 9 from current_setting (reg 14), so an early write is lost.
Current flowing is the signal that charging has actually begun.
"""
if cfg.get('current') is None:
return
if _wait(cfg['timeout'], stop_when=lambda: _has_current(monitor)):
_safe(rcu.set_current, cfg['current'])
else:
logger.warning("No output current within %.0fs - runtime current not "
"applied.", cfg['timeout'])
def _scope_live(cfg):
"""Put the scope back to free-running at the start of a cycle."""
scope = cfg.get('scope')
if scope is not None:
_safe(scope.run)
def _arm_scope(cfg):
"""Sample the RCU operating point and arm the scope, both just before the
cut: the relay also cuts RCU power, so afterwards it answers nothing."""
pre_cut = dict(cfg['monitor'].latest().get('rcu', {}))
scope = cfg.get('scope')
if scope is not None:
try:
scope.arm()
except Exception as exc:
# Silently armed-or-not was the worst of both: the run carried on
# and the missing capture only showed up at the end.
logger.warning("scope arm failed, this cycle has no capture: %s",
exc)
return pre_cut
def _record_scope(cfg, cycle, result, pre_cut):
"""Read the scope after the cut and append one row to the scope CSV."""
scope = cfg.get('scope')
if scope is None:
return
if not scope.wait_triggered():
logger.warning("Cycle %d: scope did not trigger.", cycle)
_safe(scope.stop)
gap = scope.trip_gap_ns()
name = "ovp_c%03d_%s" % (cycle, cfg['stamp'])
image = scope.capture(name)
params = scope.read_params()
cfg['scope_csv'].write(
[cycle,
datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
result,
pre_cut.get('vf_status'), pre_cut.get('vf_freq'),
pre_cut.get('output_current'), pre_cut.get('output_voltage'),
"" if cfg.get('current') is None else cfg['current']]
+ params
+ [image, "" if gap is None else gap])
# Keep a full recall point for the cycles that are actually interesting.
# It stays on the scope only - it is for Flashback there, not for the run
# folder, so it is never queued for the copy to the network drive.
if (gap is not None and cfg.get('save_wf_below')
and gap < cfg['save_wf_below']):
if scope.save_labnotebook(name):
logger.info("Cycle %d: gap %.1f ns - LabNotebook saved as %s.lnb",
cycle, gap, name)
cfg['monitor'].write_event(
"scope", "trip gap %s" % ("n/a" if gap is None else "%.1f ns" % gap),
extra=params + [image, "" if gap is None else gap])
cfg.setdefault('gaps', []).append(gap)
logger.info("Cycle %d: trip gap %s, image %s", cycle,
"not latched" if gap is None else ("%.1f ns" % gap),
image or "none")
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'])
_scope_live(cfg)
# 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)
_apply_runtime_current(rcu, monitor, cfg)
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)
pre_cut = _arm_scope(cfg)
# 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)
_record_scope(cfg, cycle, result, pre_cut)
# 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(cfg['observe'])
# 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:
# One attempt: this is the belt-and-braces stop for a relay that failed
# to open. If it did open the RCU is dead and cannot answer, so the
# further attempts only cost a serial timeout each.
rcu.set_charging(False, retries=1)
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 run_cycle_ramp(cycle, rcu, relay, jig, monitor, cfg):
"""One OVP cycle with the relay cut while the current is moving between
two setpoints.
Charging settles at --current, the setpoint is then moved to --current2,
and the cut lands the moment the measured current has covered --ramp-at
percent of the distance between the two: 25 A -> 15 A at the default 50
percent cuts at 20 A.
Returns 'PASS', 'TIMEOUT', 'RAMP_TIMEOUT' or 'CUT_FAILED'.
"""
i1, i2 = cfg['current'], cfg['current2']
trip_at = i1 + (i2 - i1) * cfg['ramp_at'] / 100.0
falling = i2 < i1
logger.info("=" * 62)
logger.info("CYCLE %d/%d (ramp scenario)", cycle, cfg['cycles'])
_scope_live(cfg)
# 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: charging position, then settle.
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: charge at the first setpoint and let it establish.
monitor.reset_dwell()
monitor.set_context(phase="charging", charging_cmd=1)
rcu.set_charging(True)
_apply_runtime_current(rcu, monitor, cfg)
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)
# Arm before the setpoint moves: once the ramp starts there is no time.
pre_cut = _arm_scope(cfg)
# Step 4: move the setpoint and cut partway through the ramp.
monitor.set_context(phase="ramp")
logger.info("Cycle %d: ramp %.1f A -> %.1f A, cutting at %.1f A",
cycle, i1, i2, trip_at)
_safe(rcu.set_current, i2)
# The monitor samples far too slowly to catch the crossing, so poll the
# current directly. One Modbus read is ~20 ms at 9600 baud, and that is
# the time resolution of the cut.
reached = False
measured = None
deadline = time.time() + cfg['ramp_timeout']
while time.time() < deadline:
try:
value = _q128(rcu.read_u16(RCU_OUTPUT_CURRENT_ADDRESS))
except Exception:
continue
if value is None:
continue
measured = value
if (value <= trip_at) if falling else (value >= trip_at):
reached = True
break
# 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)
cut_ok = True
except Exception as exc:
logger.warning("Relay cut not acknowledged: %s", exc)
cut_ok = _cut_relay(relay, retry=False)
monitor.set_context(phase="relay_off", relay_cmd=0, rcu_powered=False)
if result == "PASS" and not reached:
result = "RAMP_TIMEOUT"
if not cut_ok:
result = "CUT_FAILED"
note = "cut at %s A (target %.1f A, %.1f -> %.1f)" % (
"n/a" if measured is None else "%.2f" % measured, trip_at, i1, i2)
if not reached:
note = "RAMP_TIMEOUT: " + note
logger.info("RELAY %s", note)
monitor.write_event("relay_cut", note)
monitor.set_context(note=result)
# The polled value is fresher than anything the monitor has, and it is the
# current the cut actually landed on.
if measured is not None:
pre_cut['output_current'] = measured
_record_scope(cfg, cycle, result, pre_cut)
# Step 5: observation window with the relay open.
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(cfg['observe'])
# Step 6: stop charging and settle. Best-effort: the RCU is unpowered.
try:
# One attempt: this is the belt-and-braces stop for a relay that failed
# to open. If it did open the RCU is dead and cannot answer, so the
# further attempts only cost a serial timeout each.
rcu.set_charging(False, retries=1)
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'])
_scope_live(cfg)
# 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)
_apply_runtime_current(rcu, monitor, cfg)
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)
pre_cut = _arm_scope(cfg)
# 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)
_record_scope(cfg, cycle, result, pre_cut)
_wait(max(0.0, JIG_MOVE_MAX_S - achieved))
# Step 5: stop charging. Best-effort - the RCU lost power at the cut.
try:
# One attempt: this is the belt-and-braces stop for a relay that failed
# to open. If it did open the RCU is dead and cannot answer, so the
# further attempts only cost a serial timeout each.
rcu.set_charging(False, retries=1)
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)
scope_csv_path = os.path.join(RESULTS_DIR, "ovp_scope_%s.csv" % stamp)
rcu = tpu = relay = jig = None
monitor = None
recorder = None
results = []
cfg['stamp'] = stamp
cfg['scope'] = None
cfg['scope_csv'] = None
if cfg.get('scope_resource'):
try:
cfg['scope'] = Scope(cfg['scope_resource'], cfg['scope_dir'],
cfg['scope_copy_dir'], SCOPE_PARAMS)
cfg['scope_csv'] = ScopeCSV(scope_csv_path, cfg['scope'].labels)
logger.info("Scope: %s", cfg['scope'].idn)
logger.info("Scope images -> %s (.%s)", cfg['scope'].image_dir,
cfg['scope'].ext)
except Exception as exc:
logger.warning("Scope unavailable, continuing without it: %s", exc)
cfg['scope'] = None
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)
scope_cols = []
if cfg.get('scope') is not None:
scope_cols = list(cfg['scope'].labels) + ['Scope_Image', 'Trip_Gap_ns']
recorder = CSVRecorder(csv_path, scope_cols)
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()
cfg['monitor'] = monitor
runner = {'move': run_cycle_move,
'ramp': run_cycle_ramp}.get(cfg['scenario'], 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)
if cfg.get('scope') is not None:
_safe(cfg['scope'].copy_images)
cfg['scope'].close()
if cfg.get('scope_csv') is not None:
cfg['scope_csv'].close()
logger.info("Scope CSV: %s", scope_csv_path)
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)
# The gap floor is the number the run exists to produce: the
# smallest separation seen between SR2H turning off and SR2L
# turning on. Negative means the two gates overlapped.
gaps = [g for g in cfg.get('gaps', []) if g is not None]
if cfg.get('gaps'):
logger.info("-" * 62)
logger.info("Trip gap readings: %d valid / %d cycles",
len(gaps), len(cfg['gaps']))
if gaps:
worst = min(gaps)
logger.info("Smallest trip gap: %+.1f ns%s", worst,
" <-- OVERLAP" if worst < 0 else "")
else:
logger.info("No valid gap: every trip arrived while SR2L "
"was already on")
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", type=float, default=None, metavar="A",
help="runtime charging current, amps. Written to RCU "
"holding reg 9 once current starts flowing - it "
"cannot be set before that, the charger overwrites "
"it from current_setting. Keep --current-threshold "
"below this value or the dwell never completes.")
parser.add_argument("--scope-profile", choices=sorted(SCOPE_PROFILES),
default=DEFAULT_SCOPE_PROFILE,
help="which scope on the bench: " + ", ".join(
"%s = %s" % (k, v["resource"])
for k, v in sorted(SCOPE_PROFILES.items()))
+ " (default: %(default)s)")
parser.add_argument("--scope", dest="scope_resource", metavar="VISA",
help="scope VISA resource, overriding the profile's")
parser.add_argument("--scope-ip", metavar="IP",
help="reach the scope over LAN at this address instead "
"of USB, as " + SCOPE_LAN_RESOURCE % "IP" +
". Needs the scope's Utilities > Utilities Setup > "
"Remote set to TCPIP (VXI11). Overrides --scope")
parser.add_argument("--no-scope", action="store_true",
help="run without the scope")
parser.add_argument("--save-wf-below", type=float, default=SAVE_WF_BELOW_NS,
metavar="NS",
help="also save a LabNotebook (.lnb), into the scope "
"folder and under the same base name as the image, "
"whenever the trip gap is below this. 0 disables. "
"Needs 'Annotate Screen Image Before Saving' off on "
"the scope (default: %(default)s)")
parser.add_argument("--scope-dir", metavar="DIR",
help="folder on the SCOPE for the captures, overriding "
"the profile's")
parser.add_argument("--scope-copy-dir", metavar="DIR",
help="share the scope copies the captures to at the "
"end, overriding the profile's. Empty to keep them "
"on the scope")
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("--observe", type=float, default=FAR_OBSERVE_S,
help="Seconds to keep sampling after the cut with the "
"relay open, which is what shows whether the RCU "
"really went down. Scenarios static and ramp only "
"- move waits out the jig travel instead "
"(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", "ramp"),
default="static",
help="static: cut the relay with the jig still. "
"move: cut it partway through a jig move, which "
"needs a jig. ramp: cut it partway through a "
"change of current setpoint, which needs "
"--current and --current2 (default: %(default)s)")
parser.add_argument("--current2", type=float, default=None, metavar="A",
help="second current setpoint for --scenario ramp. "
"Charging settles at --current, then moves here.")
parser.add_argument("--ramp-at", type=float, default=RAMP_TRIGGER_PCT,
metavar="PCT",
help="how far between the two setpoints the cut lands, "
"percent. 25 A -> 15 A at 50 cuts at 20 A "
"(default: %(default)s)")
parser.add_argument("--ramp-timeout", type=float, default=RAMP_TIMEOUT_S,
metavar="S",
help="give up waiting for the ramp to reach the cut "
"point after this long (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")
# The profile supplies the three scope settings; each explicit flag then
# overrides its own, and --scope-ip overrides the address whichever way it
# arrived. Only what is passed is overridden, so a profile stays usable when
# a single value needs changing for one run.
profile = SCOPE_PROFILES[args.scope_profile]
scope_resource = args.scope_resource or profile["resource"]
if args.scope_ip:
scope_resource = SCOPE_LAN_RESOURCE % args.scope_ip
scope_dir = profile["dir"] if args.scope_dir is None else args.scope_dir
scope_copy_dir = (profile["copy_dir"] if args.scope_copy_dir is None
else args.scope_copy_dir)
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': args.current,
'current2': args.current2,
'ramp_at': args.ramp_at,
'ramp_timeout': args.ramp_timeout,
'scope_resource': None if args.no_scope else scope_resource,
'scope_dir': scope_dir,
'save_wf_below': args.save_wf_below,
'scope_copy_dir': scope_copy_dir,
'current_threshold': args.current_threshold,
'dwell': args.dwell,
'observe': args.observe,
'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 == "ramp":
if args.current is None or args.current2 is None:
parser.error("--scenario ramp needs both --current and --current2")
trip_at = args.current + (args.current2 - args.current) * args.ramp_at / 100.0
logger.info("Scenario ramp: %.1f A -> %.1f A, cut at %.0f%% = %.1f A",
args.current, args.current2, args.ramp_at, trip_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()
Reference in New Issue View Git Blame Copy Permalink