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

Phase 6: strip signature-restating docstrings; correct README/SETUP

Browse Source 
- Collapsed Args:/Returns:/Raises: blocks that only restated the
  signature (364 lines): tektronix_base 48% -> ~20% doc density,
  helios_laser and uc480_camera likewise. Only docstrings whose entire
  body was those sections were touched.
- Preserved verbatim the comments that carry hardware knowledge the code
  can't express: uc480's USB split-transaction contention note (with its
  measured fps), the IS_ALLOW_STARTER_FW_UPLOAD segfault explanation, the
  QImage-copy rationale, and tektronix's NUMFRAMESACQuired warning.
- README: project structure, quick start, and every usage example now
  describe code that exists (they referenced hardware/bbd202.py,
  CoherentHOPSLaser, get_curve_binary, and 'python -m scanengine.app',
  none of which do). Added a headless-scan example and a read-a-scan-file
  example, since reuse without the GUI is the point of the refactor.
- SETUP: structure section defers to README instead of keeping a second
  stale copy; documents the vendored uEye SDK and the Genesis quarantine.
- ruff is now clean repo-wide: fixed the remaining raise-from, unused
  loop variables, placeholder f-strings, and a non-strict zip; the
  widget-layout semicolon idiom is an explicit config ignore rather than
  22 standing warnings.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
Thomas Ales
2026-07-28 11:27:36 -05:00
parent 44febe34b8
commit 709dc529df
10 changed files with 264 additions and 584 deletions
Download Patch File Download Diff File Expand all files Collapse all files
hardware/helios_laser.py
+8 -70
View File
@@ -29,13 +29,7 @@ class HeliosLaser:
"""
def __init__(self, port: str = None, timeout: float = 1.0):
"""
Initialize Helios laser driver.
Args:
port: Serial port (e.g., '/dev/ttyUSB0' or 'COM5')
timeout: Serial timeout in seconds
"""
"""Initialize Helios laser driver."""
self.port = port
self.timeout = timeout
self.serial = None
@@ -48,15 +42,7 @@ class HeliosLaser:
return list_port_devices()
def connect(self, port: str = None) -> bool:
"""
Connect to the Helios laser.
Args:
port: Serial port (uses stored port if None)
Returns:
True if connection successful
"""
"""Connect to the Helios laser."""
if port:
self.port = port
@@ -91,15 +77,7 @@ class HeliosLaser:
self.serial = None
def _send_command(self, command: str) -> bool:
"""
Send a command to the laser.
Args:
command: ASCII command string (without CR)
Returns:
True if sent successfully
"""
"""Send a command to the laser."""
if not self.is_connected or not self.serial:
logger.error("Not connected to laser")
return False
@@ -164,15 +142,7 @@ class HeliosLaser:
return None
def set_frequency_hz(self, frequency: int) -> bool:
"""
Set laser pulse frequency in Hz.
Args:
frequency: Frequency in Hz (16700 - 125000)
Returns:
True if successful
"""
"""Set laser pulse frequency in Hz."""
if not (16700 <= frequency <= 125000):
logger.error(f"Frequency {frequency} Hz out of range (16700-125000)")
return False
@@ -189,15 +159,7 @@ class HeliosLaser:
return self._send_command(command)
def set_current_ma(self, current: int) -> bool:
"""
Set pump diode current in mA.
Args:
current: Current in mA (0 - 2000 for this model)
Returns:
True if successful
"""
"""Set pump diode current in mA."""
if not (0 <= current <= 2000):
logger.error(f"Current {current} mA out of range (0-2000)")
return False
@@ -206,28 +168,12 @@ class HeliosLaser:
return self._send_command(command)
def set_pulse_mode(self, mode: PulseMode) -> bool:
"""
Set pulse mode.
Args:
mode: PulseMode enumeration value
Returns:
True if successful
"""
"""Set pulse mode."""
command = f"LDG {mode.value}"
return self._send_command(command)
def set_laser_enable(self, enable: bool) -> bool:
"""
Enable or disable laser emission.
Args:
enable: True to enable, False to disable
Returns:
True if successful
"""
"""Enable or disable laser emission."""
command = f"LDO {1 if enable else 0}"
success = self._send_command(command)
@@ -329,15 +275,7 @@ class HeliosLaser:
return None
def set_remote_enable(self, enable: bool) -> bool:
"""
Set the remote enable state (LRE - utility connector pin 8).
Args:
enable: True to activate remote enable, False to deactivate
Returns:
True if successful
"""
"""Set the remote enable state (LRE - utility connector pin 8)."""
command = f"LRE {1 if enable else 0}"
return self._send_command(command)
hardware/pybbd202/apt_messages.py
+1 -2
View File
@@ -280,8 +280,7 @@ class APTProtocol():
raise ValueError(f"No data fields have been defined for {msg_spec['name']}!")
unpacked_payload = struct.unpack(fmt_string, payload)
data = {field: value for field, value in zip(data_fields,
unpacked_payload)}
data = dict(zip(data_fields, unpacked_payload, strict=True))
data['destination'] = dest
data['source'] = src
hardware/pybbd202/bbd20x.py
+1 -1
View File
@@ -344,7 +344,7 @@ class ThorlabsServoDriver():
power up. Default timeout is 60s, but 20-30s is fine as well if
you're in that much of a hurry.
'''
ch = self._channel_for(axis)
self._channel_for(axis) # validate the axis address
self.send_and_wait(0x0443, timeout=timeout, retries=0, chan_ident=1,
destination=axis, source=0x01)
hardware/tektronix_base.py
+30 -262
View File
@@ -93,13 +93,7 @@ class TektronixOscilloscopeBase:
self._connected = False
def connect(self):
"""
Establish TCP socket connection to the oscilloscope.
Raises:
ValueError: If resource_name is not provided
ConnectionError: If connection fails
"""
"""Establish TCP socket connection to the oscilloscope."""
if not self.resource_name:
raise ValueError("resource_name (IP address/hostname) must be provided")
@@ -114,7 +108,8 @@ class TektronixOscilloscopeBase:
except socket.error as e:
self.socket = None
self._connected = False
raise ConnectionError(f"Failed to connect to {self.resource_name}:{self.port} - {e}")
raise ConnectionError(
f"Failed to connect to {self.resource_name}:{self.port} - {e}") from e
def disconnect(self):
"""
@@ -130,18 +125,7 @@ class TektronixOscilloscopeBase:
self._connected = False
def _normalize_channel(self, channel):
"""
Normalize channel input to integer (1-4).
Args:
channel: Channel as int (1-4) or string ('CH1'-'CH4')
Returns:
int: Channel number (1-4)
Raises:
ValueError: If channel is invalid
"""
"""Normalize channel input to integer (1-4)."""
if isinstance(channel, str):
channel_upper = channel.upper()
if not (channel_upper.startswith('CH') and len(channel_upper) == 3 and channel_upper[2].isdigit()):
@@ -154,27 +138,13 @@ class TektronixOscilloscopeBase:
return channel
def set_acquire_mode(self, mode):
"""
Set the acquisition mode.
Args:
mode: Acquisition mode to set (case-insensitive). Valid options:
'SAMple' or 'SAM' or 'SAMPLE' - Sample mode (default)
'PEAKdetect' or 'PEAK' or 'PEAKDETECT' - Peak detect mode
'HIRes' or 'HIR' or 'HIRES' - High resolution mode
'AVErage' or 'AVE' or 'AVERAGE' - Average mode
'ENVelope' or 'ENV' or 'ENVELOPE' - Envelope mode
Raises:
ValueError: If mode is not valid
RuntimeError: If not connected
"""
"""Set the acquisition mode."""
# Normalize mode to uppercase for comparison
mode_upper = mode.upper()
# Check if mode is valid (any short or long form)
valid = False
for long_form, variants in self.ACQUIRE_MODES.items():
for variants in self.ACQUIRE_MODES.values():
if mode_upper in [v.upper() for v in variants]:
valid = True
break
@@ -189,30 +159,12 @@ class TektronixOscilloscopeBase:
self.write(f"ACQuire:MODe {mode}")
def get_fastframe_state(self):
"""
Query the current FastFrame state.
Returns:
int: 0 if FastFrame is off, 1 if active
Raises:
RuntimeError: If not connected
"""
"""Query the current FastFrame state."""
response = self.query("HORizontal:FASTframe:STATE?")
return int(response)
def set_fastframe_state(self, state):
"""
Enable or disable FastFrame mode.
Args:
state: FastFrame state (0 = off, 1 = active)
Can be int (0/1) or bool (False/True)
Raises:
ValueError: If state is not valid
RuntimeError: If not connected
"""
"""Enable or disable FastFrame mode."""
# Convert boolean to int if needed
if isinstance(state, bool):
state = 1 if state else 0
@@ -224,16 +176,7 @@ class TektronixOscilloscopeBase:
self.write(f"HORizontal:FASTframe:STATE {state}")
def set_fastframe_count(self, count):
"""
Set the number of FastFrame frames.
Args:
count: Number of frames to capture (must be positive integer)
Raises:
ValueError: If count is not valid
RuntimeError: If not connected
"""
"""Set the number of FastFrame frames."""
# Validate count
if not isinstance(count, int) or count <= 0:
raise ValueError(f"Invalid frame count: {count}. Must be a positive integer")
@@ -241,29 +184,12 @@ class TektronixOscilloscopeBase:
self.write(f"HORizontal:FASTframe:COUNt {count}")
def get_record_length(self):
"""
Query the current horizontal record length.
Returns:
int: Current record length in samples
Raises:
RuntimeError: If not connected
"""
"""Query the current horizontal record length."""
response = self.query("HORizontal:MODe:RECOrdlength?")
return int(response)
def set_sample_rate(self, rate):
"""
Set the horizontal sample rate.
Args:
rate: Sample rate in samples per second (must be positive number)
Raises:
ValueError: If rate is not valid
RuntimeError: If not connected
"""
"""Set the horizontal sample rate."""
# Validate rate
if not isinstance(rate, (int, float)) or rate <= 0:
raise ValueError(f"Invalid sample rate: {rate}. Must be a positive number")
@@ -271,24 +197,12 @@ class TektronixOscilloscopeBase:
self.write(f"HORizontal:MODe:SAMPLERate {rate}")
def set_trigger_slope(self, slope):
"""
Set the trigger slope.
Args:
slope: Trigger slope (case-insensitive). Valid options:
'RISe' or 'RISE' - Rising edge
'FALL' - Falling edge
'EITher' or 'EITHER' - Either edge
Raises:
ValueError: If slope is not valid
RuntimeError: If not connected
"""
"""Set the trigger slope."""
slope_upper = slope.upper()
# Check if slope is valid
valid = False
for long_form, variants in self.TRIGGER_SLOPE.items():
for variants in self.TRIGGER_SLOPE.values():
if slope_upper in [v.upper() for v in variants]:
valid = True
break
@@ -303,18 +217,7 @@ class TektronixOscilloscopeBase:
self.write(f"TRIGger:A:EDGE:SLOpe {slope}")
def set_trigger_source(self, source):
"""
Set the trigger source.
Args:
source: Trigger source channel. Valid options:
'CH1', 'CH2', 'CH3', 'CH4' (case-insensitive)
Can also pass as integer 1-4
Raises:
ValueError: If source is not valid
RuntimeError: If not connected
"""
"""Set the trigger source."""
# Convert integer to channel string if needed
if isinstance(source, int):
if source < 1 or source > 4:
@@ -333,17 +236,7 @@ class TektronixOscilloscopeBase:
self.write(f"TRIGger:A:EDGE:SOUrce {source}")
def set_trigger_level(self, channel, level):
"""
Set the trigger level for a specific channel.
Args:
channel: Channel number (1-4) or channel string ('CH1'-'CH4')
level: Trigger level in volts
Raises:
ValueError: If channel or level is not valid
RuntimeError: If not connected
"""
"""Set the trigger level for a specific channel."""
# Convert to channel number if string
if isinstance(channel, str):
channel_upper = channel.upper()
@@ -362,23 +255,12 @@ class TektronixOscilloscopeBase:
self.write(f"TRIGger:A:LEVel:CH{channel} {level}")
def set_trigger_mode(self, mode):
"""
Set the trigger mode.
Args:
mode: Trigger mode (case-insensitive). Valid options:
'AUTO' - Auto trigger mode
'NORMal' or 'NORMAL' - Normal trigger mode
Raises:
ValueError: If mode is not valid
RuntimeError: If not connected
"""
"""Set the trigger mode."""
mode_upper = mode.upper()
# Check if mode is valid
valid = False
for long_form, variants in self.TRIGGER_MODE.items():
for variants in self.TRIGGER_MODE.values():
if mode_upper in [v.upper() for v in variants]:
valid = True
break
@@ -395,38 +277,18 @@ class TektronixOscilloscopeBase:
# ========== Channel Control Methods ==========
def set_channel_bandwidth(self, channel, bandwidth):
"""
Set the bandwidth for a specific channel.
Args:
channel: Channel number (1-4) or channel string ('CH1'-'CH4')
bandwidth: Bandwidth setting (depends on scope model)
Raises:
ValueError: If channel is not valid
RuntimeError: If not connected
"""
"""Set the bandwidth for a specific channel."""
channel = self._normalize_channel(channel)
self.write(f"CH{channel}:BANdwidth {bandwidth}")
def set_channel_coupling(self, channel, coupling):
"""
Set the coupling mode for a specific channel.
Args:
channel: Channel number (1-4) or channel string ('CH1'-'CH4')
coupling: Coupling mode (AC or DC, case-insensitive)
Raises:
ValueError: If channel or coupling is not valid
RuntimeError: If not connected
"""
"""Set the coupling mode for a specific channel."""
channel = self._normalize_channel(channel)
coupling_upper = coupling.upper()
# Validate coupling
valid = False
for long_form, variants in self.CHANNEL_COUPLING.items():
for variants in self.CHANNEL_COUPLING.items():
if coupling_upper in [v.upper() for v in variants]:
valid = True
break
@@ -441,17 +303,7 @@ class TektronixOscilloscopeBase:
self.write(f"CH{channel}:COUPling {coupling}")
def set_channel_label_name(self, channel, name):
"""
Set the label name for a specific channel.
Args:
channel: Channel number (1-4) or channel string ('CH1'-'CH4')
name: Label name as string
Raises:
ValueError: If channel is not valid
RuntimeError: If not connected
"""
"""Set the label name for a specific channel."""
channel = self._normalize_channel(channel)
if not isinstance(name, str):
@@ -461,17 +313,7 @@ class TektronixOscilloscopeBase:
self.write(f'CH{channel}:LABel:NAMe "{name}"')
def set_channel_position(self, channel, position):
"""
Set the vertical position for a specific channel.
Args:
channel: Channel number (1-4) or channel string ('CH1'-'CH4')
position: Vertical position in divisions (can be negative or positive)
Raises:
ValueError: If channel or position is not valid
RuntimeError: If not connected
"""
"""Set the vertical position for a specific channel."""
channel = self._normalize_channel(channel)
if not isinstance(position, (int, float)):
@@ -480,17 +322,7 @@ class TektronixOscilloscopeBase:
self.write(f"CH{channel}:POSition {position}")
def set_channel_scale(self, channel, scale):
"""
Set the vertical scale for a specific channel.
Args:
channel: Channel number (1-4) or channel string ('CH1'-'CH4')
scale: Vertical scale in volts per division (must be positive)
Raises:
ValueError: If channel or scale is not valid
RuntimeError: If not connected
"""
"""Set the vertical scale for a specific channel."""
channel = self._normalize_channel(channel)
if not isinstance(scale, (int, float)) or scale <= 0:
@@ -499,17 +331,7 @@ class TektronixOscilloscopeBase:
self.write(f"CH{channel}:SCAle {scale}")
def set_channel_termination(self, channel, termination):
"""
Set the termination for a specific channel.
Args:
channel: Channel number (1-4) or channel string ('CH1'-'CH4')
termination: Termination in ohms (50 or 1000000)
Raises:
ValueError: If channel or termination is not valid
RuntimeError: If not connected
"""
"""Set the termination for a specific channel."""
channel = self._normalize_channel(channel)
if termination not in self.CHANNEL_TERMINATION:
@@ -521,18 +343,7 @@ class TektronixOscilloscopeBase:
# ========== Waveform Transfer Methods ==========
def set_data_source(self, source):
"""
Set the data source for waveform transfer.
Args:
source: Data source channel. Valid options:
'CH1', 'CH2', 'CH3', 'CH4' (case-insensitive)
Can also pass as integer 1-4
Raises:
ValueError: If source is not valid
RuntimeError: If not connected
"""
"""Set the data source for waveform transfer."""
# Convert integer to channel string if needed
if isinstance(source, int):
if source < 1 or source > 4:
@@ -551,15 +362,7 @@ class TektronixOscilloscopeBase:
self.write(f"DATa:SOUrce {source}")
def query_wfmoutpre(self):
"""
Query all waveform output preamble parameters.
Returns:
str: Complete waveform preamble string with all parameters
Raises:
RuntimeError: If not connected
"""
"""Query all waveform output preamble parameters."""
return self.query("WFMOutpre?")
def transfer_curve(self):
@@ -627,7 +430,7 @@ class TektronixOscilloscopeBase:
# fewer frames than configured — reading the configured count would block.
acquired = int(self.query("ACQuire:NUMFRAMESACQuired?"))
if acquired <= 0:
raise RuntimeError(f"Scope acquired 0 FastFrame frames — no data to read")
raise RuntimeError("Scope acquired 0 FastFrame frames — no data to read")
frame_count = acquired
# Send a single CURVe? query - scope will return all frames
@@ -653,21 +456,7 @@ class TektronixOscilloscopeBase:
return waveforms
def parse_curve_data(self, curve_bytes, byte_count=1, signed=True, byte_order='MSB'):
"""
Parse raw curve data into integer array.
Args:
curve_bytes: Raw binary curve data
byte_count: Number of bytes per sample (1 or 2)
signed: True for signed integer, False for unsigned
byte_order: 'MSB' or 'LSB' for byte order
Returns:
list: List of integer values from the curve data
Raises:
ValueError: If parameters are invalid
"""
"""Parse raw curve data into integer array."""
if byte_count not in (1, 2):
raise ValueError(f"Invalid byte count: {byte_count}. Must be 1 or 2")
@@ -709,16 +498,7 @@ class TektronixOscilloscopeBase:
return values
def write(self, command):
"""
Send a raw SCPI command to the instrument.
Args:
command: SCPI command string
Raises:
RuntimeError: If not connected
socket.error: If communication fails
"""
"""Send a raw SCPI command to the instrument."""
if not self._connected or not self.socket:
raise RuntimeError("Not connected to instrument")
@@ -728,19 +508,7 @@ class TektronixOscilloscopeBase:
self.socket.sendall(command.encode('ascii'))
def query(self, command):
"""
Send a SCPI query and return the response.
Args:
command: SCPI query command string
Returns:
str: Response from the instrument (stripped of trailing newline)
Raises:
RuntimeError: If not connected
socket.error: If communication fails
"""
"""Send a SCPI query and return the response."""
if not self._connected or not self.socket:
raise RuntimeError("Not connected to instrument")
hardware/uc480_camera.py
+15 -84
View File
@@ -91,12 +91,7 @@ class UC480Camera(QObject):
error_occurred = pyqtSignal(str) # Emitted when an error occurs
def __init__(self, camera_id: int = 1):
"""
Initialize the uC480 camera driver.
Args:
camera_id: Camera ID (1-based; use is_GetCameraList to find IDs)
"""
"""Initialize the uC480 camera driver."""
super().__init__()
self.camera_id = camera_id
@@ -131,12 +126,7 @@ class UC480Camera(QObject):
self._settings_lock = threading.Lock()
def initialize(self) -> bool:
"""
Initialize the camera and allocate memory.
Returns:
True if successful, False otherwise
"""
"""Initialize the camera and allocate memory."""
try:
# Initialize camera. After is_ExitCamera the UI124x series
# resets and re-enumerates on USB (firmware reload), so retry
@@ -264,12 +254,7 @@ class UC480Camera(QObject):
logger.error(f"is_ExitCamera failed: {ret} — camera handle may still be held by daemon")
def start_capture(self) -> bool:
"""
Start continuous video capture.
Returns:
True if successful, False otherwise
"""
"""Start continuous video capture."""
if not self.is_initialized:
logger.error("Camera not initialized")
return False
@@ -311,12 +296,7 @@ class UC480Camera(QObject):
return ret == ueye.IS_SUCCESS
def stop_capture(self) -> bool:
"""
Stop continuous video capture.
Returns:
True if successful, False otherwise
"""
"""Stop continuous video capture."""
if not self.is_capturing:
return True
@@ -331,12 +311,7 @@ class UC480Camera(QObject):
return True
def get_frame(self) -> Optional[QImage]:
"""
Capture a single frame from the camera.
Returns:
QImage if successful, None otherwise
"""
"""Capture a single frame from the camera."""
if not self.is_initialized:
logger.error("Camera not initialized")
return None
@@ -377,15 +352,7 @@ class UC480Camera(QObject):
return None
def set_exposure(self, exposure_ms: float) -> bool:
"""
Set camera exposure time.
Args:
exposure_ms: Exposure time in milliseconds
Returns:
True if successful, False otherwise
"""
"""Set camera exposure time."""
if not self.is_initialized:
return False
@@ -405,12 +372,7 @@ class UC480Camera(QObject):
return False
def get_exposure(self) -> Optional[float]:
"""
Get current exposure time.
Returns:
Exposure time in milliseconds, or None if failed
"""
"""Get current exposure time."""
if not self.is_initialized:
return None
@@ -428,12 +390,7 @@ class UC480Camera(QObject):
return None
def get_pixel_clock_range(self) -> Optional[Tuple[int, int, int]]:
"""
Query the sensor's supported pixel clock range.
Returns:
(min_mhz, max_mhz, increment_mhz), or None if the query failed
"""
"""Query the sensor's supported pixel clock range."""
if not self.is_initialized:
return None
@@ -452,15 +409,7 @@ class UC480Camera(QObject):
return None
def set_pixel_clock(self, pixel_clock_mhz: int) -> bool:
"""
Set camera pixel clock.
Args:
pixel_clock_mhz: Pixel clock in MHz
Returns:
True if successful, False otherwise
"""
"""Set camera pixel clock."""
if not self.is_initialized:
return False
@@ -512,15 +461,7 @@ class UC480Camera(QObject):
return False
def set_gain(self, master_gain: int) -> bool:
"""
Set camera master gain.
Args:
master_gain: Gain value (0-100)
Returns:
True if successful, False otherwise
"""
"""Set camera master gain."""
if not self.is_initialized:
return False
@@ -541,9 +482,9 @@ class UC480Camera(QObject):
return True
elif ret == ueye.IS_CANT_COMMUNICATE_WITH_DRIVER:
logger.error(
f"Hardware gain not supported by this camera model "
f"(IS_CANT_COMMUNICATE_WITH_DRIVER). "
f"Consider using gain boost instead."
"Hardware gain not supported by this camera model "
"(IS_CANT_COMMUNICATE_WITH_DRIVER). "
"Consider using gain boost instead."
)
return False
else:
@@ -551,12 +492,7 @@ class UC480Camera(QObject):
return False
def get_sensor_info(self) -> dict:
"""
Get camera sensor information.
Returns:
Dictionary with sensor information
"""
"""Get camera sensor information."""
if not self.is_initialized:
return {}
@@ -582,12 +518,7 @@ class CameraStreamThread(QThread):
error_occurred = pyqtSignal(str)
def __init__(self, camera: UC480Camera):
"""
Initialize the camera stream thread.
Args:
camera: UC480Camera instance
"""
"""Initialize the camera stream thread."""
super().__init__()
self.camera = camera
self.running = False