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

Compare commits

base: tka/sras-viewer:refactor/parallel-and-dedup
Branches Tags
tka/sras-viewer:main tka/sras-viewer:refactor/parallel-and-dedup tka/sras-viewer:v7-cache-format tka/sras-viewer:feature/angle-alignment tka/sras-viewer:remove-timegate-saw-add-v6
..
compare: tka/sras-viewer:b0bfe6e8c6fb0bd88f6329e7b53bc2bd11aa698d
Branches Tags
tka/sras-viewer:main tka/sras-viewer:refactor/parallel-and-dedup tka/sras-viewer:v7-cache-format tka/sras-viewer:feature/angle-alignment tka/sras-viewer:remove-timegate-saw-add-v6

1 Commits
refactor/parallel-and-dedup .. b0bfe6e8c6

Author SHA1 Message Date
tka b0bfe6e8c6 Merge pull request 'Refactor/parallel and dedup' (#1) from refactor/parallel-and-dedup into main 
Reviewed-on: #1
 2026-07-31 15:20:33 -05:00
27 changed files with 5414 additions and 6092 deletions
Expand all files Collapse all files
.gitignore
-7
View File
@@ -1,9 +1,2 @@
__pycache__/ __pycache__/
.venv/ .venv/
.DS_Store
.pytest_cache/
*.png
*.sras
baseline*.txt
after*.txt
sras_viewer.egg-info/
SRAS_FORMAT.html
+761
View File
@@ -0,0 +1,761 @@
<!DOCTYPE html>
<html>
<head>
<meta charset="UTF-8">
<title>SRAS File Format Specification</title>
<style>
/* From extension vscode.github */
/*---------------------------------------------------------------------------------------------
* Copyright (c) Microsoft Corporation. All rights reserved.
* Licensed under the MIT License. See License.txt in the project root for license information.
*--------------------------------------------------------------------------------------------*/
.vscode-dark img[src$=\#gh-light-mode-only],
.vscode-light img[src$=\#gh-dark-mode-only],
.vscode-high-contrast:not(.vscode-high-contrast-light) img[src$=\#gh-light-mode-only],
.vscode-high-contrast-light img[src$=\#gh-dark-mode-only] {
display: none;
}
</style>
<link rel="stylesheet" href="https://cdn.jsdelivr.net/gh/Microsoft/vscode/extensions/markdown-language-features/media/markdown.css">
<link rel="stylesheet" href="https://cdn.jsdelivr.net/gh/Microsoft/vscode/extensions/markdown-language-features/media/highlight.css">
<style>
body {
font-family: -apple-system, BlinkMacSystemFont, 'Segoe WPC', 'Segoe UI', system-ui, 'Ubuntu', 'Droid Sans', sans-serif;
font-size: 14px;
line-height: 1.6;
}
</style>
<style>
.task-list-item {
list-style-type: none;
}
.task-list-item-checkbox {
margin-left: -20px;
vertical-align: middle;
pointer-events: none;
}
</style>
<style>
:root {
--color-note: #0969da;
--color-tip: #1a7f37;
--color-warning: #9a6700;
--color-severe: #bc4c00;
--color-caution: #d1242f;
--color-important: #8250df;
}
</style>
<style>
@media (prefers-color-scheme: dark) {
:root {
--color-note: #2f81f7;
--color-tip: #3fb950;
--color-warning: #d29922;
--color-severe: #db6d28;
--color-caution: #f85149;
--color-important: #a371f7;
}
}
</style>
<style>
.markdown-alert {
padding: 0.5rem 1rem;
margin-bottom: 16px;
color: inherit;
border-left: .25em solid #888;
}
.markdown-alert>:first-child {
margin-top: 0
}
.markdown-alert>:last-child {
margin-bottom: 0
}
.markdown-alert .markdown-alert-title {
display: flex;
font-weight: 500;
align-items: center;
line-height: 1
}
.markdown-alert .markdown-alert-title .octicon {
margin-right: 0.5rem;
display: inline-block;
overflow: visible !important;
vertical-align: text-bottom;
fill: currentColor;
}
.markdown-alert.markdown-alert-note {
border-left-color: var(--color-note);
}
.markdown-alert.markdown-alert-note .markdown-alert-title {
color: var(--color-note);
}
.markdown-alert.markdown-alert-important {
border-left-color: var(--color-important);
}
.markdown-alert.markdown-alert-important .markdown-alert-title {
color: var(--color-important);
}
.markdown-alert.markdown-alert-warning {
border-left-color: var(--color-warning);
}
.markdown-alert.markdown-alert-warning .markdown-alert-title {
color: var(--color-warning);
}
.markdown-alert.markdown-alert-tip {
border-left-color: var(--color-tip);
}
.markdown-alert.markdown-alert-tip .markdown-alert-title {
color: var(--color-tip);
}
.markdown-alert.markdown-alert-caution {
border-left-color: var(--color-caution);
}
.markdown-alert.markdown-alert-caution .markdown-alert-title {
color: var(--color-caution);
}
</style>
</head>
<body class="vscode-body vscode-light">
<h1 id="sras-file-format-specification">SRAS File Format Specification</h1>
<p><strong>Format family:</strong> <code>.sras</code><br>
<strong>Byte order:</strong> Big-endian (network byte order) throughout, unless noted.<br>
<strong>Version history:</strong> v2 (baseline), v3 (scope calibration), v4 (background waveform), v5 (precomputed images + guaranteed frame count).</p>
<hr>
<h2 id="table-of-contents">Table of Contents</h2>
<ol>
<li><a href="#overview">Overview</a></li>
<li><a href="#type-notation">Type notation</a></li>
<li><a href="#version-history">Version history</a></li>
<li><a href="#file-structure">File structure</a>
<ul>
<li><a href="#1-fixed-header-43-bytes-all-versions">Fixed header (all versions)</a></li>
<li><a href="#2-angle-table-all-versions">Angle table (all versions)</a></li>
<li><a href="#3-row-position-table-all-versions">Row position table (all versions)</a></li>
<li><a href="#4-channel-preambles-v3">Channel preambles (v3+)</a></li>
<li><a href="#5-background-waveform-v4">Background waveform (v4+)</a></li>
<li><a href="#6-waveform-data-all-versions">Waveform data (all versions)</a></li>
<li><a href="#7-prec-section-v5">PREC section (v5)</a></li>
</ul>
</li>
<li><a href="#derived-quantities">Derived quantities</a></li>
<li><a href="#adc-calibration">ADC calibration</a></li>
<li><a href="#waveform-data-layout-detail">Waveform data layout detail</a></li>
<li><a href="#size-reference">Size reference</a></li>
<li><a href="#compatibility-notes">Compatibility notes</a></li>
</ol>
<hr>
<h2 id="overview">Overview</h2>
<p>An SRAS file stores the raw RF waveforms captured during a Surface-acoustic-wave Resonance And Spectroscopy (SRAS) scan, along with the scan geometry and scope calibration metadata needed to interpret them.</p>
<p>A scan consists of one or more <strong>angles</strong> (rotation positions of the sample), each containing a 2-D raster of <strong>rows</strong> × <strong>frames</strong>. At every pixel, <code>n_channels</code> waveforms of <code>samples_per_frame</code> ADC counts are stored. Channel order is fixed:</p>
<table>
<thead>
<tr>
<th>Index</th>
<th>Hardware channel</th>
<th>Signal</th>
</tr>
</thead>
<tbody>
<tr>
<td>0</td>
<td>CH1</td>
<td>RF acoustic packet (AC-coupled)</td>
</tr>
<tr>
<td>1</td>
<td>CH3</td>
<td>Bias A — DC mean used for masking</td>
</tr>
<tr>
<td>2</td>
<td>CH4</td>
<td>Bias B — DC mean used for masking</td>
</tr>
</tbody>
</table>
<hr>
<h2 id="type-notation">Type notation</h2>
<table>
<thead>
<tr>
<th>Symbol</th>
<th>C type</th>
<th>Size</th>
<th>Notes</th>
</tr>
</thead>
<tbody>
<tr>
<td><code>u8</code></td>
<td><code>uint8_t</code></td>
<td>1 byte</td>
<td>unsigned</td>
</tr>
<tr>
<td><code>u16</code></td>
<td><code>uint16_t</code></td>
<td>2 bytes</td>
<td>big-endian</td>
</tr>
<tr>
<td><code>u32</code></td>
<td><code>uint32_t</code></td>
<td>4 bytes</td>
<td>big-endian</td>
</tr>
<tr>
<td><code>i8</code></td>
<td><code>int8_t</code></td>
<td>1 byte</td>
<td>signed, used for ADC samples when <code>bytes_per_sample == 1</code></td>
</tr>
<tr>
<td><code>i16</code></td>
<td><code>int16_t</code></td>
<td>2 bytes</td>
<td>big-endian signed, used when <code>bytes_per_sample == 2</code></td>
</tr>
<tr>
<td><code>f32</code></td>
<td><code>float</code></td>
<td>4 bytes</td>
<td>big-endian IEEE 754 single</td>
</tr>
<tr>
<td><code>f64</code></td>
<td><code>double</code></td>
<td>8 bytes</td>
<td>big-endian IEEE 754 double</td>
</tr>
<tr>
<td><code>char[N]</code></td>
<td>—</td>
<td>N bytes</td>
<td>raw bytes, no null terminator unless noted</td>
</tr>
<tr>
<td><code>utf8[N]</code></td>
<td>—</td>
<td>N bytes</td>
<td>UTF-8 string, length-prefixed (see preamble section)</td>
</tr>
</tbody>
</table>
<hr>
<h2 id="version-history">Version history</h2>
<table>
<thead>
<tr>
<th>Version</th>
<th>Added</th>
</tr>
</thead>
<tbody>
<tr>
<td>2</td>
<td>Baseline: fixed header, angle table, row table, raw waveform data. No scope calibration (fallback constants used by readers).</td>
</tr>
<tr>
<td>3</td>
<td>Per-channel Tektronix WFMOutpre preamble strings carrying YMULT / YOFF / YZERO calibration.</td>
</tr>
<tr>
<td>4</td>
<td>Background waveform section: one CH1 reference shot subtracted from each CH1 frame before FFT.</td>
</tr>
<tr>
<td>5</td>
<td><strong>(this document)</strong> Version byte incremented to 5. <code>n_frames_hdr</code> is now the <em>actual</em> acquired frame count (authoritative). PREC section appended after waveform data with precomputed FFT-peak and DC images for instant re-display.</td>
</tr>
</tbody>
</table>
<blockquote>
<p><strong>v2 note:</strong> Version 1 is not defined; version 2 is the lowest observed in the field.</p>
</blockquote>
<hr>
<h2 id="file-structure">File structure</h2>
<pre><code>┌─────────────────────────────────────────────┐
│ 1. Fixed header (43 bytes) │ all versions
├─────────────────────────────────────────────┤
│ 2. Angle table (n_angles × 4 bytes)│ all versions
├─────────────────────────────────────────────┤
│ 3. Row position table (n_rows × 4 bytes)│ all versions
├─────────────────────────────────────────────┤
│ 4. Channel preambles (variable) │ v3+
├─────────────────────────────────────────────┤
│ 5. Background waveform (variable) │ v4+
├─────────────────────────────────────────────┤
│ 6. Waveform data (variable) │ all versions
├─────────────────────────────────────────────┤
│ 7. PREC section (variable) │ v5 only
└─────────────────────────────────────────────┘
</code></pre>
<hr>
<h3 id="1-fixed-header-43-bytes-all-versions">1. Fixed header (43 bytes, all versions)</h3>
<table>
<thead>
<tr>
<th>Offset</th>
<th>Size</th>
<th>Type</th>
<th>Field</th>
<th>Description</th>
</tr>
</thead>
<tbody>
<tr>
<td>0</td>
<td>4</td>
<td><code>char[4]</code></td>
<td><code>magic</code></td>
<td><code>SRAS</code> (ASCII, no null terminator). Reject file if this does not match.</td>
</tr>
<tr>
<td>4</td>
<td>1</td>
<td><code>u8</code></td>
<td><code>version</code></td>
<td>Format version. This document describes version <strong>5</strong>.</td>
</tr>
<tr>
<td>5</td>
<td>2</td>
<td><code>u16</code></td>
<td><code>n_angles</code></td>
<td>Number of scan angles (rotation positions). ≥ 1.</td>
</tr>
<tr>
<td>7</td>
<td>2</td>
<td><code>u16</code></td>
<td><code>n_rows</code></td>
<td>Number of scan rows (Y positions). ≥ 1.</td>
</tr>
<tr>
<td>9</td>
<td>4</td>
<td><code>f32</code></td>
<td><code>x_start_mm</code></td>
<td>X position of the first frame in the first row, in millimetres.</td>
</tr>
<tr>
<td>13</td>
<td>4</td>
<td><code>f32</code></td>
<td><code>x_delta_mm</code></td>
<td>Pre-computed pixel pitch in mm (<code>velocity_mm_s / laser_freq_hz</code>). Provided for convenience; readers should prefer the derived value.</td>
</tr>
<tr>
<td>17</td>
<td>4</td>
<td><code>f32</code></td>
<td><code>velocity_mm_s</code></td>
<td>Scanner stage velocity, mm s⁻¹. Used together with <code>laser_freq_hz</code> to compute pixel pitch.</td>
</tr>
<tr>
<td>21</td>
<td>4</td>
<td><code>f32</code></td>
<td><code>laser_freq_hz</code></td>
<td>Laser repetition rate, Hz.</td>
</tr>
<tr>
<td>25</td>
<td>4</td>
<td><code>u32</code></td>
<td><code>n_frames_hdr</code></td>
<td><strong>v2–v4:</strong> the <em>configured</em> frame count written before acquisition; may exceed actual frames acquired (use file-size arithmetic to obtain the true count). <strong>v5:</strong> the <em>actual</em> acquired frame count — authoritative; readers must not re-derive it from file size.</td>
</tr>
<tr>
<td>29</td>
<td>4</td>
<td><code>u32</code></td>
<td><code>samples_per_frame</code></td>
<td>ADC samples per waveform (<code>spf</code>).</td>
</tr>
<tr>
<td>33</td>
<td>8</td>
<td><code>f64</code></td>
<td><code>sample_rate_hz</code></td>
<td>Oscilloscope sample rate, Hz (e.g. 5 × 10⁹ for 5 GS/s).</td>
</tr>
<tr>
<td>41</td>
<td>1</td>
<td><code>u8</code></td>
<td><code>bytes_per_sample</code></td>
<td>ADC word size: <code>1</code> → <code>i8</code>, <code>2</code> → <code>i16</code> (big-endian).</td>
</tr>
<tr>
<td>42</td>
<td>1</td>
<td><code>u8</code></td>
<td><code>n_channels</code></td>
<td>Number of channels per frame. Currently always <code>3</code>.</td>
</tr>
</tbody>
</table>
<hr>
<h3 id="2-angle-table-all-versions">2. Angle table (all versions)</h3>
<p>Immediately follows the fixed header.</p>
<pre><code>n_angles × f32 — scan angle in degrees
</code></pre>
<p>Each entry is a big-endian <code>f32</code> giving the sample rotation angle in degrees at which that angle index was acquired.</p>
<hr>
<h3 id="3-row-position-table-all-versions">3. Row position table (all versions)</h3>
<p>Immediately follows the angle table.</p>
<pre><code>n_rows × f32 — Y position of each row, in millimetres
</code></pre>
<hr>
<h3 id="4-channel-preambles-v3">4. Channel preambles (v3+)</h3>
<p>One entry per channel, in channel-index order (CH1 first).</p>
<pre><code>for each channel:
u16 preamble_length — byte count of the UTF-8 string that follows
utf8[N] preamble — Tektronix WFMOutpre string
</code></pre>
<p>The preamble is the oscilloscope's <code>WFMOutpre</code> response string. Readers extract the following keys (case-insensitive, space-separated value):</p>
<table>
<thead>
<tr>
<th>Key</th>
<th>Stored unit</th>
<th>Conversion to mV</th>
</tr>
</thead>
<tbody>
<tr>
<td><code>YMULT</code></td>
<td>V count⁻¹</td>
<td>multiply by 1000</td>
</tr>
<tr>
<td><code>YOFF</code></td>
<td>ADC counts</td>
<td>used directly</td>
</tr>
<tr>
<td><code>YZERO</code></td>
<td>V</td>
<td>multiply by 1000</td>
</tr>
</tbody>
</table>
<p><strong>v2 fallback:</strong> when preambles are absent, readers use:</p>
<ul>
<li><code>YMULT</code> = 1.5625 mV count⁻¹ (50 mV/div, 8 div, 8-bit ADC)</li>
<li><code>YOFF</code> = −87.04 ADC counts (scope position = −2.72 div)</li>
<li><code>YZERO</code> = 0 mV</li>
</ul>
<hr>
<h3 id="5-background-waveform-v4">5. Background waveform (v4+)</h3>
<pre><code>u32 n_bg_samples — number of i8 ADC samples that follow
i8[n_bg] background — one representative CH1 background shot
</code></pre>
<p>The background waveform has the same <code>samples_per_frame</code> length as a normal CH1 waveform. It is subtracted from each CH1 waveform before FFT processing when background subtraction is enabled. When <code>n_bg_samples == 0</code> the section is present but empty.</p>
<hr>
<h3 id="6-waveform-data-all-versions">6. Waveform data (all versions)</h3>
<p>Begins immediately after the fixed header (v2), preambles (v3), or background waveform (v4+). The waveform data is a flat, contiguous array with the following logical shape, stored in row-major (C) order:</p>
<pre><code>waveform_data[n_angles][n_rows][n_channels][n_frames][samples_per_frame]
</code></pre>
<p>Each element is a signed ADC count of size <code>bytes_per_sample</code>:</p>
<ul>
<li><code>bytes_per_sample == 1</code> → <code>i8</code></li>
<li><code>bytes_per_sample == 2</code> → <code>i16</code> big-endian</li>
</ul>
<p><strong>Total byte count:</strong></p>
<pre><code>waveform_bytes = n_angles × n_rows × n_channels × n_frames × samples_per_frame × bytes_per_sample
</code></pre>
<h4 id="index-semantics">Index semantics</h4>
<table>
<thead>
<tr>
<th>Dimension</th>
<th>Range</th>
<th>Meaning</th>
</tr>
</thead>
<tbody>
<tr>
<td><code>[a]</code></td>
<td>0 … n_angles−1</td>
<td>Scan angle (rotation position)</td>
</tr>
<tr>
<td><code>[r]</code></td>
<td>0 … n_rows−1</td>
<td>Row (Y position); row 0 is the first acquired</td>
</tr>
<tr>
<td><code>[c]</code></td>
<td>0 … n_channels−1</td>
<td>Channel (0=CH1 RF, 1=CH3 Bias A, 2=CH4 Bias B)</td>
</tr>
<tr>
<td><code>[f]</code></td>
<td>0 … n_frames−1</td>
<td>Frame (X position) within the row</td>
</tr>
<tr>
<td><code>[s]</code></td>
<td>0 … spf−1</td>
<td>Sample index within the waveform</td>
</tr>
</tbody>
</table>
<h4 id="frame-count-determination">Frame-count determination</h4>
<ul>
<li><strong>v5:</strong> use <code>n_frames_hdr</code> directly; do not use file-size arithmetic.</li>
<li><strong>v2–v4:</strong> <code>n_frames = floor((file_bytes_after_header_sections) / (bytes_per_sample × n_angles × n_rows × n_channels × samples_per_frame))</code>. Any remainder bytes are a partial trailing row and are discarded.</li>
</ul>
<hr>
<h3 id="7-prec-section-v5">7. PREC section (v5)</h3>
<p>The PREC section is appended immediately after the waveform data and is present if and only if <code>version == 5</code> and the file size exceeds <code>waveform_end_offset</code>.</p>
<pre><code>waveform_end_offset = data_offset + waveform_bytes
</code></pre>
<p>where <code>data_offset</code> is the file offset of the first waveform byte (the byte immediately after the background waveform, or after the angle/row tables for v2 files).</p>
<h4 id="prec-header-8-bytes">PREC header (8 bytes)</h4>
<table>
<thead>
<tr>
<th>Offset (relative)</th>
<th>Size</th>
<th>Type</th>
<th>Field</th>
<th>Description</th>
</tr>
</thead>
<tbody>
<tr>
<td>0</td>
<td>4</td>
<td><code>char[4]</code></td>
<td><code>prec_magic</code></td>
<td><code>PREC</code> (ASCII). Absent or wrong magic → ignore section.</td>
</tr>
<tr>
<td>4</td>
<td>1</td>
<td><code>u8</code></td>
<td><code>prec_version</code></td>
<td>PREC format version. Currently <code>1</code>.</td>
</tr>
<tr>
<td>5</td>
<td>1</td>
<td><code>u8</code></td>
<td><code>flags</code></td>
<td>Bitmask (see below).</td>
</tr>
<tr>
<td>6</td>
<td>2</td>
<td><code>u16</code></td>
<td><code>n_stored</code></td>
<td>Number of angle entries that follow. 0 ≤ <code>n_stored</code> ≤ <code>n_angles</code>.</td>
</tr>
</tbody>
</table>
<h5 id="flags-byte">Flags byte</h5>
<table>
<thead>
<tr>
<th>Bit</th>
<th>Mask</th>
<th>Meaning when set</th>
</tr>
</thead>
<tbody>
<tr>
<td>0</td>
<td><code>0x01</code></td>
<td><code>bg_sub_applied</code> — background waveform was subtracted from CH1 before the FFT when these images were computed.</td>
</tr>
<tr>
<td>1–7</td>
<td>—</td>
<td>Reserved, must be zero on write; readers must ignore.</td>
</tr>
</tbody>
</table>
<h4 id="prec-angle-entries">PREC angle entries</h4>
<p>Repeated <code>n_stored</code> times, in arbitrary angle-index order:</p>
<pre><code>for each stored angle:
u16 angle_idx — index into the angle table (0-based)
f32[n_rows×n_frames] peak_freq_mhz — CH1 FFT peak frequency, MHz, row-major
f32[n_rows×n_frames] dc4_mv — CH4 waveform mean, mV, row-major
f32[n_rows×n_frames] dc3_mv — CH3 waveform mean, mV, row-major
</code></pre>
<p>All image arrays are <code>f32</code> big-endian, stored in row-major order: element <code>[r][f]</code> is at offset <code>(r × n_frames + f) × 4</code> bytes within the array.</p>
<p><strong><code>peak_freq_mhz</code></strong> is computed without any DC-threshold masking (i.e. the FFT is run on every pixel unconditionally). Readers apply the <code>dc4_mv</code> threshold at display time:</p>
<pre><code>pixel is valid ⟺ dc4_mv[r][f] ≥ threshold_mv
display_value = peak_freq_mhz[r][f] if valid, else 0
</code></pre>
<p><strong><code>dc4_mv</code> / <code>dc3_mv</code></strong> are the mean of all ADC samples in the respective channel waveform, converted to millivolts using the channel calibration:</p>
<pre><code>dc_mv = (adc_mean − YOFF) × YMULT + YZERO
</code></pre>
<h4 id="when-readers-must-bypass-the-prec-fast-path">When readers must bypass the PREC fast path</h4>
<p>Readers must fall back to real-time FFT computation (ignoring stored <code>peak_freq_mhz</code>) when any of the following are true:</p>
<ul>
<li>Time-domain gating is active (zeroing samples outside a time window changes peak frequency).</li>
<li>Zero-padding (<code>n_fft ≠ samples_per_frame</code>) is requested (changes bin spacing).</li>
<li>The reader's background-subtraction setting does not match <code>flags.bg_sub_applied</code>.</li>
</ul>
<hr>
<h2 id="derived-quantities">Derived quantities</h2>
<pre><code>pixel_pitch_mm = velocity_mm_s / laser_freq_hz
x_axis_mm[f] = x_start_mm + f × pixel_pitch_mm (f = 0 … n_frames−1)
time_axis_ns[s] = s / sample_rate_hz × 1e9 (s = 0 … spf−1)
freq_axis_mhz[k] = k × sample_rate_hz / (n_fft × 1e6) (k = 0 … n_fft/2)
where n_fft = samples_per_frame unless zero-padding is active
velocity_ms[r][f] = peak_freq_mhz[r][f] × grating_um (grating_um user-supplied)
</code></pre>
<hr>
<h2 id="adc-calibration">ADC calibration</h2>
<p>Convert raw ADC counts to millivolts:</p>
<pre><code>voltage_mv = (adc_count − YOFF) × YMULT_mv + YZERO_mv
</code></pre>
<p>Invert (mV → ADC count):</p>
<pre><code>adc_count = (voltage_mv − YZERO_mv) / YMULT_mv + YOFF
</code></pre>
<p>where <code>YMULT_mv</code> is YMULT in mV count⁻¹ (= scope YMULT in V count⁻¹ × 1000).</p>
<hr>
<h2 id="waveform-data-layout-detail">Waveform data layout detail</h2>
<p>For a scan with <code>n_angles=2</code>, <code>n_rows=3</code>, <code>n_channels=3</code>, <code>n_frames=4</code>, <code>spf=5</code> the layout is:</p>
<pre><code>angle 0
row 0
CH1: [s0 s1 s2 s3 s4] [s0 s1 s2 s3 s4] [s0 s1 s2 s3 s4] [s0 s1 s2 s3 s4]
frame 0 frame 1 frame 2 frame 3
CH3: …(same layout)…
CH4: …(same layout)…
row 1
…
row 2
…
angle 1
…
</code></pre>
<p>The flat byte offset of sample <code>s</code> of frame <code>f</code>, channel <code>c</code>, row <code>r</code>, angle <code>a</code> is:</p>
<pre><code>offset = data_offset
+ (a × n_rows × n_channels × n_frames × spf
+ r × n_channels × n_frames × spf
+ c × n_frames × spf
+ f × spf
+ s)
× bytes_per_sample
</code></pre>
<hr>
<h2 id="size-reference">Size reference</h2>
<p>Approximate sizes for representative scans (<code>bytes_per_sample = 1</code>, <code>n_channels = 3</code>).</p>
<table>
<thead>
<tr>
<th>n_angles</th>
<th>n_rows</th>
<th>n_frames</th>
<th>spf</th>
<th>Waveform data</th>
<th>PREC section</th>
</tr>
</thead>
<tbody>
<tr>
<td>1</td>
<td>500</td>
<td>500</td>
<td>400</td>
<td>300 MB</td>
<td>12 MB</td>
</tr>
<tr>
<td>4</td>
<td>500</td>
<td>500</td>
<td>400</td>
<td>1.2 GB</td>
<td>48 MB</td>
</tr>
<tr>
<td>1</td>
<td>2000</td>
<td>2000</td>
<td>400</td>
<td>4.8 GB</td>
<td>48 MB</td>
</tr>
<tr>
<td>4</td>
<td>2000</td>
<td>2000</td>
<td>400</td>
<td>19.2 GB</td>
<td>192 MB</td>
</tr>
<tr>
<td>8</td>
<td>2000</td>
<td>2000</td>
<td>400</td>
<td>38.4 GB</td>
<td>384 MB</td>
</tr>
<tr>
<td>16</td>
<td>2000</td>
<td>2000</td>
<td>400</td>
<td>76.8 GB</td>
<td>768 MB</td>
</tr>
</tbody>
</table>
<p><strong>PREC section size formula:</strong></p>
<pre><code>prec_bytes = 8 + n_stored × (2 + 3 × n_rows × n_frames × 4)
</code></pre>
<hr>
<h2 id="compatibility-notes">Compatibility notes</h2>
<h3 id="reading-v5-files-with-a-v4-reader">Reading v5 files with a v4 reader</h3>
<p>A v4 reader that only accepts versions <code>{2, 3, 4}</code> will reject a v5 file with an &quot;unsupported version&quot; error. This is intentional: a v4 reader would derive <code>n_frames</code> from the file size, incorrectly including the PREC bytes in the sample count, producing a silently wrong reshape.</p>
<h3 id="producing-v5-files">Producing v5 files</h3>
<p>v5 files are produced by the SRAS viewer's <strong>&quot;Pre-process and Save as v5&quot;</strong> action. The procedure is:</p>
<ol>
<li>Copy the source file (any version) verbatim.</li>
<li>Set <code>version = 5</code> at byte offset 4.</li>
<li>Set <code>n_frames_hdr</code> at byte offset 25 to the actual acquired frame count.</li>
<li>Truncate the copy to <code>data_offset + waveform_bytes</code> (removes any pre-existing stale PREC tail).</li>
<li>Compute <code>peak_freq_mhz</code>, <code>dc4_mv</code>, and <code>dc3_mv</code> for every angle using chunked FFT.</li>
<li>Append the PREC section.</li>
</ol>
<h3 id="partially-written-prec-sections">Partially-written PREC sections</h3>
<p>If <code>n_stored &lt; n_angles</code> (e.g. pre-processing was interrupted), the file is still valid. Readers use stored images for the angles present in the PREC section and fall back to real-time FFT for the remainder. Readers must check <code>angle_idx</code> bounds on each entry and stop parsing on an out-of-range value.</p>
</body>
</html>
SRAS_FORMAT.md
+379
View File
@@ -0,0 +1,379 @@
# SRAS File Format Specification
**Format family:** `.sras`
**Byte order:** Big-endian (network byte order) throughout, unless noted.
**Version history:** v2 (baseline), v3 (scope calibration), v4 (background waveform), v5 (precomputed images + guaranteed frame count).
---
## Table of Contents
1. [Overview](#overview)
2. [Type notation](#type-notation)
3. [Version history](#version-history)
4. [File structure](#file-structure)
- [Fixed header (all versions)](#1-fixed-header-43-bytes-all-versions)
- [Angle table (all versions)](#2-angle-table-all-versions)
- [Row position table (all versions)](#3-row-position-table-all-versions)
- [Channel preambles (v3+)](#4-channel-preambles-v3)
- [Background waveform (v4+)](#5-background-waveform-v4)
- [Waveform data (all versions)](#6-waveform-data-all-versions)
- [PREC section (v5)](#7-prec-section-v5)
5. [Derived quantities](#derived-quantities)
6. [ADC calibration](#adc-calibration)
7. [Waveform data layout detail](#waveform-data-layout-detail)
8. [Size reference](#size-reference)
9. [Compatibility notes](#compatibility-notes)
---
## Overview
An SRAS file stores the raw RF waveforms captured during a Surface-acoustic-wave Resonance And Spectroscopy (SRAS) scan, along with the scan geometry and scope calibration metadata needed to interpret them.
A scan consists of one or more **angles** (rotation positions of the sample), each containing a 2-D raster of **rows** × **frames**. At every pixel, `n_channels` waveforms of `samples_per_frame` ADC counts are stored. Channel order is fixed:
| Index | Hardware channel | Signal |
|-------|-----------------|--------|
| 0 | CH1 | RF acoustic packet (AC-coupled) |
| 1 | CH3 | Bias A — DC mean used for masking |
| 2 | CH4 | Bias B — DC mean used for masking |
---
## Type notation
| Symbol | C type | Size | Notes |
|--------|--------|------|-------|
| `u8` | `uint8_t` | 1 byte | unsigned |
| `u16` | `uint16_t` | 2 bytes | big-endian |
| `u32` | `uint32_t` | 4 bytes | big-endian |
| `i8` | `int8_t` | 1 byte | signed, used for ADC samples when `bytes_per_sample == 1` |
| `i16` | `int16_t` | 2 bytes | big-endian signed, used when `bytes_per_sample == 2` |
| `f32` | `float` | 4 bytes | big-endian IEEE 754 single |
| `f64` | `double` | 8 bytes | big-endian IEEE 754 double |
| `char[N]` | — | N bytes | raw bytes, no null terminator unless noted |
| `utf8[N]` | — | N bytes | UTF-8 string, length-prefixed (see preamble section) |
---
## Version history
| Version | Added |
|---------|-------|
| 2 | Baseline: fixed header, angle table, row table, raw waveform data. No scope calibration (fallback constants used by readers). |
| 3 | Per-channel Tektronix WFMOutpre preamble strings carrying YMULT / YOFF / YZERO calibration. |
| 4 | Background waveform section: one CH1 reference shot subtracted from each CH1 frame before FFT. |
| 5 | **(this document)** Version byte incremented to 5. `n_frames_hdr` is now the *actual* acquired frame count (authoritative). PREC section appended after waveform data with precomputed FFT-peak and DC images for instant re-display. |
> **v2 note:** Version 1 is not defined; version 2 is the lowest observed in the field.
---
## File structure
```
┌─────────────────────────────────────────────┐
│ 1. Fixed header (43 bytes) │ all versions
├─────────────────────────────────────────────┤
│ 2. Angle table (n_angles × 4 bytes)│ all versions
├─────────────────────────────────────────────┤
│ 3. Row position table (n_rows × 4 bytes)│ all versions
├─────────────────────────────────────────────┤
│ 4. Channel preambles (variable) │ v3+
├─────────────────────────────────────────────┤
│ 5. Background waveform (variable) │ v4+
├─────────────────────────────────────────────┤
│ 6. Waveform data (variable) │ all versions
├─────────────────────────────────────────────┤
│ 7. PREC section (variable) │ v5 only
└─────────────────────────────────────────────┘
```
---
### 1. Fixed header (43 bytes, all versions)
| Offset | Size | Type | Field | Description |
|--------|------|------|-------|-------------|
| 0 | 4 | `char[4]` | `magic` | `SRAS` (ASCII, no null terminator). Reject file if this does not match. |
| 4 | 1 | `u8` | `version` | Format version. This document describes version **5**. |
| 5 | 2 | `u16` | `n_angles` | Number of scan angles (rotation positions). ≥ 1. |
| 7 | 2 | `u16` | `n_rows` | Number of scan rows (Y positions). ≥ 1. |
| 9 | 4 | `f32` | `x_start_mm` | X position of the first frame in the first row, in millimetres. |
| 13 | 4 | `f32` | `x_delta_mm` | Pre-computed pixel pitch in mm (`velocity_mm_s / laser_freq_hz`). Provided for convenience; readers should prefer the derived value. |
| 17 | 4 | `f32` | `velocity_mm_s` | Scanner stage velocity, mm s⁻¹. Used together with `laser_freq_hz` to compute pixel pitch. |
| 21 | 4 | `f32` | `laser_freq_hz` | Laser repetition rate, Hz. |
| 25 | 4 | `u32` | `n_frames_hdr` | **v2–v4:** the *configured* frame count written before acquisition; may exceed actual frames acquired (use file-size arithmetic to obtain the true count). **v5:** the *actual* acquired frame count — authoritative; readers must not re-derive it from file size. |
| 29 | 4 | `u32` | `samples_per_frame` | ADC samples per waveform (`spf`). |
| 33 | 8 | `f64` | `sample_rate_hz` | Oscilloscope sample rate, Hz (e.g. 5 × 10⁹ for 5 GS/s). |
| 41 | 1 | `u8` | `bytes_per_sample` | ADC word size: `1` → `i8`, `2` → `i16` (big-endian). |
| 42 | 1 | `u8` | `n_channels` | Number of channels per frame. Currently always `3`. |
---
### 2. Angle table (all versions)
Immediately follows the fixed header.
```
n_angles × f32 — scan angle in degrees
```
Each entry is a big-endian `f32` giving the sample rotation angle in degrees at which that angle index was acquired.
---
### 3. Row position table (all versions)
Immediately follows the angle table.
```
n_rows × f32 — Y position of each row, in millimetres
```
---
### 4. Channel preambles (v3+)
One entry per channel, in channel-index order (CH1 first).
```
for each channel:
u16 preamble_length — byte count of the UTF-8 string that follows
utf8[N] preamble — Tektronix WFMOutpre string
```
The preamble is the oscilloscope's `WFMOutpre` response string. Readers extract the following keys (case-insensitive, space-separated value):
| Key | Stored unit | Conversion to mV |
|-----|-------------|-----------------|
| `YMULT` | V count⁻¹ | multiply by 1000 |
| `YOFF` | ADC counts | used directly |
| `YZERO` | V | multiply by 1000 |
**v2 fallback:** when preambles are absent, readers use:
- `YMULT` = 1.5625 mV count⁻¹ (50 mV/div, 8 div, 8-bit ADC)
- `YOFF` = −87.04 ADC counts (scope position = −2.72 div)
- `YZERO` = 0 mV
---
### 5. Background waveform (v4+)
```
u32 n_bg_samples — number of i8 ADC samples that follow
i8[n_bg] background — one representative CH1 background shot
```
The background waveform has the same `samples_per_frame` length as a normal CH1 waveform. It is subtracted from each CH1 waveform before FFT processing when background subtraction is enabled. When `n_bg_samples == 0` the section is present but empty.
---
### 6. Waveform data (all versions)
Begins immediately after the fixed header (v2), preambles (v3), or background waveform (v4+). The waveform data is a flat, contiguous array with the following logical shape, stored in row-major (C) order:
```
waveform_data[n_angles][n_rows][n_channels][n_frames][samples_per_frame]
```
Each element is a signed ADC count of size `bytes_per_sample`:
- `bytes_per_sample == 1` → `i8`
- `bytes_per_sample == 2` → `i16` big-endian
**Total byte count:**
```
waveform_bytes = n_angles × n_rows × n_channels × n_frames × samples_per_frame × bytes_per_sample
```
#### Index semantics
| Dimension | Range | Meaning |
|-----------|-------|---------|
| `[a]` | 0 … n_angles−1 | Scan angle (rotation position) |
| `[r]` | 0 … n_rows−1 | Row (Y position); row 0 is the first acquired |
| `[c]` | 0 … n_channels−1 | Channel (0=CH1 RF, 1=CH3 Bias A, 2=CH4 Bias B) |
| `[f]` | 0 … n_frames−1 | Frame (X position) within the row |
| `[s]` | 0 … spf−1 | Sample index within the waveform |
#### Frame-count determination
- **v5:** use `n_frames_hdr` directly; do not use file-size arithmetic.
- **v2–v4:** `n_frames = floor((file_bytes_after_header_sections) / (bytes_per_sample × n_angles × n_rows × n_channels × samples_per_frame))`. Any remainder bytes are a partial trailing row and are discarded.
---
### 7. PREC section (v5)
The PREC section is appended immediately after the waveform data and is present if and only if `version == 5` and the file size exceeds `waveform_end_offset`.
```
waveform_end_offset = data_offset + waveform_bytes
```
where `data_offset` is the file offset of the first waveform byte (the byte immediately after the background waveform, or after the angle/row tables for v2 files).
#### PREC header (8 bytes)
| Offset (relative) | Size | Type | Field | Description |
|-------------------|------|------|-------|-------------|
| 0 | 4 | `char[4]` | `prec_magic` | `PREC` (ASCII). Absent or wrong magic → ignore section. |
| 4 | 1 | `u8` | `prec_version` | PREC format version. Currently `1`. |
| 5 | 1 | `u8` | `flags` | Bitmask (see below). |
| 6 | 2 | `u16` | `n_stored` | Number of angle entries that follow. 0 ≤ `n_stored` ≤ `n_angles`. |
##### Flags byte
| Bit | Mask | Meaning when set |
|-----|------|-----------------|
| 0 | `0x01` | `bg_sub_applied` — background waveform was subtracted from CH1 before the FFT when these images were computed. |
| 1–7 | — | Reserved, must be zero on write; readers must ignore. |
#### PREC angle entries
Repeated `n_stored` times, in arbitrary angle-index order:
```
for each stored angle:
u16 angle_idx — index into the angle table (0-based)
f32[n_rows×n_frames] peak_freq_mhz — CH1 FFT peak frequency, MHz, row-major
f32[n_rows×n_frames] dc4_mv — CH4 waveform mean, mV, row-major
f32[n_rows×n_frames] dc3_mv — CH3 waveform mean, mV, row-major
```
All image arrays are `f32` big-endian, stored in row-major order: element `[r][f]` is at offset `(r × n_frames + f) × 4` bytes within the array.
**`peak_freq_mhz`** is computed without any DC-threshold masking (i.e. the FFT is run on every pixel unconditionally). Readers apply the `dc4_mv` threshold at display time:
```
pixel is valid ⟺ dc4_mv[r][f] ≥ threshold_mv
display_value = peak_freq_mhz[r][f] if valid, else 0
```
**`dc4_mv` / `dc3_mv`** are the mean of all ADC samples in the respective channel waveform, converted to millivolts using the channel calibration:
```
dc_mv = (adc_mean − YOFF) × YMULT + YZERO
```
#### When readers must bypass the PREC fast path
Readers must fall back to real-time FFT computation (ignoring stored `peak_freq_mhz`) when any of the following are true:
- Time-domain gating is active (zeroing samples outside a time window changes peak frequency).
- Zero-padding (`n_fft ≠ samples_per_frame`) is requested (changes bin spacing).
- The reader's background-subtraction setting does not match `flags.bg_sub_applied`.
---
## Derived quantities
```
pixel_pitch_mm = velocity_mm_s / laser_freq_hz
x_axis_mm[f] = x_start_mm + f × pixel_pitch_mm (f = 0 … n_frames−1)
time_axis_ns[s] = s / sample_rate_hz × 1e9 (s = 0 … spf−1)
freq_axis_mhz[k] = k × sample_rate_hz / (n_fft × 1e6) (k = 0 … n_fft/2)
where n_fft = samples_per_frame unless zero-padding is active
velocity_ms[r][f] = peak_freq_mhz[r][f] × grating_um (grating_um user-supplied)
```
---
## ADC calibration
Convert raw ADC counts to millivolts:
```
voltage_mv = (adc_count − YOFF) × YMULT_mv + YZERO_mv
```
Invert (mV → ADC count):
```
adc_count = (voltage_mv − YZERO_mv) / YMULT_mv + YOFF
```
where `YMULT_mv` is YMULT in mV count⁻¹ (= scope YMULT in V count⁻¹ × 1000).
---
## Waveform data layout detail
For a scan with `n_angles=2`, `n_rows=3`, `n_channels=3`, `n_frames=4`, `spf=5` the layout is:
```
angle 0
row 0
CH1: [s0 s1 s2 s3 s4] [s0 s1 s2 s3 s4] [s0 s1 s2 s3 s4] [s0 s1 s2 s3 s4]
frame 0 frame 1 frame 2 frame 3
CH3: …(same layout)…
CH4: …(same layout)…
row 1
…
row 2
…
angle 1
…
```
The flat byte offset of sample `s` of frame `f`, channel `c`, row `r`, angle `a` is:
```
offset = data_offset
+ (a × n_rows × n_channels × n_frames × spf
+ r × n_channels × n_frames × spf
+ c × n_frames × spf
+ f × spf
+ s)
× bytes_per_sample
```
---
## Size reference
Approximate sizes for representative scans (`bytes_per_sample = 1`, `n_channels = 3`).
| n_angles | n_rows | n_frames | spf | Waveform data | PREC section |
|----------|--------|----------|-----|---------------|-------------|
| 1 | 500 | 500 | 400 | 300 MB | 12 MB |
| 4 | 500 | 500 | 400 | 1.2 GB | 48 MB |
| 1 | 2000 | 2000 | 400 | 4.8 GB | 48 MB |
| 4 | 2000 | 2000 | 400 | 19.2 GB | 192 MB |
| 8 | 2000 | 2000 | 400 | 38.4 GB | 384 MB |
| 16 | 2000 | 2000 | 400 | 76.8 GB | 768 MB |
**PREC section size formula:**
```
prec_bytes = 8 + n_stored × (2 + 3 × n_rows × n_frames × 4)
```
---
## Compatibility notes
### Reading v5 files with a v4 reader
A v4 reader that only accepts versions `{2, 3, 4}` will reject a v5 file with an "unsupported version" error. This is intentional: a v4 reader would derive `n_frames` from the file size, incorrectly including the PREC bytes in the sample count, producing a silently wrong reshape.
### Producing v5 files
v5 files are produced by the SRAS viewer's **"Pre-process and Save as v5"** action. The procedure is:
1. Copy the source file (any version) verbatim.
2. Set `version = 5` at byte offset 4.
3. Set `n_frames_hdr` at byte offset 25 to the actual acquired frame count.
4. Truncate the copy to `data_offset + waveform_bytes` (removes any pre-existing stale PREC tail).
5. Compute `peak_freq_mhz`, `dc4_mv`, and `dc3_mv` for every angle using chunked FFT.
6. Append the PREC section.
### Partially-written PREC sections
If `n_stored < n_angles` (e.g. pre-processing was interrupted), the file is still valid. Readers use stored images for the angles present in the PREC section and fall back to real-time FFT for the remainder. Readers must check `angle_idx` bounds on each entry and stop parsing on an out-of-range value.
docs/design.md
-137
View File
@@ -1,137 +0,0 @@
# sras-viewer design notes
Rationale that outgrew code comments. Each section is referenced by a short
pointer comment at the relevant definition, so the code stays scannable and
the reasoning stays findable.
## Memory budget and row chunking (`sras_compute.py`)
DC images are computed over row chunks so the float32 working buffers for one
chunk stay under a memory budget. A fixed row count (the original design)
works fine for small legacy scans but is catastrophic for a v6 scan with a
large per-angle frame/sample count — e.g. a 7500-frame × 2500-sample angle
needs ~2.4 GB for a single 32-row chunk.
With chunks running concurrently the budget has to cover *all* live chunks at
once. On a large scan `chunk_rows` is already clamped to its floor of one row
(one row alone is ~75 MB of float32 at 7507×2500), so shrinking the per-chunk
size cannot buy more concurrency — the worker count must be derived from the
budget instead: `_plan_chunks` picks the worker count *first* and sizes the
chunk to it. Sizing the chunk first is the trap: a single chunk would always
consume the whole budget and leave room for exactly one worker, precisely on
the large scans that need concurrency most.
The 1024 MB default (`SRAS_MEM_BUDGET_MB`) is the measured knee on a 16-core
machine against a 7507-frame × 2500-sample angle: 512 MB left ~20% of the
speedup on the table, and 1536+ MB cost ~0.4 GB more resident memory for no
further gain.
A caller that itself runs several computations concurrently (angle-level
parallelism, `plan_angle_level`) must pass *both* `max_workers=1` and its
share of the budget. Capping the workers alone is not enough: the chunk would
still be sized against the whole budget, and N concurrent callers would each
allocate all of it.
## FFT peak search: block-parallel zoom refinement (`sras_compute.py`)
The displayed RF value per pixel is the argmax of the zero-padded power
spectrum of that pixel's CH1 waveform. At the pad factor of 40 needed for
mapping resolution, materialising padded spectra is hopeless: ~9 GB per scan
row, which is what used to collapse the old row-chunk planner to one worker
and make synthesis single-threaded.
`_peak_bins_zoom` never materialises the padded spectrum:
1. a coarse rfft at `next_fast_len(2*spf)` — 2× oversampled, so the padded
power spectrum (a trig polynomial of degree spf−1) cannot hide its global
max between coarse samples;
2. every coarse bin within `_ZOOM_CAND_RATIO` (0.7) of its row's coarse max
becomes a refinement candidate. Quarter-natural-bin scalloping at the 2×
grid can understate a peak's power by at most ~19%, so 0.7 keeps a wide
margin. The DC-adjacent window is always refined too: the coarse DC bin
is zeroed for suppression, which would otherwise blind the scan to fine
bins closer to DC than the first coarse sample (where the leakage skirt
of an un-subtracted offset peaks);
3. each candidate window (±`_ZOOM_HALFWIDTH` = 0.75 coarse spacings; every
fine bin lies within 0.5 spacings of its nearest coarse bin) is evaluated
on the exact `n_fft` grid by one small complex gemm, with np.argmax's
lowest-bin tie-break preserved across windows.
The selected bin is bit-identical to the full padded argmax — enforced by
`tests/test_compute.py::test_zoom_identity`, a fuzz test over adversarial
spectra, and the golden-hash harness (`tools/check_equivalence.py`), whose
baseline was captured on the old full-padded path.
Work fans out over a persistent thread pool in `_FFT_BLOCK` = 512-waveform
tasks: smaller blocks serialise on GIL-held numpy dispatch, larger ones lose
cache residency and task granularity (measured on a 16-core machine, where
this path runs ~35× faster than the old serial padded transform at pad 40).
pyFFTW runs through per-thread `builders` plans (FFTW_MEASURE, wisdom
persisted under `~/.cache/sras-viewer/`), and `threadpoolctl` clamps BLAS to
one thread under the pool so the refinement gemm cannot oversubscribe.
`compute_rf_image(exact=True)` (or `SRAS_FFT_EXACT=1`) keeps the reference
full-padded path for audits.
## Angle alignment coordinate frames (`sras_compute.py`)
Alignment puts every angle's images onto one shared, zero-padded pixel grid
using a rigid transform only — rotation + translation, never scale.
Angle 0 (the reference) is the sole coordinate authority: it is the only
angle whose stage XY (`x_start_mm` / `y_positions_mm`) is ever read, and the
shared canvas is literally an extension of angle 0's own pixel grid, so the
aligned view carries angle 0's real X/Y axes. Every *other* angle is placed
purely by content — its rotation and translation come from cross-correlating
its CH4 image against angle 0's (`register_angle_to_reference`) — and its own
stage XY is deliberately never consulted. That is not an oversight: the
rotation stage moves the sample relative to the scan window, so where a
window sat in stage coordinates says nothing about where the sample is, and
an earlier design that pivoted each angle on a signal-weighted centroid of
its own window put every angle on a ~20 mm circle around the optical center
instead of stacking them into one shape.
Only two coordinate frames exist:
* **local mm** — one angle's own physical frame: origin at the *center of its
own pixel array*, x along +column, y along +row, scaled by that angle's own
pitches. Carries no stage position whatsoever.
* **ref mm** — the reference angle's local mm. A registration result
`(rotation_deg, shift_mm)` is exactly the rigid map from an angle's local
mm to ref mm: `q = R(rotation_deg) @ l + shift_mm`. Stage coordinates
re-enter once, at the very end, when the canvas origin is converted to
angle 0's stage mm (`AlignmentResult.canvas_origin_mm`).
Rotation is done in mm, never on raw pixel indices: the x pitch
(`SrasFile.pixel_x_mm`, 5 µm on a real scan) and the y/row pitch (50 µm)
differ by 10×, so rotating the raw index grid would shear the image — an
unwanted anisotropic scale. Registration runs on a resampled *isotropic* grid
for the same reason, and every affine maps shared-grid index → mm → undo
rotation/shift → that angle's own local mm → that angle's own raw index,
matching the output→input convention `scipy.ndimage.affine_transform` wants.
## Manual-alignment sidecar (`sras_compute.py`)
`<name>.sras.align.json` lives next to the scan file. The code lives in
`sras_compute`, not `sras_format`: `sras_format` is scoped to the versioned
binary .sras spec itself (see `scan_format.md`), while a manual alignment is
a viewer-computed *derived* artifact, analogous in kind to `AlignmentResult`
— so it belongs with the alignment math it serialises. json + pathlib are
stdlib, so this adds no dependency to a module whose load-bearing constraint
is staying free of Qt/matplotlib for cheap multiprocessing-child imports.
### Schema history
The stored `rotation_deg`/`shift_mm` are meaningless without the frame they
were measured in, so `_SIDECAR_SCHEMA_VERSION` is bumped whenever that frame
changes. Each bump makes older files describe a different (and, for the bugs
each bump fixed, actively wrong) transform than the same numbers would today;
loading one unchanged would silently reproduce the very "scans show up
everywhere" symptom the bump fixed — so older sidecars are treated as absent
rather than migrated.
* **1 → 2** — pivot moved from the scan-window bbox center to a
content-derived centroid, and the rotation sign convention was corrected.
* **2 → 3** — the content centroid was abandoned entirely: rotation is now
about each angle's own array center, mapped onto the reference's array
center, with `shift_mm` in the reference's local mm frame. No angle but the
reference contributes stage coordinates any more.
pyproject.toml
-40
View File
@@ -1,40 +0,0 @@
[build-system]
requires = ["setuptools>=68"]
build-backend = "setuptools.build_meta"
[project]
name = "sras-viewer"
version = "0.1.0"
description = "Viewer and processing tools for SRAS .sras scan files"
requires-python = ">=3.12"
dependencies = [
"PyQt6==6.10.2",
"numpy==2.4.1",
"matplotlib==3.10.8",
"scipy==1.18.0",
# Angle alignment only: masked FFT phase correlation (skimage.registration).
"scikit-image==0.26.0",
# Faster rfft backend; the viewer falls back to scipy.fft without it.
"pyFFTW==0.15.1",
# Clamps BLAS threading under the FFT worker pool.
"threadpoolctl==3.6.0",
]
[project.optional-dependencies]
dev = ["pytest"]
[project.scripts]
sras-viewer = "sras_viewer.main_window:main"
[tool.setuptools]
py-modules = [
"sras_format",
"sras_compute",
"sras_workers",
"sras_average",
"sras_edit_scans",
]
packages = ["sras_viewer"]
[tool.pytest.ini_options]
testpaths = ["tests"]
sras_average.py
+1 -1
View File
@@ -47,7 +47,7 @@ def read_header_sections(sras: SrasFile) -> bytes:
lost in a re-encode.""" lost in a re-encode."""
with open(sras.path, "rb") as f: with open(sras.path, "rb") as f:
f.seek(HDR_SIZE) f.seek(HDR_SIZE)
return f.read(sras.data_offset - HDR_SIZE) return f.read(sras._data_offset - HDR_SIZE)
def average_rows(block: np.ndarray, n: int, discard_remainder: bool) -> np.ndarray: def average_rows(block: np.ndarray, n: int, discard_remainder: bool) -> np.ndarray:
sras_compute.py
+535 -1131
View File
File diff suppressed because it is too large Load Diff
sras_edit_scans.py
-219
View File
@@ -1,219 +0,0 @@
#!/usr/bin/env python3
"""
sras_edit_scans.py — Remove one or more angle scans from a .sras file.
A .sras file holds one or more "angles" (rotation positions); the viewer
cross-correlates each non-reference angle against the reference to align
them. If one angle's acquisition went wrong (stage glitch, bad trigger,
laser dropout, ...) it throws off that alignment for the whole file. This
tool drops the bad angle(s) and renumbers the rest, writing a new .sras file
with everything else — waveform samples, calibration preambles, background
waveform, row/geometry tables — carried over byte-for-byte.
Handles v2-v7. Any precomputed FFT/DC cache (v5 PREC tail, v7 CACH tail) is
dropped on write, since it's indexed by angle and would be stale/misaligned
after renumbering; the viewer just recomputes it next time the file opens.
Usage:
python sras_edit_scans.py input.sras --list
python sras_edit_scans.py input.sras output.sras --drop 2,5
python sras_edit_scans.py input.sras output.sras --keep 0,1,3,4,6
"""
import argparse
import struct
import sys
from pathlib import Path
from sras_format import GEO_FMT_V6, HDR_FMT, HDR_FMT_V6, HDR_SIZE, SrasFile
_LEGACY_VERSIONS = (2, 3, 4, 5)
_V6_VERSIONS = (6, 7)
def _die(msg: str):
print(f"Error: {msg}", file=sys.stderr)
sys.exit(1)
def parse_args():
p = argparse.ArgumentParser(
description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
p.add_argument("input", help="Input .sras file")
p.add_argument("output", nargs="?", help="Output .sras file (omit with --list)")
p.add_argument("--list", action="store_true",
help="Print each angle's index/degrees/geometry and exit")
g = p.add_mutually_exclusive_group()
g.add_argument("--drop", metavar="I,J,...",
help="Comma-separated angle indices to remove")
g.add_argument("--keep", metavar="I,J,...",
help="Comma-separated angle indices to keep (all others dropped)")
return p.parse_args()
def _parse_index_list(s: str, n_angles: int) -> set[int]:
out = set()
for piece in s.split(","):
piece = piece.strip()
if not piece:
continue
i = int(piece)
if not (0 <= i < n_angles):
raise ValueError(f"angle index {i} out of range [0, {n_angles - 1}]")
out.add(i)
return out
def print_listing(sras: SrasFile):
print(f"\n{'idx':>4} {'angle_deg':>10} {'x_start_mm':>11} {'rows':>6} {'frames':>7}")
for a in range(sras.n_angles):
print(f"{a:>4} {sras.angles_deg[a]:>10.4f} {sras.x_start_mm[a]:>11.4f} "
f"{int(sras.n_rows[a]):>6} {int(sras.n_frames[a]):>7}")
def _copy_range(fin, fout, offset: int, nbytes: int, chunk: int = 64 * 1024 * 1024):
"""Stream *nbytes* raw bytes from *fin* at *offset* into *fout*, without
ever holding more than one chunk in memory (waveform blocks can be
hundreds of MB to low GB each)."""
fin.seek(offset)
remaining = nbytes
while remaining:
buf = fin.read(min(chunk, remaining))
if not buf:
raise IOError("unexpected EOF while copying waveform data")
fout.write(buf)
remaining -= len(buf)
# ---------------------------------------------------------------------------
# Legacy (v2-v5): uniform geometry across angles, one flat waveform block
# ---------------------------------------------------------------------------
def _write_legacy(sras: SrasFile, keep: list[int], out_path: Path):
n_rows = int(sras.n_rows[0])
n_frames = int(sras.n_frames[0]) # uniform across angles for v2-v5
n_ch = sras.n_channels
spf = sras.samples_per_frame
bps = sras.bytes_per_sample
header = struct.pack(
HDR_FMT, b"SRAS", sras.version, len(keep), n_rows,
float(sras.x_start_mm[0]), float(sras.x_delta_mm),
sras.velocity_mm_s, sras.laser_freq_hz,
n_frames, spf, sras.sample_rate_hz, bps, n_ch,
)
# Row table + preambles + background sit right after the angle table and
# don't vary per angle — copy that whole span through unmodified.
angle_table_size = sras.n_angles * 4
with open(sras.path, "rb") as f:
f.seek(HDR_SIZE + angle_table_size)
shared_mid = f.read(sras.data_offset - (HDR_SIZE + angle_table_size))
angle_bytes = n_rows * n_ch * n_frames * spf * bps
with open(sras.path, "rb") as fin, open(out_path, "wb") as fout:
fout.write(header)
fout.write(sras.angles_deg[keep].astype(">f4").tobytes())
fout.write(shared_mid)
for a in keep:
_copy_range(fin, fout, sras.data_offset + a * angle_bytes, angle_bytes)
# ---------------------------------------------------------------------------
# v6/v7: per-angle geometry, ragged waveform blocks
# ---------------------------------------------------------------------------
def _write_v6(sras: SrasFile, keep: list[int], out_path: Path):
header = struct.pack(
HDR_FMT_V6, b"SRAS", sras.version, len(keep),
sras.x_start_nominal_mm, sras.y_start_nominal_mm,
sras.x_delta_nominal_mm, sras.y_delta_nominal_mm,
sras.row_spacing_mm, sras.velocity_mm_s, sras.laser_freq_hz,
sras.samples_per_frame, sras.sample_rate_hz,
sras.bytes_per_sample, sras.n_channels,
)
blocks = {a: (offset, nbytes) for a, offset, nbytes in sras.iter_angle_blocks()}
with open(sras.path, "rb") as fin, open(out_path, "wb") as fout:
fout.write(header)
fout.write(sras.angles_deg[keep].astype(">f4").tobytes())
for i in keep:
fout.write(struct.pack(
GEO_FMT_V6, float(sras.x_start_mm[i]),
float(sras.x_delta_mm_per_angle[i]),
int(sras.n_frames[i]), int(sras.n_rows[i])))
for i in keep:
fout.write(sras.y_pos_per_angle[i].astype(">f4").tobytes())
fout.write(sras.preambles_raw)
fout.write(sras.background_raw)
for i in keep:
offset, nbytes = blocks[i]
_copy_range(fin, fout, offset, nbytes)
def main():
args = parse_args()
in_path = Path(args.input)
if not in_path.exists():
_die(f"input file not found: {in_path}")
print(f"Reading {in_path} ...", flush=True)
try:
sras = SrasFile(str(in_path))
except ValueError as e:
_die(str(e))
if sras.version not in (*_LEGACY_VERSIONS, *_V6_VERSIONS):
_die(f"unsupported .sras version: {sras.version}")
aborted_note = " (scan aborted; trailing angle(s) already excluded)" if sras.scan_aborted else ""
print(f" Version : v{sras.version}", flush=True)
print(f" Angles : {sras.n_angles}{aborted_note}", flush=True)
if args.list:
print_listing(sras)
return
if not args.output:
_die("output path required unless --list is given.")
if not (args.drop or args.keep):
_die("specify --drop or --keep (see --list for indices).")
out_path = Path(args.output)
if out_path.resolve() == in_path.resolve():
_die("output path must differ from input path.")
try:
if args.drop:
drop = _parse_index_list(args.drop, sras.n_angles)
keep = [a for a in range(sras.n_angles) if a not in drop]
else:
keep = sorted(_parse_index_list(args.keep, sras.n_angles))
except ValueError as e:
_die(str(e))
if not keep:
_die("at least one angle must remain.")
dropped = [a for a in range(sras.n_angles) if a not in keep]
print(f"\nDropping angle(s): {dropped}")
print(f"Keeping angle(s) : {keep} ({len(keep)} of {sras.n_angles})")
print(f"\nWriting {out_path} ...", flush=True)
if sras.version in _LEGACY_VERSIONS:
_write_legacy(sras, keep, out_path)
else:
_write_v6(sras, keep, out_path)
in_mb = in_path.stat().st_size / 1024**2
out_mb = out_path.stat().st_size / 1024**2
print(f" Input size : {in_mb:.1f} MB")
print(f" Output size: {out_mb:.1f} MB")
print("Done.")
print("Note: any precomputed FFT/DC cache was dropped (it's indexed by "
"angle); the viewer will recompute it next time this file opens.")
if __name__ == "__main__":
main()
sras_format.py
+21 -93
View File
@@ -60,12 +60,7 @@ PREC_FLAG_BG_SUB = 0x01
CACH_MAGIC = b"CACH" CACH_MAGIC = b"CACH"
CACH_HDR_FMT = ">4sBB" # magic, cach_version, block_flags CACH_HDR_FMT = ">4sBB" # magic, cach_version, block_flags
CACH_HDR_SIZE = struct.calcsize(CACH_HDR_FMT) CACH_HDR_SIZE = struct.calcsize(CACH_HDR_FMT)
# v2 added the SFFT pad factor. Writers always emit v2; readers still accept CACH_VERSION = 1
# v1, whose FFT block is natural-resolution by construction. An older reader
# meeting a v2 tail rejects the whole section on the version check and
# recomputes — no cache, but no misreading either.
CACH_VERSION = 2
CACH_VERSION_MIN = 1
CACH_FLAG_DC = 0x01 CACH_FLAG_DC = 0x01
CACH_FLAG_FFT = 0x02 CACH_FLAG_FFT = 0x02
@@ -74,12 +69,9 @@ SDCB_HDR_FMT = ">4sBH" # magic, reserved, n_stored
SDCB_HDR_SIZE = struct.calcsize(SDCB_HDR_FMT) SDCB_HDR_SIZE = struct.calcsize(SDCB_HDR_FMT)
SFFT_MAGIC = b"SFFT" SFFT_MAGIC = b"SFFT"
SFFT_HDR_FMT_V1 = ">4sBH" # magic, flags, n_stored SFFT_HDR_FMT = ">4sBH" # magic, flags, n_stored
SFFT_HDR_FMT = ">4sBHH" # magic, flags, n_stored, pad_factor
SFFT_HDR_SIZE = struct.calcsize(SFFT_HDR_FMT) SFFT_HDR_SIZE = struct.calcsize(SFFT_HDR_FMT)
SFFT_FLAG_BG_SUB = 0x01 SFFT_FLAG_BG_SUB = 0x01
# The viewer clamps its pad factor to this, and the field is a uint16.
PAD_FACTOR_MAX = 256
# Fixed channel indices into the .sras data array (CH1=RF, CH3/CH4=Bias DC) # Fixed channel indices into the .sras data array (CH1=RF, CH3/CH4=Bias DC)
CH1_IDX, CH3_IDX, CH4_IDX = 0, 1, 2 CH1_IDX, CH3_IDX, CH4_IDX = 0, 1, 2
@@ -177,9 +169,7 @@ class SrasFile:
``precomputed_dc3_mv`` / ``precomputed_dc4_mv`` / ``precomputed_freq_mhz``, ``precomputed_dc3_mv`` / ``precomputed_dc4_mv`` / ``precomputed_freq_mhz``,
always as ragged per-angle lists (``list[np.ndarray | None]``, one entry always as ragged per-angle lists (``list[np.ndarray | None]``, one entry
per angle, ``None`` where that angle was never stored) regardless of per angle, ``None`` where that angle was never stored) regardless of
source version. The settings the FFT images were computed under travel source version.
with them as ``precomputed_bg_sub`` / ``precomputed_pad_factor``, since a
stored image is only usable by a view asking for the same two.
""" """
def __init__(self, path: str): def __init__(self, path: str):
@@ -236,9 +226,6 @@ class SrasFile:
self.precomputed_dc4_mv: list[np.ndarray | None] = [None] * n_angles self.precomputed_dc4_mv: list[np.ndarray | None] = [None] * n_angles
self.precomputed_dc3_mv: list[np.ndarray | None] = [None] * n_angles self.precomputed_dc3_mv: list[np.ndarray | None] = [None] * n_angles
self.precomputed_bg_sub: bool = False self.precomputed_bg_sub: bool = False
# Zero-padding the stored FFT images were computed at; 1 = natural
# resolution, which is all a v5 PREC or CACH v1 tail can hold.
self.precomputed_pad_factor: int = 1
def cached_dc_mv(self, angle_idx: int, ch_idx: int) -> np.ndarray | None: def cached_dc_mv(self, angle_idx: int, ch_idx: int) -> np.ndarray | None:
"""A stored DC image (already in mV) for (angle, channel), or None.""" """A stored DC image (already in mV) for (angle, channel), or None."""
@@ -388,34 +375,21 @@ class SrasFile:
angles = np.frombuffer(f.read(n_angles * 4), dtype=">f4").astype(np.float32) angles = np.frombuffer(f.read(n_angles * 4), dtype=">f4").astype(np.float32)
x_start = np.empty(n_angles, dtype=np.float64) x_start = np.empty(n_angles, dtype=np.float64)
x_delta = np.empty(n_angles, dtype=np.float64)
n_frames = np.empty(n_angles, dtype=np.int64) n_frames = np.empty(n_angles, dtype=np.int64)
n_rows = np.empty(n_angles, dtype=np.int64) n_rows = np.empty(n_angles, dtype=np.int64)
for a in range(n_angles): for a in range(n_angles):
xs, xd, nf, nr = _read_struct(f, GEO_FMT_V6) xs, xd, nf, nr = _read_struct(f, GEO_FMT_V6)
x_start[a], x_delta[a], n_frames[a], n_rows[a] = xs, xd, nf, nr x_start[a], n_frames[a], n_rows[a] = xs, nf, nr
y_pos_per_angle = [ y_pos_per_angle = [
np.frombuffer(f.read(int(n_rows[a]) * 4), dtype=">f4").astype(np.float32) np.frombuffer(f.read(int(n_rows[a]) * 4), dtype=">f4").astype(np.float32)
for a in range(n_angles) for a in range(n_angles)
] ]
# Verbatim on-disk spans of the preamble and background sections,
# kept so file-rewriting tools (sras_edit_scans) can carry them
# over byte-for-byte without re-parsing.
span_start = f.tell()
self._set_calibration(_read_preambles(f, n_ch), n_ch) self._set_calibration(_read_preambles(f, n_ch), n_ch)
span_end = f.tell()
f.seek(span_start)
self.preambles_raw = f.read(span_end - span_start)
span_start = span_end
self.background = _read_background(f) self.background = _read_background(f)
span_end = f.tell()
f.seek(span_start)
self.background_raw = f.read(span_end - span_start)
data_offset = span_end data_offset = f.tell()
self._data_offset = data_offset self._data_offset = data_offset
@@ -454,7 +428,6 @@ class SrasFile:
self.scan_aborted = n_complete < n_angles_declared self.scan_aborted = n_complete < n_angles_declared
self.angles_deg = angles[:n_complete] self.angles_deg = angles[:n_complete]
self.x_start_mm = x_start[:n_complete] self.x_start_mm = x_start[:n_complete]
self.x_delta_mm_per_angle = x_delta[:n_complete]
self.n_frames = n_frames[:n_complete] self.n_frames = n_frames[:n_complete]
self.n_rows = n_rows[:n_complete] self.n_rows = n_rows[:n_complete]
self._y_pos_per_angle = y_pos_per_angle[:n_complete] self._y_pos_per_angle = y_pos_per_angle[:n_complete]
@@ -463,37 +436,6 @@ class SrasFile:
if self.version == 7 and offset < file_size: if self.version == 7 and offset < file_size:
self._parse_cach_section(offset) self._parse_cach_section(offset)
# ------------------------------------------------------------------
# Byte-layout accessors (public: used by file-rewriting tools)
# ------------------------------------------------------------------
@property
def data_offset(self) -> int:
"""File offset where the waveform data begins (headers end)."""
return self._data_offset
@property
def y_pos_per_angle(self) -> list[np.ndarray]:
"""Per-angle Y row positions (mm). The list and its arrays are the
live parsed state — tools that reproject may replace entries."""
return self._y_pos_per_angle
@y_pos_per_angle.setter
def y_pos_per_angle(self, value: list[np.ndarray]):
self._y_pos_per_angle = value
def iter_angle_blocks(self):
"""Yields (angle_idx, byte_offset, byte_count) for each complete
angle's waveform block. Works for every version: legacy files have
uniform per-angle geometry, so the same walk applies."""
offset = self._data_offset
for a in range(self.n_angles):
nbytes = (int(self.n_rows[a]) * self.n_channels
* int(self.n_frames[a]) * self.samples_per_frame
* self.bytes_per_sample)
yield a, offset, nbytes
offset += nbytes
# ------------------------------------------------------------------ # ------------------------------------------------------------------
# v7 cache tail (CACH section: precomputed DC / FFT images) # v7 cache tail (CACH section: precomputed DC / FFT images)
# ------------------------------------------------------------------ # ------------------------------------------------------------------
@@ -503,24 +445,24 @@ class SrasFile:
start), derived purely from the header + Per-Angle Geometry Table — start), derived purely from the header + Per-Angle Geometry Table —
independent of whether a cache tail is actually present. Used by independent of whether a cache tail is actually present. Used by
both the parser and the in-place writer.""" both the parser and the in-place writer."""
end = self._data_offset waveform_bytes = sum(
for _, offset, nbytes in self.iter_angle_blocks(): int(self.n_rows[a]) * self.n_channels * int(self.n_frames[a])
end = offset + nbytes * self.samples_per_frame * self.bytes_per_sample
return end for a in range(self.n_angles)
)
return self._data_offset + int(waveform_bytes)
def _read_cache_block(self, f, hdr_fmt: str, magic: bytes, def _read_cache_block(self, f, hdr_fmt: str, magic: bytes,
stores: list[list]) -> tuple | None: stores: list[list]) -> int | None:
"""Read one CACH sub-block header, then its per-angle image entries """Read one CACH sub-block header, then its per-angle image entries
into *stores* (one list per image the block stores per angle). into *stores* (one list per image the block stores per angle).
Every sub-block header is (magic, flags, n_stored, *extras). Returns Returns the header's flags byte, or None if the block is malformed.
everything after the magic, so a caller that has extras knows how to
read them, or None if the block is malformed.
""" """
raw = f.read(struct.calcsize(hdr_fmt)) raw = f.read(struct.calcsize(hdr_fmt))
if len(raw) < struct.calcsize(hdr_fmt): if len(raw) < struct.calcsize(hdr_fmt):
return None return None
block_magic, flags, n_stored, *extras = struct.unpack(hdr_fmt, raw) block_magic, flags, n_stored = struct.unpack(hdr_fmt, raw)
if block_magic != magic: if block_magic != magic:
return None return None
for _ in range(n_stored): for _ in range(n_stored):
@@ -530,7 +472,7 @@ class SrasFile:
shape = self.image_shape(angle_idx) shape = self.image_shape(angle_idx)
for store in stores: for store in stores:
store[angle_idx] = _read_f32_image(f, shape) store[angle_idx] = _read_f32_image(f, shape)
return (flags, *extras) return flags
def _parse_cach_section(self, offset: int): def _parse_cach_section(self, offset: int):
"""Parse the v7 CACH tail that holds precomputed DC/FFT images.""" """Parse the v7 CACH tail that holds precomputed DC/FFT images."""
@@ -540,8 +482,7 @@ class SrasFile:
if len(header_raw) < CACH_HDR_SIZE: if len(header_raw) < CACH_HDR_SIZE:
return return
magic, cach_version, block_flags = struct.unpack(CACH_HDR_FMT, header_raw) magic, cach_version, block_flags = struct.unpack(CACH_HDR_FMT, header_raw)
if (magic != CACH_MAGIC if magic != CACH_MAGIC or cach_version != CACH_VERSION:
or not CACH_VERSION_MIN <= cach_version <= CACH_VERSION):
return return
if block_flags & CACH_FLAG_DC: if block_flags & CACH_FLAG_DC:
@@ -551,23 +492,17 @@ class SrasFile:
return return
if block_flags & CACH_FLAG_FFT: if block_flags & CACH_FLAG_FFT:
# v1 has no pad field: those images are natural-resolution. flags = self._read_cache_block(
hdr_fmt = SFFT_HDR_FMT if cach_version >= 2 else SFFT_HDR_FMT_V1 f, SFFT_HDR_FMT, SFFT_MAGIC, [self.precomputed_freq_mhz])
header = self._read_cache_block( if flags is None:
f, hdr_fmt, SFFT_MAGIC, [self.precomputed_freq_mhz])
if header is None:
return return
flags, *extras = header
self.precomputed_bg_sub = bool(flags & SFFT_FLAG_BG_SUB) self.precomputed_bg_sub = bool(flags & SFFT_FLAG_BG_SUB)
self.precomputed_pad_factor = (
min(max(1, extras[0]), PAD_FACTOR_MAX) if extras else 1)
def write_v7_cache(self, *, def write_v7_cache(self, *,
new_dc3_mv: list[np.ndarray | None] | None = None, new_dc3_mv: list[np.ndarray | None] | None = None,
new_dc4_mv: list[np.ndarray | None] | None = None, new_dc4_mv: list[np.ndarray | None] | None = None,
new_freq_mhz: list[np.ndarray | None] | None = None, new_freq_mhz: list[np.ndarray | None] | None = None,
new_bg_sub: bool | None = None, new_bg_sub: bool | None = None):
new_pad_factor: int | None = None):
"""Store computed DC and/or FFT images into this file's CACH tail, """Store computed DC and/or FFT images into this file's CACH tail,
in place, converting a v6 source to v7 (or updating an existing v7 in place, converting a v6 source to v7 (or updating an existing v7
file). Only the block(s) passed in are recomputed; whichever block file). Only the block(s) passed in are recomputed; whichever block
@@ -587,11 +522,6 @@ class SrasFile:
final_dc4 = new_dc4_mv if new_dc4_mv is not None else self.precomputed_dc4_mv final_dc4 = new_dc4_mv if new_dc4_mv is not None else self.precomputed_dc4_mv
final_freq = new_freq_mhz if new_freq_mhz is not None else self.precomputed_freq_mhz final_freq = new_freq_mhz if new_freq_mhz is not None else self.precomputed_freq_mhz
final_bg_sub = new_bg_sub if new_bg_sub is not None else self.precomputed_bg_sub final_bg_sub = new_bg_sub if new_bg_sub is not None else self.precomputed_bg_sub
final_pad = (new_pad_factor if new_pad_factor is not None
else self.precomputed_pad_factor)
if not 1 <= final_pad <= PAD_FACTOR_MAX:
raise ValueError(
f"pad factor {final_pad} outside 1..{PAD_FACTOR_MAX}")
dc_entries = [a for a in range(self.n_angles) if final_dc3[a] is not None] dc_entries = [a for a in range(self.n_angles) if final_dc3[a] is not None]
fft_entries = [a for a in range(self.n_angles) if final_freq[a] is not None] fft_entries = [a for a in range(self.n_angles) if final_freq[a] is not None]
@@ -611,8 +541,7 @@ class SrasFile:
if fft_entries: if fft_entries:
fft_flags = SFFT_FLAG_BG_SUB if final_bg_sub else 0 fft_flags = SFFT_FLAG_BG_SUB if final_bg_sub else 0
payload += struct.pack(SFFT_HDR_FMT, SFFT_MAGIC, fft_flags, payload += struct.pack(SFFT_HDR_FMT, SFFT_MAGIC, fft_flags, len(fft_entries))
len(fft_entries), final_pad)
for a in fft_entries: for a in fft_entries:
payload += struct.pack(">H", a) payload += struct.pack(">H", a)
payload += final_freq[a].astype(">f4").tobytes() payload += final_freq[a].astype(">f4").tobytes()
@@ -640,7 +569,6 @@ class SrasFile:
self.precomputed_dc4_mv = final_dc4 self.precomputed_dc4_mv = final_dc4
self.precomputed_freq_mhz = final_freq self.precomputed_freq_mhz = final_freq
self.precomputed_bg_sub = final_bg_sub self.precomputed_bg_sub = final_bg_sub
self.precomputed_pad_factor = final_pad
# ------------------------------------------------------------------ # ------------------------------------------------------------------
# Axes helpers # Axes helpers
sras_viewer.py
+2730
View File
File diff suppressed because it is too large Load Diff
sras_viewer/__init__.py
-24
View File
@@ -1,24 +0,0 @@
"""
SRAS Scan File Viewer
PyQt6 application for visualizing channel data from .sras binary scan files.
Channel semantics (fixed by sc3_aui_app.py acquisition settings):
CH1 — RF Acoustic Packet (AC-coupled, 100 mV/div): FFT → peak frequency
CH3 — Bias A (DC-coupled, 50 mV/div): waveform mean
CH4 — Bias B (DC-coupled, 50 mV/div): waveform mean
RF images are masked: pixels where CH4_dc < dc_threshold show 0.
File parsing lives in sras_format, image/alignment math in sras_compute, and
background workers in sras_workers — none of which import Qt or matplotlib,
so multiprocessing children can load them cheaply.
"""
import faulthandler
faulthandler.enable() # print a native stack trace on SIGSEGV/SIGABRT/etc.
from .canvases import ImageCanvas, RoiQuad, WaveformCanvas # noqa: E402,F401
from .common import CH_LABELS, CMAPS, VELOCITY_MODE_IDX # noqa: E402,F401
from .dialogs import FftOptionsDialog, ManualAlignmentDialog # noqa: E402,F401
from .main_window import SrasViewerWindow, main # noqa: E402,F401
sras_viewer/__main__.py
-4
View File
@@ -1,4 +0,0 @@
from .main_window import main
if __name__ == "__main__":
main()
sras_viewer/canvases.py
-542
View File
@@ -1,542 +0,0 @@
"""Matplotlib canvases and the ROI primitive."""
import numpy as np
from matplotlib.backends.backend_qtagg import FigureCanvasQTAgg
from matplotlib.figure import Figure
from matplotlib.patches import Polygon
from matplotlib.path import Path as MplPath
from PyQt6.QtCore import Qt, pyqtSignal
from PyQt6.QtGui import QKeyEvent
from PyQt6.QtWidgets import QSizePolicy
from sras_format import CH1_IDX, CH3_IDX, CH4_IDX, CH_NAMES, SrasFile, adc_to_mv
# ---------------------------------------------------------------------------
# ROI (free quadrilateral in data coordinates)
# ---------------------------------------------------------------------------
class RoiQuad:
"""Free quadrilateral defined in data coordinates (mm).
Stored as 4 corner points (shape (4, 2)) in CCW order: BL, BR, TR, TL.
Each corner can be positioned independently, allowing skewed /
non-orthogonal regions of interest. Because it lives in scan/data
coords it persists unchanged when the displayed channel/mode switches.
"""
def __init__(self, pts: np.ndarray):
"""pts : array-like, shape (4, 2)."""
self._pts = np.asarray(pts, dtype=np.float64).reshape(4, 2).copy()
@classmethod
def from_bbox(cls, x0: float, y0: float, x1: float, y1: float) -> "RoiQuad":
"""Create an axis-aligned rectangle from two opposite corners."""
lx, rx = min(x0, x1), max(x0, x1)
by, ty = min(y0, y1), max(y0, y1)
return cls(np.array([[lx, by], [rx, by], [rx, ty], [lx, ty]]))
def copy(self) -> "RoiQuad":
return RoiQuad(self._pts.copy())
def corners(self) -> np.ndarray:
"""World-coord corners, shape (4, 2), CCW: BL, BR, TR, TL."""
return self._pts.copy()
def centroid(self) -> np.ndarray:
return self._pts.mean(axis=0)
def bbox_size(self) -> np.ndarray:
"""Width and height of the axis-aligned bounding box, shape (2,)."""
return self._pts.max(axis=0) - self._pts.min(axis=0)
def contains(self, x: float, y: float) -> bool:
return bool(MplPath(self._pts).contains_point((x, y)))
def mask_for_grid(self, x_axis: np.ndarray,
y_axis: np.ndarray) -> np.ndarray:
"""Boolean mask (n_rows, n_frames) of pixels whose centres lie
inside the quadrilateral.
Only the quad's axis-aligned bounding box is tested — meshgrid and
contains_points over the *whole* grid would be tens of millions of
point-in-polygon tests (and hundreds of MB of float64 temporaries)
on a large scan, on every ROI edit.
"""
x = np.asarray(x_axis, dtype=np.float64)
y = np.asarray(y_axis, dtype=np.float64)
mask = np.zeros((y.size, x.size), dtype=bool)
(x0, y0), (x1, y1) = self._pts.min(axis=0), self._pts.max(axis=0)
cols = np.nonzero((x >= x0) & (x <= x1))[0]
rows = np.nonzero((y >= y0) & (y <= y1))[0]
if cols.size == 0 or rows.size == 0:
return mask
c0, c1 = int(cols[0]), int(cols[-1]) + 1
r0, r1 = int(rows[0]), int(rows[-1]) + 1
X, Y = np.meshgrid(x[c0:c1], y[r0:r1])
inside = MplPath(self._pts).contains_points(
np.column_stack([X.ravel(), Y.ravel()]))
mask[r0:r1, c0:c1] = inside.reshape(X.shape)
return mask
# ---------------------------------------------------------------------------
# Matplotlib canvases
# ---------------------------------------------------------------------------
class ImageCanvas(FigureCanvasQTAgg):
pixel_clicked = pyqtSignal(int, int) # row_idx, frame_idx
roi_changed = pyqtSignal() # ROI created / edited / cleared
draw_mode_changed = pyqtSignal(bool) # "draw new ROI" arm toggled
# Interaction state values
_IDLE = "idle"
_DRAW_NEW = "draw_new"
_MOVE = "move"
_DRAG_CORNER = "drag_corner"
# Hit tolerance (display pixels) for handles.
_HANDLE_PX = 12
_CLICK_THRESH_PX = 4 # releases within this of press count as a click
def __init__(self, parent=None):
fig = Figure(figsize=(7, 5), tight_layout=True)
self.ax = fig.add_subplot(111)
super().__init__(fig)
self.setParent(parent)
self.setSizePolicy(QSizePolicy.Policy.Expanding, QSizePolicy.Policy.Expanding)
self._extent = None
self._img_shape = None
# ROI state
self._roi: RoiQuad | None = None
self._roi_artists: list = []
self._state = self._IDLE
self._draw_mode = False
# Per-interaction snapshots / anchors
self._press_xy: tuple[float, float] | None = None
self._press_pixel: tuple[float, float] | None = None
self._press_button = None
self._snapshot: RoiQuad | None = None
self._drag_corner_idx: int = -1
self._move_anchor = None # press-point in world coords
self._draw_previous: RoiQuad | None = None
self.mpl_connect("button_press_event", self._on_press)
self.mpl_connect("motion_notify_event", self._on_motion)
self.mpl_connect("button_release_event", self._on_release)
# ------------------------------------------------------------------
# Public API
# ------------------------------------------------------------------
def show_image(self, img: np.ndarray, extent: list[float], cmap: str,
vmin: float, vmax: float, xlabel: str, ylabel: str, title: str,
colorbar_label: str = ""):
self.figure.clf()
self.ax = self.figure.add_subplot(111)
# Patches and lines are destroyed by figure.clf(); drop stale refs.
self._roi_artists = []
self._extent = extent
self._img_shape = img.shape
im = self.ax.imshow(
img, aspect="auto", origin="upper",
extent=extent, cmap=cmap, vmin=vmin, vmax=vmax,
interpolation="nearest",
)
cb = self.figure.colorbar(im, ax=self.ax, fraction=0.046, pad=0.04)
if colorbar_label:
cb.set_label(colorbar_label)
self.ax.set_xlabel(xlabel)
self.ax.set_ylabel(ylabel)
self.ax.set_title(title)
# Re-draw the ROI (if any) on top of the fresh image so it persists
# unchanged across mode / angle / channel switches.
self._draw_roi()
self.draw()
def get_roi(self) -> RoiQuad | None:
return self._roi
def set_roi(self, roi: RoiQuad | None):
self._roi = roi.copy() if roi is not None else None
self._draw_roi()
self.draw_idle()
self.roi_changed.emit()
def clear_roi(self):
self._roi = None
self._remove_roi_artists()
self.draw_idle()
self.roi_changed.emit()
def start_drawing(self):
"""Arm the next click+drag on the image to create a new ROI,
replacing any existing one."""
self._draw_mode = True
self.setCursor(Qt.CursorShape.CrossCursor)
self.draw_mode_changed.emit(True)
def cancel_drawing(self):
if self._draw_mode:
self._draw_mode = False
self.setCursor(Qt.CursorShape.ArrowCursor)
self.draw_mode_changed.emit(False)
# ------------------------------------------------------------------
# Rendering
# ------------------------------------------------------------------
def _remove_roi_artists(self):
for a in self._roi_artists:
try:
a.remove()
except (ValueError, AttributeError, NotImplementedError):
pass
self._roi_artists = []
def _draw_roi(self):
self._remove_roi_artists()
if self._roi is None or self.ax is None:
return
corners = self._roi.corners()
# Filled quad, then a sharp unfilled edge for visibility over bright
# images, then draggable corner handles.
for kwargs in (
dict(fill=True, facecolor="#ffd93a", edgecolor="#e53935",
alpha=0.22, linewidth=2.0, zorder=10),
dict(fill=False, edgecolor="#e53935", linewidth=1.8, zorder=11),
):
patch = Polygon(corners, closed=True, **kwargs)
self.ax.add_patch(patch)
self._roi_artists.append(patch)
self._roi_artists.append(self.ax.scatter(
corners[:, 0], corners[:, 1], s=60, c="white",
edgecolors="#e53935", linewidths=1.6, zorder=13))
# ------------------------------------------------------------------
# Hit testing (display pixels for handles, data coords for "inside")
# ------------------------------------------------------------------
def _hit_test(self, event) -> tuple[str, int | None] | None:
if self._roi is None or self.ax is None:
return None
if event.x is None or event.y is None:
return None
corners_disp = self.ax.transData.transform(self._roi.corners())
click = np.array([event.x, event.y])
for i in range(4):
if np.hypot(*(corners_disp[i] - click)) <= self._HANDLE_PX:
return ("corner", i)
if event.xdata is not None and event.ydata is not None:
if self._roi.contains(event.xdata, event.ydata):
return ("inside", None)
return None
# ------------------------------------------------------------------
# Mouse event handlers
# ------------------------------------------------------------------
def _on_press(self, event):
if event.inaxes is not self.ax or self._extent is None:
return
if event.button != 1: # only left mouse button
return
# If the matplotlib toolbar is in pan / zoom mode, let it handle
# the interaction instead of starting a ROI manipulation.
tb = getattr(self, "toolbar", None)
if tb is not None and getattr(tb, "mode", ""):
return
self._press_xy = (event.xdata, event.ydata)
self._press_pixel = (event.x, event.y)
self._press_button = event.button
if self._draw_mode:
self._draw_previous = self._roi.copy() if self._roi else None
self._roi = RoiQuad.from_bbox(event.xdata, event.ydata,
event.xdata, event.ydata)
self._state = self._DRAW_NEW
self._draw_roi()
self.draw_idle()
return
hit = self._hit_test(event)
if hit is None:
self._state = self._IDLE
return
kind, idx = hit
self._snapshot = self._roi.copy()
if kind == "corner":
self._state = self._DRAG_CORNER
self._drag_corner_idx = idx
else:
self._state = self._MOVE
self._move_anchor = (event.xdata, event.ydata)
def _on_motion(self, event):
if self._state == self._IDLE:
return
if event.xdata is None or event.ydata is None:
return
if event.inaxes is not self.ax:
return
if self._state == self._DRAW_NEW:
x0, y0 = self._press_xy
self._roi = RoiQuad.from_bbox(x0, y0, event.xdata, event.ydata)
elif self._state == self._MOVE:
delta = np.array([event.xdata - self._move_anchor[0],
event.ydata - self._move_anchor[1]])
self._roi._pts = self._snapshot.corners() + delta
elif self._state == self._DRAG_CORNER:
self._roi._pts[self._drag_corner_idx] = [event.xdata, event.ydata]
self._draw_roi()
self.draw_idle()
def _on_release(self, event):
if event.button != 1 and self._press_button != 1:
return
prev_state = self._state
self._state = self._IDLE
try:
if prev_state == self._DRAW_NEW:
self._finish_draw()
elif prev_state in (self._MOVE, self._DRAG_CORNER):
self._draw_roi()
self.draw_idle()
self.roi_changed.emit()
else:
self._maybe_emit_pixel_click(event)
finally:
self._press_xy = self._press_pixel = None
self._press_button = None
def _finish_draw(self):
"""Commit (or reject) a freshly-dragged quad."""
if self._extent is not None:
x0, x1, y_bot, y_top = self._extent
min_w = abs(x1 - x0) * 0.01 # minimum: 1% of each axis range
min_h = abs(y_bot - y_top) * 0.01
else:
min_w = min_h = 1e-6
if self._roi is None:
too_small = True
else:
bbox = self._roi.bbox_size()
too_small = bbox[0] < min_w or bbox[1] < min_h
if too_small:
self._roi = self._draw_previous
self._draw_previous = None
self.cancel_drawing()
self._draw_roi()
self.draw_idle()
self.roi_changed.emit()
def _maybe_emit_pixel_click(self, event):
"""A release close enough to its press counts as a pixel click."""
if (self._press_pixel is None or event.x is None or event.y is None
or self._extent is None or event.inaxes is not self.ax
or event.xdata is None):
return
dx_px = event.x - self._press_pixel[0]
dy_px = event.y - self._press_pixel[1]
if dx_px * dx_px + dy_px * dy_px > self._CLICK_THRESH_PX ** 2:
return
x0, x1, y_bot, y_top = self._extent
n_rows, n_frames = self._img_shape
col = int((event.xdata - x0) / (x1 - x0) * n_frames)
row = int((event.ydata - y_top) / (y_bot - y_top) * n_rows)
self.pixel_clicked.emit(max(0, min(row, n_rows - 1)),
max(0, min(col, n_frames - 1)))
class WaveformCanvas(FigureCanvasQTAgg):
def __init__(self, parent=None):
fig = Figure(figsize=(8, 3), tight_layout=True)
self.ax_wave = fig.add_subplot(121)
self.ax_right = fig.add_subplot(122)
super().__init__(fig)
self.setParent(parent)
self.setSizePolicy(QSizePolicy.Policy.Expanding, QSizePolicy.Policy.Expanding)
def show_rf_waveform(self, sras: SrasFile, angle_idx: int,
row_idx: int, frame_idx: int,
apply_bg_sub: bool = True):
"""CH1 RF: time-domain + FFT spectrum.
If apply_bg_sub is True and sras.background is not None, the background
waveform is overlaid on the time-domain plot and the FFT is computed
on the subtracted signal. The unsubtracted FFT is also shown faintly
for comparison.
"""
data = sras.data[angle_idx]
waveform = data[row_idx, CH1_IDX, frame_idx, :].astype(np.float32)
t_ns = sras.time_axis_ns()
f_mhz = sras.freq_axis_mhz()
dc3_val = data[row_idx, CH3_IDX, frame_idx, :].astype(np.float32).mean()
dc4_val = data[row_idx, CH4_IDX, frame_idx, :].astype(np.float32).mean()
bg = sras.background if (apply_bg_sub and sras.background is not None) else None
waveform_plot = waveform - bg if bg is not None else waveform
self.ax_wave.cla()
self.ax_right.cla()
if bg is not None:
self.ax_wave.plot(t_ns, waveform, linewidth=0.5, color="#aaaaaa",
label="raw", zorder=1)
self.ax_wave.plot(t_ns, bg, linewidth=0.5, color="#e07030",
linestyle="--", label="background", zorder=2)
self.ax_wave.plot(t_ns, waveform_plot, linewidth=0.7, color="#4488cc",
label="subtracted", zorder=3)
self.ax_wave.legend(fontsize=7, loc="upper right")
else:
self.ax_wave.plot(t_ns, waveform, linewidth=0.7, color="#4488cc")
self.ax_wave.set_xlabel("Time (ns)")
self.ax_wave.set_ylabel("ADC counts")
bg_tag = " [bg sub]" if bg is not None else ""
dc3_mv = adc_to_mv(dc3_val, *sras.cal(CH3_IDX))
dc4_mv = adc_to_mv(dc4_val, *sras.cal(CH4_IDX))
self.ax_wave.set_title(
f"CH1 RF row={row_idx} frame={frame_idx}{bg_tag}\n"
f"CH3={dc3_val:.1f} CH4={dc4_val:.1f} "
f"({dc3_mv:.2f} / {dc4_mv:.2f} mV)",
fontsize=8,
)
# FFT of the (possibly subtracted) waveform
power_sub = np.abs(np.fft.rfft(waveform_plot)) ** 2
power_sub[0] = 0.0
peak_mhz = f_mhz[int(np.argmax(power_sub))]
if bg is not None:
# Also show the unsubtracted FFT for reference
power_raw = np.abs(np.fft.rfft(waveform)) ** 2
power_raw[0] = 0.0
self.ax_right.plot(f_mhz, power_raw, linewidth=0.5, color="#aaaaaa",
label="raw FFT", zorder=1)
self.ax_right.plot(f_mhz, power_sub, linewidth=0.7, color="#4488cc",
label="subtracted FFT" if bg is not None else None, zorder=2)
self.ax_right.axvline(peak_mhz, color="tomato", linestyle="--",
linewidth=1.2, label=f"peak = {peak_mhz:.1f} MHz")
self.ax_right.set_xlabel("Frequency (MHz)")
self.ax_right.set_ylabel("Power (arb.)")
self.ax_right.set_title("FFT Power Spectrum")
self.ax_right.set_xlim(0, 500)
self.ax_right.legend(fontsize=8)
self.draw()
def show_dc_waveform(self, sras: SrasFile, angle_idx: int, ch_idx: int,
row_idx: int, frame_idx: int):
"""CH3 or CH4 DC: time-domain + mean annotation."""
waveform = sras.data[angle_idx][row_idx, ch_idx, frame_idx, :].astype(np.float32)
mean_val = float(waveform.mean())
mean_mv = adc_to_mv(mean_val, *sras.cal(ch_idx))
self.ax_wave.cla()
self.ax_right.cla()
self.ax_wave.plot(sras.time_axis_ns(), waveform, linewidth=0.7, color="#4488cc")
self.ax_wave.axhline(mean_val, color="tomato", linestyle="--",
linewidth=1.2, label=f"mean = {mean_val:.2f} ADC")
self.ax_wave.set_xlabel("Time (ns)")
self.ax_wave.set_ylabel("ADC counts")
self.ax_wave.set_title(
f"{CH_NAMES[ch_idx]} DC row={row_idx} frame={frame_idx}")
self.ax_wave.legend(fontsize=8)
self.ax_right.text(
0.5, 0.5,
f"DC mode\n\nmean = {mean_val:.3f} ADC\n = {mean_mv:.3f} mV",
ha="center", va="center",
transform=self.ax_right.transAxes, fontsize=11,
)
self.ax_right.set_axis_off()
self.draw()
class ManualAlignOverlayCanvas(FigureCanvasQTAgg):
"""Renders ManualAlignmentDialog's multi-angle mask overlay and turns
keyboard input into translate/rotate nudge requests for whichever angle
the dialog currently has active.
A pure input+render widget — it holds no alignment state and never
touches SrasFile itself; ManualAlignmentDialog owns all of that and
decides, from these signals, whether a cheap single-layer refresh or a
full preview-canvas rebuild is needed.
FigureCanvasQTAgg is a real QWidget, so keyPressEvent works like on any
other widget, but Qt only ever delivers key events to whichever widget
currently has focus — StrongFocus, plus grabbing focus on click and once
right after the dialog is shown, are both required or arrow keys
silently do nothing.
Rotate keys are letters (Q/E), not punctuation (comma/period or
brackets): Shift+letter still reports the same Qt.Key on every platform,
whereas Shift+comma/bracket can report a different virtual key
(Key_Less / Key_BraceLeft) depending on platform and keyboard layout —
which would silently break the "Shift = coarse step" modifier for
rotation specifically. Arrow keys have no such hazard.
"""
nudge_translate = pyqtSignal(int, int, bool) # dir_x, dir_y in {-1,0,1}; coarse
nudge_rotate = pyqtSignal(int, bool) # dir in {-1,1} (CCW/CW); coarse
_TRANSLATE_KEYS = {
Qt.Key.Key_Left: (-1, 0),
Qt.Key.Key_Right: (1, 0),
Qt.Key.Key_Up: (0, -1),
Qt.Key.Key_Down: (0, 1),
}
_ROTATE_KEYS = {Qt.Key.Key_Q: 1, Qt.Key.Key_E: -1} # CCW, CW
def __init__(self, parent=None):
fig = Figure(figsize=(6, 6), tight_layout=True)
self.ax = fig.add_subplot(111)
super().__init__(fig)
self.setParent(parent)
self.setFocusPolicy(Qt.FocusPolicy.StrongFocus)
self.setSizePolicy(QSizePolicy.Policy.Expanding, QSizePolicy.Policy.Expanding)
self.mpl_connect("button_press_event", lambda _e: self.setFocus())
def show_overlay(self, rgba: np.ndarray, extent: list[float], title: str):
self.figure.clf()
self.ax = self.figure.add_subplot(111)
self.ax.imshow(rgba, extent=extent, origin="upper", aspect="auto")
self.ax.set_xlabel("X (mm)")
self.ax.set_ylabel("Y (mm)")
self.ax.set_title(title)
self.draw_idle() # coalesces rapid redraws — matters for key-repeat.
def keyPressEvent(self, event: QKeyEvent):
key = event.key()
coarse = bool(event.modifiers() & Qt.KeyboardModifier.ShiftModifier)
if key in self._TRANSLATE_KEYS:
dx, dy = self._TRANSLATE_KEYS[key]
self.nudge_translate.emit(dx, dy, coarse)
event.accept()
elif key in self._ROTATE_KEYS:
self.nudge_rotate.emit(self._ROTATE_KEYS[key], coarse)
event.accept()
else:
super().keyPressEvent(event)
sras_viewer/common.py
-151
View File
@@ -1,151 +0,0 @@
"""Shared constants and small layout helpers for the viewer widgets."""
from PyQt6.QtCore import Qt
from PyQt6.QtWidgets import (
QDoubleSpinBox, QFormLayout, QFrame, QGroupBox, QLabel, QScrollArea,
QSizePolicy, QVBoxLayout, QWidget,
)
from sras_format import CH1_IDX, CH3_IDX, CH4_IDX
# ---------------------------------------------------------------------------
# Display constants
# ---------------------------------------------------------------------------
CH_LABELS = [
"CH1 — RF (FFT peak freq)",
"CH3 — Bias A (DC mean)",
"CH4 — Bias B (DC mean)",
"CH1 — Velocity (SRAS)",
]
# Combo index for the derived velocity mode (uses CH1_IDX data)
VELOCITY_MODE_IDX = 3
# All modes that operate on CH1 waveforms
CH1_DERIVED_MODES = (CH1_IDX, VELOCITY_MODE_IDX)
CMAPS = ["gray", "viridis", "plasma", "inferno", "hot", "jet", "RdBu_r", "seismic"]
# (mode_str, status-bar unit, colorbar label) per channel index
_CHANNEL_DISPLAY = {
CH1_IDX: ("RF", "Peak frequency (MHz)", "MHz"),
CH3_IDX: ("DC", "DC mean (mV)", "mV"),
CH4_IDX: ("DC", "DC mean (mV)", "mV"),
VELOCITY_MODE_IDX: ("Velocity", "Velocity (m/s)", "m/s"),
}
_CSS_HINT = "font-size: 11px; color: #aaa;"
_CSS_INFO = "font-size: 11px;"
_CSS_MUTED = "color: #888; font-size: 11px;"
_CSS_WARN = "color: #e07000; font-size: 11px;"
_CSS_BUSY = "color: #4a90d9; font-size: 11px;"
# Side-panel column widths (the scroll areas that hold the controls).
_LEFT_PANEL_W = 288
_RIGHT_PANEL_W = 272
# Minimum width for a spin box so its value + suffix are never clipped.
_SPIN_MIN_W = 96
# ---------------------------------------------------------------------------
# Small layout helpers
# ---------------------------------------------------------------------------
class Jobs:
"""Keys for SrasViewerWindow's background-job registry (_run_worker /
_job_running) and its progress dialogs — one place instead of string
literals scattered across window and dialogs."""
LOAD = "load"
COMPUTE = "compute"
DC_PRECOMPUTE = "dc_precompute"
BATCH = "batch"
ALIGN = "align"
MANUAL_ALIGN_MASKS = "manual_align_masks"
MANUAL_ALIGN_CORRELATE = "manual_align_correlate"
def _make_dspin(lo: float, hi: float, decimals: int, *, suffix: str = "",
value: float | None = None, step: float | None = None) -> QDoubleSpinBox:
"""A QDoubleSpinBox with the panel-standard construction."""
spin = QDoubleSpinBox()
spin.setRange(lo, hi)
spin.setDecimals(decimals)
if suffix:
spin.setSuffix(suffix)
if step is not None:
spin.setSingleStep(step)
if value is not None:
spin.setValue(value)
spin.setMinimumWidth(_SPIN_MIN_W)
return spin
def _axes_extent(x_axis, y_axis, dx: float, dy: float) -> list[float]:
"""Matplotlib imshow extent with half-pixel margins, Y flipped so row 0
renders at the top."""
return [x_axis[0] - dx / 2, x_axis[-1] + dx / 2,
y_axis[-1] + dy / 2, y_axis[0] - dy / 2]
def _wrap_label(text: str = "", css: str | None = None) -> QLabel:
"""A word-wrapped QLabel that reports its *wrapped* height to the layout.
A plain word-wrapped QLabel advertises a single-line minimum height, so in a
fixed-width column the layout happily shrinks it and the extra lines get
clipped. Enabling height-for-width makes the box layout ask for the real
height at the column's width instead.
"""
lbl = QLabel(text)
lbl.setWordWrap(True)
sp = lbl.sizePolicy()
sp.setVerticalPolicy(QSizePolicy.Policy.Minimum)
sp.setHeightForWidth(True)
lbl.setSizePolicy(sp)
if css:
lbl.setStyleSheet(css)
return lbl
def _group(title: str) -> tuple[QGroupBox, QVBoxLayout]:
"""A group box with consistent, non-cramped internal margins."""
grp = QGroupBox(title)
lay = QVBoxLayout(grp)
lay.setContentsMargins(10, 8, 10, 10)
lay.setSpacing(6)
return grp, lay
def _form() -> QFormLayout:
"""A label/field form layout for a narrow side panel."""
form = QFormLayout()
form.setContentsMargins(0, 0, 0, 0)
form.setHorizontalSpacing(8)
form.setVerticalSpacing(6)
form.setLabelAlignment(Qt.AlignmentFlag.AlignRight
| Qt.AlignmentFlag.AlignVCenter)
form.setFormAlignment(Qt.AlignmentFlag.AlignLeft | Qt.AlignmentFlag.AlignTop)
form.setFieldGrowthPolicy(
QFormLayout.FieldGrowthPolicy.AllNonFixedFieldsGrow)
form.setRowWrapPolicy(QFormLayout.RowWrapPolicy.DontWrapRows)
return form
def _scroll_panel(inner: QWidget, width: int) -> QScrollArea:
"""Put a side panel in a fixed-width scroll area.
Without this the panels are sized by the window: a short window squeezes the
controls past their minimum heights, which is what makes text overlap the
widget below it. Scrolling keeps every control at its natural size.
"""
area = QScrollArea()
area.setWidget(inner)
area.setWidgetResizable(True)
area.setFrameShape(QFrame.Shape.NoFrame)
area.setHorizontalScrollBarPolicy(Qt.ScrollBarPolicy.ScrollBarAlwaysOff)
area.setVerticalScrollBarPolicy(Qt.ScrollBarPolicy.ScrollBarAsNeeded)
area.setFixedWidth(width)
area.viewport().setAutoFillBackground(False)
inner.setAutoFillBackground(False)
return area
sras_viewer/dialogs.py
-755
View File
@@ -1,755 +0,0 @@
"""FFT Options and Manual Alignment dialogs."""
from typing import TYPE_CHECKING
import matplotlib as mpl
import numpy as np
from matplotlib.backends.backend_qtagg import NavigationToolbar2QT
from PyQt6.QtCore import QSignalBlocker, pyqtSignal
from PyQt6.QtWidgets import (
QButtonGroup, QComboBox, QDialog, QDialogButtonBox, QGroupBox,
QHBoxLayout, QLabel, QMessageBox, QPushButton, QRadioButton, QSpinBox,
QVBoxLayout, QWidget,
)
import sras_compute as compute
from sras_compute import (
PYFFTW_AVAILABLE, ManualAngleParams, build_manual_alignment,
delete_manual_alignment, save_manual_alignment,
)
from sras_format import SrasFile
from sras_workers import Ch4MaskWorker, CrossCorrelateWorker
from .canvases import ManualAlignOverlayCanvas
from .common import (
_CSS_HINT, _CSS_MUTED, _CSS_WARN, Jobs, _axes_extent, _form, _group, _make_dspin,
_scroll_panel, _wrap_label,
)
if TYPE_CHECKING:
from .main_window import SrasViewerWindow
# ---------------------------------------------------------------------------
# FFT Options dialog
# ---------------------------------------------------------------------------
class FftOptionsDialog(QDialog):
"""Configure FFT backend and zero-padding.
Changes take effect only when the user clicks Apply. Cancel discards
all pending edits. The live 'frequency resolution' label updates as
the user adjusts the pad factor so they can see the trade-off before
committing.
"""
def __init__(self, parent=None, *,
current_backend: str,
current_pad_factor: int,
samples_per_frame: int | None,
sample_rate_hz: float | None,
grating_um: float,
cached_pad_factor: int | None = None):
super().__init__(parent)
self.setWindowTitle("FFT Options")
self.setModal(True)
self.setMinimumWidth(380)
self._samples_per_frame = samples_per_frame
self._sample_rate_hz = sample_rate_hz
self._grating_um = grating_um
# The pad the open file's stored FFT cache was computed at, if it has
# one — picking anything else here makes that cache unreadable, which
# is worth saying before Apply rather than after.
self._cached_pad_factor = cached_pad_factor
layout = QVBoxLayout(self)
# ---- Backend ---------------------------------------------------
grp_backend = QGroupBox("FFT Backend")
bl = QVBoxLayout(grp_backend)
self._btn_scipy = QRadioButton("SciPy FFT (pocketfft) (always available)")
self._btn_pyfftw = QRadioButton(
"pyFFTW (faster for large arrays)" if PYFFTW_AVAILABLE
else "pyFFTW (not installed — run: pip install pyfftw)")
self._btn_pyfftw.setEnabled(PYFFTW_AVAILABLE)
self._backend_group = QButtonGroup(self)
self._backend_group.addButton(self._btn_scipy, id=0)
self._backend_group.addButton(self._btn_pyfftw, id=1)
if current_backend == "pyfftw" and PYFFTW_AVAILABLE:
self._btn_pyfftw.setChecked(True)
else:
self._btn_scipy.setChecked(True)
bl.addWidget(self._btn_scipy)
bl.addWidget(self._btn_pyfftw)
layout.addWidget(grp_backend)
# ---- Zero-padding ----------------------------------------------
grp_zp = QGroupBox("Zero-Padding")
zl = QVBoxLayout(grp_zp)
pad_row = QHBoxLayout()
pad_row.addWidget(QLabel("Pad factor:"))
self._spin_pad = QSpinBox()
self._spin_pad.setRange(1, 256)
self._spin_pad.setValue(max(1, current_pad_factor))
self._spin_pad.setToolTip(
"Multiply the waveform length by this factor via zero-padding\n"
"before computing the FFT.\n"
"1 = no padding (natural length).\n"
"Powers of 2 (2, 4, 8 …) give the best performance."
)
self._spin_pad.valueChanged.connect(self._update_info)
pad_row.addWidget(self._spin_pad)
zl.addLayout(pad_row)
self._lbl_nfft = QLabel()
self._lbl_freq_res = QLabel()
self._lbl_vel_res = QLabel()
for lbl in (self._lbl_nfft, self._lbl_freq_res, self._lbl_vel_res):
lbl.setStyleSheet(_CSS_HINT)
zl.addWidget(lbl)
self._lbl_cache = QLabel()
self._lbl_cache.setWordWrap(True)
self._lbl_cache.setStyleSheet(_CSS_WARN)
zl.addWidget(self._lbl_cache)
layout.addWidget(grp_zp)
# ---- Buttons ---------------------------------------------------
buttons = QDialogButtonBox()
buttons.addButton("Apply", QDialogButtonBox.ButtonRole.AcceptRole
).clicked.connect(self.accept)
buttons.addButton("Cancel", QDialogButtonBox.ButtonRole.RejectRole
).clicked.connect(self.reject)
layout.addWidget(buttons)
self._update_info()
def _update_info(self):
spf = self._samples_per_frame
sr = self._sample_rate_hz
pad = self._spin_pad.value()
self._lbl_cache.setText(
"" if self._cached_pad_factor in (None, pad) else
f"! This file's stored FFT cache was computed at pad "
f"{self._cached_pad_factor}x, so at {pad}x it can't be used and "
f"CH1/Velocity will recompute. Re-run Convert -> Batch Compute FFT "
f"to cache at {pad}x.")
if spf is None or sr is None:
self._lbl_nfft.setText("Load a file to preview FFT parameters.")
self._lbl_freq_res.setText("")
self._lbl_vel_res.setText("")
return
n_fft = spf * pad
freq_res_hz = sr / n_fft
freq_res_mhz = freq_res_hz / 1e6
# v (m/s) = freq (MHz) × grating (µm)
vel_res_ms = freq_res_mhz * self._grating_um
self._lbl_nfft.setText(f"FFT points: {spf} × {pad} = {n_fft:,}")
self._lbl_freq_res.setText(
f"Frequency bin: {freq_res_mhz:.4f} MHz ({freq_res_hz / 1e3:.2f} kHz)")
self._lbl_vel_res.setText(
f"Velocity bin: {vel_res_ms:.3f} m/s "
f"(at grating = {self._grating_um:.2f} µm)")
def get_backend(self) -> str:
return "pyfftw" if self._btn_pyfftw.isChecked() and PYFFTW_AVAILABLE else "scipy"
def get_pad_factor(self) -> int:
return max(1, self._spin_pad.value())
class ManualAlignmentDialog(QDialog):
"""Non-modal manual angle-alignment editor (Fusion -> Manual Alignment...).
Shows every angle's binarized CH4 (Bias B) mask overlaid in a distinct
color at partial opacity on one shared canvas, so translation/rotation
misalignment is visible by eye. Reference angle (always index 0) is
ground truth and never moves; every other angle is aligned to it. The
user picks an "active" angle and nudges its rotation+translation with
the keyboard; Auto Cross-Correlate finds every non-reference angle's
rotation *and* translation by registering its image against the
reference's (see compute.register_angle_to_reference) — meant to get every
angle stacked on top of each other so keyboard nudging only has to make
small corrections, not find an alignment from scratch; Auto De-rotate is
the weaker fallback that just seeds rotation from the stage's reported
angle, leaving translation alone. Save writes a JSON sidecar next to the
.sras file and hands a freshly-built, full-resolution AlignmentResult back
to the main window — the exact same object shape compute_angle_alignment
produces, so every existing Aligned-View code path (apply_alignment,
_aligned_canvas_axes, the pixel-inspector inverse-transform) works
completely unmodified.
Non-modal by design (shown via .show(), never .exec() or setModal(True))
so the user can still interact with the main window. Talks back to
SrasViewerWindow two ways: it reuses parent._run_worker/_jobs directly
for its background mask-fetch and cross-correlate steps, so the main
window's existing shutdown/lifecycle plumbing covers both for free, and
it emits alignment_saved / alignment_cleared signals for the two moments
that should actually mutate the main window's persistent state —
everything else (nudging, Auto De-rotate, Auto Cross-Correlate, threshold
edits) stays purely local to this dialog until Save.
"""
alignment_saved = pyqtSignal(object, str) # AlignmentResult, sidecar path (str)
alignment_cleared = pyqtSignal()
_PREVIEW_MARGIN_FRAC = 0.15
_BASE_ALPHA = 0.42
_ACTIVE_ALPHA = 0.75
_MAX_PREVIEW_DIM = 1024
# (label, sources passed to compute.register_angle_to_reference). "Both"
# registers on each and keeps whichever scores higher per angle, which
# costs roughly double but removes the failure mode where the single
# chosen source is the one that happens to be uninformative for one angle.
_CORRELATE_SOURCES = (
("Both, keep best (recommended)", ("signal", "mask")),
("Raw signal", ("signal",)),
("Thresholded mask", ("mask",)),
)
def __init__(self, parent: "SrasViewerWindow", sras: SrasFile, *,
ref_angle_idx: int, dc_threshold_mv: float,
seed_per_angle: dict[int, ManualAngleParams] | None,
cached_dc4_mv: dict[int, np.ndarray]):
super().__init__(parent)
self._parent = parent
self._sras = sras
self._ref_angle_idx = ref_angle_idx
self._downsample = (1, 1) # (rows, cols) block-mean factors
self._dc4_mv: dict[int, np.ndarray] = {}
self._masks_small: dict[int, np.ndarray] = {}
self._preview_layers: dict[int, np.ndarray] = {}
self._preview_origin_mm = (0.0, 0.0)
self._preview_shape = (1, 1)
self._preview_pitch_mm = (1.0, 1.0)
self._masks_ready = False
self._fit_notes: dict[int, tuple[float, str]] = {}
self._derotate_sign_flipped = False
self.setWindowTitle(f"Manual Alignment — {sras.path.name}")
self.resize(1150, 760)
self._seed_initial_params(seed_per_angle)
n = sras.n_angles
cmap = mpl.colormaps["tab10"] if n <= 10 else mpl.colormaps["tab20"]
self._angle_colors = {a: cmap(a % cmap.N)[:3] for a in range(n)}
self._active_angle = 1 if ref_angle_idx == 0 and n > 1 else 0
self._build_ui(dc_threshold_mv)
self._set_controls_enabled(False) # re-enabled once masks are ready
self._start_mask_prep(cached_dc4_mv)
def showEvent(self, event):
super().showEvent(event)
self.canvas.setFocus()
# ------------------------------------------------------------------
# Construction
# ------------------------------------------------------------------
def _seed_initial_params(self, seed_per_angle: dict[int, ManualAngleParams] | None):
seed = seed_per_angle or {}
self._angle_params: dict[int, ManualAngleParams] = {
a: (ManualAngleParams(seed[a].rotation_deg, seed[a].shift_mm)
if a in seed else ManualAngleParams())
for a in range(self._sras.n_angles)
}
self._angle_params[self._ref_angle_idx] = ManualAngleParams()
def _build_ui(self, dc_threshold_mv: float):
root = QHBoxLayout(self)
self.canvas = ManualAlignOverlayCanvas()
left = QWidget()
left_l = QVBoxLayout(left)
left_l.setContentsMargins(0, 0, 0, 0)
left_l.setSpacing(4)
left_l.addWidget(NavigationToolbar2QT(self.canvas, left))
left_l.addWidget(self.canvas)
root.addWidget(left, stretch=1)
panel = QWidget()
panel_l = QVBoxLayout(panel)
panel_l.setContentsMargins(0, 0, 0, 0)
panel_l.setSpacing(8)
panel_l.addWidget(self._build_angle_group())
panel_l.addWidget(self._build_adjust_group())
panel_l.addWidget(self._build_step_group())
panel_l.addWidget(self._build_threshold_group(dc_threshold_mv))
panel_l.addWidget(self._build_correlate_group())
panel_l.addWidget(self._build_actions_group())
self.lbl_status = _wrap_label("", _CSS_MUTED)
panel_l.addWidget(self.lbl_status)
panel_l.addStretch()
root.addWidget(_scroll_panel(panel, 320))
self._connect_controls()
def _build_angle_group(self) -> QWidget:
grp_angle, al = _group("Active Angle")
self.combo_active_angle = QComboBox()
for a in range(self._sras.n_angles):
label = f"Angle {a} ({self._sras.angles_deg[a]:.1f}°)"
if a == self._ref_angle_idx:
label += " [reference]"
self.combo_active_angle.addItem(label)
al.addWidget(self.combo_active_angle)
self.lbl_active_note = _wrap_label("", _CSS_WARN)
al.addWidget(self.lbl_active_note)
return grp_angle
def _build_adjust_group(self) -> QWidget:
self.grp_manual_adjust, mform_box = _group("Manual Adjustment")
mform = _form()
self.spin_active_rotation_deg = _make_dspin(-3600.0, 3600.0, 3, suffix=" °")
mform.addRow("Rotation:", self.spin_active_rotation_deg)
self.spin_active_shift_x_mm = _make_dspin(-1e5, 1e5, 4, suffix=" mm")
mform.addRow("Shift X:", self.spin_active_shift_x_mm)
self.spin_active_shift_y_mm = _make_dspin(-1e5, 1e5, 4, suffix=" mm")
mform.addRow("Shift Y:", self.spin_active_shift_y_mm)
mform_box.addLayout(mform)
return self.grp_manual_adjust
def _build_step_group(self) -> QWidget:
self.grp_step_sizes, sl = _group("Nudge Step Sizes")
sform = _form()
self.spin_step_translate_mm = _make_dspin(0.0001, 1000.0, 4,
suffix=" mm", value=0.01)
sform.addRow("Translate step:", self.spin_step_translate_mm)
self.spin_step_rotate_deg = _make_dspin(0.001, 90.0, 3,
suffix=" °", value=0.1)
sform.addRow("Rotate step:", self.spin_step_rotate_deg)
self.spin_step_multiplier = _make_dspin(1.0, 1000.0, 1, value=10.0)
sform.addRow("Coarse × (Shift):", self.spin_step_multiplier)
sl.addLayout(sform)
sl.addWidget(_wrap_label(
"Arrow keys nudge X/Y translation; Q/E nudge rotation (CCW/CW). "
"Hold Shift for the coarse step. Click the image once so it has "
"keyboard focus.", _CSS_HINT))
return self.grp_step_sizes
def _build_threshold_group(self, dc_threshold_mv: float) -> QWidget:
self.grp_mask_threshold, tl = _group("Mask Threshold")
tform = _form()
self.spin_mask_threshold_mv = _make_dspin(-500.0, 500.0, 3,
suffix=" mV", value=dc_threshold_mv)
tform.addRow("DC threshold:", self.spin_mask_threshold_mv)
tl.addLayout(tform)
return self.grp_mask_threshold
def _build_correlate_group(self) -> QWidget:
self.grp_correlate, cl = _group("Cross-Correlate (FFT)")
cform = _form()
self.combo_correlate_source = QComboBox()
for label, sources in self._CORRELATE_SOURCES:
self.combo_correlate_source.addItem(label, sources)
cform.addRow("Correlate on:", self.combo_correlate_source)
self.spin_correlate_search_deg = _make_dspin(0.0, 180.0, 1, suffix=" °",
value=6.0, step=1.0)
cform.addRow("Rotation search (±):", self.spin_correlate_search_deg)
cl.addLayout(cform)
self.btn_auto_correlate = QPushButton("Auto Cross-Correlate (vs Reference)")
cl.addWidget(self.btn_auto_correlate)
cl.addWidget(_wrap_label(
"Finds each non-reference angle's rotation *and* translation by "
"cross-correlating its image against the reference's — the stage's "
"reported angle is only the starting point of the search, and both "
"of its signs are tried. Run this first, then nudge only for small "
"corrections.", _CSS_HINT))
return self.grp_correlate
def _build_actions_group(self) -> QWidget:
grp_actions, acl = _group("Actions")
self.btn_auto_derotate = QPushButton("Auto De-rotate (use known angles)")
self.btn_save = QPushButton("Save Alignment")
self.btn_clear = QPushButton("Clear Alignment…")
self.btn_close = QPushButton("Close")
for btn in (self.btn_auto_derotate, self.btn_save, self.btn_clear, self.btn_close):
acl.addWidget(btn)
return grp_actions
def _connect_controls(self):
self.combo_active_angle.currentIndexChanged.connect(self._on_active_angle_changed)
self.spin_active_rotation_deg.editingFinished.connect(self._on_rotation_spin_edited)
self.spin_active_shift_x_mm.editingFinished.connect(self._on_shift_spin_edited)
self.spin_active_shift_y_mm.editingFinished.connect(self._on_shift_spin_edited)
self.spin_mask_threshold_mv.editingFinished.connect(self._on_mask_threshold_edited)
self.btn_auto_derotate.clicked.connect(self._on_auto_derotate)
self.btn_auto_correlate.clicked.connect(self._on_auto_correlate)
self.btn_save.clicked.connect(self._on_save)
self.btn_clear.clicked.connect(self._on_clear)
self.btn_close.clicked.connect(self.close)
self.canvas.nudge_translate.connect(self._on_nudge_translate)
self.canvas.nudge_rotate.connect(self._on_nudge_rotate)
with QSignalBlocker(self.combo_active_angle):
self.combo_active_angle.setCurrentIndex(self._active_angle)
self._on_active_angle_changed(self._active_angle)
# ------------------------------------------------------------------
# Mask preparation (initial CH4 fetch + threshold + downsample)
# ------------------------------------------------------------------
def _start_mask_prep(self, cached_dc4_mv: dict[int, np.ndarray]):
self._dc4_mv = dict(cached_dc4_mv)
missing = [a for a in range(self._sras.n_angles) if a not in self._dc4_mv]
if not missing:
self._finish_mask_prep()
return
self.lbl_status.setText(f"Preparing masks: 0/{len(missing)} angle(s) needed…")
started = self._parent._run_worker(
Jobs.MANUAL_ALIGN_MASKS, Ch4MaskWorker(self._sras, missing),
connect=(
("angle_done", self._on_mask_angle_done),
("error", lambda msg: self.lbl_status.setText(f"Mask prep error: {msg}")),
),
on_done=self._finish_mask_prep)
if not started:
self.lbl_status.setText(
"Could not start mask preparation (busy) — close and reopen.")
def _on_mask_angle_done(self, angle_idx: int, dc4_mv: np.ndarray):
self._dc4_mv[angle_idx] = dc4_mv
self.lbl_status.setText(
f"Preparing masks: {len(self._dc4_mv)}/{self._sras.n_angles} ready…")
def _finish_mask_prep(self):
if len(self._dc4_mv) < self._sras.n_angles:
return # a mask-worker error left some angles unfetched
# Rows and columns get their own factor. A real scan is ~7500 frames
# wide but only ~750 rows tall, so one shared factor sized for the
# frames would throw away 8x more row detail than the preview needs and
# leave the overlay too coarse in y to judge alignment by eye.
max_rows = max(img.shape[0] for img in self._dc4_mv.values())
max_cols = max(img.shape[1] for img in self._dc4_mv.values())
self._downsample = (
max(1, int(np.ceil(max_rows / self._MAX_PREVIEW_DIM))),
max(1, int(np.ceil(max_cols / self._MAX_PREVIEW_DIM))))
self._recompute_masks_small()
self._rebuild_preview_canvas()
self._set_controls_enabled(True)
self.lbl_status.setText("Ready.")
def _recompute_masks_small(self):
"""Threshold + downsample every angle's already-in-memory full-res
CH4 mV image. Cheap (a compare + block-mean), so this re-runs in
full whenever the mask-threshold spin box changes — no re-fetch.
Purely for the overlay's visuals: no alignment geometry depends on this
threshold, only which pixels the overlay paints."""
threshold = self.spin_mask_threshold_mv.value()
fy, fx = self._downsample
self._masks_small = {
a: compute.block_mean_2d((img >= threshold).astype(np.float32), fy, fx)
for a, img in self._dc4_mv.items()
}
# ------------------------------------------------------------------
# Preview canvas: full rebuild vs. incremental single-layer refresh
# ------------------------------------------------------------------
def _rebuild_preview_canvas(self):
"""Full geometry rebuild: recomputes the shared preview canvas's
origin/shape (rotation can grow the union bbox — translation alone
cannot, per the padding baked in via _PREVIEW_MARGIN_FRAC) and every
angle's reprojected mask layer. Triggered by: dialog open,
mask-threshold change, Auto De-rotate, a rotation nudge/edit of the
active angle. NOT triggered by a translation-only nudge — see
_refresh_active_preview_layer."""
dx_ref, dy_ref = compute.pixel_pitch_mm(self._sras, self._ref_angle_idx)
fy, fx = self._downsample
pitch = (dx_ref * fx, dy_ref * fy)
origin, shape = compute.canvas_for_params(
self._sras, self._ref_angle_idx, pitch, self._angle_params,
margin_frac=self._PREVIEW_MARGIN_FRAC, snap=False)
self._preview_origin_mm, self._preview_shape = origin, shape
self._preview_pitch_mm = pitch
self._preview_layers = {
a: self._reproject(a) for a in range(self._sras.n_angles)
}
self._redraw_overlay()
def _reproject(self, angle_idx: int) -> np.ndarray:
"""One angle's downsampled mask on the current preview canvas.
src_downsample must match _masks_small's block-mean factors, or the
layer lands magnified and offset instead of where the alignment
actually puts it."""
p = self._angle_params[angle_idx]
return compute.reproject_mask(
self._sras, angle_idx, self._ref_angle_idx,
self._masks_small[angle_idx], p.rotation_deg, p.shift_mm,
self._preview_pitch_mm, self._preview_origin_mm, self._preview_shape,
src_downsample=self._downsample)
def _refresh_active_preview_layer(self):
"""Cheap path for a translation-only nudge/edit of the active angle:
reproject just that one angle's downsampled mask onto the *existing*
preview canvas — every other angle's cached layer is untouched."""
self._preview_layers[self._active_angle] = self._reproject(self._active_angle)
self._redraw_overlay()
def _redraw_overlay(self):
"""Alpha-composite every angle's colored mask layer into one RGBA
image ("all thresholds overlaid with varying opacity"). Each angle
keeps a fixed, distinct color regardless of which is active; the
active angle is drawn last (on top) at a visibly higher alpha so
it's easy to track while nudging."""
if not self._preview_layers:
return # mask prep hasn't finished yet — nothing to draw
n_rows, n_cols = self._preview_shape
rgba = np.zeros((n_rows, n_cols, 4), dtype=np.float32)
order = sorted(range(self._sras.n_angles), key=lambda a: a == self._active_angle)
for a in order:
layer = self._preview_layers.get(a)
if layer is None:
continue
alpha = self._ACTIVE_ALPHA if a == self._active_angle else self._BASE_ALPHA
color = self._angle_colors[a]
fg_a = layer * alpha
for c in range(3):
rgba[..., c] = color[c] * fg_a + rgba[..., c] * rgba[..., 3] * (1 - fg_a)
rgba[..., 3] = fg_a + rgba[..., 3] * (1 - fg_a)
x0, y0 = self._preview_origin_mm
dx, dy = self._preview_pitch_mm
x_axis = x0 + np.arange(n_cols) * dx
y_axis = y0 + np.arange(n_rows) * dy
extent = _axes_extent(x_axis, y_axis, dx, dy)
title = (f"Angle {self._active_angle} active "
f"({self._sras.angles_deg[self._active_angle]:.1f}°)")
self.canvas.show_overlay(rgba, extent, title)
# ------------------------------------------------------------------
# Angle selection / nudge / edit handlers
# ------------------------------------------------------------------
def _on_active_angle_changed(self, angle_idx: int):
self._active_angle = angle_idx
is_ref = angle_idx == self._ref_angle_idx
self.grp_manual_adjust.setEnabled(self._masks_ready and not is_ref)
self.lbl_active_note.setText(
"Reference angle — defines the shared origin, not adjustable." if is_ref else "")
self._sync_active_spinboxes()
self._redraw_overlay()
def _sync_active_spinboxes(self):
p = self._angle_params[self._active_angle]
for spin, val in ((self.spin_active_rotation_deg, p.rotation_deg),
(self.spin_active_shift_x_mm, p.shift_mm[0]),
(self.spin_active_shift_y_mm, p.shift_mm[1])):
with QSignalBlocker(spin):
spin.setValue(val)
def _on_nudge_translate(self, dir_x: int, dir_y: int, coarse: bool):
if not self._masks_ready or self._active_angle == self._ref_angle_idx:
return
step = self.spin_step_translate_mm.value()
if coarse:
step *= self.spin_step_multiplier.value()
p = self._angle_params[self._active_angle]
p.shift_mm = (p.shift_mm[0] + dir_x * step, p.shift_mm[1] + dir_y * step)
self._sync_active_spinboxes()
self._refresh_active_preview_layer()
def _on_nudge_rotate(self, direction: int, coarse: bool):
if not self._masks_ready or self._active_angle == self._ref_angle_idx:
return
step = self.spin_step_rotate_deg.value()
if coarse:
step *= self.spin_step_multiplier.value()
self._angle_params[self._active_angle].rotation_deg += direction * step
self._sync_active_spinboxes()
self._rebuild_preview_canvas()
def _on_rotation_spin_edited(self):
if self._active_angle == self._ref_angle_idx:
return
self._angle_params[self._active_angle].rotation_deg = self.spin_active_rotation_deg.value()
self._rebuild_preview_canvas()
def _on_shift_spin_edited(self):
if self._active_angle == self._ref_angle_idx:
return
p = self._angle_params[self._active_angle]
p.shift_mm = (self.spin_active_shift_x_mm.value(), self.spin_active_shift_y_mm.value())
self._refresh_active_preview_layer()
def _on_mask_threshold_edited(self):
if not self._masks_ready:
return
self._recompute_masks_small()
self._rebuild_preview_canvas()
# ------------------------------------------------------------------
# Actions
# ------------------------------------------------------------------
def _on_auto_derotate(self):
"""Seed every angle's rotation from the stage's reported angle.
A starting point for nudging by eye, not an alignment: the stage's
sign convention relative to this module's is not knowable from the
file, so the sign that lines the scans up is whichever of the two looks
right in the overlay. Auto Cross-Correlate decides that from the images
instead, and is the button to reach for first.
"""
sign = -1.0 if self._derotate_sign_flipped else 1.0
self._derotate_sign_flipped = not self._derotate_sign_flipped
n_changed = 0
for a in range(self._sras.n_angles):
if a == self._ref_angle_idx:
continue
self._angle_params[a].rotation_deg = sign * compute.nominal_delta_deg(
self._sras, a, self._ref_angle_idx)
n_changed += 1
self._sync_active_spinboxes()
self._rebuild_preview_canvas()
self.lbl_status.setText(
f"Rotation set to the stage angle ({'−' if sign < 0 else '+'}delta) "
f"for {n_changed} angle(s); translation untouched. Click again to "
"try the opposite sign.")
def _on_auto_correlate(self):
if not self._masks_ready:
return
angles = [a for a in range(self._sras.n_angles) if a != self._ref_angle_idx]
if not angles:
return
worker = CrossCorrelateWorker(
self._sras, self._ref_angle_idx, angles, self._dc4_mv,
sources=self.combo_correlate_source.currentData(),
dc_threshold_mv=self.spin_mask_threshold_mv.value(),
search_deg=self.spin_correlate_search_deg.value())
self._correlate_done_count = 0
self._correlate_total = len(angles)
self._fit_notes = {}
self._set_controls_enabled(False)
self.lbl_status.setText(f"Cross-correlating: 0/{self._correlate_total} angle(s)…")
started = self._parent._run_worker(
Jobs.MANUAL_ALIGN_CORRELATE, worker,
connect=(
("angle_done", self._on_correlate_angle_done),
("error", self._on_correlate_error),
),
on_done=self._finish_auto_correlate)
if not started:
self._set_controls_enabled(True)
self.lbl_status.setText("Could not start cross-correlation (busy) — try again.")
def _on_correlate_angle_done(self, angle_idx: int, rotation_deg: float,
shift_x_mm: float, shift_y_mm: float,
score: float, source: str):
self._angle_params[angle_idx] = ManualAngleParams(rotation_deg, (shift_x_mm, shift_y_mm))
self._fit_notes[angle_idx] = (score, source)
self._correlate_done_count += 1
self.lbl_status.setText(
f"Cross-correlating: {self._correlate_done_count}/{self._correlate_total} angle(s)…")
def _on_correlate_error(self, msg: str):
self.lbl_status.setText(f"Cross-correlation error: {msg}")
def _finish_auto_correlate(self):
self._sync_active_spinboxes()
self._rebuild_preview_canvas()
self._set_controls_enabled(True)
self.lbl_status.setText(
f"Cross-correlated {self._correlate_done_count} angle(s) against "
f"Angle {self._ref_angle_idx}.\n" + self._fit_report())
def _fit_report(self) -> str:
"""Per-angle registration quality, worst first.
Surfaced rather than buried because a single bad acquisition (stage
glitch, laser dropout) registers poorly and would otherwise be fused in
silently — seeing which angle it is, is what makes dropping it with
sras_edit_scans.py actionable. The deviation from the stage's own
reported angle is shown alongside: a large one means the search and the
stage disagree, which is either a genuine mechanical error or a sign
that this angle's fit is not to be trusted.
"""
if not self._fit_notes:
return ""
rows = sorted(self._fit_notes.items(), key=lambda kv: kv[1][0])
worst = rows[0]
lines = [f"Worst fit: angle {worst[0]} (score {worst[1][0]:.3f}, "
f"{worst[1][1]})."]
drifted = []
for a, _note in rows:
nominal = compute.nominal_delta_deg(self._sras, a, self._ref_angle_idx)
got = self._angle_params[a].rotation_deg
dev = min(abs(got - nominal), abs(got + nominal))
if dev > 1.0:
drifted.append(f"{a} ({dev:.2f}°)")
if drifted:
lines.append("Rotation differs from the stage angle by >1° for "
"angle(s) " + ", ".join(drifted) + ".")
lines.append("Nudge from here for any remaining fine correction.")
return " ".join(lines)
def _on_save(self):
threshold = self.spin_mask_threshold_mv.value()
resolved = dict(self._angle_params) # already concrete floats
try:
path = save_manual_alignment(self._sras, self._ref_angle_idx, threshold, resolved)
result = build_manual_alignment(self._sras, self._ref_angle_idx,
threshold, resolved)
except OSError as exc:
QMessageBox.warning(self, "Save Alignment Failed", str(exc))
return
self.lbl_status.setText(f"Saved to {path.name}.")
self.alignment_saved.emit(result, str(path))
def _on_clear(self):
reply = QMessageBox.question(
self, "Clear Alignment",
"This resets every angle back to raw/unaligned (0° rotation, no "
"shift) and deletes the saved alignment file for this scan, if "
"any. This cannot be undone. Continue?",
QMessageBox.StandardButton.Yes | QMessageBox.StandardButton.No,
QMessageBox.StandardButton.No)
if reply != QMessageBox.StandardButton.Yes:
return
try:
existed = delete_manual_alignment(self._sras)
except OSError as exc:
QMessageBox.warning(self, "Clear Alignment Failed",
f"Could not delete the saved alignment file: {exc}")
return
self._angle_params = {a: ManualAngleParams() for a in range(self._sras.n_angles)}
self._fit_notes = {}
self._sync_active_spinboxes()
self._rebuild_preview_canvas()
self.lbl_status.setText(
"Alignment cleared; saved file removed." if existed
else "Alignment cleared (there was no saved file).")
self.alignment_cleared.emit()
def _set_controls_enabled(self, enabled: bool):
self._masks_ready = enabled
self.combo_active_angle.setEnabled(enabled)
self.grp_manual_adjust.setEnabled(enabled and self._active_angle != self._ref_angle_idx)
self.grp_step_sizes.setEnabled(enabled)
self.grp_mask_threshold.setEnabled(enabled)
self.grp_correlate.setEnabled(enabled)
self.btn_auto_derotate.setEnabled(enabled)
self.btn_save.setEnabled(enabled)
self.btn_clear.setEnabled(enabled)
sras_viewer/main_window.py
-1474
View File
File diff suppressed because it is too large Load Diff
sras_viewer_requirements.txt
+3
View File
@@ -0,0 +1,3 @@
PyQt6==6.10.2
numpy==2.4.1
matplotlib==3.10.8
sras_workers.py
+99 -104
View File
@@ -54,34 +54,6 @@ class CancellableWorker(QObject):
return self._stop return self._stop
class _PooledWorker(CancellableWorker):
"""Fans a per-item computation across a thread pool, emitting each result
from this worker's own thread as it lands (never from a pool thread).
Subclasses provide _plan() -> n_workers (stashing whatever per-run
context they need), _items(), _one(item) -> result, and _emit(result).
On stop(): queued items are dropped, in-flight ones are not waited for —
that is what keeps closing the window responsive on a large scan.
"""
finished = pyqtSignal()
error = pyqtSignal(str)
def run(self):
try:
pool = ThreadPoolExecutor(max_workers=max(1, self._plan()))
try:
futures = [pool.submit(self._one, it) for it in self._items()]
for fut in as_completed(futures):
if self._stop:
break
self._emit(fut.result())
finally:
pool.shutdown(wait=not self._stop, cancel_futures=True)
self.finished.emit()
except Exception as exc:
self.error.emit(str(exc))
class LoadWorker(QObject): class LoadWorker(QObject):
finished = pyqtSignal(object) # SrasFile | None finished = pyqtSignal(object) # SrasFile | None
error = pyqtSignal(str) error = pyqtSignal(str)
@@ -145,29 +117,29 @@ class ComputeWorker(CancellableWorker):
self.error.emit(str(exc)) self.error.emit(str(exc))
class DcPrecomputeWorker(_PooledWorker): class DcPrecomputeWorker(CancellableWorker):
"""Computes CH3/CH4 DC images for every angle in the background. """Computes CH3/CH4 DC images for every angle in the background.
DC images are cheap (a per-waveform mean, no FFT) compared to the DC images are cheap (a per-waveform mean, no FFT) compared to the
CH1/Velocity FFT, so precomputing them for the whole file right after load CH1/Velocity FFT, so precomputing them for the whole file right after load
makes switching angles instant while on a DC channel, and also means the makes switching angles instant while on a DC channel, and also means the
FFT masking step (which needs a DC4 image) rarely has to wait on anything. FFT masking step (which needs a DC4 image) rarely has to wait on anything.
Angles are computed on a thread pool — the work is a pure mean over the
waveform block, so it is I/O- and bandwidth-bound and embarrassingly
parallel. Results are emitted one at a time as they land (out of angle
order), and always from this worker's own thread: nothing emits a Qt
signal from a pool thread.
""" """
angle_done = pyqtSignal(int, np.ndarray, np.ndarray) # angle_idx, dc3_mv, dc4_mv angle_done = pyqtSignal(int, np.ndarray, np.ndarray) # angle_idx, dc3_mv, dc4_mv
finished = pyqtSignal()
error = pyqtSignal(str)
def __init__(self, sras: SrasFile): def __init__(self, sras: SrasFile):
super().__init__() super().__init__()
self._sras = sras self._sras = sras
self._angle_budget = 0
def _plan(self) -> int: def _one_angle(self, a: int) -> tuple[int, np.ndarray, np.ndarray]:
n_workers, self._angle_budget = compute.plan_angle_level(self._sras)
return n_workers
def _items(self):
return range(self._sras.n_angles)
def _one(self, a: int) -> tuple[int, np.ndarray, np.ndarray]:
# max_workers=1 *and* a budget share: this call is one of several # max_workers=1 *and* a budget share: this call is one of several
# concurrent angles, and both the thread count and the buffer size # concurrent angles, and both the thread count and the buffer size
# have to be divided (see compute.plan_angle_level). # have to be divided (see compute.plan_angle_level).
@@ -177,8 +149,26 @@ class DcPrecomputeWorker(_PooledWorker):
dc_image_mv(self._sras, a, CH3_IDX, **kw), dc_image_mv(self._sras, a, CH3_IDX, **kw),
dc_image_mv(self._sras, a, CH4_IDX, **kw)) dc_image_mv(self._sras, a, CH4_IDX, **kw))
def _emit(self, result): def run(self):
self.angle_done.emit(*result) try:
n = self._sras.n_angles
n_workers, self._angle_budget = compute.plan_angle_level(self._sras)
pool = ThreadPoolExecutor(max_workers=n_workers)
try:
futures = {pool.submit(self._one_angle, a): a for a in range(n)}
for fut in as_completed(futures):
if self._stop:
break
a, dc3, dc4 = fut.result()
self.angle_done.emit(a, dc3, dc4)
finally:
# cancel_futures drops the queued angles; should_stop lets the
# in-flight ones bail within a chunk. Not waiting here is what
# keeps closing the window responsive on a large scan.
pool.shutdown(wait=not self._stop, cancel_futures=True)
self.finished.emit()
except Exception as exc:
self.error.emit(str(exc))
class BatchCacheWorker(QObject): class BatchCacheWorker(QObject):
@@ -189,10 +179,7 @@ class BatchCacheWorker(QObject):
*mode* is ``"dc"`` (CH3/CH4 mean images) or ``"fft"`` (CH1 peak-frequency *mode* is ``"dc"`` (CH3/CH4 mean images) or ``"fft"`` (CH1 peak-frequency
images, unmasked — masking is applied at display time, same as v5's PREC images, unmasked — masking is applied at display time, same as v5's PREC
convention). FFT images are computed and stored at *pad_factor*, the convention).
viewer's own zero-padding setting: a stored image only serves a view
asking for the same padding, so caching at any other one produces a file
the viewer will never read back.
Files are processed one per subprocess: they are fully independent, each Files are processed one per subprocess: they are fully independent, each
opens its own memmap and writes only its own bytes, and only path strings opens its own memmap and writes only its own bytes, and only path strings
@@ -204,13 +191,11 @@ class BatchCacheWorker(QObject):
file_done = pyqtSignal(str, str) file_done = pyqtSignal(str, str)
finished = pyqtSignal() finished = pyqtSignal()
def __init__(self, paths: list[str], mode: str, apply_bg_sub: bool, def __init__(self, paths: list[str], mode: str, apply_bg_sub: bool):
pad_factor: int = 1):
super().__init__() super().__init__()
self._paths = paths self._paths = paths
self._mode = mode self._mode = mode
self._apply_bg_sub = apply_bg_sub self._apply_bg_sub = apply_bg_sub
self._pad_factor = pad_factor
def _report(self, path: str, err: str, done: int, total: int): def _report(self, path: str, err: str, done: int, total: int):
self.file_done.emit(path, err) self.file_done.emit(path, err)
@@ -235,8 +220,7 @@ class BatchCacheWorker(QObject):
with ProcessPoolExecutor(max_workers=n_procs) as executor: with ProcessPoolExecutor(max_workers=n_procs) as executor:
futures = { futures = {
executor.submit(cache_file, p, self._mode, self._apply_bg_sub, executor.submit(cache_file, p, self._mode, self._apply_bg_sub,
compute.get_fft_backend(), per_proc_workers, compute.get_fft_backend(), per_proc_workers): p
self._pad_factor): p
for p in paths for p in paths
} }
for fut in as_completed(futures): for fut in as_completed(futures):
@@ -259,9 +243,7 @@ class BatchCacheWorker(QObject):
for path in paths: for path in paths:
try: try:
err = cache_file(path, self._mode, self._apply_bg_sub, err = cache_file(path, self._mode, self._apply_bg_sub,
compute.get_fft_backend(), compute.get_fft_backend(), compute._MAX_WORKERS)
compute.default_max_workers(),
self._pad_factor)
except Exception as exc: except Exception as exc:
err = str(exc) err = str(exc)
done += 1 done += 1
@@ -300,10 +282,10 @@ class BatchCacheWorker(QObject):
class AngleAlignmentWorker(QObject): class AngleAlignmentWorker(QObject):
"""Computes the rigid (rotation + translation, never scale) alignment for """Computes rotation+translation alignment for every angle in *sras*,
every angle in *sras* against *ref_angle_idx*, by cross-correlating each referenced to *ref_angle_idx*, from each angle's binarized CH4 mask.
angle's CH4 image against the reference's. Both the rotation and the Rotation is analytic (from sras.angles_deg); only translation is found by
translation are found from image content — see compute_angle_alignment. phase correlation.
""" """
progress = pyqtSignal(int) # 0–100 progress = pyqtSignal(int) # 0–100
finished = pyqtSignal(object, str) # AlignmentResult|None, error ("" = success) finished = pyqtSignal(object, str) # AlignmentResult|None, error ("" = success)
@@ -324,7 +306,7 @@ class AngleAlignmentWorker(QObject):
self.finished.emit(None, str(exc)) self.finished.emit(None, str(exc))
class Ch4MaskWorker(_PooledWorker): class Ch4MaskWorker(QObject):
"""Fetches each requested angle's CH4 (Bias B) DC image in mV, for """Fetches each requested angle's CH4 (Bias B) DC image in mV, for
ManualAlignmentDialog's initial threshold-mask overlay. ManualAlignmentDialog's initial threshold-mask overlay.
@@ -338,70 +320,83 @@ class Ch4MaskWorker(_PooledWorker):
ManualAlignmentDialog._start_mask_prep). ManualAlignmentDialog._start_mask_prep).
""" """
angle_done = pyqtSignal(int, np.ndarray) # angle_idx, dc4_mv angle_done = pyqtSignal(int, np.ndarray) # angle_idx, dc4_mv
finished = pyqtSignal()
error = pyqtSignal(str)
def __init__(self, sras: SrasFile, angle_indices: list[int]): def __init__(self, sras: SrasFile, angle_indices: list[int]):
super().__init__() super().__init__()
self._sras = sras self._sras = sras
self._angles = angle_indices self._angles = angle_indices
self._budget = 0
def _plan(self) -> int: def run(self):
n_workers, self._budget = compute.plan_angle_level(self._sras) try:
return n_workers n_workers, budget = compute.plan_angle_level(self._sras)
pool = ThreadPoolExecutor(max_workers=n_workers)
def _items(self): try:
return self._angles futures = {
pool.submit(dc_image_mv, self._sras, a, CH4_IDX,
def _one(self, a: int) -> tuple[int, np.ndarray]: max_workers=1, budget=budget): a
return a, dc_image_mv(self._sras, a, CH4_IDX, for a in self._angles
max_workers=1, budget=self._budget) }
for fut in as_completed(futures):
def _emit(self, result): a = futures[fut]
self.angle_done.emit(*result) self.angle_done.emit(a, fut.result())
finally:
pool.shutdown(wait=True)
self.finished.emit()
except Exception as exc:
self.error.emit(str(exc))
class CrossCorrelateWorker(_PooledWorker): class CrossCorrelateWorker(QObject):
"""Rigid registration (rotation + translation, never scale) of each of """FFT phase-correlation translation for each of *angle_indices* against
*angle_indices* against *ref_angle_idx*, for ManualAlignmentDialog's Auto *ref_angle_idx*, for ManualAlignmentDialog's Auto Cross-Correlate button.
Cross-Correlate button.
Runs on a background thread — registering a real many-angle, Runs on a background thread — a real many-angle, high-resolution scan's
high-resolution scan takes long enough that doing it on the GUI thread correlation (even at its downsampled working resolution) can take long
would visibly freeze the dialog. Rotation is *searched*, not taken from the enough that doing all of them on the GUI thread would visibly freeze the
stage's reported angle: see compute.register_angle_to_reference, which dialog. Rotation is set to the same analytic scan-angle delta Auto
seeds from that angle but scores both of its signs and refines from there. De-rotate uses alongside the correlated shift, since a translation
dc4_mv is the dialog's own already-in-memory per-angle CH4 image — this search is only meaningful once both angles' content is already oriented
worker does no fetching of its own. the same way. dc4_mv/pivot_mm are the dialog's own already-in-memory
per-angle images/pivots — this worker does no fetching of its own.
""" """
# angle_idx, rotation_deg, shift_x_mm, shift_y_mm, score, source angle_done = pyqtSignal(int, float, float, float) # angle_idx, rotation_deg, shift_x_mm, shift_y_mm
angle_done = pyqtSignal(int, float, float, float, float, str) finished = pyqtSignal()
error = pyqtSignal(str)
def __init__(self, sras: SrasFile, ref_angle_idx: int, angle_indices: list[int], def __init__(self, sras: SrasFile, ref_angle_idx: int, angle_indices: list[int],
dc4_mv: dict[int, np.ndarray], *, dc4_mv: dict[int, np.ndarray], pivot_mm: dict[int, tuple[float, float]],
sources: tuple[str, ...], dc_threshold_mv: float, *, use_mask: bool, dc_threshold_mv: float, margin_frac: float):
search_deg: float):
super().__init__() super().__init__()
self._sras = sras self._sras = sras
self._ref = ref_angle_idx self._ref = ref_angle_idx
self._angles = angle_indices self._angles = angle_indices
self._dc4_mv = dc4_mv self._dc4_mv = dc4_mv
self._sources = sources self._pivot_mm = pivot_mm
self._use_mask = use_mask
self._threshold = dc_threshold_mv self._threshold = dc_threshold_mv
self._search_deg = search_deg self._margin = margin_frac
def _plan(self) -> int: def _one(self, a: int) -> tuple[int, float, float, float]:
return compute.registration_workers(self._sras) theta = compute._theta_deg(self._sras, a, self._ref)
dx, dy = compute.correlate_translation_mm(
self._sras, a, self._ref, self._dc4_mv, self._pivot_mm,
use_mask=self._use_mask, dc_threshold_mv=self._threshold,
margin_frac=self._margin)
return a, theta, dx, dy
def _items(self): def run(self):
return self._angles try:
n_workers, _budget = compute.plan_angle_level(self._sras)
def _one(self, a: int) -> tuple[int, compute.RigidFit]: pool = ThreadPoolExecutor(max_workers=max(1, n_workers))
return a, compute.register_angle_to_reference( try:
self._sras, a, self._ref, self._dc4_mv, futures = [pool.submit(self._one, a) for a in self._angles]
dc_threshold_mv=self._threshold, sources=self._sources, for fut in as_completed(futures):
search_deg=self._search_deg) a, theta, dx, dy = fut.result()
self.angle_done.emit(a, theta, dx, dy)
def _emit(self, result): finally:
a, fit = result pool.shutdown(wait=True)
self.angle_done.emit(a, fit.rotation_deg, fit.shift_mm[0], self.finished.emit()
fit.shift_mm[1], fit.score, fit.source) except Exception as exc:
self.error.emit(str(exc))
tests/conftest.py
-16
View File
@@ -1,16 +0,0 @@
"""Shared test setup: repo-root imports, the offscreen Qt platform, and
hermetic QSettings (tests must not read or write the user's real viewer
settings)."""
import os
import sys
import tempfile
from pathlib import Path
os.environ.setdefault("QT_QPA_PLATFORM", "offscreen")
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
from PyQt6.QtCore import QSettings # noqa: E402
QSettings.setPath(QSettings.Format.IniFormat, QSettings.Scope.UserScope,
tempfile.mkdtemp(prefix="sras_qsettings_"))
tests/test_alignment.py
-212
View File
@@ -1,212 +0,0 @@
"""Angle-alignment tests: does registration actually stack the scans?
Builds a synthetic scan in which one sample is imaged at several *known*
rotations and offsets (tools/make_test_sras.write_rotating) and checks that the
alignment path recovers them, that the shared canvas is angle 0's own pixel
grid extended, and that nothing in the result depends on any other angle's
stage coordinates.
No Qt — this exercises sras_compute directly. See tests/test_gui.py for the
dialog and Aligned-View plumbing.
"""
from types import SimpleNamespace
import numpy as np
import pytest
import sras_compute as compute
from sras_format import CH4_IDX, SrasFile, adc_to_mv
import tools.make_test_sras as gen
# Registration is limited by how far a feature moves per degree: with this
# sample's ~1 mm radius and a ~16 µm registration pitch, a quarter degree is
# already sub-pixel, so it is the floor of what any metric can resolve here.
_ROT_TOL_DEG = 0.5
_SHIFT_TOL_MM = 0.02
_STACK_IOU_MIN = 0.90
_THRESHOLD_MV = 80.0
def dc4_images(sras: SrasFile) -> dict[int, np.ndarray]:
return {a: adc_to_mv(compute.compute_dc_image(sras, a, CH4_IDX), *sras.cal(CH4_IDX))
for a in range(sras.n_angles)}
def mm_transform(sras: SrasFile, result, angle_idx: int) -> np.ndarray:
"""Recover the pure mm-space rotation from a canvas->raw affine.
matrix == D @ R^T @ A_out, where A_out and D only carry the canvas and
per-angle pixel pitches; undoing both must leave something orthonormal, or
the transform is smuggling in a scale or a shear.
"""
dx_a, dy_a = compute.pixel_pitch_mm(sras, angle_idx)
A_out = np.array([[0.0, result.canvas_dx_mm], [result.canvas_dy_mm, 0.0]])
D = np.array([[0.0, 1.0 / dy_a], [1.0 / dx_a, 0.0]])
return np.linalg.inv(D) @ result.per_angle[angle_idx].matrix @ np.linalg.inv(A_out)
@pytest.fixture(scope="module")
def rig(tmp_path_factory):
"""The rotating-sample scan plus everything computed from it once."""
tmpdir = tmp_path_factory.mktemp("sras_align")
path = tmpdir / "rotating.sras"
meta = gen.write_rotating(path, n_angles=5)
sras = SrasFile(str(path))
dc4 = dc4_images(sras)
fits = {a: compute.register_angle_to_reference(
sras, a, 0, dc4, dc_threshold_mv=_THRESHOLD_MV)
for a in range(sras.n_angles)}
result = compute.compute_angle_alignment(sras, 0, _THRESHOLD_MV)
return SimpleNamespace(path=path, sras=sras, truth=meta["truth"],
dc4=dc4, fits=fits, result=result)
def test_registration_recovers_truth(rig):
"""Per-angle rigid registration (rotation + translation, no scale)."""
for a, fit in rig.fits.items():
t_rot, t_shift = rig.truth[a]
rot_err = abs(fit.rotation_deg - t_rot)
shift_err = float(np.hypot(fit.shift_mm[0] - t_shift[0],
fit.shift_mm[1] - t_shift[1]))
assert rot_err <= _ROT_TOL_DEG, \
(f"angle {a}: got {fit.rotation_deg:.3f}°, truth {t_rot:.3f}° "
f"(err {rot_err:.3f}°)")
assert shift_err <= _SHIFT_TOL_MM, f"angle {a}: err {shift_err:.4f} mm"
assert rig.fits[0] == compute.RigidFit(0.0, (0.0, 0.0), 1.0, "reference"), \
"reference angle registers as exact identity"
def test_stage_angle_sign_is_not_trusted(rig):
# The stage's rotational sense relative to this module's math-positive
# convention is not knowable from the file, and the old code hardcoded a
# guess. Flipping every reported angle must therefore change nothing: the
# search scores both signs and the images decide.
flipped = SrasFile(str(rig.path))
flipped.angles_deg = -flipped.angles_deg
flipped_fits = {a: compute.register_angle_to_reference(
flipped, a, 0, rig.dc4, dc_threshold_mv=_THRESHOLD_MV)
for a in range(1, flipped.n_angles)}
mismatches = {a: (flipped_fits[a].rotation_deg, rig.fits[a].rotation_deg)
for a in flipped_fits if flipped_fits[a] != rig.fits[a]}
assert not mismatches, \
f"negating every reported stage angle changed fits: {mismatches}"
def test_stage_coordinates_are_not_consulted(rig):
# Move every non-reference angle's scan window somewhere else entirely.
# Only angle 0's coordinates may matter, so every fit must be untouched.
moved = SrasFile(str(rig.path))
for a in range(1, moved.n_angles):
moved.x_start_mm[a] += 13.5 * a
moved.y_pos_per_angle[a] = moved.y_pos_per_angle[a] - 9.25 * a
moved_dc4 = dc4_images(moved)
moved_fits = {a: compute.register_angle_to_reference(
moved, a, 0, moved_dc4, dc_threshold_mv=_THRESHOLD_MV)
for a in range(1, moved.n_angles)}
mismatches = {a: (round(moved_fits[a].rotation_deg, 4), rig.fits[a].rotation_deg)
for a in moved_fits if moved_fits[a] != rig.fits[a]}
assert not mismatches, \
f"relocating every other angle's scan window changed fits: {mismatches}"
def test_canvas_is_reference_grid_extended(rig):
sras, result = rig.sras, rig.result
t0 = result.per_angle[0]
assert np.allclose(t0.matrix, np.eye(2)), \
f"angle 0's transform has rotation/scale/shear: {t0.matrix}"
assert np.allclose(t0.offset, np.round(t0.offset)), \
f"angle 0 does not land on whole canvas pixels: {t0.offset}"
assert ((result.canvas_dx_mm, result.canvas_dy_mm)
== compute.pixel_pitch_mm(sras, 0)), \
"canvas pitch is angle 0's own pitch"
n_rows, n_cols = result.canvas_shape
x_axis = result.canvas_origin_mm[0] + np.arange(n_cols) * result.canvas_dx_mm
y_axis = result.canvas_origin_mm[1] + np.arange(n_rows) * result.canvas_dy_mm
row0, col0 = int(round(-t0.offset[0])), int(round(-t0.offset[1]))
a0_rows, a0_cols = sras.image_shape(0)
assert np.allclose(x_axis[col0:col0 + a0_cols], sras.x_axis_mm(0)), \
"canvas X axis reproduces angle 0's own X coordinates"
assert np.allclose(y_axis[row0:row0 + a0_rows], sras.y_positions_mm(0)), \
"canvas Y axis reproduces angle 0's own Y coordinates"
assert (n_rows >= max(int(sras.n_rows[a]) for a in range(sras.n_angles))
and n_cols >= max(int(sras.n_frames[a]) for a in range(sras.n_angles))), \
f"canvas does not cover every angle's footprint: {result.canvas_shape}"
def test_transforms_are_pure_rotations(rig):
"""No scaling anywhere in the per-angle transforms."""
for a in range(rig.sras.n_angles):
R = mm_transform(rig.sras, rig.result, a)
assert (np.allclose(R @ R.T, np.eye(2), atol=1e-9)
and abs(abs(np.linalg.det(R)) - 1.0) < 1e-9), \
f"angle {a}: det={np.linalg.det(R):.6f}"
def test_all_angles_stack(rig):
aligned = {a: compute.apply_alignment(rig.result, a, rig.dc4[a])
for a in range(rig.sras.n_angles)}
base = aligned[0] >= _THRESHOLD_MV
for a in range(1, rig.sras.n_angles):
other = aligned[a] >= _THRESHOLD_MV
iou = float((base & other).sum()) / max(1, int((base | other).sum()))
assert iou >= _STACK_IOU_MIN, f"angle {a}: IoU {iou:.4f}"
def test_downsampled_preview_lands_with_full_res(rig):
# ManualAlignmentDialog reprojects block-mean-downsampled masks, so the
# affine has to account for the factor. When it did not, every preview
# layer came out magnified by that factor and offset — the overlay showed a
# blown-up crop of each mask, which is not something you can align by eye.
sras, result = rig.sras, rig.result
pitch = (result.canvas_dx_mm, result.canvas_dy_mm)
a = sras.n_angles - 1
p = result.per_angle[a]
full_mask = (rig.dc4[a] >= _THRESHOLD_MV).astype(np.float32)
full = compute.reproject_mask(
sras, a, 0, full_mask, p.rotation_deg, p.shift_mm, pitch,
result.canvas_origin_mm, result.canvas_shape)
fy, fx = 4, 16
small = compute.reproject_mask(
sras, a, 0, compute.block_mean_2d(full_mask, fy, fx),
p.rotation_deg, p.shift_mm, (pitch[0] * fx, pitch[1] * fy),
result.canvas_origin_mm,
(result.canvas_shape[0] // fy, result.canvas_shape[1] // fx),
src_downsample=(fy, fx))
# Compare in mm, via each layer's own center of mass.
def com_mm(layer, px, py):
rows, cols = np.nonzero(layer > 0.5)
return np.array([cols.mean() * px, rows.mean() * py])
d = com_mm(small, pitch[0] * fx, pitch[1] * fy) - com_mm(full, *pitch)
assert (abs(d[0]) <= abs(pitch[0] * fx) and abs(d[1]) <= abs(pitch[1] * fy)), \
f"downsampled preview offset {d[0]:+.4f}, {d[1]:+.4f} mm"
def test_manual_path_reproduces_geometry(rig):
sras, result = rig.sras, rig.result
params = {a: compute.ManualAngleParams(t.rotation_deg, t.shift_mm)
for a, t in result.per_angle.items()}
manual = compute.build_manual_alignment(sras, 0, _THRESHOLD_MV, params)
assert (manual.canvas_shape == result.canvas_shape
and np.allclose(manual.canvas_origin_mm, result.canvas_origin_mm)
and all(np.allclose(manual.per_angle[a].matrix, result.per_angle[a].matrix)
and np.allclose(manual.per_angle[a].offset, result.per_angle[a].offset)
for a in range(sras.n_angles))), \
"build_manual_alignment matches compute_angle_alignment for the same params"
def test_sidecar_roundtrip(rig):
sras, result = rig.sras, rig.result
params = {a: compute.ManualAngleParams(t.rotation_deg, t.shift_mm)
for a, t in result.per_angle.items()}
compute.save_manual_alignment(sras, 0, _THRESHOLD_MV, params)
loaded = compute.load_manual_alignment(sras)
assert (loaded is not None
and all(np.isclose(loaded.per_angle[a].rotation_deg, params[a].rotation_deg)
and np.allclose(loaded.per_angle[a].shift_mm, params[a].shift_mm)
for a in range(sras.n_angles))), \
"sidecar reloads every angle's params"
assert compute.delete_manual_alignment(sras), "sidecar deletes cleanly"
tests/test_compute.py
-356
View File
@@ -1,356 +0,0 @@
"""Behavioural tests for the compute/format layer.
Covers what the golden-hash harness can't: the v6->v7 cache round-trip
(including block carry-forward), parallel-vs-serial identity, the no-mask
fast path, and the ROI bounding-box mask optimisation.
"""
import subprocess
import sys
from pathlib import Path
import numpy as np
import pytest
import sras_compute as compute
from sras_compute import (
cache_file, compute_dc_image, compute_rf_image, dc_image_mv,
)
from sras_format import CH3_IDX, CH4_IDX, SrasFile, adc_to_mv
import tools.make_test_sras as gen
REPO = Path(__file__).resolve().parent.parent
def test_cache_roundtrip(tmp_path):
"""v6 -> v7 for DC, then FFT, asserting the first block survives the
second write (the carry-forward path in write_v7_cache)."""
path = tmp_path / "roundtrip.sras"
gen.write(path, n_angles=3, seed=1, samples_per_frame=64)
src = SrasFile(str(path))
assert src.version == 6, f"got v{src.version}"
expect_dc3 = [dc_image_mv(src, a, CH3_IDX) for a in range(src.n_angles)]
expect_dc4 = [dc_image_mv(src, a, CH4_IDX) for a in range(src.n_angles)]
expect_fft = [compute_rf_image(src, a, dc_threshold_mv=None, apply_bg_sub=True)
for a in range(src.n_angles)]
err = cache_file(str(path), "dc", True)
assert err == "", err
after_dc = SrasFile(str(path))
assert after_dc.version == 7, f"got v{after_dc.version}"
assert all(x is not None for x in after_dc.precomputed_dc3_mv)
assert all(np.allclose(after_dc.precomputed_dc3_mv[a], expect_dc3[a], atol=1e-4)
for a in range(after_dc.n_angles))
assert all(np.allclose(after_dc.precomputed_dc4_mv[a], expect_dc4[a], atol=1e-4)
for a in range(after_dc.n_angles))
assert all(x is None for x in after_dc.precomputed_freq_mhz), "no fft block yet"
assert (after_dc.precomputed_dc3_mv[0].dtype == np.float32
and after_dc.precomputed_dc3_mv[0].dtype.byteorder in ("=", "|")), \
"cached images are native float32"
assert after_dc.precomputed_dc3_mv[0].flags.writeable
err = cache_file(str(path), "fft", True)
assert err == "", err
both = SrasFile(str(path))
assert all(x is not None for x in both.precomputed_freq_mhz)
assert all(np.allclose(both.precomputed_freq_mhz[a], expect_fft[a], atol=1e-3)
for a in range(both.n_angles))
assert all(np.allclose(both.precomputed_dc3_mv[a], expect_dc3[a], atol=1e-4)
for a in range(both.n_angles)), \
"DC block carried forward through the FFT write"
assert both.precomputed_bg_sub is True
# The fast path must reproduce a fresh compute, and masking must still
# apply on top of a cached (unmasked) image.
fresh = SrasFile(str(path))
fresh.precomputed_freq_mhz = [None] * fresh.n_angles
dc4 = dc_image_mv(both, 0, CH4_IDX)
thr = float(np.median(dc4))
assert np.allclose(
compute_rf_image(both, 0, dc_threshold_mv=None, apply_bg_sub=True),
compute_rf_image(fresh, 0, dc_threshold_mv=None, apply_bg_sub=True),
atol=1e-3), "cached fast path == fresh compute (unmasked)"
assert np.allclose(
compute_rf_image(both, 0, dc_threshold_mv=thr, apply_bg_sub=True),
compute_rf_image(fresh, 0, dc_threshold_mv=thr, apply_bg_sub=True),
atol=1e-3), "cached fast path == fresh compute (masked)"
# Waveform data must be byte-identical to the pre-cache file.
orig = tmp_path / "roundtrip_orig.sras"
gen.write(orig, n_angles=3, seed=1, samples_per_frame=64)
o, n = SrasFile(str(orig)), SrasFile(str(path))
assert all(np.array_equal(np.asarray(o.data[a]), np.asarray(n.data[a]))
for a in range(o.n_angles)), \
"waveform data untouched by the cache write"
def test_partial_v7_cache(tmp_path):
"""Only some angles cached: uncached angles must compute, not read zeros.
This is the v5 bug the ragged normalisation fixed, checked via v7."""
path = tmp_path / "partial.sras"
gen.write(path, n_angles=3, seed=2, samples_per_frame=64)
src = SrasFile(str(path))
expected = [compute_rf_image(src, a, dc_threshold_mv=None, apply_bg_sub=True)
for a in range(src.n_angles)]
partial = [expected[0], None, expected[2]] # angle 1 deliberately absent
src.write_v7_cache(new_freq_mhz=partial, new_bg_sub=True)
reread = SrasFile(str(path))
assert reread.precomputed_freq_mhz[1] is None
assert (reread.precomputed_freq_mhz[0] is not None
and reread.precomputed_freq_mhz[2] is not None)
img1 = compute_rf_image(reread, 1, dc_threshold_mv=None, apply_bg_sub=True)
assert np.any(img1 != 0) and np.allclose(img1, expected[1], atol=1e-3), \
"uncached angle computes rather than returning zeros"
def test_parallel_identity(tmp_path, monkeypatch):
"""Forcing 1 worker vs many must give identical output — catches
chunk-boundary and race bugs."""
path = tmp_path / "parallel.sras"
# Many rows, so the row loop actually splits into several chunks.
n_rows, n_frames, spf = 48, 9, 256
gen.write(path, n_angles=1, seed=3, samples_per_frame=spf,
geometry=[(n_rows, n_frames)])
sras = SrasFile(str(path))
# Shrink the budget so the outer row loop splits into many chunks, and
# the block size so every chunk splits into many FFT tasks — the worst
# case for boundary bugs.
monkeypatch.setattr(compute, "_TOTAL_BYTES_BUDGET", 8 * n_frames * spf * 4)
monkeypatch.setattr(compute, "_FFT_BLOCK", 4)
fft_rows = compute._plan_fft_rows(n_frames, spf, compute._TOTAL_BYTES_BUDGET)
assert fft_rows < n_rows, \
f"FFT work actually splits into multiple chunks ({fft_rows} of {n_rows})"
dc_rows = compute._chunk_rows_for(n_frames, spf, compute._TOTAL_BYTES_BUDGET)
assert dc_rows < n_rows, \
f"DC work actually splits into multiple chunks ({dc_rows} of {n_rows})"
monkeypatch.setattr(compute, "_MAX_WORKERS", 1)
dc_serial = compute_dc_image(sras, 0, CH4_IDX)
rf_serial = compute_rf_image(sras, 0, dc_threshold_mv=None, apply_bg_sub=True)
dc4 = adc_to_mv(dc_serial, *sras.cal(CH4_IDX))
thr = float(np.median(dc4))
rf_masked_serial = compute_rf_image(sras, 0, dc_threshold_mv=thr,
apply_bg_sub=True)
rf_pad_serial = compute_rf_image(sras, 0, dc_threshold_mv=thr,
apply_bg_sub=True, n_fft=spf * 8)
monkeypatch.setattr(compute, "_MAX_WORKERS", 8)
dc_par = compute_dc_image(sras, 0, CH4_IDX)
rf_par = compute_rf_image(sras, 0, dc_threshold_mv=None, apply_bg_sub=True)
rf_masked_par = compute_rf_image(sras, 0, dc_threshold_mv=thr, apply_bg_sub=True)
rf_pad_par = compute_rf_image(sras, 0, dc_threshold_mv=thr,
apply_bg_sub=True, n_fft=spf * 8)
assert np.array_equal(dc_serial, dc_par), "dc image identical"
assert np.array_equal(rf_serial, rf_par), "rf image identical (unmasked)"
assert np.array_equal(rf_masked_serial, rf_masked_par), \
"rf image identical (masked)"
assert np.array_equal(rf_pad_serial, rf_pad_par), \
"rf image identical (masked, padded/zoom)"
@pytest.mark.parametrize("spf,bps", [(64, 2), (37, 1)])
def test_zoom_identity(tmp_path, monkeypatch, spf, bps):
"""The zoom peak search must reproduce the full padded-rfft argmax
bit-for-bit, across pad factors, masking, bg-sub, dtype, and backend."""
path = tmp_path / f"zoom_{spf}.sras"
gen.write(path, n_angles=2, seed=6, samples_per_frame=spf, bps=bps)
sras = SrasFile(str(path))
dc4 = dc_image_mv(sras, 0, CH4_IDX)
thr = float(np.median(dc4))
backends = ["scipy"] + (["pyfftw"] if compute.PYFFTW_AVAILABLE else [])
for backend in backends:
monkeypatch.setattr(compute, "_fft_backend", backend)
for pad in (4, 8, 40):
n_fft = spf * pad
for thr_v in (None, thr):
for bg in (False, True):
ref = compute_rf_image(sras, 0, dc_threshold_mv=thr_v,
apply_bg_sub=bg, n_fft=n_fft,
exact=True)
zoom = compute_rf_image(sras, 0, dc_threshold_mv=thr_v,
apply_bg_sub=bg, n_fft=n_fft)
diff = int((ref != zoom).sum())
assert diff == 0, \
(f"{diff} px differ: backend={backend} pad={pad} "
f"thr={thr_v} bg={bg} spf={spf}")
# A threshold above every pixel masks everything: both paths must agree
# on an all-zero image.
all_masked = compute_rf_image(sras, 0, dc_threshold_mv=1e9, n_fft=spf * 8)
assert not all_masked.any()
def test_zoom_identity_fuzz():
"""Hammer _peak_bins_zoom directly with adversarial spectra: noise,
un-subtracted DC offsets, on-bin and off-bin tones, near-tie tone pairs,
and all-zero rows."""
import scipy.fft as scipy_fft
rng = np.random.default_rng(42)
for _ in range(25):
spf = int(rng.integers(16, 220))
pad = int(rng.choice([4, 5, 8, 16, 40]))
n_fft = spf * pad
n_wf = 24
w = rng.normal(scale=20.0, size=(n_wf, spf))
t = np.arange(spf)
# rows 0-5: pure/noisy tones (some off-bin), row 6-7: near-tie pair,
# row 8: big DC offset, row 9: all zeros, rest: plain noise.
for r in range(6):
f = rng.uniform(1.0, spf / 2 - 1)
w[r] = 60 * np.sin(2 * np.pi * f * t / spf) + w[r] * (r % 2)
f1, f2 = rng.uniform(2.0, spf / 2 - 2, size=2)
w[6] = 50 * np.sin(2 * np.pi * f1 * t / spf) \
+ 49.9 * np.sin(2 * np.pi * f2 * t / spf)
w[7] = 50 * np.sin(2 * np.pi * f1 * t / spf) \
+ 50 * np.cos(2 * np.pi * f2 * t / spf)
w[8] = 90 + rng.normal(scale=5.0, size=spf)
w[9] = 0.0
w = w.astype(np.float32)
S = scipy_fft.rfft(w, n=n_fft, axis=-1, workers=1)
P = S.real ** 2
P += S.imag ** 2
P[:, 0] = 0.0
ref = np.argmax(P, axis=1)
zp = compute._zoom_plan(spf, n_fft)
got = compute._peak_bins_zoom(w, zp)
bad = np.nonzero(ref != got)[0]
assert not len(bad), \
(f"spf={spf} pad={pad}: rows {bad.tolist()} picked "
f"{got[bad].tolist()} instead of {ref[bad].tolist()}")
def test_nomask_equals_low_threshold(tmp_path):
"""dc_threshold_mv=None must equal a threshold below every pixel, while
skipping the CH4 read."""
path = tmp_path / "nomask.sras"
gen.write(path, n_angles=2, seed=4, samples_per_frame=128)
sras = SrasFile(str(path))
for a in range(sras.n_angles):
none_img = compute_rf_image(sras, a, dc_threshold_mv=None, apply_bg_sub=True)
low_img = compute_rf_image(sras, a, dc_threshold_mv=-1e9, apply_bg_sub=True)
assert np.array_equal(none_img, low_img), \
f"angle {a}: None == -1e9 threshold"
assert len(np.unique(none_img)) > 1, \
f"angle {a}: image is degenerate ({len(np.unique(none_img))} unique)"
def test_roi_mask():
"""The bbox-restricted mask must equal a full-grid point-in-polygon test."""
from matplotlib.path import Path as MplPath
from sras_viewer import RoiQuad
rng = np.random.default_rng(0)
x = np.linspace(-2.0, 3.0, 137)
y = np.linspace(1.0, 4.0, 91)
cases = {
"axis-aligned rect": np.array([[0.0, 1.5], [1.0, 1.5], [1.0, 3.0], [0.0, 3.0]]),
"skewed quad": np.array([[-0.5, 1.2], [1.7, 1.9], [1.2, 3.4], [-1.0, 2.6]]),
"entirely outside": np.array([[8.0, 8.0], [9.0, 8.0], [9.0, 9.0], [8.0, 9.0]]),
"covers whole grid": np.array([[-9.0, -9.0], [9.0, -9.0], [9.0, 9.0], [-9.0, 9.0]]),
"straddles left edge": np.array([[-4.0, 2.0], [0.5, 2.0], [0.5, 3.0], [-4.0, 3.0]]),
}
for _ in range(5):
cases[f"random {_}"] = rng.uniform([-2.5, 0.5], [3.5, 4.5], size=(4, 2))
for name, pts in cases.items():
roi = RoiQuad(pts)
fast = roi.mask_for_grid(x, y)
X, Y = np.meshgrid(x.astype(np.float64), y.astype(np.float64))
slow = MplPath(pts).contains_points(
np.column_stack([X.ravel(), Y.ravel()])).reshape(X.shape)
assert np.array_equal(fast, slow), f"{name} ({int(slow.sum())} px inside)"
# Descending y axis (images are stored top-down in some scans).
roi = RoiQuad(cases["skewed quad"])
y_desc = y[::-1]
fast = roi.mask_for_grid(x, y_desc)
X, Y = np.meshgrid(x.astype(np.float64), y_desc.astype(np.float64))
slow = MplPath(cases["skewed quad"]).contains_points(
np.column_stack([X.ravel(), Y.ravel()])).reshape(X.shape)
assert np.array_equal(fast, slow), "descending y axis"
def test_legacy_parse(tmp_path):
"""v2-v4 parsing against known written data."""
for version in (2, 3, 4):
path = tmp_path / f"legacy_v{version}.sras"
meta = gen.write_legacy(path, version=version, n_angles=2, n_rows=4,
n_frames=12, samples_per_frame=32, seed=version)
s = SrasFile(str(path))
assert s.version == version, f"got v{s.version}"
assert list(s.n_rows) == [4, 4] and list(s.n_frames) == [12, 12], \
f"rows={list(s.n_rows)} frames={list(s.n_frames)}"
assert all(np.array_equal(np.asarray(s.data[a]), meta["data"][a])
for a in range(s.n_angles)), \
f"v{version} waveform data matches what was written"
assert (s.background is not None) == (version >= 4), \
f"v{version} background {'present' if version >= 4 else 'absent'}"
assert (isinstance(s.precomputed_freq_mhz, list)
and len(s.precomputed_freq_mhz) == s.n_angles), \
f"v{version} precomputed stores are ragged lists"
# DC image must equal a direct mean of the known input.
expect = meta["data"][0][:, CH3_IDX, :, :].astype(np.float64).mean(axis=-1)
assert np.allclose(compute_dc_image(s, 0, CH3_IDX), expect, atol=1e-3), \
f"v{version} DC image equals a direct mean"
def test_sras_average(tmp_path):
"""The sras_average.py CLI: frame averaging with remainder handling."""
src = tmp_path / "legacy_v4.sras"
meta = gen.write_legacy(src, version=4, n_angles=2, n_rows=4, n_frames=12,
samples_per_frame=32, seed=4)
dst = tmp_path / "legacy_v4_avg.sras"
proc = subprocess.run(
[sys.executable, str(REPO / "sras_average.py"), str(src), str(dst), "--n", "4"],
capture_output=True, text=True, cwd=REPO)
assert proc.returncode == 0, (proc.stderr or proc.stdout).strip()[-200:]
avg = SrasFile(str(dst))
assert avg.version == 4
assert list(avg.n_frames) == [3, 3], f"{list(avg.n_frames)}"
assert (avg.n_angles == 2 and list(avg.n_rows) == [4, 4]
and avg.n_channels == meta["n_channels"])
assert np.allclose(avg.ch_ymult_mv, SrasFile(str(src)).ch_ymult_mv), \
"calibration preserved"
assert np.array_equal(avg.background, SrasFile(str(src)).background), \
"background preserved"
src_data = meta["data"]
expect0 = src_data[0][:, :, 0:4, :].astype(np.float32).mean(axis=2).astype(np.int16)
assert np.array_equal(np.asarray(avg.data[0])[:, :, 0, :], expect0), \
"first averaged group equals the mean of its 4 source frames"
# Remainder handling: 12 frames / 5 -> 2 full groups + 1 partial.
dst2 = tmp_path / "legacy_v4_avg5.sras"
subprocess.run([sys.executable, str(REPO / "sras_average.py"),
str(src), str(dst2), "--n", "5"],
capture_output=True, text=True, cwd=REPO)
assert list(SrasFile(str(dst2)).n_frames) == [3, 3], \
"partial trailing group kept by default"
dst3 = tmp_path / "legacy_v4_avg5d.sras"
subprocess.run([sys.executable, str(REPO / "sras_average.py"),
str(src), str(dst3), "--n", "5", "--discard-remainder"],
capture_output=True, text=True, cwd=REPO)
assert list(SrasFile(str(dst3)).n_frames) == [2, 2], \
"--discard-remainder drops the partial group"
def test_unsupported_version_reported(tmp_path):
"""cache_file must report, not raise, for a file it can't handle."""
bogus = tmp_path / "bogus.sras"
bogus.write_bytes(b"SRAS" + bytes([99]) + b"\x00" * 200)
err = cache_file(str(bogus), "dc", True)
assert err, "bad version returns an error string"
missing = cache_file(str(tmp_path / "does_not_exist.sras"), "dc", True)
assert missing, "missing file returns an error string"
tests/test_gui.py
-546
View File
@@ -1,546 +0,0 @@
"""Headless GUI test: drives SrasViewerWindow through the real Qt widgets,
signals and worker threads under the offscreen platform plugin.
Covers the interactions a manual smoke test would: load, switch angles and
channels, background DC precompute, lazy FFT compute, threshold and bg-sub
changes, angle alignment, manual angle alignment, aligned view, ROI
draw/move, and CSV export.
NOTE: this module is one ordered integration sequence over a single shared
window — the tests build on each other's state and must run in definition
order (pytest's default within a module). Run the whole module, not single
tests.
"""
import json
from types import SimpleNamespace
from unittest.mock import patch
import numpy as np
import pytest
from PyQt6.QtCore import QEventLoop, Qt, QTimer
from PyQt6.QtTest import QTest
from PyQt6.QtWidgets import QApplication, QMessageBox
import sras_compute as compute
from sras_format import CH1_IDX, CH3_IDX, CH4_IDX, SrasFile
from sras_viewer import RoiQuad, SrasViewerWindow, VELOCITY_MODE_IDX
import tools.make_test_sras as gen
def pump(ms: int = 250):
"""Run the event loop for a while so queued signals and worker threads
make progress."""
loop = QEventLoop()
QTimer.singleShot(ms, loop.quit)
loop.exec()
def wait_until(pred, timeout_ms: int = 20000, step: int = 100) -> bool:
waited = 0
while waited < timeout_ms:
if pred():
return True
pump(step)
waited += step
return pred()
@pytest.fixture(scope="module")
def ctx(tmp_path_factory):
"""The shared window, test file, and cross-test state for the sequence."""
app = QApplication.instance() or QApplication([])
tmpdir = tmp_path_factory.mktemp("sras_gui")
path = tmpdir / "gui.sras"
gen.write(path, n_angles=4, seed=11, samples_per_frame=256)
win = SrasViewerWindow()
win.show()
errors: list[str] = []
# Capture anything the app reports as an error via the status bar.
win.statusBar().messageChanged.connect(
lambda m: errors.append(m) if m and "error" in m.lower() else None)
c = SimpleNamespace(app=app, win=win, path=path, tmpdir=tmpdir,
errors=errors, s=None)
yield c
if win.isVisible():
win.close()
pump(400)
def test_load(ctx):
win = ctx.win
win._load_file(str(ctx.path))
assert wait_until(lambda: win._sras is not None), "file loaded"
ctx.s = s = win._sras
assert s.version == 6, f"v{s.version}"
assert win.combo_channel.currentIndex() == CH4_IDX, "defaults to CH4"
assert win._current_image is not None, "image displayed"
assert win.spin_angle.maximum() == s.n_angles - 1, \
"angle spinbox ranges over all angles"
assert win._info["Angles"].text() == f"Angles: {s.n_angles}", \
win._info["Angles"].text()
def test_dc_precompute_all_angles(ctx):
win, s = ctx.win, ctx.s
ok = wait_until(lambda: all((a, CH4_IDX) in win._dc_cache
and (a, CH3_IDX) in win._dc_cache
for a in range(s.n_angles)))
assert ok, f"every angle cached for CH3 and CH4 ({len(win._dc_cache)} entries)"
assert "ready for all angles" in win.lbl_dc_precompute.text(), \
win.lbl_dc_precompute.text()
def test_angle_switching_from_cache(ctx):
"""Setting the angle must redraw on its own — no nudge from the test.
The spinbox used to be wired to editingFinished alone, so stepping it with
the arrows changed the number and left the plot on the previous angle until
the box happened to lose focus.
"""
win, s = ctx.win, ctx.s
for a in range(s.n_angles):
win.spin_angle.setValue(a)
pump(60)
assert win._current_angle == a, \
f"angle {a} redrawn from the spinbox alone, showing {win._current_angle}"
expected = win._sras.image_shape(a)
assert win._current_image.shape == expected, \
f"angle {a} shows its own geometry {expected}, got {win._current_image.shape}"
assert not win._job_running("compute"), \
"no compute job needed for cached DC angles"
def test_angle_steps_and_typing(ctx):
"""The arrows step and redraw one angle at a time; typing commits once,
without a stop at every intermediate value."""
win, s = ctx.win, ctx.s
assert s.n_angles > 2, "fixture needs several angles"
win.spin_angle.setValue(0)
pump(60)
seen = []
win.spin_angle.valueChanged.connect(seen.append)
try:
win.spin_angle.stepUp()
pump(60)
assert win._current_angle == 1, f"arrow stepped to 1, showing {win._current_angle}"
win.spin_angle.stepDown()
pump(60)
assert win._current_angle == 0, f"arrow stepped to 0, showing {win._current_angle}"
# Typing must not redraw per keystroke — that is what editingFinished
# bought and what keyboard tracking has to keep buying.
seen.clear()
last = s.n_angles - 1
win.spin_angle.lineEdit().selectAll()
QTest.keyClicks(win.spin_angle.lineEdit(), str(last))
pump(30)
assert seen == [] and win._current_angle == 0, \
f"half-typed text must not redraw ({seen}, showing {win._current_angle})"
QTest.keyClick(win.spin_angle, Qt.Key.Key_Return)
pump(60)
assert seen == [last], f"committed exactly once on Return: {seen}"
assert win._current_angle == last
finally:
win.spin_angle.valueChanged.disconnect(seen.append)
win.spin_angle.clearFocus()
win.spin_angle.setValue(0)
pump(60)
def test_channel_switching(ctx):
win = ctx.win
win.spin_angle.setValue(0)
win._on_view_changed()
pump(60)
win.combo_channel.setCurrentIndex(CH3_IDX)
assert wait_until(lambda: win._current_ch == CH3_IDX), "CH3 displayed"
win.combo_channel.setCurrentIndex(CH1_IDX)
assert wait_until(
lambda: win._current_ch == CH1_IDX and not win._job_running("compute")), \
"CH1 (FFT) computed"
assert len(win._fft_cache) > 0, "FFT result cached"
ctx.rf_img = win._current_image
assert len(np.unique(ctx.rf_img)) > 1, \
f"FFT image is degenerate ({len(np.unique(ctx.rf_img))} unique values)"
def test_velocity_mode(ctx):
"""Velocity mode is a pure post-multiply, no recompute."""
win = ctx.win
ctx.n_fft_before = len(win._fft_cache)
win.combo_channel.setCurrentIndex(VELOCITY_MODE_IDX)
assert wait_until(
lambda: win._current_ch == VELOCITY_MODE_IDX
and not win._job_running("compute")), "velocity displayed"
grating = win.spin_grating_um.value()
assert np.allclose(win._current_image, ctx.rf_img * grating, atol=1e-3), \
"velocity == freq x grating"
assert len(win._fft_cache) == ctx.n_fft_before, \
f"velocity reused the cached FFT ({ctx.n_fft_before} -> {len(win._fft_cache)})"
assert win.grp_velocity.isVisible(), "grating spinbox visible in velocity mode"
def test_threshold_change_recomputes(ctx):
"""A threshold change is a genuine cache-key change."""
win = ctx.win
win.combo_channel.setCurrentIndex(CH1_IDX)
wait_until(lambda: not win._job_running("compute"))
dc4 = win._dc_cache[(0, CH4_IDX)]
win.spin_threshold_mv.setValue(float(np.median(dc4)))
win._on_threshold_changed()
assert wait_until(
lambda: not win._job_running("compute")
and len(win._fft_cache) > ctx.n_fft_before), "recomputed at new threshold"
n_zero = int((win._current_image == 0).sum())
assert n_zero > 0, \
f"masking zeroed some pixels ({n_zero} of {win._current_image.size})"
def test_bg_sub_toggle(ctx):
win = ctx.win
n_before = len(win._fft_cache)
win.chk_bg_sub.setChecked(False)
assert wait_until(
lambda: not win._job_running("compute") and len(win._fft_cache) > n_before), \
"recomputed without bg-sub"
win.chk_bg_sub.setChecked(True)
pump(200)
assert not win._job_running("compute"), \
"returning to bg-sub was a cache hit (no recompute)"
def test_roi_and_csv_export(ctx):
win, s = ctx.win, ctx.s
x = s.x_axis_mm(0)
y = s.y_positions_mm(0)
roi = RoiQuad.from_bbox(float(x[1]), float(y[1]),
float(x[-2]), float(y[-2]))
win.image_canvas.set_roi(roi)
pump(120)
assert win.image_canvas.get_roi() is not None, "ROI registered"
assert ("pixels inside" in win.lbl_roi_npix.text()
and win.lbl_roi_npix.text() != "pixels inside: —"), \
win.lbl_roi_npix.text()
npix = int(win.lbl_roi_npix.text().split(":")[1])
assert 0 < npix <= win._current_image.size, f"{npix}"
assert win.btn_export_roi.isEnabled(), "Export ROI enabled"
csv_path = ctx.tmpdir / "roi.csv"
with patch("sras_viewer.main_window.QFileDialog.getSaveFileName",
return_value=(str(csv_path), "")):
win._on_export_roi_csv()
assert csv_path.exists(), "ROI CSV written"
body = [l for l in csv_path.read_text().splitlines() if not l.startswith("#")]
assert len(body) == npix + 1, \
f"ROI CSV has {len(body)} lines for {npix} pixels (want header + one per pixel)"
img_csv = ctx.tmpdir / "img.csv"
with patch("sras_viewer.main_window.QFileDialog.getSaveFileName",
return_value=(str(img_csv), "")):
win._on_export_csv()
assert img_csv.exists(), "image CSV written"
arr = np.loadtxt(img_csv, delimiter=",")
assert (arr.shape == win._current_image.shape
and np.allclose(arr, win._current_image, rtol=1e-5, atol=1e-4)), \
"image CSV round-trips the displayed image"
def test_roi_survives_switches(ctx):
win = ctx.win
win.spin_angle.setValue(1)
win._on_view_changed()
wait_until(lambda: not win._job_running("compute"))
assert win.image_canvas.get_roi() is not None, \
"ROI still present after angle switch"
win.combo_channel.setCurrentIndex(CH4_IDX)
wait_until(lambda: win._current_ch == CH4_IDX)
assert win.image_canvas.get_roi() is not None, \
"ROI still present after channel switch"
def test_angle_alignment(ctx):
win, s = ctx.win, ctx.s
win.spin_angle.setValue(0)
win._on_view_changed()
wait_until(lambda: not win._job_running("compute"))
assert win._alignment_act.isEnabled(), "alignment action enabled"
win._on_angle_alignment()
assert wait_until(
lambda: win._alignment_result is not None and not win._job_running("align"),
timeout_ms=60000), "alignment completed"
r = win._alignment_result
assert len(r.per_angle) == s.n_angles, "transform for every angle"
assert all(r.canvas_shape[0] >= int(s.n_rows[a])
and r.canvas_shape[1] >= int(s.n_frames[a])
for a in range(s.n_angles)), \
f"canvas is at least as large as any single angle: {r.canvas_shape}"
assert r.per_angle[r.ref_angle_idx].shift_mm == (0.0, 0.0), \
"reference angle has zero shift"
assert win.chk_aligned_view.isEnabled() and win.chk_aligned_view.isChecked(), \
"Aligned View auto-enabled and checked"
pump(200)
assert win.image_canvas._img_shape == r.canvas_shape, \
f"{win.image_canvas._img_shape} vs {r.canvas_shape}"
win.chk_aligned_view.setChecked(False)
pump(200)
assert win.image_canvas._img_shape == s.image_shape(0), \
f"unchecking returns to the raw per-angle grid: {win.image_canvas._img_shape}"
def test_manual_alignment_geometry(ctx):
"""Local mm is anchored on each angle's array center, not its stage
position: that is what makes a scan's placement independent of where its
window happened to sit. (Registration accuracy itself is covered by
tests/test_alignment.py, which has a synthetic sample to register.)"""
win, s = ctx.win, ctx.s
assert win._manual_align_act.isEnabled(), "manual alignment action enabled"
n_rows, n_frames = s.image_shape(0)
assert np.allclose(compute._center_idx(s, 0),
[(n_rows - 1) / 2, (n_frames - 1) / 2]), \
"array center is the geometric center of the pixel grid"
dx0, dy0 = compute.pixel_pitch_mm(s, 0)
assert np.allclose(compute._local_half_extent_mm(s, 0),
[(n_frames - 1) / 2 * abs(dx0), (n_rows - 1) / 2 * abs(dy0)]), \
"local half-extent is derived from shape and pitch alone"
identity = {a: compute.ManualAngleParams() for a in range(s.n_angles)}
origin_a, shape_a = compute.canvas_for_params(s, 0, (dx0, dy0), identity)
moved = SrasFile(str(ctx.path))
for a in range(1, moved.n_angles):
moved.x_start_mm[a] += 7.5
moved.y_pos_per_angle[a] = moved.y_pos_per_angle[a] + 3.25
origin_b, shape_b = compute.canvas_for_params(moved, 0, (dx0, dy0), identity)
assert shape_a == shape_b and np.allclose(origin_a, origin_b), \
("moving every non-reference angle's scan window must leave the canvas "
f"unchanged: {origin_a} {shape_a} vs {origin_b} {shape_b}")
# Both signs of the stage's reported angle are searched.
cands = compute._rotation_candidates(30.0, 6.0, 2.0)
assert min(cands) < -29.0 and max(cands) > 29.0, f"{min(cands)}..{max(cands)}"
# Whole-pixel translation must not wrap content around the edge.
arr = np.zeros((6, 6), dtype=np.float32)
arr[0, 0] = 1.0
assert compute._shift_into(arr, -1, -1).sum() == 0.0, \
"_shift_into zero-fills rather than wrapping"
assert compute._shift_into(arr, 2, 3)[2, 3] == 1.0, \
"_shift_into moves content by exactly the requested offset"
def test_manual_dialog_opens_at_identity(ctx):
"""Open must NOT seed from the still-live automatic AlignmentResult.
Manual mode exists to fix up whatever the automatic registration got
wrong, so it must start from identity (every angle centered on the
reference, no rotation) regardless of whatever the automatic run last
computed. Only a previously *saved manual* alignment (sidecar) should
ever seed this dialog."""
win, s = ctx.win, ctx.s
win._on_manual_alignment()
assert win._manual_align_dialog is not None, "dialog opened"
ctx.dlg = dlg = win._manual_align_dialog
assert not win._job_running("manual_align_masks"), \
"mask prep needed no background worker (already DC-cached)"
assert all(dlg._angle_params[a] == compute.ManualAngleParams()
for a in range(s.n_angles)), \
"no manual sidecar yet -> dialog starts at identity, not the automatic result"
def test_reference_angle_is_locked(ctx):
dlg = ctx.dlg
dlg.combo_active_angle.setCurrentIndex(dlg._ref_angle_idx)
pump(30)
before_ref = dlg._angle_params[dlg._ref_angle_idx]
dlg._on_nudge_translate(1, 0, False)
dlg._on_nudge_rotate(1, False)
assert not dlg.grp_manual_adjust.isEnabled(), "reference angle group disabled"
assert dlg._angle_params[dlg._ref_angle_idx] == before_ref, \
"reference angle untouched by nudge attempts"
def test_nudges(ctx):
"""Nudging a real angle (fine + coarse, translate + rotate)."""
dlg, s = ctx.dlg, ctx.s
ctx.active = active = 1 if s.n_angles > 1 else 0
dlg.combo_active_angle.setCurrentIndex(active)
pump(30)
before = dlg._angle_params[active].shift_mm
dlg._on_nudge_translate(1, 0, False) # fine +X
fine_step = dlg.spin_step_translate_mm.value()
assert abs(dlg._angle_params[active].shift_mm[0] - (before[0] + fine_step)) < 1e-9, \
"fine translate nudge moved shift_x by exactly one fine step"
before = dlg._angle_params[active].shift_mm
dlg._on_nudge_translate(0, -1, True) # coarse -Y
coarse_step = fine_step * dlg.spin_step_multiplier.value()
assert abs(dlg._angle_params[active].shift_mm[1] - (before[1] - coarse_step)) < 1e-9, \
"coarse translate nudge uses the multiplier"
before_rot = dlg._angle_params[active].rotation_deg
dlg._on_nudge_rotate(1, False)
assert dlg._angle_params[active].rotation_deg != before_rot, \
"rotate nudge changed rotation_deg"
assert len(dlg._preview_layers) == s.n_angles, \
"preview canvas rebuilt for every angle after a rotation nudge"
# Real key-event wiring (proves keyPressEvent -> signal -> slot).
before = dlg._angle_params[active].shift_mm
QTest.keyClick(dlg.canvas, Qt.Key.Key_Right)
assert dlg._angle_params[active].shift_mm[0] > before[0], \
"a real Right-arrow key event nudged shift_x"
def test_auto_derotate(ctx):
"""Auto De-rotate: seeds rotation from the stage angle, no translation."""
dlg, s, active = ctx.dlg, ctx.s, ctx.active
shift_before_derotate = dlg._angle_params[active].shift_mm
dlg._on_auto_derotate()
nominal = compute.nominal_delta_deg(s, active, dlg._ref_angle_idx)
assert abs(dlg._angle_params[active].rotation_deg - nominal) < 1e-6, \
"auto de-rotate seeded rotation from the stage's reported angle"
assert dlg._angle_params[active].shift_mm == shift_before_derotate, \
"auto de-rotate left translation untouched"
assert dlg._angle_params[dlg._ref_angle_idx].rotation_deg == 0.0, \
"reference angle stays identity after auto de-rotate"
# Clicking again offers the other sign, since which one lines the scans up
# is not knowable from the file.
dlg._on_auto_derotate()
assert abs(dlg._angle_params[active].rotation_deg + nominal) < 1e-6, \
"auto de-rotate offers the opposite sign on a second click"
def test_auto_cross_correlate(ctx):
"""Auto Cross-Correlate: searches rotation *and* translation."""
win, dlg, s = ctx.win, ctx.dlg, ctx.s
assert dlg.btn_auto_correlate.isEnabled(), \
"cross-correlate action enabled once masks are ready"
for label_idx, (label, _sources) in enumerate(dlg._CORRELATE_SOURCES):
dlg.combo_correlate_source.setCurrentIndex(label_idx)
dlg._on_auto_correlate()
assert wait_until(
lambda: not win._job_running("manual_align_correlate"),
timeout_ms=60000), f"auto cross-correlate completed ({label})"
assert all(a in dlg._fit_notes for a in range(s.n_angles)
if a != dlg._ref_angle_idx), \
f"every non-reference angle got a fit ({label})"
assert dlg._angle_params[dlg._ref_angle_idx] == compute.ManualAngleParams(), \
"auto cross-correlate reference angle stays identity"
assert dlg.grp_correlate.isEnabled() and dlg.btn_save.isEnabled(), \
"auto cross-correlate re-enabled controls when done"
assert len(dlg._preview_layers) == s.n_angles, \
"preview canvas rebuilt after cross-correlate"
assert dlg._fit_report(), "fit quality is reported per angle"
def test_save_sidecar(ctx):
win, dlg, s, active = ctx.win, ctx.dlg, ctx.s, ctx.active
dlg._on_save()
sidecar = compute.sidecar_path(s.path)
assert sidecar.exists(), "sidecar file written"
ctx.sidecar = sidecar
ctx.sidecar_raw = raw = json.loads(sidecar.read_text())
assert raw.get("schema_version") == compute._SIDECAR_SCHEMA_VERSION, \
"sidecar schema_version is current"
assert all(raw.get("per_angle", {}).get(str(a), {}).get("rotation_deg")
== dlg._angle_params[a].rotation_deg for a in range(s.n_angles)), \
"sidecar per_angle round-trips the dialog's resolved params"
assert (win._alignment_result is not None
and win._alignment_result.per_angle[active].rotation_deg
== dlg._angle_params[active].rotation_deg), \
"main window's alignment_result replaced by the manual build"
assert win.chk_aligned_view.isEnabled() and win.chk_aligned_view.isChecked(), \
"Aligned View auto-enabled after Save"
def test_stale_schema_sidecar_ignored(ctx):
"""An old-schema sidecar (pre-pivot/sign fix) is treated as absent."""
s, raw, sidecar = ctx.s, ctx.sidecar_raw, ctx.sidecar
stale = dict(raw)
stale["schema_version"] = compute._SIDECAR_SCHEMA_VERSION - 1
sidecar.write_text(json.dumps(stale))
assert compute.load_manual_alignment(s) is None, \
"a sidecar with an old schema_version is not loaded"
sidecar.write_text(json.dumps(raw)) # restore for the rest of the sequence
def test_clear_with_confirmation(ctx):
win, dlg, s = ctx.win, ctx.dlg, ctx.s
with patch("sras_viewer.dialogs.QMessageBox.question",
return_value=QMessageBox.StandardButton.Yes):
dlg._on_clear()
assert not ctx.sidecar.exists(), "sidecar file deleted"
assert all(dlg._angle_params[a] == compute.ManualAngleParams()
for a in range(s.n_angles)), "dialog params reset to identity"
assert win._alignment_result is None, "main window alignment_result cleared"
assert (not win.chk_aligned_view.isEnabled()
and not win.chk_aligned_view.isChecked()), \
"Aligned View disabled after Clear"
dlg.close()
pump(150)
assert win._manual_align_dialog is None, "dialog reference released on close"
def test_sidecar_restored_on_reload(ctx):
win, active = ctx.win, ctx.active
win._on_manual_alignment()
dlg = win._manual_align_dialog
dlg.combo_active_angle.setCurrentIndex(active)
pump(30)
dlg._on_auto_derotate()
dlg._on_nudge_translate(1, 1, True)
saved_rotation = dlg._angle_params[active].rotation_deg
saved_shift = dlg._angle_params[active].shift_mm
dlg._on_save()
dlg.close()
pump(150)
old_sras_id = id(win._sras)
win._load_file(str(ctx.path)) # reload the same file fresh
assert wait_until(
lambda: win._sras is not None and id(win._sras) != old_sras_id), \
"file reloaded"
ctx.s = win._sras
assert win._manual_align_dialog is None, \
"manual dialog force-closed by a reload"
assert win._alignment_result is not None, \
"reload restores the saved manual alignment automatically"
assert abs(win._alignment_result.per_angle[active].rotation_deg
- saved_rotation) < 1e-9, "restored rotation matches what was saved"
assert win._alignment_result.per_angle[active].shift_mm == saved_shift, \
"restored shift matches what was saved"
assert win.chk_aligned_view.isChecked(), \
"Aligned View auto-checked after restoring a saved alignment"
def test_pixel_inspector(ctx):
win = ctx.win
win.chk_aligned_view.setChecked(False)
pump(100)
win._on_pixel_clicked(0, 0)
pump(150)
assert win.lbl_wave_hint.isHidden(), "waveform hint hidden after a click"
win.combo_channel.setCurrentIndex(CH1_IDX)
wait_until(lambda: not win._job_running("compute"))
win._on_pixel_clicked(1, 1)
pump(150)
assert len(win.wave_canvas.ax_wave.lines) > 0, \
f"RF waveform panel rendered ({len(win.wave_canvas.ax_wave.lines)} lines)"
def test_shutdown(ctx):
win = ctx.win
win.close()
pump(400)
assert len(win._jobs) == 0, f"all background jobs released: {list(win._jobs)}"
def test_no_status_bar_errors(ctx):
unexpected = [e for e in ctx.errors if e]
assert not unexpected, f"status-bar errors seen: {unexpected}"
tools/bench_fft.py
-106
View File
@@ -1,106 +0,0 @@
#!/usr/bin/env python3
"""Benchmark the FFT peak-search path: exact vs zoom, serial vs pooled.
Reports wall time, waveforms/s, CPU utilization (utime+stime over wall, in
cores), and verifies every variant against the exact reference image.
Usage:
python tools/bench_fft.py # synthetic, pads 1/8/40
python tools/bench_fft.py --pads 40 --spf 2500 --rows 8 --frames 1024
python tools/bench_fft.py --real /path/big.sras --real-rows 32 --pads 40
"""
import argparse
import resource
import sys
import tempfile
import time
from pathlib import Path
import numpy as np
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
import sras_compute as compute # noqa: E402
from sras_compute import compute_rf_image, set_fft_backend # noqa: E402
from sras_format import SrasFile # noqa: E402
import tools.make_test_sras as gen # noqa: E402
from tools.check_equivalence import row_slice # noqa: E402
def _timed(fn):
r0 = resource.getrusage(resource.RUSAGE_SELF)
t0 = time.perf_counter()
out = fn()
wall = time.perf_counter() - t0
r1 = resource.getrusage(resource.RUSAGE_SELF)
cpu = (r1.ru_utime - r0.ru_utime) + (r1.ru_stime - r0.ru_stime)
return out, wall, cpu / max(wall, 1e-9)
def bench(sras, pads, backends):
n_wf = sum(int(sras.n_rows[a]) * int(sras.n_frames[a])
for a in range(sras.n_angles))
spf = sras.samples_per_frame
print(f"{n_wf} waveforms x {spf} samples, {sras.n_angles} angle(s)")
print(f"{'pad':>4} {'backend':>8} {'variant':>16} {'wall':>9} "
f"{'wf/s':>10} {'util':>6} match")
for pad in pads:
n_fft = spf * pad if pad > 1 else None
for backend in backends:
set_fft_backend(backend)
def run(**kw):
imgs = [compute_rf_image(sras, a, dc_threshold_mv=None,
apply_bg_sub=True, n_fft=n_fft, **kw)
for a in range(sras.n_angles)]
return np.concatenate([i.ravel() for i in imgs])
ref, wall, util = _timed(lambda: run(exact=True))
rows = [("exact(serial)", ref, wall, util, True)]
for label, kw in (("zoom(serial)", dict(max_workers=1)),
("zoom(pool)", {})):
img, wall, util = _timed(lambda: run(**kw))
rows.append((label, img, wall, util, bool(np.array_equal(img, ref))))
for label, img, wall, util, ok in rows:
print(f"{pad:>4} {backend:>8} {label:>16} {wall:>8.2f}s "
f"{n_wf / wall:>10.0f} {util:>5.1f}x "
f"{'OK' if ok else 'MISMATCH'}")
def main():
p = argparse.ArgumentParser(description=__doc__)
p.add_argument("--pads", default="1,8,40",
help="comma-separated pad factors (default 1,8,40)")
p.add_argument("--spf", type=int, default=2500)
p.add_argument("--rows", type=int, default=8)
p.add_argument("--frames", type=int, default=1024)
p.add_argument("--backends", default=None,
help="comma-separated (default: scipy,pyfftw if available)")
p.add_argument("--real", help="path to a real .sras file")
p.add_argument("--real-rows", type=int, default=32,
help="rows of angle 0 to use from the real file")
args = p.parse_args()
pads = [int(x) for x in args.pads.split(",")]
if args.backends:
backends = args.backends.split(",")
else:
backends = ["scipy"] + (["pyfftw"] if compute.PYFFTW_AVAILABLE else [])
if args.real:
sras = row_slice(SrasFile(args.real), 0, args.real_rows)
sras.data = [sras.data[0]]
sras.n_angles = 1
bench(sras, pads, backends)
else:
with tempfile.TemporaryDirectory(prefix="sras_bench_") as tmp:
path = Path(tmp) / "bench.sras"
gen.write(path, n_angles=1, seed=0, samples_per_frame=args.spf,
geometry=[(args.rows, args.frames)])
bench(SrasFile(str(path)), pads, backends)
if __name__ == "__main__":
main()
tools/check_equivalence.py
+28 -25
View File
@@ -1,9 +1,13 @@
#!/usr/bin/env python3 #!/usr/bin/env python3
"""Golden-output equivalence harness for compute-path refactors. """Golden-output equivalence harness for the sras-viewer refactor.
Computes a battery of DC / FFT / alignment outputs and prints a stable hash Computes a battery of DC / FFT / alignment outputs and prints a stable hash
for each. Run it before a refactor to capture a baseline, then again after for each. Run it on the pre-refactor commit to capture a baseline, then again
and diff the two reports — every line must match. after the refactor and diff the two reports — every line must match.
Imports work against both the pre-refactor monolith (`sras_viewer`) and the
post-refactor split (`sras_format` + `sras_compute`), so the *same* script
produces both sides of the comparison.
Hashes canonicalise to native little-endian float64 before hashing, so a Hashes canonicalise to native little-endian float64 before hashing, so a
deliberate dtype/byte-order change that preserves values does not show up as deliberate dtype/byte-order change that preserves values does not show up as
@@ -25,10 +29,22 @@ import numpy as np
sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
from sras_format import SrasFile, CH3_IDX, CH4_IDX, adc_to_mv # noqa: E402 # --- Import shim: split modules if present, else the monolith --------------
from sras_compute import ( # noqa: E402 try:
apply_alignment, compute_angle_alignment, compute_dc_image, compute_rf_image, from sras_format import SrasFile, CH1_IDX, CH3_IDX, CH4_IDX, adc_to_mv
) import sras_compute as C
_LAYOUT = "split"
except ImportError:
import sras_viewer as _V
from sras_viewer import SrasFile, CH1_IDX, CH3_IDX, CH4_IDX, adc_to_mv
C = _V
_LAYOUT = "monolith"
compute_dc_image = C.compute_dc_image
compute_rf_image = C.compute_rf_image
compute_alignment = C._compute_angle_alignment
apply_alignment = C.apply_alignment
import tools.make_test_sras as gen # noqa: E402 import tools.make_test_sras as gen # noqa: E402
@@ -47,8 +63,8 @@ def row_slice(sras: SrasFile, angle_idx: int, n_rows: int) -> SrasFile:
view.n_rows[angle_idx] = n view.n_rows[angle_idx] = n
view.data = list(sras.data) view.data = list(sras.data)
view.data[angle_idx] = sras.data[angle_idx][:n] view.data[angle_idx] = sras.data[angle_idx][:n]
view.y_pos_per_angle = list(sras.y_pos_per_angle) view._y_pos_per_angle = list(sras._y_pos_per_angle)
view.y_pos_per_angle[angle_idx] = sras.y_pos_per_angle[angle_idx][:n] view._y_pos_per_angle[angle_idx] = sras._y_pos_per_angle[angle_idx][:n]
return view return view
@@ -93,7 +109,7 @@ def check_file(path: Path, lines: list[str], tag: str,
for bg in (False, True): for bg in (False, True):
if bg and s.background is None: if bg and s.background is None:
continue continue
for pad in (1, 2, 4, 8, 40): for pad in (1, 2):
n_fft = s.samples_per_frame * pad if pad > 1 else None n_fft = s.samples_per_frame * pad if pad > 1 else None
for ti, thr in enumerate(thresholds): for ti, thr in enumerate(thresholds):
img = compute_rf_image(s, a, dc_threshold_mv=thr, img = compute_rf_image(s, a, dc_threshold_mv=thr,
@@ -119,7 +135,7 @@ def check_alignment(path: Path, lines: list[str], tag: str):
sras.ch_ymult_mv[CH4_IDX], sras.ch_yoff_adc[CH4_IDX], sras.ch_ymult_mv[CH4_IDX], sras.ch_yoff_adc[CH4_IDX],
sras.ch_yzero_mv[CH4_IDX]) sras.ch_yzero_mv[CH4_IDX])
thr = float(np.median(dc4)) thr = float(np.median(dc4))
res = compute_angle_alignment(sras, 0, thr) res = compute_alignment(sras, 0, thr)
report(lines, f"[{tag}] align canvas_shape", str(res.canvas_shape)) report(lines, f"[{tag}] align canvas_shape", str(res.canvas_shape))
report(lines, f"[{tag}] align canvas_origin", report(lines, f"[{tag}] align canvas_origin",
f"{res.canvas_origin_mm[0]:.9g},{res.canvas_origin_mm[1]:.9g}") f"{res.canvas_origin_mm[0]:.9g},{res.canvas_origin_mm[1]:.9g}")
@@ -149,7 +165,7 @@ def main():
help="directory for generated synthetic files") help="directory for generated synthetic files")
args = p.parse_args() args = p.parse_args()
lines = [f"# numpy: {np.__version__}"] lines = [f"# layout: {_LAYOUT}", f"# numpy: {np.__version__}"]
scratch = Path(args.scratch) scratch = Path(args.scratch)
synth = scratch / "equiv_synth.sras" synth = scratch / "equiv_synth.sras"
@@ -163,19 +179,6 @@ def main():
gen.write(synth_odd, n_angles=2, seed=7, samples_per_frame=37) gen.write(synth_odd, n_angles=2, seed=7, samples_per_frame=37)
check_file(synth_odd, lines, "odd", angles=[0, 1], n_rows=None) check_file(synth_odd, lines, "odd", angles=[0, 1], n_rows=None)
# A legacy v4 file exercises the uniform-geometry legacy layout through
# the same DC/FFT battery.
synth_v4 = scratch / "equiv_synth_v4.sras"
gen.write_legacy(synth_v4, version=4, n_angles=2, n_rows=6,
n_frames=14, samples_per_frame=48, seed=5)
check_file(synth_v4, lines, "v4", angles=[0, 1], n_rows=None)
# A big-endian int16 v6 file (real acquisitions are >i2; the other
# synthetics are int8).
synth_i16 = scratch / "equiv_synth_i16.sras"
gen.write(synth_i16, n_angles=2, seed=9, samples_per_frame=64, bps=2)
check_file(synth_i16, lines, "int16", angles=[0, 1], n_rows=None)
if args.real: if args.real:
real = Path(args.real) real = Path(args.real)
if real.exists(): if real.exists():
tools/make_test_sras.py
+8 -143
View File
@@ -11,19 +11,12 @@ Usage:
import argparse import argparse
import struct import struct
import sys
from pathlib import Path from pathlib import Path
import numpy as np import numpy as np
sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) HDR_FMT_V6 = ">4sBHfffffffIdBB"
GEO_FMT_V6 = ">ffIH"
# Single source of truth for the byte layout: the reader's own constants.
# The byte *assembly* below stays independent, so a writer bug can't be
# masked by a matching reader bug.
from sras_format import HDR_FMT as HDR_FMT_LEGACY # noqa: E402
from sras_format import GEO_FMT_V6, HDR_FMT_V6 # noqa: E402
from sras_compute import _rotation_matrix as _rot # noqa: E402
# Per-angle (n_rows, n_frames) — deliberately different per angle so ragged # Per-angle (n_rows, n_frames) — deliberately different per angle so ragged
# geometry handling is actually exercised. # geometry handling is actually exercised.
@@ -49,12 +42,12 @@ def _preamble(ymult_v: float, yoff_adc: float, yzero_v: float) -> bytes:
def build(n_angles: int, seed: int, samples_per_frame: int, def build(n_angles: int, seed: int, samples_per_frame: int,
geometry: list[tuple[int, int]] | None = None, geometry: list[tuple[int, int]] | None = None) -> tuple[bytes, dict]:
bps: int = 1) -> tuple[bytes, dict]:
rng = np.random.default_rng(seed) rng = np.random.default_rng(seed)
src_geom = geometry or _GEOMETRY src_geom = geometry or _GEOMETRY
geom = [src_geom[a % len(src_geom)] for a in range(n_angles)] geom = [src_geom[a % len(src_geom)] for a in range(n_angles)]
n_ch = 3 n_ch = 3
bps = 1
angles_deg = np.linspace(0.0, 60.0, n_angles, dtype=np.float32) angles_deg = np.linspace(0.0, 60.0, n_angles, dtype=np.float32)
# Distinct calibration per channel so a swapped-channel bug is visible. # Distinct calibration per channel so a swapped-channel bug is visible.
@@ -107,9 +100,7 @@ def build(n_angles: int, seed: int, samples_per_frame: int,
block[r, 1, f] = np.int8((a * 7 + r * 3 + f) % 100 - 50) block[r, 1, f] = np.int8((a * 7 + r * 3 + f) % 100 - 50)
block[r, 2, f] = np.int8((a * 5 + r * 11 + f * 2) % 120 - 60) block[r, 2, f] = np.int8((a * 5 + r * 11 + f * 2) % 120 - 60)
waveforms.append(block) waveforms.append(block)
# bps=2 stores the same values big-endian int16, exercising the out += block.tobytes()
# reader's >i2 memmap path.
out += (block.astype(">i2") if bps == 2 else block).tobytes()
meta = { meta = {
"n_angles": n_angles, "n_angles": n_angles,
@@ -128,139 +119,13 @@ def build(n_angles: int, seed: int, samples_per_frame: int,
def write(path: Path, n_angles: int = 3, seed: int = 0, def write(path: Path, n_angles: int = 3, seed: int = 0,
samples_per_frame: int = 64, samples_per_frame: int = 64,
geometry: list[tuple[int, int]] | None = None, geometry: list[tuple[int, int]] | None = None) -> dict:
bps: int = 1) -> dict: payload, meta = build(n_angles, seed, samples_per_frame, geometry)
payload, meta = build(n_angles, seed, samples_per_frame, geometry, bps=bps)
path.write_bytes(payload) path.write_bytes(payload)
return meta return meta
# --------------------------------------------------------------------------- HDR_FMT_LEGACY = ">4sBHHffffIIdBB"
# Rotating-sample scan: one shape, imaged at several known rotations
# ---------------------------------------------------------------------------
#
# The scan the angle-alignment path actually has to solve: every angle images
# the *same* sample at a different known rotation and offset, and a correct
# alignment stacks them all back into one shape. Two properties are
# deliberately hostile:
#
# * every angle gets a different window size and a different, meaningless
# stage x_start / y0 — alignment must ignore per-angle stage coordinates
# entirely, so any code that reads them will visibly fail here;
# * the pixel grid is strongly anisotropic (5 µm along x, 50 µm along y),
# like the real instrument, so any registration that rotates raw indices
# instead of millimetres shears the image and cannot converge.
_ROT_DX_MM = 0.005 # x pitch, from velocity/laser_freq below
_ROT_DY_MM = 0.05 # row spacing
_ROT_BG_MV = 4.0
_ROT_FG_MV = 160.0
# How far the sample sits from the rotation axis. Non-zero on purpose: on the
# real instrument every angle's scan window is centred on the rotation axis
# while the sample is not, so each scan sees the sample somewhere else along a
# circle. That offset is exactly what a wrong rotation pivot turns into a ring
# of scans instead of a stack, so a centred test sample would hide the bug.
_ROT_SAMPLE_OFFSET_MM = (0.55, 0.40)
def _sample_shape_mv(u: np.ndarray, v: np.ndarray) -> np.ndarray:
"""An asymmetric test sample in its own mm frame, chirally distinct at
every rotation (no 180° ambiguity) and with structure at several radii so
rotation is well determined."""
u = u - _ROT_SAMPLE_OFFSET_MM[0]
v = v - _ROT_SAMPLE_OFFSET_MM[1]
img = np.full(u.shape, _ROT_BG_MV, dtype=np.float32)
img[((u / 0.85) ** 2 + (v / 0.40) ** 2) <= 1.0] = _ROT_FG_MV # bar
img[(np.abs(u - 0.55) <= 0.22) & (np.abs(v - 0.62) <= 0.22)] = _ROT_FG_MV # nub
img[((u + 0.75) ** 2 + (v + 0.30) ** 2) <= 0.20 ** 2] = _ROT_FG_MV # dot
return img
def write_rotating(path: Path, n_angles: int = 5, samples_per_frame: int = 4,
seed: int = 0) -> dict:
"""Write a v6 file whose CH4 DC image is one sample seen at n_angles known
rotations, and return the ground truth each angle should register to.
``truth[a] = (rotation_deg, (shift_x_mm, shift_y_mm))`` is the rigid map
from angle *a*'s local mm (origin at its own array center) to angle 0's —
exactly what ``register_angle_to_reference`` is supposed to recover.
"""
rng = np.random.default_rng(seed)
n_ch, bps = 3, 1
cal = [(1.5625e-3, -87.04, 0.0), (2.0e-3, -60.0, 1.0e-3), (2.5e-3, -40.0, -2.0e-3)]
ymult_mv, yoff, yzero_mv = cal[2][0] * 1000, cal[2][1], cal[2][2] * 1000
stage_angles, geom, x_starts, y_starts, thetas, offsets = [], [], [], [], [], []
for a in range(n_angles):
stage = -37.0 * a # what the rotation stage reports
stage_angles.append(stage)
# The true image rotation is the negative of the stage's reported
# angle: the stage's positive sense is the opposite of math-positive
# (x toward y) in scan mm. Nothing may depend on knowing that — the
# registration search tries both signs.
thetas.append(-stage)
offsets.append((0.0, 0.0) if a == 0
else (float(rng.uniform(-0.3, 0.3)), float(rng.uniform(-0.3, 0.3))))
# A different window per angle, all centred on the same array center —
# the real instrument grows each angle's axis-aligned bounding box to
# cover the rotated ROI. Sized so the off-axis sample stays inside every
# window at every angle, keeping the expected result unambiguous.
geom.append((88 + 8 * a, 780 + 60 * a))
# Meaningless per-angle stage positions: correct alignment never reads
# them, so scattering them proves it.
x_starts.append(float(20.0 + rng.uniform(-6.0, 6.0)))
y_starts.append(float(30.0 + rng.uniform(-6.0, 6.0)))
out = bytearray()
out += struct.pack(
HDR_FMT_V6, b"SRAS", 6, n_angles,
x_starts[0], y_starts[0], 1.0, 1.0, _ROT_DY_MM,
_VELOCITY_MM_S, _VELOCITY_MM_S / _ROT_DX_MM, # velocity/freq -> 5 µm pitch
samples_per_frame, _SAMPLE_RATE_HZ, bps, n_ch,
)
out += np.array(stage_angles, dtype=">f4").tobytes()
for a, (n_rows, n_frames) in enumerate(geom):
out += struct.pack(GEO_FMT_V6, x_starts[a], 1.0, n_frames, n_rows)
for a, (n_rows, _) in enumerate(geom):
out += (y_starts[a] + np.arange(n_rows) * _ROT_DY_MM).astype(">f4").tobytes()
for ymult_v, yoff_a, yzero_v in cal:
p = _preamble(ymult_v, yoff_a, yzero_v)
out += struct.pack(">H", len(p)) + p
background = rng.integers(-8, 9, size=samples_per_frame, dtype=np.int8)
out += struct.pack(">I", samples_per_frame) + background.tobytes()
truth, dc4_images = {}, []
for a, (n_rows, n_frames) in enumerate(geom):
# Local mm of every pixel, measured from this angle's own array center.
lx = (np.arange(n_frames) - (n_frames - 1) / 2.0) * _ROT_DX_MM
ly = (np.arange(n_rows) - (n_rows - 1) / 2.0) * _ROT_DY_MM
gx, gy = np.meshgrid(lx, ly)
# local = R(theta) @ sample + offset, so sample = R(theta)^T @ (local - offset)
rel = np.stack([gx - offsets[a][0], gy - offsets[a][1]], axis=-1)
s = rel @ _rot(thetas[a]) # == rel @ R^T.T == R^T @ rel
dc4 = _sample_shape_mv(s[..., 0], s[..., 1])
dc4_images.append(dc4)
inv = _rot(-thetas[a])
truth[a] = (-thetas[a],
tuple(float(v) for v in -(inv @ np.array(offsets[a]))))
adc4 = np.clip(np.round((dc4 - yzero_mv) / ymult_mv + yoff), -128, 127).astype(np.int8)
block = np.zeros((n_rows, n_ch, n_frames, samples_per_frame), dtype=np.int8)
block[:, 2] = adc4[:, :, None] # CH4 carries the sample
block[:, 1] = 10 # CH3 flat
block[:, 0] = rng.integers(-40, 41, size=(n_rows, n_frames, samples_per_frame),
dtype=np.int8) # CH1 noise
out += block.tobytes()
path.write_bytes(bytes(out))
return {"n_angles": n_angles, "geometry": geom, "stage_angles_deg": stage_angles,
"truth": truth, "dc4_mv": dc4_images, "x_starts": x_starts,
"y_starts": y_starts, "dx_mm": _ROT_DX_MM, "dy_mm": _ROT_DY_MM}
def write_legacy(path: Path, version: int = 4, n_angles: int = 2, def write_legacy(path: Path, version: int = 4, n_angles: int = 2,
tools/test_gui.py
+485
View File
@@ -0,0 +1,485 @@
#!/usr/bin/env python3
"""Headless GUI test: drives SrasViewerWindow through the real Qt widgets,
signals and worker threads under the offscreen platform plugin.
Covers the interactions a manual smoke test would: load, switch angles and
channels, background DC precompute, lazy FFT compute, threshold and bg-sub
changes, angle alignment, manual angle alignment, aligned view, ROI
draw/move, and CSV export.
Usage: QT_QPA_PLATFORM=offscreen python tools/test_gui.py [file.sras]
"""
import json
import os
import sys
import tempfile
from pathlib import Path
from unittest.mock import patch
os.environ.setdefault("QT_QPA_PLATFORM", "offscreen")
import numpy as np # noqa: E402
from PyQt6.QtCore import QEventLoop, Qt, QTimer # noqa: E402
from PyQt6.QtTest import QTest # noqa: E402
from PyQt6.QtWidgets import QApplication, QMessageBox # noqa: E402
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
import sras_compute as compute # noqa: E402
from sras_format import CH1_IDX, CH3_IDX, CH4_IDX # noqa: E402
from sras_viewer import RoiQuad, SrasViewerWindow, VELOCITY_MODE_IDX # noqa: E402
import tools.make_test_sras as gen # noqa: E402
_failures: list[str] = []
def check(name: str, ok: bool, detail: str = ""):
print(f" {'PASS' if ok else 'FAIL'} {name}" + (f" — {detail}" if detail else ""))
if not ok:
_failures.append(name)
def pump(ms: int = 250):
"""Run the event loop for a while so queued signals and worker threads
make progress."""
loop = QEventLoop()
QTimer.singleShot(ms, loop.quit)
loop.exec()
def wait_until(pred, timeout_ms: int = 20000, step: int = 100) -> bool:
waited = 0
while waited < timeout_ms:
if pred():
return True
pump(step)
waited += step
return pred()
def main():
app = QApplication(sys.argv)
errors: list[str] = []
tmpdir = Path(tempfile.mkdtemp(prefix="sras_gui_"))
path = Path(sys.argv[1]) if len(sys.argv) > 1 else tmpdir / "gui.sras"
if len(sys.argv) <= 1:
gen.write(path, n_angles=4, seed=11, samples_per_frame=256)
print(f"\nloading {path.name}")
win = SrasViewerWindow()
win.show()
# Capture anything the app reports as an error via the status bar.
win.statusBar().messageChanged.connect(
lambda m: errors.append(m) if m and "error" in m.lower() else None)
win._load_file(str(path))
check("file loaded", wait_until(lambda: win._sras is not None))
s = win._sras
check("parsed as v6", s.version == 6, f"v{s.version}")
check("defaults to CH4", win.combo_channel.currentIndex() == CH4_IDX)
check("image displayed", win._current_image is not None)
check("angle spinbox ranges over all angles",
win.spin_angle.maximum() == s.n_angles - 1)
check("scan info populated",
win._info["Angles"].text() == f"Angles: {s.n_angles}",
win._info["Angles"].text())
print("\nbackground DC precompute (all angles)")
ok = wait_until(lambda: all((a, CH4_IDX) in win._dc_cache
and (a, CH3_IDX) in win._dc_cache
for a in range(s.n_angles)))
check("every angle cached for CH3 and CH4", ok,
f"{len(win._dc_cache)} entries")
check("status label reports completion",
"ready for all angles" in win.lbl_dc_precompute.text(),
win.lbl_dc_precompute.text())
print("\nangle switching (DC, should be served from cache)")
for a in range(s.n_angles):
win.spin_angle.setValue(a)
win._on_view_changed()
pump(60)
expected = win._sras.image_shape(a)
check(f"angle {a} shows its own geometry {expected}",
win._current_image.shape == expected,
str(win._current_image.shape))
check("no compute job needed for cached DC angles",
not win._job_running("compute"))
print("\nchannel switching")
win.spin_angle.setValue(0)
win._on_view_changed()
pump(60)
win.combo_channel.setCurrentIndex(CH3_IDX)
check("CH3 displayed", wait_until(lambda: win._current_ch == CH3_IDX))
win.combo_channel.setCurrentIndex(CH1_IDX)
check("CH1 (FFT) computed", wait_until(
lambda: win._current_ch == CH1_IDX and not win._job_running("compute")))
check("FFT result cached", len(win._fft_cache) > 0, f"{len(win._fft_cache)} keys")
rf_img = win._current_image
check("FFT image is non-degenerate", len(np.unique(rf_img)) > 1,
f"{len(np.unique(rf_img))} unique values")
print("\nvelocity mode (pure post-multiply, no recompute)")
n_fft_before = len(win._fft_cache)
win.combo_channel.setCurrentIndex(VELOCITY_MODE_IDX)
check("velocity displayed", wait_until(
lambda: win._current_ch == VELOCITY_MODE_IDX and not win._job_running("compute")))
grating = win.spin_grating_um.value()
check("velocity == freq x grating",
np.allclose(win._current_image, rf_img * grating, atol=1e-3))
check("velocity reused the cached FFT", len(win._fft_cache) == n_fft_before,
f"{n_fft_before} -> {len(win._fft_cache)}")
check("grating spinbox visible in velocity mode", win.grp_velocity.isVisible())
print("\nthreshold change (genuine cache-key change)")
win.combo_channel.setCurrentIndex(CH1_IDX)
wait_until(lambda: not win._job_running("compute"))
dc4 = win._dc_cache[(0, CH4_IDX)]
win.spin_threshold_mv.setValue(float(np.median(dc4)))
win._on_threshold_changed()
check("recomputed at new threshold", wait_until(
lambda: not win._job_running("compute") and len(win._fft_cache) > n_fft_before))
check("masking zeroed some pixels",
int((win._current_image == 0).sum()) > 0,
f"{int((win._current_image == 0).sum())} of {win._current_image.size}")
print("\nbackground subtraction toggle")
n_before = len(win._fft_cache)
win.chk_bg_sub.setChecked(False)
check("recomputed without bg-sub", wait_until(
lambda: not win._job_running("compute") and len(win._fft_cache) > n_before))
win.chk_bg_sub.setChecked(True)
pump(200)
check("returning to bg-sub was a cache hit (no recompute)",
not win._job_running("compute"))
print("\nROI")
x = s.x_axis_mm(0)
y = s.y_positions_mm(0)
roi = RoiQuad.from_bbox(float(x[1]), float(y[1]),
float(x[-2]), float(y[-2]))
win.image_canvas.set_roi(roi)
pump(120)
check("ROI registered", win.image_canvas.get_roi() is not None)
check("pixel count reported",
"pixels inside" in win.lbl_roi_npix.text()
and win.lbl_roi_npix.text() != "pixels inside: —",
win.lbl_roi_npix.text())
npix = int(win.lbl_roi_npix.text().split(":")[1])
check("ROI pixel count is plausible",
0 < npix <= win._current_image.size, f"{npix}")
check("Export ROI enabled", win.btn_export_roi.isEnabled())
csv_path = tmpdir / "roi.csv"
with patch("sras_viewer.QFileDialog.getSaveFileName",
return_value=(str(csv_path), "")):
win._on_export_roi_csv()
check("ROI CSV written", csv_path.exists())
if csv_path.exists():
body = [l for l in csv_path.read_text().splitlines() if not l.startswith("#")]
check("ROI CSV has header + one line per pixel",
len(body) == npix + 1, f"{len(body)} lines for {npix} pixels")
img_csv = tmpdir / "img.csv"
with patch("sras_viewer.QFileDialog.getSaveFileName",
return_value=(str(img_csv), "")):
win._on_export_csv()
check("image CSV written", img_csv.exists())
if img_csv.exists():
arr = np.loadtxt(img_csv, delimiter=",")
check("image CSV round-trips the displayed image",
arr.shape == win._current_image.shape
and np.allclose(arr, win._current_image, rtol=1e-5, atol=1e-4))
print("\nROI survives angle and channel switches")
win.spin_angle.setValue(1)
win._on_view_changed()
wait_until(lambda: not win._job_running("compute"))
check("ROI still present after angle switch",
win.image_canvas.get_roi() is not None)
win.combo_channel.setCurrentIndex(CH4_IDX)
wait_until(lambda: win._current_ch == CH4_IDX)
check("ROI still present after channel switch",
win.image_canvas.get_roi() is not None)
print("\nangle alignment (Fusion)")
win.spin_angle.setValue(0)
win._on_view_changed()
wait_until(lambda: not win._job_running("compute"))
check("alignment action enabled", win._alignment_act.isEnabled())
win._on_angle_alignment()
check("alignment completed", wait_until(
lambda: win._alignment_result is not None and not win._job_running("align"),
timeout_ms=60000))
if win._alignment_result is not None:
r = win._alignment_result
check("transform for every angle", len(r.per_angle) == s.n_angles)
check("canvas is at least as large as any single angle",
all(r.canvas_shape[0] >= int(s.n_rows[a])
and r.canvas_shape[1] >= int(s.n_frames[a])
for a in range(s.n_angles)), str(r.canvas_shape))
check("reference angle has zero shift",
r.per_angle[r.ref_angle_idx].shift_mm == (0.0, 0.0))
check("Aligned View auto-enabled and checked",
win.chk_aligned_view.isEnabled() and win.chk_aligned_view.isChecked())
pump(200)
check("displayed image is on the alignment canvas",
win.image_canvas._img_shape == r.canvas_shape,
f"{win.image_canvas._img_shape} vs {r.canvas_shape}")
win.chk_aligned_view.setChecked(False)
pump(200)
check("unchecking returns to the raw per-angle grid",
win.image_canvas._img_shape == s.image_shape(0),
str(win.image_canvas._img_shape))
print("\nmanual alignment (Fusion)")
check("manual alignment action enabled", win._manual_align_act.isEnabled())
# --- Alignment pivot is a signal-weighted centroid, not the raw bbox --
# center, and is independent of any DC threshold (so a threshold that
# happens to leave a real angle's binary mask empty can't silently
# degrade the pivot back to the bbox center).
corner_signal = np.zeros(s.image_shape(0), dtype=np.float32)
corner_signal[0, 0] = 1.0 # single spike -> weighted centroid is exact
expected_corner = (float(s.x_axis_mm(0)[0]), float(s.y_positions_mm(0)[0]))
centroid = compute._signal_centroid_mm(s, 0, corner_signal)
check("signal-weighted centroid of a single spike pixel is that pixel exactly",
np.allclose(centroid, expected_corner), f"{centroid} vs {expected_corner}")
bbox_center = compute._bbox_center_mm(s, 0)
check("signal centroid differs from the raw scan-window bbox center",
not np.allclose(centroid, bbox_center),
f"centroid {centroid} vs bbox center {bbox_center}")
# compute_pivot_points_mm should reuse a pre-computed dc4_mv dict rather
# than recomputing from the real DC4 image (which has no such spike and
# would give a different answer if silently recomputed).
reused_pivot = compute.compute_pivot_points_mm(s, dc4_mv={0: corner_signal})[0]
check("compute_pivot_points_mm reuses a pre-computed dc4_mv dict",
np.allclose(reused_pivot, expected_corner))
# A perfectly flat signal carries no information to weight by, so it
# falls back to the bbox center rather than producing a NaN/degenerate
# centroid.
flat_signal = np.full(s.image_shape(0), 5.0, dtype=np.float32)
flat_centroid = compute._signal_centroid_mm(s, 0, flat_signal)
check("a perfectly flat signal falls back to the bbox center",
np.allclose(flat_centroid, bbox_center))
# --- Rotation sign convention: negative of the raw angles_deg delta ----
check("_theta_deg negates the raw angles_deg delta (GR stage's positive "
"angle is the opposite rotational sense from this module's CCW "
"math convention)",
all(np.isclose(compute._theta_deg(s, a, 0),
-(float(s.angles_deg[a]) - float(s.angles_deg[0])))
for a in range(s.n_angles)))
# --- FFT phase correlation recovers a known synthetic pixel shift ------
rng = np.random.default_rng(0)
corr_ref = np.zeros((40, 50), dtype=np.float32)
corr_ref[10:25, 15:35] = 1.0
corr_ref += 0.05 * rng.standard_normal(corr_ref.shape).astype(np.float32)
corr_mov = np.roll(corr_ref, shift=(4, -7), axis=(0, 1))
dr, dc = compute._phase_correlate_shift(corr_ref, corr_mov)
check("phase correlation recovers the shift that aligns mov onto ref",
(dr, dc) == (-4, 7), f"got (dr, dc)={(dr, dc)}")
# --- Open: must NOT seed from the still-live automatic AlignmentResult --
# The automatic result's translation comes from FFT phase correlation --
# the very thing manual mode exists to work around -- so manual mode
# must start from identity (centroids coincide, zero shift) regardless
# of whatever the automatic run last computed. Only a previously *saved
# manual* alignment (sidecar) should ever seed this dialog.
win._on_manual_alignment()
check("dialog opened", win._manual_align_dialog is not None)
dlg = win._manual_align_dialog
check("mask prep needed no background worker (already DC-cached)",
not win._job_running("manual_align_masks"))
check("no manual sidecar yet -> dialog starts at identity, not the "
"automatic result",
all(dlg._angle_params[a] == compute.ManualAngleParams()
for a in range(s.n_angles)))
# --- Reference angle is locked -------------------------------------------
dlg.combo_active_angle.setCurrentIndex(dlg._ref_angle_idx)
pump(30)
before_ref = dlg._angle_params[dlg._ref_angle_idx]
dlg._on_nudge_translate(1, 0, False)
dlg._on_nudge_rotate(1, False)
check("reference angle group disabled", not dlg.grp_manual_adjust.isEnabled())
check("reference angle untouched by nudge attempts",
dlg._angle_params[dlg._ref_angle_idx] == before_ref)
# --- Nudging a real angle (fine + coarse, translate + rotate) -----------
active = 1 if s.n_angles > 1 else 0
dlg.combo_active_angle.setCurrentIndex(active)
pump(30)
before = dlg._angle_params[active].shift_mm
dlg._on_nudge_translate(1, 0, False) # fine +X
fine_step = dlg.spin_step_translate_mm.value()
check("fine translate nudge moved shift_x by exactly one fine step",
abs(dlg._angle_params[active].shift_mm[0] - (before[0] + fine_step)) < 1e-9)
before = dlg._angle_params[active].shift_mm
dlg._on_nudge_translate(0, -1, True) # coarse -Y
coarse_step = fine_step * dlg.spin_step_multiplier.value()
check("coarse translate nudge uses the multiplier",
abs(dlg._angle_params[active].shift_mm[1] - (before[1] - coarse_step)) < 1e-9)
before_rot = dlg._angle_params[active].rotation_deg
dlg._on_nudge_rotate(1, False)
check("rotate nudge changed rotation_deg",
dlg._angle_params[active].rotation_deg != before_rot)
check("preview canvas rebuilt for every angle after a rotation nudge",
len(dlg._preview_layers) == s.n_angles)
# --- Real key-event wiring (proves keyPressEvent -> signal -> slot) -----
before = dlg._angle_params[active].shift_mm
QTest.keyClick(dlg.canvas, Qt.Key.Key_Right)
check("a real Right-arrow key event nudged shift_x",
dlg._angle_params[active].shift_mm[0] > before[0])
# --- Auto De-rotate: rotation only, translation untouched ---------------
shift_before_derotate = dlg._angle_params[active].shift_mm
dlg._on_auto_derotate()
expected_theta = compute._theta_deg(s, active, dlg._ref_angle_idx)
check("auto de-rotate set the known analytic angle",
abs(dlg._angle_params[active].rotation_deg - expected_theta) < 1e-6)
check("auto de-rotate left translation untouched",
dlg._angle_params[active].shift_mm == shift_before_derotate)
check("reference angle stays identity after auto de-rotate",
dlg._angle_params[dlg._ref_angle_idx].rotation_deg == 0.0)
# --- Auto Cross-Correlate: rotation + FFT-correlated shift, backgrounded -
check("cross-correlate action enabled once masks are ready",
dlg.btn_auto_correlate.isEnabled())
dlg._on_auto_correlate()
check("auto cross-correlate completed", wait_until(
lambda: not win._job_running("manual_align_correlate"), timeout_ms=30000))
check("auto cross-correlate set the known analytic angle for every angle",
all(abs(dlg._angle_params[a].rotation_deg
- compute._theta_deg(s, a, dlg._ref_angle_idx)) < 1e-6
for a in range(s.n_angles) if a != dlg._ref_angle_idx))
check("auto cross-correlate reference angle stays identity",
dlg._angle_params[dlg._ref_angle_idx] == compute.ManualAngleParams())
check("auto cross-correlate re-enabled controls when done",
dlg.grp_correlate.isEnabled() and dlg.btn_save.isEnabled())
check("preview canvas rebuilt after cross-correlate",
len(dlg._preview_layers) == s.n_angles)
# The thresholded-mask option should also work end to end.
dlg.combo_correlate_source.setCurrentIndex(1) # thresholded mask
dlg._on_auto_correlate()
check("auto cross-correlate (thresholded-mask option) completed", wait_until(
lambda: not win._job_running("manual_align_correlate"), timeout_ms=30000))
# --- Save -----------------------------------------------------------------
dlg._on_save()
sidecar = compute.sidecar_path(s.path)
check("sidecar file written", sidecar.exists())
raw = json.loads(sidecar.read_text()) if sidecar.exists() else {}
check("sidecar schema_version is current",
raw.get("schema_version") == compute._SIDECAR_SCHEMA_VERSION)
check("sidecar per_angle round-trips the dialog's resolved params",
all(raw.get("per_angle", {}).get(str(a), {}).get("rotation_deg")
== dlg._angle_params[a].rotation_deg for a in range(s.n_angles)))
check("main window's alignment_result replaced by the manual build",
win._alignment_result is not None
and win._alignment_result.per_angle[active].rotation_deg
== dlg._angle_params[active].rotation_deg)
check("Aligned View auto-enabled after Save",
win.chk_aligned_view.isEnabled() and win.chk_aligned_view.isChecked())
# --- An old-schema sidecar (pre-pivot/sign fix) is treated as absent ------
stale = dict(raw)
stale["schema_version"] = compute._SIDECAR_SCHEMA_VERSION - 1
sidecar.write_text(json.dumps(stale))
check("a sidecar with an old schema_version is not loaded",
compute.load_manual_alignment(s) is None)
sidecar.write_text(json.dumps(raw)) # restore for the rest of this section
# --- Clear (with confirmation) --------------------------------------------
with patch("sras_viewer.QMessageBox.question",
return_value=QMessageBox.StandardButton.Yes):
dlg._on_clear()
check("sidecar file deleted", not sidecar.exists())
check("dialog params reset to identity",
all(dlg._angle_params[a] == compute.ManualAngleParams()
for a in range(s.n_angles)))
check("main window alignment_result cleared", win._alignment_result is None)
check("Aligned View disabled after Clear",
not win.chk_aligned_view.isEnabled() and not win.chk_aligned_view.isChecked())
dlg.close()
pump(150)
check("dialog reference released on close", win._manual_align_dialog is None)
# --- Sidecar auto-restore on next load ------------------------------------
win._on_manual_alignment()
dlg = win._manual_align_dialog
dlg.combo_active_angle.setCurrentIndex(active)
pump(30)
dlg._on_auto_derotate()
dlg._on_nudge_translate(1, 1, True)
saved_rotation = dlg._angle_params[active].rotation_deg
saved_shift = dlg._angle_params[active].shift_mm
dlg._on_save()
dlg.close()
pump(150)
old_sras_id = id(win._sras)
win._load_file(str(path)) # reload the same file fresh
check("file reloaded", wait_until(
lambda: win._sras is not None and id(win._sras) != old_sras_id))
s = win._sras
check("manual dialog force-closed by a reload", win._manual_align_dialog is None)
check("reload restores the saved manual alignment automatically",
win._alignment_result is not None)
if win._alignment_result is not None:
check("restored rotation matches what was saved",
abs(win._alignment_result.per_angle[active].rotation_deg
- saved_rotation) < 1e-9)
check("restored shift matches what was saved",
win._alignment_result.per_angle[active].shift_mm == saved_shift)
check("Aligned View auto-checked after restoring a saved alignment",
win.chk_aligned_view.isChecked())
print("\npixel inspector")
win.chk_aligned_view.setChecked(False)
pump(100)
win._on_pixel_clicked(0, 0)
pump(150)
check("waveform hint hidden after a click", win.lbl_wave_hint.isHidden())
win.combo_channel.setCurrentIndex(CH1_IDX)
wait_until(lambda: not win._job_running("compute"))
win._on_pixel_clicked(1, 1)
pump(150)
check("RF waveform panel rendered",
len(win.wave_canvas.ax_wave.lines) > 0,
f"{len(win.wave_canvas.ax_wave.lines)} lines")
print("\nshutdown")
win.close()
pump(400)
check("all background jobs released", len(win._jobs) == 0,
f"{list(win._jobs)}")
print()
unexpected = [e for e in errors if e]
if unexpected:
print(f"status-bar errors seen: {unexpected}")
_failures.append("status-bar errors")
if _failures:
print(f"{len(_failures)} FAILURE(S): " + ", ".join(_failures))
return 1
print("All GUI checks passed.")
return 0
if __name__ == "__main__":
sys.exit(main())
tools/test_refactor.py
+358
View File
@@ -0,0 +1,358 @@
#!/usr/bin/env python3
"""Behavioural tests for the sras-viewer refactor.
Covers what the golden-hash harness can't: the v6->v7 cache round-trip
(including block carry-forward), parallel-vs-serial identity, the no-mask
fast path, and the ROI bounding-box mask optimisation.
Usage: python tools/test_refactor.py [--scratch DIR]
"""
import argparse
import shutil
import sys
import tempfile
from pathlib import Path
import numpy as np
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
import sras_compute as compute # noqa: E402
from sras_compute import ( # noqa: E402
cache_file, compute_dc_image, compute_rf_image, dc_image_mv,
)
from sras_format import CH3_IDX, CH4_IDX, SrasFile, adc_to_mv # noqa: E402
import tools.make_test_sras as gen # noqa: E402
_failures: list[str] = []
def check(name: str, ok: bool, detail: str = ""):
print(f" {'PASS' if ok else 'FAIL'} {name}" + (f" — {detail}" if detail else ""))
if not ok:
_failures.append(name)
def test_cache_roundtrip(scratch: Path):
"""v6 -> v7 for DC, then FFT, asserting the first block survives the
second write (the carry-forward path in write_v7_cache)."""
print("\ncache round-trip (v6 -> v7, both blocks)")
path = scratch / "roundtrip.sras"
gen.write(path, n_angles=3, seed=1, samples_per_frame=64)
src = SrasFile(str(path))
check("source is v6", src.version == 6, f"got v{src.version}")
expect_dc3 = [dc_image_mv(src, a, CH3_IDX) for a in range(src.n_angles)]
expect_dc4 = [dc_image_mv(src, a, CH4_IDX) for a in range(src.n_angles)]
expect_fft = [compute_rf_image(src, a, dc_threshold_mv=None, apply_bg_sub=True)
for a in range(src.n_angles)]
err = cache_file(str(path), "dc", True)
check("dc cache_file succeeded", err == "", err)
after_dc = SrasFile(str(path))
check("version flipped to 7", after_dc.version == 7, f"got v{after_dc.version}")
check("dc3 stored for every angle",
all(x is not None for x in after_dc.precomputed_dc3_mv))
check("dc3 values round-trip",
all(np.allclose(after_dc.precomputed_dc3_mv[a], expect_dc3[a], atol=1e-4)
for a in range(after_dc.n_angles)))
check("dc4 values round-trip",
all(np.allclose(after_dc.precomputed_dc4_mv[a], expect_dc4[a], atol=1e-4)
for a in range(after_dc.n_angles)))
check("no fft block yet",
all(x is None for x in after_dc.precomputed_freq_mhz))
check("cached images are native float32",
after_dc.precomputed_dc3_mv[0].dtype == np.float32
and after_dc.precomputed_dc3_mv[0].dtype.byteorder in ("=", "|"),
str(after_dc.precomputed_dc3_mv[0].dtype.byteorder))
check("cached images are writable",
after_dc.precomputed_dc3_mv[0].flags.writeable)
err = cache_file(str(path), "fft", True)
check("fft cache_file succeeded", err == "", err)
both = SrasFile(str(path))
check("fft stored for every angle",
all(x is not None for x in both.precomputed_freq_mhz))
check("fft values round-trip",
all(np.allclose(both.precomputed_freq_mhz[a], expect_fft[a], atol=1e-3)
for a in range(both.n_angles)))
check("DC block carried forward through the FFT write",
all(np.allclose(both.precomputed_dc3_mv[a], expect_dc3[a], atol=1e-4)
for a in range(both.n_angles)))
check("bg_sub flag persisted", both.precomputed_bg_sub is True)
# The fast path must reproduce a fresh compute, and masking must still
# apply on top of a cached (unmasked) image.
fresh = SrasFile(str(path))
fresh.precomputed_freq_mhz = [None] * fresh.n_angles
dc4 = dc_image_mv(both, 0, CH4_IDX)
thr = float(np.median(dc4))
check("cached fast path == fresh compute (unmasked)",
np.allclose(compute_rf_image(both, 0, dc_threshold_mv=None, apply_bg_sub=True),
compute_rf_image(fresh, 0, dc_threshold_mv=None, apply_bg_sub=True),
atol=1e-3))
check("cached fast path == fresh compute (masked)",
np.allclose(compute_rf_image(both, 0, dc_threshold_mv=thr, apply_bg_sub=True),
compute_rf_image(fresh, 0, dc_threshold_mv=thr, apply_bg_sub=True),
atol=1e-3))
# Waveform data must be byte-identical to the pre-cache file.
orig = scratch / "roundtrip_orig.sras"
gen.write(orig, n_angles=3, seed=1, samples_per_frame=64)
o, n = SrasFile(str(orig)), SrasFile(str(path))
check("waveform data untouched by the cache write",
all(np.array_equal(np.asarray(o.data[a]), np.asarray(n.data[a]))
for a in range(o.n_angles)))
def test_partial_v7_cache(scratch: Path):
"""Only some angles cached: uncached angles must compute, not read zeros.
This is the v5 bug the ragged normalisation fixed, checked via v7."""
print("\npartial cache (only some angles stored)")
path = scratch / "partial.sras"
gen.write(path, n_angles=3, seed=2, samples_per_frame=64)
src = SrasFile(str(path))
expected = [compute_rf_image(src, a, dc_threshold_mv=None, apply_bg_sub=True)
for a in range(src.n_angles)]
partial = [expected[0], None, expected[2]] # angle 1 deliberately absent
src.write_v7_cache(new_freq_mhz=partial, new_bg_sub=True)
reread = SrasFile(str(path))
check("angle 1 is not cached", reread.precomputed_freq_mhz[1] is None)
check("angles 0 and 2 are cached",
reread.precomputed_freq_mhz[0] is not None
and reread.precomputed_freq_mhz[2] is not None)
img1 = compute_rf_image(reread, 1, dc_threshold_mv=None, apply_bg_sub=True)
check("uncached angle computes rather than returning zeros",
np.any(img1 != 0) and np.allclose(img1, expected[1], atol=1e-3))
def test_parallel_identity(scratch: Path):
"""Forcing 1 worker vs many must give identical output — catches
chunk-boundary and race bugs."""
print("\nparallel vs serial identity")
path = scratch / "parallel.sras"
# Many rows, so the row loop actually splits into several chunks.
n_rows, n_frames, spf = 48, 9, 256
gen.write(path, n_angles=1, seed=3, samples_per_frame=spf,
geometry=[(n_rows, n_frames)])
sras = SrasFile(str(path))
saved_budget, saved_workers = compute._TOTAL_BYTES_BUDGET, compute._MAX_WORKERS
try:
# Shrink the budget so chunk_rows collapses to 1 and every row is
# its own chunk — the worst case for boundary bugs.
compute._TOTAL_BYTES_BUDGET = 8 * n_frames * spf * 4
compute._MAX_WORKERS = 1
dc_serial = compute_dc_image(sras, 0, CH4_IDX)
rf_serial = compute_rf_image(sras, 0, dc_threshold_mv=None, apply_bg_sub=True)
dc4 = adc_to_mv(dc_serial, *sras.cal(CH4_IDX))
thr = float(np.median(dc4))
rf_masked_serial = compute_rf_image(sras, 0, dc_threshold_mv=thr,
apply_bg_sub=True)
chunk_rows, n_workers = compute._plan_chunks(
n_rows, n_frames, spf, live_multiplier=compute._FFT_LIVE_MULTIPLIER)
check("serial plan uses 1 worker", n_workers == 1, f"chunk_rows={chunk_rows}")
check("work actually splits into multiple chunks", chunk_rows < n_rows,
f"chunk_rows={chunk_rows} of {n_rows} rows")
compute._MAX_WORKERS = 8
chunk_rows, n_workers = compute._plan_chunks(
n_rows, n_frames, spf, live_multiplier=compute._FFT_LIVE_MULTIPLIER)
check("parallel plan uses >1 worker", n_workers > 1,
f"chunk_rows={chunk_rows} workers={n_workers}")
dc_par = compute_dc_image(sras, 0, CH4_IDX)
rf_par = compute_rf_image(sras, 0, dc_threshold_mv=None, apply_bg_sub=True)
rf_masked_par = compute_rf_image(sras, 0, dc_threshold_mv=thr, apply_bg_sub=True)
check("dc image identical", np.array_equal(dc_serial, dc_par))
check("rf image identical (unmasked)", np.array_equal(rf_serial, rf_par))
check("rf image identical (masked)",
np.array_equal(rf_masked_serial, rf_masked_par))
finally:
compute._TOTAL_BYTES_BUDGET, compute._MAX_WORKERS = saved_budget, saved_workers
def test_nomask_equals_low_threshold(scratch: Path):
"""dc_threshold_mv=None must equal a threshold below every pixel, while
skipping the CH4 read."""
print("\nno-mask path")
path = scratch / "nomask.sras"
gen.write(path, n_angles=2, seed=4, samples_per_frame=128)
sras = SrasFile(str(path))
for a in range(sras.n_angles):
none_img = compute_rf_image(sras, a, dc_threshold_mv=None, apply_bg_sub=True)
low_img = compute_rf_image(sras, a, dc_threshold_mv=-1e9, apply_bg_sub=True)
check(f"angle {a}: None == -1e9 threshold",
np.array_equal(none_img, low_img))
check(f"angle {a}: image is non-degenerate",
len(np.unique(none_img)) > 1, f"{len(np.unique(none_img))} unique")
def test_roi_mask():
"""The bbox-restricted mask must equal a full-grid point-in-polygon test."""
print("\nROI mask (bbox fast path vs full grid)")
from matplotlib.path import Path as MplPath
from sras_viewer import RoiQuad
rng = np.random.default_rng(0)
x = np.linspace(-2.0, 3.0, 137)
y = np.linspace(1.0, 4.0, 91)
cases = {
"axis-aligned rect": np.array([[0.0, 1.5], [1.0, 1.5], [1.0, 3.0], [0.0, 3.0]]),
"skewed quad": np.array([[-0.5, 1.2], [1.7, 1.9], [1.2, 3.4], [-1.0, 2.6]]),
"entirely outside": np.array([[8.0, 8.0], [9.0, 8.0], [9.0, 9.0], [8.0, 9.0]]),
"covers whole grid": np.array([[-9.0, -9.0], [9.0, -9.0], [9.0, 9.0], [-9.0, 9.0]]),
"straddles left edge": np.array([[-4.0, 2.0], [0.5, 2.0], [0.5, 3.0], [-4.0, 3.0]]),
}
for _ in range(5):
cases[f"random {_}"] = rng.uniform([-2.5, 0.5], [3.5, 4.5], size=(4, 2))
for name, pts in cases.items():
roi = RoiQuad(pts)
fast = roi.mask_for_grid(x, y)
X, Y = np.meshgrid(x.astype(np.float64), y.astype(np.float64))
slow = MplPath(pts).contains_points(
np.column_stack([X.ravel(), Y.ravel()])).reshape(X.shape)
check(f"{name} ({int(slow.sum())} px inside)", np.array_equal(fast, slow))
# Descending y axis (images are stored top-down in some scans).
roi = RoiQuad(cases["skewed quad"])
y_desc = y[::-1]
fast = roi.mask_for_grid(x, y_desc)
X, Y = np.meshgrid(x.astype(np.float64), y_desc.astype(np.float64))
slow = MplPath(cases["skewed quad"]).contains_points(
np.column_stack([X.ravel(), Y.ravel()])).reshape(X.shape)
check("descending y axis", np.array_equal(fast, slow))
def test_legacy_parse_and_average(scratch: Path):
"""v2-v4 parsing plus the sras_average.py rewrite (which now streams via
SrasFile rather than slurping the whole file)."""
import subprocess
print("\nlegacy formats (v2-v4) and sras_average")
repo = Path(__file__).resolve().parent.parent
for version in (2, 3, 4):
path = scratch / f"legacy_v{version}.sras"
meta = gen.write_legacy(path, version=version, n_angles=2, n_rows=4,
n_frames=12, samples_per_frame=32, seed=version)
s = SrasFile(str(path))
check(f"v{version} parses", s.version == version, f"got v{s.version}")
check(f"v{version} geometry uniform across angles",
list(s.n_rows) == [4, 4] and list(s.n_frames) == [12, 12],
f"rows={list(s.n_rows)} frames={list(s.n_frames)}")
check(f"v{version} waveform data matches what was written",
all(np.array_equal(np.asarray(s.data[a]), meta["data"][a])
for a in range(s.n_angles)))
check(f"v{version} background {'present' if version >= 4 else 'absent'}",
(s.background is not None) == (version >= 4))
check(f"v{version} precomputed stores are ragged lists",
isinstance(s.precomputed_freq_mhz, list)
and len(s.precomputed_freq_mhz) == s.n_angles)
# DC image must equal a direct mean of the known input.
expect = meta["data"][0][:, CH3_IDX, :, :].astype(np.float64).mean(axis=-1)
check(f"v{version} DC image equals a direct mean",
np.allclose(compute_dc_image(s, 0, CH3_IDX), expect, atol=1e-3))
src = scratch / "legacy_v4.sras"
meta = gen.write_legacy(src, version=4, n_angles=2, n_rows=4, n_frames=12,
samples_per_frame=32, seed=4)
dst = scratch / "legacy_v4_avg.sras"
if dst.exists():
dst.unlink()
proc = subprocess.run(
[sys.executable, str(repo / "sras_average.py"), str(src), str(dst), "--n", "4"],
capture_output=True, text=True, cwd=repo)
check("sras_average ran", proc.returncode == 0,
(proc.stderr or proc.stdout).strip()[-200:])
if dst.exists():
avg = SrasFile(str(dst))
check("averaged file parses", avg.version == 4)
check("frame count divided by 4",
list(avg.n_frames) == [3, 3], f"{list(avg.n_frames)}")
check("angles/rows/channels unchanged",
avg.n_angles == 2 and list(avg.n_rows) == [4, 4]
and avg.n_channels == meta["n_channels"])
check("calibration preserved",
np.allclose(avg.ch_ymult_mv, SrasFile(str(src)).ch_ymult_mv))
check("background preserved",
np.array_equal(avg.background, SrasFile(str(src)).background))
src_data = meta["data"]
expect0 = src_data[0][:, :, 0:4, :].astype(np.float32).mean(axis=2).astype(np.int16)
check("first averaged group equals the mean of its 4 source frames",
np.array_equal(np.asarray(avg.data[0])[:, :, 0, :], expect0))
# Remainder handling: 12 frames / 5 -> 2 full groups + 1 partial.
dst2 = scratch / "legacy_v4_avg5.sras"
subprocess.run([sys.executable, str(repo / "sras_average.py"),
str(src), str(dst2), "--n", "5"],
capture_output=True, text=True, cwd=repo)
if dst2.exists():
check("partial trailing group kept by default",
list(SrasFile(str(dst2)).n_frames) == [3, 3],
f"{list(SrasFile(str(dst2)).n_frames)}")
dst3 = scratch / "legacy_v4_avg5d.sras"
subprocess.run([sys.executable, str(repo / "sras_average.py"),
str(src), str(dst3), "--n", "5", "--discard-remainder"],
capture_output=True, text=True, cwd=repo)
if dst3.exists():
check("--discard-remainder drops the partial group",
list(SrasFile(str(dst3)).n_frames) == [2, 2],
f"{list(SrasFile(str(dst3)).n_frames)}")
def test_unsupported_version_reported(scratch: Path):
"""cache_file must report, not raise, for a file it can't handle."""
print("\nerror reporting")
bogus = scratch / "bogus.sras"
bogus.write_bytes(b"SRAS" + bytes([99]) + b"\x00" * 200)
err = cache_file(str(bogus), "dc", True)
check("bad version returns an error string", bool(err), err)
missing = cache_file(str(scratch / "does_not_exist.sras"), "dc", True)
check("missing file returns an error string", bool(missing), missing)
def main():
p = argparse.ArgumentParser(description=__doc__)
p.add_argument("--scratch")
args = p.parse_args()
tmp = None
if args.scratch:
scratch = Path(args.scratch)
scratch.mkdir(parents=True, exist_ok=True)
else:
tmp = tempfile.mkdtemp(prefix="sras_test_")
scratch = Path(tmp)
try:
test_cache_roundtrip(scratch)
test_partial_v7_cache(scratch)
test_parallel_identity(scratch)
test_nomask_equals_low_threshold(scratch)
test_roi_mask()
test_legacy_parse_and_average(scratch)
test_unsupported_version_reported(scratch)
finally:
if tmp:
shutil.rmtree(tmp, ignore_errors=True)
print()
if _failures:
print(f"{len(_failures)} FAILURE(S): " + ", ".join(_failures))
sys.exit(1)
print("All checks passed.")
if __name__ == "__main__":
main()