"""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 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 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" 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 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): pattern = r"(?im)^\s*0x%08x\s*:\s*([0-9a-f]{8})\s*$" % address match = re.search(pattern, output) require(match is not None, "ST-LINK output did not contain address 0x%08X" % address) return int(match.group(1), 16) 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 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 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) self.last_request_finished = time.monotonic() require(response, "no Modbus response to %s" % request.hex(" ")) require(expected_length is not None, "response header is incomplete") require(len(response) == expected_length, "incomplete or 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 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 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 restore_default_configuration(client): 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) 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(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", ) def test_positive_negative_and_absolute_zero(client): clear_position(client) common = make_common(position_mode=0, start_hz=500) configure(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") configure(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") absolute = make_common(position_mode=1, start_hz=500) configure(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_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(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)) 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(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)) def test_wait_time(client): clear_position(client) configure( 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") def test_act_time(client): clear_position(client) configure( 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( 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(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") 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(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") 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(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") 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(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( 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") def run_x45(client, fixture, keep_config): snapshot = ( client.read_holding(CONFIG_BASE, COMMON_WORDS), client.read_holding(SEGMENT_1_BASE, SEGMENT_WORDS), client.read_holding(SEGMENT_2_BASE, SEGMENT_WORDS), ) try: 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: fixture.all_off() finally: try: safe_stop(client) if not keep_config: restore_configuration(client, snapshot) finally: fixture.release() def safe_stop(client): try: status = read_status(client) if status.state in ACTIVE_STATES: command(client, COMMAND_STOP) wait_terminal(client, 3.0) except (ModbusException, TestFailure, serial.SerialException): pass def restore_configuration(client, snapshot): safe_stop(client) command(client, COMMAND_CLEAR) client.write_multiple(CONFIG_BASE, snapshot[0]) client.write_multiple(SEGMENT_1_BASE, snapshot[1]) client.write_multiple(SEGMENT_2_BASE, snapshot[2]) 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(CONFIG_BASE, COMMON_WORDS), client.read_holding(SEGMENT_1_BASE, SEGMENT_WORDS), client.read_holding(SEGMENT_2_BASE, SEGMENT_WORDS), ) try: run_smoke(client) run_case( "positive_negative_absolute_zero", lambda: test_positive_negative_and_absolute_zero(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: safe_stop(client) else: restore_configuration(client, snapshot) def persistence_prepare(client): clear_position(client) configure( client, make_common(start_hz=500), make_segment(500, 7), ) command(client, COMMAND_START) status = wait_terminal(client, 3.0) require(status.state == STATUS_COMPLETED, "persistence move did not complete") require(status.position == 7, "persistence position expected 7") time.sleep(1.2) require(read_status(client).position == 7, "position changed before reset") 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") 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") 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)) def self_test(): 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 \n" require(parse_stlink_word(sample, X4_INPUT_BIT) == 1, "ST-LINK read parser failed") 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)) print(" - Y0 pulse and Y12 direction 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 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( "--phase", choices=("smoke", "all", "x45", "persistence-prepare", "persistence-verify", "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 pulse and direction outputs may toggle") parser.add_argument("--keep-config", action="store_true", help="do not restore the three modified config blocks") parser.add_argument("--cleanup", action="store_true", help="CLEAR position after persistence verification") 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", "x45", "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 == "x45": fixture = X45Fixture( args.stlink_cli, args.stlink_id, args.stlink_timeout, ) run_x45(client, fixture, args.keep_config) elif args.phase == "persistence-prepare": persistence_prepare(client) elif args.phase == "persistence-verify": persistence_verify(client, args.cleanup) 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)