Jungfraujoch/viewer/JFJochProcessController.cpp

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

#include "JFJochProcessController.h"
#include "../reader/JFJochCBFReader.h"
#include "../reader/JFJochHDF5Reader.h"

#include <QMetaType>

#include <cmath>

JFJochProcessController::JFJochProcessController(QObject *parent) : QObject(parent) {
    qRegisterMetaType<ProcessResult>("ProcessResult");
    qRegisterMetaType<std::shared_ptr<const JFJochReaderDataset>>("std::shared_ptr<const JFJochReaderDataset>");
}

JFJochProcessController::~JFJochProcessController() {
    cancel();
    joinWorker_();
}

void JFJochProcessController::joinWorker_() {
    if (worker_.joinable())
        worker_.join();
}

void JFJochProcessController::start(const QString &file_path, DiffractionExperiment experiment,
                                    PixelMask pixel_mask, ProcessConfig config) {
    if (running_.exchange(true))
        return;   // a job is already running

    cancel_pending_ = false;
    joinWorker_();   // reap the previous (finished) worker, if any
    worker_ = std::thread(&JFJochProcessController::run_, this,
                          file_path, std::move(experiment), std::move(pixel_mask), std::move(config));
    emit started();
}

void JFJochProcessController::cancel() {
    cancel_pending_ = true;
    if (auto *p = active_.load())
        p->Cancel();
}

void JFJochProcessController::run_(QString file_path, DiffractionExperiment experiment,
                                   PixelMask pixel_mask, ProcessConfig config) {
    try {
        JFJochHDF5Reader hdf5_reader;
        JFJochCBFReader cbf_reader;
        JFJochReader *reader_ptr;
        if (JFJochCBFReader::CanRead(file_path.toStdString())) {
            cbf_reader.ReadFiles(file_path.toStdString());
            reader_ptr = &cbf_reader;
        } else {
            hdf5_reader.ReadFile(file_path.toStdString());
            reader_ptr = &hdf5_reader;
        }
        JFJochReader &reader = *reader_ptr;

        // Seed the live dataset with the experiment so the chart has geometry context; per-image
        // results are filled in by OnImageProcessed as the run progresses.
        {
            auto base = std::make_shared<JFJochReaderDataset>();
            base->experiment = experiment;
            std::lock_guard lock(live_mutex_);
            live_dataset_ = std::move(base);
            last_live_emit_ = {};
        }

        Rugnux process(reader, std::move(experiment), std::move(pixel_mask), std::move(config));
        active_ = &process;
        if (cancel_pending_)
            process.Cancel();

        ProcessResult result = process.Run(this);

        active_ = nullptr;
        running_ = false;
        emit finished(result);
    } catch (const std::exception &e) {
        active_ = nullptr;
        running_ = false;
        emit failed(QString::fromStdString(e.what()));
    }
}

void JFJochProcessController::OnPhase(const std::string &phase) {
    emit phaseChanged(QString::fromStdString(phase));
}

void JFJochProcessController::OnProgress(uint64_t done, uint64_t total) {
    // Throttle to ~200 updates so a long run does not flood the GUI event queue.
    const uint64_t step = total > 200 ? total / 200 : 1;
    if (done == total || done % step == 0)
        emit progress(done, total);
}

void JFJochProcessController::OnImageProcessed(const DataMessage &msg) {
    std::shared_ptr<JFJochReaderDataset> snapshot;
    {
        std::lock_guard lock(live_mutex_);
        if (!live_dataset_)
            return;

        // Place each available per-image result at its ordinal; gaps (images still being processed
        // by other threads) read back as NaN.
        const int64_t i = msg.number;
        auto put = [i](std::vector<float> &v, float val) {
            if (static_cast<int64_t>(v.size()) <= i)
                v.resize(i + 1, NAN);
            v[i] = val;
        };
        auto &d = *live_dataset_;
        // Map this ordinal back to its original image number (for the x-axis of subset/strided runs).
        if (static_cast<int64_t>(d.source_image_number.size()) <= i)
            d.source_image_number.resize(i + 1, 0);
        d.source_image_number[i] = static_cast<int>(msg.original_number.value_or(msg.number));
        if (msg.spot_count)            put(d.spot_count, *msg.spot_count);
        if (msg.spot_count_indexed)    put(d.spot_count_indexed, *msg.spot_count_indexed);
        if (msg.spot_count_low_res)    put(d.spot_count_low_res, *msg.spot_count_low_res);
        if (msg.spot_count_ice_rings)  put(d.spot_count_ice_rings, *msg.spot_count_ice_rings);
        if (msg.indexing_result)       put(d.indexing_result, *msg.indexing_result ? 1.0f : 0.0f);
        if (msg.indexing_lattice_count) put(d.indexing_lattice_count, *msg.indexing_lattice_count);
        if (msg.bkg_estimate)          put(d.bkg_estimate, *msg.bkg_estimate);
        if (msg.spindle_blind_fraction) put(d.spindle_blind_fraction, *msg.spindle_blind_fraction);
        if (msg.resolution_estimate)   put(d.resolution_estimate, *msg.resolution_estimate);
        if (msg.profile_radius)        put(d.profile_radius, *msg.profile_radius);
        if (msg.mosaicity_deg)         put(d.mosaicity_deg, *msg.mosaicity_deg);
        if (msg.b_factor)              put(d.b_factor, *msg.b_factor);
        if (msg.integrated_reflections) put(d.integrated_reflections, *msg.integrated_reflections);
        if (msg.image_scale_factor)    put(d.image_scale_factor, *msg.image_scale_factor);
        if (msg.image_scale_cc)        put(d.image_scale_cc, *msg.image_scale_cc);

        // Throttle to ~4 Hz so the GUI plots refresh smoothly without flooding the event queue.
        const auto now = std::chrono::steady_clock::now();
        if (now - last_live_emit_ < std::chrono::milliseconds(250))
            return;
        last_live_emit_ = now;
        snapshot = std::make_shared<JFJochReaderDataset>(d);   // immutable copy for the GUI thread
    }
    emit liveDataset(snapshot);
}