"""Does a batch-computed FFT cache actually spare the viewer the FFT?

Storing a peak-frequency image per angle in the file is only worth doing if
displaying it is then free. The regression this module pins down is the
viewer's *dispatch* decision: it used to find the stored image only inside
ComputeWorker, so after a batch every angle change still queued a background
job behind a "Computing FFT…" popup for an image already on disk.

Both layers are covered — cached_rf_image's accept/reject rules, and the
window never reaching _start_compute for a batch-cached angle.
"""

import struct
from types import SimpleNamespace
from unittest.mock import patch

import numpy as np
import pytest
from PyQt6.QtCore import QEventLoop, QTimer
from PyQt6.QtWidgets import QApplication, QDialog

import sras_compute as compute
import sras_format as fmt
from sras_compute import cache_file, cached_rf_image, compute_rf_image, dc_image_mv
from sras_format import CH1_IDX, CH3_IDX, CH4_IDX, SrasFile
from sras_viewer import SrasViewerWindow, VELOCITY_MODE_IDX
import tools.make_test_sras as gen

_THRESHOLD_MV = 50.0    # the viewer's own default


def pump(ms: int = 200):
    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 rig(tmp_path_factory):
    """A v6 file, the FFT images a from-scratch compute gives for it, and the
    same file after Batch Compute FFT has written them into its v7 cache."""
    path = tmp_path_factory.mktemp("stored_cache") / "cached.sras"
    gen.write(path, n_angles=4, seed=7, samples_per_frame=256)

    src = SrasFile(str(path))
    fresh = {a: compute_rf_image(src, a, dc_threshold_mv=_THRESHOLD_MV,
                                 apply_bg_sub=True)
             for a in range(src.n_angles)}
    assert src.background is not None, "the fixture file must have a background"

    err = cache_file(str(path), "fft", True)
    assert err == "", err
    cached = SrasFile(str(path))
    assert all(x is not None for x in cached.precomputed_freq_mhz)
    assert all(x is None for x in cached.precomputed_dc4_mv), \
        "FFT-only batch: the mask has to come from the viewer, not the file"

    return SimpleNamespace(path=path, fresh=fresh, sras=cached,
                           n_angles=cached.n_angles)


@pytest.fixture(scope="module")
def dc_rig(tmp_path_factory):
    """A file that has been through Batch Compute DC and Store."""
    path = tmp_path_factory.mktemp("stored_dc") / "dc_cached.sras"
    gen.write(path, n_angles=4, seed=9, samples_per_frame=128)
    err = cache_file(str(path), "dc", True)
    assert err == "", err

    sras = SrasFile(str(path))
    assert all(x is not None for x in sras.precomputed_dc3_mv)
    assert all(x is not None for x in sras.precomputed_dc4_mv)
    assert len(np.unique(sras.precomputed_dc4_mv[0])) > 1, \
        "a degenerate DC image would make the comparisons below vacuous"
    return SimpleNamespace(path=path, sras=sras, n_angles=sras.n_angles)


@pytest.fixture
def no_fft(monkeypatch):
    """Make any real FFT work loud: returns a list that stays empty unless a
    peak search actually runs."""
    calls = []
    for name in ("_peak_bins",):
        original = getattr(compute, name)

        def spy(*args, _f=original, **kwargs):
            calls.append(_f.__name__)
            return _f(*args, **kwargs)

        monkeypatch.setattr(compute, name, spy)
    return calls


# ---------------------------------------------------------------------------
# cached_rf_image: when may the stored image stand in for a compute?
# ---------------------------------------------------------------------------

def test_stored_image_matches_a_fresh_compute(rig, no_fft):
    for a in range(rig.n_angles):
        dc4 = dc_image_mv(SrasFile(str(rig.path)), a, CH4_IDX)
        img = cached_rf_image(rig.sras, a, dc_threshold_mv=_THRESHOLD_MV,
                              apply_bg_sub=True, dc4_mv=dc4)
        assert img is not None, f"angle {a} is cached in the file"
        assert np.allclose(img, rig.fresh[a], atol=1e-3), \
            f"angle {a} differs from a from-scratch compute"
    assert not no_fft, f"the stored image was used, no FFT ran: {no_fft}"


def test_unmasked_when_no_threshold(rig):
    img = cached_rf_image(rig.sras, 0, dc_threshold_mv=None, apply_bg_sub=True)
    assert img is not None and np.array_equal(img, rig.sras.precomputed_freq_mhz[0])
    img[:] = -1.0
    assert not np.any(rig.sras.precomputed_freq_mhz[0] == -1.0), \
        "callers get a copy, never the file's own array"


def test_settings_the_stored_image_cannot_serve(rig):
    """A stored image carries one bg-sub state and one padding, so anything
    else must fall through to a real compute rather than lie."""
    spf = rig.sras.samples_per_frame
    assert cached_rf_image(rig.sras, 0, _THRESHOLD_MV, apply_bg_sub=True,
                           n_fft=spf * 4) is None, \
        "cache was written at pad 1, a pad-4 view resolves different peaks"
    assert cached_rf_image(rig.sras, 0, _THRESHOLD_MV, apply_bg_sub=False) is None, \
        "cache was written with bg-sub on"

    uncached = SrasFile(str(rig.path))
    uncached.precomputed_freq_mhz[1] = None
    assert cached_rf_image(uncached, 1, _THRESHOLD_MV, apply_bg_sub=True) is None


@pytest.mark.parametrize("pad", [2, 10])
def test_cache_is_stored_at_the_configured_pad(tmp_path, pad):
    """Batching at a pad factor must produce a cache that view can read back —
    a pad-1-only cache is one the padded viewer can never use."""
    path = tmp_path / f"pad{pad}.sras"
    gen.write(path, n_angles=2, seed=13, samples_per_frame=256)
    spf = SrasFile(str(path)).samples_per_frame

    assert cache_file(str(path), "fft", True, 0, pad) == ""
    sras = SrasFile(str(path))
    assert sras.precomputed_pad_factor == pad, "pad factor survives the round trip"

    assert cached_rf_image(sras, 0, None, apply_bg_sub=True,
                           n_fft=spf * pad) is not None, f"usable at pad {pad}"
    assert cached_rf_image(sras, 0, None, apply_bg_sub=True) is None, \
        "not usable unpadded"
    assert cached_rf_image(sras, 0, None, apply_bg_sub=True,
                           n_fft=spf * (pad + 1)) is None, "not usable at another pad"

    # The stored numbers must be the padded ones, not pad-1 relabelled.
    fresh = SrasFile(str(path))
    fresh.precomputed_freq_mhz = [None] * fresh.n_angles
    for a in range(sras.n_angles):
        assert np.allclose(
            compute_rf_image(sras, a, dc_threshold_mv=None, apply_bg_sub=True,
                             n_fft=spf * pad),
            compute_rf_image(fresh, a, dc_threshold_mv=None, apply_bg_sub=True,
                             n_fft=spf * pad), atol=1e-3), \
            f"angle {a}: stored image is the pad-{pad} answer"


def test_cach_v1_reads_as_natural_resolution(tmp_path):
    """Files cached before the pad factor existed must keep working: a v1 tail
    has no pad field and is pad 1 by construction."""
    path = tmp_path / "v1.sras"
    gen.write(path, n_angles=2, seed=14, samples_per_frame=256)
    assert cache_file(str(path), "fft", True) == ""

    # Rewrite the tail as a genuine CACH v1 block (old header, no pad field).
    v2 = SrasFile(str(path))
    freq, entries = v2.precomputed_freq_mhz, list(range(v2.n_angles))
    tail_offset = v2._cache_tail_offset()
    payload = struct.pack(fmt.CACH_HDR_FMT, fmt.CACH_MAGIC, 1, fmt.CACH_FLAG_FFT)
    payload += struct.pack(fmt.SFFT_HDR_FMT_V1, fmt.SFFT_MAGIC,
                           fmt.SFFT_FLAG_BG_SUB, len(entries))
    for a in entries:
        payload += struct.pack(">H", a) + freq[a].astype(">f4").tobytes()
    # Windows refuses to truncate a file with a live mapping (write_bytes
    # opens 'wb'), and every SrasFile holds its waveform memmaps for life —
    # drop the instance first. The parsed freq arrays are plain copies and
    # stay usable.
    del v2
    head = path.read_bytes()[:tail_offset]
    path.write_bytes(head + payload)

    v1 = SrasFile(str(path))
    assert v1.precomputed_pad_factor == 1
    assert v1.precomputed_bg_sub is True
    assert all(np.array_equal(v1.precomputed_freq_mhz[a], freq[a]) for a in entries), \
        "v1 images read back unchanged"
    assert cached_rf_image(v1, 0, None, apply_bg_sub=True) is not None
    assert cached_rf_image(v1, 0, None, apply_bg_sub=True,
                           n_fft=v1.samples_per_frame * 10) is None


def test_mask_read_can_be_refused(rig, no_fft):
    """With no DC4 in hand, building the mask means reading a whole channel —
    the GUI thread asks for None instead."""
    assert cached_rf_image(rig.sras, 0, _THRESHOLD_MV, apply_bg_sub=True,
                           allow_dc_recompute=False) is None
    dc4 = dc_image_mv(SrasFile(str(rig.path)), 0, CH4_IDX)
    img = cached_rf_image(rig.sras, 0, _THRESHOLD_MV, apply_bg_sub=True,
                          dc4_mv=dc4, allow_dc_recompute=False)
    assert img is not None and np.allclose(img, rig.fresh[0], atol=1e-3)
    assert not no_fft, f"no FFT on either branch: {no_fft}"


# ---------------------------------------------------------------------------
# The viewer: no compute job at all for a batch-cached angle
# ---------------------------------------------------------------------------

def test_viewer_shows_stored_angles_without_computing(rig, no_fft, monkeypatch):
    app = QApplication.instance() or QApplication([])   # noqa: F841
    win = SrasViewerWindow()
    win.show()

    dispatched = []
    original_start = type(win)._start_compute
    monkeypatch.setattr(type(win), "_start_compute",
                        lambda self: (dispatched.append(self.spin_angle.value()),
                                      original_start(self))[1])
    try:
        win._load_file(str(rig.path))
        assert wait_until(lambda: win._sras is not None), "file loaded"
        # The file carries no DC block, so the mask comes from the window's own
        # background precompute — the state a user is in by the time they click.
        assert wait_until(lambda: all((a, CH4_IDX) in win._dc_cache
                                      for a in range(rig.n_angles))), \
            "DC precompute finished"

        dispatched.clear()
        win.combo_channel.setCurrentIndex(CH1_IDX)
        assert wait_until(lambda: win._current_ch == CH1_IDX), "CH1 displayed"

        for a in list(range(rig.n_angles)) + [1, 0]:
            win.spin_angle.setValue(a)
            win._on_view_changed()
            pump(60)
            assert win._current_angle == a, f"angle {a} displayed"
            assert np.allclose(win._current_image, rig.fresh[a], atol=1e-3), \
                f"angle {a} shows the stored image"

        assert dispatched == [], \
            f"stored angles need no compute job, dispatched for {dispatched}"
        assert not no_fft, f"no FFT ran for any stored angle: {no_fft}"

        # Velocity is still a post-multiply of the same stored image.
        win.combo_channel.setCurrentIndex(VELOCITY_MODE_IDX)
        pump(120)
        assert np.allclose(win._current_image,
                           rig.fresh[0] * win.spin_grating_um.value(), atol=1e-3)
        assert dispatched == [] and not no_fft

        # ...and a live control that no longer matches the stored image's own
        # provenance must NOT force a recompute either — the stored image is
        # shown as-is; only an explicit batch recompute changes what's shown.
        win.combo_channel.setCurrentIndex(CH1_IDX)
        pump(60)
        win.chk_bg_sub.setChecked(False)
        pump(200)
        assert not win._job_running("compute") and not no_fft, \
            "bg-sub off still shows the stored image, no real FFT"
        assert np.allclose(win._current_image, rig.fresh[0], atol=1e-3)
    finally:
        win.close()
        pump(300)


def test_batch_caches_at_the_viewers_pad_factor(tmp_path, no_fft, monkeypatch):
    """The bug a pad-10 user hits: Batch Compute FFT used to store pad-1
    images regardless, so the padded view recomputed every angle forever.
    """
    path = tmp_path / "padded_gui.sras"
    gen.write(path, n_angles=3, seed=15, samples_per_frame=256)

    app = QApplication.instance() or QApplication([])   # noqa: F841
    win = SrasViewerWindow()
    win.show()

    dispatched = []
    original_start = type(win)._start_compute
    monkeypatch.setattr(type(win), "_start_compute",
                        lambda self: (dispatched.append(self.spin_angle.value()),
                                      original_start(self))[1])
    try:
        win._fft_pad_factor = 10
        win._load_file(str(path))
        assert wait_until(lambda: win._sras is not None), "file loaded"
        assert wait_until(lambda: all((a, CH4_IDX) in win._dc_cache
                                      for a in range(win._sras.n_angles))), \
            "DC precompute finished"

        # Convert -> Batch Compute FFT, on the open file, through the real slot.
        with patch("sras_viewer.main_window.QFileDialog.getOpenFileNames",
                   return_value=([str(path)], "")):
            win._on_batch_compute("fft")
        assert wait_until(lambda: not win._job_running("batch"), 60000), "batch ran"
        assert wait_until(lambda: win._sras is not None
                          and win._sras.version == 7), "file reloaded as v7"
        pump(200)
        assert win._sras.precomputed_pad_factor == 10, \
            f"cached at the viewer's pad, got {win._sras.precomputed_pad_factor}"

        expected = {a: compute.cached_rf_image(
                        win._sras, a, dc_threshold_mv=win.spin_threshold_mv.value(),
                        apply_bg_sub=win.chk_bg_sub.isChecked(),
                        n_fft=win._current_n_fft(),
                        dc4_mv=win._dc_cache.get((a, CH4_IDX)))
                    for a in range(win._sras.n_angles)}
        assert all(v is not None for v in expected.values()), "cache is readable at pad 10"

        no_fft.clear()
        dispatched.clear()
        win.combo_channel.setCurrentIndex(CH1_IDX)
        assert wait_until(lambda: win._current_ch == CH1_IDX), "CH1 displayed"
        for a in range(win._sras.n_angles):
            win.spin_angle.setValue(a)
            pump(60)
            assert np.allclose(win._current_image, expected[a], atol=1e-3), \
                f"angle {a} served from the pad-10 cache"
        assert dispatched == [] and not no_fft, \
            f"no recompute at pad 10 (jobs={dispatched}, fft={no_fft})"
        assert "unusable" not in win.lbl_frame_warn.text()

        # Change the live pad control so it no longer matches the stored
        # image's own provenance — the info panel has to say so rather than
        # leave it a mystery, but the stored pad-10 image keeps displaying;
        # only an explicit batch recompute would ever produce a pad-4 one.
        win._fft_pad_factor = 4
        win._update_scan_info_labels()
        assert "differ from the stored cache" in win.lbl_frame_warn.text(), \
            win.lbl_frame_warn.text()
        assert "pad 10x" in win.lbl_frame_warn.text()
        last_angle = win._current_angle
        win._refresh_display()
        assert wait_until(lambda: not win._job_running("compute")), "settled"
        assert not no_fft, "no real FFT ran — the pad-10 cache still served the view"
        assert np.allclose(win._current_image, expected[last_angle], atol=1e-3), \
            "pad-10 cache still shown after the live pad control diverged"
    finally:
        win.close()
        pump(300)


def test_viewer_shows_stored_dc_without_computing(dc_rig, monkeypatch):
    """Same for the DC half, with the background precompute silenced so the
    file's stored block is the only thing that can be carrying the display."""
    app = QApplication.instance() or QApplication([])   # noqa: F841
    win = SrasViewerWindow()
    win.show()

    dispatched = []
    original_start = type(win)._start_compute
    monkeypatch.setattr(type(win), "_start_compute",
                        lambda self: (dispatched.append(self.spin_angle.value()),
                                      original_start(self))[1])
    monkeypatch.setattr(type(win), "_start_dc_precompute", lambda self: None)
    try:
        win._load_file(str(dc_rig.path))
        assert wait_until(lambda: win._sras is not None), "file loaded"
        pump(120)

        for ch in (CH3_IDX, CH4_IDX):
            win.combo_channel.setCurrentIndex(ch)
            for a in range(dc_rig.n_angles):
                win.spin_angle.setValue(a)
                pump(60)
                assert (win._current_angle, win._current_ch) == (a, ch), \
                    f"angle {a} on channel {ch} displayed"
                assert np.array_equal(win._current_image,
                                      dc_rig.sras.cached_dc_mv(a, ch)), \
                    f"angle {a} channel {ch} shows the file's stored DC image"

        assert dispatched == [], \
            f"stored DC angles need no compute job, dispatched for {dispatched}"
    finally:
        win.close()
        pump(300)


# ---------------------------------------------------------------------------
# Row-averaged FFT: cache_file("fft_rowavg", ...) and its on-disk provenance
# ---------------------------------------------------------------------------

def test_row_average_flag_and_n_round_trip(tmp_path):
    path = tmp_path / "rowavg_roundtrip.sras"
    gen.write(path, n_angles=2, seed=20, samples_per_frame=128)
    err = cache_file(str(path), "fft_rowavg", True, dc_threshold_mv=-1e9, row_avg_n=5)
    assert err == "", err

    sras = SrasFile(str(path))
    assert sras.version == 7
    assert sras.precomputed_row_avg_n == 5
    assert all(x is not None for x in sras.precomputed_freq_mhz)


def test_fft_rowavg_mode_requires_positive_n_and_threshold(tmp_path):
    path = tmp_path / "rowavg_bad_args.sras"
    gen.write(path, n_angles=1, seed=27, samples_per_frame=64)
    assert cache_file(str(path), "fft_rowavg", True, dc_threshold_mv=0.0, row_avg_n=0)
    assert cache_file(str(path), "fft_rowavg", True, dc_threshold_mv=None, row_avg_n=5)


def test_raw_and_row_averaged_caches_never_cross_served(tmp_path):
    """The central regression this feature must never allow: a raw request
    served a row-averaged image (or vice versa), or a request at one window
    size served a cache stored at a different one."""
    path = tmp_path / "cross_serve.sras"
    gen.write(path, n_angles=1, seed=21, samples_per_frame=128)
    err = cache_file(str(path), "fft_rowavg", True, dc_threshold_mv=-1e9, row_avg_n=5)
    assert err == "", err

    sras = SrasFile(str(path))
    assert cached_rf_image(sras, 0, None, apply_bg_sub=True, row_avg_n=0) is None, \
        "a raw request must not be served a row-averaged cache"
    assert cached_rf_image(sras, 0, None, apply_bg_sub=True, row_avg_n=3) is None, \
        "a request at the wrong window size must not be served either"
    served = cached_rf_image(sras, 0, None, apply_bg_sub=True, row_avg_n=5)
    assert served is not None
    assert np.array_equal(served, sras.precomputed_freq_mhz[0])


def test_write_v7_cache_row_avg_n_carries_forward(tmp_path):
    """A later DC-only write must leave a previously-written row-averaged
    FFT block -- including its row_avg_n -- byte-for-byte unchanged."""
    path = tmp_path / "carry_forward.sras"
    gen.write(path, n_angles=2, seed=22, samples_per_frame=64)
    err = cache_file(str(path), "fft_rowavg", True, dc_threshold_mv=-1e9, row_avg_n=7)
    assert err == "", err

    before = SrasFile(str(path))
    assert before.precomputed_row_avg_n == 7
    freq_before = [x.copy() for x in before.precomputed_freq_mhz]

    err = cache_file(str(path), "dc", True)
    assert err == "", err

    after = SrasFile(str(path))
    assert after.precomputed_row_avg_n == 7, "row_avg_n survives a DC-only write"
    assert all(np.array_equal(after.precomputed_freq_mhz[a], freq_before[a])
               for a in range(after.n_angles)), \
        "the row-averaged FFT block itself is untouched by a DC-only write"


def test_row_average_never_touches_dc_images(tmp_path):
    path = tmp_path / "dc_untouched.sras"
    gen.write(path, n_angles=2, seed=23, samples_per_frame=64)

    src = SrasFile(str(path))
    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)]

    assert cache_file(str(path), "dc", True) == ""
    assert cache_file(str(path), "fft_rowavg", True,
                      dc_threshold_mv=-1e9, row_avg_n=6) == ""

    after = SrasFile(str(path))
    assert all(np.allclose(after.precomputed_dc3_mv[a], expect_dc3[a], atol=1e-4)
               for a in range(after.n_angles))
    assert all(np.allclose(after.precomputed_dc4_mv[a], expect_dc4[a], atol=1e-4)
               for a in range(after.n_angles))


def test_row_average_never_modifies_raw_waveform_data(tmp_path):
    path = tmp_path / "waveform_untouched.sras"
    gen.write(path, n_angles=2, seed=24, samples_per_frame=64)
    orig = tmp_path / "waveform_untouched_orig.sras"
    gen.write(orig, n_angles=2, seed=24, samples_per_frame=64)

    assert cache_file(str(path), "fft_rowavg", True,
                      dc_threshold_mv=-1e9, row_avg_n=5) == ""

    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 a row-averaged cache write"


def test_cach_v1_backward_compat_defaults_row_avg_n_zero(tmp_path):
    """A v1 CACH tail predates row-averaged FFT caching entirely (no
    row_avg_n byte at all) -- readers must still parse it in full, treating
    it as row_avg_n=0. This is what protects an existing real-world v7
    file's already-stored FFT cache from silently becoming unusable after
    this change ships."""
    path = tmp_path / "v1_rowavg.sras"
    gen.write(path, n_angles=2, seed=25, samples_per_frame=128)
    assert cache_file(str(path), "fft", True) == ""

    v2 = SrasFile(str(path))
    freq, entries = v2.precomputed_freq_mhz, list(range(v2.n_angles))
    tail_offset = v2._cache_tail_offset()
    payload = struct.pack(fmt.CACH_HDR_FMT, fmt.CACH_MAGIC, 1, fmt.CACH_FLAG_FFT)
    payload += struct.pack(fmt.SFFT_HDR_FMT_V1, fmt.SFFT_MAGIC,
                           fmt.SFFT_FLAG_BG_SUB, len(entries))
    for a in entries:
        payload += struct.pack(">H", a) + freq[a].astype(">f4").tobytes()
    # See test_cach_v1_reads_as_natural_resolution: release the memmaps
    # before write_bytes truncates, or Windows raises EINVAL.
    del v2
    head = path.read_bytes()[:tail_offset]
    path.write_bytes(head + payload)

    v1 = SrasFile(str(path))
    assert v1.precomputed_row_avg_n == 0
    assert all(np.array_equal(v1.precomputed_freq_mhz[a], freq[a]) for a in entries), \
        "v1 images read back unchanged"
    assert cached_rf_image(v1, 0, None, apply_bg_sub=True, row_avg_n=0) is not None
    assert cached_rf_image(v1, 0, None, apply_bg_sub=True, row_avg_n=5) is None, \
        "a v1 tail (predating this feature) can never satisfy a row-averaged request"


# ---------------------------------------------------------------------------
# Min peak frequency floor: serve-time masking, on-disk provenance, batch
# ---------------------------------------------------------------------------

def _biting_floor(stored: np.ndarray) -> float:
    """A floor that zeroes some-but-not-all of *stored*'s positive peaks:
    the median distinct positive value, so pixels below it get masked and
    pixels at/above it survive (the mask is a strict <)."""
    positive = np.unique(stored[stored > 0])
    assert len(positive) >= 2, "fixture must have varied peak frequencies"
    return float(positive[len(positive) // 2])


def test_stored_image_served_with_raised_floor_masked(rig, no_fft):
    """The core of the 5 m/s bug fix: a floor above the stored one (here 0)
    is re-applied when the stored image is served — pixels whose stored
    peak falls below it come back as the 0.0 invalid sentinel, everything
    else passes through, and no FFT runs. Both directly through
    cached_rf_image and through compute_rf_image's fast path."""
    assert rig.sras.precomputed_min_freq_mhz == 0.0, "batched without a floor"
    stored = rig.sras.precomputed_freq_mhz[0]
    floor = _biting_floor(stored)
    expect = np.where(stored < floor, np.float32(0.0), stored)

    img = cached_rf_image(rig.sras, 0, None, apply_bg_sub=True,
                          min_freq_mhz=floor)
    assert img is not None, "an equal-or-higher floor is servable"
    assert np.array_equal(img, expect)
    assert (img == 0).any() and (img > 0).any(), \
        "the floor bites some pixels but not all"

    via_compute = compute_rf_image(rig.sras, 0, dc_threshold_mv=None,
                                   apply_bg_sub=True, min_freq_mhz=floor)
    assert np.array_equal(via_compute, expect), \
        "compute_rf_image's fast path applies the same serve-time mask"
    assert not no_fft, f"serving + masking must not run an FFT: {no_fft}"


def test_min_freq_floor_round_trips_and_gates_serving(tmp_path):
    """cache_file records the floor in the SFFT header and the accept rule
    is asymmetric: an equal-or-higher request is servable, a lower one is
    refused (the stored search never looked below its floor). 20.1 pins the
    fixed-point kHz encoding — a float32 header field would read back as
    20.10000038…, above the requested 20.1, and mismatch forever."""
    path = tmp_path / "floor_roundtrip.sras"
    gen.write(path, n_angles=2, seed=31, samples_per_frame=128)
    floor = 20.1
    assert cache_file(str(path), "fft", True, min_freq_mhz=floor) == ""

    sras = SrasFile(str(path))
    assert sras.precomputed_min_freq_mhz == floor, "exact fixed-point round-trip"
    assert all(((img == 0) | (img >= floor)).all()
               for img in sras.precomputed_freq_mhz), \
        "no stored peak below the floor"

    def reasons(f):
        return compute.cache_mismatch_reasons(
            sras, n_fft=None, apply_bg_sub=True, row_avg_n=0, min_freq_mhz=f)

    assert reasons(floor) == []
    assert reasons(floor + 5.0) == [], "a higher request is servable (masked)"
    low = reasons(0.0)
    assert low and "min-peak-freq floor" in low[0], \
        "a lower request cannot be answered by the stored search"
    assert cached_rf_image(sras, 0, None, apply_bg_sub=True) is None, \
        "default floor-0 request refused against a floored store"
    assert cached_rf_image(sras, 0, None, apply_bg_sub=True,
                           min_freq_mhz=floor) is not None


def test_batch_recompute_resolves_not_masks(tmp_path, no_fft):
    """Re-batching an already-cached file at a raised floor must run the
    real FFT and store re-resolved peaks — never let compute_rf_image's
    fast path serve the file's own stale cache back to it and bake the
    masked copy in as if it were a recompute (silent, permanent data
    loss: a masked pixel's true above-floor peak is unrecoverable)."""
    path = tmp_path / "rebatch.sras"
    gen.write(path, n_angles=2, seed=32, samples_per_frame=256)
    assert cache_file(str(path), "fft", True) == ""

    first = SrasFile(str(path))
    n_angles = first.n_angles
    # On the header's kHz grid up front (as the spinbox value would be), so
    # the recorded floor reads back equal — cache_file quantizes whatever it
    # is given, and this test wants that to be the identity.
    floor = round(_biting_floor(first.precomputed_freq_mhz[0]) * 1000) / 1000.0
    bites = [img < floor for img in first.precomputed_freq_mhz]
    assert bites[0].any(), "the floor must actually bite this fixture"
    # What a real floored compute gives, from a view blinded to the cache.
    first.precomputed_freq_mhz = [None] * n_angles
    expected = [compute_rf_image(first, a, dc_threshold_mv=None,
                                 apply_bg_sub=True, min_freq_mhz=floor)
                for a in range(n_angles)]
    del first   # release memmaps before cache_file rewrites the tail

    no_fft.clear()
    assert cache_file(str(path), "fft", True, min_freq_mhz=floor) == ""
    assert no_fft, "the re-batch ran a real FFT"

    after = SrasFile(str(path))
    assert after.precomputed_min_freq_mhz == floor
    for a in range(n_angles):
        assert np.array_equal(after.precomputed_freq_mhz[a], expected[a]), \
            f"angle {a}: stored image is a real floored recompute"
        assert (after.precomputed_freq_mhz[a][bites[a]] >= floor).all(), \
            f"angle {a}: bitten pixels re-resolved above the floor, not zeroed"


def test_min_freq_carries_forward_through_dc_write(tmp_path):
    """A later DC-only write must leave the FFT block's recorded floor
    untouched, like row_avg_n and pad_factor."""
    path = tmp_path / "floor_carry.sras"
    gen.write(path, n_angles=2, seed=33, samples_per_frame=128)
    assert cache_file(str(path), "fft", True, min_freq_mhz=75.0) == ""
    assert cache_file(str(path), "dc", True) == ""
    after = SrasFile(str(path))
    assert after.precomputed_min_freq_mhz == 75.0, \
        "floor survives a DC-only write"


def test_cach_v3_backward_compat_defaults_floor_zero(tmp_path):
    """A v3 CACH tail predates the min peak frequency floor entirely (no
    min_freq_khz field) — readers must still parse it in full, treating it
    as floor 0: servable as-is at floor 0, and serve-maskable at any higher
    one. This is what protects existing real-world v7 caches from silently
    becoming unusable after the v4 bump ships."""
    path = tmp_path / "v3_floor.sras"
    gen.write(path, n_angles=2, seed=34, samples_per_frame=128)
    assert cache_file(str(path), "fft", True) == ""

    # Rewrite the tail as a genuine CACH v3 block (no min_freq field).
    v4 = SrasFile(str(path))
    freq, entries = v4.precomputed_freq_mhz, list(range(v4.n_angles))
    tail_offset = v4._cache_tail_offset()
    payload = struct.pack(fmt.CACH_HDR_FMT, fmt.CACH_MAGIC, 3, fmt.CACH_FLAG_FFT)
    payload += struct.pack(fmt.SFFT_HDR_FMT_V3, fmt.SFFT_MAGIC,
                           fmt.SFFT_FLAG_BG_SUB, len(entries), 0, 1)
    for a in entries:
        payload += struct.pack(">H", a) + freq[a].astype(">f4").tobytes()
    # See test_cach_v1_reads_as_natural_resolution: release the memmaps
    # before write_bytes truncates, or Windows raises EINVAL.
    del v4
    head = path.read_bytes()[:tail_offset]
    path.write_bytes(head + payload)

    v3 = SrasFile(str(path))
    assert v3.precomputed_min_freq_mhz == 0.0
    assert v3.precomputed_pad_factor == 1 and v3.precomputed_bg_sub is True
    assert all(np.array_equal(v3.precomputed_freq_mhz[a], freq[a])
               for a in entries), "v3 images read back unchanged"
    assert cached_rf_image(v3, 0, None, apply_bg_sub=True) is not None
    floor = _biting_floor(freq[0])
    served = cached_rf_image(v3, 0, None, apply_bg_sub=True,
                             min_freq_mhz=floor)
    assert served is not None
    assert np.array_equal(served,
                          np.where(freq[0] < floor, np.float32(0.0), freq[0]))


def test_min_freq_validation(tmp_path):
    path = tmp_path / "floor_bad.sras"
    gen.write(path, n_angles=1, seed=35, samples_per_frame=64)
    assert cache_file(str(path), "fft", True, min_freq_mhz=-1.0), \
        "negative floor must be an error, not a write"
    assert cache_file(str(path), "fft", True, min_freq_mhz=float("nan")), \
        "NaN floor must be an error, not a write"
    sras = SrasFile(str(path))
    with pytest.raises(ValueError):
        sras.write_v7_cache(new_min_freq_mhz=-0.5)


def test_viewer_reapplies_floor_to_stored_images_without_computing(
        rig, no_fft, monkeypatch):
    """The session-cache poisoning bug behind the '95 MHz peak but 5 m/s'
    report: changing 'Min peak freq' against a stored cache used to re-file
    the identical un-floored image under a key claiming the new floor — the
    UI looked updated, the pixels weren't. Now the floor really is
    re-applied on serve (masked, still no compute), and clearing it
    restores the unmasked image, still without computing."""
    app = QApplication.instance() or QApplication([])   # noqa: F841
    win = SrasViewerWindow()
    win.show()

    dispatched = []
    original_start = type(win)._start_compute
    monkeypatch.setattr(type(win), "_start_compute",
                        lambda self: (dispatched.append(self.spin_angle.value()),
                                      original_start(self))[1])
    try:
        win._load_file(str(rig.path))
        assert wait_until(lambda: win._sras is not None), "file loaded"
        assert wait_until(lambda: all((a, CH4_IDX) in win._dc_cache
                                      for a in range(rig.n_angles))), \
            "DC precompute finished"

        win.combo_channel.setCurrentIndex(CH1_IDX)
        assert wait_until(lambda: win._current_ch == CH1_IDX), "CH1 displayed"
        assert np.allclose(win._current_image, rig.fresh[0], atol=1e-3)

        # Round to the spinbox's 3-decimal granularity; the chosen bin value
        # still survives its own (strict-<) floor after rounding down.
        floor = round(_biting_floor(rig.fresh[0]), 3)
        expect = np.where(rig.fresh[0] < floor, np.float32(0.0), rig.fresh[0])

        dispatched.clear()
        no_fft.clear()
        win.spin_min_freq_mhz.setValue(floor)
        win._on_min_freq_changed()
        pump(120)
        assert np.allclose(win._current_image, expect, atol=1e-3), \
            "raised floor re-masks the stored image on serve"
        assert (win._current_image == 0).any() and (win._current_image > 0).any()
        assert dispatched == [] and not no_fft, \
            f"re-masked serve needs no compute (jobs={dispatched}, fft={no_fft})"
        key = (0, win.spin_threshold_mv.value(), win.spin_min_freq_mhz.value())
        assert key in win._fft_cache
        assert np.allclose(win._fft_cache[key], expect, atol=1e-3), \
            "the session cache holds the value its key claims"

        win.spin_min_freq_mhz.setValue(0.0)
        win._on_min_freq_changed()
        pump(120)
        assert np.allclose(win._current_image, rig.fresh[0], atol=1e-3), \
            "clearing the floor restores the unmasked stored image"
        assert dispatched == [] and not no_fft
    finally:
        win.close()
        pump(300)


def test_viewer_batch_fft_records_the_floor(tmp_path, no_fft, monkeypatch):
    """Convert → Batch Compute FFT with a floor set: the live spinbox value
    reaches cache_file, lands in the reloaded file's provenance and the
    info panel, and the viewer then serves the floored cache without
    recomputing — the tooltip's promised remedy, end to end."""
    path = tmp_path / "floor_gui.sras"
    gen.write(path, n_angles=2, seed=36, samples_per_frame=256)

    app = QApplication.instance() or QApplication([])   # noqa: F841
    win = SrasViewerWindow()
    win.show()

    dispatched = []
    original_start = type(win)._start_compute
    monkeypatch.setattr(type(win), "_start_compute",
                        lambda self: (dispatched.append(self.spin_angle.value()),
                                      original_start(self))[1])
    try:
        win._load_file(str(path))
        assert wait_until(lambda: win._sras is not None), "file loaded"
        assert wait_until(lambda: all((a, CH4_IDX) in win._dc_cache
                                      for a in range(win._sras.n_angles))), \
            "DC precompute finished"

        floor = 100.0
        win.spin_min_freq_mhz.setValue(floor)
        with patch("sras_viewer.main_window.QFileDialog.getOpenFileNames",
                   return_value=([str(path)], "")):
            win._on_batch_compute("fft")
        assert wait_until(lambda: not win._job_running("batch"), 60000), "batch ran"
        assert wait_until(lambda: win._sras is not None
                          and win._sras.version == 7), "file reloaded as v7"
        pump(200)
        assert win._sras.precomputed_min_freq_mhz == floor, \
            "the viewer's floor reached the stored provenance"
        assert "floor ≥ 100 MHz" in win.lbl_frame_warn.text()

        no_fft.clear()
        dispatched.clear()
        win.combo_channel.setCurrentIndex(CH1_IDX)
        assert wait_until(lambda: win._current_ch == CH1_IDX), "CH1 displayed"
        pump(120)
        assert dispatched == [] and not no_fft, \
            f"floored cache serves the view directly (jobs={dispatched}, fft={no_fft})"
        img = win._current_image
        assert ((img == 0) | (img >= floor)).all(), \
            "no displayed peak below the floor"
    finally:
        win.close()
        pump(300)


def test_viewer_batch_row_average_dispatch(tmp_path, monkeypatch, no_fft):
    """Driving the new 'Batch Compute Row-Averaged FFT and Store' action
    end-to-end through the real menu handler: dialog values reach the
    worker, the worker reaches cache_file, and the written file is
    self-describing afterward. Also the exact scenario the row-averaged-FFT
    recompute bug reported: before the fix, the display always asked for
    row_avg_n=0 regardless of what the file actually had stored, so viewing
    an angle after this batch action saw a phantom mismatch and launched a
    full raw recompute on every view switch. This asserts that no longer
    happens -- the stored row-averaged image is shown directly, with no
    dispatched compute job and no real FFT."""
    path = tmp_path / "rowavg_gui.sras"
    gen.write(path, n_angles=2, seed=26, samples_per_frame=128)

    class _StubDialog:
        def __init__(self, *a, **k):
            pass

        def exec(self):
            return QDialog.DialogCode.Accepted

        def get_half_width(self):
            return 6

        def get_threshold_mv(self):
            return -1e9   # mask nothing, keep the comparison simple

    monkeypatch.setattr("sras_viewer.main_window.RowAverageFftOptionsDialog", _StubDialog)

    app = QApplication.instance() or QApplication([])   # noqa: F841
    win = SrasViewerWindow()
    win.show()

    dispatched = []
    original_start = type(win)._start_compute
    monkeypatch.setattr(type(win), "_start_compute",
                        lambda self: (dispatched.append(self.spin_angle.value()),
                                      original_start(self))[1])
    try:
        win._load_file(str(path))
        assert wait_until(lambda: win._sras is not None), "file loaded"
        assert wait_until(lambda: all((a, CH4_IDX) in win._dc_cache
                                      for a in range(win._sras.n_angles))), \
            "DC precompute finished"

        with patch("sras_viewer.main_window.QFileDialog.getOpenFileNames",
                   return_value=([str(path)], "")):
            win._on_batch_compute_row_avg()
        assert wait_until(lambda: not win._job_running("batch"), 60000), "batch ran"
        assert wait_until(lambda: win._sras is not None
                          and win._sras.version == 7), "file reloaded as v7"
        pump(200)

        assert win._sras.precomputed_row_avg_n == 6
        assert "row-averaged n=6" in win.lbl_frame_warn.text(), win.lbl_frame_warn.text()

        expected = compute.cached_rf_image(win._sras, 0, dc_threshold_mv=None,
                                           apply_bg_sub=win.chk_bg_sub.isChecked(),
                                           row_avg_n=6)
        assert expected is not None, "the batch write left a readable row-averaged cache"

        # The regression this batch action used to leave unfixed: viewing an
        # angle afterward must show the stored row-averaged image directly,
        # never fall through to a real (raw) recompute.
        no_fft.clear()
        dispatched.clear()
        win.combo_channel.setCurrentIndex(CH1_IDX)
        assert wait_until(lambda: win._current_ch == CH1_IDX), "CH1 displayed"
        for a in range(win._sras.n_angles):
            win.spin_angle.setValue(a)
            pump(60)
            assert np.allclose(win._current_image,
                               compute.cached_rf_image(
                                   win._sras, a,
                                   dc_threshold_mv=win.spin_threshold_mv.value(),
                                   apply_bg_sub=win.chk_bg_sub.isChecked(),
                                   row_avg_n=6), atol=1e-3), \
                f"angle {a} shows the stored row-averaged image"
        assert dispatched == [] and not no_fft, \
            f"no recompute for row-averaged angles (jobs={dispatched}, fft={no_fft})"
    finally:
        win.close()
        pump(300)


def test_compute_worker_fallback_honors_row_avg_n(tmp_path, monkeypatch, no_fft):
    """The other half of the row-averaged recompute bug: _stored_fft_image
    isn't the only path that can serve a view. When the DC4 mask isn't known
    yet without I/O (e.g. background DC precompute hasn't reached this angle),
    _stored_fft_image's allow_dc_recompute=False guard bails and
    _refresh_display falls through to _start_compute's ComputeWorker instead.
    That worker must still ask compute_rf_image for the file's own
    row_avg_n -- not silently default to 0 -- so its own internal cache
    fast path also serves the stored row-averaged image rather than running
    a real, non-averaged FFT."""
    path = tmp_path / "rowavg_fallback.sras"
    gen.write(path, n_angles=2, seed=33, samples_per_frame=128)
    # Only fft_rowavg -- deliberately no DC block, exactly what "Batch
    # Compute Row-Averaged FFT and Store" leaves on disk by itself.
    err = cache_file(str(path), "fft_rowavg", True, dc_threshold_mv=-1e9, row_avg_n=5)
    assert err == "", err

    app = QApplication.instance() or QApplication([])   # noqa: F841
    win = SrasViewerWindow()
    win.show()

    # Keep _dc_cache empty so _stored_fft_image can't resolve a mask without
    # I/O and _refresh_display must fall through to _start_compute.
    monkeypatch.setattr(type(win), "_start_dc_precompute", lambda self: None)

    dispatched = []
    original_start = type(win)._start_compute
    monkeypatch.setattr(type(win), "_start_compute",
                        lambda self: (dispatched.append(self.spin_angle.value()),
                                      original_start(self))[1])
    try:
        win._load_file(str(path))
        assert wait_until(lambda: win._sras is not None), "file loaded"
        assert win._sras.precomputed_row_avg_n == 5
        # The initial default view (CH4, DC) has nothing cached either, so
        # it dispatches its own one-off DC compute for angle 0 on load --
        # unrelated to this bug. Let that settle, then clear it so no DC4 is
        # available for any angle, simulating "background DC precompute
        # hasn't reached this angle yet".
        assert wait_until(lambda: not win._job_running("compute")), "initial DC view settled"
        dispatched.clear()
        win._dc_cache.clear()

        no_fft.clear()
        win.combo_channel.setCurrentIndex(CH1_IDX)
        assert wait_until(lambda: win._current_ch == CH1_IDX
                          and not win._job_running("compute")), "CH1 displayed"

        assert dispatched == [0], \
            "with no DC cache available yet, the fallback compute must run"
        assert not no_fft, \
            f"the fallback's own compute_rf_image call must still hit the " \
            f"stored row-averaged cache internally: {no_fft}"
        expected = compute.cached_rf_image(
            win._sras, 0, dc_threshold_mv=win.spin_threshold_mv.value(),
            apply_bg_sub=win.chk_bg_sub.isChecked(), row_avg_n=5)
        assert expected is not None
        assert np.allclose(win._current_image, expected, atol=1e-3), \
            "the displayed image is the stored row-averaged one, not a raw recompute"
    finally:
        win.close()
        pump(300)