Receiver: prototype FPGAImagePuller, PSI detector frames through the stream receivers

A prototype puller over the acquisition devices: one thread per module takes the converted frames off the
card and writes them into their image, worker threads compress, and the images leave in image-number order
as a filled DataMessage between a start and an end message. JFJochReceiverLite and JFJochReceiverRugnux then
take them as they take a DECTRIS stream. Off by default and not wired to the broker:
JFJochReceiverService::UseFPGAImagePuller() selects it, the FPGA receiver stays the default, and pedestal
runs and summation on the host stay with the FPGA receiver.

No module waits for another and nothing waits for the consumer. Images are held in a bounded pool of slots
(no ring, so a slow image pins nothing behind it). An image finding no free slot is dropped whole and counted
in images_dropped and in the efficiency. A module or a card behind the fastest by the device's own threshold
is taken as missing, and its late frames go back to the card.

What Lite assumed of a DECTRIS stream, now changed: ApplyStartMessage no longer applies to a PSI detector (it
forced pixel_mask_applied false). Calibration goes out right after the start message, as a calibration
message could reach the writer after the first image and be refused. Efficiency counts images expected and
each image's collection efficiency, not frames. The ADU histogram comes through the end message. An image in
raw module layout is forwarded without analysis, as the preprocessor needs the assembled geometry.

Measured against the FPGA receiver on the HLS simulation, JUNGFRAU and PSI EIGER, conversion and raw: pixels
and source metadata identical.

Open:
- The azimuthal profile of a PSI EIGER differs between the card and the host (cosine 0.91 on synthetic frames,
  0.999996 on JUNGFRAU int16). Cause not known yet.
- Unsigned images: the card writes saturated and error pixels as the same marker, which the host preprocessor
  counts as error (saturated 8 to 0, error 8 to 16 on the same frames).
- Not run on a real card or the broker. Missing: per-module packet statistics, plots of empty images, the
  card's delay in the messages. Above the card's summation limit the run falls back to the FPGA receiver.
- JFJochReceiverRugnux: forwarding tested, analysis results not. It has no raw layout guard, its efficiency
  still divides by frames, and it does not forward the ADU histogram.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

image_puller/ImagePuller.h

@@ -7,6 +7,7 @@
 #include <chrono>
 #include <string>
 #include <memory>
+#include <vector>
 
 #include "../common/ZMQWrappers.h"
 #include "../frame_serialize/CBORStream2Deserializer.h"
@@ -81,6 +82,12 @@ public:
     virtual bool SendAck(const PullerAckMessage &) { return false; }
 
     virtual void Disconnect() = 0;
+
+    // To go to the writer right after the start message, before any image: what a detector stream sends
+    // as calibration messages can reach the writer after the first image, where it is refused.
+    virtual std::vector<CompressedImage> GetCalibration() const { return {}; }
+    // Images the puller had no room for, which the receiver therefore never gets.
+    virtual size_t GetDroppedImages() const { return 0; }
 };
 
 

receiver/CMakeLists.txt

@@ -11,6 +11,7 @@ ADD_LIBRARY(JFJochReceiver STATIC
         JFJochReceiverCurrentStatus.h
         JFJochReceiverOutput.h
         JFJochReceiverLite.h JFJochReceiverLite.cpp
+        FPGAImagePuller.h FPGAImagePuller.cpp
         JFJochReceiverRugnux.h JFJochReceiverRugnux.cpp
         StillsPrePass.h StillsPrePass.cpp
         StillsGPUAnalysis.h StillsGPUAnalysis.cpp

receiver/FPGAImagePuller.cpp

@@ -0,0 +1,344 @@
+// SPDX-FileCopyrightText: 2026 Filip Leonarski, Paul Scherrer Institute <filip.leonarski@psi.ch>
+// SPDX-License-Identifier: GPL-3.0-only
+
+#include "FPGAImagePuller.h"
+
+#include <thread>
+#include <utility>
+
+#include "JFJochCompressor.h"
+
+FPGAImagePuller::FPGAImagePuller(const DiffractionExperiment &in_experiment, const PixelMask &mask,
+                                 const JFCalibration *calibration, AcquisitionDeviceGroup &in_devices,
+                                 Logger &in_logger, int64_t compression_threads, size_t in_max_slots)
+    : ImagePuller(in_max_slots + 2), // every slot, the start and the end message: Put never finds it full
+      experiment(in_experiment),
+      logger(in_logger),
+      devices(in_devices),
+      max_slots(in_max_slots),
+      image_dropped(experiment.GetImageNum(), 0) {
+    const int ndatastreams = experiment.GetDataStreamsNum();
+    if (devices.size() < ndatastreams)
+        throw JFJochException(JFJochExceptionCategory::InputParameterInvalid,
+                              "Number of acquisition devices has to match data streams");
+
+    for (int m = 0; m < experiment.GetModulesNum(); m++)
+        adu_histogram.emplace_back(std::make_unique<ADUHistogram>());
+
+    for (int d = 0; d < ndatastreams; d++) {
+        devices[d].PrepareAction(experiment);
+        expected_packets_per_image += devices[d].Counters().GetExpectedPacketsPerImage();
+    }
+
+    if (experiment.GetDetectorType() != DetectorType::JUNGFRAU)
+        calibration = nullptr;
+    if (calibration != nullptr && experiment.GetSaveCalibration())
+        PrepareCalibration(*calibration);
+
+    // Making a slot takes tens of milliseconds on a large detector, under the lock all modules go through:
+    // the first ones are ready when the frames start. The rest only if the receiver falls behind.
+    for (size_t i = 0; i < std::min<size_t>(max_slots, InitialSlots); i++) {
+        free_slots.push_back(std::make_shared<Slot>(experiment));
+        slots_created++;
+    }
+
+    // The receiver's own start message is built from its experiment; this one only says the run has begun.
+    StartMessage start{};
+    experiment.FillMessage(start);
+    outside_fifo.Put({.cbor = std::make_shared<CBORStream2DeserializerOutput>(start),
+                      .received = std::chrono::steady_clock::now()});
+
+    for (int i = 0; i < compression_threads; i++)
+        compress_futures.emplace_back(std::async(std::launch::async, &FPGAImagePuller::CompressImages, this));
+
+    for (int d = 0; d < ndatastreams; d++)
+        for (int m = 0; m < experiment.GetModulesNum(d); m++)
+            module_futures.emplace_back(std::async(std::launch::async, &FPGAImagePuller::ReadModule, this, d, m));
+
+    std::latch ready(ndatastreams);
+    for (int d = 0; d < ndatastreams; d++)
+        acquire_futures.emplace_back(std::async(std::launch::async, &FPGAImagePuller::Acquire, this,
+                                                static_cast<uint16_t>(d), std::cref(mask), calibration,
+                                                std::ref(ready)));
+    ready.wait();
+
+    finish_future = std::async(std::launch::async, &FPGAImagePuller::Finish, this);
+
+    if (failed) {
+        try {
+            Stop();
+        } catch (...) {
+            outside_fifo.Clear(); // the destructor does not run
+            throw;
+        }
+    }
+}
+
+FPGAImagePuller::~FPGAImagePuller() {
+    try {
+        Stop();
+    } catch (const std::exception &e) {
+        logger.Error("FPGA image puller teardown failed: {}", e.what());
+    }
+    // These hold slots, which go back to this object
+    outside_fifo.Clear();
+}
+
+void FPGAImagePuller::SetError(std::exception_ptr e) {
+    std::unique_lock ul(error_mutex);
+    if (!error)
+        error = e;
+    failed = true;
+}
+
+void FPGAImagePuller::Cancel() {
+    for (int d = 0; d < experiment.GetDataStreamsNum(); d++)
+        devices[d].Cancel();
+}
+
+void FPGAImagePuller::Stop() {
+    Cancel();
+    std::call_once(finish_once, [this] { finish_future.get(); });
+    std::unique_lock ul(error_mutex);
+    if (error)
+        std::rethrow_exception(std::exchange(error, nullptr));
+}
+
+void FPGAImagePuller::PrepareCalibration(const JFCalibration &calibration) {
+    JFJochBitShuffleCompressor compressor(CompressionAlgorithm::BSHUF_LZ4);
+
+    const auto add = [&](const std::string &channel, std::vector<uint8_t> v) {
+        calibration_data.emplace_back(std::move(v));
+        calibration_images.emplace_back(calibration_data.back().data(), calibration_data.back().size(),
+                                        RAW_MODULE_COLS, experiment.GetModulesNum() * RAW_MODULE_LINES,
+                                        CompressedImageMode::Uint16, CompressionAlgorithm::BSHUF_LZ4, channel);
+    };
+    for (int sc = 0; sc < experiment.GetStorageCellNumber(); sc++) {
+        for (int gain = 0; gain < 3; gain++) {
+            if (experiment.IsFixedGainG1() && (gain != 1))
+                continue;
+            const std::string suffix = (experiment.GetStorageCellNumber() > 1)
+                                           ? fmt::format("_g{:d}_sc{:d}", gain, sc)
+                                           : fmt::format("_g{:d}", gain);
+            add("pedestal" + suffix, compressor.Compress(calibration.GetPedestal(gain, sc)));
+            add("pedestal_rms" + suffix, compressor.Compress(calibration.GetPedestalRMS(gain, sc)));
+        }
+    }
+}
+
+void FPGAImagePuller::Acquire(uint16_t data_stream, const PixelMask &mask, const JFCalibration *calibration,
+                              std::latch &ready) {
+    auto &device = devices[data_stream];
+    try {
+        // Only what the card needs for the pixels: its analysis results are not read.
+        if (calibration != nullptr)
+            device.InitializeCalibration(experiment, *calibration);
+        device.InitializePixelMask(experiment, mask);
+        device.StartAction(experiment);
+    } catch (...) {
+        SetError(std::current_exception());
+        Cancel(); // the module threads of the other cards have to come out as well
+        ready.count_down();
+        return;
+    }
+    ready.count_down();
+
+    try {
+        device.WaitForActionComplete();
+    } catch (...) {
+        SetError(std::current_exception());
+        Cancel();
+    }
+}
+
+int64_t FPGAImagePuller::FastestFrame() const {
+    int64_t ret = 0;
+    for (int d = 0; d < experiment.GetDataStreamsNum(); d++)
+        ret = std::max(ret, static_cast<int64_t>(devices[d].Counters().GetFastestFrameNumber()));
+    return ret;
+}
+
+// Polled rather than waited on: a dead card has to be noticed from the other cards' frames, and the
+// counters wake every waiter on every completion.
+void FPGAImagePuller::WaitForFrame(uint16_t data_stream, uint16_t module_number, int64_t frame) const {
+    const auto &counters = devices[data_stream].Counters();
+    // A frame is complete once the module is two further on, as in JFJochReceiverFPGA
+    while ((static_cast<int64_t>(counters.GetCurrFrameNumber(module_number)) < frame + 2)
+           && !counters.IsAcquisitionFinished() && !failed
+           && (FastestFrame() < frame + 2 + MissingAfterFrames))
+        std::this_thread::sleep_for(std::chrono::milliseconds(1));
+}
+
+// The last frame any card got, for a module whose own card has finished. Waits for the other cards to
+// finish too, so every module of the detector stops at the same image.
+int64_t FPGAImagePuller::LastFrame() const {
+    for (int d = 0; d < experiment.GetDataStreamsNum(); d++)
+        while (!devices[d].Counters().IsAcquisitionFinished() && !failed)
+            std::this_thread::sleep_for(std::chrono::milliseconds(1));
+    return FastestFrame();
+}
+
+std::shared_ptr<FPGAImagePuller::Slot> FPGAImagePuller::GetSlot(int64_t image_number) {
+    std::unique_lock ul(images_mutex);
+    if (image_dropped[image_number])
+        return nullptr;
+    if (const auto it = images_in_progress.find(image_number); it != images_in_progress.end())
+        return it->second;
+
+    std::shared_ptr<Slot> slot;
+    if (!free_slots.empty()) {
+        slot = free_slots.back();
+        free_slots.pop_back();
+    } else if (slots_created < max_slots) {
+        slots_created++;
+        slot = std::make_shared<Slot>(experiment);
+    } else {
+        image_dropped[image_number] = 1;
+        dropped_images++;
+        Emit(image_number, std::nullopt);
+        return nullptr;
+    }
+    slot->image_number = image_number;
+    slot->metadata = std::make_unique<ImageMetadata>(experiment);
+    slot->modules_left = static_cast<int>(experiment.GetModulesNum());
+    images_in_progress[image_number] = slot;
+    return slot;
+}
+
+void FPGAImagePuller::ReturnSlot(const std::shared_ptr<Slot> &slot) {
+    std::unique_lock ul(images_mutex);
+    free_slots.push_back(slot);
+}
+
+void FPGAImagePuller::ReadModule(uint16_t data_stream, uint16_t module_number) {
+    try {
+        auto &device = devices[data_stream];
+        const auto &counters = device.Counters();
+        const size_t module_abs = experiment.GetFirstModuleOfDataStream(data_stream) + module_number;
+        int64_t oldest_held = 0;
+
+        for (int64_t image = 0; image < experiment.GetImageNum(); image++) {
+            WaitForFrame(data_stream, module_number, image);
+
+            if (failed)
+                break;
+
+            if (counters.IsAcquisitionFinished()
+                && (image > static_cast<int64_t>(counters.GetFastestFrameNumber()))
+                && (image > LastFrame()))
+                break;
+
+            // A frame given up on can still arrive, late: its buffer goes back to the card.
+            while ((oldest_held < image)
+                   && (oldest_held < static_cast<int64_t>(counters.GetCurrFrameNumber(module_number))))
+                device.FrameBufferRelease(oldest_held++, module_number);
+
+            const bool collected = counters.IsAnyPacketCollected(image, module_number);
+            if (const auto slot = GetSlot(image)) {
+                if (collected) {
+                    const DeviceOutput *output = device.GetDeviceOutput(image, module_number);
+                    slot->metadata->Process(output);
+                    adu_histogram[module_abs]->Add(*output);
+                    slot->transformation.ProcessModule(output, data_stream);
+                } else
+                    slot->transformation.FillNotCollectedModule(module_number, data_stream);
+
+                if (--slot->modules_left == 0) {
+                    {
+                        std::unique_lock ul(images_mutex);
+                        images_in_progress.erase(image);
+                    }
+                    images_to_compress.Put(slot);
+                }
+            }
+            if (collected)
+                device.FrameBufferRelease(image, module_number);
+        }
+    } catch (...) {
+        SetError(std::current_exception());
+        Cancel();
+    }
+}
+
+void FPGAImagePuller::CompressImages() {
+    while (const auto slot = images_to_compress.GetBlocking()) {
+        const int64_t image_number = slot->image_number;
+        try {
+            DataMessage message{};
+            message.number = image_number;
+            message.original_number = image_number;
+            slot->metadata->Export(message, expected_packets_per_image);
+            // The card's spot finder was not set up for this run; the receiver reports its own.
+            message.strong_pixel_count.reset();
+
+            if (message.image_collection_efficiency == 0.0f) { // no module of it came
+                ReturnSlot(slot);
+                Emit(image_number, std::nullopt);
+                continue;
+            }
+
+            message.image = slot->transformation.GetCompressedImage();
+
+            ImagePullerOutput output;
+            output.cbor = std::make_shared<CBORStream2DeserializerOutput>(message);
+            // The image lives in the slot: it goes back when the last copy of this output is gone.
+            output.raw = std::shared_ptr<ReceiveBufferRegion>(
+                new ReceiveBufferRegion{const_cast<uint8_t *>(message.image.GetCompressed()),
+                                        message.image.GetCompressedSize()},
+                [this, slot](ReceiveBufferRegion *region) {
+                    delete region;
+                    ReturnSlot(slot);
+                });
+            Emit(image_number, std::move(output));
+        } catch (...) {
+            SetError(std::current_exception());
+            Cancel();
+            ReturnSlot(slot);
+            Emit(image_number, std::nullopt);
+        }
+    }
+}
+
+// Whichever thread finishes the image next in line lets it, and those waiting behind it, out.
+void FPGAImagePuller::Emit(int64_t image_number, std::optional<ImagePullerOutput> output) {
+    std::unique_lock ul(order_mutex);
+    images_done.emplace(image_number, std::move(output));
+    for (auto it = images_done.begin(); (it != images_done.end()) && (it->first == next_image);
+         it = images_done.erase(it), next_image++) {
+        if (it->second) {
+            it->second->received = std::chrono::steady_clock::now();
+            outside_fifo.Put(std::move(*it->second));
+        }
+    }
+}
+
+void FPGAImagePuller::Finish() {
+    for (auto &f: module_futures)
+        f.get();
+    for (size_t i = 0; i < compress_futures.size(); i++)
+        images_to_compress.Put(nullptr);
+    for (auto &f: compress_futures)
+        f.get();
+
+    Cancel();
+    for (auto &f: acquire_futures)
+        f.get();
+
+    // Only an error leaves images unfinished, holding back those behind them
+    {
+        std::unique_lock ul(order_mutex);
+        for (auto &[n, output]: images_done)
+            if (output)
+                outside_fifo.Put(std::move(*output));
+        images_done.clear();
+    }
+
+    if (dropped_images > 0)
+        logger.Warning("FPGA image puller: {} images dropped, no free slot", dropped_images.load());
+
+    EndMessage end{};
+    for (size_t i = 0; i < adu_histogram.size(); i++)
+        end.adu_histogram["module" + std::to_string(i)] = adu_histogram[i]->GetHistogram();
+    outside_fifo.Put({.cbor = std::make_shared<CBORStream2DeserializerOutput>(end),
+                      .received = std::chrono::steady_clock::now()});
+}

receiver/FPGAImagePuller.h

@@ -0,0 +1,108 @@
+// SPDX-FileCopyrightText: 2026 Filip Leonarski, Paul Scherrer Institute <filip.leonarski@psi.ch>
+// SPDX-License-Identifier: GPL-3.0-only
+
+#pragma once
+
+#include <atomic>
+#include <future>
+#include <latch>
+#include <map>
+#include <memory>
+#include <mutex>
+#include <optional>
+#include <vector>
+
+#include "FrameTransformation.h"
+#include "ImageMetadata.h"
+#include "../acquisition_device/AcquisitionDeviceGroup.h"
+#include "../common/ADUHistogram.h"
+#include "../image_puller/ImagePuller.h"
+#include "../jungfrau/JFCalibration.h"
+
+// The images of a collection from the FPGA cards, as a detector stream for JFJochReceiverLite: start
+// message, images assembled and compressed here in image-number order, end message. The card converts
+// the pixels; its analysis results are not used, the receiver analyses the assembled image.
+//
+// One thread per module takes that module's frames off the card and writes them into the image they
+// belong to, so no module ever waits for another and none waits for the consumer. Images are held in a
+// bounded pool of slots, from the first module to reach them until the receiver is done with them. An
+// image finding no free slot is dropped whole, and its frames go back to the card unread.
+class FPGAImagePuller : public ImagePuller {
+public:
+    static constexpr size_t DefaultSlots = 128;
+    static constexpr size_t InitialSlots = 32;
+    // A module this far behind the fastest one is taken as missing from that frame. It is what the
+    // device itself allows before it starts handing late frames straight back to the card.
+    static constexpr int64_t MissingAfterFrames = AcquisitionCounters::ThresholdFramesLost + 2;
+private:
+    struct Slot {
+        explicit Slot(const DiffractionExperiment &experiment) : transformation(experiment) {}
+        FrameTransformation transformation;
+        std::unique_ptr<ImageMetadata> metadata;
+        std::atomic<int> modules_left = 0;
+        int64_t image_number = 0;
+    };
+
+    const DiffractionExperiment experiment;
+    Logger &logger;
+    AcquisitionDeviceGroup &devices;
+    const size_t max_slots;
+    uint64_t expected_packets_per_image = 0;
+
+    std::mutex images_mutex;
+    std::vector<std::shared_ptr<Slot>> free_slots;
+    size_t slots_created = 0;
+    std::map<int64_t, std::shared_ptr<Slot>> images_in_progress;
+    std::vector<uint8_t> image_dropped;
+    std::atomic<size_t> dropped_images = 0;
+
+    // What is done, waiting for the images before it. No value: that image will not come.
+    std::mutex order_mutex;
+    std::map<int64_t, std::optional<ImagePullerOutput>> images_done;
+    int64_t next_image = 0;
+
+    ThreadSafeFIFO<std::shared_ptr<Slot>> images_to_compress;
+    std::vector<std::unique_ptr<ADUHistogram>> adu_histogram;
+
+    std::vector<std::vector<uint8_t>> calibration_data;
+    std::vector<CompressedImage> calibration_images;
+
+    std::mutex error_mutex;
+    std::exception_ptr error;
+    std::atomic<bool> failed = false; // a failed card may never report finished: nothing waits for it then
+    void SetError(std::exception_ptr e);
+
+    std::vector<std::future<void>> acquire_futures;
+    std::vector<std::future<void>> module_futures;
+    std::vector<std::future<void>> compress_futures;
+    std::future<void> finish_future;
+    std::once_flag finish_once;
+
+    void Acquire(uint16_t data_stream, const PixelMask &mask, const JFCalibration *calibration, std::latch &ready);
+    void ReadModule(uint16_t data_stream, uint16_t module_number);
+    void CompressImages();
+    void Finish();
+
+    void PrepareCalibration(const JFCalibration &calibration);
+    void WaitForFrame(uint16_t data_stream, uint16_t module_number, int64_t frame) const;
+    int64_t FastestFrame() const;
+    int64_t LastFrame() const;
+    std::shared_ptr<Slot> GetSlot(int64_t image_number);
+    void ReturnSlot(const std::shared_ptr<Slot> &slot);
+    void Emit(int64_t image_number, std::optional<ImagePullerOutput> output);
+public:
+    // Arms the cards; when it returns the detector can be started.
+    FPGAImagePuller(const DiffractionExperiment &experiment, const PixelMask &mask, const JFCalibration *calibration,
+                    AcquisitionDeviceGroup &devices, Logger &logger, int64_t compression_threads,
+                    size_t max_slots = DefaultSlots);
+    ~FPGAImagePuller() override;
+
+    void Disconnect() override {}
+    std::vector<CompressedImage> GetCalibration() const override { return calibration_images; }
+    size_t GetDroppedImages() const override { return dropped_images; }
+
+    // The collection is over, whatever the cards have got: what is on its way still comes, then the end message.
+    void Cancel();
+    // After the receiver is done: stops the cards, waits for the threads, throws what went wrong with the cards.
+    void Stop();
+};

receiver/JFJochReceiver.cpp

@@ -100,6 +100,7 @@ JFJochReceiverStatus JFJochReceiver::GetStatus() const {
     ret.images_collected = images_collected;
     ret.images_sent = images_sent;
     ret.images_skipped = images_skipped;
+    ret.images_dropped = GetImagesDropped();
     ret.images_written = image_pusher.GetImagesWritten();
     ret.cancelled = cancelled;
     ret.efficiency = GetEfficiency();

receiver/JFJochReceiver.h

@@ -103,6 +103,7 @@ protected:
     // The last word on the end message before it is sent: a receiver whose analysis is not the
     // per-image one fills in what it found.
     virtual void AddToEndMessage(EndMessage &message) {}
+    virtual uint64_t GetImagesDropped() const { return 0; }
     virtual void Cancel(const JFJochException &e);
 public:
     JFJochReceiver(const DiffractionExperiment& experiment,

receiver/JFJochReceiverCurrentStatus.h

@@ -23,6 +23,7 @@ struct JFJochReceiverStatus {
     uint64_t images_collected;
     uint64_t images_sent;
     uint64_t images_skipped;
+    uint64_t images_dropped = 0; // no room for them before the receiver: they never reached it
     std::optional<uint64_t> images_written;
     bool cancelled;
 };

receiver/JFJochReceiverLite.cpp

@@ -156,6 +156,9 @@ void JFJochReceiverLite::MeasurementThread() {
         start_time = std::chrono::system_clock::now();
         // Send new start message out
         SendStartMessage();
+        if (push_images_to_writer)
+            for (const auto &c: image_puller.GetCalibration())
+                image_pusher.SendCalibration(c);
         measurement_started.count_down();
 
     } catch (const JFJochException &e) {
@@ -203,6 +206,10 @@ void JFJochReceiverLite::MeasurementThread() {
 }
 
 void JFJochReceiverLite::ApplyStartMessage(DiffractionExperiment &experiment, const StartMessage &msg) {
+    // A PSI detector's images are what the experiment set the card up to produce: nothing to learn here.
+    if (experiment.GetDetectorType() != DetectorType::DECTRIS)
+        return;
+
     if ((experiment.GetXPixelsNum() != msg.image_size_x)
     || (experiment.GetYPixelsNum() != msg.image_size_y))
         throw JFJochException(JFJochExceptionCategory::InputParameterInvalid,
@@ -298,8 +305,10 @@ void JFJochReceiverLite::DataAnalysisThread(uint32_t id) {
     measurement_started.wait();
 
     try {
-        analysis = std::make_unique<MXAnalysisWithoutFPGA>(experiment, *az_int_mapping, pixel_mask, indexer,
-                                                           /*enable_fused_adaptive_gpu=*/true);
+        // The analysis works on the assembled image; an image in raw module layout is only forwarded.
+        if (experiment.IsGeometryTransformed())
+            analysis = std::make_unique<MXAnalysisWithoutFPGA>(experiment, *az_int_mapping, pixel_mask, indexer,
+                                                               /*enable_fused_adaptive_gpu=*/true);
     } catch (const JFJochException &e) {
         Cancel(e);
         return;
@@ -314,6 +323,7 @@ void JFJochReceiverLite::DataAnalysisThread(uint32_t id) {
                 continue;
             } else if (msg->cbor->end_message) {
                 // Message is end message
+                end_adu_histogram = msg->cbor->end_message->adu_histogram;
                 end_message_received = true;
                 logger.Debug("Thread {} End message received in JFJochReceiverLite", id);
             } else if (msg->cbor->calibration) {
@@ -325,6 +335,7 @@ void JFJochReceiverLite::DataAnalysisThread(uint32_t id) {
                 DataMessage data_msg = msg->cbor->data_message.value();
 
                 ++images_collected;
+                collection_efficiency_sum += data_msg.image_collection_efficiency.value_or(1.0f);
 
                 auto compressed_size_img = data_msg.image.GetCompressedSize();
                 auto uncompressed_size_img = data_msg.image.GetUncompressedSize();
@@ -337,7 +348,8 @@ void JFJochReceiverLite::DataAnalysisThread(uint32_t id) {
                 auto image_start_time = std::chrono::high_resolution_clock::now();
 
                 AzimuthalIntegrationProfile profile(*az_int_mapping);
-                analysis->Analyze(data_msg, profile, GetSpotFindingSettings());
+                if (analysis)
+                    analysis->Analyze(data_msg, profile, GetSpotFindingSettings());
 
                 auto image_end_time = std::chrono::high_resolution_clock::now();
                 std::chrono::duration<float> image_duration = image_end_time - image_start_time;
@@ -434,16 +446,28 @@ void JFJochReceiverLite::StopReceiver() {
     }
 }
 
+void JFJochReceiverLite::AddToEndMessage(EndMessage &message) {
+    if (!end_adu_histogram.empty())
+        message.adu_histogram = end_adu_histogram;
+}
+
+// With the card summing, an image is many frames. A dark mask run has frames and no images.
+int64_t JFJochReceiverLite::ExpectedImages() const {
+    return (experiment.GetDetectorMode() == DetectorMode::DarkMask) ? experiment.GetFrameNum() : experiment.GetImageNum();
+}
+
 float JFJochReceiverLite::GetEfficiency() const {
-    if (experiment.GetFrameNum() == 0)
+    if (ExpectedImages() == 0)
         return 0;
-    return static_cast<float>(images_collected) / static_cast<float>(experiment.GetFrameNum());
+    if (experiment.GetDetectorMode() == DetectorMode::DarkMask)
+        return static_cast<float>(images_collected) / static_cast<float>(ExpectedImages());
+    return static_cast<float>(collection_efficiency_sum / static_cast<double>(ExpectedImages()));
 }
 
 float JFJochReceiverLite::GetProgress() const {
-    if (experiment.GetFrameNum() == 0)
+    if (ExpectedImages() == 0)
         return 0.0;
-    return static_cast<float>(max_image_number_received) / static_cast<float>(experiment.GetFrameNum());
+    return static_cast<float>(max_image_number_received) / static_cast<float>(ExpectedImages());
 }
 
 void JFJochReceiverLite::SetSpotFindingSettings(const SpotFindingSettings &in_spot_finding_settings) {

receiver/JFJochReceiverLite.h

@@ -53,6 +53,13 @@ class JFJochReceiverLite : public JFJochReceiver {
 
     DarkMaskAnalysis dark_mask_analysis;
 
+    // Over the images received, as a detector stream without it counts each as whole
+    std::atomic<double> collection_efficiency_sum{0.0};
+    std::map<std::string, std::vector<uint64_t>> end_adu_histogram;
+    int64_t ExpectedImages() const;
+    void AddToEndMessage(EndMessage &message) override;
+    uint64_t GetImagesDropped() const override { return image_puller.GetDroppedImages(); }
+
     void MaskThread(uint32_t id);
     void MeasurementThread();
     void DataAnalysisThread(uint32_t id);

receiver/JFJochReceiverRugnux.cpp

@@ -281,6 +281,9 @@ void JFJochReceiverRugnux::MeasurementThread() {
             StartStills();
         else {
             SendStartMessage();
+            if (push_images_to_writer)
+                for (const auto &c: image_puller.GetCalibration())
+                    image_pusher.SendCalibration(c);
 
             // Rugnux reads the images back from the buffer, so it needs one slot per image: a slot taken
             // again would hand it a later image. Without that the collection is only forwarded.

receiver/JFJochReceiverRugnux.h

@@ -86,6 +86,7 @@ class JFJochReceiverRugnux : public JFJochReceiver {
     // the end message goes out with that. Reason: "deadline" or "cancel".
     void Cut(const std::string &reason);
     void AddToEndMessage(EndMessage &message) override;
+    uint64_t GetImagesDropped() const override { return image_puller.GetDroppedImages(); }
 public:
     JFJochReceiverRugnux(const DiffractionExperiment &experiment,
                          const PixelMask &pixel_mask,

receiver/JFJochReceiverService.cpp

@@ -32,6 +32,16 @@ JFJochReceiverService &JFJochReceiverService::RugnuxLive(bool enable, std::share
     return *this;
 }
 
+JFJochReceiverService &JFJochReceiverService::UseFPGAImagePuller(bool enable, size_t slots) {
+    std::unique_lock ul_state(state_mutex);
+    if (state != ReceiverState::Idle)
+        throw JFJochException(JFJochExceptionCategory::WrongDAQState,
+                              "Cannot change the receiver path during data collection");
+    fpga_image_puller = enable;
+    fpga_image_puller_slots = slots;
+    return *this;
+}
+
 RugnuxLiveSnapshot JFJochReceiverService::GetRugnuxStatus() const {
     return rugnux_status.Get();
 }
@@ -52,6 +62,8 @@ void JFJochReceiverService::FinalizeMeasurementChangeState() {
 void JFJochReceiverService::FinalizeMeasurement() {
     try {
         receiver->StopReceiver();
+        if (fpga_puller)
+            fpga_puller->Stop();
     } catch (...) {
         FinalizeMeasurementChangeState();
         throw;
@@ -96,10 +108,27 @@ void JFJochReceiverService::Start(const DiffractionExperiment &experiment,
 
         // First clean-up old measurement
         receiver.reset();
+        if (fpga_puller) {
+            image_puller.reset();
+            fpga_puller.reset();
+        }
 
         preview_image.Configure(experiment, pixel_mask);
 
-        switch (experiment.GetDetectorType()) {
+        // Pedestal runs and host summation stay with JFJochReceiverFPGA
+        auto detector_type = experiment.GetDetectorType();
+        if (fpga_image_puller && (detector_type != DetectorType::DECTRIS) && (experiment.GetImageNum() > 0)) {
+            if (experiment.IsCPUSummation())
+                logger.Warning("Summation above what the card can do needs JFJochReceiverFPGA: FPGA image puller not used");
+            else {
+                fpga_puller = std::make_shared<FPGAImagePuller>(experiment, pixel_mask, calibration, aq_devices, logger,
+                                                                nthreads_local, fpga_image_puller_slots);
+                puller = fpga_puller;
+                detector_type = DetectorType::DECTRIS; // from here on, one stream receiver input like any other
+            }
+        }
+
+        switch (detector_type) {
             case DetectorType::EIGER:
             case DetectorType::JUNGFRAU:
                 receiver = std::make_unique<JFJochReceiverFPGA>(experiment, pixel_mask,
@@ -155,6 +184,10 @@ void JFJochReceiverService::Start(const DiffractionExperiment &experiment,
         // until the next start overwrites it.
         receiver_status.Clear();
         receiver_status.SetProgress({});
+        if (fpga_puller) {
+            image_puller.reset();
+            fpga_puller.reset();
+        }
         logger.ErrorException(e);
         throw;
     }
@@ -162,8 +195,11 @@ void JFJochReceiverService::Start(const DiffractionExperiment &experiment,
 
 void JFJochReceiverService::Cancel(bool silent) {
     std::unique_lock ul(state_mutex);
-    if (state == ReceiverState::Running)
+    if (state == ReceiverState::Running) {
         receiver->Cancel(silent);
+        if (fpga_puller)
+            fpga_puller->Cancel();
+    }
 }
 
 JFJochReceiverOutput JFJochReceiverService::Stop() {

receiver/JFJochReceiverService.h

@@ -9,6 +9,7 @@
 #include <memory>
 
 #include "JFJochReceiver.h"
+#include "FPGAImagePuller.h"
 #include "../acquisition_device/AcquisitionDeviceGroup.h"
 
 #include "../preview/ZMQMetadataSocket.h"
@@ -33,6 +34,12 @@ class JFJochReceiverService {
     double rugnux_deadline_s = 60;   // the analysis is cut this long after the last image (0 = never)
     RugnuxLiveStatus rugnux_status;   // of the last collection, kept after it ends
 
+    // A PSI detector's frames go through FPGAImagePuller and the stream receivers, instead of
+    // JFJochReceiverFPGA. Off unless asked for.
+    bool fpga_image_puller = false;
+    size_t fpga_image_puller_slots = FPGAImagePuller::DefaultSlots;
+    std::shared_ptr<FPGAImagePuller> fpga_puller;
+
     enum class ReceiverState {Idle, Running};
     ReceiverState state = ReceiverState::Idle;
     mutable std::mutex state_mutex;
@@ -58,6 +65,7 @@ public:
     JFJochReceiverService& RugnuxLive(bool enable, std::shared_ptr<const RugnuxModel> model, int reserved_cpus,
                                       double deadline_s);
     RugnuxLiveSnapshot GetRugnuxStatus() const;
+    JFJochReceiverService& UseFPGAImagePuller(bool enable, size_t slots = FPGAImagePuller::DefaultSlots);
     JFJochReceiverService& PreviewSocket(const std::string &addr, const std::optional<int32_t> &watermark = {});
     JFJochReceiverService& MetadataSocket(const std::string &addr);
 

tests/CMakeLists.txt

@@ -150,6 +150,8 @@ ADD_EXECUTABLE(jfjoch_test
         ImageMetadataTest.cpp
         JFJochReceiverLiteTest.cpp
         JFJochReceiverRugnuxTest.cpp
+        FPGAImagePullerTest.cpp
+        FPGAImagePullerPerfTest.cpp
         GridScanSettingsTest.cpp
         JFJochReceiverPlotsTest.cpp
         GoniometerAxisTest.cpp

tests/FPGAImagePullerPerfTest.cpp

@@ -0,0 +1,412 @@
+// SPDX-FileCopyrightText: 2026 Filip Leonarski, Paul Scherrer Institute <filip.leonarski@psi.ch>
+// SPDX-License-Identifier: GPL-3.0-only
+
+// Host-side measurements of FPGAImagePuller without the HLS simulation: a device that hands out already
+// converted module frames at a chosen rate. Hidden: run by hand, with FPGA_PULLER_* in the environment.
+
+#include <catch2/catch_all.hpp>
+
+#include <sys/resource.h>
+
+#include <cstdlib>
+#include <functional>
+#include <iostream>
+#include <random>
+#include <thread>
+
+#include "../acquisition_device/AcquisitionDevice.h"
+#include "../compression/JFJochDecompress.h"
+#include "../frame_serialize/CBORStream2Deserializer.h"
+#include "../image_pusher/ImagePusher.h"
+#include "../receiver/FPGAImagePuller.h"
+#include "../receiver/JFJochReceiverService.h"
+
+namespace {
+    int64_t Env(const char *name, int64_t def) {
+        const char *v = std::getenv(name);
+        return v ? std::atoll(v) : def;
+    }
+
+    // Converted frame of a module: sparse Poisson photons plus a few spots
+    std::vector<int16_t> PhotonFrame(uint32_t seed, double photons_per_pixel) {
+        std::vector<int16_t> ret(RAW_MODULE_SIZE);
+        std::mt19937 g(seed);
+        std::poisson_distribution<int> p(photons_per_pixel);
+        for (auto &v: ret)
+            v = static_cast<int16_t>(p(g));
+        std::uniform_int_distribution<int> where(0, RAW_MODULE_SIZE - 1);
+        for (int i = 0; i < 2000; i++)
+            ret[where(g)] = 500;
+        return ret;
+    }
+
+    // Like the card, it writes the frame into a buffer it was given and reports the handle; with no buffer
+    // given back in time the frame is lost. The pixels are put in the buffers beforehand, so that the
+    // device itself costs nothing: the rate is limited by what is measured, not by this thread.
+    class ReplayDevice : public AcquisitionDevice {
+        std::vector<std::unique_ptr<uint8_t[]>> storage;
+        std::thread generator;
+        std::atomic<bool> stop{false};
+        const int64_t rate_hz;
+        int64_t nimages = 0, nmodules = 0;
+        std::function<bool(int64_t, int)> present;
+
+        void HW_RunInternalGenerator(const FrameGeneratorConfig &) override {}
+        void HW_SetSpotFinderParameters(const SpotFinderParameters &) override {}
+
+        void Run() {
+            const auto t0 = std::chrono::steady_clock::now();
+            for (int64_t f = 0; (f < nimages) && !stop; f++) {
+                std::this_thread::sleep_until(t0 + std::chrono::nanoseconds(f * 1000000000 / rate_hz));
+                for (int m = 0; m < nmodules; m++) {
+                    if (!present(f, m))
+                        continue;
+                    WorkRequest wr{};
+                    if (work_request_queue.Get(wr) == 0) {
+                        lost++;
+                        continue;
+                    }
+                    auto &s = GetDeviceOutput(static_cast<size_t>(wr.handle))->module_statistics;
+                    s.frame_number = f;
+                    s.module_number = m;
+                    s.packet_count = 512;
+                    s.timestamp = f * 100;
+                    s.exptime = 100;
+                    Completion c{};
+                    c.type = Completion::Type::Image;
+                    c.handle = wr.handle;
+                    work_completion_queue.Put(c);
+                }
+            }
+            Completion c{};
+            c.type = Completion::Type::End;
+            work_completion_queue.Put(c);
+        }
+
+        void Start(const DiffractionExperiment &x, uint32_t) override {
+            if (generator.joinable())
+                generator.join();
+            nimages = x.GetImageNum();
+            nmodules = x.GetModulesNum(data_stream);
+            stop = false;
+            lost = 0;
+            Completion c{};
+            c.type = Completion::Type::Start;
+            work_completion_queue.Put(c);
+            generator = std::thread(&ReplayDevice::Run, this);
+        }
+    public:
+        std::atomic<int64_t> lost{0};
+
+        ReplayDevice(uint16_t stream, size_t nbuffers, int64_t in_rate_hz, double photons_per_pixel,
+                     std::function<bool(int64_t, int)> in_present = [](int64_t, int) { return true; })
+            : AcquisitionDevice(stream), rate_hz(in_rate_hz), present(std::move(in_present)) {
+            max_modules = 32;
+            std::vector<std::vector<int16_t>> patterns;
+            for (int i = 0; i < 16; i++)
+                patterns.push_back(PhotonFrame(i + 1, photons_per_pixel));
+            for (size_t i = 0; i < nbuffers; i++) {
+                storage.emplace_back(new uint8_t[FPGA_BUFFER_LOCATION_SIZE]);
+                memset(storage.back().get(), 0, FPGA_BUFFER_LOCATION_SIZE);
+                buffer_device.push_back(reinterpret_cast<DeviceOutput *>(storage.back().get()));
+                memcpy(buffer_device.back()->pixels, patterns[i % patterns.size()].data(), RAW_MODULE_SIZE * 2);
+            }
+        }
+        ~ReplayDevice() override {
+            stop = true;
+            if (generator.joinable())
+                generator.join();
+        }
+        void Cancel() override { stop = true; }
+        uint32_t GetExpectedDescriptorsPerModule() const override { return 1; }
+    };
+
+    double CpuSeconds() {
+        rusage r{};
+        getrusage(RUSAGE_SELF, &r);
+        return static_cast<double>(r.ru_utime.tv_sec + r.ru_stime.tv_sec)
+               + static_cast<double>(r.ru_utime.tv_usec + r.ru_stime.tv_usec) * 1e-6;
+    }
+
+    DiffractionExperiment Detector(int64_t modules, int64_t images, CompressionAlgorithm algorithm) {
+        DiffractionExperiment x(DetJF(modules));
+        x.Conversion();
+        x.PedestalG0Frames(0).NumTriggers(1).UseInternalPacketGenerator(false).ImagesPerTrigger(images)
+                .ImagesPerFile(1000).IncidentEnergy_keV(12.4).DetectorDistance_mm(75)
+                .BeamX_pxl(500).BeamY_pxl(600).Compression(algorithm);
+        x.DataStreams(1);
+        return x;
+    }
+
+    CompressionAlgorithm Algorithm() {
+        switch (Env("FPGA_PULLER_ALG", 0)) {
+            case 1: return CompressionAlgorithm::BSHUF_ZSTD;
+            case 2: return CompressionAlgorithm::NO_COMPRESSION;
+            default: return CompressionAlgorithm::BSHUF_LZ4;
+        }
+    }
+
+    class TimingPusher : public ImagePusher {
+    public:
+        std::mutex m;
+        std::vector<float> decode_s, preprocess_s;
+        void StartDataCollection(StartMessage &) override {}
+        bool EndDataCollection(const EndMessage &) override { return true; }
+        bool SendCalibration(const CompressedImage &) override { return true; }
+        bool SendImage(const uint8_t *data, size_t size, int64_t) override {
+            auto d = CBORStream2Deserialize(data, size);
+            std::unique_lock ul(m);
+            decode_s.push_back(d->data_message->compression_time_s.value_or(NAN));
+            preprocess_s.push_back(d->data_message->preprocessing_time_s.value_or(NAN));
+            return true;
+        }
+        std::string PrintSetup() const override { return "timing"; }
+        ImagePusherType GetType() const override { return ImagePusherType::Test; }
+    };
+
+    float Median(std::vector<float> v) {
+        if (v.empty())
+            return NAN;
+        std::sort(v.begin(), v.end());
+        return v[v.size() / 2];
+    }
+}
+
+// One thread, one image: what assembly and compression cost per image.
+TEST_CASE("FPGAImagePullerPerf_SingleThread", "[.FPGAImagePullerPerf]") {
+    for (const double photons: {0.1, 1.0}) {
+        for (const int64_t modules: {2, 18}) {
+            for (const auto alg: {CompressionAlgorithm::BSHUF_LZ4, CompressionAlgorithm::BSHUF_ZSTD}) {
+                auto x = Detector(modules, 1, alg);
+                std::vector<std::vector<int16_t>> frames;
+                for (int64_t m = 0; m < modules; m++)
+                    frames.push_back(PhotonFrame(m + 1, photons));
+
+                FrameTransformation t(x);
+                std::vector<uint8_t> out(MaxCompressedSize(alg, x.GetPixelsNum(), 2));
+                const int rep = 10;
+                double assemble = 0, compress = 0, decompress = 0;
+                size_t size = 0;
+                std::vector<uint8_t> decoded;
+                for (int r = 0; r < rep; r++) {
+                    const auto t0 = std::chrono::steady_clock::now();
+                    for (int64_t m = 0; m < modules; m++)
+                        t.ProcessModule(frames[m].data(), m, 0);
+                    const auto t1 = std::chrono::steady_clock::now();
+                    size = t.CompressImage(out.data(), out.size());
+                    const auto t2 = std::chrono::steady_clock::now();
+                    CompressedImage(out.data(), size, x.GetXPixelsNum(), x.GetYPixelsNum(), x.GetImageMode(), alg)
+                            .GetUncompressed(decoded);
+                    const auto t3 = std::chrono::steady_clock::now();
+                    assemble += std::chrono::duration<double>(t1 - t0).count();
+                    compress += std::chrono::duration<double>(t2 - t1).count();
+                    decompress += std::chrono::duration<double>(t3 - t2).count();
+                }
+                std::cout << "single thread: " << modules << " modules, " << photons << " ph/pixel, "
+                          << (alg == CompressionAlgorithm::BSHUF_LZ4 ? "BSHUF_LZ4" : "BSHUF_ZSTD")
+                          << ": raw " << x.GetPixelsNum() * 2 / 1000000.0 << " MB -> " << size / 1000000.0 << " MB"
+                          << "; assemble " << assemble / rep * 1e3 << " ms, compress " << compress / rep * 1e3
+                          << " ms, host decompress " << decompress / rep * 1e3 << " ms" << std::endl;
+            }
+        }
+    }
+}
+
+// The whole puller at a rate, the consumer letting go of each image at once.
+TEST_CASE("FPGAImagePullerPerf_Rate", "[.FPGAImagePullerPerf]") {
+    const int64_t modules = Env("FPGA_PULLER_MODULES", 2);
+    const int64_t rate = Env("FPGA_PULLER_RATE", 100);
+    const int64_t images = Env("FPGA_PULLER_IMAGES", 500);
+    const int64_t threads = Env("FPGA_PULLER_THREADS", 8);
+    const double photons = static_cast<double>(Env("FPGA_PULLER_MILLIPHOTONS", 100)) * 1e-3;
+
+    auto x = Detector(modules, images, Algorithm());
+    AcquisitionDeviceGroup devices;
+    auto device = std::make_unique<ReplayDevice>(0, 512, rate, photons);
+    auto *replay = device.get();
+    devices.Add(std::move(device));
+
+    Logger logger("FPGAImagePullerPerf");
+    PixelMask mask(x);
+    const double cpu0 = CpuSeconds();
+    const auto t0 = std::chrono::steady_clock::now();
+    FPGAImagePuller puller(x, mask, nullptr, devices, logger, threads);
+
+    int64_t received = 0, last = -1;
+    bool in_order = true, end = false;
+    size_t compressed = 0;
+    while (!end) {
+        const auto m = puller.PollImage(std::chrono::milliseconds(10));
+        if (!m)
+            continue;
+        if (m->cbor->data_message) {
+            received++;
+            in_order = in_order && (m->cbor->data_message->number > last);
+            last = m->cbor->data_message->number;
+            compressed += m->cbor->data_message->image.GetCompressedSize();
+        }
+        end = m->cbor->end_message.has_value();
+    }
+    const double wall = std::chrono::duration<double>(std::chrono::steady_clock::now() - t0).count();
+    puller.Stop();
+    const double cpu = CpuSeconds() - cpu0;
+    std::cout << "rate: " << modules << " modules, " << rate << " Hz asked, " << threads << " compression threads, "
+              << photons << " ph/pixel: " << received << "/" << images << " images in " << wall << " s = "
+              << received / wall << " Hz, dropped " << puller.GetDroppedImages() << ", lost at the device "
+              << replay->lost.load() << " module frames, in order " << in_order << ", "
+              << compressed / std::max<int64_t>(received, 1) / 1000000.0 << " MB/image, CPU " << cpu / wall
+              << " cores" << std::endl;
+}
+
+// Through Lite, on whatever device the analysis picks: what decoding (or just getting the image across) costs.
+TEST_CASE("FPGAImagePullerPerf_Lite", "[.FPGAImagePullerPerf]") {
+    const int64_t modules = Env("FPGA_PULLER_MODULES", 2);
+    const int64_t rate = Env("FPGA_PULLER_RATE", 20);
+    const int64_t images = Env("FPGA_PULLER_IMAGES", 100);
+    const int64_t threads = Env("FPGA_PULLER_THREADS", 2);
+    const double photons = static_cast<double>(Env("FPGA_PULLER_MILLIPHOTONS", 100)) * 1e-3;
+
+    auto x = Detector(modules, images, Algorithm());
+    AcquisitionDeviceGroup devices;
+    devices.Add(std::make_unique<ReplayDevice>(0, 512, rate, photons));
+
+    Logger logger("FPGAImagePullerPerf");
+    TimingPusher pusher;
+    JFJochReceiverService service(devices, logger, pusher, 2048);
+    service.NumThreads(threads);
+    SpotFindingSettings s = DiffractionExperiment::DefaultDataProcessingSettings();
+    s.enable = false;
+    s.indexing = false;
+    service.SetSpotFindingSettings(s);
+    service.UseFPGAImagePuller(true);
+
+    PixelMask mask(x);
+    service.Start(x, mask, nullptr);
+    const auto out = service.Stop();
+    std::cout << "lite: " << modules << " modules, " << rate << " Hz, " << threads << " threads, algorithm "
+              << Env("FPGA_PULLER_ALG", 0) << ", " << photons << " ph/pixel: collected " << out.status.images_collected
+              << " dropped " << out.status.images_dropped << " sent " << out.status.images_sent
+              << "; ratio " << out.status.compressed_ratio.value_or(0)
+              << "; median compression_time_s (decode on the analysis device) " << Median(pusher.decode_s) * 1e3
+              << " ms, preprocessing_time_s (rest of getting it into the preprocessed buffer) "
+              << Median(pusher.preprocess_s) * 1e3 << " ms" << std::endl;
+}
+
+namespace {
+    struct Stream {
+        std::vector<int64_t> numbers;
+        std::vector<float> efficiency;
+        bool start_first = false, end = false;
+    };
+
+    // Reads until the end message, or until the deadline when `until_image` has not come by then
+    Stream ReadAll(FPGAImagePuller &puller) {
+        Stream ret;
+        bool first = true;
+        const auto t0 = std::chrono::steady_clock::now();
+        while (!ret.end && (std::chrono::steady_clock::now() - t0 < std::chrono::seconds(120))) {
+            const auto m = puller.PollImage(std::chrono::milliseconds(10));
+            if (!m)
+                continue;
+            if (first)
+                ret.start_first = m->cbor->start_message.has_value();
+            first = false;
+            if (m->cbor->data_message) {
+                ret.numbers.push_back(m->cbor->data_message->number);
+                ret.efficiency.push_back(m->cbor->data_message->image_collection_efficiency.value_or(-1));
+            }
+            ret.end = m->cbor->end_message.has_value();
+        }
+        return ret;
+    }
+
+    DiffractionExperiment TwoCards(int64_t images) {
+        auto x = Detector(4, images, CompressionAlgorithm::BSHUF_LZ4);
+        x.DataStreams(2);
+        return x;
+    }
+}
+
+// Replay rather than simulation: the HLS simulation keeps its state in statics, so only one card at a time.
+TEST_CASE("FPGAImagePuller_Replay_TwoCards_InOrder", "[FPGAImagePuller]") {
+    const auto x = TwoCards(300);
+    REQUIRE(x.GetDataStreamsNum() == 2);
+    AcquisitionDeviceGroup devices;
+    devices.Add(std::make_unique<ReplayDevice>(0, 64, 500, 0.1));
+    devices.Add(std::make_unique<ReplayDevice>(1, 64, 500, 0.1));
+
+    Logger logger("FPGAImagePuller_TwoCards");
+    PixelMask mask(x);
+    FPGAImagePuller puller(x, mask, nullptr, devices, logger, 8);
+    const auto s = ReadAll(puller);
+    REQUIRE_NOTHROW(puller.Stop());
+
+    CHECK(s.start_first);
+    CHECK(s.end);
+    REQUIRE(s.numbers.size() == 300);
+    for (int i = 0; i < 300; i++)
+        CHECK(s.numbers[i] == i);
+    CHECK(puller.GetDroppedImages() == 0);
+}
+
+// A module that dies: nobody stops the run, the images behind it keep coming in order, with the module
+// taken as missing once the others are far enough ahead.
+TEST_CASE("FPGAImagePuller_Replay_DeadModule", "[FPGAImagePuller]") {
+    const auto x = Detector(2, 300, CompressionAlgorithm::BSHUF_LZ4);
+    AcquisitionDeviceGroup devices;
+    devices.Add(std::make_unique<ReplayDevice>(0, 64, 200, 0.1, [](int64_t f, int m) { return (m == 0) || (f < 100); }));
+
+    Logger logger("FPGAImagePuller_DeadModule");
+    PixelMask mask(x);
+    FPGAImagePuller puller(x, mask, nullptr, devices, logger, 4);
+    const auto s = ReadAll(puller);
+    REQUIRE_NOTHROW(puller.Stop());
+
+    CHECK(s.end);
+    REQUIRE(s.numbers.size() == 300);
+    for (int i = 0; i < 300; i++) {
+        CHECK(s.numbers[i] == i);
+        CHECK(s.efficiency[i] == ((i < 100) ? 1.0f : 0.5f));
+    }
+    CHECK(puller.GetDroppedImages() == 0);
+}
+
+// A whole card dies, the other one goes on: as above, with no module of that card ever reporting again,
+// until the card is told to stop.
+TEST_CASE("FPGAImagePuller_Replay_DeadCard", "[FPGAImagePuller]") {
+    const auto x = TwoCards(200);
+    AcquisitionDeviceGroup devices;
+    devices.Add(std::make_unique<ReplayDevice>(0, 64, 200, 0.1));
+    devices.Add(std::make_unique<ReplayDevice>(1, 64, 200, 0.1, [](int64_t f, int) { return f < 20; }));
+
+    Logger logger("FPGAImagePuller_DeadCard");
+    PixelMask mask(x);
+    FPGAImagePuller puller(x, mask, nullptr, devices, logger, 4);
+    const auto s = ReadAll(puller);
+    REQUIRE_NOTHROW(puller.Stop());
+
+    CHECK(s.end);
+    REQUIRE(s.numbers.size() == 200);
+    for (int i = 0; i < 200; i++)
+        CHECK(s.numbers[i] == i);
+    CHECK(s.efficiency[150] == 0.5f);
+    CHECK(puller.GetDroppedImages() == 0);
+}
+
+TEST_CASE("FPGAImagePuller_Replay_9M", "[FPGAImagePuller]") {
+    const auto x = Detector(18, 100, CompressionAlgorithm::BSHUF_LZ4);
+    AcquisitionDeviceGroup devices;
+    auto device = std::make_unique<ReplayDevice>(0, 512, 100, 0.1);
+    auto *replay = device.get();
+    devices.Add(std::move(device));
+
+    Logger logger("FPGAImagePuller_FewBuffers");
+    PixelMask mask(x);
+    FPGAImagePuller puller(x, mask, nullptr, devices, logger, 8);
+    const auto s = ReadAll(puller);
+    REQUIRE_NOTHROW(puller.Stop());
+    std::cout << "9M replay at 100 Hz: " << s.numbers.size() << " images, dropped " << puller.GetDroppedImages()
+              << ", module frames lost at the device " << replay->lost.load() << std::endl;
+    CHECK(s.end);
+    CHECK(s.numbers.size() + puller.GetDroppedImages() == 100);
+}

tests/FPGAImagePullerTest.cpp

@@ -0,0 +1,705 @@
+// SPDX-FileCopyrightText: 2026 Filip Leonarski, Paul Scherrer Institute <filip.leonarski@psi.ch>
+// SPDX-License-Identifier: GPL-3.0-only
+
+// The same frames through JFJochReceiverFPGA and through FPGAImagePuller + JFJochReceiverLite, on the HLS
+// simulation of the card. The pixels must be the same; the analysis differs by design, and its numbers are printed.
+
+#include <catch2/catch_all.hpp>
+
+#include <cmath>
+#include <iostream>
+#include <map>
+#include <mutex>
+#include <random>
+#include <thread>
+
+#include "../acquisition_device/HLSSimulatedDevice.h"
+#include "../compression/JFJochDecompress.h"
+#include "../frame_serialize/CBORStream2Deserializer.h"
+#include "../image_pusher/HDF5FilePusher.h"
+#include "../image_pusher/ImagePusher.h"
+#include "../writer/HDF5Objects.h"
+#include "../writer/StreamWriter.h"
+#include "../receiver/FPGAImagePuller.h"
+#include "../receiver/JFJochReceiverService.h"
+#include "../rugnux/RugnuxDefaults.h"
+
+using namespace std::chrono_literals;
+
+namespace {
+    struct Received {
+        DataMessage message;
+        std::vector<uint8_t> compressed;
+    };
+
+    class CollectingPusher : public ImagePusher {
+    public:
+        std::mutex m;
+        std::map<int64_t, Received> images;
+        std::vector<int64_t> arrival;
+        std::vector<std::string> sequence;
+        EndMessage end;
+        std::chrono::milliseconds delay{0};
+
+        void StartDataCollection(StartMessage &) override {
+            std::unique_lock ul(m);
+            sequence.emplace_back("start");
+        }
+        bool EndDataCollection(const EndMessage &message) override {
+            std::unique_lock ul(m);
+            sequence.emplace_back("end");
+            end = message;
+            return true;
+        }
+        bool SendCalibration(const CompressedImage &c) override {
+            std::unique_lock ul(m);
+            sequence.emplace_back("calibration " + c.GetChannel());
+            return true;
+        }
+        bool SendImage(const uint8_t *image_data, size_t image_size, int64_t image_number) override {
+            std::this_thread::sleep_for(delay);
+            auto d = CBORStream2Deserialize(image_data, image_size);
+            Received r;
+            r.message = d->data_message.value();
+            const uint8_t *p = r.message.image.GetCompressed();
+            r.compressed.assign(p, p + r.message.image.GetCompressedSize());
+            r.message.image = CompressedImage(r.compressed.data(), r.compressed.size(), r.message.image.GetWidth(),
+                                              r.message.image.GetHeight(), r.message.image.GetMode(),
+                                              r.message.image.GetCompressionAlgorithm());
+            std::unique_lock ul(m);
+            sequence.emplace_back("image");
+            arrival.push_back(image_number);
+            images.emplace(image_number, std::move(r));
+            return true;
+        }
+        std::string PrintSetup() const override { return "collecting"; }
+        ImagePusherType GetType() const override { return ImagePusherType::Test; }
+    };
+
+    float Pedestal(int module, int pixel) {
+        return 3000 + 10 * module + pixel % 50;
+    }
+
+    JFCalibration Calibration(const DiffractionExperiment &x) {
+        JFCalibration ret(x);
+        std::vector<float> g0(RAW_MODULE_SIZE), g1(RAW_MODULE_SIZE), g2(RAW_MODULE_SIZE);
+        for (int s = 0; s < x.GetStorageCellNumber(); s++)
+            for (int m = 0; m < x.GetModulesNum(); m++) {
+                for (int i = 0; i < RAW_MODULE_SIZE; i++) {
+                    g0[i] = Pedestal(m, i);
+                    g1[i] = 15000 + 10 * m + i % 50;
+                    g2[i] = 14000 + 10 * m + i % 50;
+                }
+                ret.Pedestal(m, 0, s).LoadPedestal(g0);
+                ret.Pedestal(m, 1, s).LoadPedestal(g1);
+                ret.Pedestal(m, 2, s).LoadPedestal(g2);
+            }
+        return ret;
+    }
+
+    // Raw detector frame of a module: background of a few photons, some spots, one saturated and one bad
+    // pixel, different for each frame.
+    std::vector<uint16_t> Frame(int64_t frame, int module, bool jungfrau) {
+        std::vector<uint16_t> ret(RAW_MODULE_SIZE);
+        std::mt19937 g(static_cast<uint32_t>(frame * 100 + module));
+        std::uniform_int_distribution<int> background(0, jungfrau ? 300 : 3);
+        for (int i = 0; i < RAW_MODULE_SIZE; i++)
+            ret[i] = (jungfrau ? static_cast<uint16_t>(Pedestal(module, i)) : 0) + background(g);
+
+        std::uniform_int_distribution<int> where_x(20, RAW_MODULE_COLS - 20), where_y(20, RAW_MODULE_LINES - 20);
+        for (int s = 0; s < 40; s++) {
+            const int x0 = where_x(g), y0 = where_y(g);
+            for (int y = y0; y < y0 + 3; y++)
+                for (int x = x0; x < x0 + 3; x++)
+                    ret[y * RAW_MODULE_COLS + x] = jungfrau ? 12000 : 2000;
+        }
+        ret[50 * RAW_MODULE_COLS + 500] = jungfrau ? 0xC000 : 0xFFFE;
+        ret[60 * RAW_MODULE_COLS + 600] = 0xFFFF;
+        return ret;
+    }
+
+    SpotFindingSettings Spots() {
+        SpotFindingSettings s = DiffractionExperiment::DefaultDataProcessingSettings();
+        s.signal_to_noise_threshold = 3;
+        s.photon_count_threshold = 2;
+        s.min_pix_per_spot = 1;
+        s.indexing = false;
+        return s;
+    }
+
+    AcquisitionDeviceGroup Devices(const DiffractionExperiment &x, size_t buffers = 64) {
+        AcquisitionDeviceGroup ret;
+        for (int d = 0; d < x.GetDataStreamsNum(); d++)
+            ret.Add(std::make_unique<HLSSimulatedDevice>(d, buffers));
+        return ret;
+    }
+
+    struct PathResult {
+        std::map<int64_t, Received> images;
+        std::vector<int64_t> arrival;
+        std::vector<std::string> sequence;
+        EndMessage end;
+        JFJochReceiverOutput output;
+    };
+
+    PathResult Run(const DiffractionExperiment &x, bool puller, int64_t distinct_frames) {
+        auto devices = Devices(x);
+        Logger logger("FPGAImagePullerTest");
+        CollectingPusher pusher;
+        JFJochReceiverService service(devices, logger, pusher, 256);
+        service.NumThreads(4);
+        service.SetSpotFindingSettings(Spots());
+        service.UseFPGAImagePuller(puller);
+
+        const bool jungfrau = x.GetDetectorType() == DetectorType::JUNGFRAU;
+        for (int64_t f = 0; f < distinct_frames; f++) {
+            std::vector<uint16_t> raw;
+            for (int m = 0; m < x.GetModulesNum(); m++) {
+                const auto frame = Frame(f, m, jungfrau);
+                raw.insert(raw.end(), frame.begin(), frame.end());
+            }
+            service.LoadInternalGeneratorImage(x, raw, f);
+        }
+
+        PixelMask mask(x);
+        const JFCalibration calib = Calibration(x);
+        service.Start(x, mask, &calib);
+        PathResult ret;
+        ret.output = service.Stop();
+        ret.images = std::move(pusher.images);
+        ret.arrival = std::move(pusher.arrival);
+        ret.sequence = std::move(pusher.sequence);
+        ret.end = pusher.end;
+        return ret;
+    }
+
+    std::vector<uint8_t> Pixels(const CompressedImage &image) {
+        std::vector<uint8_t> ret;
+        image.GetUncompressed(ret);
+        return ret;
+    }
+
+    template<class T>
+    void Row(const std::string &name, const std::optional<T> &a, const std::optional<T> &b) {
+        if (a == b)
+            return;
+        std::cout << "    differs " << name << ": FPGA ";
+        if (a) std::cout << *a; else std::cout << "-";
+        std::cout << " puller ";
+        if (b) std::cout << *b; else std::cout << "-";
+        std::cout << std::endl;
+    }
+
+    // Must be the same: asserted. Analysis: printed for the first image and summed over all.
+    void Compare(const std::string &name, const PathResult &a, const PathResult &b) {
+        std::cout << "=== " << name << std::endl;
+        REQUIRE(a.images.size() == b.images.size());
+        REQUIRE(!a.images.empty());
+
+        int64_t spots_a = 0, spots_b = 0, sat_a = 0, sat_b = 0, err_a = 0, err_b = 0, strong_a = 0, strong_b = 0;
+        double dot = 0, norm_a = 0, norm_b = 0, ratio_sum = 0;
+        size_t ratio_n = 0, bins_a = 0, bins_b = 0;
+
+        for (const auto &[n, ia]: a.images) {
+            INFO("image " << n);
+            REQUIRE(b.images.contains(n));
+            const auto &ib = b.images.at(n);
+            const auto &ma = ia.message;
+            const auto &mb = ib.message;
+
+            CHECK(ma.image.GetMode() == mb.image.GetMode());
+            CHECK(ma.image.GetCompressionAlgorithm() == mb.image.GetCompressionAlgorithm());
+            CHECK(ma.image.GetWidth() == mb.image.GetWidth());
+            CHECK(ma.image.GetHeight() == mb.image.GetHeight());
+            CHECK(Pixels(ma.image) == Pixels(mb.image));
+            CHECK(ia.compressed.size() == ib.compressed.size());
+
+            // From the source
+            CHECK(ma.number == mb.number);
+            CHECK(ma.original_number == mb.original_number);
+            CHECK(ma.packets_expected == mb.packets_expected);
+            CHECK(ma.packets_received == mb.packets_received);
+            CHECK(ma.image_collection_efficiency == mb.image_collection_efficiency);
+            CHECK(ma.timestamp == mb.timestamp);
+            CHECK(ma.timestamp_base == mb.timestamp_base);
+            CHECK(ma.exptime == mb.exptime);
+            CHECK(ma.exptime_base == mb.exptime_base);
+            CHECK(ma.jf_info == mb.jf_info);
+            CHECK(ma.storage_cell == mb.storage_cell);
+            CHECK(ma.xfel_pulse_id == mb.xfel_pulse_id);
+            CHECK(ma.xfel_event_code == mb.xfel_event_code);
+            CHECK(ma.pixel_sum == mb.pixel_sum);
+            CHECK(ma.user_data == mb.user_data);
+            CHECK(ma.run_number == mb.run_number);
+            CHECK(ma.run_name == mb.run_name);
+
+            if (n == a.images.begin()->first) {
+                Row("saturated_pixel_count", ma.saturated_pixel_count, mb.saturated_pixel_count);
+                Row("error_pixel_count", ma.error_pixel_count, mb.error_pixel_count);
+                Row("min_viable_pixel_value", ma.min_viable_pixel_value, mb.min_viable_pixel_value);
+                Row("max_viable_pixel_value", ma.max_viable_pixel_value, mb.max_viable_pixel_value);
+                Row("strong_pixel_count", ma.strong_pixel_count, mb.strong_pixel_count);
+                Row("spot_count", ma.spot_count, mb.spot_count);
+                Row("bkg_estimate", ma.bkg_estimate, mb.bkg_estimate);
+                Row("ice_ring_score", ma.ice_ring_score, mb.ice_ring_score);
+                Row("receiver_aq_dev_delay", ma.receiver_aq_dev_delay, mb.receiver_aq_dev_delay);
+                Row("receiver_buf_in_preparation", ma.receiver_buf_in_preparation, mb.receiver_buf_in_preparation);
+                Row("receiver_buf_in_sending", ma.receiver_buf_in_sending, mb.receiver_buf_in_sending);
+                std::cout << "    compression_ratio FPGA " << ma.compression_ratio.value_or(0)
+                          << " puller " << mb.compression_ratio.value_or(0) << std::endl;
+            }
+            spots_a += ma.spots.size();
+            spots_b += mb.spots.size();
+            sat_a += ma.saturated_pixel_count.value_or(0);
+            sat_b += mb.saturated_pixel_count.value_or(0);
+            err_a += ma.error_pixel_count.value_or(0);
+            err_b += mb.error_pixel_count.value_or(0);
+            strong_a += ma.strong_pixel_count.value_or(0);
+            strong_b += mb.strong_pixel_count.value_or(0);
+
+            bins_a = ma.az_int_profile.size();
+            bins_b = mb.az_int_profile.size();
+            for (size_t i = 0; i < std::min(bins_a, bins_b); i++) {
+                if (std::isfinite(ma.az_int_profile[i]) && std::isfinite(mb.az_int_profile[i])) {
+                    dot += ma.az_int_profile[i] * mb.az_int_profile[i];
+                    norm_a += ma.az_int_profile[i] * ma.az_int_profile[i];
+                    norm_b += mb.az_int_profile[i] * mb.az_int_profile[i];
+                    if (ma.az_int_profile[i] != 0) {
+                        ratio_sum += mb.az_int_profile[i] / ma.az_int_profile[i];
+                        ratio_n++;
+                    }
+                }
+            }
+        }
+        std::cout << "    over " << a.images.size() << " images:"
+                  << " spots FPGA " << spots_a << " puller " << spots_b
+                  << " | strong " << strong_a << " / " << strong_b
+                  << " | saturated " << sat_a << " / " << sat_b
+                  << " | error " << err_a << " / " << err_b
+                  << " | az bins " << bins_a << " / " << bins_b
+                  << " cosine " << (norm_a * norm_b > 0 ? dot / std::sqrt(norm_a * norm_b) : 0)
+                  << " mean ratio puller/FPGA " << (ratio_n > 0 ? ratio_sum / ratio_n : 0) << std::endl;
+
+        const auto &sa = a.output.status;
+        const auto &sb = b.output.status;
+        std::cout << "    status: efficiency " << a.output.efficiency << " / " << b.output.efficiency
+                  << " | collected " << sa.images_collected << " / " << sb.images_collected
+                  << " | sent " << sa.images_sent << " / " << sb.images_sent
+                  << " | dropped " << sa.images_dropped << " / " << sb.images_dropped
+                  << " | max_image_sent " << sa.max_image_number_sent << " / " << sb.max_image_number_sent
+                  << " | compressed_size " << sa.compressed_size << " / " << sb.compressed_size
+                  << " | ratio " << sa.compressed_ratio.value_or(0) << " / " << sb.compressed_ratio.value_or(0)
+                  << " | max_delay " << sa.max_receive_delay.value_or(0) << " / " << sb.max_receive_delay.value_or(0)
+                  << std::endl;
+
+        CHECK(a.output.efficiency == b.output.efficiency);
+        CHECK(sa.images_collected == sb.images_collected);
+        CHECK(sa.images_sent == sb.images_sent);
+        CHECK(sb.images_dropped == 0);
+        CHECK(sa.max_image_number_sent == sb.max_image_number_sent);
+        CHECK(sa.compressed_size == sb.compressed_size);
+        CHECK(a.end.images_collected_count == b.end.images_collected_count);
+        CHECK(a.end.images_sent_to_write_count == b.end.images_sent_to_write_count);
+        CHECK(a.end.max_image_number == b.end.max_image_number);
+        CHECK(a.end.adu_histogram == b.end.adu_histogram);
+
+        // Calibration between start and the first image, in the order the FPGA path sends it
+        CHECK(a.sequence.front() == "start");
+        CHECK(b.sequence.front() == "start");
+        CHECK(a.sequence.back() == "end");
+        CHECK(b.sequence.back() == "end");
+        CHECK(a.sequence == b.sequence);
+
+        int64_t displacement = 0;
+        for (size_t i = 0; i < b.arrival.size(); i++)
+            displacement = std::max(displacement, std::abs(b.arrival[i] - static_cast<int64_t>(i)));
+        std::cout << "    max displacement at the pusher (Lite's threads): " << displacement << " images" << std::endl;
+    }
+
+    void CompareBoth(const std::string &name, const DiffractionExperiment &x, int64_t distinct_frames) {
+        Compare(name, Run(x, false, distinct_frames), Run(x, true, distinct_frames));
+    }
+
+    DiffractionExperiment Base(const DetectorSetup &detector, int64_t images, int64_t distinct_frames) {
+        DiffractionExperiment x(detector);
+        x.PedestalG0Frames(0).NumTriggers(1).UseInternalPacketGenerator(true).InternalPacketGeneratorImages(distinct_frames)
+                .ImagesPerTrigger(images).ImagesPerFile(10).IncidentEnergy_keV(12.4)
+                .DetectorDistance_mm(75).BeamX_pxl(500).BeamY_pxl(600);
+        return x;
+    }
+}
+
+TEST_CASE("FPGAImagePuller_SameAsFPGA_JF_Conversion", "[FPGAImagePuller]") {
+    auto x = Base(DetJF(2), 6, 3);
+    x.Conversion();
+    x.Compression(CompressionAlgorithm::BSHUF_LZ4);
+    CompareBoth("JF conversion, int16, BSHUF_LZ4", x, 3);
+}
+
+TEST_CASE("FPGAImagePuller_SameAsFPGA_JF_Conversion_ApplyMask_ROI", "[FPGAImagePuller]") {
+    auto x = Base(DetJF(2), 4, 2);
+    x.Conversion();
+    x.Compression(CompressionAlgorithm::BSHUF_ZSTD).ApplyPixelMask(true);
+    x.ROI().SetROI(ROIDefinition{.boxes = {ROIBox("box", 380, 450, 280, 320)}, .circles = {}, .azimuthal = {}});
+    const auto a = Run(x, false, 2);
+    const auto b = Run(x, true, 2);
+    Compare("JF conversion, applied mask, ROI, BSHUF_ZSTD", a, b);
+    const auto &ra = a.images.begin()->second.message.roi.at("box");
+    const auto &rb = b.images.begin()->second.message.roi.at("box");
+    std::cout << "    ROI sum " << ra.sum << " / " << rb.sum << " pixels " << ra.pixels << " / " << rb.pixels
+              << " max " << ra.max_count << " / " << rb.max_count << std::endl;
+}
+
+TEST_CASE("FPGAImagePuller_SameAsFPGA_JF_Conversion_Unsigned", "[FPGAImagePuller]") {
+    auto x = Base(DetJF(2), 4, 2);
+    x.Conversion();
+    x.Compression(CompressionAlgorithm::BSHUF_LZ4).BitDepthImage(16).PixelSigned(false);
+    CompareBoth("JF conversion, uint16, BSHUF_LZ4", x, 2);
+}
+
+TEST_CASE("FPGAImagePuller_SameAsFPGA_JF_Conversion_Summation", "[FPGAImagePuller]") {
+    auto x = Base(DetJF(1), 3, 2);
+    x.Conversion();
+    x.Summation(2).Compression(CompressionAlgorithm::BSHUF_LZ4);
+    CompareBoth("JF conversion, card sums 2 frames, int32", x, 2);
+}
+
+TEST_CASE("FPGAImagePuller_SameAsFPGA_JF_Conversion_NoGeometry", "[FPGAImagePuller]") {
+    auto x = Base(DetJF(2), 4, 2);
+    x.Conversion();
+    x.GeometryTransformation(false).Compression(CompressionAlgorithm::BSHUF_LZ4);
+    CompareBoth("JF conversion, no geometry transform", x, 2);
+}
+
+TEST_CASE("FPGAImagePuller_SameAsFPGA_JF_Raw", "[FPGAImagePuller]") {
+    auto x = Base(DetJF(2), 4, 2);
+    x.Raw();
+    x.Compression(CompressionAlgorithm::BSHUF_LZ4);
+    CompareBoth("JF raw, BSHUF_LZ4", x, 2);
+}
+
+TEST_CASE("FPGAImagePuller_SameAsFPGA_JF_Raw_NoCompression", "[FPGAImagePuller]") {
+    auto x = Base(DetJF(1), 3, 1);
+    x.Raw();
+    x.Compression(CompressionAlgorithm::NO_COMPRESSION);
+    CompareBoth("JF raw, no compression", x, 1);
+}
+
+TEST_CASE("FPGAImagePuller_SameAsFPGA_EIGER_Conversion", "[FPGAImagePuller]") {
+    auto x = Base(DetEIGER(2), 4, 2);
+    x.Conversion();
+    x.Compression(CompressionAlgorithm::BSHUF_LZ4);
+    CompareBoth("EIGER assembled, 16 bit, BSHUF_LZ4", x, 2);
+}
+
+TEST_CASE("FPGAImagePuller_SameAsFPGA_EIGER_Raw", "[FPGAImagePuller]") {
+    auto x = Base(DetEIGER(2), 4, 2);
+    x.Raw();
+    x.Compression(CompressionAlgorithm::BSHUF_ZSTD);
+    CompareBoth("EIGER raw layout, 16 bit, BSHUF_ZSTD", x, 2);
+}
+
+TEST_CASE("FPGAImagePuller_SameAsFPGA_EIGER_32bit", "[FPGAImagePuller]") {
+    auto x = Base(DetEIGER(2), 3, 1);
+    x.Compression(CompressionAlgorithm::BSHUF_LZ4);
+    x.EigerBitDepth(32);
+    // Pattern from the generator's memory, as the 16-bit raw frames do not map onto a 32-bit stream
+    auto devices = Devices(x);
+    Logger logger("FPGAImagePuller_EIGER32");
+    for (const bool puller: {false, true}) {
+        CollectingPusher pusher;
+        JFJochReceiverService service(devices, logger, pusher, 256);
+        service.NumThreads(4).SetSpotFindingSettings(Spots());
+        service.UseFPGAImagePuller(puller);
+        PixelMask mask(x);
+        service.Start(x, mask, nullptr);
+        const auto out = service.Stop();
+        CHECK(out.efficiency == 1.0f);
+        CHECK(pusher.images.size() == 3);
+        CHECK(pusher.images.at(0).message.image.GetMode() == x.GetImageMode());
+    }
+}
+
+// Nobody takes the images: the slots run out and later images are dropped whole, without any waiting.
+TEST_CASE("FPGAImagePuller_SlowConsumer_Puller", "[FPGAImagePuller]") {
+    auto x = Base(DetJF(1), 20, 1);
+    x.Conversion();
+    x.Compression(CompressionAlgorithm::BSHUF_LZ4);
+
+    auto devices = Devices(x);
+    Logger logger("FPGAImagePuller_SlowConsumer");
+    PixelMask mask(x);
+    const JFCalibration calib = Calibration(x);
+    FPGAImagePuller puller(x, mask, &calib, devices, logger, 2, 4);
+
+    // Not read until the card is through, however long the simulation takes
+    while (!devices[0].Counters().IsAcquisitionFinished())
+        std::this_thread::sleep_for(10ms);
+
+    std::vector<int64_t> numbers;
+    bool end = false;
+    const auto t0 = std::chrono::steady_clock::now();
+    while (!end && std::chrono::steady_clock::now() - t0 < 60s) {
+        const auto m = puller.PollImage(10ms);
+        if (!m)
+            continue;
+        if (m->cbor->data_message)
+            numbers.push_back(m->cbor->data_message->number);
+        end = m->cbor->end_message.has_value();
+    }
+    CHECK(end);
+    puller.Stop();
+    std::cout << "slow consumer, puller alone: " << numbers.size() << " images, "
+              << puller.GetDroppedImages() << " dropped" << std::endl;
+    CHECK(numbers.size() + puller.GetDroppedImages() == 20);
+    CHECK(puller.GetDroppedImages() > 0);
+    for (size_t i = 1; i < numbers.size(); i++)
+        CHECK(numbers[i] > numbers[i - 1]);
+}
+
+// The same behind Lite: a writer that takes its time keeps Lite's only thread, and the drops reach the statistics.
+TEST_CASE("FPGAImagePuller_SlowConsumer_Receiver", "[FPGAImagePuller]") {
+    auto x = Base(DetJF(1), 24, 1);
+    x.Conversion();
+    x.Compression(CompressionAlgorithm::BSHUF_LZ4);
+
+    auto devices = Devices(x);
+    Logger logger("FPGAImagePuller_SlowReceiver");
+    CollectingPusher pusher;
+    pusher.delay = 3s;
+    JFJochReceiverService service(devices, logger, pusher, 256);
+    service.NumThreads(1).SetSpotFindingSettings(Spots());
+    service.UseFPGAImagePuller(true, 2);
+
+    PixelMask mask(x);
+    const JFCalibration calib = Calibration(x);
+    service.Start(x, mask, &calib);
+    const auto out = service.Stop();
+
+    std::cout << "slow consumer, Lite: collected " << out.status.images_collected << " dropped "
+              << out.status.images_dropped << " efficiency " << out.efficiency << std::endl;
+    CHECK(out.status.images_dropped > 0);
+    CHECK(out.status.images_collected + out.status.images_dropped == 24);
+    CHECK(out.efficiency == Catch::Approx(static_cast<float>(out.status.images_collected) / 24.0f));
+    CHECK(pusher.sequence.back() == "end");
+}
+
+// A module frame missing in the middle, and a module that dies: the images keep coming with no help
+// from the end of the run, and the missing module is filled, as in JFJochReceiverFPGA.
+TEST_CASE("FPGAImagePuller_MissingModule", "[FPGAImagePuller]") {
+    constexpr int64_t images = 62;
+    constexpr int64_t dies_at = 3;
+    auto x = Base(DetJF(2), images, 1);
+    x.UseInternalPacketGenerator(false).Conversion();
+    x.Compression(CompressionAlgorithm::BSHUF_LZ4);
+
+    auto device = std::make_unique<HLSSimulatedDevice>(0, 64);
+    for (int64_t f = 0; f < images; f++)
+        for (int m = 0; m < 2; m++) {
+            if ((m == 1) && (f == 1 || f >= dies_at))
+                continue;
+            const auto frame = Frame(f, m, true);
+            for (int p = 0; p < 128; p++)
+                device->CreateJFPacket(x, f + 1, p, m, frame.data() + p * 4096);
+        }
+    AcquisitionDeviceGroup devices;
+    devices.Add(std::move(device));
+
+    Logger logger("FPGAImagePuller_MissingModule");
+    PixelMask mask(x);
+    const JFCalibration calib = Calibration(x);
+    FPGAImagePuller puller(x, mask, &calib, devices, logger, 2);
+
+    // No final packet and no cancel yet: images must come through anyway, up to the lag behind the card
+    std::vector<int64_t> numbers;
+    std::vector<float> efficiency;
+    bool stop_requested = false, end = false;
+    size_t images_before_cancel = 0;
+    const auto t0 = std::chrono::steady_clock::now();
+    auto last = t0;
+    while (!end && std::chrono::steady_clock::now() - t0 < 900s) {
+        const auto m = puller.PollImage(10ms);
+        if (!m) {
+            if (!stop_requested && std::chrono::steady_clock::now() - last > 60s) {
+                stop_requested = true;
+                std::cout << "dead module, before cancel: fastest " << devices[0].Counters().GetFastestFrameNumber()
+                          << " module 0 at " << devices[0].Counters().GetCurrFrameNumber(0)
+                          << " module 1 at " << devices[0].Counters().GetCurrFrameNumber(1)
+                          << " images so far " << numbers.size() << std::endl;
+                puller.Cancel();
+            }
+            continue;
+        }
+        last = std::chrono::steady_clock::now();
+        if (m->cbor->data_message) {
+            const auto &d = *m->cbor->data_message;
+            numbers.push_back(d.number);
+            efficiency.push_back(d.image_collection_efficiency.value_or(-1));
+            if (!stop_requested)
+                images_before_cancel = numbers.size();
+
+            if (d.number == 1) {
+                // Module 1 absent from this frame: its pixels are the fill value
+                std::vector<uint8_t> tmp;
+                const auto *p = reinterpret_cast<const int16_t *>(d.image.GetUncompressedPtr(tmp));
+                CHECK(p[x.GetPixel0OfModuleConv(1)] == x.GetImageFillValue());
+                CHECK(p[x.GetPixel0OfModuleConv(0)] != x.GetImageFillValue());
+            }
+        }
+        end = m->cbor->end_message.has_value();
+    }
+    CHECK(end);
+    REQUIRE_NOTHROW(puller.Stop());
+    std::cout << "dead module: " << numbers.size() << " images, before any cancel: "
+              << images_before_cancel << std::endl;
+    // Image i is let go once the card reports frame i + 2 + MissingAfterFrames of the other module
+    CHECK(images_before_cancel >= 4);
+    REQUIRE(numbers.size() >= 4);
+    for (size_t i = 0; i < numbers.size(); i++)
+        CHECK(numbers[i] == static_cast<int64_t>(i));
+    CHECK(efficiency[0] == 1.0f);
+    CHECK(efficiency[1] == 0.5f);
+    // Frame 2 of module 1 is sent, but the card holds a module's last frames until newer ones push them
+    // out: given up on, it comes with the cancel. Its buffer went back.
+    CHECK(efficiency[3] == 0.5f);
+    CHECK(efficiency.back() == 0.5f);
+}
+
+TEST_CASE("FPGAImagePuller_Cancel", "[FPGAImagePuller]") {
+    for (const bool silent: {false, true}) {
+        auto x = Base(DetJF(2), 6, 1);
+        x.UseInternalPacketGenerator(false).Conversion();
+        x.Compression(CompressionAlgorithm::BSHUF_LZ4);
+
+        auto device = std::make_unique<HLSSimulatedDevice>(0, 64);
+        for (int64_t f = 0; f < 3; f++)
+            for (int m = 0; m < 2; m++) {
+                const auto frame = Frame(f, m, true);
+                for (int p = 0; p < 128; p++)
+                    device->CreateJFPacket(x, f + 1, p, m, frame.data() + p * 4096);
+            }
+        AcquisitionDeviceGroup devices;
+        devices.Add(std::move(device));
+
+        Logger logger("FPGAImagePuller_Cancel");
+        CollectingPusher pusher;
+        JFJochReceiverService service(devices, logger, pusher, 256);
+        service.NumThreads(2).SetSpotFindingSettings(Spots());
+        service.UseFPGAImagePuller(true);
+        PixelMask mask(x);
+        const JFCalibration calib = Calibration(x);
+        service.Start(x, mask, &calib);
+
+        std::this_thread::sleep_for(30s);
+        service.Cancel(silent);
+        const auto out = service.Stop();
+
+        std::cout << "cancel silent=" << silent << ": cancelled " << out.status.cancelled << " collected "
+                  << out.status.images_collected << " sent " << out.status.images_sent << " efficiency "
+                  << out.efficiency << std::endl;
+        CHECK(out.status.cancelled == !silent);
+        CHECK(pusher.sequence.front() == "start");
+        CHECK(pusher.sequence.back() == "end");
+        CHECK(out.status.images_collected <= 3);
+        if (silent) {
+            CHECK(out.status.images_collected == 3);
+            CHECK(out.efficiency == Catch::Approx(0.5f));
+        }
+    }
+}
+
+// Default stays the FPGA receiver; pedestal and summation above the card's limit never take the puller
+TEST_CASE("FPGAImagePuller_NotForPedestalAndHostSummation", "[FPGAImagePuller]") {
+    auto x = Base(DetJF(1), 2, 1);
+    x.Conversion();
+    x.Summation(2).CPUSummation(true).Compression(CompressionAlgorithm::BSHUF_LZ4);
+    REQUIRE(x.IsCPUSummation());
+
+    auto devices = Devices(x);
+    Logger logger("FPGAImagePuller_HostSummation");
+    CollectingPusher pusher;
+    JFJochReceiverService service(devices, logger, pusher, 256);
+    service.NumThreads(4).SetSpotFindingSettings(Spots());
+    service.UseFPGAImagePuller(true);
+    PixelMask mask(x);
+    const JFCalibration calib = Calibration(x);
+    service.Start(x, mask, &calib);
+    const auto out = service.Stop();
+    CHECK(out.efficiency == 1.0f);
+    CHECK(!out.received_packets.empty()); // only the FPGA receiver fills these
+    CHECK(out.status.images_collected == 2);
+}
+
+// JFJochReceiverRugnux takes the same stream, with its own forwarding thread.
+TEST_CASE("FPGAImagePuller_RugnuxReceiver", "[FPGAImagePuller]") {
+    auto x = Base(DetJF(2), 6, 3);
+    x.Conversion();
+    x.Compression(CompressionAlgorithm::BSHUF_LZ4);
+    x.Goniometer(GoniometerAxis("omega", 0, 0.1, Coord(-1, 0, 0), std::nullopt));
+    x.FilePrefix("fpga_puller_rugnux");
+    REQUIRE(RugnuxRotationData(x));
+
+    auto devices = Devices(x);
+    Logger logger("FPGAImagePuller_Rugnux");
+    CollectingPusher pusher;
+    JFJochReceiverService service(devices, logger, pusher, 256);
+    service.NumThreads(2).SetSpotFindingSettings(Spots());
+    service.UseFPGAImagePuller(true);
+    service.RugnuxLive(true, nullptr, 2, 30);
+
+    for (int64_t f = 0; f < 3; f++) {
+        std::vector<uint16_t> raw;
+        for (int m = 0; m < 2; m++) {
+            const auto frame = Frame(f, m, true);
+            raw.insert(raw.end(), frame.begin(), frame.end());
+        }
+        service.LoadInternalGeneratorImage(x, raw, f);
+    }
+
+    PixelMask mask(x);
+    const JFCalibration calib = Calibration(x);
+    service.Start(x, mask, &calib);
+    const auto out = service.Stop();
+    std::cout << "rugnux receiver: efficiency " << out.efficiency << " sent " << out.status.images_sent
+              << " warning '" << out.analysis_warning << "'" << std::endl;
+    CHECK(out.efficiency == 1.0f);
+    CHECK(out.status.images_sent == 6);
+    CHECK(pusher.images.size() == 6);
+    CHECK(pusher.sequence.front() == "start");
+    CHECK(pusher.sequence.back() == "end");
+    CHECK(pusher.sequence[1].starts_with("calibration"));
+    const auto a = Pixels(pusher.images.at(0).message.image);
+    const auto b = Pixels(pusher.images.at(3).message.image);
+    CHECK(a == b); // the third frame in the generator's memory repeats
+}
+
+// The real writer: the master file states what the card did, not what a DECTRIS stream implies.
+TEST_CASE("FPGAImagePuller_Writer", "[FPGAImagePuller]") {
+    RegisterHDF5Filter();
+    auto x = Base(DetJF(2), 5, 2);
+    x.Conversion();
+    x.Compression(CompressionAlgorithm::BSHUF_LZ4).ApplyPixelMask(true).ImagesPerFile(2)
+            .FilePrefix("fpga_image_puller_writer").SetFileWriterFormat(FileWriterFormat::NXmxVDS)
+            .OverwriteExistingFiles(true).SaveCalibration(true);
+
+    auto devices = Devices(x);
+    Logger logger("FPGAImagePuller_Writer");
+    HDF5FilePusher pusher;
+    JFJochReceiverService service(devices, logger, pusher, 256);
+    service.NumThreads(2).SetSpotFindingSettings(Spots());
+    service.UseFPGAImagePuller(true);
+    PixelMask mask(x);
+    const JFCalibration calib = Calibration(x);
+    service.Start(x, mask, &calib);
+    const auto out = service.Stop();
+    CHECK(out.efficiency == 1.0f);
+    CHECK(out.writer_err.empty());
+    CHECK(out.status.images_sent == 5);
+
+    HDF5ReadOnlyFile master("fpga_image_puller_writer_master.h5");
+    CHECK(master.GetBool("/entry/instrument/detector/pixel_mask_applied"));
+    HDF5DataSet dataset(master, "/entry/data/data");
+    HDF5DataSpace space(dataset);
+    CHECK(space.GetDimensions()[0] == 5);
+    CHECK(space.GetDimensions()[1] == x.GetYPixelsNum());
+}