Jungfraujoch/broker/JFJochStateMachine.cpp

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

#include <thread>

#include "JFJochStateMachine.h"
#include "../preview/JFJochTIFF.h"
#include "../common/CUDAWrapper.h"
#include "../common/GitInfo.h"
#include "../common/JFJochException.h"

JFJochStateMachine::JFJochStateMachine(const DiffractionExperiment& in_experiment,
                                       JFJochServices &in_services,
                                       Logger &in_logger,
                                       const SpotFindingSettings &spot_finding_settings)
        : logger(in_logger),
          services(in_services),
          experiment(in_experiment),
          pixel_mask(experiment),
          current_detector_setup(0),
          data_processing_settings(spot_finding_settings),
          pixel_mask_statistics({0, 0, 0}),
          gpu_count(get_gpu_count()) {

    SuppressTIFFErrors();
}

bool JFJochStateMachine::ImportPedestalG0(const JFJochReceiverOutput &receiver_output) {
    if (receiver_output.pedestal_result.empty())
        return false;

    if (receiver_output.pedestal_result.size() != experiment.GetModulesNum() * experiment.GetStorageCellNumber())
        throw JFJochException(JFJochExceptionCategory::InputParameterInvalid, "Mismatch in pedestal output");

    size_t gain_level = experiment.IsFixedGainG1() ? 1 : 0;

    for (int s = 0; s < experiment.GetStorageCellNumber(); s++) {
        for (int module = 0; module < experiment.GetModulesNum(); module++)
            calibration->Pedestal(module, gain_level, s)
                    = receiver_output.pedestal_result[module + s * experiment.GetModulesNum()];
    }
    SetCalibrationStatistics(calibration->GetModuleStatistics());
    return true;
}

bool JFJochStateMachine::ImportPedestalG1G2(const JFJochReceiverOutput &receiver_output, size_t gain_level,
                                            size_t storage_cell) {
    if (receiver_output.pedestal_result.empty())
        return false;

    if (receiver_output.pedestal_result.size() != experiment.GetModulesNum())
        throw JFJochException(JFJochExceptionCategory::InputParameterInvalid, "Mismatch in pedestal output");

    for (int i = 0; i < receiver_output.pedestal_result.size(); i++)
        calibration->Pedestal(i, gain_level, storage_cell) = receiver_output.pedestal_result[i];
    SetCalibrationStatistics(calibration->GetModuleStatistics());
    return true;
}

bool JFJochStateMachine::CalibrateJUNGFRAU(std::unique_lock<std::mutex> &ul) {
    if (!gain_calibration.empty()) {
        if (gain_calibration.size() != experiment.GetModulesNum())
            throw JFJochException(JFJochExceptionCategory::InputParameterInvalid,
                                  "Mismatch in gain files number");
        for (int i = 0; i < gain_calibration.size(); i++)
            calibration->GainCalibration(i) = gain_calibration[i];
    }

    // Abandon the sequence on the first failure. Collecting G1 on top of a G0 that was never
    // measured only produces a calibration that looks complete.
    if (!TakePedestalInternalG0(ul))
        return false;
    if (!experiment.IsFixedGainG1()) {
        for (int i = 0; i < experiment.GetStorageCellNumber(); i++) {
            if (!TakePedestalInternalG1(ul, i))
                return false;
            if (!TakePedestalInternalG2(ul, i))
                return false;
        }
    }
    pixel_mask.LoadDetectorBadPixelMask(experiment, calibration.get());
    return true;
}

void JFJochStateMachine::CalibrateDetector(std::unique_lock<std::mutex> ul) {
    cancel_sequence = false;

    pixel_mask = PixelMask(experiment);
    UpdatePixelMaskStatistics(pixel_mask.GetStatistics());

    logger.Info("Calibration sequence started");
    bool calibrated;
    try {
        if (experiment.GetDetectorType() == DetectorType::EIGER) {
            // PSI EIGER - only reset calibration
            calibration.reset();
            calibrated = true;
        } else if (experiment.GetDetectorType() == DetectorType::DECTRIS) {
            // DECTRIS - take dark data for mask
            calibration.reset();
            calibrated = TakeDarkMaskInternal(ul);
        } else {
            // PSI JUNGFRAU - take pedestal
            calibration = std::make_unique<JFCalibration>(experiment);
            calibrated = CalibrateJUNGFRAU(ul);
        }
        // Update pixel mask statistics
        UpdatePixelMaskStatistics(pixel_mask.GetStatistics());
        // configure detector for standard operation - only worth doing if there is a calibration to
        // operate with, and a cancelled sequence has left the detector mid-abort anyway
        if (calibrated)
            services.ConfigureDetector(experiment);
    } catch (const std::exception &e) {
        logger.Error("Calibration sequence error {}", e.what());
        // The calibration is in an undefined state, so the detector has to be initialized again.
        // Error rather than Inactive: /wait_till_done and /wait_until_running answer Error with a 500
        // carrying this message, while Inactive is a bodiless 502 that loses the reason - and Inactive
        // is also what a deliberate Deactivate() leaves behind, so a failure would be indistinguishable
        // from a detector someone powered down. Initialize() clears Error, so recovery is unchanged.
        SetState(JFJochState::Error, e.what(), BrokerStatus::MessageSeverity::Error);
        c.notify_all(); // ul unlocks on the way out
        throw;
    }
    // The steps above report a cancellation or a failure through SetState and return false; that
    // must not be overwritten with success here.
    if (calibrated) {
        SetState(JFJochState::Idle, "Calibration sequence done", BrokerStatus::MessageSeverity::Success);
        logger.Info("Calibration sequence done");
    }
    ul.unlock(); // Notify all outside of mutex
    c.notify_all();
}

bool JFJochStateMachine::TakeDarkMaskInternal(std::unique_lock<std::mutex> &ul) {
    if (cancel_sequence) {
        SetState(JFJochState::Inactive,
                 "Mask sequence cancelled",
                 BrokerStatus::MessageSeverity::Error);
        return false;
    }

    services.LoadDetectorPixelMask(pixel_mask);

    if (experiment.GetDarkMaskNumberOfFrames() == 0)
        return true;

    DiffractionExperiment local_experiment(experiment);
    local_experiment.Mode(DetectorMode::DarkMask);

    SetState(JFJochState::Calibration, "Dark sequence for mask calculation", BrokerStatus::MessageSeverity::Info);
    services.ConfigureDetector(local_experiment);
    services.Start(local_experiment, pixel_mask, nullptr);

    services.Trigger();

    ul.unlock();
    // Allow to cancel/abort during the mask data collection
    auto mask_output = services.Stop();
    ul.lock();

    // A cancelled collection still hands back a full-size mask - it is built from whatever dark
    // frames arrived, and every pixel that saw none reads as good - so the size check below accepts
    // it and the sequence goes on to report a calibration it never took as done. The dark mask is a
    // single step, so unlike a pedestal sequence there is no later step whose cancel_sequence check
    // would catch it. Refuse it the way the check at the top of this function does.
    if (cancel_sequence || mask_output.receiver_output.status.cancelled) {
        SetState(JFJochState::Inactive,
                 "Mask sequence cancelled",
                 BrokerStatus::MessageSeverity::Error);
        return false;
    }

    if (mask_output.receiver_output.dark_mask_result.size() != local_experiment.GetPixelsNum()) {
        SetState(JFJochState::Error, "Mask not collected properly", BrokerStatus::MessageSeverity::Error);
        return false;
    }
    pixel_mask.LoadDarkBadPixelMask(local_experiment, mask_output.receiver_output.dark_mask_result);
    SetState(JFJochState::Idle);
    return true;
}

bool JFJochStateMachine::TakePedestalInternalG0(std::unique_lock<std::mutex> &ul) {
    DiffractionExperiment local_experiment(experiment);
    std::string message;
    if (local_experiment.IsFixedGainG1()) {
        local_experiment.Mode(DetectorMode::PedestalG1);
        message = "Pedestal G1";
    } else {
        local_experiment.Mode(DetectorMode::PedestalG0);
        message = "Pedestal G0";
    }

    if (local_experiment.GetStorageCellNumber() == 1)
        local_experiment.StorageCellStart(15);
    else
        local_experiment.StorageCellStart(0);

    if (cancel_sequence) {
        SetState(JFJochState::Inactive,
                 "Pedestal sequence cancelled",
                 BrokerStatus::MessageSeverity::Error);
        return false;
    }

    if (local_experiment.GetPedestalG0Frames() == 0)
        return true;

    SetState(JFJochState::Calibration, message, BrokerStatus::MessageSeverity::Info);
    services.ConfigureDetector(local_experiment);
    services.Start(local_experiment, pixel_mask, calibration.get());

    services.Trigger();

    ul.unlock();
    // Allow to cancel/abort during the pedestal data collection
    // Must ensure that while state is Pedestal, nothing can take lock for longer time, to avoid deadlock
    auto pedestal_output = services.Stop();
    ul.lock();

    if (!ImportPedestalG0(pedestal_output.receiver_output)) {
        SetState(JFJochState::Error,
                 "Pedestal not collected properly",
                 BrokerStatus::MessageSeverity::Error);
        return false;
    }
    SetState(JFJochState::Idle);
    return true;
}

bool JFJochStateMachine::TakePedestalInternalG1(std::unique_lock<std::mutex> &ul, int32_t storage_cell) {
    DiffractionExperiment local_experiment(experiment);
    local_experiment.Mode(DetectorMode::PedestalG1);

    if (local_experiment.GetStorageCellNumber() == 2)
        local_experiment.StorageCellStart((storage_cell + 15) % 16); // one previous
    else
        local_experiment.StorageCellStart(15);


    if (cancel_sequence) {
        SetState(JFJochState::Inactive,
                 "Pedestal sequence cancelled",
                 BrokerStatus::MessageSeverity::Error);
        return false;
    }

    if (local_experiment.GetPedestalG1Frames() == 0)
        return true;


    SetState(JFJochState::Calibration,
             "Pedestal G1   SC" + std::to_string(storage_cell),
             BrokerStatus::MessageSeverity::Info);
    services.ConfigureDetector(local_experiment);
    services.Start(local_experiment, pixel_mask, calibration.get());

    services.Trigger();

    ul.unlock();
    // Allow to cancel/abort during the pedestal data collection
    // Must ensure that while state is Pedestal, nothing can take lock for longer time, to avoid deadlock
    auto pedestal_output = services.Stop();
    ul.lock();

    if (!ImportPedestalG1G2(pedestal_output.receiver_output, 1, storage_cell)) {
        SetState(JFJochState::Error,
                 "Pedestal not collected properly",
                 BrokerStatus::MessageSeverity::Error);
        return false;
    }
    return true;
}

bool JFJochStateMachine::TakePedestalInternalG2(std::unique_lock<std::mutex> &ul, int32_t storage_cell) {
    DiffractionExperiment local_experiment(experiment);
    local_experiment.Mode(DetectorMode::PedestalG2);

    if (local_experiment.GetStorageCellNumber() == 2)
        local_experiment.StorageCellStart((storage_cell + 15) % 16); // one previous
    else
        local_experiment.StorageCellStart(15);

    if (cancel_sequence) {
        SetState(JFJochState::Inactive,
                 "Pedestal sequence cancelled",
                 BrokerStatus::MessageSeverity::Error);
        return false;
    }

    if (local_experiment.GetPedestalG2Frames() == 0)
        return true;


    SetState(JFJochState::Calibration,
             "Pedestal G2   SC" + std::to_string(storage_cell),
             BrokerStatus::MessageSeverity::Info);
    services.ConfigureDetector(local_experiment);
    services.Start(local_experiment, pixel_mask, calibration.get());

    services.Trigger();

    ul.unlock();
    // Allow to cancel/abort during the pedestal data collection
    // Must ensure that while state is Pedestal, nothing can take lock for longer time, to avoid deadlock
    auto pedestal_output = services.Stop();
    ul.lock();

    if (!ImportPedestalG1G2(pedestal_output.receiver_output, 2, storage_cell)) {
        SetState(JFJochState::Error,
                 "Pedestal not collected properly",
                 BrokerStatus::MessageSeverity::Error);
        return false;
    }
    return true;
}

void JFJochStateMachine::Initialize() {
    std::unique_lock ul(m);

    if (IsRunning())
        throw WrongDAQStateException("Cannot initialize during measurement");

    if (detector_setup.empty())
        throw JFJochException(JFJochExceptionCategory::InputParameterInvalid, "Detector information not provided");

    ResetError(); // Clear error, we don't care what was it
    start_exception = nullptr; // Re-initialising discards a pending start failure

    logger.Info("Initialize");
    SetState(JFJochState::Busy, "Configuring indexing threads", BrokerStatus::MessageSeverity::Info);
    try {
        services.SetupIndexing(experiment.GetIndexingSettings());
    } catch (const JFJochException &e) {
        SetState(JFJochState::Error, e.what(), BrokerStatus::MessageSeverity::Error);
        throw;
    }
    SetState(JFJochState::Busy, "Configuring detector", BrokerStatus::MessageSeverity::Info);

    scan_result = {}; // Clear scan result

    measurement = std::async(std::launch::async, &JFJochStateMachine::InitializeThread, this, std::move(ul));
}

void JFJochStateMachine::Pedestal() {
    std::unique_lock ul(m);

    if (state != JFJochState::Idle)
        throw WrongDAQStateException("Must be idle to take pedestal");

    start_exception = nullptr; // A new operation supersedes a pending start failure
    SetState(JFJochState::Busy, "Updating calibration", BrokerStatus::MessageSeverity::Info);

    measurement = std::async(std::launch::async, &JFJochStateMachine::CalibrateDetector, this, std::move(ul));
}

void JFJochStateMachine::InitializeThread(std::unique_lock<std::mutex> ul) {
    try {
        // services.On can potentially take a lot of time, so better to unlock main mutex
        // Since On might modify the experiment (reads DECTRIS configuration), one has to have a local copy for unlocked part
        DiffractionExperiment local_experiment(experiment);
        if (state != JFJochState::Busy)
            throw JFJochException(JFJochExceptionCategory::InputParameterInvalid,
                                  "State must be busy for safe operation");
        ul.unlock();
        services.On(local_experiment);
        ul.lock();

        experiment = local_experiment;
        detector_setup[current_detector_setup] = experiment.GetDetectorSetup();

    } catch (const std::exception &e) {
        logger.Error("Initialize error {}", e.what());
        SetState(JFJochState::Error, e.what(), BrokerStatus::MessageSeverity::Error);
        // Wake anyone in WaitTillNotBusy/WaitTillMeasurementDone - the state has left Busy, and
        // without this they sleep out their whole timeout before noticing.
        c.notify_all(); // ul unlocks on the way out
        throw;
    }
    CalibrateDetector(std::move(ul));
}

void JFJochStateMachine::Trigger() {
    services.Trigger();
}

void JFJochStateMachine::Start(const DatasetSettings &settings, bool async, std::vector<std::string> in_tokens) {
    std::unique_lock ul(m);

    if (state != JFJochState::Idle)
        throw WrongDAQStateException("Must be idle to start measurement");

    if (measurement.valid())
        measurement.get(); // In case measurement was running - clear thread

    // Clear before ImportDatasetSettings, which can throw: a rejected /start must not leave the
    // previous run's failure behind for the next wait call to report.
    start_exception = nullptr;

    // The same for the previous run's scan result. It is only ever replaced at the END of a
    // successful measurement, so a run that fails to start - or is rejected here - would otherwise
    // leave /scan_result answering with the last run's images, under this run's number and with
    // nothing to say they are not its own. Initialize() already clears it for the same reason.
    scan_result = {};

    // The settings are checked on a copy first, so a rejected start changes nothing below: the
    // previous run stays readable to its own token holders, and nothing of the new one is seen.
    {
        DiffractionExperiment check = experiment;
        check.ImportDatasetSettings(settings);
    }

    // The order matters. First nothing of the previous run is left readable, then the tokens
    // become the new run's, then the new run's name and cell go into the experiment: at no moment
    // are the previous run's results served under the new tokens, nor the new run's identity
    // under the previous ones. (The receiver resets these structures again when it starts.)
    services.ClearResults();
    tokens.Replace(std::move(in_tokens));
    experiment.ImportDatasetSettings(settings);

    cancel_sequence = false;
    if (experiment.GetStorageCellNumber() == 1)
        experiment.StorageCellStart(15);
    else
        experiment.StorageCellStart(0);

    experiment.IncrementRunNumber();

    SetState(JFJochState::Busy, "Preparing measurement", BrokerStatus::MessageSeverity::Info);
    measurement = std::async(std::launch::async, &JFJochStateMachine::MeasurementThread, this);
    if (!async) {
        c.wait(ul, [&]() { return state != JFJochState::Busy; });
        // A synchronous start propagates the failure to the caller. The state has already been set
        // by MeasurementThread (Idle for an ordinary failure, Error for a critical detector fault).
        // start_exception is left in place - the next Start() or Initialize() clears it - so that a
        // wait call made afterwards reports the same failure instead of an apparent timeout.
        if (start_exception)
            std::rethrow_exception(start_exception);
    }
}

bool JFJochStateMachine::AcceptToken(const std::optional<std::string> &authorization) const {
    return tokens.Accept(authorization);
}

BrokerStatus JFJochStateMachine::WaitTillNotBusy(std::chrono::milliseconds timeout) {
    std::unique_lock ul(m);
    c.wait_for(ul, timeout, [&]() { return state != JFJochState::Busy; });
    // An asynchronous start reports its failure here, since /start itself returned before the
    // measurement thread ran. Without this the state is plain Idle and the caller cannot tell a
    // failed start from a timeout. rethrow_exception does not consume the exception_ptr, so
    // repeated calls all report the same failure.
    if (start_exception)
        std::rethrow_exception(start_exception);
    return GetStatus();
}

void JFJochStateMachine::UpdatePixelMaskStatistics(const PixelMaskStatistics &input) {
    std::unique_lock ul(pixel_mask_statistics_mutex);
    pixel_mask_statistics = input;
}

PixelMaskStatistics JFJochStateMachine::GetPixelMaskStatistics() const {
    std::unique_lock ul(pixel_mask_statistics_mutex);
    return pixel_mask_statistics;
}

void JFJochStateMachine::MeasurementThread() {
    try {
        // Before anything is started: ask the writer whether it could write this run. A run that
        // would be refused for a file already there, or a directory that cannot be made, is
        // refused now - with no receiver built and, above all, no detector armed, so there is
        // nothing to tear down and no series left waiting for a trigger that will not come.
        services.Preflight(experiment);
        services.SetSpotFindingSettings(GetSpotFindingSettings());
        services.Start(experiment, pixel_mask, calibration.get());
        {
            std::unique_lock ul(m);
            SetState(JFJochState::Measuring, "Measuring ...", BrokerStatus::MessageSeverity::Info);
        }
        c.notify_all();
    } catch (const JFJochCriticalException &e) {
        // Detector left in an undefined state - force re-initialisation via the Error state.
        logger.Error("Critical error starting measurement: {}", e.what());
        {
            std::unique_lock ul(m);
            SetState(JFJochState::Error, e.what(), BrokerStatus::MessageSeverity::Error);
            start_exception = std::current_exception();
        }
        c.notify_all();
        return;
    } catch (const std::exception &e) {
        // Ordinary acquisition failure - the detector is still configured/calibrated, so return to
        // Idle and let the user retry without re-initialising. services.Start has already stopped the
        // receiver it launched.
        logger.Error("Error starting measurement: {}", e.what());
        {
            std::unique_lock ul(m);
            SetState(JFJochState::Idle, e.what(), BrokerStatus::MessageSeverity::Error);
            start_exception = std::current_exception();
        }
        c.notify_all();
        return;
    }

    try {
        auto tmp_output = services.Stop();
        {
            std::unique_lock ul(m);
            scan_result = tmp_output.receiver_output.scan_result;

            auto image_mean_time = tmp_output.receiver_output.processing_time;
            logger.Info("Per-image mean processing time (microseconds): compression {:.0f} preprocess {:.0f} azint {:.0f} spot finding {:.0f} indexing {:.0f} refinement {:.0f} indexing analysis {:.0f} prediction {:.0f} integration {:.0f} total {:.0f}",
                        image_mean_time.compression * 1e6,
                        image_mean_time.preprocessing * 1e6,
                        image_mean_time.azint * 1e6,
                        image_mean_time.spot_finding * 1e6,
                        image_mean_time.indexing * 1e6,
                        image_mean_time.refinement * 1e6,
                        image_mean_time.indexing_analysis * 1e6,
                        image_mean_time.bragg_prediction * 1e6,
                        image_mean_time.integration * 1e6,
                        image_mean_time.processing * 1e6);

            // Priority order matters. A cancel is checked first (it legitimately leaves efficiency < 1),
            // then the hard errors (missing packets, truncated writer output). The queue-full warning is
            // only the primary status when the run otherwise succeeded - it must not mask a real error
            // by downgrading an incomplete/truncated dataset to a "reduce frame rate" warning.
            if (tmp_output.receiver_output.status.cancelled)
                SetState(JFJochState::Idle,
                         "Data collection cancelled",
                         BrokerStatus::MessageSeverity::Info);
            else if (tmp_output.receiver_output.efficiency != 1.0)
                SetState(JFJochState::Idle,
                         "Missing packets in data collection; reduce frame rate",
                         BrokerStatus::MessageSeverity::Error);
            else if (!tmp_output.receiver_output.writer_err.empty())
                SetState(JFJochState::Idle,
                         "Writer error, written data may be incomplete: " + tmp_output.receiver_output.writer_err,
                         BrokerStatus::MessageSeverity::Error);
            else if (tmp_output.receiver_output.writer_queue_full_warning)
                SetState(JFJochState::Idle,
                         "Stream receiver (writer or downstream analysis) cannot cope with data; reduce frame rate",
                         BrokerStatus::MessageSeverity::Warning);
            else
                SetState(JFJochState::Idle,
                         "Data collection without problems",
                         BrokerStatus::MessageSeverity::Success);
        }
    } catch (const JFJochCriticalException &e) {
        // Detector faulted during the run - it needs re-initialisation, so go to the Error state.
        logger.Error("Critical error finishing measurement: {}", e.what());
        std::unique_lock ul(m);
        SetState(JFJochState::Error, e.what(), BrokerStatus::MessageSeverity::Error);
    } catch (const std::exception &e) {
        // Receiver/writer problem - the data may be incomplete, but the detector is still usable, so
        // return to Idle rather than forcing re-initialisation.
        logger.Error("Error finishing measurement: {}", e.what());
        std::unique_lock ul(m);
        SetState(JFJochState::Idle, e.what(), BrokerStatus::MessageSeverity::Error);
    }
    c.notify_all();
}

void JFJochStateMachine::Cancel() {
    // This is inconsistency in naming - need to solve later
    std::unique_lock ul(m);
    // Cancelling supersedes a pending start failure: the caller has taken the run in hand, and the
    // next wait must report what the cancel did, not what the start before it failed to do.
    start_exception = nullptr;
    if ((state == JFJochState::Calibration) || (state == JFJochState::Measuring)) {
        services.Cancel();
        cancel_sequence = true;
    }
}

void JFJochStateMachine::DebugOnly_SetState(JFJochState in_state,
                                            const std::optional<std::string> &message,
                                            BrokerStatus::MessageSeverity message_severity) {
    std::unique_lock ul(m);
    SetState(in_state, message, message_severity);
}

void JFJochStateMachine::Deactivate() {
    std::unique_lock ul(m);

    // Powering the detector off holds m for the whole sequence, so it must not be started while a
    // measurement, calibration or initialisation thread is still live - those re-acquire m to
    // finish, and waiting for them here would deadlock the whole control plane.
    if (IsRunning())
        throw WrongDAQStateException("Cannot deactivate while the detector is busy");

    // Powering down supersedes a pending start failure. Without this it outlives the detector it
    // belongs to: every later wait call keeps rethrowing a failure from a run on hardware that is no
    // longer even on, until the next Initialize() or Start() happens to clear it.
    start_exception = nullptr;

    // The same for the last run's scan result, and for the same reason: it describes a measurement on
    // a detector that is being turned off, and Initialize() - the only way back from here - clears it
    // anyway. Leaving it would let /scan_result answer for a session that has ended.
    scan_result = {};

    // Reap the finished thread, but do not let a failure it stored stop the power-off: the state is
    // Error precisely because that run failed, and leaving the detector powered is worse than
    // losing an error message that was already reported when it happened.
    if (measurement.valid()) {
        try {
            measurement.get();
        } catch (const std::exception &e) {
            logger.Warning("Deactivating after an earlier failure: {}", e.what());
        }
    }

    try {
        services.Off();
        SetState(JFJochState::Inactive,
                 "Detector safe to turn off",
                 BrokerStatus::MessageSeverity::Info);
    } catch (const std::exception &e) {
        SetState(JFJochState::Error,
                 e.what(),
                 BrokerStatus::MessageSeverity::Error);
        throw;
    }
}

JFJochStateMachine::~JFJochStateMachine() {
    ResetError();
}

std::optional<MeasurementStatistics> JFJochStateMachine::GetMeasurementStatistics() const {
    MeasurementStatistics tmp{};

    tmp.file_prefix = experiment.GetFilePrefix();
    tmp.run_number = experiment.GetRunNumber();
    tmp.experiment_group = experiment.GetExperimentGroup();

    tmp.detector_width = experiment.GetXPixelsNum();
    tmp.detector_height = experiment.GetYPixelsNum();
    tmp.detector_pixel_depth = experiment.GetByteDepthImage();
    tmp.images_expected = experiment.GetImageNum();
    tmp.unit_cell = experiment.GetUnitCellString();

    auto rcv_status = services.GetReceiverStatus();
    if (rcv_status) {
        tmp.compression_ratio = rcv_status->compressed_ratio;
        tmp.images_collected = rcv_status->images_collected;
        tmp.images_sent = rcv_status->images_sent;
        tmp.images_skipped = rcv_status->images_skipped;
        tmp.cancelled = rcv_status->cancelled;
        tmp.max_image_number_sent = rcv_status->max_image_number_sent;
        tmp.max_receive_delay = rcv_status->max_receive_delay;
        tmp.indexing_rate = rcv_status->indexing_rate;
        tmp.bkg_estimate = rcv_status->bkg_estimate;
        tmp.collection_efficiency = rcv_status->efficiency;
        tmp.error_pixels = rcv_status->error_pixels;
        tmp.saturated_pixels = rcv_status->saturated_pixels;
        tmp.roi_beam_sum = rcv_status->roi_beam_sum;
        tmp.roi_beam_npixel = rcv_status->roi_beam_npixel;
        tmp.images_written = rcv_status->images_written;
    }
    return tmp;
}

std::vector<JFCalibrationModuleStatistics> JFJochStateMachine::GetCalibrationStatistics() const {
    std::unique_lock ul(calibration_statistics_mutex);
    return calibration_statistics;
}

void JFJochStateMachine::SetCalibrationStatistics(const std::vector<JFCalibrationModuleStatistics> &input) {
    std::unique_lock ul(calibration_statistics_mutex);
    calibration_statistics = input;
}

DetectorSettings JFJochStateMachine::GetDetectorSettings() const {
    std::unique_lock ul(experiment_detector_settings_mutex);
    return experiment.GetDetectorSettings();
}

bool JFJochStateMachine::ImportDetectorSettings(const DetectorSettings &input) {
    std::unique_lock ul(experiment_detector_settings_mutex);
    // For JUNGFRAU detector, if detector settings changes key parameters
    // need to recalibrate the detector
    bool recalib = input.NeedsJUNGFRAURecalibration(experiment.GetDetectorSettings())
                   && experiment.GetDetectorType() == DetectorType::JUNGFRAU;
    experiment.ImportDetectorSettings(input);
    return recalib;
}

void JFJochStateMachine::LoadDetectorSettings(const DetectorSettings &settings) {
    std::unique_lock ul(m);
    switch (state) {
        case JFJochState::Inactive:
        case JFJochState::Error:
            ImportDetectorSettings(settings);
            break;
        case JFJochState::Idle:
            if (ImportDetectorSettings(settings)) {
                start_exception = nullptr; // A new operation supersedes a pending start failure
                SetState(JFJochState::Busy, "Loading settings", BrokerStatus::MessageSeverity::Info);
                measurement = std::async(std::launch::async, &JFJochStateMachine::CalibrateDetector, this, std::move(ul));
            } else {
                try {
                    SetState(JFJochState::Busy, "Configure detector", BrokerStatus::MessageSeverity::Info);
                    services.ConfigureDetector(experiment);
                    SetState(JFJochState::Idle, "Detector configured", BrokerStatus::MessageSeverity::Info);
                } catch (const std::exception &e) {
                    logger.Error("Detector configuration error {}", e.what());
                    SetState(JFJochState::Error, e.what(), BrokerStatus::MessageSeverity::Error);
                }
            }
            break;
        case JFJochState::Measuring:
        case JFJochState::Busy:
        case JFJochState::Calibration:
            throw WrongDAQStateException("Cannot change detector settings during data collection");
    }
}

DiffractionExperiment JFJochStateMachine::Experiment() {
    return experiment;
}

BrokerStatus JFJochStateMachine::GetStatus() const {
    std::unique_lock ul(broker_status_mutex);
    BrokerStatus ret = broker_status;
    ret.progress = services.GetReceiverProgress();
    ret.gpu_count = gpu_count;
    ret.broker_version = jfjoch_version();
    return ret;
}

void JFJochStateMachine::SetState(JFJochState curr_state,
                                  const std::optional<std::string> &message,
                                  BrokerStatus::MessageSeverity message_severity) {
    std::unique_lock ul(broker_status_mutex);
    state = curr_state;
    broker_status = BrokerStatus{
            .state = curr_state,
            .message = message,
            .message_severity = message_severity
    };
}

MultiLinePlot JFJochStateMachine::GetPlots(const PlotRequest &request) const {
    return services.GetPlots(request);
}

void JFJochStateMachine::GetPlotRaw(std::vector<float> &v, PlotType type, const std::string &roi) const {
    services.GetPlotRaw(v, type, roi);
}

void JFJochStateMachine::SetSpotFindingSettings(const SpotFindingSettings &settings) {
    std::unique_lock ul(data_processing_settings_mutex);
    DiffractionExperiment::CheckDataProcessingSettings(settings);
    // The adaptive threshold is a property of the software spot finder, which only the DECTRIS
    // (SIMPLON) workflow runs - JUNGFRAU and EIGER find spots on the FPGA, at its own fixed
    // threshold. Refuse it there rather than accept it and do nothing: a silently ignored detection
    // setting is indistinguishable from one that had no effect on the data.
    if (settings.adaptive_threshold && experiment.GetDetectorType() != DetectorType::DECTRIS)
        throw JFJochException(JFJochExceptionCategory::InputParameterInvalid,
                              "Adaptive spot-finding threshold is not available on this detector: spots "
                              "are found on the FPGA, which applies its own fixed threshold");

    data_processing_settings = settings;

    services.SetSpotFindingSettings(data_processing_settings);
}

SpotFindingSettings JFJochStateMachine::GetSpotFindingSettings() const {
    std::unique_lock ul(data_processing_settings_mutex);
    return data_processing_settings;
}

void JFJochStateMachine::AddDetectorSetup(const DetectorSetup &setup) {
    // Not thread safe, only during setup

    if (detector_setup.empty()) {
        experiment.Detector(setup);
        UpdateROIDefinition();
        gain_calibration = setup.GetGainCalibration();
        current_detector_setup = 0;
        pixel_mask = PixelMask(experiment);
    }
    detector_setup.emplace_back(setup);
}

DetectorList JFJochStateMachine::GetDetectorsList() const {
    DetectorList ret;

    for (const auto &i: detector_setup) {
        DetectorListElement tmp;
        tmp.description = i.GetDescription();
        tmp.nmodules = i.GetModulesNum();
        tmp.width = i.GetGeometry().GetWidth(true);
        tmp.height = i.GetGeometry().GetHeight(true);
        tmp.serial_number = i.GetSerialNumber();
        tmp.base_ipv4_addr = i.GetBaseIPv4Addr();
        tmp.udp_interface_count = i.GetUDPInterfaceCount();
        tmp.min_frame_time = i.GetMinFrameTime();
        tmp.min_count_time = i.GetMinCountTime();
        tmp.readout_time = i.GetReadOutTime();
        tmp.detector_type = i.GetDetectorType();
        tmp.pixel_size_mm = i.GetPixelSize_mm();
        ret.detector.emplace_back(std::move(tmp));
    }
    ret.current_id = current_detector_setup;
    return ret;
}

std::optional<DetectorStatus> JFJochStateMachine::GetDetectorStatus() const {
    return services.GetDetectorStatus();
}

void JFJochStateMachine::SelectDetector(int64_t id) {
    std::unique_lock ul(m);

    if ((id < 0) || (id >= detector_setup.size()))
        throw JFJochException(JFJochExceptionCategory::ArrayOutOfBounds, "Detector doesn't exist");

    if (IsRunning())
        throw WrongDAQStateException("Cannot change detector during data collection");

    // Do nothing if this is the same detector as currently used
    if (id == current_detector_setup)
        return;

    // Try to deactivate current detector (if actually running)
    if (state != JFJochState::Inactive) {
        try {
            SetState(JFJochState::Busy, "Deactivating existing detector");
            ul.unlock();
            services.Off();
            ul.lock();
        } catch (const std::exception &e) {
            logger.ErrorException(e);
            logger.Warning("Cannot turn off existing detector - proceeding anyway");
        }
    }

    try {
        experiment.Detector(detector_setup[id]);
        UpdateROIDefinition();
        gain_calibration = detector_setup[id].GetGainCalibration();
        pixel_mask = PixelMask(experiment);
        SetState(JFJochState::Inactive, detector_setup[id].GetDescription() + " selected; please initialize");
        current_detector_setup = id;
    } catch (const JFJochException &e) {
        logger.ErrorException(e);
        SetState(JFJochState::Error, e.what(), BrokerStatus::MessageSeverity::Error);
        throw; // re-throw the exception, so it is populated to caller
    }
}

void JFJochStateMachine::SetRadialIntegrationSettings(const AzimuthalIntegrationSettings &settings) {
    std::unique_lock ul(m);

    if (IsRunning())
        throw WrongDAQStateException("Cannot change radial integration settings during data collection");
    {
        std::unique_lock ul2(experiment_azimuthal_integration_settings_mutex);
        experiment.ImportAzimuthalIntegrationSettings(settings);
    }
}

AzimuthalIntegrationSettings JFJochStateMachine::GetRadialIntegrationSettings() const {
    std::unique_lock ul(experiment_azimuthal_integration_settings_mutex);
    return experiment.GetAzimuthalIntegrationSettings();
}

bool JFJochStateMachine::IsRunning() const {
    switch (state) {
        case JFJochState::Inactive:
        case JFJochState::Error:
        case JFJochState::Idle:
            return false;
        case JFJochState::Measuring:
        case JFJochState::Busy:
        case JFJochState::Calibration:
            return true;
        default:
            throw JFJochException(JFJochExceptionCategory::InputParameterInvalid, "State unknown");
    }
}

BrokerStatus JFJochStateMachine::WaitTillMeasurementDone() {
    std::unique_lock ul(m);

    c.wait(ul, [&] { return !IsRunning(); });

    // A start that failed asynchronously never reached Measuring, so the state is Idle and would
    // otherwise be reported as a successfully finished collection.
    if (start_exception)
        std::rethrow_exception(start_exception);

    return GetStatus();
}

BrokerStatus JFJochStateMachine::WaitTillMeasurementDone(std::chrono::milliseconds timeout) {
    std::unique_lock ul(m);

    c.wait_for(ul, timeout, [&] { return !IsRunning(); });

    if (start_exception)
        std::rethrow_exception(start_exception);

    return GetStatus();
}

void JFJochStateMachine::ResetError() noexcept {
    try {
        if (measurement.valid())
            measurement.get();
    } catch (...) {
    }
}

std::string JFJochStateMachine::GetPreviewJPEG(const PreviewImageSettings &settings, int64_t image_number) const {
    return services.GetPreviewJPEG(settings, image_number);
}

std::string JFJochStateMachine::GetPreviewTIFF(int64_t image_number) const {
    return services.GetPreviewTIFF(image_number);
}

std::string JFJochStateMachine::GetPedestalTIFF(size_t gain_level, size_t sc) const {
    std::unique_lock ul(m);

    if (state != JFJochState::Idle)
        throw WrongDAQStateException("Pedestal can be only retrieved in Idle state");

    if ((experiment.GetDetectorSetup().GetDetectorType() == DetectorType::JUNGFRAU) && calibration) {
        auto tmp = calibration->GetPedestal(gain_level, sc);
        CompressedImage image(tmp, RAW_MODULE_COLS, RAW_MODULE_LINES * experiment.GetModulesNum());
        return WriteTIFFToString(image);
    } else
        return {};
}

void JFJochStateMachine::LoadInternalGeneratorImage(const void *data, size_t size, uint64_t image_number) {
    std::unique_lock ul(m);

    if (state != JFJochState::Idle)
        throw WrongDAQStateException("Can change internal generator image only when detector in Idle state");

    if ((size != experiment.GetPixelsNum() * sizeof(uint16_t))
        && (size != experiment.GetModulesNum() * RAW_MODULE_SIZE * sizeof(uint16_t)))
        throw JFJochException(JFJochExceptionCategory::InputParameterInvalid,
                              "Image size doesn't match current detector");

    if (image_number >= experiment.GetInternalPacketGeneratorImages())
        throw JFJochException(JFJochExceptionCategory::InputParameterInvalid,
                              "Image for internal generator out of bounds");

    std::vector<uint16_t> image(size / sizeof(uint16_t));
    memcpy(image.data(), data, size);

    services.LoadInternalGeneratorImage(experiment, image, image_number);
}

void JFJochStateMachine::LoadInternalGeneratorImageTIFF(const std::string &s, uint64_t image_number) {
    std::unique_lock ul(m);

    if (state != JFJochState::Idle)
        throw WrongDAQStateException("Can change internal generator image only when detector in Idle state");

    uint32_t cols, lines;
    auto v = ReadTIFFFromString16(s, cols, lines);
    if (((cols == experiment.GetXPixelsNum()) && (lines == experiment.GetYPixelsNum()))
        || ((cols == RAW_MODULE_COLS) && (lines == RAW_MODULE_LINES * experiment.GetModulesNum())))
        services.LoadInternalGeneratorImage(experiment, v, image_number);
    else
        throw JFJochException(JFJochExceptionCategory::InputParameterInvalid,
                              "Image size doesn't match current detector");
}

void JFJochStateMachine::UpdateROIDefinition() {
    std::unique_lock ul(roi_mutex);
    roi = experiment.ROI().GetROIDefinition();
}

void JFJochStateMachine::SetROIDefinition(const ROIDefinition &input) {
    std::unique_lock ul(m);

    if (IsRunning())
        throw WrongDAQStateException("ROI can be modified only when detector is not running");

    experiment.ROI().SetROI(input);
    UpdateROIDefinition();
}

ROIDefinition JFJochStateMachine::GetROIDefintion() const {
    std::unique_lock ul(roi_mutex);
    return roi;
}

std::vector<uint64_t> JFJochStateMachine::GetXFELPulseID() const {
    std::vector<uint64_t> ret;
    services.GetXFELPulseID(ret);
    return ret;
}

std::vector<uint64_t> JFJochStateMachine::GetXFELEventCode() const {
    std::vector<uint64_t> ret;
    services.GetXFELEventCode(ret);
    return ret;
}

std::string JFJochStateMachine::GetFullPixelMaskTIFF() const {
    std::unique_lock ul(m);
    if (state == JFJochState::Inactive)
        return {};

    std::vector v = pixel_mask.GetMask(experiment);
    CompressedImage mask_image(v, experiment.GetXPixelsNum(), experiment.GetYPixelsNum());
    return WriteTIFFToString(mask_image);
}

std::string JFJochStateMachine::GetUserPixelMaskTIFF() const {
    std::unique_lock ul(m);

    if (state == JFJochState::Inactive)
        return {};

    std::vector v = pixel_mask.GetUserMask(experiment);
    CompressedImage mask_image(v, experiment.GetXPixelsNum(), experiment.GetYPixelsNum());
    return WriteTIFFToString(mask_image);
}

std::vector<uint32_t> JFJochStateMachine::GetFullPixelMask() const {
    std::unique_lock ul(m);
    if (state == JFJochState::Inactive)
        return {};

    return pixel_mask.GetMask(experiment);
}

std::vector<uint32_t> JFJochStateMachine::GetUserPixelMask() const {
    std::unique_lock ul(m);
    if (state == JFJochState::Inactive)
        return {};

    return pixel_mask.GetUserMask(experiment);
}

void JFJochStateMachine::SetUserPixelMask(const std::vector<uint32_t> &v) {
    std::unique_lock ul(m);

    if (state != JFJochState::Idle)
        throw WrongDAQStateException("User mask can be only modified in Idle state");

    try {
        pixel_mask.LoadUserMask(experiment, v);
        UpdatePixelMaskStatistics(pixel_mask.GetStatistics());
    } catch (const JFJochException &e) {
        throw JFJochException(JFJochExceptionCategory::InputParameterInvalid,
                              "Problem handling user mask " + std::string(e.what()));
    }
}

void JFJochStateMachine::SetUserPixelMask(const CompressedImage &image) {
    std::unique_lock ul(m);

    if (state != JFJochState::Idle)
        throw WrongDAQStateException("User mask can be only modified in Idle state");

    try {
        pixel_mask.LoadUserMask(experiment, image);
        UpdatePixelMaskStatistics(pixel_mask.GetStatistics());
    } catch (const JFJochException &e) {
        throw JFJochException(JFJochExceptionCategory::InputParameterInvalid,
                              "Problem handling user mask " + std::string(e.what()));
    }
}

InstrumentMetadata JFJochStateMachine::GetInstrumentMetadata() const {
    std::unique_lock ul(experiment_instrument_metadata_mutex);
    return experiment.GetInstrumentMetadata();
}

void JFJochStateMachine::LoadInstrumentMetadata(const InstrumentMetadata &settings) {
    std::unique_lock ul(m);

    if (IsRunning())
        throw WrongDAQStateException("Cannot change instrument metadata during data collection");
    {
        std::unique_lock ul2(experiment_instrument_metadata_mutex);
        experiment.ImportInstrumentMetadata(settings);
    }
}

ImageFormatSettings JFJochStateMachine::GetImageFormatSettings() const {
    std::unique_lock ul(experiment_image_format_settings_mutex);
    return experiment.GetImageFormatSettings();
}

void JFJochStateMachine::LoadImageFormatSettings(const ImageFormatSettings &settings) {
    std::unique_lock ul(m);

    if (IsRunning())
        throw WrongDAQStateException("Cannot change image format settings during data collection");

    bool recalc_mask = (experiment.GetPedestalG0RMSLimit() != settings.GetPedestalG0RMSLimit());

    {
        std::unique_lock ul2(experiment_image_format_settings_mutex);
        experiment.ImportImageFormatSettings(settings);
    }

    if (recalc_mask)
        pixel_mask.LoadDetectorBadPixelMask(experiment, calibration.get());
    else
        pixel_mask.CalcEdgePixels(experiment);

    UpdatePixelMaskStatistics(pixel_mask.GetStatistics());
}

void JFJochStateMachine::RawImageFormatSettings() {
    std::unique_lock ul(m);

    if (IsRunning())
        throw WrongDAQStateException("Cannot change instrument metadata during data collection");

    experiment.Raw();
}

void JFJochStateMachine::ConvImageFormatSettings() {
    std::unique_lock ul(m);

    if (IsRunning())
        throw WrongDAQStateException("Cannot change instrument metadata during data collection");

    experiment.Conversion();
}

std::vector<DeviceStatus> JFJochStateMachine::GetDeviceStatus() const {
    return services.GetDeviceStatus();
}

void JFJochStateMachine::SetPreviewSocketSettings(const ZMQPreviewSettings &input) {
    services.SetPreviewSocketSettings(input);
}

ZMQPreviewSettings JFJochStateMachine::GetPreviewSocketSettings() {
    return services.GetPreviewSocketSettings();
}

void JFJochStateMachine::SetMetadataSocketSettings(const ZMQMetadataSettings &input) {
    services.SetMetadataSocketSettings(input);
}

ZMQMetadataSettings JFJochStateMachine::GetMetadataSocketSettings() {
    return services.GetMetadataSocketSettings();
}

void JFJochStateMachine::GetStartMessageFromBuffer(std::vector<uint8_t> &v) {
    return services.GetStartMessageFromBuffer(v);
}

void JFJochStateMachine::GetImageFromBuffer(std::vector<uint8_t> &v, int64_t image_number) {
    services.GetImageFromBuffer(v, image_number);
}

ImageBufferStatus JFJochStateMachine::GetImageBufferStatus() const {
    return services.GetImageBufferStatus();
}

void JFJochStateMachine::ClearImageBuffer() const {
    std::unique_lock ul(m);

    if (IsRunning())
        throw WrongDAQStateException("Cannot clear image buffer during data collection");

    services.ClearImageBuffer();
}

FileWriterSettings JFJochStateMachine::GetFileWriterSettings() const {
    std::unique_lock ul(experiment_file_writer_settings_mutex);
    return experiment.GetFileWriterSettings();
}

void JFJochStateMachine::LoadFileWriterSettings(const FileWriterSettings &settings) {
    std::unique_lock ul(m);

    if (IsRunning())
        throw WrongDAQStateException("Cannot change instrument metadata during data collection");
    {
        std::unique_lock ul2(experiment_file_writer_settings_mutex);
        experiment.ImportFileWriterSettings(settings);
    }
}

IndexingSettings JFJochStateMachine::GetIndexingSettings() const {
    std::unique_lock ul(experiment_indexing_settings_mutex);
    return experiment.GetIndexingSettings();
}

void JFJochStateMachine::SetIndexingSettings(const IndexingSettings &input) {
    std::unique_lock ul(m);
    if (IsRunning())
        throw WrongDAQStateException("Cannot change instrument metadata during data collection");
    {
        std::unique_lock ul2(experiment_indexing_settings_mutex);
        experiment.ImportIndexingSettings(input);
        try {
            services.SetupIndexing(input);
        } catch (const JFJochException &e) {
            logger.ErrorException(e);
            SetState(JFJochState::Error,
                     e.what(),
                     BrokerStatus::MessageSeverity::Error);
            throw;
        }
    }
}

BraggIntegrationSettings JFJochStateMachine::GetBraggIntegrationSettings() const {
    std::unique_lock ul(experiment_indexing_settings_mutex);
    return experiment.GetBraggIntegrationSettings();
}

void JFJochStateMachine::SetBraggIntegrationSettings(const BraggIntegrationSettings &input) {
    std::unique_lock ul(m);
    if (IsRunning())
        throw WrongDAQStateException("Cannot change Bragg integration settings during data collection");
    // The analysis engines read the integrator mode off the experiment when they are built at the start
    // of the next run, so importing it here is all that is needed.
    std::unique_lock ul2(experiment_indexing_settings_mutex);
    experiment.ImportBraggIntegrationSettings(input);
}

std::optional<ScanResult> JFJochStateMachine::GetScanResult() const {
    std::unique_lock ul(m);
    if (IsRunning())
        throw WrongDAQStateException("Cannot check scan result, when running");

    return scan_result;
}

DarkMaskSettings JFJochStateMachine::GetDarkMaskSettings() const {
    std::unique_lock ul(experiment_dark_mask_settings_mutex);
    return experiment.GetDarkMaskSettings();
}

void JFJochStateMachine::SetDarkMaskSettings(const DarkMaskSettings &settings) {
    std::unique_lock ul(m);
    if (IsRunning())
        throw WrongDAQStateException("Cannot change dark mask calculation settings during data collection");
    {
        // Setting dark mask settings in experiment requires BOTH mutexes
        std::unique_lock ul2(experiment_dark_mask_settings_mutex);
        experiment.ImportDarkMaskSettings(settings);
    }
    if ((experiment.GetDetectorType() == DetectorType::DECTRIS) && (state == JFJochState::Idle)) {
        // Need to redo the calibration
        start_exception = nullptr; // A new operation supersedes a pending start failure
        SetState(JFJochState::Busy, "Loading settings", BrokerStatus::MessageSeverity::Info);
        measurement = std::async(std::launch::async, &JFJochStateMachine::CalibrateDetector, this, std::move(ul));
    }
}

ImagePusherStatus JFJochStateMachine::GetImagePusherStatus() const {
    return services.GetImagePusherStatus();
}