Jungfraujoch/receiver/JFJochReceiverFPGA.cpp

// SPDX-FileCopyrightText: 2024 Filip Leonarski, Paul Scherrer Institute <filip.leonarski@psi.ch>
// SPDX-License-Identifier: GPL-3.0-only

#include "JFJochReceiverFPGA.h"

#include "../image_analysis/WriteReflections.h"

#include <thread>

#include "ImageMetadata.h"
#include "JFJochReceiverLite.h"
#include "../common/CUDAWrapper.h"
#include "../image_analysis/MXAnalysisWithoutFPGA.h"

JFJochReceiverFPGA::JFJochReceiverFPGA(const DiffractionExperiment &in_experiment,
                                       const PixelMask &in_pixel_mask,
                                       const JFCalibration *in_calibration,
                                       AcquisitionDeviceGroup &in_aq_device,
                                       ImagePusher &in_image_sender,
                                       Logger &in_logger, int64_t in_forward_and_sum_nthreads,
                                       const SpotFindingSettings &in_spot_finding_settings,
                                       JFJochReceiverCurrentStatus &in_current_status,
                                       JFJochReceiverPlots &in_plots,
                                       ImageBuffer &in_send_buf_ctrl,
                                       ZMQPreviewSocket *in_zmq_preview_socket,
                                       ZMQMetadataSocket *in_zmq_metadata_socket,
                                       IndexerThreadPool *indexing_thread_pool,
                                       bool in_gpu_analysis,
                                       RugnuxLiveStatus *in_rugnux_status)
    : JFJochReceiver(in_experiment,
                     in_send_buf_ctrl,
                     in_image_sender,
                     in_current_status,
                     in_plots,
                     in_spot_finding_settings,
                     in_logger,
                     in_pixel_mask,
                     in_zmq_preview_socket,
                     in_zmq_metadata_socket,
                     indexing_thread_pool),
      calibration(nullptr),
      acquisition_device(in_aq_device),
      ndatastreams(experiment.GetDataStreamsNum()),
      frame_transformation_nthreads(
          in_forward_and_sum_nthreads),
      pedestal_nthreads((experiment.GetStorageCellNumber() > 2) ? 1 : 4),
      summation_nthreads(2),
      frame_transformation_ready((experiment.GetImageNum() > 0) ? frame_transformation_nthreads : 0),
      data_acquisition_ready(ndatastreams),
      gpu_analysis(in_gpu_analysis && experiment.GetImageNum() > 0
                   && experiment.GetDetectorMode() == DetectorMode::Standard),
      rugnux_status(in_rugnux_status) {
    // The FPGA integration core has a fixed bin-count limit. Reject configurations it cannot
    // handle before starting acquisition, unless azimuthal integration is forced onto the CPU.
    const auto &azint_settings = experiment.GetAzimuthalIntegrationSettings();
    if (!azint_settings.IsForceCPUinFPGAWorkflow() && azint_settings.GetBinCount() > FPGA_INTEGRATION_BIN_COUNT)
        throw JFJochException(JFJochExceptionCategory::InputParameterAboveMax,
                              fmt::format("Azimuthal integration bin count ({}) exceeds the FPGA limit ({}). "
                                          "Enable ForceCPUinFPGAWorkflow to compute it on the CPU.",
                                          azint_settings.GetBinCount(), FPGA_INTEGRATION_BIN_COUNT));

    image_buffer.StartMeasurement(experiment);

    for (int m = 0; m < experiment.GetModulesNum(); m++)
        adu_histogram_module.emplace_back(std::make_unique<ADUHistogram>());

    roi_map = experiment.ExportROIMap();

    if (experiment.GetDetectorSetup().GetDetectorType() == DetectorType::JUNGFRAU)
        calibration = in_calibration;

    if (acquisition_device.size() < ndatastreams)
        throw JFJochException(JFJochExceptionCategory::InputParameterInvalid,
                              "Number of acquisition devices has to match data streams");
    if (frame_transformation_nthreads <= 0)
        throw JFJochException(JFJochExceptionCategory::InputParameterInvalid,
                              "Number of threads must be more than zero");

    expected_packets_per_image = 0;
    for (int d = 0; d < ndatastreams; d++) {
        acquisition_device[d].PrepareAction(experiment);
        acquisition_device[d].SetSpotFinderParameters(spot_finding_settings);

        expected_packets_per_image += acquisition_device[d].Counters().GetExpectedPacketsPerImage();

        logger.Debug("Acquisition device {} prepared", d);
    }

    if (experiment.IsCPUSummation())
        expected_packets_per_image *= experiment.GetSummation();

    logger.Debug("Data acquisition devices ready");

    if ((experiment.GetDetectorMode() == DetectorMode::PedestalG0)
        || (experiment.GetDetectorMode() == DetectorMode::PedestalG1)
        || (experiment.GetDetectorMode() == DetectorMode::PedestalG2)) {
        if (experiment.GetImageNum() > 0) {
            throw JFJochException(JFJochExceptionCategory::InputParameterInvalid,
                                  "Saving and calculating pedestal is not supported for the time being");
        }

        switch (experiment.GetDetectorMode()) {
            case DetectorMode::PedestalG1:
                only_2nd_sc_pedestal = !experiment.IsFixedGainG1() && (experiment.GetStorageCellNumber() == 2);
                break;
            case DetectorMode::PedestalG2:
                only_2nd_sc_pedestal = (experiment.GetStorageCellNumber() == 2);
                break;
            default:
                only_2nd_sc_pedestal = false;
                break;
        }

        int64_t pedestal_count = (only_2nd_sc_pedestal)
                                     ? experiment.GetModulesNum()
                                     : experiment.GetModulesNum() * experiment.GetStorageCellNumber();
        for (int i = 0; i < pedestal_count; i++)
            pedestal.emplace_back(std::make_unique<JFPedestalCalc>(experiment));

        for (int s = 0; s < experiment.GetStorageCellNumber(); s++) {
            bool ignore = only_2nd_sc_pedestal && (s == 0);
            for (int d = 0; d < ndatastreams; d++)
                for (int m = 0; m < experiment.GetModulesNum(d); m++)
                    for (int n = 0; n < pedestal_nthreads; n++)
                        frame_transformation_futures.emplace_back(std::async(std::launch::async,
                                                                             &JFJochReceiverFPGA::MeasurePedestalThread,
                                                                             this, d, m, s, n, ignore));
        }
        logger.Debug("Pedestal threads ready ({} threads/module*SC)", pedestal_nthreads);
    } else if (experiment.GetImageNum() > 0) {
        if (gpu_analysis) {
            // The start message and the calibration go out once the pre-pass is done.
            gpu = std::make_unique<StillsGPUAnalysis>(
                    experiment, pixel_mask, MakeStartMessage(experiment, pixel_mask, *az_int_mapping),
                    indexing_thread_pool, std::min<int64_t>(16, frame_transformation_nthreads),
                    // Half the image buffer for the frames waiting, the other half for those sent.
                    image_buffer.GetStatus().total_slots / 2, logger,
                    [this](const StillsGPUAnalysis::Setup &s) {
                        // The profiles are binned on the pre-pass's geometry.
                        plots.Setup(s.experiment, *s.mapping);
                        SendStartMessage(MakeStartMessage(s.experiment, s.mask, *s.mapping));
                        SendCalibration();
                    },
                    [this](DataMessage &m, const AzimuthalIntegrationProfile &p) { GPUSink(m, p); },
                    [this] {
                        // The broker's settings, with the spot finder rugnux uses on stills (the pre-pass
                        // decided the centre and mask with it): adaptive threshold, single-pixel spots kept.
                        // With the broker's default fixed threshold 3 points fewer frames index.
                        SpotFindingSettings s = GetSpotFindingSettings();
                        s.adaptive_threshold = true;
                        s.min_pix_per_spot = std::nullopt;
                        return s;
                    },
                    // The start message did not go out (or the analysis could not be set up): nothing can
                    // be written, so the collection stops, as it does on an error in a frame thread.
                    [this](const JFJochException &e) { Cancel(e); },
                    rugnux_status);
        } else {
            SendStartMessage();
            SendCalibration();
        }

        for (int i = 0; i < experiment.GetImageNum(); i++)
            images_to_go.Put(i);

        // Setup frames summation and forwarding
        for (uint32_t i = 0; i < frame_transformation_nthreads; i++) {
            auto handle = std::async(std::launch::async, &JFJochReceiverFPGA::FrameTransformationThread,
                                     this, i);
            frame_transformation_futures.emplace_back(std::move(handle));
        }

        logger.Debug("Image compression/forwarding threads started ({} threads)", frame_transformation_nthreads);
        frame_transformation_ready.wait();
        logger.Debug("Image compression/forwarding threads ready");
    }

    for (int d = 0; d < ndatastreams; d++)
        data_acquisition_futures.emplace_back(std::async(std::launch::async, &JFJochReceiverFPGA::AcquireThread,
                                                         this, d));

    data_acquisition_ready.wait();
    start_time = std::chrono::system_clock::now();
    measurement = std::async(std::launch::async, &JFJochReceiverFPGA::FinalizeMeasurement, this);
    logger.Info("Receiving data started");
}

void JFJochReceiverFPGA::SendPedestal(const std::string &prefix, const std::vector<uint8_t> &v, int gain, int sc) {
    size_t xpixel = RAW_MODULE_COLS;
    size_t ypixel = experiment.GetModulesNum() * RAW_MODULE_LINES;

    std::string channel;

    if (experiment.GetStorageCellNumber() > 1)
        channel = fmt::format("{:s}_g{:d}_sc{:d}", prefix, gain, sc);
    else
        channel = fmt::format("{:s}_g{:d}", prefix, gain);

    CompressedImage image(v.data(), v.size(), xpixel, ypixel, CompressedImageMode::Uint16,
                          CompressionAlgorithm::BSHUF_LZ4, channel);

    image_pusher.SendCalibration(image);
}

void JFJochReceiverFPGA::SendCalibration() {
    if ((calibration == nullptr) || !experiment.GetSaveCalibration() || !push_images_to_writer)
        return;

    JFJochBitShuffleCompressor compressor(CompressionAlgorithm::BSHUF_LZ4);

    for (int sc = 0; sc < experiment.GetStorageCellNumber(); sc++) {
        for (int gain = 0; gain < 3; gain++) {
            if (experiment.IsFixedGainG1() && (gain != 1))
                continue;
            SendPedestal("pedestal", compressor.Compress(calibration->GetPedestal(gain, sc)), gain, sc);
            SendPedestal("pedestal_rms", compressor.Compress(calibration->GetPedestalRMS(gain, sc)), gain, sc);
        }
    }
}

void JFJochReceiverFPGA::AcquireThread(uint16_t data_stream) {
    try {
        LoadCalibrationToFPGA(data_stream);
        frame_transformation_ready.wait();
        logger.Debug("Device thread {} start FPGA action", data_stream);
        acquisition_device[data_stream].StartAction(experiment);
    } catch (const JFJochException &e) {
        Cancel(e);
        data_acquisition_ready.count_down();
        logger.ErrorException(e);
        logger.Warning("Device thread {} done due to an error", data_stream);
        return;
    }

    data_acquisition_ready.count_down();

    try {
        logger.Debug("Device thread {} wait for FPGA action complete", data_stream);
        acquisition_device[data_stream].WaitForActionComplete();
    } catch (const JFJochException &e) {
        logger.ErrorException(e);
        Cancel(e);
        logger.ErrorException(e);
        logger.Warning("Device thread {} done due to an error", data_stream);
        return;
    }
    logger.Debug("Device thread {} done", data_stream);
}


void JFJochReceiverFPGA::MeasurePedestalThread(uint16_t data_stream, uint16_t module_number, uint16_t storage_cell,
                                               uint32_t threadid, bool ignore) {
    JFPedestalCalc pedestal_calc(experiment);

    uint64_t starting_frame = storage_cell + threadid * experiment.GetStorageCellNumber();
    uint64_t frame_stride = experiment.GetStorageCellNumber() * pedestal_nthreads;
    uint32_t storage_cell_header = UINT32_MAX;

    try {
        for (size_t frame = starting_frame; frame < experiment.GetFrameNum(); frame += frame_stride) {
            // Frame will be processed only if one already collects frame+2
            acquisition_device[data_stream].Counters().WaitForFrame(frame + 2, module_number);
            if (acquisition_device[data_stream].Counters().IsFullModuleCollected(frame, module_number) && !ignore) {
                auto output = acquisition_device[data_stream].GetDeviceOutput(frame, module_number);
                // Partial packets will bring more problems, than benefit
                pedestal_calc.AnalyzeImage((uint16_t *) output->pixels);
                storage_cell_header = (output->module_statistics.debug >> 8) & 0xF;
            }
            acquisition_device[data_stream].FrameBufferRelease(frame, module_number);

            UpdateMaxDelay(acquisition_device[data_stream].Counters().CalculateDelay(frame, module_number));
            current_status.SetProgress(GetProgress());
            current_status.SetStatus(GetStatus());
        }

        uint64_t offset = experiment.GetFirstModuleOfDataStream(data_stream) + module_number;
        if (!only_2nd_sc_pedestal)
            offset += experiment.GetModulesNum() * storage_cell;

        if (!ignore)
            *pedestal[offset] += pedestal_calc;
    } catch (const JFJochException &e) {
        Cancel(e);
    }
    logger.Debug("Pedestal calculation thread for data stream {} module {} storage cell {} -> header {} done",
                 data_stream, module_number, storage_cell, storage_cell_header);
}

int64_t JFJochReceiverFPGA::SummationThread(uint16_t data_stream,
                                            int64_t image_number,
                                            uint16_t module_number,
                                            uint32_t threadid,
                                            ModuleSummation &summation) {
    ModuleSummation local_summation(experiment);
    int64_t starting_frame = image_number * experiment.GetSummation();

    for (int64_t i = threadid; i < experiment.GetSummation(); i += summation_nthreads) {
        const int64_t frame = starting_frame + i;

        // Frame will be processed only if one already collects frame+2
        acquisition_device[data_stream].Counters().WaitForFrame(frame + 2, module_number);
        if (acquisition_device[data_stream].Counters().IsAnyPacketCollected(frame, module_number)) {
            const auto output = acquisition_device[data_stream].GetDeviceOutput(frame, module_number);
            local_summation.AddFPGAOutput(*output);
        } else
            local_summation.AddEmptyOutput();

        acquisition_device[data_stream].FrameBufferRelease(frame, module_number);

        UpdateMaxDelay(acquisition_device[data_stream].Counters().CalculateDelay(frame, module_number));
        current_status.SetProgress(GetProgress());
        current_status.SetStatus(GetStatus());
    }

    if (!summation.empty())
        summation.AddFPGAOutput(local_summation.GetOutput(), 4);
    return 0;
}

void JFJochReceiverFPGA::FrameTransformationThread(uint32_t threadid) {
    std::unique_ptr<MXAnalysisAfterFPGA> analyzer;

    try {
        analyzer = std::make_unique<MXAnalysisAfterFPGA>(experiment, *az_int_mapping, indexer);
    } catch (const JFJochException &e) {
        frame_transformation_ready.count_down();
        logger.Error("Thread setup error {}", e.what());
        Cancel(e);
        return;
    }

    FrameTransformation transformation(experiment);

    frame_transformation_ready.count_down();

    uint16_t az_int_min_bin = std::floor(az_int_mapping->QToBin(experiment.GetLowQForBkgEstimate_recipA()));
    uint16_t az_int_max_bin = std::ceil(az_int_mapping->QToBin(experiment.GetHighQForBkgEstimate_recipA()));

    // When forced, azimuthal integration is computed on the CPU from the assembled image
    // instead of being read back from the FPGA per module (lifts the FPGA bin-count limit).
    const bool force_cpu_azint = experiment.GetAzimuthalIntegrationSettings().IsForceCPUinFPGAWorkflow();

    int64_t image_number;
    while (images_to_go.Get(image_number) != 0) {
        try {
            int64_t expected_frame = image_number;
            if (experiment.IsCPUSummation())
                expected_frame *= experiment.GetSummation();

            logger.Debug("Frame transformation thread - trying to get image {}", expected_frame);
            // If data acquisition is finished and fastest frame for the first device is behind
            acquisition_device[0].Counters().WaitForFrame(expected_frame);
            logger.Debug("Frame transformation thread - frame arrived {}", expected_frame);
            if (acquisition_device[0].Counters().IsAcquisitionFinished() &&
                (acquisition_device[0].Counters().GetFastestFrameNumber() < expected_frame)) {
                logger.Debug("Frame transformation thread - skipping image {}", expected_frame);
                continue;
            }

            DataMessage message{};
            message.number = image_number;
            message.original_number = image_number;
            message.user_data = experiment.GetImageAppendix();
            message.run_number = experiment.GetRunNumber();
            message.run_name = experiment.GetRunName();

            ImageMetadata metadata(experiment);

            AzimuthalIntegrationProfile az_int_profile_image(*az_int_mapping);

            auto local_spot_finding_settings = GetSpotFindingSettings();

            const auto preprocessing_start_time = std::chrono::steady_clock::now();

            if (experiment.IsCPUSummation()) {
                std::vector<std::unique_ptr<ModuleSummation>> summation;
                for (int i = 0; i < experiment.GetModulesNum(); i++)
                    summation.emplace_back(std::make_unique<ModuleSummation>(experiment));

                std::vector<std::future<int64_t> > futures;

                for (int d = 0; d < ndatastreams; d++) {
                    for (int m = 0; m < experiment.GetModulesNum(d); m++) {
                        size_t module_abs_number = experiment.GetFirstModuleOfDataStream(d) + m;
                        for (int i = 0; i < summation_nthreads; i++) {
                            futures.emplace_back(
                                std::async(std::launch::async,
                                           &JFJochReceiverFPGA::SummationThread,
                                           this,
                                           d, image_number, m, i,
                                           std::ref(*summation[module_abs_number]))
                            );
                        }
                    }
                }

                for (auto &f: futures)
                    f.get();

                for (int d = 0; d < ndatastreams; d++) {
                    for (int m = 0; m < experiment.GetModulesNum(d); m++) {
                        size_t i = experiment.GetFirstModuleOfDataStream(d) + m;

                        if (!summation[i]->empty()) {
                            adu_histogram_module[i]->Add(summation[i]->GetOutput());
                            transformation.ProcessModule(&summation[i]->GetOutput(), d);
                            metadata.Process(&summation[i]->GetOutput());
                            if (!force_cpu_azint)
                                az_int_profile_image.Add(summation[i]->GetOutput());
                            analyzer->ReadFromCPU(&summation[i]->GetOutput(), GetSpotFindingSettings(), i);
                        } else
                            transformation.FillNotCollectedModule(m, d);
                    }
                }
            } else {
                logger.Debug("Frame transformation thread - processing image from FPGA {}", image_number);
                for (int d = 0; d < ndatastreams; d++) {
                    for (int m = 0; m < experiment.GetModulesNum(d); m++) {
                        acquisition_device[d].Counters().WaitForFrame(image_number + 2, m);
                        if (acquisition_device[d].Counters().IsAnyPacketCollected(image_number, m)) {
                            const DeviceOutput *output = acquisition_device[d].GetDeviceOutput(image_number, m);

                            metadata.Process(output);

                            size_t module_abs_number = experiment.GetFirstModuleOfDataStream(d) + m;
                            adu_histogram_module[module_abs_number]->Add(*output);
                            if (!force_cpu_azint)
                                az_int_profile_image.Add(*output);

                            analyzer->ReadFromFPGA(output, local_spot_finding_settings, module_abs_number);

                            transformation.ProcessModule(output, d);
                        } else
                            transformation.FillNotCollectedModule(m, d);
                        acquisition_device[d].FrameBufferRelease(image_number, m);
                    }
                    auto delay = acquisition_device[d].Counters().CalculateDelay(image_number);
                    UpdateMaxDelay(delay);
                    if (delay > message.receiver_aq_dev_delay)
                        message.receiver_aq_dev_delay = delay;
                }
            }
            const auto preprocessing_end_time = std::chrono::steady_clock::now();
            message.preprocessing_time_s = std::chrono::duration<float>(preprocessing_end_time - preprocessing_start_time).count();

            auto image_start_time = std::chrono::high_resolution_clock::now();

            metadata.Export(message, expected_packets_per_image);

            if (message.image_collection_efficiency == 0.0f) {
                plots.AddEmptyImage(message);
                continue;
            }

            message.image = CompressedImage(transformation.GetImage(),
                                            experiment.GetPixelsNum() * experiment.GetByteDepthImage(),
                                            experiment.GetXPixelsNum(),
                                            experiment.GetYPixelsNum(),
                                            experiment.GetImageMode(),
                                            CompressionAlgorithm::NO_COMPRESSION);

            if (gpu_analysis) {
                // Compressed here into a slot of the image buffer, where it waits for the GPU analysis; GPUSink
                // writes it, with its analysis, to a second slot, which is sent. A frame that finds no slot
                // free, or as many frames waiting as the analysis is to hold, is dropped, as the receiver drops
                // any frame that finds the image buffer full.
                ++images_collected;
                uncompressed_size += experiment.GetModulesNum() * RAW_MODULE_SIZE * experiment.GetByteDepthImage();
                auto held = std::make_shared<HeldImage>();
                held->slot = image_buffer.GetImageSlot();
                if (held->slot == nullptr) {
                    writer_queue_full = true;
                    continue;
                }
                const size_t size = transformation.CompressImage(static_cast<uint8_t *>(held->slot->GetImage()),
                                                                 experiment.GetImageBufferLocationSize());
                held->message = std::move(message);
                held->message.image = CompressedImage(held->slot->GetImage(), size,
                                                      experiment.GetXPixelsNum(), experiment.GetYPixelsNum(),
                                                      experiment.GetImageMode(),
                                                      experiment.GetCompressionAlgorithm());
                if (!gpu->Hold(held)) {
                    writer_queue_full = true;
                    continue;
                }
                compressed_size += size;
                continue;
            }

            // No-op unless the CPU backend is forced; fills az_int_profile_image from the assembled image
            analyzer->RunAzimuthalIntegration(transformation.GetImage(), az_int_profile_image);

            analyzer->Process(message, local_spot_finding_settings);

            auto status = image_buffer.GetStatus();
            message.receiver_buf_available = status.available_slots;
            message.receiver_buf_in_preparation = status.preparation_slots;
            message.receiver_buf_in_sending = status.sending_slots;
            message.az_int_profile = az_int_profile_image.GetResult();
            message.az_int_profile_count = az_int_profile_image.GetPixelCount();
            if (force_cpu_azint)
                message.az_int_profile_std = az_int_profile_image.GetStd();
            message.bkg_estimate = az_int_profile_image.GetBkgEstimate(experiment.GetAzimuthalIntegrationSettings());
            message.ice_ring_score = az_int_profile_image.GetIceRingScore(
                experiment.GetAzimuthalIntegrationSettings(), spot_finding_settings.ice_ring_width_Q_recipA);

            scan_result.Add(message);

            auto image_end_time = std::chrono::high_resolution_clock::now();
            std::chrono::duration<float> duration = image_end_time - image_start_time;
            message.processing_time_s = duration.count();

            // Store overload/error pixel count
            if (message.image_collection_efficiency == 1.0f) {
                saturated_pixels.Add(message.saturated_pixel_count);
                error_pixels.Add(message.error_pixel_count);

                if (message.roi.contains("beam")) {
                    roi_beam_npixel.Add(message.roi["beam"].pixels);
                    roi_beam_sum.Add(message.roi["beam"].sum);
                }
            }

            ++images_collected;

            uncompressed_size += experiment.GetModulesNum() * RAW_MODULE_SIZE * experiment.GetByteDepthImage();

            if (!serialmx_filter.ApplyFilter(message))
                ++images_skipped;
            else {
                auto loc = image_buffer.GetImageSlot();
                if (loc == nullptr) {
                    // No free buffer locations - continue
                    writer_queue_full = true;
                } else {
                    auto writer_buffer = (uint8_t *) loc->GetImage();

                    CBORStream2Serializer serializer(writer_buffer, experiment.GetImageBufferLocationSize());
                    message.image = CompressedImage(nullptr, 0,
                                                    experiment.GetXPixelsNum(),
                                                    experiment.GetYPixelsNum(),
                                                    experiment.GetImageMode(),
                                                    experiment.GetCompressionAlgorithm());
                    serializer.SerializeImage(message);

                    const size_t metadata_size = serializer.GetImageAppendOffset();
                    const size_t slot_size = experiment.GetImageBufferLocationSize();

                    const auto compression_start_time = std::chrono::steady_clock::now();
                    try {
                        // Reserve 32 bytes for close, etc. If the metadata already fills the
                        // slot the subtraction below would wrap (size_t), so bail out here and
                        // let the catch drop just this frame.
                        if (metadata_size + 32 >= slot_size)
                            throw CompressionBufferTooSmallException("CBOR metadata leaves no room for the compressed image");

                        size_t image_size = transformation.CompressImage(
                                writer_buffer + serializer.GetImageAppendOffset(),
                                slot_size - (metadata_size + 32));
                        const auto compression_end_time = std::chrono::steady_clock::now();
                        message.compression_time_s = std::chrono::duration<float>(
                            compression_end_time - compression_start_time).count();

                        serializer.AppendImage(image_size);
                        compressed_size += image_size;
                        if (image_size > 0)
                            message.compression_ratio =
                                    static_cast<float>(experiment.GetPixelsNum() * experiment.GetByteDepthImage())
                                    / static_cast<float>(image_size);

                        loc->SetImageNumber(image_number);
                        loc->SetImageSize(serializer.GetBufferSize());
                        loc->SetIndexed(message.indexing_result.value_or(false));
                        loc->ReadyToSend();

                        if (zmq_preview_socket != nullptr)
                            zmq_preview_socket->SendImage(writer_buffer, serializer.GetBufferSize());

                        if (zmq_metadata_socket != nullptr)
                            zmq_metadata_socket->AddDataMessage(message);

                        if (push_images_to_writer) {
                            // Only count images the pusher accepted; TCP refuses them on a broken/absent
                            // connection. Images actually written are GetImagesWritten() (ACK-based).
                            if (image_pusher.SendImage(*loc))
                                ++images_sent;
                        } else
                            loc->release();
                        UpdateMaxImageSent(message.number);
                    } catch (const CompressionBufferTooSmallException &) {
                        // The per-image CBOR metadata (spot/reflection lists, ...) leaves
                        // too little room for the compressed image in the buffer slot. Drop
                        // just this frame instead of aborting the whole data collection.
                        logger.Error("Dropping image {} - CBOR metadata too large to store image: "
                                     "metadata {} do not fit in buffer slot {} B",
                                     message.number, metadata_size, slot_size);
                        loc->release();
                    }
                }
            }

            logger.Debug("Frame transformation thread - done sending image {} / {}", image_number, message.number);

            plots.Add(message, az_int_profile_image);
            current_status.SetProgress(GetProgress());
            current_status.SetStatus(GetStatus());
        } catch (const JFJochException &e) {
            logger.ErrorException(e);
            Cancel(e);
        }
    }
    logger.Debug("Sum&compression thread done");
}

float JFJochReceiverFPGA::GetEfficiency() const {

    uint64_t expected_packets;
    if (experiment.GetImageNum() == 0)
        expected_packets = expected_packets_per_image * experiment.GetFrameNum();
    else
        expected_packets = expected_packets_per_image * experiment.GetImageNum();
    uint64_t received_packets = 0;

    for (int d = 0; d < ndatastreams; d++) {
        received_packets += acquisition_device[d].Counters().GetTotalPackets();
    }

    if ((expected_packets == received_packets) || (expected_packets == 0))
        return 1.0;
    return received_packets / static_cast<double>(expected_packets);
}

void JFJochReceiverFPGA::Cancel(bool silent) {
   JFJochReceiver::Cancel(silent);

    for (int d = 0; d < ndatastreams; d++)
        acquisition_device[d].Cancel();
}

void JFJochReceiverFPGA::Cancel(const JFJochException &e) {
    JFJochReceiver::Cancel(e);

    for (int d = 0; d < ndatastreams; d++)
        acquisition_device[d].Cancel();
}

float JFJochReceiverFPGA::GetProgress() const {
    int64_t frames = experiment.GetImageNum();
    if (experiment.IsCPUSummation())
        frames *= experiment.GetSummation();

    if (frames == 0)
        frames = experiment.GetFrameNum();
    if ((frames == 0) || (acquisition_device[0].Counters().IsAcquisitionFinished()))
        return 1.0;

    return static_cast<float>(acquisition_device[0].Counters().GetSlowestFrameNumber()) / static_cast<float>(frames);
}

void JFJochReceiverFPGA::FinalizeMeasurement() {
    if (!frame_transformation_futures.empty()) {
        for (auto &future: frame_transformation_futures)
            future.get();

        logger.Debug("All processing threads done");
    }

    // Where the start message never went out (or the analysis could not be set up), there is no end
    // message either: the devices are stopped all the same, and the error is the collection's result.
    std::exception_ptr gpu_error;
    if (gpu) {
        try {
            gpu->Finish();
        } catch (const std::exception &) {
            gpu_error = std::current_exception();
        }
    }

    current_status.SetProgress(1.0);
    current_status.SetStatus(GetStatus());

    if (!gpu_error)
        SendEndMessage();

    if (experiment.GetImageNum() > 0) {
        for (int d = 0; d < ndatastreams; d++)
            acquisition_device[d].Cancel();
    }

    end_time = std::chrono::system_clock::now();

    for (auto &future: data_acquisition_futures)
        future.get();

    logger.Debug("Devices stopped");

    if (!image_buffer.CheckIfBufferReturned(SendFinalizeTimeout)) {
        logger.Error("Send commands not finalized in {} seconds", SendFinalizeTimeout.count());
        throw JFJochException(JFJochExceptionCategory::ZeroMQ, "Send commands not finalized in "
                                                               + std::to_string(SendFinalizeTimeout.count()) + " seconds");
    }
    if (gpu_error) {
        current_status.SetProgress({});
        current_status.SetStatus(GetStatus());
        std::rethrow_exception(gpu_error);
    }
    logger.Info("Receiving data done");

    if (push_images_to_writer)
        writer_error = image_pusher.Finalize();
    // Last image -> all written, as the rugnux receiver reports it
    if (gpu)
        gpu->Written();

    current_status.SetProgress({});
    current_status.SetStatus(GetStatus());

    logger.Debug("Writing process finalized");
}

void JFJochReceiverFPGA::SetSpotFindingSettings(const SpotFindingSettings &in_spot_finding_settings) {
    std::unique_lock ul(spot_finding_settings_mutex);
    DiffractionExperiment::CheckDataProcessingSettings(in_spot_finding_settings);
    spot_finding_settings = in_spot_finding_settings;
    for (int i = 0; i < ndatastreams; i++)
        acquisition_device[i].SetSpotFinderParameters(spot_finding_settings);
}

void JFJochReceiverFPGA::StopReceiver() {
    if (measurement.valid()) {
        measurement.get();
        logger.Debug("Receiver stopped");
    }
}

JFJochReceiverFPGA::~JFJochReceiverFPGA() {
    try {
        if (measurement.valid())
            measurement.get();
    } catch (const std::exception& e) {
        logger.Error("ReceiverFPGA teardown failed: {}", e.what());
    } catch (...) {
        logger.Error("ReceiverFPGA teardown failed with unknown exception");
    }
}

JFJochReceiverOutput JFJochReceiverFPGA::GetFinalStatistics() const {
    JFJochReceiverOutput ret = JFJochReceiver::GetFinalStatistics();
    for (int d = 0; d < ndatastreams; d++) {
        for (int m = 0; m < acquisition_device[d].Counters().GetModuleNumber(); m++) {
            if (experiment.IsCPUSummation())
                ret.expected_packets.push_back(acquisition_device[d].Counters().GetTotalExpectedPacketsPerModule() * experiment.GetSummation());
            else
                ret.expected_packets.push_back(acquisition_device[d].Counters().GetTotalExpectedPacketsPerModule());
            ret.received_packets.push_back(acquisition_device[d].Counters().GetTotalPackets(m));
        }
    }
    RetrievePedestal(ret.pedestal_result);
    return ret;
}

void JFJochReceiverFPGA::RetrievePedestal(std::vector<JFModulePedestal> &output) const {
    time_t curr_time = std::chrono::system_clock::to_time_t(start_time);
    for (const auto &pc: pedestal) {
        JFModulePedestal mp;
        if (experiment.GetDetectorMode() == DetectorMode::PedestalG0)
            pc->Export(mp, PEDESTAL_G0_WRONG_GAIN_ALLOWED_COUNT);
        else
            pc->Export(mp);
        mp.SetCollectionTime(curr_time);
        output.emplace_back(std::move(mp));
    }
}

void JFJochReceiverFPGA::LoadCalibrationToFPGA(uint16_t data_stream) {
    if (experiment.IsPedestalRun()) {
        acquisition_device[data_stream].InitializeEmptyPixelMask(experiment);
        return; // No calibration loaded for pedestal
    }

    if (calibration != nullptr)
        acquisition_device[data_stream].InitializeCalibration(experiment, *calibration);

    // Initialize pixel_mask
    acquisition_device[data_stream].InitializePixelMask(experiment, pixel_mask);

    // Initialize roi_map
    acquisition_device[data_stream].InitializeROIMap(experiment, roi_map);

    // Initialize data processing
    acquisition_device[data_stream].InitializeDataProcessing(experiment, *az_int_mapping);
}

void JFJochReceiverFPGA::GPUSink(DataMessage &message, const AzimuthalIntegrationProfile &profile) {
    try {
        const auto status = image_buffer.GetStatus();
        message.receiver_buf_available = status.available_slots;
        message.receiver_buf_in_preparation = status.preparation_slots;
        message.receiver_buf_in_sending = status.sending_slots;
        if (message.image.GetCompressedSize() > 0)
            message.compression_ratio = static_cast<float>(experiment.GetPixelsNum() * experiment.GetByteDepthImage())
                                        / static_cast<float>(message.image.GetCompressedSize());

        if (message.image_collection_efficiency == 1.0f) {
            saturated_pixels.Add(message.saturated_pixel_count);
            error_pixels.Add(message.error_pixel_count);
            if (message.roi.contains("beam")) {
                roi_beam_npixel.Add(message.roi["beam"].pixels);
                roi_beam_sum.Add(message.roi["beam"].sum);
            }
        }

        scan_result.Add(message);

        if (!serialmx_filter.ApplyFilter(message))
            ++images_skipped;
        else if (auto loc = image_buffer.GetImageSlot(); loc == nullptr)
            writer_queue_full = true;
        else {
            auto writer_buffer = (uint8_t *) loc->GetImage();
            CBORStream2Serializer serializer(writer_buffer, experiment.GetImageBufferLocationSize());
            if (JFJochReceiverLite::SerializeImageOrDrop(serializer, message, *loc,
                                                         experiment.GetImageBufferLocationSize(), logger)) {
                loc->SetImageNumber(message.number);
                loc->SetImageSize(serializer.GetBufferSize());
                loc->SetIndexed(message.indexing_result.value_or(false));
                loc->ReadyToSend();

                if (zmq_preview_socket != nullptr)
                    zmq_preview_socket->SendImage(writer_buffer, serializer.GetBufferSize());
                if (zmq_metadata_socket != nullptr)
                    zmq_metadata_socket->AddDataMessage(message);

                if (push_images_to_writer) {
                    if (image_pusher.SendImage(*loc))
                        ++images_sent;
                } else
                    loc->release();
                UpdateMaxImageSent(message.number);
            }
        }

        plots.Add(message, profile);
        current_status.SetProgress(GetProgress());
        current_status.SetStatus(GetStatus());
    } catch (const JFJochException &e) {
        logger.ErrorException(e);
        Cancel(e);
    } catch (const std::exception &e) {
        const JFJochException error(JFJochExceptionCategory::WrongDAQState,
                                    fmt::format("Image {} not written - {}", message.number, e.what()));
        logger.ErrorException(error);
        Cancel(error);
    }
}

void JFJochReceiverFPGA::AddToEndMessage(EndMessage &message) {
    if (!gpu)
        return;
    message.unit_cell = gpu->GetConsensusUnitCell();
    message.refined_beam_center_x = gpu->GetBeamX_pxl();
    message.refined_beam_center_y = gpu->GetBeamY_pxl();
    message.refined_detector_distance_mm = gpu->GetDetectorDistance_mm();
    // The merged data, for the writer to put beside the master file.
    if (const auto &m = gpu->GetMerged(); m && gpu->GetMergeExperiment().GetUnitCell())
        message.merged = MergedMtzTable(m->merge.merged, *gpu->GetMergeExperiment().GetUnitCell(),
                                        gpu->GetMergeExperiment(), OutputProvenance{});
    if (!message.merged)
        return;
    // Every pusher serialises END into a buffer of MESSAGE_SIZE_FOR_START_END, and an END that does not fit there
    // fails the collection. So it is serialised here first, as JFJochReceiverRugnux does; one that does not fit
    // goes without the merged data.
    const auto scratch = std::make_unique_for_overwrite<uint8_t[]>(MESSAGE_SIZE_FOR_START_END);
    CBORStream2Serializer serializer(scratch.get(), MESSAGE_SIZE_FOR_START_END);
    try {
        serializer.SerializeSequenceEnd(message);
    } catch (const JFJochException &e) {
        const std::string text = fmt::format("the merged data do not fit in the end message ({} MiB) and are not "
                                             "written", MESSAGE_SIZE_FOR_START_END / (1024 * 1024));
        logger.Error("GPU analysis: {} - {}", text, e.what());
        if (rugnux_status)
            rugnux_status->Warning(text);
        message.merged.reset();
    }
}