2 Commits

Author SHA1 Message Date
Thomas Ales 7bcdff9756 Remove nonessential files
- photorec.ses / photorec.se2: accidentally committed PhotoRec
  data-recovery session files
- sc3-aui-focusing.py: dead module — imports a T3RStepperDriver that
  doesn't exist anywhere; superseded by t3r_control_panel.py
- adc_bug.md: stale debugging note for an already-applied fix
- app_style.qss: empty stylesheet; app.py already handles its absence

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
2026-07-28 09:27:30 -05:00
Thomas Ales 2041fc8439 pre merge cleanup commit 2026-07-28 09:20:58 -05:00
50 changed files with 1628 additions and 582 deletions
Regular → Executable
View File
Regular → Executable
View File
Regular → Executable
View File
Regular → Executable
View File
Regular → Executable
View File
-37
View File
@@ -1,37 +0,0 @@
# ADC YOFF Sign Bug — sras_viewer.py
## Status
Fix applied, awaiting user testing.
## What was wrong
`DC_YOFF_ADC` in `sras_viewer.py` was `+87.04` instead of `-87.04`.
The Tektronix scope stores CH3/CH4 waveform data as **signed int8** (−128 to +127), where ADC 0 = screen center. The scope's vertical position for CH3/CH4 is set to `−2.72 div` in `sc3_aui_app.py`, which places 0 V **below** center at ADC count `−2.72 × 32 = −87.04`. The comment in the code had the formula as `-position × (256/8)` (sign flipped), producing `+87.04` instead of the correct `−87.04`.
## Effect of the bug
- `adc_to_mv` was off by 272 mV in the negative direction
- ADC −87 (true 0 V signal) → −272 mV (should be ≈ 0 mV)
- ADC 0 (screen center, above ground) → −136 mV (should be +136 mV)
- DC images for CH3/CH4 (Bias A/B) showed large negative voltages, physically impossible for DC bias signals
- RF mask threshold (`mv_to_adc`) was also broken: threshold ADC value ~+87 was being compared against pixel means clustered around −87, so nearly every pixel would have been incorrectly masked
## The fix
`sras_viewer.py` line 48:
```python
# Before
DC_YOFF_ADC = 87.04 # ADC count that represents 0 V
# After
DC_YOFF_ADC = -87.04 # ADC count that represents 0 V
```
Comment on line 47 also corrected from `-position × (256/8)` to `position × (256/8)`.
## What to verify during testing
1. CH3 and CH4 DC images show positive (or near-zero) voltages consistent with the bias signal levels
2. RF (CH1) image is not excessively masked — pixels with a genuine bias signal above the threshold should appear
3. `mv_to_adc(0.0)` should now return −87.04 (not +87.04)
4. The default threshold of 0.125 mV should correspond to ADC ≈ −87.0, not +87.1
Regular → Executable
+74 -4
View File
@@ -14,9 +14,11 @@ from hardware.coherent_hops_laser import CoherentHOPSLaser, DummyLaser
from hardware.helios_laser import HeliosLaser, PulseMode
from hardware.uc480_camera import UC480Camera, CameraStreamThread
from hardware.pybbd202 import ThorlabsServoDriver, AXIS_X, AXIS_Y, TriggerBitsServo
from hardware.t3r_driver import T3RDriver
from motion_worker import MotionWorker
from scanning.stage_scan_plan_generator import StageScanPlanGenerator
from genesis_worker import GenesisWorker, GenesisCommand
from t3r_control_panel import T3RControlPanel
from ui_mainwindow import Ui_MainWindow
# Page indices in stackedWidget
@@ -95,10 +97,11 @@ class ScanWorker(QtCore.QObject):
overall_progress = QtCore.pyqtSignal(int)
status_message = QtCore.pyqtSignal(str)
def __init__(self, scan_params, motion_worker):
def __init__(self, scan_params, motion_worker, t3r_driver=None):
super().__init__()
self.scan_params = scan_params
self.motion_worker = motion_worker
self.t3r_driver = t3r_driver
self.should_stop = False
@QtCore.pyqtSlot()
@@ -108,7 +111,30 @@ class ScanWorker(QtCore.QObject):
self.motion_worker.scanning_active = True
try:
self.scan_started.emit()
# TODO: implement scan execution logic
num_angles = self.scan_params.get("num_angles", 1)
angle_step = 360.0 / num_angles if num_angles > 1 else 0.0
gr_microsteps = self.scan_params.get("gr_axis_microsteps", 16)
for angle_idx in range(num_angles):
if self.should_stop:
break
self.angle_started.emit(angle_idx, num_angles)
self.status_message.emit(
f"Scanning angle {angle_idx + 1}/{num_angles}")
# TODO: execute scan lines for this angle via motion_worker
if angle_idx < num_angles - 1 and angle_step and self.t3r_driver:
if self.t3r_driver.is_open:
self.status_message.emit(
f"Rotating stage {angle_step:.3f}° for next angle…")
self.t3r_driver.rotate_stage(
angle_step,
gr_microsteps,
self.scan_params.get("rotation_velocity", 8000),
self.scan_params.get("rotation_accel", 4000),
)
# TODO: wait for MOTION_DONE event before proceeding
self.scan_completed.emit()
except Exception as e:
self.scan_failed.emit(str(e))
@@ -140,8 +166,13 @@ class MainWindow(QtWidgets.QMainWindow):
self.scan_worker: Optional[ScanWorker] = None
self.scan_thread: Optional[QtCore.QThread] = None
# T3R focusing / rotation driver (lives in main thread; reader runs internally)
self.t3r_driver = T3RDriver(self)
self.t3r_panel: Optional[T3RControlPanel] = None
self._connect_signals()
self._init_genesis_worker()
self._init_t3r_menu()
self.ui.stackedWidget.setCurrentIndex(PAGE_START)
# ------------------------------------------------------------------
@@ -358,7 +389,13 @@ class MainWindow(QtWidgets.QMainWindow):
pass # TODO
def _on_t3r_connect(self):
pass # TODO
port = self.ui.t3r_serial_port_edit.currentText().split(" ")[0]
self._show_t3r_panel()
if port and not self.t3r_driver.is_open:
try:
self.t3r_driver.connect(port)
except Exception as exc:
QtWidgets.QMessageBox.warning(self, "T3R Connect", str(exc))
# ------------------------------------------------------------------
# New scan page
@@ -419,6 +456,10 @@ class MainWindow(QtWidgets.QMainWindow):
"save_directory": self.ui.newscan_save_directory_edit.text(),
"scan_velocity_mm_s": self.config["stage"]["scan_velocity_mm_s"],
"scan_acceleration_mm_s2": self.config["stage"]["scan_acceleration_mm_s2"],
# T3R rotation between angles (GR-axis, ch1)
"gr_axis_microsteps": 16,
"rotation_velocity": 8000,
"rotation_accel": 4000,
}
# ------------------------------------------------------------------
@@ -456,7 +497,7 @@ class MainWindow(QtWidgets.QMainWindow):
def _start_scan(self, scan_params: dict):
self.scan_thread = QtCore.QThread()
self.scan_worker = ScanWorker(scan_params, self.motion_worker)
self.scan_worker = ScanWorker(scan_params, self.motion_worker, self.t3r_driver)
self.scan_worker.moveToThread(self.scan_thread)
self.scan_thread.started.connect(self.scan_worker.run_scan)
@@ -497,6 +538,33 @@ class MainWindow(QtWidgets.QMainWindow):
self.scan_thread = None
self.scan_worker = None
# ------------------------------------------------------------------
# T3R focusing / rotation panel
# ------------------------------------------------------------------
def _init_t3r_menu(self):
"""Add a Hardware menu with a T3R panel toggle action."""
hw_menu = self.menuBar().addMenu("Hardware")
self._t3r_action = hw_menu.addAction("T3R Focusing && Rotation…")
self._t3r_action.setCheckable(True)
self._t3r_action.setShortcut("Ctrl+T")
self._t3r_action.triggered.connect(self._on_t3r_action_toggled)
def _show_t3r_panel(self):
if self.t3r_panel is None:
self.t3r_panel = T3RControlPanel(self.t3r_driver, self)
self.t3r_panel.finished.connect(
lambda: self._t3r_action.setChecked(False))
self.t3r_panel.show()
self.t3r_panel.raise_()
self._t3r_action.setChecked(True)
def _on_t3r_action_toggled(self, checked: bool):
if checked:
self._show_t3r_panel()
elif self.t3r_panel is not None:
self.t3r_panel.hide()
# ------------------------------------------------------------------
# Genesis laser worker
# ------------------------------------------------------------------
@@ -523,6 +591,8 @@ class MainWindow(QtWidgets.QMainWindow):
# ------------------------------------------------------------------
def closeEvent(self, event):
if self.t3r_driver.is_open:
self.t3r_driver.disconnect()
self._cleanup_genesis_worker()
self._cleanup_scan_thread()
if self.motion_thread:
View File
Regular → Executable
+3 -3
View File
@@ -1,8 +1,8 @@
{
"t3r_port": "/dev/ttyACM0",
"bbd_port": "/dev/ttyAPT",
"bbd_port": "/dev/ttyUSB0",
"oscope_ip": "192.168.100.105",
"laser_freq_hz": 20000.0,
"save_dir": "/opt/scanengine-3/scans",
"helios_port": "/dev/ttyUSB0"
"save_dir": "/data/SRAS",
"helios_port": "/dev/ttyUSB1"
}
Regular → Executable
View File
Regular → Executable
View File
Regular → Executable
View File
View File
View File
View File
View File
View File
View File
View File
View File
View File
View File
Regular → Executable
View File
Regular → Executable
+1
View File
@@ -1,5 +1,6 @@
"""Hardware driver modules for ScanEngine-3"""
from .pybbd202 import ThorlabsServoDriver, TriggerBitsServo, AXIS_X, AXIS_Y, CONTROLLER
from .t3r_driver import T3RDriver
from .uc480_camera import *
from .tektronix_base import *
from .coherent_hops_laser import *
Regular → Executable
View File
Regular → Executable
View File
Regular → Executable
View File
Regular → Executable
View File
View File
View File
Regular → Executable
View File
View File
Regular → Executable
View File
Regular → Executable
+153 -20
View File
@@ -4,6 +4,9 @@ Driver for IDS/Thorlabs uEye uC480 cameras using pyueye library.
Provides camera control, live streaming, and image capture capabilities.
"""
import glob
import os
import re
import time
import numpy as np
from pyueye import ueye
@@ -12,10 +15,71 @@ from PyQt6.QtGui import QImage
import logging
import threading
from contextlib import contextmanager
from typing import Optional, Tuple
from typing import List, Optional, Tuple
logger = logging.getLogger(__name__)
IDS_USB_VENDOR_ID = "1409"
def find_camera_bus_conflicts() -> List[Tuple[str, str, str]]:
"""Return (tty_name, bus, product) for serial adapters sharing a USB host
controller with the IDS camera.
A uC480 USB2 camera at high pixel clock needs nearly the whole 480 Mbit/s
of its host controller. When a full-speed serial adapter (ESP32 CDC,
FTDI, …) on the same controller has its port *open*, the kernel's
periodic split transactions starve the camera's bulk stream — measured
on this rig: 22 fps with all ports closed, <1.5 fps with either the T3R
(ttyACM0) or BBD202 (ttyUSB1) port open, full recovery on close. The
only real fix is plugging the camera into a port on a different
controller (e.g. a USB-3/xHCI port); this check exists so the UI can
say that instead of silently dropping frames.
"""
cam_buses = set()
for vid_path in glob.glob("/sys/bus/usb/devices/*/idVendor"):
try:
with open(vid_path) as f:
if f.read().strip() != IDS_USB_VENDOR_ID:
continue
with open(os.path.join(os.path.dirname(vid_path), "busnum")) as f:
cam_buses.add(f.read().strip())
except OSError:
continue
conflicts: List[Tuple[str, str, str]] = []
if not cam_buses:
return conflicts
tty_paths = glob.glob("/sys/class/tty/ttyUSB*") + glob.glob("/sys/class/tty/ttyACM*")
for tty_path in sorted(tty_paths):
real = os.path.realpath(os.path.join(tty_path, "device"))
m = re.search(r"/usb(\d+)/", real)
if not m or m.group(1) not in cam_buses:
continue
# Walk up from the interface dir to the USB device dir for its name
product = ""
d = real
for _ in range(6):
d = os.path.dirname(d)
if os.path.exists(os.path.join(d, "busnum")):
try:
with open(os.path.join(d, "product")) as f:
product = f.read().strip()
except OSError:
pass
break
conflicts.append((os.path.basename(tty_path), m.group(1), product))
if conflicts:
devs = ", ".join(f"/dev/{n} ({p})" if p else f"/dev/{n}" for n, _, p in conflicts)
logger.warning(
f"Camera shares USB bus {sorted(cam_buses)} with serial adapters: {devs}. "
f"Opening any of these ports will collapse the camera frame rate — "
f"move the camera to a port on another USB controller."
)
return conflicts
class UC480Camera(QObject):
"""
@@ -60,7 +124,9 @@ class UC480Camera(QObject):
self.height = 0
self.bits_per_pixel = 24 # Default to 24-bit color
self.bytes_per_pixel = 3
self.color_mode = ueye.IS_CM_BGR8_PACKED
# RGB (not BGR) so get_frame() can hand the buffer straight to QImage
# without a per-frame channel-reversal copy.
self.color_mode = ueye.IS_CM_RGB8_PACKED
# Lock to serialize parameter changes that require stopping live video
self._settings_lock = threading.Lock()
@@ -159,10 +225,19 @@ class UC480Camera(QObject):
self.is_initialized = True
logger.info(f"Camera initialized: {self.width}x{self.height}, {self.bits_per_pixel}bpp")
# Set default settings
# Set default settings. Pixel clock caps the max sensor readout
# rate, which in turn caps achievable fps regardless of the
# requested framerate below — use the sensor's max rather than
# a hardcoded guess, since a too-low clock silently forces
# is_SetFrameRate to negotiate down to a much lower actual fps.
self.set_exposure(10.0) # 10ms default exposure
self.set_pixel_clock(30) # 30MHz default pixel clock
self.set_framerate(30.0) # 30fps default
clock_range = self.get_pixel_clock_range()
if clock_range is not None:
min_clock, max_clock, _ = clock_range
self.set_pixel_clock(max_clock)
else:
self.set_pixel_clock(30) # fallback if range query fails
self.set_framerate(30.0) # 30fps default (actual may be lower)
return True
@@ -210,10 +285,32 @@ class UC480Camera(QObject):
self.error_occurred.emit(f"Failed to start capture: {ret}")
return False
ret = ueye.is_EnableEvent(self.h_cam, ueye.IS_SET_EVENT_FRAME)
if ret != ueye.IS_SUCCESS:
logger.error(f"Failed to enable frame event: {ret}")
self.is_capturing = True
logger.info("Video capture started")
return True
def wait_for_frame(self, timeout_ms: int = 200) -> bool:
"""
Block until the camera signals that a new frame has landed in
image memory (or until timeout_ms elapses).
Without this, a caller polling get_frame() in a tight loop just
re-reads the same still-unfinished/unchanged buffer as fast as the
GIL allows — burning CPU without raising the delivered frame rate,
and occasionally reading a frame mid-write (tearing).
Returns:
True if a new frame arrived, False on timeout/error.
"""
if not self.is_capturing:
return False
ret = ueye.is_WaitEvent(self.h_cam, ueye.IS_SET_EVENT_FRAME, timeout_ms)
return ret == ueye.IS_SUCCESS
def stop_capture(self) -> bool:
"""
Stop continuous video capture.
@@ -224,6 +321,7 @@ class UC480Camera(QObject):
if not self.is_capturing:
return True
ueye.is_DisableEvent(self.h_cam, ueye.IS_SET_EVENT_FRAME)
ret = ueye.is_StopLiveVideo(self.h_cam, ueye.IS_WAIT)
self.is_capturing = False # Always reset, even if the call fails
if ret != ueye.IS_SUCCESS:
@@ -278,25 +376,23 @@ class UC480Camera(QObject):
copy=True
)
# Reshape to image dimensions
# Reshape to image dimensions (view, no copy — array already owns
# its memory since get_data() was called with copy=True above)
frame = np.reshape(array, (self.height, self.width, self.bytes_per_pixel))
# Convert to QImage (BGR to RGB)
height, width, channel = frame.shape
bytes_per_line = self.bytes_per_pixel * width
# Convert BGR to RGB
rgb_frame = frame[:, :, ::-1].copy()
q_image = QImage(
rgb_frame.data,
frame.data,
width,
height,
bytes_per_line,
QImage.Format.Format_RGB888
)
# Make a copy since the numpy array will be deleted
# Must copy: this QImage crosses threads via a queued signal,
# which hands the slot a new Python wrapper around the same
# frame that gets delivered after `frame` may already be GC'd —
# confirmed by testing that skipping this copy corrupts pixels.
return q_image.copy()
except Exception as e:
@@ -355,6 +451,30 @@ class UC480Camera(QObject):
else:
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
"""
if not self.is_initialized:
return None
clock_range = (ueye.c_uint * 3)()
ret = ueye.is_PixelClock(
self.h_cam,
ueye.IS_PIXELCLOCK_CMD_GET_RANGE,
clock_range,
ueye.sizeof(clock_range)
)
if ret == ueye.IS_SUCCESS:
return clock_range[0].value, clock_range[1].value, clock_range[2].value
else:
logger.error(f"Failed to get pixel clock range: {ret}")
return None
def set_pixel_clock(self, pixel_clock_mhz: int) -> bool:
"""
Set camera pixel clock.
@@ -376,7 +496,7 @@ class UC480Camera(QObject):
)
if ret == ueye.IS_SUCCESS:
logger.debug(f"Pixel clock set to {pixel_clock_mhz}MHz")
logger.info(f"Pixel clock set to {pixel_clock_mhz}MHz")
return True
else:
logger.error(f"Failed to set pixel clock: {ret}")
@@ -384,7 +504,10 @@ class UC480Camera(QObject):
def set_framerate(self, fps: float) -> bool:
"""
Set camera framerate.
Set camera framerate. The SDK negotiates the requested value against
the current pixel clock/exposure/AOI and may return a lower actual
rate — that negotiated value is what's logged and returned, not the
request, since silently trusting the request hides the real cap.
Args:
fps: Frames per second
@@ -400,7 +523,13 @@ class UC480Camera(QObject):
ret = ueye.is_SetFrameRate(self.h_cam, new_fps, actual_fps)
if ret == ueye.IS_SUCCESS:
logger.debug(f"Framerate set to {fps}fps")
if actual_fps.value < fps * 0.9:
logger.warning(
f"Requested {fps}fps but camera negotiated only "
f"{actual_fps.value:.1f}fps (pixel clock/exposure/AOI-limited)"
)
else:
logger.info(f"Framerate set to {actual_fps.value:.1f}fps")
return True
else:
logger.error(f"Failed to set framerate: {ret}")
@@ -514,12 +643,16 @@ class CameraStreamThread(QThread):
return
while self.running:
# Block until the camera actually has a new frame ready, instead
# of re-reading (and re-copying/re-emitting) the same buffer as
# fast as possible. The short timeout just bounds how quickly a
# stop() request is noticed.
if not self.camera.wait_for_frame(200):
continue
frame = self.camera.get_frame()
if frame is not None:
self.frame_ready.emit(frame)
else:
# Small delay on error to prevent CPU spinning
self.msleep(10)
self.camera.stop_capture()
Regular → Executable
View File
Regular → Executable
View File
Regular → Executable
View File
Regular → Executable
View File
Regular → Executable
+98 -8
View File
@@ -662,6 +662,12 @@
<layout class="QGridLayout" name="gridLayout_4">
<item row="0" column="4">
<widget class="QPushButton" name="bbd_enable_all_btn">
<property name="minimumSize">
<size>
<width>0</width>
<height>36</height>
</size>
</property>
<property name="text">
<string>Toggle Axes Enable</string>
</property>
@@ -699,9 +705,15 @@
</item>
<item row="4" column="4" alignment="Qt::AlignmentFlag::AlignLeft">
<widget class="QLabel" name="bbd_current_y_position_indicator">
<property name="minimumSize">
<size>
<width>170</width>
<height>44</height>
</size>
</property>
<property name="maximumSize">
<size>
<width>100</width>
<width>220</width>
<height>16777215</height>
</size>
</property>
@@ -734,6 +746,12 @@
</item>
<item row="0" column="0">
<widget class="QPushButton" name="bbd_home_all_btn">
<property name="minimumSize">
<size>
<width>0</width>
<height>36</height>
</size>
</property>
<property name="text">
<string>Home All Axes</string>
</property>
@@ -741,12 +759,23 @@
</item>
<item row="1" column="2" alignment="Qt::AlignmentFlag::AlignHCenter">
<widget class="QPushButton" name="bbd_jog_y_pos_btn">
<property name="minimumSize">
<size>
<width>0</width>
<height>44</height>
</size>
</property>
<property name="maximumSize">
<size>
<width>150</width>
<width>190</width>
<height>16777215</height>
</size>
</property>
<property name="font">
<font>
<pointsize>14</pointsize>
</font>
</property>
<property name="text">
<string>Y+</string>
</property>
@@ -756,10 +785,15 @@
<widget class="QLabel" name="label_21">
<property name="maximumSize">
<size>
<width>100</width>
<width>130</width>
<height>16777215</height>
</size>
</property>
<property name="font">
<font>
<pointsize>12</pointsize>
</font>
</property>
<property name="text">
<string>X Position:</string>
</property>
@@ -770,9 +804,15 @@
</item>
<item row="4" column="1" alignment="Qt::AlignmentFlag::AlignLeft">
<widget class="QLabel" name="bbd_current_x_position_indicator">
<property name="minimumSize">
<size>
<width>170</width>
<height>44</height>
</size>
</property>
<property name="maximumSize">
<size>
<width>100</width>
<width>220</width>
<height>16777215</height>
</size>
</property>
@@ -792,12 +832,23 @@
</item>
<item row="2" column="1">
<widget class="QPushButton" name="bbd_jog_x_neg_btn">
<property name="minimumSize">
<size>
<width>0</width>
<height>44</height>
</size>
</property>
<property name="maximumSize">
<size>
<width>150</width>
<width>190</width>
<height>16777215</height>
</size>
</property>
<property name="font">
<font>
<pointsize>14</pointsize>
</font>
</property>
<property name="text">
<string>X-</string>
</property>
@@ -818,12 +869,23 @@
</item>
<item row="3" column="2" alignment="Qt::AlignmentFlag::AlignHCenter">
<widget class="QPushButton" name="bbd_jog_y_neg_btn">
<property name="minimumSize">
<size>
<width>0</width>
<height>44</height>
</size>
</property>
<property name="maximumSize">
<size>
<width>150</width>
<width>190</width>
<height>16777215</height>
</size>
</property>
<property name="font">
<font>
<pointsize>14</pointsize>
</font>
</property>
<property name="text">
<string>Y-</string>
</property>
@@ -831,12 +893,23 @@
</item>
<item row="2" column="3">
<widget class="QPushButton" name="bbd_jog_x_pos_btn">
<property name="minimumSize">
<size>
<width>0</width>
<height>44</height>
</size>
</property>
<property name="maximumSize">
<size>
<width>150</width>
<width>190</width>
<height>16777215</height>
</size>
</property>
<property name="font">
<font>
<pointsize>14</pointsize>
</font>
</property>
<property name="text">
<string>X+</string>
</property>
@@ -859,10 +932,15 @@
<widget class="QLabel" name="label_23">
<property name="maximumSize">
<size>
<width>100</width>
<width>130</width>
<height>16777215</height>
</size>
</property>
<property name="font">
<font>
<pointsize>12</pointsize>
</font>
</property>
<property name="text">
<string>Y Position:</string>
</property>
@@ -873,6 +951,12 @@
</item>
<item row="5" column="0">
<widget class="QPushButton" name="bbd_set_current_start_btn">
<property name="minimumSize">
<size>
<width>0</width>
<height>36</height>
</size>
</property>
<property name="text">
<string>Set Current Coords as Start Coords</string>
</property>
@@ -880,6 +964,12 @@
</item>
<item row="5" column="4">
<widget class="QPushButton" name="bbd_set_delta_current_btn">
<property name="minimumSize">
<size>
<width>0</width>
<height>36</height>
</size>
</property>
<property name="text">
<string>Calculate Delta (Current - Start)</string>
</property>
Regular → Executable
+16
View File
@@ -118,6 +118,22 @@
</property>
</widget>
</item>
<item>
<widget class="QPushButton" name="pause_btn">
<property name="font">
<font>
<family>Noto Sans Condensed ExtraBold</family>
<pointsize>24</pointsize>
</font>
</property>
<property name="text">
<string>PAUSE</string>
</property>
<property name="checkable">
<bool>true</bool>
</property>
</widget>
</item>
<item>
<widget class="QPushButton" name="abort_btn">
<property name="font">
Regular → Executable
View File
+726 -422
View File
File diff suppressed because it is too large Load Diff
Regular → Executable
+86 -88
View File
@@ -1,19 +1,27 @@
# SRAS Scan Binary Format — Version 4
# SRAS Scan Binary Format — Version 6
Each `.sras` file contains **one complete scan**: all GR rotation angles and all
Y rows. Files are named `{prefix}.sras`.
Starting in v6, each angle only scans the **bounding box of the nominal ROI
rotated by that specific angle** — not the worst case across all angles — so
`x_start`, `x_delta` (and therefore `n_frames`, the points/row count) and
`n_rows` all vary per angle. A 0°/180° scan of a wide, short ROI needs far
fewer rows than a 45° scan of the same ROI, and the file format reflects that
instead of forcing every angle to the largest bounding box.
---
## File Layout
```
[Global Header — 43 bytes]
[Angle Table — n_angles × 4 bytes (float32 per angle)]
[Row Table — n_rows × 4 bytes (float32 per row)]
[Preamble Blocks — n_channels × (uint16 length + UTF-8 WFMOutpre string)]
[Background Block — uint32 n_bg_samples + n_bg_samples × int8 bytes]
[Waveform Data — n_angles × n_rows × n_channels × n_frames × samples_per_frame × bps bytes]
[Global Header — 49 bytes]
[Angle Table — n_angles × 4 bytes (float32 per angle, degrees)]
[Per-Angle Geometry Table— n_angles × 14 bytes (x_start f32, x_delta f32, n_frames u32, n_rows u16)]
[Row Table (ragged) — sum(n_rows) × 4 bytes (float32 per row, angle-major)]
[Preamble Blocks — n_channels × (uint16 length + UTF-8 WFMOutpre string)]
[Background Block — uint32 n_bg_samples + n_bg_samples × int8 bytes]
[Waveform Data (ragged) — per angle: n_rows[a] × n_channels × n_frames[a] × samples_per_frame × bps bytes]
```
All multi-byte integers and floats use **big-endian** byte order
@@ -21,33 +29,40 @@ All multi-byte integers and floats use **big-endian** byte order
---
## Global Header (42 bytes)
## Global Header (49 bytes)
| Offset | Size | Type | Field | Description |
|--------|------|-----------|--------------------|--------------------------------------------------|
| 0 | 4 | `4s` | `magic` | Always `SRAS` (0x53 0x52 0x41 0x53) |
| 4 | 1 | `uint8` | `version` | Format version — `4` |
| 4 | 1 | `uint8` | `version` | Format version — `6` |
| 5 | 2 | `uint16` | `n_angles` | Number of GR rotation angles |
| 7 | 2 | `uint16` | `n_rows` | Number of Y rows per angle |
| 9 | 4 | `float32` | `x_start_mm` | X scan start position in mm |
| 13 | 4 | `float32` | `x_delta_mm` | X scan width in mm |
| 17 | 4 | `float32` | `velocity_mm_s` | Stage scan velocity in mm/s |
| 21 | 4 | `float32` | `laser_freq_hz` | Laser repetition rate in Hz |
| 25 | 4 | `uint32` | `n_frames` | A-scans per row (= FastFrame count per channel) |
| 29 | 4 | `uint32` | `samples_per_frame`| Time samples per waveform |
| 33 | 8 | `float64` | `sample_rate_hz` | Oscilloscope sample rate in Hz (e.g. 6.25e9) |
| 41 | 1 | `uint8` | `bytes_per_sample` | Bytes per ADC sample: `1` = int8, `2` = int16 |
| 42 | 1 | `uint8` | `n_channels` | Number of channels recorded (currently `3`) |
| 7 | 4 | `float32` | `x_start_nominal` | Nominal (pre-rotation) X scan start, mm |
| 11 | 4 | `float32` | `y_start_nominal` | Nominal (pre-rotation) Y scan start, mm |
| 15 | 4 | `float32` | `x_delta_nominal` | Nominal (pre-rotation) X scan width, mm |
| 19 | 4 | `float32` | `y_delta_nominal` | Nominal (pre-rotation) Y scan height, mm |
| 23 | 4 | `float32` | `row_spacing_mm` | Y spacing between rows, mm |
| 27 | 4 | `float32` | `velocity_mm_s` | Stage scan velocity in mm/s |
| 31 | 4 | `float32` | `laser_freq_hz` | Laser repetition rate in Hz |
| 35 | 4 | `uint32` | `samples_per_frame`| Time samples per waveform |
| 39 | 8 | `float64` | `sample_rate_hz` | Oscilloscope sample rate in Hz (e.g. 6.25e9) |
| 47 | 1 | `uint8` | `bytes_per_sample` | Bytes per ADC sample: `1` = int8, `2` = int16 |
| 48 | 1 | `uint8` | `n_channels` | Number of channels recorded (currently `3`) |
**Total header size:** 43 bytes — verified:
`struct.calcsize(">4sBHHffffIIdBB") == 43`.
**Total header size:** 49 bytes — verified:
`struct.calcsize(">4sBHfffffffIdBB") == 49`.
The `*_nominal` fields describe the ROI as originally entered on the New Scan
page (XS/YS/XD/YD), **before** per-angle bounding-box expansion. They are for
reference/reconstruction only — the actual per-angle scan geometry used for
acquisition is in the Per-Angle Geometry Table below.
---
## Angle Table
Immediately after the header: **n_angles** big-endian float32 values, one per
GR angle (degrees, 0–180).
GR angle (degrees, signed; magnitude 0–180, sign gives physical rotation
direction — negative for the current CW-rotating GR stage).
```
angle[0], angle[1], …, angle[n_angles - 1]
@@ -55,15 +70,36 @@ angle[0], angle[1], …, angle[n_angles - 1]
---
## Row Table
## Per-Angle Geometry Table
Immediately after the angle table: **n_rows** big-endian float32 values, one
per Y row (mm).
Immediately after the angle table: **n_angles** fixed-size records, one per
angle (same order as the angle table), each 14 bytes:
| Size | Type | Field | Description |
|------|-----------|------------|-------------------------------------------------------|
| 4 | `float32` | `x_start` | X scan start for this angle's bounding box, mm |
| 4 | `float32` | `x_delta` | X scan width for this angle's bounding box, mm |
| 4 | `uint32` | `n_frames` | A-scans per row for this angle (FastFrame count) |
| 2 | `uint16` | `n_rows` | Number of Y rows scanned for this angle |
Format string per record: `">ffIH"`.
---
## Row Table (ragged)
Immediately after the per-angle geometry table: for each angle in order,
that angle's `n_rows` big-endian float32 Y positions (mm), concatenated with
no padding between angles.
```
y_mm[0], y_mm[1], …, y_mm[n_rows - 1]
# angle 0's rows, then angle 1's rows, …
y_mm[0][0], …, y_mm[0][n_rows[0]-1], y_mm[1][0], …, y_mm[n_angles-1][n_rows[-1]-1]
```
Row-table boundaries for angle *a* are derived from the per-angle geometry
table: `sum(n_rows[0:a])` gives the starting index into the flattened array.
---
## Preamble Blocks
@@ -97,17 +133,20 @@ int8[] bg_data — raw ADC samples (same encoding as waveform data)
---
## Waveform Data
## Waveform Data (ragged)
Immediately after the background block. Data is stored in **angle-major, row-minor**
order. Within each row, channels are interleaved in ascending channel-index
Immediately after the background block. Data is stored in **angle-major,
row-minor** order, but unlike earlier versions each angle contributes a
different number of rows (`n_rows[a]`) and a different number of frames per
row (`n_frames[a]`), both taken from that angle's Per-Angle Geometry Table
entry. Within each row, channels are interleaved in ascending channel-index
order, with each channel's FastFrame data written in frame order.
```
for angle in 0 … n_angles-1:
for row in 0 … n_rows-1:
for angle a in 0 … n_angles-1:
for row in 0 … n_rows[a]-1:
for channel in [CH1, CH3, CH4]: # 3 channels, fixed order
for frame in 0 … n_frames-1:
for frame in 0 … n_frames[a]-1:
samples[0 … samples_per_frame-1] # bps bytes each
```
@@ -117,81 +156,37 @@ int16**.
Total data size:
```
n_angles × n_rows × 3 × n_frames × samples_per_frame × bytes_per_sample
sum over angles a of: n_rows[a] × 3 × n_frames[a] × samples_per_frame × bytes_per_sample
```
> **Incomplete files:** If a scan is aborted the file is closed immediately and
> the data block will be shorter than the expected size. Readers should check
> `file_size >= header + angle_table + row_table + data` before reshaping.
> the data block will be shorter than the expected size. Readers should
> reconstruct the expected per-angle byte offsets from the Per-Angle Geometry
> Table and check `file_size` against the running total before reshaping —
> a fixed `(n_angles, n_rows, ...)` reshape (as in pre-v6 readers) will not
> work since row/frame counts are no longer uniform across angles.
---
## Spatial Mapping
The *k*-th waveform (frame) in a row corresponds to the *k*-th laser pulse that
hit the sample. The physical X position of that pulse is:
hit the sample. For a row belonging to angle *a*, the physical X position of
that pulse is:
```
x_k = x_start_mm + k * (velocity_mm_s / laser_freq_hz)
x_k = x_start[a] + k * (velocity_mm_s / laser_freq_hz)
```
---
## Python Read Example
```python
import struct, numpy as np
from pathlib import Path
HDR_FMT = ">4sBHHffffIIdBB"
HDR_SIZE = struct.calcsize(HDR_FMT) # 43 bytes
def read_sras(path):
with open(path, "rb") as f:
hdr = struct.unpack(HDR_FMT, f.read(HDR_SIZE))
magic, ver, n_angles, n_rows, xs, xd, vel, freq, nf, spf, sr, bps, n_ch = hdr
assert magic == b"SRAS" and ver == 4, "Not a v4 SRAS file"
angles = np.frombuffer(f.read(n_angles * 4), dtype=">f4")
y_positions = np.frombuffer(f.read(n_rows * 4), dtype=">f4")
# Preamble blocks (one per channel)
preambles = []
for _ in range(n_ch):
(plen,) = struct.unpack(">H", f.read(2))
preambles.append(f.read(plen).decode("utf-8"))
# Background waveform block (v4+)
(n_bg,) = struct.unpack(">I", f.read(4))
background = np.frombuffer(f.read(n_bg), dtype=np.int8)
dtype = np.int8 if bps == 1 else ">i2"
data = np.frombuffer(f.read(), dtype=dtype).reshape(
n_angles, n_rows, n_ch, nf, spf
)
return {
"angles_deg": angles,
"y_positions_mm": y_positions,
"x_start_mm": xs,
"x_delta_mm": xd,
"velocity_mm_s": vel,
"laser_freq_hz": freq,
"sample_rate_hz": sr,
"n_channels": n_ch, # 3: CH1, CH3, CH4 (see Acquisition Settings)
"preambles": preambles, # WFMOutpre strings, same order as n_channels
"background": background,# shape: (n_bg_samples,) — CH1 noise reference
# shape: (n_angles, n_rows, n_channels, n_frames, samples_per_frame)
"data": data,
}
```
using that angle's `x_start` from the Per-Angle Geometry Table (not
`x_start_nominal`).
---
## Acquisition Settings (fixed by sc3_aui_app.py)
| Parameter | Value |
|----------------------|------------------------------|
|-----------------------|------------------------------|
| Oscilloscope trigger | CH2, rising edge, 1.24 V |
| Trigger offset | 0 % (trigger at left edge) |
| Sample rate | 6.25 GS/s (160 ps/sample) |
@@ -211,3 +206,6 @@ def read_sras(path):
| 2 | One file per scan; global header with `n_angles`/`n_rows`; separate angle and row tables; three channels (CH1, CH3, CH4) per row. |
| 3 | Added preamble blocks (WFMOutpre strings) after the row table, one length-prefixed UTF-8 block per channel. |
| 4 | Added background waveform block (CH1, Helios ON / Genesis OFF) after the preamble blocks; stored as `uint32` sample count followed by raw `int8` ADC bytes. |
| 5 | (skipped) |
| 6 | Each angle now scans only the bounding box of the nominal ROI rotated by that angle instead of the AABB-expanded worst case across all angles. Header no longer carries a single global `x_start`/`x_delta`/`n_rows` — replaced with `*_nominal` reference fields plus a new Per-Angle Geometry Table (`x_start`, `x_delta`, `n_frames`, `n_rows` per angle) and a ragged Row Table / Waveform Data block sized per angle. **Not compatible with v4 readers** (e.g. `sras_viewer.py`, which has not yet been updated for v6). |
Regular → Executable
View File
Regular → Executable
View File
View File
+471
View File
@@ -0,0 +1,471 @@
#!/opt/srasenv/bin/python3
"""
SRAS Scan Manager
Command-line / interactive TUI for inspecting v6 .sras files.
A .sras file (see scan_format.md) holds one acquisition run across several
GR rotation angles, each with its own geometry (x_start, x_delta, n_frames,
n_rows) and waveform data block. This tool lists those per-angle sub-scans
and lets you export a subset to a new .sras file, or delete a subset from
the file in place — both operations rewrite the angle/geometry/row tables
and stream-copy only the selected angles' waveform data, producing a file
that is itself a valid v6 .sras readable by sras_viewer.py-style tools
(once updated for v6) or sc3_aui_app.py.
Only format version 6 is supported.
"""
import argparse
import struct
import sys
from dataclasses import dataclass, field
from datetime import datetime
from pathlib import Path
BLOB_MAGIC = b"SRAS"
BLOB_VERSION = 6
HDR_FMT = ">4sBHfffffffIdBB"
HDR_SIZE = struct.calcsize(HDR_FMT) # 49 bytes
GEOM_FMT = ">ffIH"
GEOM_SIZE = struct.calcsize(GEOM_FMT) # 14 bytes
@dataclass
class AngleEntry:
index: int
angle_deg: float
x_start: float
x_delta: float
n_frames: int
n_rows_declared: int
y_positions: list # declared length; may exceed what's actually on disk
row_bytes: int
data_offset: int # byte offset into the file where this angle's data starts
n_rows_available: int = 0
data_size_available: int = 0
complete: bool = True
@property
def data_size_declared(self) -> int:
return self.row_bytes * self.n_rows_declared
class SrasScanFile:
"""Parsed view of a v6 .sras file's header/tables plus per-angle data offsets."""
def __init__(self, path: Path):
self.path = Path(path)
self._parse()
def _parse(self):
file_size = self.path.stat().st_size
with open(self.path, "rb") as f:
raw = f.read(HDR_SIZE)
if len(raw) < HDR_SIZE:
raise ValueError(f"{self.path.name}: file too short for a valid header")
(magic, version, n_angles, x_start_nom, y_start_nom, x_delta_nom,
y_delta_nom, row_spacing, velocity, laser_freq, samples_per_frame,
sample_rate, bytes_per_sample, n_channels) = struct.unpack(HDR_FMT, raw)
if magic != BLOB_MAGIC:
raise ValueError(f"{self.path.name}: bad magic {magic!r}, not a .sras file")
if version != BLOB_VERSION:
raise ValueError(
f"{self.path.name}: unsupported format version {version} "
f"(this tool only supports v{BLOB_VERSION})")
self.x_start_nominal = x_start_nom
self.y_start_nominal = y_start_nom
self.x_delta_nominal = x_delta_nom
self.y_delta_nominal = y_delta_nom
self.row_spacing_mm = row_spacing
self.velocity_mm_s = velocity
self.laser_freq_hz = laser_freq
self.samples_per_frame = samples_per_frame
self.sample_rate_hz = sample_rate
self.bytes_per_sample = bytes_per_sample
self.n_channels = n_channels
angles = list(struct.unpack(f">{n_angles}f", f.read(4 * n_angles)))
geoms = []
for _ in range(n_angles):
x_start, x_delta, n_frames, n_rows = struct.unpack(GEOM_FMT, f.read(GEOM_SIZE))
geoms.append((x_start, x_delta, n_frames, n_rows))
row_tables = []
for (_, _, _, n_rows) in geoms:
row_tables.append(list(struct.unpack(f">{n_rows}f", f.read(4 * n_rows))))
preambles_raw = []
for _ in range(n_channels):
(plen,) = struct.unpack(">H", f.read(2))
preambles_raw.append(f.read(plen))
self.preambles_raw = preambles_raw
(n_bg,) = struct.unpack(">I", f.read(4))
self.background_raw = f.read(n_bg)
data_start_offset = f.tell()
# Build angle entries and compute what's actually present on disk,
# in case the file was closed early (aborted scan) — see scan_format.md's
# "Incomplete files" note. Waveform data is angle-major/row-minor with a
# fixed per-row byte count within an angle, so we walk cumulative offsets.
self.angles = []
cursor = data_start_offset
truncated_seen = False
for i, (angle, (x_start, x_delta, n_frames, n_rows)) in enumerate(zip(angles, geoms)):
row_bytes = n_channels * n_frames * samples_per_frame * bytes_per_sample
entry = AngleEntry(
index=i, angle_deg=angle, x_start=x_start, x_delta=x_delta,
n_frames=n_frames, n_rows_declared=n_rows,
y_positions=row_tables[i], row_bytes=row_bytes,
data_offset=cursor,
)
if truncated_seen:
entry.n_rows_available = 0
entry.data_size_available = 0
entry.complete = False
else:
declared_bytes = row_bytes * n_rows
if row_bytes > 0 and cursor + declared_bytes <= file_size:
entry.n_rows_available = n_rows
entry.data_size_available = declared_bytes
entry.complete = True
cursor += declared_bytes
else:
remaining = max(0, file_size - cursor)
n_complete = remaining // row_bytes if row_bytes > 0 else 0
entry.n_rows_available = n_complete
entry.data_size_available = n_complete * row_bytes
entry.complete = (n_complete == n_rows)
cursor += entry.data_size_available
truncated_seen = True
self.angles.append(entry)
self.data_start_offset = data_start_offset
self.file_size = file_size
def get(self, index: int) -> AngleEntry:
return self.angles[index]
# ---------------------------------------------------------------------------
# Export / delete
# ---------------------------------------------------------------------------
def _write_subset(sf: SrasScanFile, indices: list, dst_path: Path) -> list:
"""Write a new v6 .sras file containing only the given angle indices
(in the given order). Returns a list of warning strings (e.g. for
angles that were truncated on disk and thus exported with fewer rows
than declared).
"""
warnings = []
selected = [sf.get(i) for i in indices]
header = struct.pack(
HDR_FMT, BLOB_MAGIC, BLOB_VERSION, len(selected),
sf.x_start_nominal, sf.y_start_nominal,
sf.x_delta_nominal, sf.y_delta_nominal,
sf.row_spacing_mm, sf.velocity_mm_s, sf.laser_freq_hz,
sf.samples_per_frame, sf.sample_rate_hz,
sf.bytes_per_sample, sf.n_channels,
)
with open(sf.path, "rb") as src, open(dst_path, "wb") as dst:
dst.write(header)
dst.write(struct.pack(f">{len(selected)}f", *[e.angle_deg for e in selected]))
for e in selected:
if e.n_rows_available != e.n_rows_declared:
warnings.append(
f"angle[{e.index}] ({e.angle_deg:.2f} deg): declared "
f"{e.n_rows_declared} rows but only {e.n_rows_available} "
f"present on disk — exporting truncated")
dst.write(struct.pack(GEOM_FMT, e.x_start, e.x_delta, e.n_frames,
e.n_rows_available))
for e in selected:
ys = e.y_positions[:e.n_rows_available]
dst.write(struct.pack(f">{len(ys)}f", *ys))
for praw in sf.preambles_raw:
dst.write(struct.pack(">H", len(praw)))
dst.write(praw)
dst.write(struct.pack(">I", len(sf.background_raw)))
dst.write(sf.background_raw)
for e in selected:
src.seek(e.data_offset)
remaining = e.data_size_available
chunk_size = 1 << 20
while remaining > 0:
chunk = src.read(min(chunk_size, remaining))
if not chunk:
break
dst.write(chunk)
remaining -= len(chunk)
return warnings
def export_angles(sf: SrasScanFile, indices: list, dst_path: Path) -> list:
if not indices:
raise ValueError("no angles selected to export")
return _write_subset(sf, indices, dst_path)
def delete_angles(sf: SrasScanFile, indices_to_delete: list, backup: bool = True) -> Path | None:
"""Rewrite sf.path in place, keeping every angle NOT in indices_to_delete.
Returns the backup file path if one was made, else None.
"""
keep = [e.index for e in sf.angles if e.index not in set(indices_to_delete)]
if not keep:
raise ValueError("refusing to delete every angle — a .sras file needs at least one")
tmp_path = sf.path.with_suffix(sf.path.suffix + ".tmp")
_write_subset(sf, keep, tmp_path)
backup_path = None
if backup:
stamp = datetime.now().strftime("%Y%m%d%H%M%S")
backup_path = sf.path.with_name(f"{sf.path.stem}.bak-{stamp}{sf.path.suffix}")
sf.path.rename(backup_path)
tmp_path.replace(sf.path)
return backup_path
# ---------------------------------------------------------------------------
# Formatting helpers
# ---------------------------------------------------------------------------
def _human_size(n: int) -> str:
size = float(n)
for unit in ("B", "KB", "MB", "GB", "TB"):
if size < 1024.0:
return f"{size:.1f} {unit}"
size /= 1024.0
return f"{size:.1f} PB"
def parse_index_spec(spec: str, max_index: int) -> list:
"""Parse '0,2,4-6' or 'all' into a sorted list of unique in-range indices."""
spec = spec.strip().lower()
if spec in ("all", "*"):
return list(range(max_index + 1))
if not spec:
return []
out = set()
for part in spec.split(","):
part = part.strip()
if not part:
continue
if "-" in part:
lo, hi = part.split("-", 1)
lo, hi = int(lo), int(hi)
if lo > hi:
lo, hi = hi, lo
for i in range(lo, hi + 1):
out.add(i)
else:
out.add(int(part))
bad = [i for i in out if i < 0 or i > max_index]
if bad:
raise ValueError(f"index out of range (0-{max_index}): {sorted(bad)}")
return sorted(out)
def print_summary(sf: SrasScanFile, selected: set):
print()
print(f"File: {sf.path} (v{BLOB_VERSION}, {_human_size(sf.file_size)})")
print(f"Nominal ROI: x_start={sf.x_start_nominal:.4f} x_delta={sf.x_delta_nominal:.4f} "
f"y_start={sf.y_start_nominal:.4f} y_delta={sf.y_delta_nominal:.4f} mm "
f"row_spacing={sf.row_spacing_mm:.4f} mm")
print(f"Velocity={sf.velocity_mm_s:.2f} mm/s Laser={sf.laser_freq_hz:.1f} Hz "
f"Samples/frame={sf.samples_per_frame} Sample rate={sf.sample_rate_hz:.3e} Hz "
f"Channels={sf.n_channels} Bytes/sample={sf.bytes_per_sample}")
print()
hdr = f"{'':>2} {'#':>3} {'Angle(deg)':>10} {'Rows':>7} {'Frames/row':>10} {'x_start':>9} {'x_delta':>9} {'Data size':>11} Status"
print(hdr)
print("-" * len(hdr))
for e in sf.angles:
mark = "*" if e.index in selected else " "
if e.complete:
status = "OK"
elif e.n_rows_available == 0:
status = "MISSING (no data on disk)"
else:
status = f"TRUNCATED ({e.n_rows_available}/{e.n_rows_declared} rows on disk)"
print(f"{mark:>2} {e.index:>3} {e.angle_deg:>10.2f} {e.n_rows_declared:>7} "
f"{e.n_frames:>10} {e.x_start:>9.3f} {e.x_delta:>9.3f} "
f"{_human_size(e.data_size_available):>11} {status}")
print()
# ---------------------------------------------------------------------------
# Interactive TUI
# ---------------------------------------------------------------------------
def interactive_loop(path: Path):
sf = SrasScanFile(path)
selected: set = set()
help_text = (
" [l] list show the angle table again\n"
" [s] select <spec> set selection, e.g. '0,2,4-6' or 'all' or 'none'\n"
" [e] export <path> write selected angles to a new .sras file\n"
" [d] delete remove selected angles from this file in place\n"
" [r] reload re-read the file from disk (after external changes)\n"
" [h] help show this help\n"
" [q] quit"
)
print_summary(sf, selected)
print(help_text)
while True:
try:
cmd_line = input("\nsras> ").strip()
except EOFError:
print()
break
if not cmd_line:
continue
parts = cmd_line.split(None, 1)
cmd = parts[0].lower()
arg = parts[1].strip() if len(parts) > 1 else ""
try:
if cmd in ("q", "quit", "exit"):
break
elif cmd in ("h", "help", "?"):
print(help_text)
elif cmd in ("l", "list"):
print_summary(sf, selected)
elif cmd in ("s", "select"):
if not arg:
arg = input("Angles to select (e.g. 0,2,4-6 / all / none): ").strip()
if arg.lower() == "none":
selected = set()
else:
selected = set(parse_index_spec(arg, len(sf.angles) - 1))
print(f"Selected {len(selected)} angle(s): {sorted(selected)}")
elif cmd in ("e", "export"):
if not selected:
print("Nothing selected — use 's' first.")
continue
dst = arg or input("Output path: ").strip()
if not dst:
print("Export cancelled — no path given.")
continue
dst_path = Path(dst)
if dst_path.exists():
ans = input(f"{dst_path} exists — overwrite? [y/N] ").strip().lower()
if ans != "y":
print("Export cancelled.")
continue
warnings = export_angles(sf, sorted(selected), dst_path)
print(f"Exported {len(selected)} angle(s) -> {dst_path}")
for w in warnings:
print(f" warning: {w}")
elif cmd in ("d", "delete"):
if not selected:
print("Nothing selected — use 's' first.")
continue
print(f"About to delete {len(selected)} angle(s) from {sf.path}: {sorted(selected)}")
ans = input("Type 'yes' to confirm (a timestamped .bak copy will be kept): ").strip()
if ans != "yes":
print("Delete cancelled.")
continue
backup_path = delete_angles(sf, sorted(selected), backup=True)
print(f"Deleted. Backup saved to {backup_path}")
sf = SrasScanFile(sf.path)
selected = set()
print_summary(sf, selected)
elif cmd in ("r", "reload"):
sf = SrasScanFile(sf.path)
selected = set()
print_summary(sf, selected)
else:
print(f"Unknown command: {cmd!r} (type 'h' for help)")
except Exception as exc:
print(f"Error: {exc}")
# ---------------------------------------------------------------------------
# CLI entry point
# ---------------------------------------------------------------------------
def main():
ap = argparse.ArgumentParser(
description="Inspect, export, or delete per-angle sub-scans in a v6 .sras file.")
ap.add_argument("file", type=Path, help="path to a .sras file")
ap.add_argument("--list", action="store_true", help="print the angle table and exit")
ap.add_argument("--export", metavar="SPEC", help="angle index spec to export, e.g. '0,2,4-6' or 'all'")
ap.add_argument("--output", metavar="PATH", type=Path, help="destination path for --export")
ap.add_argument("--delete", metavar="SPEC", help="angle index spec to delete in place, e.g. '1,3'")
ap.add_argument("--no-backup", action="store_true", help="skip the .bak copy when using --delete")
ap.add_argument("--yes", action="store_true", help="don't prompt for confirmation on --delete")
args = ap.parse_args()
if not args.file.exists():
print(f"error: {args.file} does not exist", file=sys.stderr)
sys.exit(1)
try:
sf = SrasScanFile(args.file)
except ValueError as exc:
print(f"error: {exc}", file=sys.stderr)
sys.exit(1)
non_interactive = args.list or args.export or args.delete
if args.list:
print_summary(sf, set())
try:
if args.export:
indices = parse_index_spec(args.export, len(sf.angles) - 1)
if not args.output:
print("error: --export requires --output", file=sys.stderr)
sys.exit(1)
if args.output.exists() and not args.yes:
ans = input(f"{args.output} exists — overwrite? [y/N] ").strip().lower()
if ans != "y":
print("Export cancelled.")
sys.exit(1)
warnings = export_angles(sf, indices, args.output)
print(f"Exported {len(indices)} angle(s) -> {args.output}")
for w in warnings:
print(f" warning: {w}")
if args.delete:
indices = parse_index_spec(args.delete, len(sf.angles) - 1)
if not indices:
print("error: --delete requires a non-empty angle spec", file=sys.stderr)
sys.exit(1)
if not args.yes:
print(f"About to delete {len(indices)} angle(s) from {sf.path}: {indices}")
ans = input("Type 'yes' to confirm: ").strip()
if ans != "yes":
print("Delete cancelled.")
sys.exit(1)
backup_path = delete_angles(sf, indices, backup=not args.no_backup)
if backup_path:
print(f"Deleted. Backup saved to {backup_path}")
else:
print("Deleted (no backup kept).")
except ValueError as exc:
print(f"error: {exc}", file=sys.stderr)
sys.exit(1)
if not non_interactive:
interactive_loop(args.file)
if __name__ == "__main__":
main()
Regular → Executable
View File
Regular → Executable
View File
Regular → Executable
View File