tka/scanengine-3
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

removed unnessicary files

Browse Source
This commit is contained in:
Thomas Ales [M S E]
2026-02-09 16:33:01 -06:00
parent 57613b7aca
commit 43e7b99512
13 changed files with 2 additions and 2295 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.gitignore
+2
View File
@@ -1,3 +1,5 @@
# claude
.claude
# Byte-compiled / optimized / DLL files # Byte-compiled / optimized / DLL files
__pycache__/ __pycache__/
*.py[codz] *.py[codz]
aaa_155749_20260209.json
-144
View File
@@ -1,144 +0,0 @@
{
"scan_info": {
"friendly_name": "stest",
"waveform_prefix": "aaa",
"data_directory": "/opt/scanengine-3",
"timestamp": "2026-02-09T15:57:49.753094",
"scan_finished": true,
"completion_timestamp": "2026-02-09T15:57:58.301940"
},
"scan_parameters": {
"number_of_angles": 4,
"row_spacing_mm": 0.5,
"scan_type": "standalone"
},
"scan_area": {
"x_start_mm": 48.5,
"x_delta_mm": 15.83,
"y_start_mm": 42.17,
"y_delta_mm": -15.67,
"sample_size": "1.25\""
},
"scan_boxes": [
{
"angle_index": 0,
"angle_degrees": 0.0,
"start": [
48.5,
26.5
],
"end": [
64.33,
42.17
],
"original_corners": [
[
48.5,
42.17
],
[
64.33,
42.17
],
[
64.33,
26.5
],
[
48.5,
26.5
]
]
},
{
"angle_index": 1,
"angle_degrees": 45.0,
"start": [
45.3339,
24.3934
],
"end": [
67.6077,
46.6673
],
"original_corners": [
[
45.3339,
35.4738
],
[
56.5274,
46.6673
],
[
67.6077,
35.5869
],
[
56.4142,
24.3934
]
]
},
{
"angle_index": 2,
"angle_degrees": 90.0,
"start": [
47.83,
28.5
],
"end": [
63.5,
44.33
],
"original_corners": [
[
47.83,
28.5
],
[
47.83,
44.33
],
[
63.5,
44.33
],
[
63.5,
28.5
]
]
},
{
"angle_index": 3,
"angle_degrees": 135.0,
"start": [
43.3327,
25.3339
],
"end": [
65.6066,
47.6077
],
"original_corners": [
[
54.5262,
25.3339
],
[
43.3327,
36.5274
],
[
54.4131,
47.6077
],
[
65.6066,
36.4142
]
]
}
]
}
aaa_160628_20260209.json
-144
View File
@@ -1,144 +0,0 @@
{
"scan_info": {
"friendly_name": "test2",
"waveform_prefix": "aaa",
"data_directory": "/opt/scanengine-3",
"timestamp": "2026-02-09T16:06:28.364606",
"scan_finished": true,
"completion_timestamp": "2026-02-09T16:27:37.121723"
},
"scan_parameters": {
"number_of_angles": 4,
"row_spacing_mm": 0.25,
"scan_type": "standalone"
},
"scan_area": {
"x_start_mm": 54.83,
"x_delta_mm": 9.83,
"y_start_mm": 43.83,
"y_delta_mm": -9.17,
"sample_size": "1.25\""
},
"scan_boxes": [
{
"angle_index": 0,
"angle_degrees": 0.0,
"start": [
54.83,
34.66
],
"end": [
64.66,
43.83
],
"original_corners": [
[
54.83,
43.83
],
[
64.66,
43.83
],
[
64.66,
34.66
],
[
54.83,
34.66
]
]
},
{
"angle_index": 1,
"angle_degrees": 45.0,
"start": [
48.636,
34.6394
],
"end": [
62.0711,
48.0744
],
"original_corners": [
[
48.636,
41.1235
],
[
55.5869,
48.0744
],
[
62.0711,
41.5902
],
[
55.1202,
34.6394
]
]
},
{
"angle_index": 2,
"angle_degrees": 90.0,
"start": [
46.17,
34.83
],
"end": [
55.34,
44.66
],
"original_corners": [
[
46.17,
34.83
],
[
46.17,
44.66
],
[
55.34,
44.66
],
[
55.34,
34.83
]
]
},
{
"angle_index": 3,
"angle_degrees": 135.0,
"start": [
41.9256,
28.636
],
"end": [
55.3606,
42.0711
],
"original_corners": [
[
48.8765,
28.636
],
[
41.9256,
35.5869
],
[
48.4098,
42.0711
],
[
55.3606,
35.1202
]
]
}
]
}
tests/__init__.py
-1
View File
@@ -1 +0,0 @@
"""Test modules for ScanEngine-3"""
tests/test_bbd202_diagnostic.py
-569
View File
@@ -1,569 +0,0 @@
"""
BBD202 Stage Diagnostic Tests
Diagnostic tests for the Thorlabs MLS203/BBD202 motion controller to:
1. Check if axes are properly homed
2. Move in a 15mm square pattern while outputting position data
3. Detect stage hangs through position monitoring
4. Output response information for debugging
"""
import sys
import time
from typing import Optional
from hardware.bbd202 import MotionController, MotorStatusBits
def format_status_bits(bits: Optional[int]) -> str:
"""Format status bits as a readable string with key flags."""
if bits is None:
return "None (no status received)"
flags = []
status = MotorStatusBits(bits)
# Key flags to check
if MotorStatusBits.HOMED in status:
flags.append("HOMED")
if MotorStatusBits.HOMING in status:
flags.append("HOMING")
if MotorStatusBits.ENABLED in status:
flags.append("ENABLED")
if MotorStatusBits.SETTLED in status:
flags.append("SETTLED")
if MotorStatusBits.TRACKING in status:
flags.append("TRACKING")
if MotorStatusBits.INMOTIONCW in status:
flags.append("INMOTIONCW")
if MotorStatusBits.INMOTIONCCW in status:
flags.append("INMOTIONCCW")
if MotorStatusBits.POWEROK in status:
flags.append("POWEROK")
if MotorStatusBits.ERROR in status:
flags.append("ERROR")
if MotorStatusBits.POSITIONERROR in status:
flags.append("POSITIONERROR")
if MotorStatusBits.OVERTEMP in status:
flags.append("OVERTEMP")
if MotorStatusBits.COMMUTATIONERROR in status:
flags.append("COMMUTATIONERROR")
flag_str = ", ".join(flags) if flags else "NO_FLAGS"
return f"0x{bits:08X} [{flag_str}]"
def test_homing_status(mc: MotionController) -> bool:
"""
Test 1: Check if both axes are properly homed.
Returns True if both axes are homed, False otherwise.
"""
print("\n" + "="*70)
print("TEST 1: Checking Homing Status")
print("="*70)
# Request fresh status from both axes
print("\nRequesting status update from both axes...")
mc.request_status_update(mc.DEST_X_AXIS)
mc.request_status_update(mc.DEST_Y_AXIS)
time.sleep(0.2) # Allow time for response
# Also poll status to ensure we have latest
mc.poll_status()
time.sleep(0.1)
# Check X-axis
x_bits = mc.status_bits_x
x_homed = mc.is_homed_x
print(f"\nX-Axis Status:")
print(f" Raw status bits: {format_status_bits(x_bits)}")
print(f" is_homed_x: {x_homed}")
# Check Y-axis
y_bits = mc.status_bits_y
y_homed = mc.is_homed_y
print(f"\nY-Axis Status:")
print(f" Raw status bits: {format_status_bits(y_bits)}")
print(f" is_homed_y: {y_homed}")
# Check for errors
last_err = mc.last_error
if last_err:
print(f"\n** Last Error: {last_err}")
error_check = mc.check_for_errors()
if error_check:
print(f"** Error condition detected: {error_check}")
# Summary
print(f"\n--- Homing Status Summary ---")
if x_homed and y_homed:
print("PASS: Both axes are homed")
return True
else:
print("FAIL: One or both axes are NOT homed")
if not x_homed:
print(" -> X-axis needs homing")
if not y_homed:
print(" -> Y-axis needs homing")
return False
def test_square_pattern(mc: MotionController, side_length: float = 15.0) -> bool:
"""
Test 2: Move the stage in a square pattern.
Moves in a 15mm square starting from current position, outputting
position data at each step to detect hangs.
Pattern: Start -> +X -> +Y -> -X -> -Y (back to start)
Uses the new wait_until_settled() method for reliable motion detection.
"""
print("\n" + "="*70)
print(f"TEST 2: Square Pattern Movement ({side_length}mm)")
print("="*70)
print("Using wait_until_settled() for motion detection")
# Get starting position
mc.poll_positions()
time.sleep(0.1)
start_x = mc.position_x
start_y = mc.position_y
if start_x is None or start_y is None:
print("ERROR: Could not get initial position")
return False
print(f"\nStarting position: X={start_x:.4f}mm, Y={start_y:.4f}mm")
# Define the square corners (relative to start)
corners = [
(start_x + side_length, start_y), # Move +X
(start_x + side_length, start_y + side_length), # Move +Y
(start_x, start_y + side_length), # Move -X
(start_x, start_y), # Move -Y (back to start)
]
corner_names = ["+X edge", "+Y edge (diagonal)", "-X edge", "Return to start"]
all_passed = True
for i, (target_x, target_y) in enumerate(corners):
print(f"\n--- Move {i+1}/4: {corner_names[i]} ---")
print(f"Target: X={target_x:.4f}mm, Y={target_y:.4f}mm")
# Clear any previous error
mc.clear_last_error()
# Record start time
move_start = time.time()
# Issue move command (non-blocking)
mc.move_to_fast(x=target_x, y=target_y)
# Use poll_until_idle() in a loop to show progress
print(f"\n{'Time':>8} {'X_pos':>12} {'Y_pos':>12} {'X_err':>10} {'Y_err':>10} {'Pending'}")
print("-" * 70)
sample_count = 0
while not mc.poll_until_idle(tolerance=0.005, timeout=0.05):
elapsed = time.time() - move_start
curr_x = mc.position_x
curr_y = mc.position_y
if curr_x is not None and curr_y is not None:
x_err = curr_x - target_x
y_err = curr_y - target_y
# Print every 5th sample
if sample_count % 5 == 0:
pending = mc.get_pending_targets()
pending_str = ", ".join([f"{'X' if d==0x21 else 'Y'}" for d in pending.keys()])
print(f"{elapsed:>7.3f}s {curr_x:>12.4f} {curr_y:>12.4f} {x_err:>+10.4f} {y_err:>+10.4f} {pending_str:>8}")
sample_count += 1
# Timeout check
if elapsed > 30.0:
print(f"\n** TIMEOUT after 30s - stage may be hung!")
all_passed = False
mc.clear_pending_moves()
break
# Move complete - show final status
elapsed = time.time() - move_start
curr_x = mc.position_x
curr_y = mc.position_y
if curr_x is not None and curr_y is not None:
x_err = curr_x - target_x
y_err = curr_y - target_y
print(f"{elapsed:>7.3f}s {curr_x:>12.4f} {curr_y:>12.4f} {x_err:>+10.4f} {y_err:>+10.4f} ** DONE **")
print(f"\nMove completed in {elapsed:.3f}s")
print(f"Final position: X={curr_x:.4f}mm, Y={curr_y:.4f}mm")
print(f"Position error: X={abs(x_err)*1000:.1f}um, Y={abs(y_err)*1000:.1f}um")
# Check for errors after move
err = mc.check_for_errors()
if err:
print(f"** Error after move: {err}")
all_passed = False
last_err = mc.last_error
if last_err:
print(f"** Last error: {last_err}")
all_passed = False
# Short delay between moves
time.sleep(0.1)
# Final summary
print(f"\n--- Square Pattern Summary ---")
mc.poll_positions()
time.sleep(0.1)
final_x = mc.position_x
final_y = mc.position_y
if final_x is not None and final_y is not None:
total_x_err = abs(final_x - start_x)
total_y_err = abs(final_y - start_y)
print(f"Final position: X={final_x:.4f}mm, Y={final_y:.4f}mm")
print(f"Start position: X={start_x:.4f}mm, Y={start_y:.4f}mm")
print(f"Return error: X={total_x_err*1000:.1f}um, Y={total_y_err*1000:.1f}um")
if total_x_err > 0.05 or total_y_err > 0.05:
print("FAIL: Did not return to start position accurately (>50um)")
all_passed = False
if all_passed:
print("PASS: Square pattern completed successfully")
else:
print("FAIL: Issues detected during square pattern")
return all_passed
def print_diagnostic_info(mc: MotionController):
"""Print comprehensive diagnostic information about the controller."""
print("\n" + "="*70)
print("CONTROLLER DIAGNOSTIC INFO")
print("="*70)
try:
hw_info = mc.get_hw_info()
print(f"\nHardware Info:")
print(f" Serial Number: {hw_info['serial_number']}")
print(f" Model: {hw_info['model']}")
print(f" Firmware: {hw_info['firmware_version']}")
print(f" HW Version: {hw_info['hw_version']}")
print(f" Channels: {hw_info['num_channels']}")
print(f" Notes: {hw_info['notes']}")
except Exception as e:
print(f" Error getting hardware info: {e}")
# Request and display velocity params
print(f"\nVelocity Parameters:")
mc.send_command(0x0414, param1=0x01, dest=mc.DEST_X_AXIS) # REQ_VELPARAMS
mc.send_command(0x0414, param1=0x01, dest=mc.DEST_Y_AXIS)
time.sleep(0.1)
x_vel = mc.velocity_params_x
y_vel = mc.velocity_params_y
if x_vel:
print(f" X-Axis: max_vel={x_vel['max_velocity']:.2f}mm/s, accel={x_vel['acceleration']:.2f}mm/s²")
if y_vel:
print(f" Y-Axis: max_vel={y_vel['max_velocity']:.2f}mm/s, accel={y_vel['acceleration']:.2f}mm/s²")
def test_bay_and_channel_status(mc: MotionController) -> dict:
"""
Test bay occupancy and channel enable states.
Returns dict with diagnostic info about each axis.
"""
print("\n" + "="*70)
print("BAY AND CHANNEL DIAGNOSTICS")
print("="*70)
results = {'x': {}, 'y': {}}
# Query bay status using MGMSG_RACK_REQ_BAYUSED (0x0060)
print("\n--- Querying Bay Status (RACK_REQ_BAYUSED 0x0060) ---")
print("Sending to controller (0x11)...")
# Send bay request to controller
mc.send_command(0x0060, param1=0x00, param2=0x00, dest=0x11, source=0x01)
time.sleep(0.2)
# Check for response in queue
msgs = mc.get_all_messages()
bay_response = None
for msg in msgs:
print(f" Received: msg_id=0x{msg.msg_id:04X}, source=0x{msg.source:02X}, "
f"param1=0x{msg.param1:02X}, param2=0x{msg.param2:02X}, data={msg.data.hex() if msg.data else 'none'}")
if msg.msg_id == 0x0061: # RACK_GET_BAYUSED
bay_response = msg
if bay_response:
# param1 contains bay_ident (which bay), param2 contains state
print(f"\nBay status response: bay={bay_response.param1}, state=0x{bay_response.param2:02X}")
else:
print(" No RACK_GET_BAYUSED response received")
# Check channel enable state for X-axis (0x21)
print("\n--- X-Axis (Bay 1 / dest=0x21) Channel Status ---")
print(f" Destination address: 0x{mc.DEST_X_AXIS:02X}")
try:
x_enabled = mc.get_channel_enable_state(mc.DEST_X_AXIS, timeout=2.0)
print(f" Channel enabled: {x_enabled}")
results['x']['enabled'] = x_enabled
except Exception as e:
print(f" Error querying channel state: {e}")
results['x']['enabled'] = None
# Get X position to verify communication
x_pos = mc.get_position(mc.DEST_X_AXIS, timeout=2.0)
print(f" Current position: {x_pos:.4f}mm" if x_pos is not None else " Position query failed!")
results['x']['position'] = x_pos
results['x']['communicating'] = x_pos is not None
# Check channel enable state for Y-axis (0x22)
print("\n--- Y-Axis (Bay 2 / dest=0x22) Channel Status ---")
print(f" Destination address: 0x{mc.DEST_Y_AXIS:02X}")
try:
y_enabled = mc.get_channel_enable_state(mc.DEST_Y_AXIS, timeout=2.0)
print(f" Channel enabled: {y_enabled}")
results['y']['enabled'] = y_enabled
except Exception as e:
print(f" Error querying channel state: {e}")
results['y']['enabled'] = None
# Get Y position to verify communication
y_pos = mc.get_position(mc.DEST_Y_AXIS, timeout=2.0)
print(f" Current position: {y_pos:.4f}mm" if y_pos is not None else " Position query failed!")
results['y']['position'] = y_pos
results['y']['communicating'] = y_pos is not None
return results
def test_y_axis_move_verbose(mc: MotionController) -> bool:
"""
Test Y-axis movement using the new wait_until_settled() method.
Sends a small Y-axis move and monitors progress.
"""
print("\n" + "="*70)
print("Y-AXIS VERBOSE MOVE TEST")
print("="*70)
print("Using wait_until_settled() for motion detection")
# Get current Y position
mc.poll_positions()
time.sleep(0.1)
start_y = mc.position_y
if start_y is None:
print("ERROR: Cannot get Y position")
return False
print(f"\nCurrent Y position: {start_y:.4f}mm")
# Target: move 5mm in Y
target_y = start_y + 5.0
print(f"Target Y position: {target_y:.4f}mm")
# Issue move command using move_to_fast (which tracks pending moves)
print("\n--- Issuing Y-axis move via move_to_fast() ---")
mc.move_to_fast(y=target_y)
# Show pending moves
pending = mc.get_pending_targets()
print(f" Pending moves: {pending}")
print(f" is_move_pending(): {mc.is_move_pending()}")
# Monitor using poll_until_idle
print("\n--- Monitoring Y-axis with poll_until_idle() ---")
print(f"{'Time':>6} {'Y_pos':>10} {'Y_err':>10} {'Pending'}")
print("-" * 45)
start_time = time.time()
movement_detected = False
while not mc.poll_until_idle(tolerance=0.005, timeout=0.05):
elapsed = time.time() - start_time
curr_y = mc.position_y
if curr_y is not None:
y_err = curr_y - target_y
pending = "Y" if mc.is_move_pending() else "-"
print(f"{elapsed:>5.2f}s {curr_y:>10.4f} {y_err:>+10.4f} {pending:>8}")
if abs(curr_y - start_y) > 0.01:
movement_detected = True
if elapsed > 10.0:
print("\n** TIMEOUT after 10s **")
mc.clear_pending_moves()
break
# Final status
elapsed = time.time() - start_time
final_y = mc.position_y
print(f"\n--- Y-Axis Move Summary ---")
print(f"Start position: {start_y:.4f}mm")
print(f"Target position: {target_y:.4f}mm")
print(f"Final position: {final_y:.4f}mm" if final_y else "Final position: unknown")
print(f"Move time: {elapsed:.3f}s")
if final_y is not None:
distance_moved = abs(final_y - start_y)
print(f"Distance moved: {distance_moved:.4f}mm")
if distance_moved < 0.01:
print("\n** FAIL: Y-axis did not move at all! **")
print("Possible causes:")
print(" 1. Y-axis channel is disabled")
print(" 2. Y-axis motor driver issue")
print(" 3. Command not reaching Y-axis bay")
print(" 4. Mechanical obstruction")
return False
elif abs(final_y - target_y) < 0.05:
print("\n** PASS: Y-axis moved to target **")
return True
else:
print(f"\n** PARTIAL: Y-axis moved but not to target **")
return False
return movement_detected
def test_enable_and_move_y(mc: MotionController) -> bool:
"""
Explicitly enable Y-axis channel and attempt movement.
"""
print("\n" + "="*70)
print("ENABLE Y-AXIS AND MOVE TEST")
print("="*70)
# First, explicitly enable Y-axis channel
print("\n--- Enabling Y-axis channel (MOD_SET_CHANENABLESTATE) ---")
print(f" Sending to dest=0x{mc.DEST_Y_AXIS:02X}, enable=True")
try:
mc.set_channel_enable_state(mc.DEST_Y_AXIS, enabled=True)
time.sleep(0.2)
print(" Enable command sent")
except Exception as e:
print(f" Error sending enable: {e}")
# Verify it's enabled
try:
y_enabled = mc.get_channel_enable_state(mc.DEST_Y_AXIS, timeout=2.0)
print(f" Channel enabled state: {y_enabled}")
except Exception as e:
print(f" Error checking state: {e}")
# Request fresh status
mc.request_status_update(mc.DEST_Y_AXIS)
time.sleep(0.2)
y_bits = mc.status_bits_y
print(f" Status bits: {format_status_bits(y_bits)}")
# Now try the verbose move test
return test_y_axis_move_verbose(mc)
def main():
print("="*70)
print("BBD202 Stage Diagnostic Test")
print("="*70)
mc = None
try:
# Connect to controller
print("\nConnecting to BBD202 controller...")
mc = MotionController()
mc.connect(enable_updates=True)
print("Connected!")
# Print diagnostic info
print_diagnostic_info(mc)
# Run bay and channel diagnostics
channel_results = test_bay_and_channel_status(mc)
# Run homing status test
homed = test_homing_status(mc)
if not homed:
print("\n" + "-"*70)
response = input("Axes not homed. Home now? (y/n): ").strip().lower()
if response == 'y':
print("\nHoming X-axis...")
x_result = mc.home_x_axis(timeout=30.0)
print(f"X-axis homing: {'SUCCESS' if x_result else 'FAILED'}")
print("\nHoming Y-axis...")
y_result = mc.home_y_axis(timeout=30.0)
print(f"Y-axis homing: {'SUCCESS' if y_result else 'FAILED'}")
if not (x_result and y_result):
print("\nHoming failed. Cannot continue with movement test.")
return 1
# Re-check homing status
test_homing_status(mc)
else:
print("\nSkipping movement tests (axes not homed)")
return 1
# Run Y-axis verbose test first to diagnose the issue
y_axis_ok = test_enable_and_move_y(mc)
# If Y-axis failed, skip the square pattern
if not y_axis_ok:
print("\n** Y-axis movement failed - skipping square pattern test **")
square_ok = False
else:
# Run square pattern test
square_ok = test_square_pattern(mc, side_length=15.0)
# Final summary
print("\n" + "="*70)
print("DIAGNOSTIC SUMMARY")
print("="*70)
print(f"X-Axis Channel: {'OK' if channel_results['x'].get('communicating') else 'FAIL'}")
print(f"Y-Axis Channel: {'OK' if channel_results['y'].get('communicating') else 'FAIL'}")
print(f"Homing Status: {'PASS' if homed else 'FAIL'}")
print(f"Y-Axis Move: {'PASS' if y_axis_ok else 'FAIL'}")
print(f"Square Pattern: {'PASS' if square_ok else 'FAIL/SKIPPED'}")
if homed and y_axis_ok and square_ok:
print("\nAll tests PASSED")
return 0
else:
print("\nSome tests FAILED - see details above")
return 1
except KeyboardInterrupt:
print("\n\nTest interrupted by user")
return 1
except Exception as e:
print(f"\nError during test: {e}")
import traceback
traceback.print_exc()
return 1
finally:
if mc is not None:
print("\nDisconnecting from controller...")
mc.disconnect()
print("Disconnected")
if __name__ == "__main__":
sys.exit(main())
tests/test_bbd202_snake.py
-266
View File
@@ -1,266 +0,0 @@
"""
BBD202 Snake Test
Quick test to verify the BBD202 stage driver with a snake scan pattern.
Scans a 20mm x 20mm area with 2mm row spacing in a back-and-forth pattern.
"""
import sys
import time
from hardware.bbd202 import MotionController, MotorStatusBits
def snake_test(mc: MotionController, width: float = 20.0, height: float = 20.0, row_spacing: float = 2.0) -> bool:
"""
Execute a snake scan pattern.
Args:
mc: MotionController instance
width: Scan width in mm (X direction)
height: Scan height in mm (Y direction)
row_spacing: Distance between rows in mm
Pattern:
Start (0,0) -> Move +X to (width, 0) -> Move +Y by row_spacing ->
Move -X to (0, row_spacing) -> Move +Y by row_spacing ->
... repeat until height is covered
"""
print("\n" + "="*70)
print(f"SNAKE TEST: {width}mm x {height}mm area, {row_spacing}mm row spacing")
print("="*70)
# Get starting position
mc.poll_positions()
time.sleep(0.1)
start_x = mc.position_x
start_y = mc.position_y
if start_x is None or start_y is None:
print("ERROR: Could not get initial position")
return False
print(f"\nStarting position: X={start_x:.4f}mm, Y={start_y:.4f}mm")
# Calculate number of rows
num_rows = int(height / row_spacing) + 1
print(f"Number of rows: {num_rows}")
# Generate snake path waypoints
waypoints = []
for row in range(num_rows):
y_pos = start_y + row * row_spacing
if row % 2 == 0:
# Even row: move left to right
waypoints.append((start_x + width, y_pos, f"Row {row+1}/{num_rows} (+X)"))
else:
# Odd row: move right to left
waypoints.append((start_x, y_pos, f"Row {row+1}/{num_rows} (-X)"))
print(f"Total waypoints: {len(waypoints)}")
all_passed = True
total_distance = 0.0
total_time = 0.0
# Execute the snake pattern
for i, (target_x, target_y, description) in enumerate(waypoints):
print(f"\n--- Move {i+1}/{len(waypoints)}: {description} ---")
print(f"Target: X={target_x:.4f}mm, Y={target_y:.4f}mm")
# Record start time and position
move_start = time.time()
mc.poll_positions()
time.sleep(0.02)
curr_x = mc.position_x
curr_y = mc.position_y
if curr_x is not None and curr_y is not None:
distance = ((target_x - curr_x)**2 + (target_y - curr_y)**2)**0.5
total_distance += distance
# Clear any previous error
mc.clear_last_error()
# Issue move command
mc.move_to_fast(x=target_x, y=target_y)
# Wait for move to complete with periodic progress updates
sample_count = 0
max_err = 0.0
while not mc.poll_until_idle(tolerance=0.005, timeout=0.05):
elapsed = time.time() - move_start
curr_x = mc.position_x
curr_y = mc.position_y
if curr_x is not None and curr_y is not None:
x_err = curr_x - target_x
y_err = curr_y - target_y
total_err = (x_err**2 + y_err**2)**0.5
max_err = max(max_err, total_err)
# Print every 10th sample
if sample_count % 10 == 0:
print(f" {elapsed:>5.2f}s X={curr_x:>8.4f} Y={curr_y:>8.4f} err={total_err*1000:>6.1f}um", end='\r')
sample_count += 1
# Timeout check
if elapsed > 30.0:
print(f"\n** TIMEOUT after 30s - stage may be hung!")
all_passed = False
mc.clear_pending_moves()
break
# Move complete
elapsed = time.time() - move_start
total_time += elapsed
mc.poll_positions()
time.sleep(0.02)
final_x = mc.position_x
final_y = mc.position_y
if final_x is not None and final_y is not None:
x_err = final_x - target_x
y_err = final_y - target_y
total_err = (x_err**2 + y_err**2)**0.5
print(f" {elapsed:>5.2f}s X={final_x:>8.4f} Y={final_y:>8.4f} err={total_err*1000:>6.1f}um ** DONE **")
if total_err > 0.05: # 50um tolerance
print(f" ** WARNING: Position error exceeds 50um: {total_err*1000:.1f}um")
all_passed = False
# Check for errors after move
err = mc.check_for_errors()
if err:
print(f" ** Error after move: {err}")
all_passed = False
last_err = mc.last_error
if last_err:
print(f" ** Last error: {last_err}")
all_passed = False
# Return to start position
print(f"\n--- Returning to start position ---")
move_start = time.time()
mc.move_to_fast(x=start_x, y=start_y)
while not mc.poll_until_idle(tolerance=0.005, timeout=0.05):
if time.time() - move_start > 30.0:
print("** TIMEOUT returning to start")
mc.clear_pending_moves()
break
elapsed = time.time() - move_start
total_time += elapsed
# Final summary
print(f"\n{'='*70}")
print("SNAKE TEST SUMMARY")
print(f"{'='*70}")
mc.poll_positions()
time.sleep(0.1)
final_x = mc.position_x
final_y = mc.position_y
if final_x is not None and final_y is not None:
return_x_err = abs(final_x - start_x)
return_y_err = abs(final_y - start_y)
return_err = (return_x_err**2 + return_y_err**2)**0.5
print(f"Start position: X={start_x:.4f}mm, Y={start_y:.4f}mm")
print(f"Final position: X={final_x:.4f}mm, Y={final_y:.4f}mm")
print(f"Return error: {return_err*1000:.1f}um (X={return_x_err*1000:.1f}um, Y={return_y_err*1000:.1f}um)")
print(f"Total distance: {total_distance:.2f}mm")
print(f"Total time: {total_time:.2f}s")
print(f"Average speed: {total_distance/total_time:.2f}mm/s")
print(f"Waypoints: {len(waypoints)}")
if return_err > 0.05: # 50um tolerance
print(f"\n** FAIL: Did not return to start position (error: {return_err*1000:.1f}um > 50um)")
all_passed = False
if all_passed:
print(f"\n** PASS: Snake test completed successfully **")
else:
print(f"\n** FAIL: Issues detected during snake test **")
return all_passed
def main():
print("="*70)
print("BBD202 Snake Test")
print("="*70)
mc = None
try:
# Connect to controller
print("\nConnecting to BBD202 controller...")
mc = MotionController()
mc.connect(enable_updates=True)
print("Connected!")
# Brief hardware info
try:
hw_info = mc.get_hw_info()
print(f"\nHardware: {hw_info['model']} (S/N: {hw_info['serial_number']})")
print(f"Firmware: {hw_info['firmware_version']}")
except Exception as e:
print(f"Could not get hardware info: {e}")
# Check homing status
print("\n--- Checking Homing Status ---")
mc.request_status_update(mc.DEST_X_AXIS)
mc.request_status_update(mc.DEST_Y_AXIS)
time.sleep(0.2)
mc.poll_status()
time.sleep(0.1)
x_homed = mc.is_homed_x
y_homed = mc.is_homed_y
print(f"X-axis homed: {x_homed}")
print(f"Y-axis homed: {y_homed}")
if not (x_homed and y_homed):
print("\n** Axes not homed - will home now **")
print("\nHoming X-axis...")
x_result = mc.home_x_axis(timeout=30.0)
print(f"X-axis: {'SUCCESS' if x_result else 'FAILED'}")
print("\nHoming Y-axis...")
y_result = mc.home_y_axis(timeout=30.0)
print(f"Y-axis: {'SUCCESS' if y_result else 'FAILED'}")
if not (x_result and y_result):
print("\n** Homing failed. Cannot continue. **")
return 1
# Run the snake test
success = snake_test(mc, width=20.0, height=20.0, row_spacing=2.0)
return 0 if success else 1
except KeyboardInterrupt:
print("\n\nTest interrupted by user")
return 1
except Exception as e:
print(f"\n** ERROR: {e}")
import traceback
traceback.print_exc()
return 1
finally:
if mc is not None:
print("\nDisconnecting from controller...")
mc.disconnect()
print("Disconnected")
if __name__ == "__main__":
sys.exit(main())
tests/test_bbd202_snake_20x20.py
-300
View File
@@ -1,300 +0,0 @@
"""
BBD202 Snake Scan Test - 20x20mm Area
Scans a 20mm x 20mm area centered at X=55, Y=37.5 in a snake pattern.
Row spacing: 0.5mm
Movement order: X+, Y+, X-, Y+ (snake pattern)
"""
import sys
import time
import argparse
from hardware.bbd202 import MotionController, MotorStatusBits
# Scan parameters
CENTER_X = 55.0 # mm
CENTER_Y = 37.5 # mm
SCAN_WIDTH = 20.0 # mm (X direction)
SCAN_HEIGHT = 20.0 # mm (Y direction)
ROW_SPACING = 0.5 # mm (Y step between rows)
# Calculated bounds
START_X = CENTER_X - SCAN_WIDTH / 2 # 45.0 mm
END_X = CENTER_X + SCAN_WIDTH / 2 # 65.0 mm
START_Y = CENTER_Y - SCAN_HEIGHT / 2 # 27.5 mm
END_Y = CENTER_Y + SCAN_HEIGHT / 2 # 47.5 mm
def snake_scan(mc: MotionController) -> bool:
"""
Execute a snake scan pattern over the 20x20mm area.
Pattern (X+, Y+, X-, Y+):
Row 0: X=45 -> X=65 (X+)
Step: Y += 0.5 (Y+)
Row 1: X=65 -> X=45 (X-)
Step: Y += 0.5 (Y+)
... repeat until Y reaches 47.5mm
"""
print("\n" + "=" * 70)
print("SNAKE SCAN TEST: 20mm x 20mm")
print("=" * 70)
print(f"Center: X={CENTER_X:.1f}mm, Y={CENTER_Y:.1f}mm")
print(f"Bounds: X=[{START_X:.1f}, {END_X:.1f}]mm, Y=[{START_Y:.1f}, {END_Y:.1f}]mm")
print(f"Row spacing: {ROW_SPACING}mm")
print(f"Pattern: X+, Y+, X-, Y+ (snake)")
# Calculate number of rows
num_rows = int(SCAN_HEIGHT / ROW_SPACING) + 1
print(f"Total rows: {num_rows}")
# Stop automatic ACK to prevent interference with move commands
# We'll send manual ACKs after each move completes
mc.stop_status_ack()
print("Using manual ACK mode for scan")
# First, move to the starting position
print(f"\n--- Moving to start position ({START_X:.1f}, {START_Y:.1f}) ---")
mc.move_to_fast(x=START_X, y=START_Y)
move_start = time.time()
while not mc.poll_until_idle(tolerance=0.005, timeout=0.05):
if time.time() - move_start > 30.0:
print("** TIMEOUT moving to start position!")
return False
# Verify starting position
mc.poll_positions()
time.sleep(0.1)
curr_x = mc.position_x
curr_y = mc.position_y
print(f"At start: X={curr_x:.4f}mm, Y={curr_y:.4f}mm")
# Generate snake path waypoints
waypoints = []
for row in range(num_rows):
y_pos = START_Y + row * ROW_SPACING
if row % 2 == 0:
# Even row: move X+ (left to right)
waypoints.append((END_X, y_pos, f"Row {row + 1}/{num_rows} X+ (Y={y_pos:.1f})"))
else:
# Odd row: move X- (right to left)
waypoints.append((START_X, y_pos, f"Row {row + 1}/{num_rows} X- (Y={y_pos:.1f})"))
print(f"\nTotal waypoints: {len(waypoints)}")
print("-" * 70)
all_passed = True
total_distance = 0.0
total_time = 0.0
scan_start_time = time.time()
# Execute the snake pattern
for i, (target_x, target_y, description) in enumerate(waypoints):
# Record start position for distance calculation
mc.poll_positions()
time.sleep(0.02)
prev_x = mc.position_x
prev_y = mc.position_y
if prev_x is not None and prev_y is not None:
distance = ((target_x - prev_x) ** 2 + (target_y - prev_y) ** 2) ** 0.5
total_distance += distance
# Clear any previous error
mc.clear_last_error()
# Issue move command
move_start = time.time()
mc.move_to_fast(x=target_x, y=target_y)
# Wait for move to complete
while not mc.poll_until_idle(tolerance=0.005, timeout=0.05):
elapsed = time.time() - move_start
# Timeout check
if elapsed > 30.0:
print(f"\n** TIMEOUT on {description} after 30s!")
all_passed = False
mc.clear_pending_moves()
break
# Move complete - send ACK to keep controller responsive
mc.ack_status_update()
elapsed = time.time() - move_start
total_time += elapsed
mc.poll_positions()
time.sleep(0.02)
final_x = mc.position_x
final_y = mc.position_y
if final_x is not None and final_y is not None:
x_err = final_x - target_x
y_err = final_y - target_y
total_err = (x_err ** 2 + y_err ** 2) ** 0.5
# Print progress (compact format)
status = "OK" if total_err < 0.05 else "ERR"
print(f" {description:35} -> X={final_x:7.3f} Y={final_y:7.3f} err={total_err * 1000:5.1f}um [{status}]")
if total_err > 0.05: # 50um tolerance
all_passed = False
# Check for errors after move
err = mc.check_for_errors()
if err:
print(f" ** Error: {err}")
all_passed = False
last_err = mc.last_error
if last_err:
print(f" ** Last error: {last_err}")
all_passed = False
scan_elapsed = time.time() - scan_start_time
# Return to start position
print(f"\n--- Returning to start position ({START_X:.1f}, {START_Y:.1f}) ---")
move_start = time.time()
mc.move_to_fast(x=START_X, y=START_Y)
while not mc.poll_until_idle(tolerance=0.005, timeout=0.05):
if time.time() - move_start > 30.0:
print("** TIMEOUT returning to start")
mc.clear_pending_moves()
break
# Final summary
print(f"\n{'=' * 70}")
print("SNAKE SCAN SUMMARY")
print(f"{'=' * 70}")
mc.poll_positions()
time.sleep(0.1)
final_x = mc.position_x
final_y = mc.position_y
if final_x is not None and final_y is not None:
return_x_err = abs(final_x - START_X)
return_y_err = abs(final_y - START_Y)
return_err = (return_x_err ** 2 + return_y_err ** 2) ** 0.5
print(f"Scan area: {SCAN_WIDTH:.1f}mm x {SCAN_HEIGHT:.1f}mm")
print(f"Center: X={CENTER_X:.1f}mm, Y={CENTER_Y:.1f}mm")
print(f"Row spacing: {ROW_SPACING}mm")
print(f"Rows completed: {num_rows}")
print(f"Total distance: {total_distance:.2f}mm")
print(f"Scan time: {scan_elapsed:.2f}s")
print(f"Average speed: {total_distance / scan_elapsed:.2f}mm/s")
print(f"Return error: {return_err * 1000:.1f}um (X={return_x_err * 1000:.1f}um, Y={return_y_err * 1000:.1f}um)")
if return_err > 0.05:
print(f"\n** FAIL: Return position error exceeds 50um")
all_passed = False
# Restart automatic ACK
mc.start_status_ack()
if all_passed:
print(f"\n** PASS: Snake scan completed successfully **")
else:
print(f"\n** FAIL: Issues detected during snake scan **")
return all_passed
def main():
parser = argparse.ArgumentParser(description="BBD202 Snake Scan Test - 20x20mm Area")
parser.add_argument("--auto-home", action="store_true",
help="Automatically home axes if not homed (no prompt)")
args = parser.parse_args()
print("=" * 70)
print("BBD202 Snake Scan Test - 20x20mm Area")
print("=" * 70)
mc = None
try:
# Connect to controller
print("\nConnecting to BBD202 controller...")
mc = MotionController()
mc.connect(enable_updates=True)
print("Connected!")
# Brief hardware info
try:
hw_info = mc.get_hw_info()
print(f"\nHardware: {hw_info['model']} (S/N: {hw_info['serial_number']})")
print(f"Firmware: {hw_info['firmware_version']}")
except Exception as e:
print(f"Could not get hardware info: {e}")
# Check homing status
print("\n--- Checking Homing Status ---")
mc.request_status_update(mc.DEST_X_AXIS)
mc.request_status_update(mc.DEST_Y_AXIS)
time.sleep(0.2)
mc.poll_status()
time.sleep(0.1)
x_homed = mc.is_homed_x
y_homed = mc.is_homed_y
print(f"X-axis homed: {x_homed}")
print(f"Y-axis homed: {y_homed}")
if not (x_homed and y_homed):
print("\n** Axes not homed! **")
# Check if we should auto-home or prompt
should_home = args.auto_home
if not should_home:
try:
response = input("Home axes now? (y/n): ").strip().lower()
should_home = response == 'y'
except EOFError:
print("Non-interactive mode detected. Use --auto-home flag.")
return 1
if should_home:
print("\nHoming X-axis...")
x_result = mc.home_x_axis(timeout=30.0)
print(f"X-axis: {'SUCCESS' if x_result else 'FAILED'}")
print("\nHoming Y-axis...")
y_result = mc.home_y_axis(timeout=30.0)
print(f"Y-axis: {'SUCCESS' if y_result else 'FAILED'}")
if not (x_result and y_result):
print("\n** Homing failed. Cannot continue. **")
return 1
else:
print("\nCannot run snake scan without homing. Exiting.")
return 1
# Run the snake scan
success = snake_scan(mc)
return 0 if success else 1
except KeyboardInterrupt:
print("\n\nTest interrupted by user")
return 1
except Exception as e:
print(f"\n** ERROR: {e}")
import traceback
traceback.print_exc()
return 1
finally:
if mc is not None:
print("\nDisconnecting from controller...")
mc.disconnect()
print("Disconnected")
if __name__ == "__main__":
sys.exit(main())
tests/test_camera_integration.py
-126
View File
@@ -1,126 +0,0 @@
#!/usr/bin/env python3
"""
Test script for camera integration with the scan wizard.
Tests the camera driver and integration with the UI.
"""
import sys
from PyQt6 import QtWidgets
def test_camera_import():
"""Test that the camera driver can be imported"""
print("Testing camera driver import...")
try:
from uc480_camera import UC480Camera, CameraStreamThread
print("✓ Camera driver imported successfully")
return True
except ImportError as e:
print(f"✗ Failed to import camera driver: {e}")
return False
def test_camera_class():
"""Test that the camera class can be instantiated"""
print("\nTesting camera class instantiation...")
try:
from uc480_camera import UC480Camera
camera = UC480Camera(camera_id=0)
print("✓ Camera class instantiated successfully")
print(f" Camera handle: {camera.h_cam}")
print(f" Initialized: {camera.is_initialized}")
return True
except Exception as e:
print(f"✗ Failed to instantiate camera class: {e}")
return False
def test_scanengine_import():
"""Test that the scanengine app can be imported with camera integration"""
print("\nTesting scanengine app import...")
try:
from scanengine_app import NewScanWizard, MainLauncher
print("✓ Scanengine app imported successfully")
return True
except ImportError as e:
print(f"✗ Failed to import scanengine app: {e}")
return False
def test_wizard_with_camera():
"""Test that the wizard can be created with camera integration"""
print("\nTesting wizard with camera integration...")
try:
app = QtWidgets.QApplication(sys.argv)
from scanengine_app import NewScanWizard
wizard = NewScanWizard()
print("✓ Wizard created successfully")
print(f" Camera object: {wizard.camera}")
print(f" Camera stream thread: {wizard.camera_stream_thread}")
print(f" CCD scene: {wizard.ccd_scene}")
# Check if camera methods exist
assert hasattr(wizard, 'initialize_camera'), "Missing initialize_camera method"
assert hasattr(wizard, 'start_camera_stream'), "Missing start_camera_stream method"
assert hasattr(wizard, 'stop_camera_stream'), "Missing stop_camera_stream method"
assert hasattr(wizard, 'cleanup_camera'), "Missing cleanup_camera method"
print("✓ All camera methods present")
# Test page change triggers
print("\nTesting page change behavior...")
current_page = wizard.stackedWidget.currentIndex()
print(f" Current page: {current_page}")
# Simulate page navigation to focus page (index 1)
print(" Navigating to focus page (index 1)...")
wizard.stackedWidget.setCurrentIndex(1)
print(f" Current page after navigation: {wizard.stackedWidget.currentIndex()}")
# Navigate back to first page
print(" Navigating back to page 0...")
wizard.stackedWidget.setCurrentIndex(0)
print(f" Current page after navigation: {wizard.stackedWidget.currentIndex()}")
# Cleanup
wizard.cleanup_camera()
print("✓ Camera cleanup successful")
return True
except Exception as e:
print(f"✗ Failed to test wizard with camera: {e}")
import traceback
traceback.print_exc()
return False
def main():
"""Run all tests"""
print("=" * 60)
print("Camera Integration Test Suite")
print("=" * 60)
results = []
# Run tests
results.append(("Camera Import", test_camera_import()))
results.append(("Camera Class", test_camera_class()))
results.append(("Scanengine Import", test_scanengine_import()))
results.append(("Wizard Integration", test_wizard_with_camera()))
# Print summary
print("\n" + "=" * 60)
print("Test Summary")
print("=" * 60)
for test_name, passed in results:
status = "PASS" if passed else "FAIL"
symbol = "✓" if passed else "✗"
print(f"{symbol} {test_name}: {status}")
all_passed = all(result[1] for result in results)
print("\n" + "=" * 60)
if all_passed:
print("All tests passed!")
else:
print("Some tests failed.")
print("=" * 60)
return 0 if all_passed else 1
if __name__ == "__main__":
sys.exit(main())
tests/test_genesis_connection.py
-196
View File
@@ -1,196 +0,0 @@
#!/opt/srasenv/bin/python3
"""
Test script for Genesis laser serial communication.
Tests basic connectivity and I2C protocol without GUI.
"""
import sys
import time
import serial
# Constants
NXP_START_BYTE = 0x53
NXP_STOP_BYTE = 0x50
ADDR_PCA9555_PS_GLUE_OUT = 0x4a
ADDR_ADS7828 = 0x90
CHAN_CURRENT_ACTUAL = 0x84
def test_serial_connection(port_name="/dev/ttyUSB1", baudrate=9600):
"""Test basic serial connection."""
print(f"Testing serial connection to {port_name} @ {baudrate}...")
try:
port = serial.Serial(
port=port_name,
baudrate=baudrate,
bytesize=serial.EIGHTBITS,
parity=serial.PARITY_NONE,
stopbits=serial.STOPBITS_ONE,
timeout=1.0
)
print("✓ Serial port opened successfully")
return port
except Exception as e:
print(f"✗ Failed to open serial port: {e}")
return None
def nxp_write(port, i2c_addr_write, cmd, data, data_len):
"""Build and send NXP I2C write packet."""
# Determine command length
if cmd <= 0xFF:
cmd_bytes = bytes([cmd])
else:
cmd_bytes = cmd.to_bytes(2, 'big')
# Convert data to bytes (big-endian)
data_bytes = data.to_bytes(data_len, 'big')
# Calculate total length
total_len = len(cmd_bytes) + len(data_bytes)
# Build packet: [0x53][addr][len][cmd...][data...][0x50]
packet = bytes([
NXP_START_BYTE,
i2c_addr_write,
total_len
]) + cmd_bytes + data_bytes + bytes([NXP_STOP_BYTE])
print(f" TX: {packet.hex(' ')}")
port.write(packet)
return True
def nxp_read(port, i2c_addr_write, cmd, cmd_len, data_len):
"""Build and send NXP I2C read packet."""
# Convert command to bytes
if cmd_len == 1:
cmd_bytes = bytes([cmd])
else:
cmd_bytes = cmd.to_bytes(2, 'big')
# Build packet: [0x53][addr][cmd_len][cmd...][0x53][addr|0x01][data_len][0x50]
packet = bytes([
NXP_START_BYTE,
i2c_addr_write,
cmd_len
]) + cmd_bytes + bytes([
NXP_START_BYTE,
i2c_addr_write | 0x01, # Read address
data_len,
NXP_STOP_BYTE
])
print(f" TX: {packet.hex(' ')}")
port.write(packet)
time.sleep(0.1)
# Read response
response = port.read(data_len)
if response:
print(f" RX: {response.hex(' ')} ({len(response)} bytes)")
return int.from_bytes(response, 'big')
else:
print(f" RX: (no data)")
return None
def test_read_ps_glue_out(port):
"""Test reading PS glue output port."""
print("\nTest 1: Reading PS glue output port (0x4a, reg 0x02)...")
value = nxp_read(port, ADDR_PCA9555_PS_GLUE_OUT, 0x02, 1, 1)
if value is not None:
print(f"✓ PS Glue Out status: 0x{value:02x}")
print(f" Shutter: {'OPEN' if value & 0x01 else 'CLOSED'}")
print(f" Current Mode: {'ON' if value & 0x04 else 'OFF'}")
print(f" Remote Enable: {'ON' if value & 0x08 else 'OFF'}")
print(f" Analog Enable: {'ON' if value & 0x10 else 'OFF'}")
print(f" Keyswitch: {'ON' if value & 0x20 else 'OFF'}")
return True
else:
print("✗ Failed to read PS glue out")
return False
def test_read_current_adc(port):
"""Test reading current ADC."""
print("\nTest 2: Reading current ADC (ADS7828, channel 0x84)...")
value = nxp_read(port, ADDR_ADS7828, CHAN_CURRENT_ACTUAL, 1, 2)
if value is not None:
print(f"✓ Current ADC raw value: 0x{value:04x} ({value})")
scaled = value * 0.000244140625
print(f" Scaled value: {scaled:.6f}")
return True
else:
print("✗ Failed to read current ADC")
return False
def test_write_current_zero(port):
"""Test setting current to zero."""
print("\nTest 3: Setting current to 0 (X9119 @ 0x52, cmd 0xa0)...")
success = nxp_write(port, 0x52, 0xa0, 0, 2)
if success:
print("✓ Current set to 0 command sent")
return True
else:
print("✗ Failed to set current")
return False
def test_read_status_bits(port):
"""Test reading all status bits from PS glue output."""
print("\nTest 4: Reading all control status bits...")
# Read current state
current_state = nxp_read(port, ADDR_PCA9555_PS_GLUE_OUT, 0x02, 1, 1)
if current_state is None:
print("✗ Failed to read current state")
return False
print(f" Current state: 0x{current_state:02x}")
print(f" Note: Shutter bit status: {'SET' if current_state & 0x01 else 'CLEAR'} (manual shutter - not controlled)")
print("✓ Successfully read all status bits")
return True
def main():
"""Run all tests."""
print("=" * 60)
print("Genesis SLM MX 532 - Serial Communication Test")
print("=" * 60)
# Open serial port
port = test_serial_connection()
if not port:
print("\nTest FAILED: Cannot open serial port")
return 1
try:
# Run tests
tests_passed = 0
tests_total = 4
if test_read_ps_glue_out(port):
tests_passed += 1
if test_read_current_adc(port):
tests_passed += 1
if test_write_current_zero(port):
tests_passed += 1
if test_read_status_bits(port):
tests_passed += 1
# Summary
print("\n" + "=" * 60)
print(f"Test Summary: {tests_passed}/{tests_total} tests passed")
print("=" * 60)
if tests_passed == tests_total:
print("✓ All tests PASSED - Communication working!")
print("\nYou can now run the GUI application:")
print(" ./genesis_laser_gui.py")
return 0
else:
print("✗ Some tests FAILED - Check connections")
return 1
finally:
port.close()
print("\nSerial port closed.")
if __name__ == '__main__':
sys.exit(main())
tests/test_genesis_protocol.py
-237
View File
@@ -1,237 +0,0 @@
#!/usr/bin/env python3
"""
Genesis Laser Protocol Test Script
===================================
Simple command-line script to test NXP I2C-over-serial communication
with the Genesis SLM MX 532 laser.
Usage:
python test_genesis_protocol.py [port] [baudrate]
Example:
python test_genesis_protocol.py /dev/ttyUSB0 9600
"""
import sys
import time
import serial
from typing import Optional
class NXPProtocolTester:
"""Simple NXP I2C protocol tester"""
NXP_START = 0x53
NXP_STOP = 0x50
def __init__(self, port: str = "/dev/ttyUSB0", baudrate: int = 9600):
self.port = serial.Serial(
port=port,
baudrate=baudrate,
bytesize=serial.EIGHTBITS,
parity=serial.PARITY_NONE,
stopbits=serial.STOPBITS_ONE,
timeout=1.0
)
time.sleep(0.1)
print(f"Connected to {port} at {baudrate} baud")
def send_packet(self, i2c_addr: int, cmd: bytes, data: bytes = b''):
"""Send NXP I2C write packet"""
if not isinstance(cmd, bytes):
cmd = bytes([cmd])
if not isinstance(data, bytes):
data = bytes(data)
length = len(cmd) + len(data)
packet = bytes([self.NXP_START, i2c_addr, length]) + cmd + data + bytes([self.NXP_STOP])
print(f"TX: {' '.join(f'{b:02x}' for b in packet)}")
self.port.write(packet)
def read_packet(self, i2c_addr_write: int, cmd: bytes, cmd_len: int, data_len: int) -> Optional[bytes]:
"""Send NXP I2C read packet and return data"""
if not isinstance(cmd, bytes):
cmd = bytes([cmd])
i2c_addr_read = i2c_addr_write | 0x01
packet_write = bytes([self.NXP_START, i2c_addr_write, cmd_len]) + cmd
packet_read = bytes([self.NXP_START, i2c_addr_read, data_len, self.NXP_STOP])
packet = packet_write + packet_read
print(f"TX: {' '.join(f'{b:02x}' for b in packet)}")
self.port.write(packet)
time.sleep(0.05)
response = self.port.read(data_len)
if response:
print(f"RX: {' '.join(f'{b:02x}' for b in response)}")
return response
else:
print("RX: (no response)")
return None
def test_x9119_current(self, value: int = 100):
"""Test X9119 current control"""
print(f"\n=== Testing X9119 Current Control (value={value}) ===")
# X9119 current control at 0x52
# Command 0xa0, 2 bytes data (10-bit value)
value = max(0, min(1023, value))
msb = (value >> 8) & 0x03
lsb = value & 0xFF
self.send_packet(0x52, bytes([0xa0]), bytes([msb, lsb]))
print("Current command sent")
def test_pca9555_shutter(self, state: bool):
"""Test PCA9555 shutter control"""
print(f"\n=== Testing PCA9555 Shutter Control (state={'OPEN' if state else 'CLOSED'}) ===")
# Read current output port 0 value
print("Reading current port value...")
current = self.read_packet(0x4a, bytes([0x02]), 1, 1)
if current and len(current) == 1:
current_value = current[0]
print(f"Current port value: 0x{current_value:02x}")
# Modify bit 0 (shutter)
if state:
new_value = current_value | 0x01
else:
new_value = current_value & ~0x01
print(f"New port value: 0x{new_value:02x}")
# Write back
self.send_packet(0x4a, bytes([0x02]), bytes([new_value]))
print("Shutter command sent")
else:
print("Failed to read current port value")
def test_ads7828_current_reading(self):
"""Test ADS7828 current reading"""
print(f"\n=== Testing ADS7828 Current Reading ===")
# ADS7828 at 0x90/0x91
# Command byte: 0x80 (single-ended) | (channel << 4) | 0x0c (internal ref)
# Reading channel 0
cmd_byte = 0x80 | (0 << 4) | 0x0c
data = self.read_packet(0x90, bytes([cmd_byte]), 1, 2)
if data and len(data) == 2:
value = (data[0] << 8) | data[1]
value = (value >> 4) & 0x0FFF
print(f"Current reading: {value} counts (0x{value:03x})")
else:
print("Failed to read current")
def test_pca9555_read_ports(self):
"""Test reading all PCA9555 I/O expander ports"""
print(f"\n=== Testing PCA9555 Port Reads ===")
devices = [
(0x4a, "Main DIO (0x14a)"),
(0x48, "PS Glue (0x148)"),
(0x44, "Head DIO (0x144)"),
(0x40, "LDD Control (0x140)"),
]
for addr, name in devices:
print(f"\n{name}:")
for port in range(8):
data = self.read_packet(addr, bytes([port]), 1, 1)
if data and len(data) == 1:
print(f" Register 0x{port:02x}: 0x{data[0]:02x} (0b{data[0]:08b})")
else:
print(f" Register 0x{port:02x}: read failed")
time.sleep(0.05)
def close(self):
"""Close serial port"""
self.port.close()
print("\nConnection closed")
def main():
"""Main test function"""
# Parse command line arguments
port = sys.argv[1] if len(sys.argv) > 1 else "/dev/ttyUSB0"
baudrate = int(sys.argv[2]) if len(sys.argv) > 2 else 9600
try:
# Create tester
tester = NXPProtocolTester(port, baudrate)
# Menu
while True:
print("\n" + "="*60)
print("Genesis Laser Protocol Test Menu")
print("="*60)
print("1. Test X9119 current control (set to 100)")
print("2. Test PCA9555 shutter OPEN")
print("3. Test PCA9555 shutter CLOSE")
print("4. Test ADS7828 current reading")
print("5. Test read all PCA9555 ports")
print("6. Set current to 0 (safe state)")
print("7. Emergency stop (shutter close + current 0)")
print("8. Custom X9119 current value")
print("0. Exit")
print("="*60)
choice = input("Enter choice: ").strip()
if choice == "1":
tester.test_x9119_current(100)
elif choice == "2":
tester.test_pca9555_shutter(True)
elif choice == "3":
tester.test_pca9555_shutter(False)
elif choice == "4":
tester.test_ads7828_current_reading()
elif choice == "5":
tester.test_pca9555_read_ports()
elif choice == "6":
print("\n=== Setting current to 0 ===")
tester.test_x9119_current(0)
elif choice == "7":
print("\n=== EMERGENCY STOP ===")
tester.test_pca9555_shutter(False)
time.sleep(0.1)
tester.test_x9119_current(0)
print("Emergency stop complete")
elif choice == "8":
try:
value = int(input("Enter current value (0-1023): "))
tester.test_x9119_current(value)
except ValueError:
print("Invalid value")
elif choice == "0":
break
else:
print("Invalid choice")
tester.close()
except Exception as e:
print(f"Error: {e}")
import traceback
traceback.print_exc()
if __name__ == "__main__":
main()
tests/test_rotated_aoi.py
-69
View File
@@ -1,69 +0,0 @@
#!/usr/bin/env python3
"""
Test script for the new rotated AoI bounding box approach.
"""
from decimal import Decimal
from scanning.sc3_scan_model import SC3ScanModel
def test_rotated_aoi():
"""Test the rotated AoI bounding box implementation."""
# Create scan model
model = SC3ScanModel()
# Configure scan parameters
model.x_origin = Decimal('50.0')
model.y_origin = Decimal('35.0')
model.x_delta = Decimal('10.0')
model.y_delta = Decimal('5.0')
model.row_spacing = Decimal('1.0')
model.laser_frequency = Decimal('2000.0')
model.scan_velocity = Decimal('100.0')
model.scan_angles = 4 # 0, 45, 90, 135 degrees
print("Scan Model Configuration:")
print(f" AoI Origin: ({model.x_origin}, {model.y_origin})")
print(f" AoI Size: {model.x_delta} x {model.y_delta}")
print(f" Row Spacing: {model.row_spacing}")
print(f" Optical Origin: ({model.optical_x_origin}, {model.optical_y_origin})")
print(f" Scan Angles: {model.get_angle_list()}")
print(f" Rows Required: {model.rows_required}")
print()
# Verify zero-angle scan (should be horizontal lines)
print("Zero-Angle Scan (first 3 lines):")
for i, coords in enumerate(model.scan_coordinates[:3]):
print(f" Line {i}: ({coords[0]}, {coords[1]}) -> ({coords[2]}, {coords[3]})")
print()
# Verify rotated scans
print("Rotated Scans (first line of each angle):")
for angle_idx, angle_deg in enumerate(model.get_angle_list()):
if angle_idx < len(model.rotated_coordinates):
rotated_scan = model.rotated_coordinates[angle_idx]
if rotated_scan:
coords = rotated_scan[0]
print(f" Angle {angle_deg}°: ({coords[0]:.3f}, {coords[1]:.3f}) -> ({coords[2]:.3f}, {coords[3]:.3f})")
print(f" Number of lines: {len(rotated_scan)}")
print()
# Verify that scans are horizontal (+x direction)
print("Verifying horizontal scan direction:")
for angle_idx, angle_deg in enumerate(model.get_angle_list()):
if angle_idx < len(model.rotated_coordinates):
rotated_scan = model.rotated_coordinates[angle_idx]
all_horizontal = True
for coords in rotated_scan:
# Check if y_start == y_end (horizontal line)
if abs(coords[1] - coords[3]) > 1e-10:
all_horizontal = False
break
status = "✓" if all_horizontal else "✗"
print(f" Angle {angle_deg}°: {status} All lines horizontal")
print()
print("Test completed successfully!")
if __name__ == "__main__":
test_rotated_aoi()
tests/test_status_updates.py
-172
View File
@@ -1,172 +0,0 @@
"""
Test script to examine raw status update packets from the BBD202 controller.
This script:
1. Connects to the controller at a low level
2. Sends HW_START_UPDATEMSGS to enable automatic status updates
3. Captures and displays raw packets to verify the format matches our parsing
"""
import sys
import time
import struct
from hardware.bbd202 import MotionController, MsgId
def hexdump(data: bytes, prefix: str = "") -> str:
"""Format bytes as hex dump."""
hex_str = " ".join(f"{b:02X}" for b in data)
ascii_str = "".join(chr(b) if 32 <= b < 127 else "." for b in data)
return f"{prefix}{hex_str} |{ascii_str}|"
def decode_status_update(data: bytes) -> dict:
"""Decode a MOT_GET_USTATUSUPDATE message data payload."""
if len(data) < 14:
return {"error": f"Data too short: {len(data)} bytes, expected 14"}
chan_ident = struct.unpack('<H', data[0:2])[0]
position_counts = struct.unpack('<i', data[2:6])[0]
velocity_counts = struct.unpack('<H', data[6:8])[0]
motor_current = struct.unpack('<H', data[8:10])[0]
status_bits = struct.unpack('<I', data[10:14])[0]
# Convert to physical units
ENCODER_COUNTS_PER_MM = 20000
position_mm = position_counts / ENCODER_COUNTS_PER_MM
return {
"chan_ident": chan_ident,
"position_counts": position_counts,
"position_mm": position_mm,
"velocity_counts": velocity_counts,
"motor_current": motor_current,
"status_bits": status_bits,
"status_hex": f"0x{status_bits:08X}"
}
def main():
print("=" * 70)
print("BBD202 Status Update Packet Test")
print("=" * 70)
mc = None
try:
# Connect to controller WITHOUT enabling updates yet
print("\nConnecting to BBD202 controller...")
mc = MotionController()
mc.connect(enable_updates=False) # Don't auto-enable updates
print("Connected!")
# Get hardware info
try:
hw_info = mc.get_hw_info()
print(f"Hardware: {hw_info['model']} (S/N: {hw_info['serial_number']})")
print(f"Firmware: {hw_info['firmware_version']}")
except Exception as e:
print(f"Could not get hardware info: {e}")
# Clear any pending messages
print("\nClearing RX queue...")
msgs = mc.get_all_messages()
print(f"Cleared {len(msgs)} pending messages")
# Now enable status updates
print("\n" + "-" * 70)
print("Sending HW_START_UPDATEMSGS to enable automatic status updates...")
print("-" * 70)
mc.start_update_messages()
# Wait a moment for updates to start arriving
time.sleep(0.5)
# Collect messages for a few seconds
print("\nCollecting status update packets for 3 seconds...")
print("(Looking for MOT_GET_USTATUSUPDATE = 0x0491)")
print()
start_time = time.time()
update_count = 0
other_count = 0
while time.time() - start_time < 3.0:
msg = mc.get_message(timeout=0.1)
if msg:
elapsed = time.time() - start_time
if msg.msg_id == MsgId.MOT_GET_USTATUSUPDATE:
update_count += 1
source_name = "X-axis" if msg.source == 0x21 else "Y-axis" if msg.source == 0x22 else f"0x{msg.source:02X}"
print(f"[{elapsed:5.2f}s] MOT_GET_USTATUSUPDATE from {source_name}")
print(f" Raw ({len(msg.raw)} bytes): {hexdump(msg.raw)}")
print(f" Data ({len(msg.data)} bytes): {hexdump(msg.data)}")
decoded = decode_status_update(msg.data)
if "error" in decoded:
print(f" DECODE ERROR: {decoded['error']}")
else:
print(f" Decoded: chan={decoded['chan_ident']}, "
f"pos={decoded['position_mm']:.4f}mm ({decoded['position_counts']} counts), "
f"vel={decoded['velocity_counts']}, cur={decoded['motor_current']}, "
f"status={decoded['status_hex']}")
print()
else:
other_count += 1
msg_name = MsgId(msg.msg_id).name if msg.msg_id in [m.value for m in MsgId] else f"0x{msg.msg_id:04X}"
print(f"[{elapsed:5.2f}s] Other message: {msg_name} from 0x{msg.source:02X}")
print(f" Raw: {hexdump(msg.raw)}")
print()
print("-" * 70)
print(f"Summary: Received {update_count} status updates, {other_count} other messages")
print("-" * 70)
if update_count == 0:
print("\n*** WARNING: No status updates received! ***")
print("Possible causes:")
print(" 1. HW_START_UPDATEMSGS not being processed")
print(" 2. Controller firmware doesn't support automatic updates")
print(" 3. Updates are being sent but not parsed correctly")
print("\nTrying to manually request a status update...")
# Try requesting status update manually
mc.request_status_update(mc.DEST_X_AXIS)
mc.request_status_update(mc.DEST_Y_AXIS)
time.sleep(0.5)
msgs = mc.get_all_messages()
print(f"\nReceived {len(msgs)} messages after manual request:")
for msg in msgs:
msg_name = MsgId(msg.msg_id).name if msg.msg_id in [m.value for m in MsgId] else f"0x{msg.msg_id:04X}"
print(f" {msg_name} from 0x{msg.source:02X}: {hexdump(msg.raw)}")
# Also check cached positions
print("\n" + "-" * 70)
print("Cached positions in driver:")
print("-" * 70)
print(f" X position: {mc.get_stage_position_x()}")
print(f" Y position: {mc.get_stage_position_y()}")
print(f" X encoder: {mc.encoder_count_x}")
print(f" Y encoder: {mc.encoder_count_y}")
return 0
except KeyboardInterrupt:
print("\n\nTest interrupted by user")
return 1
except Exception as e:
print(f"\n*** ERROR: {e}")
import traceback
traceback.print_exc()
return 1
finally:
if mc is not None:
print("\nDisconnecting from controller...")
mc.disconnect()
print("Disconnected")
if __name__ == "__main__":
sys.exit(main())
tests/test_temperature_scaling.py
-71
View File
@@ -1,71 +0,0 @@
#!/opt/srasenv/bin/python3
"""
Test temperature scaling calculations.
"""
import math
# Constants
ADC_TO_VOLTS = 0.000244140625
STEINHART_A = 0.0011279
STEINHART_B = 0.00023429
STEINHART_C = 8.7298e-8
def test_main_temp_calculation(raw_adc):
"""Test main temperature calculation with different circuit assumptions."""
print(f"\n{'='*60}")
print(f"Testing Main Temperature with RAW ADC = {raw_adc}")
print(f"{'='*60}")
v_thermistor = raw_adc * ADC_TO_VOLTS
print(f"V_thermistor = {v_thermistor:.6f} V")
# Try different circuit configurations
configs = [
(10000, 5.0, "10kΩ series, 5V ref"),
(10000, 3.3, "10kΩ series, 3.3V ref"),
(100000, 5.0, "100kΩ series, 5V ref"),
(10000, 2.5, "10kΩ series, 2.5V ref"),
]
for r_series, vref, desc in configs:
print(f"\n{desc}:")
print(f" R_series = {r_series} Ω, Vref = {vref} V")
if v_thermistor >= vref:
print(f" ERROR: V_thermistor >= Vref")
continue
r_thermistor = r_series * v_thermistor / (vref - v_thermistor)
print(f" R_thermistor = {r_thermistor:.2f} Ω")
if r_thermistor <= 0:
print(f" ERROR: Invalid resistance")
continue
# Steinhart-Hart
ln_r = math.log(r_thermistor)
inv_t = STEINHART_A + STEINHART_B * ln_r + STEINHART_C * (ln_r ** 3)
temp_k = 1.0 / inv_t
temp_c = temp_k - 273.15
print(f" ln(R) = {ln_r:.6f}")
print(f" Temperature = {temp_c:.2f} °C")
def test_current_scaling(raw_adc):
"""Test current scaling."""
print(f"\n{'='*60}")
print(f"Testing Current Scaling with RAW ADC = {raw_adc}")
print(f"{'='*60}")
current = raw_adc * 12.0 * ADC_TO_VOLTS
print(f"Current = {current:.6f} A")
if __name__ == '__main__':
# Test with the reported raw values
test_main_temp_calculation(2)
test_main_temp_calculation(3)
test_current_scaling(0)
test_current_scaling(100)
test_current_scaling(4095)