diff --git a/core/__init__.py b/core/__init__.py new file mode 100644 index 0000000..85e0a8e --- /dev/null +++ b/core/__init__.py @@ -0,0 +1 @@ +"""Headless scan-engine core: importable without PyQt6 or any vendor SDK.""" diff --git a/core/config.py b/core/config.py new file mode 100644 index 0000000..43c8385 --- /dev/null +++ b/core/config.py @@ -0,0 +1,47 @@ +"""Persisted user defaults (ports, scope IP, save directory). + +One flat JSON file, one dataclass. ``save()`` always writes every field, so +a partial UI update can never silently drop another field's saved value +(which is exactly what the old dict-based writer did to helios_port). +""" +from __future__ import annotations + +import json +import logging +from dataclasses import asdict, dataclass, fields +from pathlib import Path + +logger = logging.getLogger(__name__) + +DEFAULTS_PATH = Path(__file__).resolve().parent.parent / "aui_defaults.json" + + +@dataclass +class ScanDefaults: + t3r_port: str = "/dev/ttyUSB0" + bbd_port: str = "/dev/ttyUSB1" + oscope_ip: str = "192.168.0.1" + save_dir: str = str(DEFAULTS_PATH.parent / "scans") + helios_port: str = "/dev/ttyUSB2" + + @classmethod + def load(cls, path: Path = DEFAULTS_PATH) -> "ScanDefaults": + """Load defaults, tolerating a missing/corrupt file and unknown keys.""" + if path.exists(): + try: + with open(path) as f: + data = json.load(f) + known = {f.name for f in fields(cls)} + return cls(**{k: v for k, v in data.items() if k in known}) + except (OSError, ValueError, TypeError) as e: + logger.warning("Could not load %s (%s); using fallback defaults", path, e) + inst = cls() + inst.save(path) + return inst + + def save(self, path: Path = DEFAULTS_PATH) -> None: + try: + with open(path, "w") as f: + json.dump(asdict(self), f, indent=2) + except OSError as e: + logger.warning("Could not save defaults to %s: %s", path, e) diff --git a/core/scan_geometry.py b/core/scan_geometry.py new file mode 100644 index 0000000..c3ee43b --- /dev/null +++ b/core/scan_geometry.py @@ -0,0 +1,199 @@ +"""Scan geometry planning: angle sequences, rotated bounding boxes, travel +limits, and ETA math. Pure Python — no Qt, no hardware. +""" +from __future__ import annotations + +import math +import time +from collections import deque +from dataclasses import dataclass, field + + +class ScanGeometryError(ValueError): + """Scan geometry that cannot be executed (bad inputs or off-stage).""" + + +@dataclass(frozen=True) +class StageLimits: + """Usable travel of the scanning stage in mm (MLS203-1).""" + x_min: float = 0.0 + x_max: float = 110.0 + y_min: float = 0.0 + y_max: float = 75.0 + + +DEFAULT_STAGE_LIMITS = StageLimits() + + +@dataclass +class AngleGeometry: + """One rotation angle's scan extent — also the on-disk v6 geometry row.""" + angle_deg: float + x_start: float + x_delta: float + n_frames: int + n_rows: int + y_positions: list[float] = field(default_factory=list) + + +@dataclass +class ScanPlan: + x_start_nominal: float + y_start_nominal: float + x_delta_nominal: float + y_delta_nominal: float + row_spacing: float + velocity_mm_s: float + laser_freq_hz: float + per_angle: list[AngleGeometry] = field(default_factory=list) + + @property + def n_angles(self) -> int: + return len(self.per_angle) + + @property + def angles(self) -> list[float]: + return [pa.angle_deg for pa in self.per_angle] + + @property + def total_rows(self) -> int: + return sum(pa.n_rows for pa in self.per_angle) + + +def build_plan(x_start: float, y_start: float, x_delta: float, y_delta: float, + num_angles: int, row_spacing: float, *, + laser_freq_hz: float, velocity_mm_s: float, + rotation_sign: int = -1) -> ScanPlan: + """Compute the per-angle scan geometry for a nominal ROI. + + Each angle only needs to physically scan the bounding box of the nominal + (x_start, y_start, x_delta, y_delta) rectangle rotated by THAT angle -- + not the worst case across all angles -- so the X extent (and therefore + points/row) and the row count are computed per angle. + """ + if x_delta <= 0: + raise ScanGeometryError("XD must be > 0") + if row_spacing <= 0: + raise ScanGeometryError("RowSpacing must be > 0") + if num_angles < 1: + raise ScanGeometryError("NumAngles must be ≥ 1") + + # Signed so the recorded/commanded angle sequence reflects the GR + # stage's actual physical rotation direction. + if num_angles > 1: + angles = [rotation_sign * i * 180.0 / (num_angles - 1) for i in range(num_angles)] + else: + angles = [0.0] + + cx = x_start + x_delta / 2.0 + cy = y_start + y_delta / 2.0 + per_angle = [] + for a in angles: + r = math.radians(a) + bb_w = abs(x_delta * math.cos(r)) + abs(y_delta * math.sin(r)) + bb_h = abs(x_delta * math.sin(r)) + abs(y_delta * math.cos(r)) + a_x_start = cx - bb_w / 2.0 + a_y_start = cy - bb_h / 2.0 + a_n_rows = max(1, round(bb_h / row_spacing) + 1) if bb_h > 0 else 1 + a_n_frames = max(1, round(bb_w * laser_freq_hz / velocity_mm_s)) + per_angle.append(AngleGeometry( + angle_deg=a, + x_start=a_x_start, + x_delta=bb_w, + n_frames=a_n_frames, + n_rows=a_n_rows, + y_positions=[a_y_start + i * row_spacing for i in range(a_n_rows)], + )) + + return ScanPlan( + x_start_nominal=x_start, y_start_nominal=y_start, + x_delta_nominal=x_delta, y_delta_nominal=y_delta, + row_spacing=row_spacing, + velocity_mm_s=velocity_mm_s, laser_freq_hz=laser_freq_hz, + per_angle=per_angle, + ) + + +def validate_plan(plan: ScanPlan, ramp_mm: float, ramp_buffer_mm: float, + limits: StageLimits = DEFAULT_STAGE_LIMITS) -> None: + """Raise ScanGeometryError if any angle's physical move leaves the stage. + + The actual X move starts one ramp-length + buffer before x_start and ends + one ramp-length + buffer after x_start + x_delta, so the stage is at full + velocity across the whole data window. + """ + x_ramp_total = ramp_mm + ramp_buffer_mm + for pa in plan.per_angle: + x_move_start = pa.x_start - x_ramp_total + x_move_end = pa.x_start + pa.x_delta + x_ramp_total + if x_move_start < limits.x_min: + raise ScanGeometryError( + f"Angle {pa.angle_deg:.1f}°: scan pre-ramp start ({x_move_start:.3f} mm) " + f"is below the X axis minimum ({limits.x_min:g} mm). Reduce XD/YD or move XS/YS " + f"so every rotation angle's bounding box stays on-stage " + f"(SCAN_RAMP_MM={ramp_mm:.3f} + SCAN_RAMP_BUFFER_MM={ramp_buffer_mm:.3f})." + ) + if x_move_end > limits.x_max: + raise ScanGeometryError( + f"Angle {pa.angle_deg:.1f}°: scan run-off end ({x_move_end:.3f} mm) " + f"exceeds the X axis maximum ({limits.x_max:g} mm). Reduce XD/YD or move XS/YS " + f"so every rotation angle's bounding box stays on-stage." + ) + y_min = min(pa.y_positions) + y_max = max(pa.y_positions) + if y_min < limits.y_min: + raise ScanGeometryError( + f"Angle {pa.angle_deg:.1f}°: scan Y range starts at {y_min:.3f} mm, " + f"below the Y axis minimum ({limits.y_min:g} mm)." + ) + if y_max > limits.y_max: + raise ScanGeometryError( + f"Angle {pa.angle_deg:.1f}°: scan Y range ends at {y_max:.3f} mm, " + f"exceeds the Y axis maximum ({limits.y_max:g} mm)." + ) + + +def format_eta(secs: float) -> str: + secs = max(0.0, secs) + m, s = divmod(int(secs), 60) + h, m = divmod(m, 60) + if h > 0: + return f"{h}h {m:02d}m" + if m > 0: + return f"{m}m {s:02d}s" + return f"{s}s" + + +class EtaEstimator: + """Rolling average of recent row durations → remaining-time estimate. + + Duration history resets when the angle index changes, since different + angles have different row lengths. + """ + + def __init__(self, window: int = 5): + self._durations: deque[float] = deque(maxlen=window) + self._row_start: float | None = None + self._last_angle_idx: int = -1 + + def reset(self) -> None: + self._durations.clear() + self._row_start = None + self._last_angle_idx = -1 + + def row_started(self, now: float | None = None) -> None: + self._row_start = time.monotonic() if now is None else now + + def row_finished(self, angle_idx: int, now: float | None = None) -> None: + if angle_idx != self._last_angle_idx and self._last_angle_idx != -1: + self._durations.clear() + self._last_angle_idx = angle_idx + if self._row_start is not None: + end = time.monotonic() if now is None else now + self._durations.append(end - self._row_start) + self._row_start = None + + def eta_secs(self, rows_left: int) -> float | None: + if not self._durations or rows_left <= 0: + return None + return sum(self._durations) / len(self._durations) * rows_left diff --git a/core/sras_format.py b/core/sras_format.py new file mode 100644 index 0000000..5dfbe7e --- /dev/null +++ b/core/sras_format.py @@ -0,0 +1,330 @@ +"""SRAS v6 binary scan-file format — the single implementation. + +Full byte-level spec: scan_format.md. Summary: + + header >4sBHfffffffIdBB magic ver n_angles xs_nom ys_nom xd_nom yd_nom + row_spacing velocity laser_freq spf sample_rate + bytes_per_sample n_channels + angle table n_angles × >f + geometry table n_angles × >ffIH (x_start x_delta n_frames n_rows) + row tables (ragged) per angle: n_rows × >f (y positions) + preambles n_channels × (>H length + utf-8 WFMOutpre string) + background block >I length + raw int8 CH1 average + waveform data angle-major, row-minor, channel-inner: + for each angle, for each row, for each channel, + n_frames × samples_per_frame × bytes_per_sample + +Incomplete files are valid: the data block is one contiguous append-only +stream, so the readable prefix defines a single frontier past which nothing +has been written yet (see ``SrasFile.angle_status``). +""" +from __future__ import annotations + +import mmap +import struct +from dataclasses import dataclass, field +from pathlib import Path +from typing import BinaryIO + +import numpy as np + +from core.scan_geometry import AngleGeometry, ScanPlan + +MAGIC = b"SRAS" +VERSION = 6 +HDR_FMT = ">4sBHfffffffIdBB" +HDR_SIZE = struct.calcsize(HDR_FMT) # 49 bytes +GEOM_FMT = ">ffIH" +GEOM_SIZE = struct.calcsize(GEOM_FMT) # 14 bytes + +# Oscilloscope channels recorded, in on-disk order. +SCAN_CHANNELS = [1, 3, 4] + +STATUS_OK = "OK" +STATUS_TRUNCATED = "TRUNCATED" +STATUS_MISSING = "MISSING" + + +@dataclass +class ScanHeader: + """The fixed v6 global header (everything but magic/version).""" + n_angles: int + x_start_nominal: float + y_start_nominal: float + x_delta_nominal: float + y_delta_nominal: float + row_spacing: float + velocity: float + laser_freq: float + samples_per_frame: int + sample_rate: float + bytes_per_sample: int + n_channels: int + + +@dataclass +class AngleStatus: + """How much of one angle's declared data is actually on disk.""" + index: int + angle_deg: float + n_rows: int # declared + row_bytes: int + data_offset: int + n_rows_available: int + status: str # STATUS_OK / STATUS_TRUNCATED / STATUS_MISSING + + @property + def complete(self) -> bool: + return self.status == STATUS_OK + + +def create_scan_file(path: Path, plan: ScanPlan, samples_per_frame: int, + sample_rate: float, preambles: list[str], + background_waveform: bytes) -> BinaryIO: + """Create a new .sras file and write the v6 header + tables. + + Returns an open binary file positioned at the start of the data block; + the caller appends waveform rows and must close it (try/finally). + """ + path.parent.mkdir(parents=True, exist_ok=True) + f = open(path, "wb") + f.write(struct.pack( + HDR_FMT, MAGIC, VERSION, + plan.n_angles, + plan.x_start_nominal, plan.y_start_nominal, + plan.x_delta_nominal, plan.y_delta_nominal, + plan.row_spacing, + plan.velocity_mm_s, plan.laser_freq_hz, + samples_per_frame, + sample_rate, + 1, # bytes_per_sample: int8 from scope default + len(SCAN_CHANNELS), + )) + f.write(struct.pack(f">{plan.n_angles}f", *plan.angles)) + for pa in plan.per_angle: + f.write(struct.pack(GEOM_FMT, pa.x_start, pa.x_delta, pa.n_frames, pa.n_rows)) + for pa in plan.per_angle: + f.write(struct.pack(f">{pa.n_rows}f", *pa.y_positions)) + for p in preambles: + enc = p.encode("utf-8") + f.write(struct.pack(">H", len(enc))) + f.write(enc) + f.write(struct.pack(">I", len(background_waveform))) + f.write(background_waveform) + return f + + +@dataclass +class SrasFile: + """Parsed v6 .sras file: header, tables, and lazy (memmap) data access. + + Parsing reads only the header/tables — never the waveform block — so + opening a multi-GB file is cheap. ``load_angle``/``load_row`` return + read-only numpy views backed by a shared mmap; no data is copied until + the caller computes on it. + """ + path: Path + header: ScanHeader = field(init=False) + per_angle: list[AngleGeometry] = field(init=False) + preambles: list[str] = field(init=False) + preambles_raw: list[bytes] = field(init=False) + background: bytes = field(init=False) + data_start_offset: int = field(init=False) + file_size: int = field(init=False) + + def __post_init__(self): + self.path = Path(self.path) + self._mmap: mmap.mmap | None = None + self._parse() + + def _parse(self): + 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 to contain a valid header") + (magic, version, n_angles, x_start_nominal, y_start_nominal, + x_delta_nominal, y_delta_nominal, row_spacing, velocity, laser_freq, + samples_per_frame, sample_rate, bytes_per_sample, + n_channels) = struct.unpack(HDR_FMT, raw) + if magic != MAGIC: + raise ValueError(f"{self.path.name}: not a valid SRAS file (bad magic)") + if version != VERSION: + raise ValueError( + f"{self.path.name}: unsupported SRAS format version {version} " + f"(only version {VERSION} is supported)" + ) + self.header = ScanHeader( + n_angles=n_angles, + x_start_nominal=x_start_nominal, y_start_nominal=y_start_nominal, + x_delta_nominal=x_delta_nominal, y_delta_nominal=y_delta_nominal, + row_spacing=row_spacing, velocity=velocity, laser_freq=laser_freq, + samples_per_frame=samples_per_frame, sample_rate=sample_rate, + bytes_per_sample=bytes_per_sample, n_channels=n_channels, + ) + + angles = struct.unpack(f">{n_angles}f", f.read(4 * n_angles)) + + self.per_angle = [] + for a in angles: + x_start, x_delta, n_frames, n_rows = struct.unpack(GEOM_FMT, f.read(GEOM_SIZE)) + self.per_angle.append(AngleGeometry( + angle_deg=a, x_start=x_start, x_delta=x_delta, + n_frames=n_frames, n_rows=n_rows, + )) + + for pa in self.per_angle: + pa.y_positions = list(struct.unpack(f">{pa.n_rows}f", f.read(4 * pa.n_rows))) + + self.preambles_raw = [] + for _ in range(n_channels): + (plen,) = struct.unpack(">H", f.read(2)) + self.preambles_raw.append(f.read(plen)) + self.preambles = [p.decode("utf-8", errors="replace") for p in self.preambles_raw] + + (n_bg,) = struct.unpack(">I", f.read(4)) + self.background = f.read(n_bg) + + self.data_start_offset = f.tell() + + # ── Frontier / truncation analysis ─────────────────────────────────────── + + def row_bytes(self, angle_idx: int) -> int: + pa = self.per_angle[angle_idx] + return (self.header.n_channels * pa.n_frames + * self.header.samples_per_frame * self.header.bytes_per_sample) + + def angle_status(self) -> list[AngleStatus]: + """Walk declared per-row byte counts against the actual file size. + + Because the data is one contiguous append-only stream, once an angle + is found short every later angle is necessarily absent too — there is + a single frontier past which nothing has been written yet. + """ + statuses = [] + cursor = self.data_start_offset + frontier_seen = False + for ai, pa in enumerate(self.per_angle): + row_bytes = self.row_bytes(ai) + data_offset = cursor + if frontier_seen: + n_rows_available = 0 + status = STATUS_MISSING + else: + declared_bytes = row_bytes * pa.n_rows + if row_bytes > 0 and cursor + declared_bytes <= self.file_size: + n_rows_available = pa.n_rows + status = STATUS_OK + cursor += declared_bytes + else: + remaining = max(0, self.file_size - cursor) + n_rows_available = remaining // row_bytes if row_bytes > 0 else 0 + status = STATUS_MISSING if n_rows_available == 0 else STATUS_TRUNCATED + frontier_seen = True + statuses.append(AngleStatus( + index=ai, angle_deg=pa.angle_deg, n_rows=pa.n_rows, + row_bytes=row_bytes, data_offset=data_offset, + n_rows_available=n_rows_available, status=status, + )) + return statuses + + def angle_data_offset(self, angle_idx: int) -> int: + offset = self.data_start_offset + for ai in range(angle_idx): + offset += self.row_bytes(ai) * self.per_angle[ai].n_rows + return offset + + # ── Lazy data access ───────────────────────────────────────────────────── + + def _ensure_mmap(self) -> mmap.mmap: + if self._mmap is None: + # The mapping stays valid after the file object is closed, so + # don't hold the descriptor open for the (long) life of a viewer + # session. + with open(self.path, "rb") as f: + self._mmap = mmap.mmap(f.fileno(), 0, access=mmap.ACCESS_READ) + return self._mmap + + def _dtype(self) -> np.dtype: + return np.dtype(np.int16 if self.header.bytes_per_sample == 2 else np.int8) + + def load_angle(self, angle_idx: int, n_rows: int | None = None) -> np.ndarray: + """Read-only view of one angle's data block, shape + (n_rows, n_channels, n_frames, samples_per_frame). + + ``n_rows`` limits the view to the rows actually on disk (pass + ``AngleStatus.n_rows_available`` for truncated files); default is the + declared row count. + """ + pa = self.per_angle[angle_idx] + h = self.header + if n_rows is None: + n_rows = pa.n_rows + start = self.angle_data_offset(angle_idx) + count = n_rows * h.n_channels * pa.n_frames * h.samples_per_frame + arr = np.frombuffer(self._ensure_mmap(), dtype=self._dtype(), + count=count, offset=start) + arr = arr.reshape(n_rows, h.n_channels, pa.n_frames, h.samples_per_frame) + arr.flags.writeable = False + return arr + + def load_row(self, angle_idx: int, row: int, channel_idx: int) -> np.ndarray: + """Read-only view of one row/channel, shape (n_frames, samples_per_frame).""" + pa = self.per_angle[angle_idx] + h = self.header + ch_bytes = pa.n_frames * h.samples_per_frame * h.bytes_per_sample + start = (self.angle_data_offset(angle_idx) + row * self.row_bytes(angle_idx) + + channel_idx * ch_bytes) + arr = np.frombuffer(self._ensure_mmap(), dtype=self._dtype(), + count=pa.n_frames * h.samples_per_frame, offset=start) + arr = arr.reshape(pa.n_frames, h.samples_per_frame) + arr.flags.writeable = False + return arr + + def close(self): + """Release this file's hold on the mapping. + + Views handed out earlier stay valid — they keep the mapping alive + until they are garbage-collected, at which point the OS frees it. + """ + if self._mmap is not None: + try: + self._mmap.close() + except BufferError: + pass # live numpy views still reference the buffer + self._mmap = None + + def __enter__(self): + return self + + def __exit__(self, exc_type, exc_val, exc_tb): + self.close() + + # ── Axes helpers (viewer conveniences, derived from header fields) ─────── + + def pixel_pitch_x_mm(self) -> float: + """Distance between adjacent frames along X.""" + return self.header.velocity / self.header.laser_freq + + def x_axis_mm(self, angle_idx: int) -> np.ndarray: + pa = self.per_angle[angle_idx] + return pa.x_start + np.arange(pa.n_frames) * self.pixel_pitch_x_mm() + + def time_axis_ns(self) -> np.ndarray: + h = self.header + return np.arange(h.samples_per_frame) / h.sample_rate * 1e9 + + def freq_axis_mhz(self, nfft: int) -> np.ndarray: + return np.fft.rfftfreq(nfft, d=1.0 / self.header.sample_rate) / 1e6 + + +def plan_from_header(sras: SrasFile) -> ScanPlan: + """Reconstruct the ScanPlan a file was written with (for resume).""" + h = sras.header + return ScanPlan( + x_start_nominal=h.x_start_nominal, y_start_nominal=h.y_start_nominal, + x_delta_nominal=h.x_delta_nominal, y_delta_nominal=h.y_delta_nominal, + row_spacing=h.row_spacing, + velocity_mm_s=h.velocity, laser_freq_hz=h.laser_freq, + per_angle=list(sras.per_angle), + ) diff --git a/hardware/helios_laser.py b/hardware/helios_laser.py index f7ac118..1f12117 100755 --- a/hardware/helios_laser.py +++ b/hardware/helios_laser.py @@ -3,12 +3,13 @@ Helios Laser System Driver Basic implementation for controlling the Helios pulsed laser. """ -import serial import time import logging from typing import Optional, List from enum import Enum +from hardware.serial_util import open_8n1 + logger = logging.getLogger(__name__) @@ -42,10 +43,9 @@ class HeliosLaser: @staticmethod def list_available_ports() -> List[str]: - """List available serial ports""" - import serial.tools.list_ports - ports = serial.tools.list_ports.comports() - return [port.device for port in ports] + """List available serial ports, likeliest devices first.""" + from hardware.serial_util import list_port_devices + return list_port_devices() def connect(self, port: str = None) -> bool: """ @@ -65,14 +65,7 @@ class HeliosLaser: return False try: - self.serial = serial.Serial( - port=self.port, - baudrate=9600, - bytesize=serial.EIGHTBITS, - parity=serial.PARITY_NONE, - stopbits=serial.STOPBITS_ONE, - timeout=self.timeout - ) + self.serial = open_8n1(self.port, baudrate=9600, timeout=self.timeout) time.sleep(0.1) # Allow time for connection to stabilize self.is_connected = True logger.info(f"Connected to Helios laser on {self.port}") diff --git a/hardware/serial_util.py b/hardware/serial_util.py new file mode 100644 index 0000000..76f8e75 --- /dev/null +++ b/hardware/serial_util.py @@ -0,0 +1,44 @@ +"""Shared serial-port helpers: 8N1 open and scored port enumeration. + +Qt-free — GUI code adapts the (device, label) list into its own widgets. +""" +from __future__ import annotations + +import serial +from serial.tools import list_ports + +# Substrings that suggest a USB-serial adapter we actually talk to +# (ESP32-based T3R, CP210x/CH340 dongles, CDC-ACM devices); matching ports +# sort first in pickers. +DEVICE_HINTS = ("esp32", "jtag", "espressif", "usb serial", "cp210", "ch340", "cdc") + + +def open_8n1(port: str, baudrate: int, timeout: float, + write_timeout: float | None = None) -> serial.Serial: + """Open a serial port with the 8N1 framing every device here uses.""" + return serial.Serial( + port=port, + baudrate=baudrate, + bytesize=serial.EIGHTBITS, + parity=serial.PARITY_NONE, + stopbits=serial.STOPBITS_ONE, + timeout=timeout, + write_timeout=write_timeout, + ) + + +def scored_ports() -> list[tuple[str, str]]: + """Enumerate serial ports as (device, human label), likeliest-first.""" + ports = list(list_ports.comports()) + + def score(p): + text = f"{p.description} {p.manufacturer or ''} {p.product or ''}".lower() + return -sum(h in text for h in DEVICE_HINTS) + + ports.sort(key=score) + return [(p.device, f"{p.device} — {p.description or p.device}") for p in ports] + + +def list_port_devices() -> list[str]: + """Plain device-path list, likeliest-first.""" + return [dev for dev, _ in scored_ports()] diff --git a/sc3_aui_app.py b/sc3_aui_app.py index 28d50a5..2a04b43 100755 --- a/sc3_aui_app.py +++ b/sc3_aui_app.py @@ -5,8 +5,6 @@ Loads sc3-aui-main.ui, sc3-aui-camera.ui, sc3-aui-scanprogress.ui via uic and wires up T3R, BBD202, oscilloscope, and camera hardware workers. """ -import json -import math import queue import struct import sys @@ -30,6 +28,11 @@ from matplotlib.figure import Figure ROOT = Path(__file__).parent sys.path.insert(0, str(ROOT)) +from core.config import ScanDefaults +from core.scan_geometry import ScanPlan, EtaEstimator, build_plan, format_eta, validate_plan +from core.sras_format import ( + SCAN_CHANNELS, SrasFile, create_scan_file, plan_from_header, +) from hardware.helios_laser import HeliosLaser from hardware.pybbd202 import AXIS_X, AXIS_Y, ThorlabsServoDriver from hardware.t3r_driver import T3RDriver @@ -39,44 +42,7 @@ from hardware.uc480_camera import ( ) from t3r_control_panel import T3RControlPanel -# ── Config defaults ────────────────────────────────────────────────────────── - -_AUI_DEFAULTS_PATH = ROOT / "aui_defaults.json" -_AUI_DEFAULTS_FALLBACK = { - "t3r_port": "/dev/ttyUSB0", - "bbd_port": "/dev/ttyUSB1", - "oscope_ip": "192.168.0.1", - "laser_freq_hz": 2000, - "save_dir": str(ROOT / "scans"), - "helios_port": "/dev/ttyUSB2", -} - -def _load_aui_defaults() -> dict: - if _AUI_DEFAULTS_PATH.exists(): - try: - with open(_AUI_DEFAULTS_PATH) as f: - return {**_AUI_DEFAULTS_FALLBACK, **json.load(f)} - except Exception: - pass - # File absent or unreadable — write fresh copy and return fallback - _save_aui_defaults(_AUI_DEFAULTS_FALLBACK) - return dict(_AUI_DEFAULTS_FALLBACK) - -def _save_aui_defaults(d: dict) -> None: - try: - with open(_AUI_DEFAULTS_PATH, "w") as f: - json.dump(d, f, indent=2) - except Exception as e: - print(f"[AUI] Could not save defaults: {e}") - -_aui = _load_aui_defaults() - -DEFAULT_T3R_PORT = _aui["t3r_port"] -DEFAULT_BBD_PORT = _aui["bbd_port"] -DEFAULT_OSCOPE_IP = _aui["oscope_ip"] -DEFAULT_LASER_FREQ_HZ = float(_aui["laser_freq_hz"]) -DEFAULT_SAVE_DIR = _aui["save_dir"] -DEFAULT_HELIOS_PORT = _aui.get("helios_port", "/dev/ttyUSB2") +DEFAULTS = ScanDefaults.load() # ── Scan constants ─────────────────────────────────────────────────────────── @@ -155,202 +121,6 @@ class DCBiasImageWidget(FigureCanvas): BBD_JOG_SPEED_MM_S = 10.0 BBD_JOG_ACCEL_MM_S2 = 50.0 -# ── Binary blob format ─────────────────────────────────────────────────────── -# Full spec: scan_format.md -BLOB_MAGIC = b"SRAS" -BLOB_VERSION = 6 -SCAN_CHANNELS = [1, 3, 4] # oscilloscope channels recorded, in order -# v6: each angle scans only the bounding box of the nominal ROI rotated by -# that angle, so x_start/x_delta/n_frames/n_rows all vary per angle and are -# no longer in the fixed header — see the per-angle geometry table. -# ">4s B H f f f f f f f I d B B" -# magic ver n_angles xs_nom ys_nom xd_nom yd_nom row_spacing vel freq spf sr bps n_channels -BLOB_HDR_FMT = ">4sBHfffffffIdBB" - - -def _format_eta(secs: float) -> str: - secs = max(0.0, secs) - m, s = divmod(int(secs), 60) - h, m = divmod(m, 60) - if h > 0: - return f"{h}h {m:02d}m" - if m > 0: - return f"{m}m {s:02d}s" - return f"{s}s" - - -def _open_scan_file(path: Path, angles: list[float], per_angle: list[dict], - x_start_nominal: float, y_start_nominal: float, - x_delta_nominal: float, y_delta_nominal: float, - row_spacing: float, - velocity: float, laser_freq: float, - samples_per_frame: int, sample_rate: float, - preambles: list[str], - background_waveform: bytes): - """Create a new SRAS file and write the v6 global header + per-angle tables. - - Returns an open binary file object positioned at the start of the data - block. The caller must close it (use in a try/finally block). - - Each angle only scans the bounding box of the nominal - (x_start_nominal, y_start_nominal, x_delta_nominal, y_delta_nominal) ROI - rotated by that angle, so x_start, x_delta, n_frames (points/row) and - n_rows all vary per angle. `per_angle` holds one dict per angle (same - order as `angles`) with keys "x_start", "x_delta", "n_frames", "n_rows", - "y_positions". - - Data is written by appending waveforms in angle-major, row-minor, - channel-inner order: for each angle, for each of its rows, for each - channel in SCAN_CHANNELS order, that angle's n_frames waveforms are - written sequentially. - - preambles: one WFMOutpre string per channel (same order as SCAN_CHANNELS), - written as length-prefixed UTF-8 blocks after the row table. - - background_waveform: raw int8 bytes of a 64-sample averaged CH1 waveform - captured with the Helios laser enabled and Genesis laser - disabled, written as uint32 length prefix followed by - the data. - """ - path.parent.mkdir(parents=True, exist_ok=True) - n_angles = len(angles) - f = open(path, "wb") - header = struct.pack( - BLOB_HDR_FMT, - BLOB_MAGIC, BLOB_VERSION, - n_angles, - x_start_nominal, y_start_nominal, - x_delta_nominal, y_delta_nominal, - row_spacing, - velocity, laser_freq, - samples_per_frame, - sample_rate, - 1, # bytes_per_sample: int8 from scope default - len(SCAN_CHANNELS), # n_channels - ) - f.write(header) - f.write(struct.pack(f">{n_angles}f", *angles)) - # Per-angle geometry table: x_start, x_delta, n_frames, n_rows. - for pa in per_angle: - f.write(struct.pack(">ffIH", pa["x_start"], pa["x_delta"], pa["n_frames"], pa["n_rows"])) - # Ragged row table: each angle's y_positions, concatenated in order. - for pa in per_angle: - f.write(struct.pack(f">{pa['n_rows']}f", *pa["y_positions"])) - for p in preambles: - enc = p.encode("utf-8") - f.write(struct.pack(">H", len(enc))) - f.write(enc) - # v4+: background waveform block — CH1 64-sample average (Helios ON, Genesis OFF) - f.write(struct.pack(">I", len(background_waveform))) - f.write(background_waveform) - return f - - -def _read_sras_header(path: Path) -> dict: - """Parse a v6 .sras file's header, per-angle geometry, and row tables. - - Does not read the (potentially huge) waveform data block itself — only - enough to know each angle's geometry and the byte offset at which the - waveform data begins. - """ - with open(path, "rb") as f: - hdr_size = struct.calcsize(BLOB_HDR_FMT) - raw = f.read(hdr_size) - if len(raw) < hdr_size: - raise ValueError(f"{path.name}: file too short to contain a valid header") - (magic, version, n_angles, x_start_nominal, y_start_nominal, - x_delta_nominal, y_delta_nominal, row_spacing, velocity, laser_freq, - samples_per_frame, sample_rate, bytes_per_sample, - n_channels) = struct.unpack(BLOB_HDR_FMT, raw) - if magic != BLOB_MAGIC: - raise ValueError(f"{path.name}: not a valid SRAS file (bad magic)") - if version != BLOB_VERSION: - raise ValueError( - f"{path.name}: unsupported SRAS format version {version} " - f"(this app can only resume version {BLOB_VERSION} files)" - ) - - angles = list(struct.unpack(f">{n_angles}f", f.read(4 * n_angles))) - - per_angle = [] - for a in angles: - x_start, x_delta, n_frames, n_rows = struct.unpack(">ffIH", f.read(14)) - per_angle.append({ - "angle": a, "x_start": x_start, "x_delta": x_delta, - "n_frames": n_frames, "n_rows": n_rows, - }) - - for pa in per_angle: - n = pa["n_rows"] - pa["y_positions"] = list(struct.unpack(f">{n}f", f.read(4 * n))) - - for _ in range(n_channels): - (plen,) = struct.unpack(">H", f.read(2)) - f.read(plen) - - (n_bg,) = struct.unpack(">I", f.read(4)) - f.read(n_bg) - - data_start_offset = f.tell() - - return { - "version": version, "n_angles": n_angles, - "x_start_nominal": x_start_nominal, "y_start_nominal": y_start_nominal, - "x_delta_nominal": x_delta_nominal, "y_delta_nominal": y_delta_nominal, - "row_spacing": row_spacing, "velocity": velocity, "laser_freq": laser_freq, - "samples_per_frame": samples_per_frame, "sample_rate": sample_rate, - "bytes_per_sample": bytes_per_sample, "n_channels": n_channels, - "angles": angles, "per_angle": per_angle, - "data_start_offset": data_start_offset, - } - - -def _compute_angle_status(path: Path, info: dict) -> list[dict]: - """Figure out, per angle, how much waveform data is actually present on - disk versus declared in the per-angle geometry table. - - Waveform data is written angle-major, row-minor with a fixed number of - bytes per row (derivable from the per-angle geometry table), so this - walks the expected per-row byte counts against the actual file size. - Because the data is one contiguous stream, once an angle is found to be - short every angle after it is necessarily entirely absent too — there is - always a single "frontier" past which nothing has been written yet. - - Returns a list of dicts, one per angle, each with: index, angle_deg, - n_rows (declared), row_bytes, data_offset (byte offset where this - angle's data starts), n_rows_available, and status - ("OK" / "TRUNCATED" / "MISSING"). - """ - actual_size = path.stat().st_size - cursor = info["data_start_offset"] - statuses = [] - frontier_seen = False - for ai, pa in enumerate(info["per_angle"]): - row_bytes = info["n_channels"] * pa["n_frames"] * info["samples_per_frame"] * info["bytes_per_sample"] - n_rows = pa["n_rows"] - data_offset = cursor - if frontier_seen: - n_rows_available = 0 - status = "MISSING" - else: - declared_bytes = row_bytes * n_rows - if row_bytes > 0 and cursor + declared_bytes <= actual_size: - n_rows_available = n_rows - status = "OK" - cursor += declared_bytes - else: - remaining = max(0, actual_size - cursor) - n_rows_available = remaining // row_bytes if row_bytes > 0 else 0 - status = "MISSING" if n_rows_available == 0 else "TRUNCATED" - frontier_seen = True - statuses.append({ - "index": ai, "angle_deg": pa["angle"], "n_rows": n_rows, - "row_bytes": row_bytes, "data_offset": data_offset, - "n_rows_available": n_rows_available, "status": status, - }) - return statuses - - # ── BBD202 worker ───────────────────────────────────────────────────────────── class BBD202Worker(QObject): @@ -969,9 +739,8 @@ class HeliosWindow(QWidget): self.helios_connect_toggle.setText("Connecting…") self.helios_connect_toggle.setEnabled(False) port = self.helios_port_edit.text().strip() - d = _load_aui_defaults() - d["helios_port"] = port - _save_aui_defaults(d) + DEFAULTS.helios_port = port + DEFAULTS.save() self._worker.queue_connect(port) else: self._poll_timer.stop() @@ -1068,7 +837,7 @@ class ResumeAngleDialog(QDialog): completed angle that's known to be bad). """ - def __init__(self, statuses: list[dict], parent: QWidget | None = None): + def __init__(self, statuses, parent: QWidget | None = None): super().__init__(parent) self.setWindowTitle("Select Angles to Rescan") self.resize(480, 420) @@ -1082,13 +851,13 @@ class ResumeAngleDialog(QDialog): self.list_widget = QListWidget(self) for s in statuses: label = ( - f"Angle {s['index'] + 1}/{len(statuses)} — {s['angle_deg']:.2f}° — " - f"{s['n_rows_available']}/{s['n_rows']} rows — {s['status']}" + f"Angle {s.index + 1}/{len(statuses)} — {s.angle_deg:.2f}° — " + f"{s.n_rows_available}/{s.n_rows} rows — {s.status}" ) item = QListWidgetItem(label) - item.setData(Qt.ItemDataRole.UserRole, s["index"]) + item.setData(Qt.ItemDataRole.UserRole, s.index) item.setCheckState( - Qt.CheckState.Checked if s["status"] != "OK" else Qt.CheckState.Unchecked + Qt.CheckState.Unchecked if s.complete else Qt.CheckState.Checked ) self.list_widget.addItem(item) layout.addWidget(self.list_widget) @@ -1142,9 +911,7 @@ class ScanProgressWindow(QWidget): self.verticalLayout.insertWidget(insert_pos + 1, self.bias_image_widget) self.verticalLayout.setStretch(insert_pos + 1, 1) - self._row_start_time: float | None = None - self._row_durations: list[float] = [] - self._last_angle_idx: int = -1 + self._eta = EtaEstimator() def _on_pause_btn(self, checked: bool): # Pause takes effect at the next row boundary; show the intermediate @@ -1163,7 +930,7 @@ class ScanProgressWindow(QWidget): self.pause_btn.blockSignals(False) def on_row_started(self, row: int, n_rows: int, angle_idx: int, n_angles: int): - self._row_start_time = time.monotonic() + self._eta.row_started() def update_progress(self, row: int, n_rows: int, angle_idx: int, n_angles: int): n_rows = max(1, n_rows) @@ -1173,31 +940,20 @@ class ScanProgressWindow(QWidget): self.overall_scan_current_angle_indicator.setText(f"Angle {angle_idx} of {n_angles}") if row == 0: - self._row_durations = [] - self._row_start_time = None - self._last_angle_idx = -1 + self._eta.reset() self.current_scan_progbar.setMaximum(n_rows) self.current_scan_progbar.setValue(0) self.current_scan_progbar.setFormat("Row %v/%m") else: - # Reset duration history when the angle changes - if angle_idx != self._last_angle_idx and self._last_angle_idx != -1: - self._row_durations = [] - self._last_angle_idx = angle_idx - - if self._row_start_time is not None: - self._row_durations.append(time.monotonic() - self._row_start_time) - self._row_start_time = None - + self._eta.row_finished(angle_idx) self.current_scan_progbar.setMaximum(n_rows) self.current_scan_progbar.setValue(row) rows_left = n_rows - row - if self._row_durations and rows_left > 0: - recent = self._row_durations[-5:] - avg = sum(recent) / len(recent) - eta_str = _format_eta(avg * rows_left) - self.current_scan_progbar.setFormat(f"Row %v/%m — ETA: {eta_str}") + eta_secs = self._eta.eta_secs(rows_left) + if eta_secs is not None: + self.current_scan_progbar.setFormat( + f"Row %v/%m — ETA: {format_eta(eta_secs)}") elif rows_left == 0: self.current_scan_progbar.setFormat("Row %v/%m — Done") else: @@ -1230,13 +986,15 @@ class ScanWorker(QObject): paused_changed = pyqtSignal(bool) # True while paused at a row boundary def __init__(self, bbd: BBD202Worker, t3r: T3RDriver | None, - oscope: OscopeWorker, params: dict, - resume_info: dict | None = None): + oscope: OscopeWorker, plan: ScanPlan, prefix: str, + save_dir: str, resume_info: dict | None = None): super().__init__() self._bbd = bbd self._t3r = t3r self._oscope = oscope - self._params = params + self._plan = plan + self._prefix = prefix + self._save_dir = save_dir self._resume_info = resume_info self._abort = False self._prompt_event = threading.Event() @@ -1290,66 +1048,23 @@ class ScanWorker(QObject): self.failed.emit(str(e)) def _run_scan(self): - p = self._params - x_start_nominal = p["x_start_nominal"] - y_start_nominal = p["y_start_nominal"] - x_delta_nominal = p["x_delta_nominal"] - y_delta_nominal = p["y_delta_nominal"] - n_angles = max(1, p["num_angles"]) - row_spacing = p["row_spacing"] - prefix = p["prefix"] - save_dir = Path(p["save_dir"]) + plan = self._plan + per_angle = plan.per_angle + n_angles = plan.n_angles + save_dir = Path(self._save_dir) - # Each angle's bounding box (x_start, x_delta, n_frames, n_rows, - # y_positions) was pre-computed in _build_scan_params() from the - # nominal ROI rotated by *that* angle only, so every angle scans the - # minimum area needed to cover the ROI at its own rotation instead of - # the worst case across all angles. - per_angle = p["per_angle"] - angles = [pa["angle"] for pa in per_angle] - - # ── Validate scan geometry against stage travel limits ───────────────── - # X axis: 0–110 mm (bbd20x.py). The actual move starts one ramp-length - # + buffer before x_start and ends one ramp-length + buffer after - # x_start + x_delta. + # The actual X move starts one ramp-length + buffer before x_start and + # ends one ramp-length + buffer after x_start + x_delta, so the stage + # is at full velocity across the whole data window. _x_ramp_total = SCAN_RAMP_MM + SCAN_RAMP_BUFFER_MM - for pa in per_angle: - a_x_start, a_x_delta = pa["x_start"], pa["x_delta"] - x_move_start = a_x_start - _x_ramp_total - x_move_end = a_x_start + a_x_delta + _x_ramp_total - if x_move_start < 0.0: - raise ValueError( - f"Angle {pa['angle']:.1f}°: scan pre-ramp start ({x_move_start:.3f} mm) " - f"is below the X axis minimum (0 mm). Reduce XD/YD or move XS/YS " - f"so every rotation angle's bounding box stays on-stage " - f"(SCAN_RAMP_MM={SCAN_RAMP_MM:.3f} + SCAN_RAMP_BUFFER_MM={SCAN_RAMP_BUFFER_MM:.3f})." - ) - if x_move_end > 110.0: - raise ValueError( - f"Angle {pa['angle']:.1f}°: scan run-off end ({x_move_end:.3f} mm) " - f"exceeds the X axis maximum (110 mm). Reduce XD/YD or move XS/YS " - f"so every rotation angle's bounding box stays on-stage." - ) - y_min = min(pa["y_positions"]) - y_max = max(pa["y_positions"]) - if y_min < 0.0: - raise ValueError( - f"Angle {pa['angle']:.1f}°: scan Y range starts at {y_min:.3f} mm, " - f"below the Y axis minimum (0 mm)." - ) - if y_max > 75.0: - raise ValueError( - f"Angle {pa['angle']:.1f}°: scan Y range ends at {y_max:.3f} mm, " - f"exceeds the Y axis maximum (75 mm)." - ) + validate_plan(plan, SCAN_RAMP_MM, SCAN_RAMP_BUFFER_MM) - total_rows = sum(pa["n_rows"] for pa in per_angle) geometry_summary = ", ".join( - f"{pa['angle']:.1f}°: {pa['n_rows']} row(s) × {pa['n_frames']} pts/row" + f"{pa.angle_deg:.1f}°: {pa.n_rows} row(s) × {pa.n_frames} pts/row" for pa in per_angle ) self.status_msg.emit( - f"Scan geometry: {n_angles} angle(s), {total_rows} row(s) total " + f"Scan geometry: {n_angles} angle(s), {plan.total_rows} row(s) total " f"(per-angle bounding box) | save → {save_dir}\n{geometry_summary}" ) self.started.emit() @@ -1530,12 +1245,9 @@ class ScanWorker(QObject): ) else: targets_by_ai = None - fname = save_dir / f"{prefix}.sras" - scan_file = _open_scan_file( - fname, angles, per_angle, - x_start_nominal, y_start_nominal, x_delta_nominal, y_delta_nominal, - row_spacing, SCAN_VELOCITY_MM_S, LASER_FREQ_HZ, - samples_per_frame, SCOPE_SAMPLE_RATE, + fname = save_dir / f"{self._prefix}.sras" + scan_file = create_scan_file( + fname, plan, samples_per_frame, SCOPE_SAMPLE_RATE, preambles, background_waveform, ) @@ -1554,12 +1266,12 @@ class ScanWorker(QObject): if self._abort: break - angle = pa["angle"] - a_x_start = pa["x_start"] - a_x_delta = pa["x_delta"] - a_n_rows = pa["n_rows"] - a_n_frames = pa["n_frames"] - y_positions = pa["y_positions"] + angle = pa.angle_deg + a_x_start = pa.x_start + a_x_delta = pa.x_delta + a_n_rows = pa.n_rows + a_n_frames = pa.n_frames + y_positions = pa.y_positions if targets_by_ai is not None: # Interior angles may already have valid data on either @@ -1733,10 +1445,10 @@ class MainWindow(QMainWindow): # ── UI initialisation ───────────────────────────────────────────────────── def _init_ui_fields(self): - self.t3r_comport_edit.setText(DEFAULT_T3R_PORT) - self.bbd202_comport_edit.setText(DEFAULT_BBD_PORT) - self.oscope_ip_edit.setText(DEFAULT_OSCOPE_IP) - self._helios_win.helios_port_edit.setText(DEFAULT_HELIOS_PORT) + self.t3r_comport_edit.setText(DEFAULTS.t3r_port) + self.bbd202_comport_edit.setText(DEFAULTS.bbd_port) + self.oscope_ip_edit.setText(DEFAULTS.oscope_ip) + self._helios_win.helios_port_edit.setText(DEFAULTS.helios_port) self.lineEdit_10.setText(str(BBD_DEFAULT_JOG_MM)) self.x_start_edit.setText("10.000") @@ -1746,7 +1458,7 @@ class MainWindow(QMainWindow): self.num_angles_edit.setText("1") self.row_spacing_edit.setText("0.250") self.scan_prefix_edit.setText("scan") - self.scan_save_dir_edit.setText(DEFAULT_SAVE_DIR) + self.scan_save_dir_edit.setText(DEFAULTS.save_dir) # Hide the old T3R manual controls; the connect toggle becomes the panel button. for w in ( @@ -1960,13 +1672,11 @@ class MainWindow(QMainWindow): # ── Persist defaults ────────────────────────────────────────────────────── def _persist_defaults(self): - _save_aui_defaults({ - "t3r_port": self.t3r_comport_edit.text().strip(), - "bbd_port": self.bbd202_comport_edit.text().strip(), - "oscope_ip": self.oscope_ip_edit.text().strip(), - "laser_freq_hz": DEFAULT_LASER_FREQ_HZ, - "save_dir": self.scan_save_dir_edit.text().strip(), - }) + DEFAULTS.t3r_port = self.t3r_comport_edit.text().strip() + DEFAULTS.bbd_port = self.bbd202_comport_edit.text().strip() + DEFAULTS.oscope_ip = self.oscope_ip_edit.text().strip() + DEFAULTS.save_dir = self.scan_save_dir_edit.text().strip() + DEFAULTS.save() # ── Camera toggle ───────────────────────────────────────────────────────── @@ -2000,11 +1710,11 @@ class MainWindow(QMainWindow): def _on_start_scan(self): try: - params = self._build_scan_params() + plan, prefix, save_dir = self._build_scan_plan() except ValueError as e: QMessageBox.warning(self, "Invalid Scan Parameters", str(e)) return - self._launch_scan_worker(params) + self._launch_scan_worker(plan, prefix, save_dir) def _on_resume_scan_action(self): if self._scan_thread is not None and self._scan_thread.isRunning(): @@ -2014,7 +1724,7 @@ class MainWindow(QMainWindow): ) return - start_dir = self.scan_save_dir_edit.text().strip() or DEFAULT_SAVE_DIR + start_dir = self.scan_save_dir_edit.text().strip() or DEFAULTS.save_dir path_str, _ = QFileDialog.getOpenFileName( self, "Select Scan File to Resume", start_dir, "SRAS Scan Files (*.sras)" ) @@ -2023,16 +1733,17 @@ class MainWindow(QMainWindow): path = Path(path_str) try: - info = _read_sras_header(path) - statuses = _compute_angle_status(path, info) + sras = SrasFile(path) + statuses = sras.angle_status() except (ValueError, struct.error, OSError) as e: QMessageBox.warning(self, "Cannot Resume Scan", f"Could not read scan file:\n\n{e}") return - if (info["bytes_per_sample"] != 1 or info["n_channels"] != len(SCAN_CHANNELS) - or abs(info["velocity"] - SCAN_VELOCITY_MM_S) > 1e-3 - or abs(info["laser_freq"] - LASER_FREQ_HZ) > 1e-3 - or abs(info["sample_rate"] - SCOPE_SAMPLE_RATE) > 1.0): + hdr = sras.header + if (hdr.bytes_per_sample != 1 or hdr.n_channels != len(SCAN_CHANNELS) + or abs(hdr.velocity - SCAN_VELOCITY_MM_S) > 1e-3 + or abs(hdr.laser_freq - LASER_FREQ_HZ) > 1e-3 + or abs(hdr.sample_rate - SCOPE_SAMPLE_RATE) > 1.0): QMessageBox.warning( self, "Cannot Resume Scan", f"{path.name} was recorded with acquisition settings that don't " @@ -2049,11 +1760,12 @@ class MainWindow(QMainWindow): if not selected: return - frontier_idx = next((s["index"] for s in statuses if s["status"] != "OK"), len(statuses)) + # Data is one contiguous stream, so angles past the frontier (the + # first incomplete one) can't be skipped over — back-fill the range. + frontier_idx = next((s.index for s in statuses if not s.complete), len(statuses)) at_or_past_frontier = {i for i in selected if i >= frontier_idx} if at_or_past_frontier: - required = set(range(frontier_idx, max(at_or_past_frontier) + 1)) - final = selected | required + final = selected | set(range(frontier_idx, max(at_or_past_frontier) + 1)) else: final = selected @@ -2069,10 +1781,10 @@ class MainWindow(QMainWindow): targets = [ { - "ai": s["index"], "data_offset": s["data_offset"], - "n_rows": s["n_rows"], "angle_deg": s["angle_deg"], + "ai": s.index, "data_offset": s.data_offset, + "n_rows": s.n_rows, "angle_deg": s.angle_deg, } - for s in statuses if s["index"] in final + for s in statuses if s.index in final ] total_rows = sum(t["n_rows"] for t in targets) angle_list = ", ".join(f"{t['ai'] + 1}" for t in targets) @@ -2087,28 +1799,20 @@ class MainWindow(QMainWindow): if reply != QMessageBox.StandardButton.Yes: return - params = { - "x_start_nominal": info["x_start_nominal"], "y_start_nominal": info["y_start_nominal"], - "x_delta_nominal": info["x_delta_nominal"], "y_delta_nominal": info["y_delta_nominal"], - "num_angles": info["n_angles"], - "row_spacing": info["row_spacing"], - "laser_freq": info["laser_freq"], - "prefix": path.stem, - "save_dir": str(path.parent), - "per_angle": info["per_angle"], - } + plan = plan_from_header(sras) resume_info = { "path": path, "targets": targets, - "samples_per_frame": info["samples_per_frame"], + "samples_per_frame": hdr.samples_per_frame, } - self._launch_scan_worker(params, resume_info) + self._launch_scan_worker(plan, path.stem, str(path.parent), resume_info) - def _launch_scan_worker(self, params: dict, resume_info: dict | None = None): - # ── Launch scan worker ──────────────────────────────────────────────── + def _launch_scan_worker(self, plan: ScanPlan, prefix: str, save_dir: str, + resume_info: dict | None = None): self._scan_thread = QThread(self) self._scan_worker = ScanWorker( - self._bbd_worker, self._t3r_driver, self._oscope_worker, params, resume_info + self._bbd_worker, self._t3r_driver, self._oscope_worker, + plan, prefix, save_dir, resume_info ) self._scan_worker.moveToThread(self._scan_thread) self._scan_thread.started.connect(self._scan_worker.run) @@ -2124,85 +1828,38 @@ class MainWindow(QMainWindow): self.start_scan_btn.setEnabled(False) self._scan_progress.reset_pause_btn() if resume_info is None: - self._scan_progress.update_progress(0, params["per_angle"][0]["n_rows"], 0, params["num_angles"]) + self._scan_progress.update_progress(0, plan.per_angle[0].n_rows, 0, plan.n_angles) else: ai0 = resume_info["targets"][0]["ai"] self._scan_progress.update_progress( - 0, params["per_angle"][ai0]["n_rows"], ai0 + 1, params["num_angles"] + 0, plan.per_angle[ai0].n_rows, ai0 + 1, plan.n_angles ) self._scan_progress.show() self._scan_thread.start() - def _build_scan_params(self) -> dict: + def _build_scan_plan(self) -> tuple[ScanPlan, str, str]: def _f(w, label): try: return float(w.text()) except ValueError: - raise ValueError(f"'{label}' is not a valid number: {w.text()!r}") + raise ValueError(f"'{label}' is not a valid number: {w.text()!r}") from None def _i(w, label): try: return int(w.text()) except ValueError: - raise ValueError(f"'{label}' is not a valid integer: {w.text()!r}") + raise ValueError(f"'{label}' is not a valid integer: {w.text()!r}") from None - x_start = _f(self.x_start_edit, "XS") - y_start = _f(self.y_start_edit, "YS") - x_delta = _f(self.x_delta_edit, "XD") - y_delta = _f(self.y_delta_edit, "YD") - num_angles = _i(self.num_angles_edit, "NumAngles") - row_spacing = _f(self.row_spacing_edit, "RowSpacing") - prefix = self.scan_prefix_edit.text().strip() or "scan" - save_dir = self.scan_save_dir_edit.text().strip() or DEFAULT_SAVE_DIR - - if x_delta <= 0: - raise ValueError("XD must be > 0") - if row_spacing <= 0: - raise ValueError("RowSpacing must be > 0") - if num_angles < 1: - raise ValueError("NumAngles must be ≥ 1") - - # Signed so the recorded/commanded angle sequence reflects the GR - # stage's actual physical rotation direction (see GR_ROTATION_SIGN). - if num_angles > 1: - angles = [GR_ROTATION_SIGN * i * 180.0 / (num_angles - 1) for i in range(num_angles)] - else: - angles = [0.0] - - # Each angle only needs to physically scan the bounding box of the - # nominal (x_start, y_start, x_delta, y_delta) rectangle rotated by - # THAT angle -- not the worst case across all angles -- so the X - # extent (and therefore points/row) and the row count are computed - # per angle instead of once globally. - cx = x_start + x_delta / 2.0 - cy = y_start + y_delta / 2.0 - per_angle = [] - for a in angles: - r = math.radians(a) - bb_w = abs(x_delta * math.cos(r)) + abs(y_delta * math.sin(r)) - bb_h = abs(x_delta * math.sin(r)) + abs(y_delta * math.cos(r)) - a_x_start = cx - bb_w / 2.0 - a_y_start = cy - bb_h / 2.0 - a_n_rows = max(1, round(bb_h / row_spacing) + 1) if bb_h > 0 else 1 - a_n_frames = max(1, round(bb_w * LASER_FREQ_HZ / SCAN_VELOCITY_MM_S)) - per_angle.append({ - "angle": a, - "x_start": a_x_start, - "x_delta": bb_w, - "n_frames": a_n_frames, - "n_rows": a_n_rows, - "y_positions": [a_y_start + i * row_spacing for i in range(a_n_rows)], - }) - - return { - "x_start_nominal": x_start, "y_start_nominal": y_start, - "x_delta_nominal": x_delta, "y_delta_nominal": y_delta, - "num_angles": num_angles, - "row_spacing": row_spacing, - "laser_freq": DEFAULT_LASER_FREQ_HZ, - "prefix": prefix, - "save_dir": save_dir, - "per_angle": per_angle, - } + plan = build_plan( + _f(self.x_start_edit, "XS"), _f(self.y_start_edit, "YS"), + _f(self.x_delta_edit, "XD"), _f(self.y_delta_edit, "YD"), + _i(self.num_angles_edit, "NumAngles"), + _f(self.row_spacing_edit, "RowSpacing"), + laser_freq_hz=LASER_FREQ_HZ, velocity_mm_s=SCAN_VELOCITY_MM_S, + rotation_sign=GR_ROTATION_SIGN, + ) + prefix = self.scan_prefix_edit.text().strip() or "scan" + save_dir = self.scan_save_dir_edit.text().strip() or DEFAULTS.save_dir + return plan, prefix, save_dir def _on_row_done(self, row: int, n_rows: int, angle_idx: int, n_angles: int): self._scan_progress.update_progress(row, n_rows, angle_idx, n_angles) diff --git a/sras_scan_manager.py b/sras_scan_manager.py index ad3e78c..9274a01 100755 --- a/sras_scan_manager.py +++ b/sras_scan_manager.py @@ -22,12 +22,9 @@ from dataclasses import dataclass 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 +sys.path.insert(0, str(Path(__file__).resolve().parent)) + +from core.sras_format import GEOM_FMT, HDR_FMT, MAGIC, VERSION as BLOB_VERSION, SrasFile @dataclass @@ -58,94 +55,37 @@ class SrasScanFile: 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) + sras = SrasFile(self.path) + h = sras.header + self.x_start_nominal = h.x_start_nominal + self.y_start_nominal = h.y_start_nominal + self.x_delta_nominal = h.x_delta_nominal + self.y_delta_nominal = h.y_delta_nominal + self.row_spacing_mm = h.row_spacing + self.velocity_mm_s = h.velocity + self.laser_freq_hz = h.laser_freq + self.samples_per_frame = h.samples_per_frame + self.sample_rate_hz = h.sample_rate + self.bytes_per_sample = h.bytes_per_sample + self.n_channels = h.n_channels + self.preambles_raw = sras.preambles_raw + self.background_raw = sras.background + self.data_start_offset = sras.data_start_offset + self.file_size = sras.file_size - 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, + self.angles = [ + AngleEntry( + index=st.index, angle_deg=st.angle_deg, + x_start=pa.x_start, x_delta=pa.x_delta, + n_frames=pa.n_frames, n_rows_declared=pa.n_rows, + y_positions=pa.y_positions, row_bytes=st.row_bytes, + data_offset=st.data_offset, + n_rows_available=st.n_rows_available, + data_size_available=st.n_rows_available * st.row_bytes, + complete=st.complete, ) - 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 + for pa, st in zip(sras.per_angle, sras.angle_status(), strict=True) + ] def get(self, index: int) -> AngleEntry: return self.angles[index] @@ -165,7 +105,7 @@ def _write_subset(sf: SrasScanFile, indices: list, dst_path: Path) -> list: selected = [sf.get(i) for i in indices] header = struct.pack( - HDR_FMT, BLOB_MAGIC, BLOB_VERSION, len(selected), + HDR_FMT, 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, diff --git a/t3r_control_panel.py b/t3r_control_panel.py index 7b9784e..38de516 100755 --- a/t3r_control_panel.py +++ b/t3r_control_panel.py @@ -28,8 +28,8 @@ from PyQt6.QtWidgets import ( ) from hardware.t3r_driver import T3RDriver +from hardware.serial_util import scored_ports import hardware.t3r_protocol as proto -import serial.tools.list_ports # ── Utilities ───────────────────────────────────────────────────────────────── @@ -613,16 +613,8 @@ class T3RControlPanel(QDialog): def _refresh_ports(self): current = self.port_combo.currentText() self.port_combo.clear() - ports = list(serial.tools.list_ports.comports()) - - def score(p): - text = f"{p.description} {p.manufacturer or ''} {p.product or ''}".lower() - hints = ("esp32", "jtag", "espressif", "usb serial", "cp210", "ch340", "cdc") - return -sum(h in text for h in hints) - - ports.sort(key=score) - for p in ports: - self.port_combo.addItem(f"{p.device} — {p.description or p.device}", p.device) + for device, label in scored_ports(): + self.port_combo.addItem(label, device) if self.port_combo.count() == 0: self.port_combo.addItem("(no serial ports found)", None) elif current: diff --git a/tests/gen_goldens.py b/tests/gen_goldens.py deleted file mode 100644 index dd81749..0000000 --- a/tests/gen_goldens.py +++ /dev/null @@ -1,173 +0,0 @@ -"""Generate golden fixtures capturing PRE-REFACTOR behavior (Phase 0). - -Run once against the un-refactored sc3_aui_app.py; the outputs are committed -under tests/golden/. After the refactor, tests compare the new -implementations against these committed files — do NOT regenerate them -against refactored code, that would defeat the purpose. - -Usage: .venv/bin/python tests/gen_goldens.py -""" -import json -import shutil -import sys -from pathlib import Path - -sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) -import conftest # noqa: F401 (pyueye stub + offscreen) - -import sc3_aui_app as app - -GOLDEN = Path(__file__).parent / "golden" - -# ── Geometry fixtures ──────────────────────────────────────────────────────── - -GEOMETRY_CASES = [ - # label, XS, YS, XD, YD, num_angles, row_spacing - ("single_angle_square", "10.0", "10.0", "10.0", "10.0", "1", "0.25"), - ("single_angle_rect", "10.0", "10.0", "80.0", "50.0", "1", "0.25"), - ("three_angles", "5.0", "5.0", "20.0", "12.0", "3", "0.1"), - ("five_angles", "0.0", "0.0", "30.0", "30.0", "5", "0.5"), - ("seven_angles_asym", "12.5", "7.5", "42.0", "17.0", "7", "0.05"), - ("two_angles_tiny", "1.0", "1.0", "0.02", "0.02", "2", "0.01"), - ("flat_line_ydelta_zero", "10.0", "10.0", "5.0", "0.0", "1", "0.25"), - ("fine_spacing", "3.0", "4.0", "2.5", "1.5", "4", "0.025"), -] - -GEOMETRY_ERROR_CASES = [ - ("xd_zero", "10.0", "10.0", "0.0", "10.0", "1", "0.25"), - ("xd_negative", "10.0", "10.0", "-5.0", "10.0", "1", "0.25"), - ("spacing_zero", "10.0", "10.0", "10.0", "10.0", "1", "0.0"), - ("angles_zero", "10.0", "10.0", "10.0", "10.0", "0", "0.25"), - ("xs_not_number", "abc", "10.0", "10.0", "10.0", "1", "0.25"), -] - - -class _W: - """Stands in for a QLineEdit: _build_scan_params only calls .text().""" - def __init__(self, text): - self._text = text - - def text(self): - return self._text - - -def _fake_window(xs, ys, xd, yd, na, rs): - fs = type("FakeMainWindow", (), {})() - fs.x_start_edit = _W(xs) - fs.y_start_edit = _W(ys) - fs.x_delta_edit = _W(xd) - fs.y_delta_edit = _W(yd) - fs.num_angles_edit = _W(na) - fs.row_spacing_edit = _W(rs) - fs.scan_prefix_edit = _W("golden") - fs.scan_save_dir_edit = _W("/tmp/golden-scans") - return fs - - -def build_geometry_fixtures(): - out = {"constants": { - "LASER_FREQ_HZ": app.LASER_FREQ_HZ, - "SCAN_VELOCITY_MM_S": app.SCAN_VELOCITY_MM_S, - "GR_ROTATION_SIGN": app.GR_ROTATION_SIGN, - }, "cases": {}, "error_cases": {}} - - for label, *inputs in GEOMETRY_CASES: - params = app.MainWindow._build_scan_params(_fake_window(*inputs)) - out["cases"][label] = { - "inputs": dict(zip(("XS", "YS", "XD", "YD", "num_angles", "row_spacing"), inputs)), - "params": params, - } - - for label, *inputs in GEOMETRY_ERROR_CASES: - try: - app.MainWindow._build_scan_params(_fake_window(*inputs)) - raise AssertionError(f"error case {label} did not raise") - except ValueError as e: - out["error_cases"][label] = { - "inputs": dict(zip(("XS", "YS", "XD", "YD", "num_angles", "row_spacing"), inputs)), - "message": str(e), - } - - with open(GOLDEN / "geometry.json", "w") as f: - json.dump(out, f, indent=2) - print(f"geometry.json: {len(out['cases'])} cases, {len(out['error_cases'])} error cases") - - -# ── SRAS v6 file fixtures ──────────────────────────────────────────────────── - -SPF = 8 # samples per frame (synthetic, tiny) -SAMPLE_RATE = 6.25e9 -PREAMBLES = [f"WFMOUTPRE:CH{ch};SYNTHETIC;PT_FMT Y;XINCR 1.6E-10" for ch in app.SCAN_CHANNELS] -BACKGROUND = bytes(range(SPF)) - - -def _synthetic_frame(ai, ri, ci, fi): - return bytes((ai * 7 + ri * 5 + ci * 3 + fi + s) % 256 for s in range(SPF)) - - -def build_sras_fixtures(): - # Geometry via the real pre-refactor code path: 2 angles, 3 rows, 4 frames. - params = app.MainWindow._build_scan_params( - _fake_window("1.0", "1.0", "0.02", "0.02", "2", "0.01")) - per_angle = params["per_angle"] - angles = [pa["angle"] for pa in per_angle] - - complete = GOLDEN / "complete.sras" - f = app._open_scan_file( - complete, angles, per_angle, - params["x_start_nominal"], params["y_start_nominal"], - params["x_delta_nominal"], params["y_delta_nominal"], - params["row_spacing"], - app.SCAN_VELOCITY_MM_S, app.LASER_FREQ_HZ, - SPF, SAMPLE_RATE, PREAMBLES, BACKGROUND, - ) - try: - data_start = f.tell() - for ai, pa in enumerate(per_angle): - for ri in range(pa["n_rows"]): - for ci in range(len(app.SCAN_CHANNELS)): - for fi in range(pa["n_frames"]): - f.write(_synthetic_frame(ai, ri, ci, fi)) - finally: - f.close() - - info = app._read_sras_header(complete) - assert info["data_start_offset"] == data_start - row_bytes = info["n_channels"] * per_angle[0]["n_frames"] * SPF * info["bytes_per_sample"] - angle_bytes = [row_bytes * pa["n_rows"] for pa in per_angle] - - def _truncate(name, size): - dst = GOLDEN / name - shutil.copyfile(complete, dst) - with open(dst, "r+b") as g: - g.truncate(data_start + size) - return dst - - variants = { - "complete.sras": complete, - "trunc_midrow_a1.sras": _truncate( - "trunc_midrow_a1.sras", angle_bytes[0] + row_bytes + row_bytes // 2), - "trunc_rowboundary_a1.sras": _truncate( - "trunc_rowboundary_a1.sras", angle_bytes[0] + 2 * row_bytes), - "trunc_angleboundary.sras": _truncate( - "trunc_angleboundary.sras", angle_bytes[0]), - "trunc_midrow_a0.sras": _truncate( - "trunc_midrow_a0.sras", row_bytes + row_bytes // 2), - "header_only.sras": _truncate("header_only.sras", 0), - } - - expected = {"header": info, "statuses": {}} - for name, path in variants.items(): - expected["statuses"][name] = app._compute_angle_status(path, info) - - with open(GOLDEN / "sras_expected.json", "w") as f_json: - json.dump(expected, f_json, indent=2) - print(f"sras fixtures: {len(variants)} files, " - f"data block {sum(angle_bytes)} bytes, row_bytes={row_bytes}") - - -if __name__ == "__main__": - GOLDEN.mkdir(exist_ok=True) - build_geometry_fixtures() - build_sras_fixtures() - print("done") diff --git a/tests/golden_util.py b/tests/golden_util.py new file mode 100644 index 0000000..f370b81 --- /dev/null +++ b/tests/golden_util.py @@ -0,0 +1,21 @@ +"""Shared constants for the golden .sras fixtures. + +These mirror the values tests/gen_goldens.py used when the fixtures were +generated against the pre-refactor code (commit d185676); they must never +change, or the byte-identical comparisons stop meaning anything. +""" +SPF = 8 +SAMPLE_RATE = 6.25e9 +CHANNELS = [1, 3, 4] +PREAMBLES = [f"WFMOUTPRE:CH{ch};SYNTHETIC;PT_FMT Y;XINCR 1.6E-10" for ch in CHANNELS] +BACKGROUND = bytes(range(SPF)) + +# build_plan inputs for the fixture geometry: 2 angles × 3 rows × 4 frames +TINY_PLAN_ARGS = dict(x_start=1.0, y_start=1.0, x_delta=0.02, y_delta=0.02, + num_angles=2, row_spacing=0.01) +LASER_FREQ_HZ = 20000.0 +VELOCITY_MM_S = 100.0 + + +def synthetic_frame(ai, ri, ci, fi): + return bytes((ai * 7 + ri * 5 + ci * 3 + fi + s) % 256 for s in range(SPF)) diff --git a/tests/test_config.py b/tests/test_config.py new file mode 100644 index 0000000..b55959c --- /dev/null +++ b/tests/test_config.py @@ -0,0 +1,50 @@ +"""core.config: round-trip, tolerance, and the helios_port regression. + +The old dict-based writer rebuilt the JSON from only the main window's +fields, silently discarding helios_port every time a port was edited. +ScanDefaults.save() always writes every field. +""" +import json + +from core.config import ScanDefaults + + +def test_roundtrip(tmp_path): + p = tmp_path / "defaults.json" + d = ScanDefaults(t3r_port="/dev/ttyACM3", helios_port="/dev/ttyUSB9") + d.save(p) + loaded = ScanDefaults.load(p) + assert loaded == d + + +def test_missing_file_creates_defaults(tmp_path): + p = tmp_path / "defaults.json" + d = ScanDefaults.load(p) + assert d == ScanDefaults() + assert p.exists() + + +def test_corrupt_file_falls_back(tmp_path): + p = tmp_path / "defaults.json" + p.write_text("{not json") + assert ScanDefaults.load(p) == ScanDefaults() + + +def test_unknown_keys_ignored(tmp_path): + p = tmp_path / "defaults.json" + p.write_text(json.dumps({"t3r_port": "/dev/ttyACM7", "laser_freq_hz": 20000.0})) + d = ScanDefaults.load(p) + assert d.t3r_port == "/dev/ttyACM7" + assert d.helios_port == ScanDefaults().helios_port + + +def test_helios_port_survives_partial_update(tmp_path): + """Regression: editing main-window ports must not clobber helios_port.""" + p = tmp_path / "defaults.json" + ScanDefaults(helios_port="/dev/ttyUSB7").save(p) + + d = ScanDefaults.load(p) + d.t3r_port = "/dev/ttyACM1" # what _persist_defaults does + d.save(p) + + assert ScanDefaults.load(p).helios_port == "/dev/ttyUSB7" diff --git a/tests/test_golden_prerefactor.py b/tests/test_golden_prerefactor.py deleted file mode 100644 index c43152c..0000000 --- a/tests/test_golden_prerefactor.py +++ /dev/null @@ -1,81 +0,0 @@ -"""Golden-fixture consistency tests. - -Phase 0: assert the pre-refactor implementations reproduce the committed -fixtures. Phase 2 migrates these assertions onto core.sras_format / -core.scan_geometry; the fixtures themselves never change. -""" -import json -from pathlib import Path - -import pytest - -import sc3_aui_app as app -from gen_goldens import ( - BACKGROUND, PREAMBLES, SAMPLE_RATE, SPF, _fake_window, _synthetic_frame, -) - -GOLDEN = Path(__file__).parent / "golden" - - -@pytest.fixture(scope="module") -def geometry(): - with open(GOLDEN / "geometry.json") as f: - return json.load(f) - - -@pytest.fixture(scope="module") -def sras_expected(): - with open(GOLDEN / "sras_expected.json") as f: - return json.load(f) - - -def test_geometry_cases_match(geometry): - for label, case in geometry["cases"].items(): - inputs = case["inputs"] - params = app.MainWindow._build_scan_params(_fake_window(*inputs.values())) - assert params == case["params"], f"geometry mismatch for case {label}" - - -def test_geometry_error_cases_raise(geometry): - for label, case in geometry["error_cases"].items(): - with pytest.raises(ValueError): - app.MainWindow._build_scan_params(_fake_window(*case["inputs"].values())) - - -def test_writer_output_byte_identical(tmp_path, sras_expected): - params = app.MainWindow._build_scan_params( - _fake_window("1.0", "1.0", "0.02", "0.02", "2", "0.01")) - per_angle = params["per_angle"] - angles = [pa["angle"] for pa in per_angle] - - out = tmp_path / "rewrite.sras" - f = app._open_scan_file( - out, angles, per_angle, - params["x_start_nominal"], params["y_start_nominal"], - params["x_delta_nominal"], params["y_delta_nominal"], - params["row_spacing"], - app.SCAN_VELOCITY_MM_S, app.LASER_FREQ_HZ, - SPF, SAMPLE_RATE, PREAMBLES, BACKGROUND, - ) - try: - for ai, pa in enumerate(per_angle): - for ri in range(pa["n_rows"]): - for ci in range(len(app.SCAN_CHANNELS)): - for fi in range(pa["n_frames"]): - f.write(_synthetic_frame(ai, ri, ci, fi)) - finally: - f.close() - - assert out.read_bytes() == (GOLDEN / "complete.sras").read_bytes() - - -def test_header_parse_matches(sras_expected): - info = app._read_sras_header(GOLDEN / "complete.sras") - assert info == sras_expected["header"] - - -def test_angle_status_all_variants(sras_expected): - info = app._read_sras_header(GOLDEN / "complete.sras") - for name, expected in sras_expected["statuses"].items(): - statuses = app._compute_angle_status(GOLDEN / name, info) - assert statuses == expected, f"status mismatch for {name}" diff --git a/tests/test_scan_geometry.py b/tests/test_scan_geometry.py new file mode 100644 index 0000000..9663b48 --- /dev/null +++ b/tests/test_scan_geometry.py @@ -0,0 +1,151 @@ +"""core.scan_geometry vs the pre-refactor golden geometry fixtures, +plus structural invariants and travel-limit validation.""" +import json +import math +from pathlib import Path + +import pytest + +from core.scan_geometry import ( + EtaEstimator, ScanGeometryError, StageLimits, build_plan, format_eta, + validate_plan, +) + +GOLDEN = Path(__file__).parent / "golden" + + +@pytest.fixture(scope="module") +def geometry(): + with open(GOLDEN / "geometry.json") as f: + return json.load(f) + + +def _plan_from_case(case, consts): + i = case["inputs"] + return build_plan( + float(i["XS"]), float(i["YS"]), float(i["XD"]), float(i["YD"]), + int(i["num_angles"]), float(i["row_spacing"]), + laser_freq_hz=consts["LASER_FREQ_HZ"], + velocity_mm_s=consts["SCAN_VELOCITY_MM_S"], + rotation_sign=consts["GR_ROTATION_SIGN"], + ) + + +def test_all_golden_cases_match(geometry): + consts = geometry["constants"] + for label, case in geometry["cases"].items(): + plan = _plan_from_case(case, consts) + exp = case["params"] + assert plan.x_start_nominal == exp["x_start_nominal"], label + assert plan.y_start_nominal == exp["y_start_nominal"], label + assert plan.x_delta_nominal == exp["x_delta_nominal"], label + assert plan.y_delta_nominal == exp["y_delta_nominal"], label + assert plan.row_spacing == exp["row_spacing"], label + assert plan.n_angles == exp["num_angles"], label + assert len(plan.per_angle) == len(exp["per_angle"]), label + for pa, e in zip(plan.per_angle, exp["per_angle"], strict=True): + assert pa.angle_deg == e["angle"], label + assert pa.x_start == e["x_start"], label + assert pa.x_delta == e["x_delta"], label + assert pa.n_frames == e["n_frames"], label + assert pa.n_rows == e["n_rows"], label + assert pa.y_positions == e["y_positions"], label + + +def test_golden_error_cases_raise(geometry): + consts = geometry["constants"] + for case in geometry["error_cases"].values(): + with pytest.raises(ValueError): + _plan_from_case(case, consts) + + +def test_zero_degree_bbox_equals_nominal_roi(): + plan = build_plan(10.0, 5.0, 20.0, 8.0, 1, 0.5, + laser_freq_hz=20000.0, velocity_mm_s=100.0) + pa = plan.per_angle[0] + assert pa.angle_deg == 0.0 + assert math.isclose(pa.x_start, 10.0) + assert math.isclose(pa.x_delta, 20.0) + assert math.isclose(pa.y_positions[0], 5.0) + + +def test_rotated_bbox_contains_all_roi_corners(): + plan = build_plan(30.0, 20.0, 24.0, 10.0, 7, 0.1, + laser_freq_hz=20000.0, velocity_mm_s=100.0) + cy = 20.0 + 5.0 + corners = [(-12.0, -5.0), (12.0, -5.0), (-12.0, 5.0), (12.0, 5.0)] + for pa in plan.per_angle: + r = math.radians(pa.angle_deg) + half_w = pa.x_delta / 2.0 + y_lo, y_hi = min(pa.y_positions), max(pa.y_positions) + for dx, dy in corners: + # ROI corner in the rotated frame + rx = dx * math.cos(r) - dy * math.sin(r) + ry = dx * math.sin(r) + dy * math.cos(r) + assert abs(rx) <= half_w + 1e-9, pa.angle_deg + # Row grid covers within one row-spacing at the edges + assert y_lo - 0.1 - 1e-9 <= cy + ry <= y_hi + 0.1 + 1e-9, pa.angle_deg + + +def test_plus_minus_theta_symmetry(): + a = build_plan(10, 10, 20, 10, 5, 0.25, laser_freq_hz=20000.0, + velocity_mm_s=100.0, rotation_sign=1) + b = build_plan(10, 10, 20, 10, 5, 0.25, laser_freq_hz=20000.0, + velocity_mm_s=100.0, rotation_sign=-1) + for pa, pb in zip(a.per_angle, b.per_angle, strict=True): + assert pa.angle_deg == -pb.angle_deg + assert math.isclose(pa.x_delta, pb.x_delta) + assert pa.n_rows == pb.n_rows + assert pa.n_frames == pb.n_frames + + +def test_validate_plan_limit_violations(): + limits = StageLimits() + ramp, buf = 100.0**2 / (2 * 1500.0), 1.0 + + ok = build_plan(20.0, 10.0, 40.0, 30.0, 3, 0.25, + laser_freq_hz=20000.0, velocity_mm_s=100.0) + validate_plan(ok, ramp, buf, limits) # must not raise + + too_left = build_plan(2.0, 10.0, 40.0, 30.0, 1, 0.25, + laser_freq_hz=20000.0, velocity_mm_s=100.0) + with pytest.raises(ScanGeometryError, match="pre-ramp start"): + validate_plan(too_left, ramp, buf, limits) + + too_right = build_plan(80.0, 10.0, 40.0, 30.0, 1, 0.25, + laser_freq_hz=20000.0, velocity_mm_s=100.0) + with pytest.raises(ScanGeometryError, match="run-off end"): + validate_plan(too_right, ramp, buf, limits) + + too_low = build_plan(30.0, -5.0, 40.0, 30.0, 1, 0.25, + laser_freq_hz=20000.0, velocity_mm_s=100.0) + with pytest.raises(ScanGeometryError, match="Y axis minimum"): + validate_plan(too_low, ramp, buf, limits) + + too_high = build_plan(30.0, 60.0, 40.0, 30.0, 1, 0.25, + laser_freq_hz=20000.0, velocity_mm_s=100.0) + with pytest.raises(ScanGeometryError, match="Y axis maximum"): + validate_plan(too_high, ramp, buf, limits) + + +def test_format_eta(): + assert format_eta(-3) == "0s" + assert format_eta(42) == "42s" + assert format_eta(90) == "1m 30s" + assert format_eta(3720) == "1h 02m" + + +def test_eta_estimator_rolls_and_resets_on_angle_change(): + eta = EtaEstimator(window=3) + assert eta.eta_secs(5) is None + t = 100.0 + for dur in (2.0, 4.0, 6.0, 8.0): + eta.row_started(now=t) + eta.row_finished(angle_idx=1, now=t + dur) + t += dur + # window=3 keeps [4, 6, 8] → avg 6 + assert eta.eta_secs(2) == pytest.approx(12.0) + # angle change wipes history + eta.row_started(now=t) + eta.row_finished(angle_idx=2, now=t + 10.0) + assert eta.eta_secs(3) == pytest.approx(30.0) diff --git a/tests/test_sras_format.py b/tests/test_sras_format.py new file mode 100644 index 0000000..0ff2d7e --- /dev/null +++ b/tests/test_sras_format.py @@ -0,0 +1,137 @@ +"""core.sras_format vs the pre-refactor golden fixtures. + +The goldens were produced by the original sc3_aui_app implementation; the +extracted module must reproduce them byte-for-byte (writer) and +field-for-field (parser + frontier walk). +""" +import json +from dataclasses import asdict +from pathlib import Path + +import numpy as np +import pytest + +from core.scan_geometry import build_plan +from core.sras_format import SrasFile, create_scan_file +from golden_util import ( + BACKGROUND, CHANNELS, LASER_FREQ_HZ, PREAMBLES, SAMPLE_RATE, SPF, + TINY_PLAN_ARGS, VELOCITY_MM_S, synthetic_frame, +) + +GOLDEN = Path(__file__).parent / "golden" + + +@pytest.fixture(scope="module") +def expected(): + with open(GOLDEN / "sras_expected.json") as f: + return json.load(f) + + +def _tiny_plan(): + return build_plan(**TINY_PLAN_ARGS, laser_freq_hz=LASER_FREQ_HZ, + velocity_mm_s=VELOCITY_MM_S) + + +def _write_complete(path): + plan = _tiny_plan() + f = create_scan_file(path, plan, SPF, SAMPLE_RATE, PREAMBLES, BACKGROUND) + try: + for ai, pa in enumerate(plan.per_angle): + for ri in range(pa.n_rows): + for ci in range(len(CHANNELS)): + for fi in range(pa.n_frames): + f.write(synthetic_frame(ai, ri, ci, fi)) + finally: + f.close() + + +def test_writer_byte_identical_to_golden(tmp_path): + out = tmp_path / "rewrite.sras" + _write_complete(out) + assert out.read_bytes() == (GOLDEN / "complete.sras").read_bytes() + + +def test_header_matches_golden(expected): + sras = SrasFile(GOLDEN / "complete.sras") + h = expected["header"] + assert asdict(sras.header) == { + "n_angles": h["n_angles"], + "x_start_nominal": h["x_start_nominal"], "y_start_nominal": h["y_start_nominal"], + "x_delta_nominal": h["x_delta_nominal"], "y_delta_nominal": h["y_delta_nominal"], + "row_spacing": h["row_spacing"], "velocity": h["velocity"], + "laser_freq": h["laser_freq"], + "samples_per_frame": h["samples_per_frame"], "sample_rate": h["sample_rate"], + "bytes_per_sample": h["bytes_per_sample"], "n_channels": h["n_channels"], + } + assert sras.data_start_offset == h["data_start_offset"] + assert [pa.angle_deg for pa in sras.per_angle] == h["angles"] + for pa, exp in zip(sras.per_angle, h["per_angle"], strict=True): + assert pa.angle_deg == exp["angle"] + assert pa.x_start == exp["x_start"] + assert pa.x_delta == exp["x_delta"] + assert pa.n_frames == exp["n_frames"] + assert pa.n_rows == exp["n_rows"] + assert pa.y_positions == exp["y_positions"] + + +def test_frontier_all_truncation_variants(expected): + for name, exp_statuses in expected["statuses"].items(): + statuses = SrasFile(GOLDEN / name).angle_status() + assert [asdict(s) for s in statuses] == exp_statuses, f"mismatch for {name}" + + +def test_preambles_and_background_roundtrip(): + sras = SrasFile(GOLDEN / "complete.sras") + assert sras.preambles == PREAMBLES + assert sras.background == BACKGROUND + + +def test_load_angle_memmap_equals_eager(): + with SrasFile(GOLDEN / "complete.sras") as sras: + raw = (GOLDEN / "complete.sras").read_bytes() + for ai, pa in enumerate(sras.per_angle): + view = sras.load_angle(ai) + h = sras.header + assert view.shape == (pa.n_rows, h.n_channels, pa.n_frames, h.samples_per_frame) + start = sras.angle_data_offset(ai) + eager = np.frombuffer( + raw, dtype=np.int8, offset=start, count=view.size + ).reshape(view.shape) + assert np.array_equal(view, eager) + assert not view.flags.writeable + + +def test_load_row_matches_synthetic_pattern(): + with SrasFile(GOLDEN / "complete.sras") as sras: + for ai in range(2): + for ri in range(3): + for ci in range(3): + row = sras.load_row(ai, ri, ci) + expected_bytes = b"".join( + synthetic_frame(ai, ri, ci, fi) + for fi in range(sras.per_angle[ai].n_frames) + ) + assert row.tobytes() == expected_bytes + + +def test_truncated_load_angle_partial_rows(): + sras = SrasFile(GOLDEN / "trunc_rowboundary_a1.sras") + st = sras.angle_status()[1] + assert st.status == "TRUNCATED" and st.n_rows_available == 2 + view = sras.load_angle(1, n_rows=st.n_rows_available) + assert view.shape[0] == 2 + sras.close() + + +def test_bad_magic_and_version_rejected(tmp_path): + bad = tmp_path / "bad.sras" + bad.write_bytes(b"XXXX" + bytes(60)) + with pytest.raises(ValueError, match="bad magic"): + SrasFile(bad) + + data = bytearray((GOLDEN / "complete.sras").read_bytes()) + data[4] = 5 # version byte + v5 = tmp_path / "v5.sras" + v5.write_bytes(bytes(data)) + with pytest.raises(ValueError, match="version 5"): + SrasFile(v5)