"""Minimal board test for the 2026 PLSR Modbus product register map. The script is inert unless --run is supplied. Motion phases additionally require --allow-motion because they drive the configured pulse output. """ import argparse import dataclasses import io import math from pathlib import Path import re import subprocess import sys import time import serial CONFIG_BASE = 0x1000 COMMON_WORDS = 0x14 SEGMENT_1_BASE = 0x1100 SEGMENT_2_BASE = 0x1110 SEGMENT_WORDS = 8 STATUS_BASE = 0x2000 STATUS_WORDS = 7 CONTROL = 0x3000 PERSISTENCE_LIVE_CONFIG_ADDRESS = CONFIG_BASE + 0x05 PERSISTENCE_LIVE_MIN_DURATION_SECONDS = 2.5 PERSISTENCE_LIVE_DEFAULT_DURATION_SECONDS = 2.5 PERSISTENCE_LIVE_DEFAULT_INTERVAL_SECONDS = 0.05 PERSISTENCE_LIVE_MAX_INTERVAL_SECONDS = 0.1 PERSISTENCE_LIVE_REQUEST_TIMEOUT_SECONDS = 0.25 OUTPUT_MODE = 0x1200 DIAGNOSTIC_BASE = 0x2100 DIAGNOSTIC_WORDS = 0x1A DIAGNOSTIC_CONTROL = 0x3100 OUTPUT_PULSE_DIR = 0 OUTPUT_AB = 1 DIAGNOSTIC_CLEAR = 0x0001 DIAG_MONITORING = 0x0001 DIAG_COUNT_CHECKED = 0x0002 DIAG_COUNT_PASS = 0x0004 DIAG_FREQUENCY_CHECKED = 0x0008 DIAG_FREQUENCY_PASS = 0x0010 DIAG_CURVE_CHECKED = 0x0020 DIAG_CURVE_PASS = 0x0040 DIAG_FAULT_LATCHED = 0x0080 DIAG_COMPLETE_PASS = ( DIAG_COUNT_CHECKED | DIAG_COUNT_PASS | DIAG_FREQUENCY_CHECKED | DIAG_FREQUENCY_PASS | DIAG_CURVE_CHECKED | DIAG_CURVE_PASS ) COMMAND_START = 0x0001 COMMAND_STOP = 0x0002 COMMAND_CLEAR = 0x0004 STATUS_IDLE = 1 STATUS_ACCELERATING = 2 STATUS_RUNNING = 3 STATUS_DECELERATING = 4 STATUS_WAITING = 5 STATUS_COMPLETED = 7 STATUS_STOPPED = 8 STATUS_ERROR = 9 ERROR_NONE = 0 WAIT_TIME = 0 WAIT_SIGNAL = 1 ACT_TIME = 2 EXT_SIGNAL = 3 EXT_OR_COMPLETE = 4 SEND_COMPLETE = 0 SEND_SUBSEQUENT = 1 DEFAULT_STLINK_CLI = r"F:\ST-LINK Utility\ST-LINK_CLI.exe" DEFAULT_IAR_MAP = r"EWARM\Modbus\List\Modbus.map" IRQ_CYCLE_BUDGET_100KHZ = 1680 GPIOE_CLOCK_BIT = 0x42470610 GPIOI_CLOCK_BIT = 0x42470620 GPIOE_PIN6_OUTPUT_BIT = 0x42420298 GPIOI_PIN8_OUTPUT_BIT = 0x424402A0 GPIOE_PIN6_MODE_LOW_BIT = 0x42420030 GPIOE_PIN6_MODE_HIGH_BIT = 0x42420034 GPIOI_PIN8_MODE_LOW_BIT = 0x42440040 GPIOI_PIN8_MODE_HIGH_BIT = 0x42440044 GPIOE_PIN6_PULL_LOW_BIT = 0x424201B0 GPIOE_PIN6_PULL_HIGH_BIT = 0x424201B4 GPIOI_PIN8_PULL_LOW_BIT = 0x424401C0 GPIOI_PIN8_PULL_HIGH_BIT = 0x424401C4 X4_INPUT_BIT = 0x42408214 X5_INPUT_BIT = 0x42430230 OUTPUT_OFF = 1 OUTPUT_ON = 0 EX_ILLEGAL_FUNCTION = 0x01 EX_ILLEGAL_ADDRESS = 0x02 EX_ILLEGAL_VALUE = 0x03 EX_DEVICE_FAILURE = 0x04 EX_DEVICE_BUSY = 0x06 ACTIVE_STATES = { STATUS_ACCELERATING, STATUS_RUNNING, STATUS_DECELERATING, STATUS_WAITING, } TERMINAL_STATES = {STATUS_COMPLETED, STATUS_STOPPED, STATUS_ERROR} class TestFailure(RuntimeError): pass class RtuTimeout(TestFailure): pass class ModbusException(RuntimeError): def __init__(self, function, code): super().__init__( "Modbus exception: function=0x%02X code=0x%02X" % (function, code) ) self.function = function self.code = code def parse_stlink_word(output, address): return parse_stlink_words(output, address, 1)[0] def parse_iar_map_symbol_address(map_text, symbol): pattern = ( r"(?m)^" + re.escape(symbol) + r"\s+0x([0-9a-fA-F]+)'([0-9a-fA-F]+)\s+" ) match = re.search(pattern, map_text) require(match is not None, "IAR map does not contain %s" % symbol) return (int(match.group(1), 16) << 16) | int(match.group(2), 16) def parse_stlink_words(output, address, count): require(count > 0, "ST-LINK parse count must be positive") observed = {} pattern = re.compile( r"(?im)^\s*0x([0-9a-f]{8})\s*:\s*" r"((?:[0-9a-f]{8}(?:\s+|$))+)", ) for match in pattern.finditer(output): line_address = int(match.group(1), 16) for offset, value in enumerate(re.findall( r"[0-9a-f]{8}", match.group(2), flags=re.IGNORECASE)): observed[line_address + offset * 4] = int(value, 16) words = [] for offset in range(count): word_address = address + offset * 4 require(word_address in observed, "ST-LINK output did not contain address 0x%08X" % word_address) words.append(observed[word_address]) return words class X45Fixture: """Drive Y4/Y5 through SWD without adding product test registers.""" def __init__(self, executable, probe_id, timeout): self.executable = executable self.probe_id = probe_id self.timeout = timeout def _run(self, *arguments): command_line = [ self.executable, "-c", "ID=%d" % self.probe_id, "SWD", "HOTPLUG", ] + list(arguments) + ["-Q", "-NoPrompt"] try: result = subprocess.run( command_line, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True, errors="replace", timeout=self.timeout, check=False, ) except (OSError, subprocess.SubprocessError) as error: raise TestFailure("ST-LINK command failed: %s" % error) from error require( result.returncode == 0, "ST-LINK command returned %d:\n%s" % (result.returncode, result.stdout.strip()), ) return result.stdout def read_words(self, address, count): require(count > 0, "ST-LINK read count must be positive") output = self._run("-r32", hex(address), str(count * 4)) return parse_stlink_words(output, address, count) def write_words(self, address, values): require(values, "ST-LINK write values must not be empty") arguments = [] for offset, value in enumerate(values): arguments.extend(( "-w32", hex(address + offset * 4), hex(value & 0xFFFFFFFF), )) self._run(*arguments) def prepare(self): self._run( "-w32", hex(GPIOE_CLOCK_BIT), "1", "-w32", hex(GPIOI_CLOCK_BIT), "1", "-w32", hex(GPIOI_PIN8_OUTPUT_BIT), str(OUTPUT_OFF), "-w32", hex(GPIOE_PIN6_OUTPUT_BIT), str(OUTPUT_OFF), "-w32", hex(GPIOI_PIN8_MODE_HIGH_BIT), "0", "-w32", hex(GPIOI_PIN8_MODE_LOW_BIT), "1", "-w32", hex(GPIOE_PIN6_MODE_HIGH_BIT), "0", "-w32", hex(GPIOE_PIN6_MODE_LOW_BIT), "1", ) time.sleep(0.050) def drive(self, input_selection, active): require(input_selection in (0, 1), "input selection must be X4 or X5") if input_selection == 0: address = GPIOI_PIN8_OUTPUT_BIT else: address = GPIOE_PIN6_OUTPUT_BIT value = OUTPUT_ON if active else OUTPUT_OFF self._run("-w32", hex(address), str(value)) time.sleep(0.050) def read(self, input_selection): require(input_selection in (0, 1), "input selection must be X4 or X5") address = X4_INPUT_BIT if input_selection == 0 else X5_INPUT_BIT output = self._run("-r32", hex(address), "1") return parse_stlink_word(output, address) & 1 def all_off(self): self._run( "-w32", hex(GPIOI_PIN8_OUTPUT_BIT), str(OUTPUT_OFF), "-w32", hex(GPIOE_PIN6_OUTPUT_BIT), str(OUTPUT_OFF), ) time.sleep(0.050) def release(self): self._run( "-w32", hex(GPIOI_PIN8_OUTPUT_BIT), str(OUTPUT_OFF), "-w32", hex(GPIOE_PIN6_OUTPUT_BIT), str(OUTPUT_OFF), "-w32", hex(GPIOI_PIN8_MODE_LOW_BIT), "0", "-w32", hex(GPIOI_PIN8_MODE_HIGH_BIT), "0", "-w32", hex(GPIOI_PIN8_PULL_LOW_BIT), "0", "-w32", hex(GPIOI_PIN8_PULL_HIGH_BIT), "0", "-w32", hex(GPIOE_PIN6_MODE_LOW_BIT), "0", "-w32", hex(GPIOE_PIN6_MODE_HIGH_BIT), "0", "-w32", hex(GPIOE_PIN6_PULL_LOW_BIT), "0", "-w32", hex(GPIOE_PIN6_PULL_HIGH_BIT), "0", ) def require(condition, message): if not condition: raise TestFailure(message) def crc16(data): value = 0xFFFF for byte in data: value ^= byte for _ in range(8): if value & 1: value = (value >> 1) ^ 0xA001 else: value >>= 1 return value def append_crc(payload): checksum = crc16(payload) return bytes(payload) + bytes((checksum & 0xFF, checksum >> 8)) def split_u32(value): value &= 0xFFFFFFFF return [value & 0xFFFF, value >> 16] def split_i32(value): require(-(1 << 31) <= value < (1 << 31), "signed value is not int32") return split_u32(value) def join_u32(low_word, high_word): return low_word | (high_word << 16) def join_i32(low_word, high_word): value = join_u32(low_word, high_word) return value - (1 << 32) if value & 0x80000000 else value def frequency_matches(actual_hz, requested_hz): tolerance_hz = max(1, requested_hz // 1000) return abs(actual_hz - requested_hz) <= tolerance_hz @dataclasses.dataclass class Status: position: int frequency_hz: int state: int segment: int error: int @dataclasses.dataclass class Diagnostic: flags: int reason: int segment: int output_mode: int direction_positive: bool expected_pulses: int actual_pulses: int count_error: int requested_hz: int expected_timer_hz: int active_timer_hz: int request_error_hz: int sample_count: int mismatch_count: int maximum_frequency_error_hz: int first_mismatch_sample: int @dataclasses.dataclass class PersistenceLiveStats: attempts: int successful: int timeouts: int maximum_attempt_seconds: float elapsed_seconds: float @dataclasses.dataclass(frozen=True) class IrqCycleStats: count: tuple last_cycles: tuple maximum_cycles: tuple @dataclasses.dataclass(frozen=True) class ProducerCycleStats: item_count: int total_cycles: int maximum_item_cycles: int @property def average_cycles(self): require(self.item_count > 0, "producer timing did not record any generated item") return self.total_cycles / float(self.item_count) @dataclasses.dataclass(frozen=True) class FinalArmCycleStats: queue_count: tuple job_last_cycles: tuple job_maximum_cycles: tuple queue_to_stop_last_cycles: tuple queue_to_stop_maximum_cycles: tuple class TimingFixture: IRQ_SYMBOLS = ( "PlsrIrqCount", "PlsrIrqLastCycles", "PlsrIrqMaxCycles", ) PRODUCER_SYMBOLS = ( "PlsrProfileProducerItemCount", "PlsrProfileProducerTotalCycles", "PlsrProfileProducerMaxItemCycles", ) FINAL_ARM_SYMBOLS = ( "PlsrFinalArmQueueCount", "PlsrFinalArmJobLastCycles", "PlsrFinalArmJobMaxCycles", "PlsrFinalArmQueueToStopLastCycles", "PlsrFinalArmQueueToStopMaxCycles", ) def __init__(self, probe, map_path): self.probe = probe try: map_text = Path(map_path).read_text( encoding="utf-8", errors="replace") except OSError as error: raise TestFailure("could not read IAR map %s: %s" % (map_path, error)) from error self.addresses = { symbol: parse_iar_map_symbol_address(map_text, symbol) for symbol in (self.IRQ_SYMBOLS + self.PRODUCER_SYMBOLS + self.FINAL_ARM_SYMBOLS) } def reset(self): for symbol in self.IRQ_SYMBOLS: self.probe.write_words(self.addresses[symbol], [0] * 4) for symbol in self.PRODUCER_SYMBOLS: self.probe.write_words(self.addresses[symbol], [0]) for symbol in self.FINAL_ARM_SYMBOLS: self.probe.write_words(self.addresses[symbol], [0] * 4) def read(self): irq = IrqCycleStats( count=tuple(self.probe.read_words( self.addresses["PlsrIrqCount"], 4)), last_cycles=tuple(self.probe.read_words( self.addresses["PlsrIrqLastCycles"], 4)), maximum_cycles=tuple(self.probe.read_words( self.addresses["PlsrIrqMaxCycles"], 4)), ) producer = ProducerCycleStats( item_count=self.probe.read_words( self.addresses["PlsrProfileProducerItemCount"], 1)[0], total_cycles=self.probe.read_words( self.addresses["PlsrProfileProducerTotalCycles"], 1)[0], maximum_item_cycles=self.probe.read_words( self.addresses["PlsrProfileProducerMaxItemCycles"], 1)[0], ) final_arm = FinalArmCycleStats( queue_count=tuple(self.probe.read_words( self.addresses["PlsrFinalArmQueueCount"], 4)), job_last_cycles=tuple(self.probe.read_words( self.addresses["PlsrFinalArmJobLastCycles"], 4)), job_maximum_cycles=tuple(self.probe.read_words( self.addresses["PlsrFinalArmJobMaxCycles"], 4)), queue_to_stop_last_cycles=tuple(self.probe.read_words( self.addresses["PlsrFinalArmQueueToStopLastCycles"], 4)), queue_to_stop_maximum_cycles=tuple(self.probe.read_words( self.addresses["PlsrFinalArmQueueToStopMaxCycles"], 4)), ) return irq, producer, final_arm class RtuClient: def __init__(self, port, baud, slave, timeout): self.port_name = port self.baud = baud self.slave = slave self.timeout = timeout self.character_seconds = 11.0 / baud if baud > 19200: self.frame_gap_seconds = 0.00175 else: self.frame_gap_seconds = 3.5 * self.character_seconds self.serial_port = None self.last_request_finished = 0.0 def __enter__(self): self.serial_port = serial.Serial( port=self.port_name, baudrate=self.baud, bytesize=serial.EIGHTBITS, parity=serial.PARITY_EVEN, stopbits=serial.STOPBITS_ONE, timeout=0, write_timeout=1, ) self.serial_port.reset_input_buffer() self.serial_port.reset_output_buffer() return self def __exit__(self, exc_type, exc_value, traceback): if self.serial_port is not None: self.serial_port.close() self.serial_port = None def _exchange(self, pdu): request = append_crc(bytes((self.slave,)) + bytes(pdu)) now = time.monotonic() delay = self.frame_gap_seconds - (now - self.last_request_finished) if delay > 0: time.sleep(delay) self.serial_port.reset_input_buffer() self.serial_port.write(request) self.serial_port.flush() response = bytearray() deadline = time.monotonic() + self.timeout expected_length = None while time.monotonic() < deadline: waiting = self.serial_port.in_waiting if waiting: response.extend(self.serial_port.read(waiting)) if len(response) >= 2 and response[1] == (pdu[0] | 0x80): expected_length = 5 elif pdu[0] == 0x03 and len(response) >= 3: expected_length = response[2] + 5 elif pdu[0] in (0x06, 0x10): expected_length = 8 if expected_length is not None and len(response) >= expected_length: break if not waiting: time.sleep(0.0005) waiting = self.serial_port.in_waiting if waiting: response.extend(self.serial_port.read(waiting)) if len(response) >= 2 and response[1] == (pdu[0] | 0x80): expected_length = 5 elif pdu[0] == 0x03 and len(response) >= 3: expected_length = response[2] + 5 elif pdu[0] in (0x06, 0x10): expected_length = 8 self.last_request_finished = time.monotonic() if not response: raise RtuTimeout("no Modbus response to %s" % request.hex(" ")) if expected_length is None: raise RtuTimeout( "Modbus response header timed out: %s" % response.hex(" ") ) if len(response) < expected_length: raise RtuTimeout( "Modbus response timed out after %d/%d bytes: %s" % (len(response), expected_length, response.hex(" ")) ) require(len(response) == expected_length, "oversized Modbus response: %s" % response.hex(" ")) require(len(response) >= 5, "short Modbus response: %s" % response.hex(" ")) received_crc = response[-2] | (response[-1] << 8) require( received_crc == crc16(response[:-2]), "bad response CRC: %s" % response.hex(" "), ) require(response[0] == self.slave, "response slave address mismatch") requested_function = pdu[0] if response[1] == (requested_function | 0x80): require(len(response) == 5, "invalid exception response length") raise ModbusException(requested_function, response[2]) require(response[1] == requested_function, "response function mismatch") return bytes(response) def read_holding(self, address, quantity): require(1 <= quantity <= 125, "FC03 quantity must be 1..125") pdu = bytes( ( 0x03, address >> 8, address & 0xFF, quantity >> 8, quantity & 0xFF, ) ) response = self._exchange(pdu) byte_count = response[2] require(byte_count == quantity * 2, "FC03 byte count mismatch") require(len(response) == byte_count + 5, "FC03 response length mismatch") return [ (response[3 + index * 2] << 8) | response[4 + index * 2] for index in range(quantity) ] def write_single(self, address, value): value &= 0xFFFF pdu = bytes( ( 0x06, address >> 8, address & 0xFF, value >> 8, value & 0xFF, ) ) response = self._exchange(pdu) require(response[:6] == bytes((self.slave,)) + pdu, "FC06 echo mismatch") def write_multiple(self, address, values): require(1 <= len(values) <= 123, "FC10 quantity must be 1..123") data = bytearray() for value in values: value &= 0xFFFF data.extend((value >> 8, value & 0xFF)) quantity = len(values) pdu = bytes( ( 0x10, address >> 8, address & 0xFF, quantity >> 8, quantity & 0xFF, len(data), ) ) + bytes(data) response = self._exchange(pdu) expected = bytes( ( self.slave, 0x10, address >> 8, address & 0xFF, quantity >> 8, quantity & 0xFF, ) ) require(response[:6] == expected, "FC10 acknowledgement mismatch") def read_status(client): words = client.read_holding(STATUS_BASE, STATUS_WORDS) return Status( position=join_i32(words[0], words[1]), frequency_hz=join_u32(words[2], words[3]), state=words[4], segment=words[5], error=words[6], ) def parse_diagnostic(words): require(len(words) == DIAGNOSTIC_WORDS, "diagnostic register count must be %d" % DIAGNOSTIC_WORDS) mode_direction = words[3] return Diagnostic( flags=words[0], reason=words[1], segment=words[2], output_mode=mode_direction & 0xFF, direction_positive=bool(mode_direction & 0x0100), expected_pulses=join_u32(words[4], words[5]), actual_pulses=join_u32(words[6], words[7]), count_error=join_i32(words[8], words[9]), requested_hz=join_u32(words[10], words[11]), expected_timer_hz=join_u32(words[12], words[13]), active_timer_hz=join_u32(words[14], words[15]), request_error_hz=join_i32(words[16], words[17]), sample_count=join_u32(words[18], words[19]), mismatch_count=join_u32(words[20], words[21]), maximum_frequency_error_hz=join_u32(words[22], words[23]), first_mismatch_sample=join_u32(words[24], words[25]), ) def read_diagnostic(client): return parse_diagnostic( client.read_holding(DIAGNOSTIC_BASE, DIAGNOSTIC_WORDS) ) def clear_diagnostic(client): client.write_single(DIAGNOSTIC_CONTROL, DIAGNOSTIC_CLEAR) diagnostic = read_diagnostic(client) require(diagnostic.flags == 0, "diagnostic CLEAR did not clear flags") require(diagnostic.reason == 0, "diagnostic CLEAR did not clear reason") require(diagnostic.first_mismatch_sample == 0xFFFFFFFF, "diagnostic CLEAR first mismatch sentinel is invalid") def require_diagnostic_pass(diagnostic, expected_pulses, expected_mode, direction_positive, label, expected_segment=1): require((diagnostic.flags & DIAG_COMPLETE_PASS) == DIAG_COMPLETE_PASS, "%s diagnostic checks incomplete or failed: flags=0x%04X" % (label, diagnostic.flags)) require(not (diagnostic.flags & DIAG_MONITORING), "%s diagnostic remained active" % label) require(not (diagnostic.flags & DIAG_FAULT_LATCHED), "%s diagnostic latched fault %d" % (label, diagnostic.reason)) require(diagnostic.reason == 0, "%s diagnostic reason is %d" % (label, diagnostic.reason)) require(diagnostic.segment == expected_segment, "%s diagnostic segment expected %d, got %d" % (label, expected_segment, diagnostic.segment)) require(diagnostic.output_mode == expected_mode, "%s diagnostic output mode expected %d, got %d" % (label, expected_mode, diagnostic.output_mode)) require(diagnostic.direction_positive == direction_positive, "%s diagnostic direction mismatch" % label) require(diagnostic.expected_pulses == expected_pulses, "%s diagnostic expected pulse count %d, got %d" % (label, expected_pulses, diagnostic.expected_pulses)) require(diagnostic.actual_pulses == expected_pulses, "%s diagnostic actual pulse count %d, got %d" % (label, expected_pulses, diagnostic.actual_pulses)) require(diagnostic.count_error == 0, "%s diagnostic count error %d" % (label, diagnostic.count_error)) require(diagnostic.requested_hz > 0, "%s diagnostic requested frequency is zero" % label) require(diagnostic.expected_timer_hz > 0, "%s diagnostic expected timer frequency is zero" % label) require(diagnostic.active_timer_hz == diagnostic.expected_timer_hz, "%s diagnostic active/expected frequency %d/%d" % (label, diagnostic.active_timer_hz, diagnostic.expected_timer_hz)) require(diagnostic.sample_count > 0, "%s diagnostic did not capture curve samples" % label) require(diagnostic.mismatch_count == 0, "%s diagnostic curve mismatch count %d" % (label, diagnostic.mismatch_count)) require(diagnostic.maximum_frequency_error_hz == 0, "%s diagnostic maximum active frequency error %d Hz" % (label, diagnostic.maximum_frequency_error_hz)) require(diagnostic.first_mismatch_sample == 0xFFFFFFFF, "%s diagnostic first mismatch sample is %d" % (label, diagnostic.first_mismatch_sample)) def wait_for_status(client, predicate, timeout, description): deadline = time.monotonic() + timeout last_status = None while time.monotonic() < deadline: last_status = read_status(client) if last_status.state == STATUS_ERROR: raise TestFailure( "%s entered ERROR (code=%d)" % (description, last_status.error) ) if predicate(last_status): return last_status time.sleep(0.005) raise TestFailure("timeout waiting for %s; last=%r" % (description, last_status)) def wait_terminal(client, timeout): status = wait_for_status( client, lambda item: item.state in TERMINAL_STATES, timeout, "terminal state", ) require(status.state != STATUS_ERROR, "motion ended in ERROR %d" % status.error) return status def expect_exception(operation, expected_code, label): try: operation() except ModbusException as error: require( error.code == expected_code, "%s expected exception 0x%02X, got 0x%02X" % (label, expected_code, error.code), ) return raise TestFailure("%s did not return a Modbus exception" % label) def expect_test_failure(operation, label): try: operation() except TestFailure: return raise TestFailure("%s unexpectedly passed" % label) def make_common( position_mode=0, segment_count=1, start_segment=1, send_mode=SEND_COMPLETE, curve_mode=0, default_hz=1000, start_hz=500, stop_hz=100, acceleration_ms=0, deceleration_ms=0, ): words = [0] * COMMON_WORDS words[0x00] = 0 words[0x01] = 0 words[0x02] = 0 words[0x03] = 0 words[0x04] = send_mode words[0x05] = 0 words[0x06] = 0 words[0x07] = curve_mode words[0x08] = position_mode words[0x09] = segment_count words[0x0A] = start_segment words[0x0B:0x0D] = split_u32(default_hz) words[0x0D:0x0F] = split_u32(start_hz) words[0x0F] = 0 words[0x10:0x12] = split_u32(stop_hz) words[0x12] = acceleration_ms words[0x13] = deceleration_ms return words def make_segment( frequency_hz, pulses, wait_type=EXT_OR_COMPLETE, wait_time_ms=0, act_time_ms=0, jump_segment=0, ): return ( split_u32(frequency_hz) + split_i32(pulses) + [wait_type, wait_time_ms, act_time_ms, jump_segment] ) def configure(client, common, segment_1, segment_2=None): client.write_multiple(CONFIG_BASE, common) client.write_multiple(SEGMENT_1_BASE, segment_1) if segment_2 is not None: client.write_multiple(SEGMENT_2_BASE, segment_2) def configure_segments(client, common, segments): require(1 <= len(segments) <= 10, "segment configuration count must be 1..10") client.write_multiple(CONFIG_BASE, common) for index, segment in enumerate(segments): client.write_multiple(SEGMENT_1_BASE + index * 0x10, segment) def configure_pulse_dir(client, common, segment_1, segment_2=None): client.write_single(OUTPUT_MODE, OUTPUT_PULSE_DIR) require(client.read_holding(OUTPUT_MODE, 1) == [OUTPUT_PULSE_DIR], "PULSE/DIR output mode readback failed") configure(client, common, segment_1, segment_2) clear_diagnostic(client) def restore_default_configuration(client): client.write_single(OUTPUT_MODE, OUTPUT_PULSE_DIR) common = make_common( default_hz=1000, start_hz=100, stop_hz=100, acceleration_ms=100, deceleration_ms=100, ) common[0x05] = 10 client.write_multiple(CONFIG_BASE, common) for index in range(10): base = SEGMENT_1_BASE + index * 0x10 pulses = 1000 if index == 0 else 0 client.write_multiple(base, make_segment(1000, pulses)) def command(client, value): client.write_single(CONTROL, value) def clear_position(client): command(client, COMMAND_CLEAR) status = read_status(client) require(status.state == STATUS_IDLE, "CLEAR did not enter IDLE") require(status.position == 0, "CLEAR did not zero logical position") def run_case(name, function): started = time.monotonic() print("RUN %s" % name) function() print("PASS %s (%.3fs)" % (name, time.monotonic() - started)) def test_fixed_map_and_exceptions(client): common = client.read_holding(CONFIG_BASE, COMMON_WORDS) segment = client.read_holding(SEGMENT_1_BASE, SEGMENT_WORDS) status = read_status(client) control = client.read_holding(CONTROL, 1) output_mode = client.read_holding(OUTPUT_MODE, 1) diagnostic = read_diagnostic(client) diagnostic_control = client.read_holding(DIAGNOSTIC_CONTROL, 1) require(len(common) == COMMON_WORDS, "common map read failed") require(len(segment) == SEGMENT_WORDS, "segment map read failed") require(0 <= status.state <= STATUS_ERROR, "status enum is out of range") require(control == [0], "control register must read as zero") require(output_mode[0] in (OUTPUT_PULSE_DIR, OUTPUT_AB), "output mode is out of range") require(diagnostic.output_mode in (OUTPUT_PULSE_DIR, OUTPUT_AB), "diagnostic output mode is out of range") require(diagnostic_control == [0], "diagnostic control register must read as zero") require(client.read_holding(0x100F, 1) == [0], "reserved word is not zero") require(client.read_holding(0x1108, 1) == [0], "segment padding is not zero") expect_exception( lambda: client.read_holding(0x0000, 1), EX_ILLEGAL_ADDRESS, "FC03 address 0", ) expect_exception( lambda: client._exchange(bytes((0x01, 0x00, 0x00, 0x00, 0x01))), EX_ILLEGAL_FUNCTION, "FC01 unsupported function", ) expect_exception( lambda: client.read_holding(0x1197, 2), EX_ILLEGAL_ADDRESS, "range crossing 0x1197", ) expect_exception( lambda: client.write_single(0x100F, 1), EX_ILLEGAL_VALUE, "nonzero reserved write", ) expect_exception( lambda: client.write_single(0x100B, 1000), EX_ILLEGAL_ADDRESS, "single half of a DWORD", ) expect_exception( lambda: client.write_single(STATUS_BASE, 0), EX_ILLEGAL_ADDRESS, "status write", ) expect_exception( lambda: client.write_single(CONTROL, 3), EX_ILLEGAL_VALUE, "combined control command", ) expect_exception( lambda: client.read_holding(0x11FF, 2), EX_ILLEGAL_ADDRESS, "range crossing into output mode", ) expect_exception( lambda: client.read_holding(0x2119, 2), EX_ILLEGAL_ADDRESS, "range crossing diagnostic end", ) expect_exception( lambda: client.write_single(DIAGNOSTIC_BASE, 0), EX_ILLEGAL_ADDRESS, "diagnostic write", ) expect_exception( lambda: client.write_single(DIAGNOSTIC_CONTROL, 2), EX_ILLEGAL_VALUE, "invalid diagnostic command", ) expect_exception( lambda: client.write_single(OUTPUT_MODE, 2), EX_ILLEGAL_VALUE, "invalid output mode", ) def test_positive_negative_and_absolute_zero(client): clear_position(client) common = make_common(position_mode=0, start_hz=500) configure_pulse_dir(client, common, make_segment(500, 25)) command(client, COMMAND_START) status = wait_terminal(client, 3.0) require(status.state == STATUS_COMPLETED, "positive move did not complete") require(status.position == 25, "positive accumulation expected 25") require_diagnostic_pass( read_diagnostic(client), 25, OUTPUT_PULSE_DIR, True, "positive move", ) configure_pulse_dir(client, common, make_segment(500, -10)) command(client, COMMAND_START) status = wait_terminal(client, 3.0) require(status.state == STATUS_COMPLETED, "negative move did not complete") require(status.position == 15, "negative accumulation expected 15") require_diagnostic_pass( read_diagnostic(client), 10, OUTPUT_PULSE_DIR, False, "negative move", ) absolute = make_common(position_mode=1, start_hz=500) configure_pulse_dir(client, absolute, make_segment(500, 15)) before = read_status(client) command(client, COMMAND_START) status = wait_terminal(client, 1.0) require(status.state == STATUS_COMPLETED, "absolute zero move did not complete") require(status.position == before.position, "absolute zero changed position") def test_ten_segments_in_sequence(client): clear_position(client) common = make_common( segment_count=10, default_hz=1000, start_hz=1000, stop_hz=1000, ) pulse_counts = [100 + index for index in range(10)] segments = [make_segment(1000, pulses) for pulses in pulse_counts] client.write_single(OUTPUT_MODE, OUTPUT_PULSE_DIR) configure_segments(client, common, segments) clear_diagnostic(client) require(client.read_holding(CONFIG_BASE + 0x09, 2) == [10, 1], "ten-segment count/start readback mismatch") command(client, COMMAND_START) for segment_number in range(1, 11): wait_for_status( client, lambda item, expected=segment_number: item.segment == expected and item.state in ACTIVE_STATES, 2.0, "sequential segment %d" % segment_number, ) status = wait_terminal(client, 3.0) expected_position = sum(pulse_counts) require(status.state == STATUS_COMPLETED, "ten-segment sequence did not complete") require(status.position == expected_position, "ten-segment sequence position expected %d, got %d" % (expected_position, status.position)) require_diagnostic_pass( read_diagnostic(client), pulse_counts[-1], OUTPUT_PULSE_DIR, True, "ten-segment sequence final segment", expected_segment=10, ) def test_start_segment_ten(client): clear_position(client) common = make_common( segment_count=10, start_segment=10, default_hz=1000, start_hz=1000, stop_hz=1000, ) segment_ten_pulses = 200 segments = [make_segment(1000, 11 + index) for index in range(9)] segments.append(make_segment(1000, segment_ten_pulses)) client.write_single(OUTPUT_MODE, OUTPUT_PULSE_DIR) configure_segments(client, common, segments) clear_diagnostic(client) require(client.read_holding(CONFIG_BASE + 0x09, 2) == [10, 10], "startSegment=10 readback mismatch") command(client, COMMAND_START) wait_for_status( client, lambda item: item.segment == 10 and item.state in ACTIVE_STATES, 2.0, "startSegment=10 active segment", ) status = wait_terminal(client, 3.0) require(status.state == STATUS_COMPLETED, "startSegment=10 move did not complete") require(status.position == segment_ten_pulses, "startSegment=10 executed an earlier segment: position=%d" % status.position) require_diagnostic_pass( read_diagnostic(client), segment_ten_pulses, OUTPUT_PULSE_DIR, True, "startSegment=10", expected_segment=10, ) def test_direction_output_and_polarity_matrix(client): pulse_count = 25 direction_delay_ms = 500 for direction_output in range(4): for negative_logic in range(2): for direction in (1, -1): label = "Y%d DIR %s %s logic" % ( direction_output + 12, "positive" if direction > 0 else "negative", "negative" if negative_logic else "positive", ) clear_position(client) common = make_common( default_hz=500, start_hz=500, stop_hz=500, ) common[0x01] = direction_output common[0x05] = direction_delay_ms common[0x06] = negative_logic target = direction * pulse_count segment = make_segment(500, target) configure_pulse_dir(client, common, segment) common_readback = client.read_holding(CONFIG_BASE, COMMON_WORDS) segment_readback = client.read_holding( SEGMENT_1_BASE, SEGMENT_WORDS) require(common_readback[0x01] == direction_output, "%s direction output readback mismatch" % label) require(common_readback[0x05] == direction_delay_ms, "%s direction delay readback mismatch" % label) require(common_readback[0x06] == negative_logic, "%s direction polarity readback mismatch" % label) require(join_i32(segment_readback[2], segment_readback[3]) == target, "%s signed target readback mismatch" % label) command(client, COMMAND_START) before_pulses = wait_for_status( client, lambda item: item.state == STATUS_ACCELERATING and item.segment == 1, 0.25, "%s direction-delay state" % label, ) require(before_pulses.segment == 1, "%s direction delay selected wrong segment" % label) require(before_pulses.position == 0, "%s counted a logical pulse during direction delay" % label) require(before_pulses.frequency_hz == 0, "%s reported a pulse frequency during direction delay" % label) status = wait_terminal(client, 3.0) require(status.state == STATUS_COMPLETED, "%s did not complete" % label) require(status.position == target, "%s position expected %d, got %d" % (label, target, status.position)) require_diagnostic_pass( read_diagnostic(client), pulse_count, OUTPUT_PULSE_DIR, direction > 0, label, ) def test_all_outputs_at_100khz(client): for output in range(4): clear_position(client) common = make_common( default_hz=100000, start_hz=100000, stop_hz=100000, ) common[0x00] = output common[0x01] = output configure_pulse_dir(client, common, make_segment(100000, 1000)) command(client, COMMAND_START) status = wait_terminal(client, 2.0) require(status.state == STATUS_COMPLETED, "output %d did not complete" % output) require(status.position == 1000, "output %d count expected 1000, got %d" % (output, status.position)) require_diagnostic_pass( read_diagnostic(client), 1000, OUTPUT_PULSE_DIR, True, "output %d 100 kHz" % output, ) def test_short_final_profiles(client): for curve_mode in range(3): clear_position(client) common = make_common( curve_mode=curve_mode, default_hz=100000, start_hz=100000, stop_hz=100, deceleration_ms=1000, ) configure_pulse_dir(client, common, make_segment(100000, 10)) command(client, COMMAND_START) status = wait_terminal(client, 2.0) require(status.state == STATUS_COMPLETED, "short curve %d did not complete" % curve_mode) require(status.error == ERROR_NONE, "short curve %d reported error %d" % (curve_mode, status.error)) require(status.position == 10, "short curve %d count expected 10, got %d" % (curve_mode, status.position)) require_diagnostic_pass( read_diagnostic(client), 10, OUTPUT_PULSE_DIR, True, "short curve %d" % curve_mode, ) def test_long_millisecond_ramp(client): pulses = 65535 label = "65,535-pulse 1 ms ramp" clear_position(client) common = make_common( curve_mode=2, default_hz=100000, start_hz=1000, stop_hz=1000, acceleration_ms=100, deceleration_ms=100, ) configure_pulse_dir(client, common, make_segment(100000, pulses)) command(client, COMMAND_START) status = wait_terminal(client, 5.0) require(status.state == STATUS_COMPLETED, "%s did not complete" % label) require(status.position == pulses, "%s expected %d pulses, got %d" % (label, pulses, status.position)) diagnostic = read_diagnostic(client) require(frequency_matches(diagnostic.requested_hz, 1000), "%s final requested frequency expected 1000 Hz, got %d Hz" % (label, diagnostic.requested_hz)) require_diagnostic_pass( diagnostic, pulses, OUTPUT_PULSE_DIR, True, label, ) def require_irq_cycle_budget(stats, expected_outputs, label): for output in expected_outputs: require(stats.count[output] > 0, "%s output Y%d recorded no IRQ samples" % (label, output)) maximum = stats.maximum_cycles[output] require(maximum > 0, "%s output Y%d recorded a zero IRQ maximum" % (label, output)) require(maximum < IRQ_CYCLE_BUDGET_100KHZ, "%s output Y%d IRQ maximum %d cycles exceeds < %d budget" % (label, output, maximum, IRQ_CYCLE_BUDGET_100KHZ)) print("INFO %s Y%d irq_count=%d last=%d max=%d cycles" % (label, output, stats.count[output], stats.last_cycles[output], maximum)) def require_producer_cycle_budget(stats, label): average = stats.average_cycles require(stats.total_cycles >= stats.maximum_item_cycles, "%s producer timing counters are inconsistent" % label) require(average < IRQ_CYCLE_BUDGET_100KHZ, "%s producer average %.2f cycles/item exceeds < %d budget" % (label, average, IRQ_CYCLE_BUDGET_100KHZ)) print("INFO %s producer_items=%d total=%d average=%.2f max=%d cycles" % (label, stats.item_count, stats.total_cycles, average, stats.maximum_item_cycles)) def require_no_producer_activity(stats, label): require(stats.item_count == 0, "%s unexpectedly generated %d short-profile items" % (label, stats.item_count)) require(stats.total_cycles == 0, "%s unexpectedly recorded %d producer cycles" % (label, stats.total_cycles)) require(stats.maximum_item_cycles == 0, "%s unexpectedly recorded a %d-cycle producer maximum" % (label, stats.maximum_item_cycles)) print("INFO %s producer_items=0 total=0 max=0 cycles" % label) def require_final_arm_cycle_budget(stats, expected_outputs, label): for output in expected_outputs: require(stats.queue_count[output] > 0, "%s output Y%d queued no final-arm jobs" % (label, output)) job_maximum = stats.job_maximum_cycles[output] latency_maximum = stats.queue_to_stop_maximum_cycles[output] require(job_maximum > 0, "%s output Y%d recorded a zero final-arm job maximum" % (label, output)) require(latency_maximum > 0, "%s output Y%d recorded a zero queue-to-stop maximum" % (label, output)) require(job_maximum < IRQ_CYCLE_BUDGET_100KHZ, "%s output Y%d final-arm job maximum %d cycles exceeds < %d budget" % (label, output, job_maximum, IRQ_CYCLE_BUDGET_100KHZ)) require(latency_maximum < IRQ_CYCLE_BUDGET_100KHZ, "%s output Y%d queue-to-stop maximum %d cycles exceeds < %d budget" % (label, output, latency_maximum, IRQ_CYCLE_BUDGET_100KHZ)) print( "INFO %s Y%d final_arm_count=%d job_last=%d job_max=%d " "queue_to_stop_last=%d queue_to_stop_max=%d cycles" % (label, output, stats.queue_count[output], stats.job_last_cycles[output], job_maximum, stats.queue_to_stop_last_cycles[output], latency_maximum)) def test_ab_100khz_timing_paths(client): pulses_per_case = 100000 for output, direction in ((0, 1), (2, -1)): target = direction * pulses_per_case label = "AB Y%d/Y%d %s timing" % ( output, output + 1, "positive" if direction > 0 else "negative", ) clear_position(client) configure_ab( client, output, make_common( default_hz=100000, start_hz=100000, stop_hz=100000, ), make_segment(100000, target), ) clear_diagnostic(client) command(client, COMMAND_START) wait_for_status( client, lambda item: item.state in ACTIVE_STATES and abs(item.position) >= 100 and frequency_matches(item.frequency_hz, 100000), 2.0, "%s initial 100 kHz" % label, ) # Both writes are committed at an AB zero boundary. The first measures # PlsrAbLoadAndStart while the incoming cycle is still 100 kHz; the # second returns the final gate path to the worst-case 10 us period. client.write_multiple(SEGMENT_1_BASE, split_u32(80000)) wait_for_status( client, lambda item: item.state in ACTIVE_STATES and frequency_matches(item.frequency_hz, 80000), 2.0, "%s dynamic 80 kHz" % label, ) client.write_multiple(SEGMENT_1_BASE, split_u32(100000)) wait_for_status( client, lambda item: item.state in ACTIVE_STATES and frequency_matches(item.frequency_hz, 100000), 2.0, "%s restored 100 kHz" % label, ) status = wait_terminal(client, 3.0) require(status.state == STATUS_COMPLETED, "%s did not complete" % label) require(status.position == target, "%s position expected %d, got %d" % (label, target, status.position)) require_diagnostic_pass( read_diagnostic(client), pulses_per_case, OUTPUT_AB, direction > 0, label, ) def run_timing(client, timing, keep_config): snapshot = ( client.read_holding(OUTPUT_MODE, 1)[0], client.read_holding(CONFIG_BASE, COMMON_WORDS), client.read_holding(SEGMENT_1_BASE, SEGMENT_WORDS), client.read_holding(SEGMENT_2_BASE, SEGMENT_WORDS), ) try: timing.reset() run_case("timing_constant_100khz_four_outputs", lambda: test_all_outputs_at_100khz(client)) irq, _, _ = timing.read() require_irq_cycle_budget(irq, range(4), "constant 100 kHz") timing.reset() run_case("timing_short_final_profiles", lambda: test_short_final_profiles(client)) irq, producer, _ = timing.read() require_irq_cycle_budget(irq, (0,), "10-pulse profiles") require_producer_cycle_budget(producer, "10-pulse profiles") timing.reset() run_case("timing_long_millisecond_ramp", lambda: test_long_millisecond_ramp(client)) irq, producer, _ = timing.read() require_irq_cycle_budget(irq, (0,), "65,535-pulse 1 ms ramp") require_no_producer_activity(producer, "65,535-pulse 1 ms ramp") timing.reset() run_case("timing_ab_100khz_reload_and_fast_gate", lambda: test_ab_100khz_timing_paths(client)) irq, _, final_arm = timing.read() require_irq_cycle_budget( irq, (0, 2), "AB 100 kHz reload and fast gate") require_final_arm_cycle_budget( final_arm, (0, 2), "AB 100 kHz final arm") finally: if keep_config: cleanup_preserving_primary_error( lambda: safe_stop(client), "timing motion stop", ) else: cleanup_preserving_primary_error( lambda: restore_configuration(client, snapshot), "timing configuration restore", ) def test_wait_time(client): clear_position(client) configure_pulse_dir( client, make_common(start_hz=500), make_segment(500, 10, wait_type=WAIT_TIME, wait_time_ms=120), ) command(client, COMMAND_START) waiting = wait_for_status( client, lambda item: item.state == STATUS_WAITING, 2.0, "WAIT_TIME state", ) waiting_at = time.monotonic() require(waiting.position == 10, "WAIT_TIME position expected 10") status = wait_terminal(client, 2.0) observed_wait = time.monotonic() - waiting_at require(status.state == STATUS_COMPLETED, "WAIT_TIME did not complete") require(observed_wait >= 0.050, "WAIT_TIME completed implausibly early") require_diagnostic_pass( read_diagnostic(client), 10, OUTPUT_PULSE_DIR, True, "WAIT_TIME", ) def test_act_time(client): clear_position(client) configure_pulse_dir( client, make_common(start_hz=1000), make_segment(1000, 2000, wait_type=ACT_TIME, act_time_ms=80), ) command(client, COMMAND_START) status = wait_terminal(client, 3.0) require(status.state == STATUS_COMPLETED, "ACT_TIME did not complete") require(0 < status.position < 2000, "ACT_TIME did not cut the segment") def test_dynamic_frequency_and_repeated_stop(client): clear_position(client) configure_pulse_dir( client, make_common(start_hz=500, deceleration_ms=150), make_segment(500, 3000), ) command(client, COMMAND_START) wait_for_status( client, lambda item: item.segment == 1 and item.frequency_hz == 500, 2.0, "initial frequency", ) client.write_multiple(SEGMENT_1_BASE, split_u32(1200)) dynamic = wait_for_status( client, lambda item: item.segment == 1 and frequency_matches(item.frequency_hz, 1200), 2.0, "dynamic frequency", ) print("INFO dynamic frequency requested=1200 actual=%d" % dynamic.frequency_hz) expect_exception( lambda: client.write_single(CONFIG_BASE, 1), EX_DEVICE_BUSY, "pulse output write while busy", ) command(client, COMMAND_STOP) command(client, COMMAND_STOP) status = wait_terminal(client, 3.0) require(status.state == STATUS_STOPPED, "repeated STOP did not stop") require(status.position < 3000, "STOP did not cut the active move") def test_future_segment_frequency_applies_on_arrival(client): clear_position(client) common = make_common(segment_count=2, start_hz=400) segment_1 = make_segment(400, 200) segment_2 = make_segment(700, 200) configure_pulse_dir(client, common, segment_1, segment_2) command(client, COMMAND_START) wait_for_status( client, lambda item: item.segment == 1 and item.frequency_hz == 400, 2.0, "segment 1", ) client.write_multiple(SEGMENT_2_BASE, split_u32(900)) status = read_status(client) require(status.segment == 1, "future write changed the current segment") require(status.frequency_hz == 400, "future write changed current frequency") wait_for_status( client, lambda item: item.segment == 2 and frequency_matches(item.frequency_hz, 900), 3.0, "updated segment 2 frequency", ) status = wait_terminal(client, 3.0) require(status.state == STATUS_COMPLETED, "two-segment move did not complete") require_diagnostic_pass( read_diagnostic(client), 200, OUTPUT_PULSE_DIR, True, "updated segment 2 frequency", expected_segment=2, ) def test_subsequent_future_frequency_rebuilds_handoff(client): clear_position(client) common = make_common( segment_count=2, send_mode=SEND_SUBSEQUENT, start_hz=400, ) segment_1 = make_segment(400, 200) segment_2 = make_segment(700, 200) configure_pulse_dir(client, common, segment_1, segment_2) command(client, COMMAND_START) wait_for_status( client, lambda item: item.segment == 1 and frequency_matches(item.frequency_hz, 400), 2.0, "subsequent segment 1", ) client.write_multiple(SEGMENT_2_BASE, split_u32(900)) status = read_status(client) require(status.segment == 1, "future write changed the current segment") require( frequency_matches(status.frequency_hz, 400), "future write changed current frequency", ) wait_for_status( client, lambda item: item.segment == 2 and frequency_matches(item.frequency_hz, 900), 3.0, "rebuilt subsequent handoff frequency", ) status = wait_terminal(client, 3.0) require(status.state == STATUS_COMPLETED, "subsequent two-segment move did not complete") require(status.position == 400, "subsequent two-segment position expected 400") require_diagnostic_pass( read_diagnostic(client), 200, OUTPUT_PULSE_DIR, True, "subsequent segment 2 frequency", expected_segment=2, ) def require_fixture_level(fixture, input_selection, expected, label): actual = fixture.read(input_selection) require( actual == expected, "%s expected X%d=%d, got %d" % (label, input_selection + 4, expected, actual), ) def test_x45_electrical_scan(fixture): fixture.all_off() require_fixture_level(fixture, 0, 0, "both outputs off") require_fixture_level(fixture, 1, 0, "both outputs off") for input_selection in range(2): other = 1 - input_selection fixture.drive(input_selection, True) require_fixture_level(fixture, input_selection, 1, "output on") require_fixture_level(fixture, other, 0, "cross-channel isolation") fixture.drive(input_selection, False) require_fixture_level(fixture, input_selection, 0, "output off") def test_wait_signal_input(client, fixture, input_selection): fixture.drive(input_selection, False) clear_position(client) common = make_common( default_hz=1000, start_hz=1000, stop_hz=1000, ) common[0x02] = input_selection configure_pulse_dir( client, common, make_segment(1000, 50, wait_type=WAIT_SIGNAL), ) command(client, COMMAND_START) waiting = wait_for_status( client, lambda item: item.state == STATUS_WAITING, 2.0, "X%d WAIT_SIGNAL" % (input_selection + 4), ) require(waiting.position == 50, "WAIT_SIGNAL pulse count expected 50") time.sleep(0.050) require(read_status(client).state == STATUS_WAITING, "WAIT_SIGNAL advanced while input was off") fixture.drive(input_selection, True) status = wait_terminal(client, 2.0) require(status.state == STATUS_COMPLETED, "WAIT_SIGNAL did not complete") require(status.position == 50, "WAIT_SIGNAL changed completed position") require_diagnostic_pass( read_diagnostic(client), 50, OUTPUT_PULSE_DIR, True, "X%d WAIT_SIGNAL" % (input_selection + 4), ) fixture.drive(input_selection, False) def test_ext_signal_input(client, fixture, input_selection, wait_type): fixture.drive(input_selection, False) clear_position(client) common = make_common( default_hz=1000, start_hz=1000, stop_hz=1000, ) common[0x03] = input_selection configure_pulse_dir( client, common, make_segment(1000, 5000, wait_type=wait_type), ) command(client, COMMAND_START) wait_for_status( client, lambda item: item.state in ACTIVE_STATES and item.position >= 10, 2.0, "X%d external-signal active move" % (input_selection + 4), ) fixture.drive(input_selection, True) status = wait_terminal(client, 2.0) require(status.state == STATUS_COMPLETED, "external-signal move did not complete") require(0 < status.position < 5000, "external signal did not cut the active segment") fixture.drive(input_selection, False) def test_ext_or_complete_natural(client, fixture, input_selection): fixture.drive(input_selection, False) clear_position(client) common = make_common( default_hz=1000, start_hz=1000, stop_hz=1000, ) common[0x03] = input_selection configure_pulse_dir( client, common, make_segment(1000, 50, wait_type=EXT_OR_COMPLETE), ) command(client, COMMAND_START) status = wait_terminal(client, 2.0) require(status.state == STATUS_COMPLETED, "EXT_OR_COMPLETE natural branch did not complete") require(status.position == 50, "EXT_OR_COMPLETE natural branch expected 50 pulses") require_diagnostic_pass( read_diagnostic(client), 50, OUTPUT_PULSE_DIR, True, "X%d EXT_OR_COMPLETE natural" % (input_selection + 4), ) def run_x45(client, fixture, keep_config): snapshot = ( client.read_holding(OUTPUT_MODE, 1)[0], client.read_holding(CONFIG_BASE, COMMON_WORDS), client.read_holding(SEGMENT_1_BASE, SEGMENT_WORDS), client.read_holding(SEGMENT_2_BASE, SEGMENT_WORDS), ) try: client.write_single(OUTPUT_MODE, OUTPUT_PULSE_DIR) fixture.prepare() run_case("x45_electrical_scan", lambda: test_x45_electrical_scan(fixture)) for input_selection in range(2): name = "X%d" % (input_selection + 4) run_case( "%s_wait_signal" % name, lambda selection=input_selection: test_wait_signal_input(client, fixture, selection), ) run_case( "%s_ext_signal" % name, lambda selection=input_selection: test_ext_signal_input( client, fixture, selection, EXT_SIGNAL), ) run_case( "%s_ext_or_complete_natural" % name, lambda selection=input_selection: test_ext_or_complete_natural(client, fixture, selection), ) run_case( "%s_ext_or_complete_trigger" % name, lambda selection=input_selection: test_ext_signal_input( client, fixture, selection, EXT_OR_COMPLETE), ) finally: try: cleanup_preserving_primary_error( fixture.all_off, "X4/X5 output shutdown", ) finally: try: if keep_config: cleanup_preserving_primary_error( lambda: safe_stop(client), "X4/X5 motion stop", ) else: cleanup_preserving_primary_error( lambda: restore_configuration(client, snapshot), "X4/X5 configuration restore", ) finally: cleanup_preserving_primary_error( fixture.release, "X4/X5 fixture release", ) def safe_stop(client): status = read_status(client) if status.state in ACTIVE_STATES: command(client, COMMAND_STOP) wait_terminal(client, 3.0) def cleanup_preserving_primary_error(cleanup, label, warning_stream=None): primary_error = sys.exc_info()[1] try: cleanup() except Exception as cleanup_error: if primary_error is None: raise message = "%s failed during error cleanup: %s: %s" % ( label, type(cleanup_error).__name__, cleanup_error, ) if hasattr(primary_error, "add_note"): primary_error.add_note(message) if warning_stream is None: warning_stream = sys.stderr print("WARN: %s" % message, file=warning_stream) def restore_configuration(client, snapshot): safe_stop(client) command(client, COMMAND_CLEAR) client.write_single(OUTPUT_MODE, OUTPUT_PULSE_DIR) client.write_multiple(CONFIG_BASE, snapshot[1]) segments = snapshot[2] if len(snapshot) == 3 else snapshot[2:] for index, segment in enumerate(segments): client.write_multiple(SEGMENT_1_BASE + index * 0x10, segment) client.write_single(OUTPUT_MODE, snapshot[0]) def snapshot_ab_configuration(client): return ( client.read_holding(OUTPUT_MODE, 1)[0], client.read_holding(CONFIG_BASE, COMMON_WORDS), client.read_holding(SEGMENT_1_BASE, SEGMENT_WORDS), ) def restore_ab_configuration(client, snapshot): safe_stop(client) command(client, COMMAND_CLEAR) # PULSE/DIR accepts every pulse output, so use it while restoring a # possible Y1/Y3 configuration that would be invalid in AB mode. client.write_single(OUTPUT_MODE, OUTPUT_PULSE_DIR) client.write_multiple(CONFIG_BASE, snapshot[1]) client.write_multiple(SEGMENT_1_BASE, snapshot[2]) client.write_single(OUTPUT_MODE, snapshot[0]) clear_diagnostic(client) def configure_ab(client, output, common, segment): require(output in (0, 2), "AB output must select Y0/Y1 or Y2/Y3") common = list(common) common[0x00] = output client.write_multiple(CONFIG_BASE, common) client.write_multiple(SEGMENT_1_BASE, segment) client.write_single(OUTPUT_MODE, OUTPUT_AB) require(client.read_holding(OUTPUT_MODE, 1) == [OUTPUT_AB], "AB output mode readback failed") def run_isolated_ab_case(client, name, function): def isolated(): snapshot = snapshot_ab_configuration(client) try: function() finally: cleanup_preserving_primary_error( lambda: restore_ab_configuration(client, snapshot), "%s configuration restore" % name, ) run_case(name, isolated) def test_ab_completed_motion(client, output, frequency_hz, pulses, curve_mode, start_hz, stop_hz, acceleration_ms, deceleration_ms, timeout, label): clear_position(client) configure_ab( client, output, make_common( curve_mode=curve_mode, default_hz=frequency_hz, start_hz=start_hz, stop_hz=stop_hz, acceleration_ms=acceleration_ms, deceleration_ms=deceleration_ms, ), make_segment(frequency_hz, pulses), ) clear_diagnostic(client) command(client, COMMAND_START) status = wait_terminal(client, timeout) require(status.state == STATUS_COMPLETED, "%s did not complete" % label) require(status.position == pulses, "%s position expected %d, got %d" % (label, pulses, status.position)) diagnostic = read_diagnostic(client) require_diagnostic_pass( diagnostic, abs(pulses), OUTPUT_AB, pulses >= 0, label, ) print( "INFO %s pair=Y%d/Y%d direction=%s requested=%dHz " "timer=%dHz pulses=%d samples=%d" % ( label, output, output + 1, "positive" if pulses >= 0 else "negative", frequency_hz, diagnostic.active_timer_hz, diagnostic.actual_pulses, diagnostic.sample_count, ) ) def test_ab_stop(client): expected_pulses = 50000 clear_position(client) configure_ab( client, 0, make_common( default_hz=2000, start_hz=500, stop_hz=500, acceleration_ms=100, deceleration_ms=100, ), make_segment(2000, expected_pulses), ) clear_diagnostic(client) command(client, COMMAND_START) wait_for_status( client, lambda item: item.state in ACTIVE_STATES and item.position >= 10, 3.0, "AB STOP active motion", ) expect_exception( lambda: client.write_single(OUTPUT_MODE, OUTPUT_PULSE_DIR), EX_DEVICE_BUSY, "AB output mode write while busy", ) command(client, COMMAND_STOP) status = wait_terminal(client, 4.0) require(status.state == STATUS_STOPPED, "AB STOP did not enter STOPPED") require(0 < status.position < expected_pulses, "AB STOP did not cut the active move") diagnostic = read_diagnostic(client) require(not (diagnostic.flags & DIAG_MONITORING), "AB STOP diagnostic remained active") require(not (diagnostic.flags & DIAG_FAULT_LATCHED), "AB STOP latched diagnostic fault %d" % diagnostic.reason) require(not (diagnostic.flags & DIAG_COUNT_CHECKED), "AB STOP incorrectly marked incomplete count as checked") require(diagnostic.expected_pulses == expected_pulses, "AB STOP diagnostic target mismatch") require(0 < diagnostic.actual_pulses < expected_pulses, "AB STOP diagnostic actual count is implausible") require(diagnostic.output_mode == OUTPUT_AB, "AB STOP diagnostic mode mismatch") require(diagnostic.direction_positive, "AB STOP diagnostic direction mismatch") def test_ab_dynamic_frequency(client): expected_pulses = 6000 clear_position(client) configure_ab( client, 2, make_common( default_hz=800, start_hz=800, stop_hz=800, ), make_segment(800, -expected_pulses), ) clear_diagnostic(client) command(client, COMMAND_START) wait_for_status( client, lambda item: item.state in ACTIVE_STATES and item.position <= -10 and frequency_matches(item.frequency_hz, 800), 3.0, "AB dynamic initial frequency", ) client.write_multiple(SEGMENT_1_BASE, split_u32(3200)) dynamic = wait_for_status( client, lambda item: item.state in ACTIVE_STATES and frequency_matches(item.frequency_hz, 3200), 3.0, "AB dynamic updated frequency", ) print("INFO AB dynamic frequency requested=3200 actual=%d" % dynamic.frequency_hz) status = wait_terminal(client, 6.0) require(status.state == STATUS_COMPLETED, "AB dynamic-frequency move did not complete") require(status.position == -expected_pulses, "AB dynamic-frequency position mismatch") require_diagnostic_pass( read_diagnostic(client), expected_pulses, OUTPUT_AB, False, "AB dynamic frequency", ) def run_ab(client): cases = ( ( "ab_y0_y1_positive_1hz_linear_short", lambda: test_ab_completed_motion( client, 0, 1, 1, 0, 1, 1, 0, 0, 5.0, "AB Y0/Y1 positive 1 Hz linear short", ), ), ( "ab_y0_y1_negative_typical_s_curve", lambda: test_ab_completed_motion( client, 0, 2500, -1200, 1, 250, 250, 80, 80, 5.0, "AB Y0/Y1 negative typical S curve", ), ), ( "ab_y2_y3_positive_100khz_sine", lambda: test_ab_completed_motion( client, 2, 100000, 5000, 2, 1000, 1000, 20, 20, 4.0, "AB Y2/Y3 positive 100 kHz sine", ), ), ( "ab_y2_y3_negative_typical_linear", lambda: test_ab_completed_motion( client, 2, 5000, -600, 0, 500, 500, 20, 20, 4.0, "AB Y2/Y3 negative typical linear", ), ), ) for name, function in cases: run_isolated_ab_case(client, name, function) run_isolated_ab_case(client, "ab_stop", lambda: test_ab_stop(client)) run_isolated_ab_case( client, "ab_dynamic_frequency", lambda: test_ab_dynamic_frequency(client), ) def run_smoke(client): run_case("fixed_map_and_exceptions", lambda: test_fixed_map_and_exceptions(client)) def run_all(client, keep_config): snapshot = ( client.read_holding(OUTPUT_MODE, 1)[0], client.read_holding(CONFIG_BASE, COMMON_WORDS), [client.read_holding(SEGMENT_1_BASE + index * 0x10, SEGMENT_WORDS) for index in range(10)], ) try: client.write_single(OUTPUT_MODE, OUTPUT_PULSE_DIR) run_smoke(client) run_case( "positive_negative_absolute_zero", lambda: test_positive_negative_and_absolute_zero(client), ) run_case("ten_segments_sequence", lambda: test_ten_segments_in_sequence(client)) run_case("start_segment_ten", lambda: test_start_segment_ten(client)) run_case("direction_output_polarity_matrix", lambda: test_direction_output_and_polarity_matrix(client)) run_case("all_outputs_100khz", lambda: test_all_outputs_at_100khz(client)) run_case("short_final_profiles", lambda: test_short_final_profiles(client)) run_case("wait_time", lambda: test_wait_time(client)) run_case("act_time", lambda: test_act_time(client)) run_case( "dynamic_frequency_repeated_stop", lambda: test_dynamic_frequency_and_repeated_stop(client), ) run_case( "future_segment_frequency_on_arrival", lambda: test_future_segment_frequency_applies_on_arrival(client), ) run_case( "subsequent_future_frequency_handoff", lambda: test_subsequent_future_frequency_rebuilds_handoff(client), ) finally: if keep_config: cleanup_preserving_primary_error( lambda: safe_stop(client), "motion stop", ) else: cleanup_preserving_primary_error( lambda: restore_configuration(client, snapshot), "configuration restore", ) def validate_persistence_live_parameters(duration_seconds, interval_seconds): require(math.isfinite(duration_seconds), "persistence-live duration must be finite") require(duration_seconds >= PERSISTENCE_LIVE_MIN_DURATION_SECONDS, "persistence-live duration must be at least %.1f seconds" % PERSISTENCE_LIVE_MIN_DURATION_SECONDS) require(math.isfinite(interval_seconds), "persistence-live interval must be finite") require(0.0 < interval_seconds <= PERSISTENCE_LIVE_MAX_INTERVAL_SECONDS, "persistence-live interval must be > 0 and <= %.3f seconds" % PERSISTENCE_LIVE_MAX_INTERVAL_SECONDS) def require_persistence_live_stats(stats, duration_seconds, label): require(stats.attempts > 0, "%s made no FC03 attempts" % label) require(stats.successful >= 0 and stats.timeouts >= 0, "%s contains a negative counter" % label) require(stats.attempts == stats.successful + stats.timeouts, "%s counters are inconsistent" % label) require(stats.timeouts == 0, "%s had %d FC03 timeout(s)" % (label, stats.timeouts)) require(stats.successful == stats.attempts, "%s did not complete every FC03 request" % label) require(math.isfinite(stats.maximum_attempt_seconds) and stats.maximum_attempt_seconds >= 0.0, "%s maximum attempt latency is invalid" % label) require(stats.elapsed_seconds >= duration_seconds, "%s probe ran %.3f s, expected at least %.3f s" % (label, stats.elapsed_seconds, duration_seconds)) def probe_persistence_live_window(client, duration_seconds, interval_seconds): validate_persistence_live_parameters(duration_seconds, interval_seconds) start = time.monotonic() deadline = start + duration_seconds next_request = start attempts = 0 successful = 0 timeouts = 0 maximum_attempt_seconds = 0.0 reads = ( (STATUS_BASE, STATUS_WORDS), (PERSISTENCE_LIVE_CONFIG_ADDRESS, 1), ) while True: now = time.monotonic() if now >= deadline: break if now < next_request: time.sleep(min(next_request - now, deadline - now)) now = time.monotonic() if now >= deadline: break address, quantity = reads[attempts % len(reads)] request_started = time.monotonic() attempts += 1 try: client.read_holding(address, quantity) except RtuTimeout: timeouts += 1 else: successful += 1 attempt_seconds = time.monotonic() - request_started maximum_attempt_seconds = max( maximum_attempt_seconds, attempt_seconds, ) next_request += interval_seconds if next_request < time.monotonic(): next_request = time.monotonic() elapsed_seconds = time.monotonic() - start return PersistenceLiveStats( attempts=attempts, successful=successful, timeouts=timeouts, maximum_attempt_seconds=maximum_attempt_seconds, elapsed_seconds=elapsed_seconds, ) def print_persistence_live_result(label, stats, config_value): print( "INFO persistence-live window=%s attempts=%d successful=%d " "timeouts=%d max_attempt_ms=%.3f elapsed_s=%.3f " "config_0x%04X=%d" % ( label, stats.attempts, stats.successful, stats.timeouts, stats.maximum_attempt_seconds * 1000.0, stats.elapsed_seconds, PERSISTENCE_LIVE_CONFIG_ADDRESS, config_value, ) ) def persistence_live(client, duration_seconds, interval_seconds): validate_persistence_live_parameters(duration_seconds, interval_seconds) original = client.read_holding(PERSISTENCE_LIVE_CONFIG_ADDRESS, 1)[0] changed = (original + 1) & 0xFFFF original_timeout = client.timeout client.timeout = min( original_timeout, PERSISTENCE_LIVE_REQUEST_TIMEOUT_SECONDS, ) print( "INFO persistence-live duration_s=%.3f interval_ms=%.3f " "request_timeout_ms=%.3f" % ( duration_seconds, interval_seconds * 1000.0, client.timeout * 1000.0, ) ) def restore_original_configuration(): client.write_single(PERSISTENCE_LIVE_CONFIG_ADDRESS, original) restore_stats = probe_persistence_live_window( client, duration_seconds, interval_seconds, ) restore_readback = client.read_holding( PERSISTENCE_LIVE_CONFIG_ADDRESS, 1, )[0] print_persistence_live_result( "restore", restore_stats, restore_readback, ) require_persistence_live_stats( restore_stats, duration_seconds, "persistence-live restore window", ) require(restore_readback == original, "persistence-live original config was not restored") try: try: client.write_single(PERSISTENCE_LIVE_CONFIG_ADDRESS, changed) change_stats = probe_persistence_live_window( client, duration_seconds, interval_seconds, ) change_readback = client.read_holding( PERSISTENCE_LIVE_CONFIG_ADDRESS, 1, )[0] print_persistence_live_result( "change", change_stats, change_readback, ) require_persistence_live_stats( change_stats, duration_seconds, "persistence-live change window", ) require(change_readback == changed, "persistence-live changed config readback mismatch") finally: cleanup_preserving_primary_error( restore_original_configuration, "persistence-live configuration restore", ) finally: client.timeout = original_timeout def persistence_prepare(client, motion_timeout=3.0, settle_seconds=1.2, announce=True): snapshot = snapshot_ab_configuration(client) prepared = False try: clear_position(client) configure_ab( client, 0, make_common(start_hz=500), make_segment(500, 7), ) command(client, COMMAND_START) status = wait_terminal(client, motion_timeout) require(status.state == STATUS_COMPLETED, "persistence move did not complete") require(status.position == 7, "persistence position expected 7") time.sleep(settle_seconds) require(read_status(client).position == 7, "position changed before reset") prepared = True finally: if not prepared: cleanup_preserving_primary_error( lambda: restore_ab_configuration(client, snapshot), "persistence preparation restore", ) if announce: print("PREPARED: release COM, reset/power-cycle the MCU, then run verify phase") def persistence_verify(client, cleanup): status = read_status(client) require(status.state == STATUS_IDLE, "post-reset state must be IDLE") require(status.position == 7, "post-reset position expected 7") require(client.read_holding(OUTPUT_MODE, 1) == [OUTPUT_AB], "AB output mode was not restored after reset") segment = client.read_holding(SEGMENT_1_BASE, SEGMENT_WORDS) require(join_u32(segment[0], segment[1]) == 500, "frequency was not restored") require(join_i32(segment[2], segment[3]) == 7, "pulse target was not restored") if cleanup: command(client, COMMAND_CLEAR) require(read_status(client).position == 0, "cleanup CLEAR failed") restore_default_configuration(client) time.sleep(1.2) common = client.read_holding(CONFIG_BASE, COMMON_WORDS) segment = client.read_holding(SEGMENT_1_BASE, SEGMENT_WORDS) require(join_u32(common[0x0B], common[0x0C]) == 1000, "cleanup default speed was not restored") require(join_u32(segment[0], segment[1]) == 1000, "cleanup segment frequency was not restored") require(join_i32(segment[2], segment[3]) == 1000, "cleanup segment pulses were not restored") def defaults_verify(client): status = read_status(client) require(status.state == STATUS_IDLE, "default state must be IDLE") require(status.position == 0, "default position must be zero") require(client.read_holding(OUTPUT_MODE, 1) == [OUTPUT_PULSE_DIR], "default output mode must be PULSE/DIR") common = client.read_holding(CONFIG_BASE, COMMON_WORDS) require(common[0x05] == 10, "default direction delay expected 10 ms") require(join_u32(common[0x0B], common[0x0C]) == 1000, "default speed expected 1000 Hz") require(join_u32(common[0x0D], common[0x0E]) == 100, "start speed expected 100 Hz") require(join_u32(common[0x10], common[0x11]) == 100, "stop speed expected 100 Hz") for index in range(10): base = SEGMENT_1_BASE + index * 0x10 segment = client.read_holding(base, SEGMENT_WORDS) expected_pulses = 1000 if index == 0 else 0 require(join_u32(segment[0], segment[1]) == 1000, "default segment %d frequency mismatch" % (index + 1)) require(join_i32(segment[2], segment[3]) == expected_pulses, "default segment %d pulse count mismatch" % (index + 1)) class _PersistenceSelfTestClient: def __init__(self, complete_on_start): self.complete_on_start = complete_on_start self.output_mode = OUTPUT_PULSE_DIR self.common = make_common(default_hz=1234) self.segment = make_segment(2345, -17) self.state = STATUS_IDLE self.position = 41 self.calls = [] def read_holding(self, address, quantity): self.calls.append(("read", address, quantity)) if address == OUTPUT_MODE and quantity == 1: return [self.output_mode] if address == CONFIG_BASE and quantity == COMMON_WORDS: return list(self.common) if address == SEGMENT_1_BASE and quantity == SEGMENT_WORDS: return list(self.segment) if address == STATUS_BASE and quantity == STATUS_WORDS: return ( split_i32(self.position) + split_u32(500) + [self.state, 1 if self.state in ACTIVE_STATES else 0, ERROR_NONE] ) if address == DIAGNOSTIC_BASE and quantity == DIAGNOSTIC_WORDS: words = [0] * DIAGNOSTIC_WORDS words[24:26] = [0xFFFF, 0xFFFF] return words raise AssertionError("unexpected self-test read 0x%04X/%d" % (address, quantity)) def write_single(self, address, value): self.calls.append(("single", address, value)) if address == OUTPUT_MODE: self.output_mode = value elif address == DIAGNOSTIC_CONTROL: require(value == DIAGNOSTIC_CLEAR, "self-test diagnostic command mismatch") elif address == CONTROL: if value == COMMAND_CLEAR: self.state = STATUS_IDLE self.position = 0 elif value == COMMAND_START: if self.complete_on_start: self.state = STATUS_COMPLETED self.position = 7 else: self.state = STATUS_RUNNING self.position = 1 elif value == COMMAND_STOP: self.state = STATUS_STOPPED else: raise AssertionError("unexpected self-test control command") else: raise AssertionError("unexpected self-test single write 0x%04X" % address) def write_multiple(self, address, values): self.calls.append(("multiple", address, list(values))) if address == CONFIG_BASE: self.common = list(values) elif address == SEGMENT_1_BASE: self.segment = list(values) else: raise AssertionError("unexpected self-test multiple write 0x%04X" % address) def self_test(): def raise_error(error): raise error payload = bytes.fromhex("01 01 00 00 00 01") require(crc16(payload) == 0xCAFD, "CRC reference vector failed") require(append_crc(payload) == bytes.fromhex("01 01 00 00 00 01 FD CA"), "wire CRC order failed") for value in (0, 1, 0x7FFFFFFF, -1, -123456, -0x80000000): words = split_i32(value) require(join_i32(words[0], words[1]) == value, "int32 word round trip failed") require(frequency_matches(1199, 1200), "frequency tolerance rejected 1 Hz") require(not frequency_matches(1198, 1200), "frequency tolerance accepted excessive error") sample = ( "\n0x42408214 : 00000001 00000002 00000003 00000004 \n" "0x42408224 : 00000005 \n" ) require(parse_stlink_word(sample, X4_INPUT_BIT) == 1, "ST-LINK read parser failed") require(parse_stlink_words(sample, X4_INPUT_BIT, 5) == [1, 2, 3, 4, 5], "ST-LINK multi-word parser failed") map_sample = ( "PlsrIrqCount 0x2000'4d80 0x10 Data Gb object.o\n" "PlsrProfileProducerItemCount\n" " 0x2000'61c0 0x4 Data Gb object.o\n" ) require(parse_iar_map_symbol_address(map_sample, "PlsrIrqCount") == 0x20004D80, "IAR map symbol parser failed") require(parse_iar_map_symbol_address( map_sample, "PlsrProfileProducerItemCount") == 0x200061C0, "IAR wrapped map symbol parser failed") passing_irq_stats = IrqCycleStats( count=(1, 2, 3, 4), last_cycles=(100, 200, 300, 400), maximum_cycles=(1679, 1200, 900, 800), ) require_irq_cycle_budget( passing_irq_stats, range(4), "timing self-test pass") expect_test_failure( lambda: require_irq_cycle_budget( dataclasses.replace( passing_irq_stats, maximum_cycles=(IRQ_CYCLE_BUDGET_100KHZ, 1200, 900, 800), ), (0,), "timing self-test reject", ), "IRQ cycle budget equality", ) passing_final_arm_stats = FinalArmCycleStats( queue_count=(2, 0, 2, 0), job_last_cycles=(700, 0, 800, 0), job_maximum_cycles=(900, 0, 950, 0), queue_to_stop_last_cycles=(1000, 0, 1100, 0), queue_to_stop_maximum_cycles=(1200, 0, 1300, 0), ) require_final_arm_cycle_budget( passing_final_arm_stats, (0, 2), "final-arm timing self-test pass") for invalid_stats, label in ( (dataclasses.replace( passing_final_arm_stats, job_maximum_cycles=(IRQ_CYCLE_BUDGET_100KHZ, 0, 950, 0)), "final-arm job budget equality"), (dataclasses.replace( passing_final_arm_stats, queue_to_stop_maximum_cycles=(1200, 0, IRQ_CYCLE_BUDGET_100KHZ, 0)), "final-arm latency budget equality"), (dataclasses.replace( passing_final_arm_stats, queue_count=(0, 0, 2, 0)), "final-arm missing job"), ): expect_test_failure( lambda item=invalid_stats: require_final_arm_cycle_budget( item, (0, 2), "final-arm timing self-test reject"), label, ) passing_producer_stats = ProducerCycleStats( item_count=4, total_cycles=6719, maximum_item_cycles=1679, ) require_producer_cycle_budget( passing_producer_stats, "producer timing self-test pass") expect_test_failure( lambda: require_producer_cycle_budget( dataclasses.replace( passing_producer_stats, total_cycles=IRQ_CYCLE_BUDGET_100KHZ * 4, maximum_item_cycles=IRQ_CYCLE_BUDGET_100KHZ, ), "producer timing self-test reject", ), "producer cycle budget equality", ) no_producer_stats = ProducerCycleStats( item_count=0, total_cycles=0, maximum_item_cycles=0, ) require_no_producer_activity( no_producer_stats, "long-ramp producer self-test pass") for invalid_stats, label in ( (dataclasses.replace(no_producer_stats, item_count=1), "long-ramp producer item count"), (dataclasses.replace(no_producer_stats, total_cycles=1), "long-ramp producer total cycles"), (dataclasses.replace(no_producer_stats, maximum_item_cycles=1), "long-ramp producer maximum cycles"), ): expect_test_failure( lambda item=invalid_stats: require_no_producer_activity( item, "long-ramp producer self-test reject"), label, ) diagnostic_words = [ DIAG_COMPLETE_PASS, 0, 3, 0x0101, 0x5678, 0x1234, 0x5678, 0x1234, 0, 0, 0x86A0, 1, 0x869F, 1, 0x869F, 1, 0xFFFF, 0xFFFF, 0x4321, 0x0002, 0, 0, 0, 0, 0xFFFF, 0xFFFF, ] diagnostic = parse_diagnostic(diagnostic_words) require(DIAGNOSTIC_BASE == 0x2100 and DIAGNOSTIC_WORDS == 26, "diagnostic address or length changed") require(OUTPUT_MODE == 0x1200 and DIAGNOSTIC_CONTROL == 0x3100, "enhanced configuration/control address changed") require(diagnostic.output_mode == OUTPUT_AB, "diagnostic mode parser failed") require(diagnostic.direction_positive, "diagnostic direction parser failed") require(diagnostic.expected_pulses == 0x12345678, "diagnostic expected count parser failed") require(diagnostic.actual_pulses == 0x12345678, "diagnostic actual count parser failed") require(diagnostic.requested_hz == 100000, "diagnostic requested frequency parser failed") require(diagnostic.request_error_hz == -1, "diagnostic signed frequency error parser failed") require(diagnostic.sample_count == 0x00024321, "diagnostic sample count parser failed") require(diagnostic.first_mismatch_sample == 0xFFFFFFFF, "diagnostic sentinel parser failed") require((DIAG_COMPLETE_PASS & DIAG_FAULT_LATCHED) == 0, "diagnostic pass mask includes fault bit") for expected_error in ( ModbusException(0x03, EX_DEVICE_BUSY), serial.SerialException("self-test serial failure"), ): class ReadFailureClient: def read_holding(self, address, quantity): raise expected_error try: safe_stop(ReadFailureClient()) except Exception as actual_error: require(actual_error is expected_error, "safe_stop replaced the original communication error") else: raise TestFailure("safe_stop swallowed a communication error") primary_error = TestFailure("primary board-test failure") cleanup_error = ModbusException(0x06, EX_DEVICE_BUSY) cleanup_warning = io.StringIO() caught_error = None try: try: raise primary_error finally: cleanup_preserving_primary_error( lambda: raise_error(cleanup_error), "self-test configuration restore", cleanup_warning, ) except TestFailure as error: caught_error = error require(caught_error is primary_error, "cleanup failure replaced the primary board-test failure") require("self-test configuration restore failed during error cleanup" in cleanup_warning.getvalue(), "cleanup failure warning omitted its operation context") if hasattr(primary_error, "add_note"): require(any("self-test configuration restore" in note for note in getattr(primary_error, "__notes__", ())), "cleanup failure was not attached to the primary exception") cleanup_error = serial.SerialException("standalone cleanup failure") try: cleanup_preserving_primary_error( lambda: raise_error(cleanup_error), "self-test standalone cleanup", io.StringIO(), ) except serial.SerialException as error: require(error is cleanup_error, "standalone cleanup failure was replaced") else: raise TestFailure("standalone cleanup failure was swallowed") validate_persistence_live_parameters( PERSISTENCE_LIVE_MIN_DURATION_SECONDS, PERSISTENCE_LIVE_MAX_INTERVAL_SECONDS, ) invalid_live_parameters = ( (PERSISTENCE_LIVE_MIN_DURATION_SECONDS - 0.001, 0.05, "short persistence-live duration"), (float("nan"), 0.05, "NaN persistence-live duration"), (float("inf"), 0.05, "infinite persistence-live duration"), (PERSISTENCE_LIVE_MIN_DURATION_SECONDS, 0.0, "zero persistence-live interval"), (PERSISTENCE_LIVE_MIN_DURATION_SECONDS, PERSISTENCE_LIVE_MAX_INTERVAL_SECONDS + 0.001, "long persistence-live interval"), (PERSISTENCE_LIVE_MIN_DURATION_SECONDS, float("nan"), "NaN persistence-live interval"), ) for duration_seconds, interval_seconds, label in invalid_live_parameters: expect_test_failure( lambda duration=duration_seconds, interval=interval_seconds: validate_persistence_live_parameters(duration, interval), label, ) passing_live_stats = PersistenceLiveStats( attempts=50, successful=50, timeouts=0, maximum_attempt_seconds=0.012, elapsed_seconds=PERSISTENCE_LIVE_MIN_DURATION_SECONDS, ) require_persistence_live_stats( passing_live_stats, PERSISTENCE_LIVE_MIN_DURATION_SECONDS, "persistence-live self-test pass", ) failing_live_stats = ( (dataclasses.replace(passing_live_stats, successful=49, timeouts=1), "persistence-live timeout stats"), (dataclasses.replace(passing_live_stats, successful=49), "persistence-live inconsistent stats"), (dataclasses.replace(passing_live_stats, attempts=0, successful=0), "persistence-live empty stats"), (dataclasses.replace(passing_live_stats, elapsed_seconds=2.499), "persistence-live short elapsed stats"), (dataclasses.replace(passing_live_stats, maximum_attempt_seconds=float("nan")), "persistence-live invalid latency stats"), ) for stats, label in failing_live_stats: expect_test_failure( lambda item=stats: require_persistence_live_stats( item, PERSISTENCE_LIVE_MIN_DURATION_SECONDS, "persistence-live self-test reject", ), label, ) failed_prepare = _PersistenceSelfTestClient(complete_on_start=False) original_common = list(failed_prepare.common) original_segment = list(failed_prepare.segment) try: persistence_prepare( failed_prepare, motion_timeout=0.0, settle_seconds=0.0, announce=False, ) except TestFailure: pass else: raise TestFailure("failed persistence prepare unexpectedly succeeded") controls = [call[2] for call in failed_prepare.calls if call[:2] == ("single", CONTROL)] require(COMMAND_STOP in controls, "failed persistence prepare did not issue STOP") require(failed_prepare.output_mode == OUTPUT_PULSE_DIR, "failed persistence prepare did not restore output mode") require(failed_prepare.common == original_common, "failed persistence prepare did not restore common config") require(failed_prepare.segment == original_segment, "failed persistence prepare did not restore segment config") successful_prepare = _PersistenceSelfTestClient(complete_on_start=True) persistence_prepare( successful_prepare, motion_timeout=0.1, settle_seconds=0.0, announce=False, ) controls = [call[2] for call in successful_prepare.calls if call[:2] == ("single", CONTROL)] require(COMMAND_STOP not in controls, "successful persistence prepare issued STOP") require(successful_prepare.output_mode == OUTPUT_AB, "successful persistence prepare did not retain AB mode") require(join_u32(successful_prepare.segment[0], successful_prepare.segment[1]) == 500, "successful persistence prepare did not retain frequency") require(join_i32(successful_prepare.segment[2], successful_prepare.segment[3]) == 7, "successful persistence prepare did not retain pulse target") print("Self-test passed; no serial port was opened") def print_preflight(args): print("No serial port opened. Board-test preconditions:") print(" - Firmware containing the PLSR 0x1000/0x2000/0x3000 map is flashed.") print(" - RS-485 is %d baud, 8E1, slave %d on %s." % (args.baud, args.slave, args.port)) if args.phase == "ab": print(" - Y0/Y1 and Y2/Y3 AB output pairs are safe to toggle.") else: print(" - Configured pulse, direction, and AB-pair outputs are safe to toggle.") print(" - No other program has the COM port open.") print("Run smoke only:") print(" py -B HostComputer\\plsr_modbus_product_test.py --run --phase smoke") print("Run motion matrix:") print(" py -B HostComputer\\plsr_modbus_product_test.py --run --allow-motion") print("Run X4/X5 loopback tests:") print(" py -B HostComputer\\plsr_modbus_product_test.py --run --phase x45 --allow-motion") print("Run AB output and diagnostic matrix:") print(" py -B HostComputer\\plsr_modbus_product_test.py --run --phase ab --allow-motion") print("Run DWT ISR/producer timing acceptance:") print(" py -B HostComputer\\plsr_modbus_product_test.py --run --phase timing --allow-motion") print("Run persistence communication-window test:") print(" py -B HostComputer\\plsr_modbus_product_test.py --run --phase persistence-live") print("Run offline checks:") print(" py -B HostComputer\\plsr_modbus_product_test.py --self-test") def parse_arguments(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--port", default="COM5") parser.add_argument("--baud", type=int, default=9600) parser.add_argument("--slave", type=int, default=1) parser.add_argument("--timeout", type=float, default=1.5) parser.add_argument("--stlink-cli", default=DEFAULT_STLINK_CLI) parser.add_argument("--stlink-id", type=int, default=0) parser.add_argument("--stlink-timeout", type=float, default=20.0) parser.add_argument("--iar-map", default=DEFAULT_IAR_MAP) parser.add_argument( "--phase", choices=("smoke", "all", "ab", "x45", "timing", "persistence-prepare", "persistence-verify", "persistence-live", "defaults-verify"), default="all", ) parser.add_argument("--run", action="store_true", help="open the serial port and execute the selected phase") parser.add_argument("--allow-motion", action="store_true", help="confirm that configured motion outputs may toggle") parser.add_argument("--keep-config", action="store_true", help="retain matrix config; AB phase always restores each case") parser.add_argument("--cleanup", action="store_true", help="CLEAR position after persistence verification") parser.add_argument( "--persistence-duration", type=float, default=PERSISTENCE_LIVE_DEFAULT_DURATION_SECONDS, help="seconds to probe each persistence-live window (minimum 2.5)", ) parser.add_argument( "--persistence-interval", type=float, default=PERSISTENCE_LIVE_DEFAULT_INTERVAL_SECONDS, help="seconds between persistence-live FC03 requests (maximum 0.1)", ) parser.add_argument("--self-test", action="store_true", help="run offline CRC/word tests without opening a port") return parser.parse_args() def main(): args = parse_arguments() require(1 <= args.slave <= 247, "slave must be 1..247") require(args.baud > 0, "baud must be positive") require(args.timeout > 0, "timeout must be positive") require(args.stlink_id >= 0, "ST-LINK ID must not be negative") require(args.stlink_timeout > 0, "ST-LINK timeout must be positive") if args.self_test: self_test() return 0 if not args.run: print_preflight(args) return 0 motion_phase = args.phase in ( "all", "ab", "x45", "timing", "persistence-prepare") if motion_phase and not args.allow_motion: raise TestFailure("motion phase requires --allow-motion") print("Opening %s at %d 8E1, slave %d" % (args.port, args.baud, args.slave)) with RtuClient(args.port, args.baud, args.slave, args.timeout) as client: if args.phase == "smoke": run_smoke(client) elif args.phase == "all": run_all(client, args.keep_config) elif args.phase == "ab": run_ab(client) elif args.phase == "x45": fixture = X45Fixture( args.stlink_cli, args.stlink_id, args.stlink_timeout, ) run_x45(client, fixture, args.keep_config) elif args.phase == "timing": probe = X45Fixture( args.stlink_cli, args.stlink_id, args.stlink_timeout, ) timing = TimingFixture(probe, args.iar_map) run_timing(client, timing, args.keep_config) elif args.phase == "persistence-prepare": persistence_prepare(client) elif args.phase == "persistence-verify": persistence_verify(client, args.cleanup) elif args.phase == "persistence-live": persistence_live( client, args.persistence_duration, args.persistence_interval, ) else: defaults_verify(client) print("PASS phase=%s" % args.phase) return 0 if __name__ == "__main__": try: sys.exit(main()) except (TestFailure, ModbusException, serial.SerialException) as error: print("FAIL: %s" % error, file=sys.stderr) sys.exit(1)