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/JFJochMarCCDReader.h"
#include "../reader/JFJochSMVReader.h"
#include "../reader/JFJochHDF5Reader.h"
#include "../rugnux/ResultReport.h"
#include "../rugnux/RugnuxOutputFiles.h"
#include "../rugnux/RugnuxDefaults.h"

#include <QElapsedTimer>
#include <QMetaType>

#include <cmath>
#include <optional>

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

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, QString output_prefix,
                                    bool write_process_h5) {
    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),
                          std::move(output_prefix), write_process_h5);
    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, QString output_prefix_q,
                                   bool write_process_h5) {
    try {
        JFJochHDF5Reader hdf5_reader;
        JFJochCBFReader cbf_reader;
        JFJochMarCCDReader marccd_reader;
        JFJochSMVReader smv_reader;
        JFJochReader *reader_ptr;
        const std::string path = file_path.toStdString();
        if (JFJochCBFReader::CanRead(path)) {
            cbf_reader.ReadFiles(path);
            reader_ptr = &cbf_reader;
        } else if (JFJochMarCCDReader::CanRead(path)) {
            marccd_reader.ReadFiles(path);
            reader_ptr = &marccd_reader;
        } else if (JFJochSMVReader::CanRead(path)) {
            smv_reader.ReadFiles(path);
            reader_ptr = &smv_reader;
        } else {
            hdf5_reader.ReadFile(path);
            reader_ptr = &hdf5_reader;
        }
        JFJochReader &reader = *reader_ptr;
        // A CCD's pixel values are ADU: analysed at the gain its background shows, as rugnux does.
        if (reader_ptr == &marccd_reader || reader_ptr == &smv_reader) {
            const RugnuxMeasuredGain g = RugnuxMeasureGain(reader);
            const std::optional<float> gain = g.peaks ? std::optional<float>(g.peaks->gain) : g.variance;
            ApplyRugnuxPixelGain(experiment, config.spot_finding, gain.value_or(1.0f));
        }

        // 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_ = {};
        }

        // The report is rendered from the run's own results after Run(); keep what it needs
        // before Rugnux consumes the experiment and the config.
        const DiffractionExperiment report_experiment = experiment;
        const std::string output_prefix = output_prefix_q.toStdString();
        const OutputProvenance output_provenance = config.provenance;
        const size_t nthreads = static_cast<size_t>(std::max(1, config.nthreads));
        std::optional<ProcessFileSink> sink;
        if (write_process_h5 && !output_prefix.empty())
            config.image_sink = &sink.emplace(output_prefix);
        QElapsedTimer run_timer;
        run_timer.start();

        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);
        // The files, from what the run returned. A failure to write them fails the job, as it fails rugnux.
        if (!output_prefix.empty() && !result.cancelled) {
            Logger logger("Viewer");
            WriteRugnuxOutputs(result, output_prefix, output_provenance, nthreads, logger);
        }

        // The same text the CLI writes as <prefix>_report.txt, shown in the analysis window. As
        // with WriteResultReport, a run that produced results must not be lost to a report
        // failure, so a throw here leaves the report empty rather than failing the job.
        QString report;
        std::vector<ReportCheck> checks;
        if (!result.cancelled) {
            try {
                RunProvenance provenance;
                provenance.wall_time_s = run_timer.elapsed() / 1000.0;
                ReportDocument doc = BuildReportDocument(output_prefix, file_path.toStdString(),
                                                         report_experiment, result, provenance);
                report = QString::fromStdString(RenderReportText(doc, provenance.developer));
                checks = std::move(doc.checks);
            } catch (...) {
            }
        }

        active_ = nullptr;
        running_ = false;
        emit finished(result, report, checks);
        Rugnux::ReleasePinnedPool();
    } 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);
}