Jungfraujoch/receiver/JFJochReceiver.h

// Copyright (2019-2023) Paul Scherrer Institute

#ifndef JUNGFRAUJOCH_JFJOCHRECEIVER_H
#define JUNGFRAUJOCH_JFJOCHRECEIVER_H

#include <optional>
#include <condition_variable>
#include <atomic>
#include <future>
#include <latch>
#include <semaphore>

#include "../acquisition_device/AcquisitionDeviceGroup.h"

#include "../common/DiffractionExperiment.h"
#include "../common/JFJochException.h"
#include "../common/FrameTransformation.h"
#include "../common/ImagePusher.h"
#include "../common/Logger.h"
#include "../common/ThreadSafeFIFO.h"
#include "../common/ZMQPreviewPublisher.h"
#include "../common/NUMAHWPolicy.h"
#include "../common/StatusVector.h"
#include "../common/Histogram.h"

#include "../image_analysis/StrongPixelSet.h"
#include "../image_analysis/RadialIntegrationMapping.h"
#include "../image_analysis/RadialIntegrationProfile.h"

#include "../jungfrau/JFCalibration.h"
#include "../common/ADUHistogram.h"
#include "../common/Plot.h"

struct JFJochReceiverOutput {
    std::vector<JFModulePedestal> pedestal_result;
    std::vector<uint64_t> received_packets;
    std::vector<uint64_t> expected_packets;
    std::string master_file_name;
    uint64_t max_receive_delay;
    uint64_t compressed_size;
    uint64_t max_image_number_sent;
    uint64_t images_sent;
    uint64_t start_time_ms;
    uint64_t end_time_ms;
    float compressed_ratio;
    float efficiency;
    float indexing_rate;
    float bkg_estimate;
    bool cancelled;
};

class JFJochReceiver {
    const DiffractionExperiment &experiment;
    const JFCalibration *calibration;
    std::vector<JFModulePedestal> pedestal_result;

    std::vector<uint8_t> one_byte_mask;
    Logger &logger;

    std::vector<std::future<void>> frame_transformation_futures;
    std::vector<std::future<void>> data_acquisition_futures;

    ADUHistogram adu_histogram_total;
    std::vector<ADUHistogram> adu_histogram_module;

    std::unique_ptr<RadialIntegrationMapping> rad_int_mapping;
    std::unique_ptr<RadialIntegrationProfile> rad_int_profile;
    std::vector<std::unique_ptr<RadialIntegrationProfile>> rad_int_profile_per_file;
    std::vector<float> rad_int_corr;
    std::vector<float> rad_int_corr_raw;

    std::future<void> measurement;

    ThreadSafeFIFO<uint64_t> images_to_go;

    ZMQPreviewPublisher *preview_publisher = nullptr;

    ImagePusher &image_pusher;
    bool push_images_to_writer;

    volatile bool cancelled{false};

    AcquisitionDeviceGroup &acquisition_device;
    uint16_t ndatastreams{0};

    int64_t max_delay = 0;
    std::mutex max_delay_mutex;

    std::atomic<size_t> compressed_size{0};
    std::atomic<size_t> images_sent{0};

    const int64_t frame_transformation_nthreads;

    std::latch frame_transformation_ready;
    std::latch data_acquisition_ready;

    std::chrono::time_point<std::chrono::system_clock> start_time;
    std::chrono::time_point<std::chrono::system_clock> end_time;

    DataProcessingSettings data_processing_settings;
    std::mutex data_processing_settings_mutex;

    StatusVector<float> bkg_estimate;
    StatusVector<uint64_t> spot_count;
    StatusVector<uint64_t> indexing_solution;
    Histogram<uint64_t> indexing_solution_per_file;
    // TODO: Histogram a,b,c,alpha,beta,gamma

    bool find_spots = false;

    int64_t max_image_number_sent = 0;
    std::mutex max_image_number_sent_mutex;

    const size_t send_buffer_size = experiment.GetMaxCompressedSize() + 1024*1024;
    // max compressed size + 1 MiB reserve for spots and rad. integration results
    const size_t send_buffer_count;

    ThreadSafeFIFO<uint32_t> send_buffer_avail;
    uint8_t *send_buffer;
    std::vector<ZeroCopyReturnValue> send_buffer_zero_copy_ret_val;

    NUMAHWPolicy numa_policy;

    void AcquireThread(uint16_t data_stream);
    void FrameTransformationThread();
    void MeasurePedestalThread(uint16_t data_stream, uint16_t module_number, uint16_t storage_cell);
    void Cancel(const JFJochException &e);
    void FinalizeMeasurement();
    DataProcessingSettings GetDataProcessingSettings();

    void UpdateMaxImage(int64_t image_number);
    void UpdateMaxDelay(int64_t delay);
public:
    JFJochReceiver(const DiffractionExperiment& experiment,
                   const JFCalibration *calibration,
                   AcquisitionDeviceGroup &acquisition_devices,
                   ImagePusher &image_pusher,
                   Logger &logger, int64_t forward_and_sum_nthreads,
                   int64_t send_buffer_count,
                   ZMQPreviewPublisher* preview_publisher,
                   const NUMAHWPolicy &numa_policy);
    ~JFJochReceiver();
    JFJochReceiver(const JFJochReceiver &other) = delete;
    JFJochReceiver& operator=(const JFJochReceiver &other) = delete;
    void StopReceiver();
    JFJochReceiverOutput GetStatistics() const;

    float GetEfficiency() const;
    void Cancel();
    float GetProgress() const;
    float GetIndexingRate() const;
    void SetDataProcessingSettings(const DataProcessingSettings &data_processing_settings);

    float GetAvailableSendBuffers() const;
    Plot GetPlots(const PlotRequest& request);
    RadialIntegrationProfiles GetRadialIntegrationProfiles();
};

#endif //JUNGFRAUJOCH_JFJOCHRECEIVER_H