From 1ea363876ecfdc1e016f1a7a1e8b2617219bcb26 Mon Sep 17 00:00:00 2001 From: Filip Leonarski Date: Sat, 7 Nov 2026 18:56:13 +0100 Subject: [PATCH] JFJochReceiverRugnuxRotation and JFJochReceiverRugnuxStills replace JFJochReceiverRugnux The two workflows shared nothing beyond what JFJochReceiverStream now holds, so they become children of it and stills_run goes with its seven branches. The service chooses between them with RugnuxRotationData. Cut, AddToEndMessage and SetSpotFindingSettings are each one body; the stills child no longer builds an image buffer reader it does not read, and has a mutex and a cut flag of its own. JFJochReceiverService::Start no longer reassigns the detector type to steer a switch: it asks whether the input is a stream. A cancel after the images are in is told by end_message_received rather than a flag of its own, set at the same moment give or take one poll. Rotation end to end, with a model, against the build before this work: MTZ data columns, both maps, placed and refined models and the blobs are byte for byte the same; only the MTZ history's build stamp differs. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01HVeDQsU5a8AJUdJPgWevAK --- receiver/CMakeLists.txt | 3 +- receiver/JFJochReceiverRugnux.h | 99 ------- ...x.cpp => JFJochReceiverRugnuxRotation.cpp} | 273 ++++-------------- receiver/JFJochReceiverRugnuxRotation.h | 80 +++++ receiver/JFJochReceiverRugnuxStills.cpp | 211 ++++++++++++++ receiver/JFJochReceiverRugnuxStills.h | 59 ++++ receiver/JFJochReceiverService.cpp | 82 +++--- receiver/JFJochReceiverService.h | 2 +- receiver/RugnuxLiveStatus.h | 2 +- tests/FPGAImagePullerTest.cpp | 12 +- tests/JFJochReceiverRugnuxTest.cpp | 1 - tests/RugnuxResultFilesTest.cpp | 4 +- 12 files changed, 451 insertions(+), 377 deletions(-) delete mode 100644 receiver/JFJochReceiverRugnux.h rename receiver/{JFJochReceiverRugnux.cpp => JFJochReceiverRugnuxRotation.cpp} (66%) create mode 100644 receiver/JFJochReceiverRugnuxRotation.h create mode 100644 receiver/JFJochReceiverRugnuxStills.cpp create mode 100644 receiver/JFJochReceiverRugnuxStills.h diff --git a/receiver/CMakeLists.txt b/receiver/CMakeLists.txt index 1089f1c9d..416ee815e 100644 --- a/receiver/CMakeLists.txt +++ b/receiver/CMakeLists.txt @@ -14,7 +14,8 @@ ADD_LIBRARY(JFJochReceiver STATIC JFJochReceiverStream.h JFJochReceiverStream.cpp JFJochReceiverLite.h JFJochReceiverLite.cpp FPGAImagePuller.h FPGAImagePuller.cpp - JFJochReceiverRugnux.h JFJochReceiverRugnux.cpp + JFJochReceiverRugnuxRotation.h JFJochReceiverRugnuxRotation.cpp + JFJochReceiverRugnuxStills.h JFJochReceiverRugnuxStills.cpp StillsPrePass.h StillsPrePass.cpp StillsGPUAnalysis.h StillsGPUAnalysis.cpp RugnuxLiveStatus.h RugnuxLiveStatus.cpp diff --git a/receiver/JFJochReceiverRugnux.h b/receiver/JFJochReceiverRugnux.h deleted file mode 100644 index d30c6724d..000000000 --- a/receiver/JFJochReceiverRugnux.h +++ /dev/null @@ -1,99 +0,0 @@ -// SPDX-FileCopyrightText: 2026 Filip Leonarski, Paul Scherrer Institute -// SPDX-License-Identifier: GPL-3.0-only - -#pragma once - -#include -#include - -#include "JFJochReceiverStream.h" -#include "../reader/JFJochImageBufferReader.h" -#include "../rugnux/Rugnux.h" -#include "RugnuxLiveStatus.h" -#include "StillsGPUAnalysis.h" - -// A DECTRIS rotation collection forwarded to the writer as JFJochReceiverLite does, but analysed by -// rugnux while it arrives instead of image by image: the forward threads put each image into the -// image buffer and send it on, and one rugnux thread reads the collection back from that buffer. The -// end message waits for rugnux, and carries what it found - the lattice, refined geometry, per-image -// results and its merged data, map coefficients and placed model (see AddToEndMessage). -// A serial collection (no goniometer scan) is analysed frame by frame instead (StillsGPUAnalysis): each image -// the puller delivers waits in the image buffer for the pre-pass, is analysed at the geometry it decides, and -// goes to the writer with its analysis; the start message waits for the pre-pass. The merges and the model step -// run beside it, and the end message carries the last of them. Only what the puller delivers is used: a -// DataMessage with a compressed image, whatever produced it. -class JFJochReceiverRugnux : public JFJochReceiverStream { - using JFJochReceiverStream::Cancel; - - JFJochImageBufferReader reader; - - const int rugnux_nthreads; - const size_t host_budget; // what the main run and the probe plan their host memory against - const std::shared_ptr model; - const double deadline_s; // after the last image, the analysis is cut (0 = never) - RugnuxLiveStatus &live_status; - std::string skipped_warning; // why the collection was not analysed, for the broker's status - - mutable std::mutex rugnux_m; - Rugnux *rugnux = nullptr; // while it runs, for Cut - Rugnux *probe = nullptr; // the same for the lattice probe - std::optional rugnux_result; - std::promise rugnux_published; // rugnux_result is in place, or the run has failed (RugnuxThread) - bool cut = false; - std::atomic images_done{false}; // the forward threads have ended - - const bool stills_run; - IndexerThreadPool *const indexing_thread_pool; - const int64_t stills_nthreads; - int64_t stills_keep_free = 0; // slots frames do not take, for the analysis threads to send with - std::unique_ptr stills; // under rugnux_m until the frames are in, for Cut - - // rugnux_out is ready once the run's result is in place, rugnux_done once the run is gone as well. - std::future rugnux_out; - std::future probe_done; - std::future rugnux_done; - - void StartStills(); - void FinishStills(); - void HoldStillsImage(const DataMessage &message); - ZeroCopyReturnValue *AnalysedImageSlot(); - void SendAnalysedImage(DataMessage &message, const AzimuthalIntegrationProfile &profile); - - void OnStart(std::chrono::steady_clock::time_point received) override; - void OnImage(uint32_t id, DataMessage &message, std::chrono::steady_clock::time_point received) override; - void FinishCollection() override; - // Opens the reader on the collection and starts the rugnux and probe threads; a failure is the - // analysis's alone (live status Failed), never the collection's. - void StartRugnux(); - void RugnuxThread(); - void ProbeThread(); - // Stops the analysis where it is: rugnux returns what it has established (see Rugnux::Cancel), and - // the end message goes out with that. Reason: "deadline" or "cancel". - void Cut(const std::string &reason); - void AddToEndMessage(EndMessage &message) override; -public: - JFJochReceiverRugnux(const DiffractionExperiment &experiment, - const PixelMask &pixel_mask, - ImagePuller &image_puller, - ImagePusher &image_pusher, - Logger &logger, - int64_t forward_nthreads, - int rugnux_nthreads, - std::shared_ptr model, - double deadline_s, - RugnuxLiveStatus &live_status, - const SpotFindingSettings &spot_finding_settings, - JFJochReceiverCurrentStatus ¤t_status, - JFJochReceiverPlots &plots, - ImageBuffer &image_buffer, - ZMQPreviewSocket *zmq_preview_socket = nullptr, - ZMQMetadataSocket *zmq_metadata_socket = nullptr, - IndexerThreadPool *indexing_thread_pool = nullptr); - ~JFJochReceiverRugnux() override; - - void Cancel(bool silent) override; - - JFJochReceiverOutput GetFinalStatistics() const override; - - void SetSpotFindingSettings(const SpotFindingSettings &spot_finding_settings) override; -}; diff --git a/receiver/JFJochReceiverRugnux.cpp b/receiver/JFJochReceiverRugnuxRotation.cpp similarity index 66% rename from receiver/JFJochReceiverRugnux.cpp rename to receiver/JFJochReceiverRugnuxRotation.cpp index 55e24474e..19222ba0f 100644 --- a/receiver/JFJochReceiverRugnux.cpp +++ b/receiver/JFJochReceiverRugnuxRotation.cpp @@ -1,26 +1,21 @@ // SPDX-FileCopyrightText: 2026 Filip Leonarski, Paul Scherrer Institute // SPDX-License-Identifier: GPL-3.0-only -#include "JFJochReceiverRugnux.h" +#include "JFJochReceiverRugnuxRotation.h" #include -#include -#include #include "spdlog/fmt/fmt.h" #include "../common/HostMemory.h" #include "../common/time_utc.h" #include "../frame_serialize/CBORStream2Deserializer.h" -#include "../image_analysis/indexing/IndexerThreadPool.h" #include "../rugnux/RugnuxDefaults.h" #include "../rugnux/RugnuxLive.h" #include "../rugnux/ResultReport.h" #include "../writer/RugnuxResultBlocks.h" #include "RugnuxStructure.h" -using namespace std::chrono_literals; - namespace { // Kept for the rest of the broker - the receiver, the pushers, the preview - when the host memory // rugnux plans against is decided. @@ -94,23 +89,22 @@ namespace { } } -JFJochReceiverRugnux::JFJochReceiverRugnux(const DiffractionExperiment &in_experiment, - const PixelMask &in_pixel_mask, - ImagePuller &in_image_puller, - ImagePusher &in_image_pusher, - Logger &in_logger, - int64_t in_forward_nthreads, - int in_rugnux_nthreads, - std::shared_ptr in_model, - double in_deadline_s, - RugnuxLiveStatus &in_live_status, - const SpotFindingSettings &in_spot_finding_settings, - JFJochReceiverCurrentStatus &in_current_status, - JFJochReceiverPlots &in_plots, - ImageBuffer &in_image_buffer, - ZMQPreviewSocket *in_zmq_preview_socket, - ZMQMetadataSocket *in_zmq_metadata_socket, - IndexerThreadPool *in_indexing_thread_pool) +JFJochReceiverRugnuxRotation::JFJochReceiverRugnuxRotation(const DiffractionExperiment &in_experiment, + const PixelMask &in_pixel_mask, + ImagePuller &in_image_puller, + ImagePusher &in_image_pusher, + Logger &in_logger, + int64_t in_forward_nthreads, + int in_rugnux_nthreads, + std::shared_ptr in_model, + double in_deadline_s, + RugnuxLiveStatus &in_live_status, + const SpotFindingSettings &in_spot_finding_settings, + JFJochReceiverCurrentStatus &in_current_status, + JFJochReceiverPlots &in_plots, + ImageBuffer &in_image_buffer, + ZMQPreviewSocket *in_zmq_preview_socket, + ZMQMetadataSocket *in_zmq_metadata_socket) : JFJochReceiverStream(in_experiment, in_pixel_mask, in_image_puller, in_image_pusher, in_logger, in_forward_nthreads, in_spot_finding_settings, in_current_status, in_plots, in_image_buffer, in_zmq_preview_socket, in_zmq_metadata_socket, nullptr), @@ -119,25 +113,20 @@ JFJochReceiverRugnux::JFJochReceiverRugnux(const DiffractionExperiment &in_exper host_budget(RugnuxHostBudget(in_image_buffer)), model(std::move(in_model)), deadline_s(in_deadline_s), - live_status(in_live_status), - stills_run(!RugnuxRotationData(in_experiment)), - indexing_thread_pool(in_indexing_thread_pool), - stills_nthreads(std::min(16, in_forward_nthreads)) { - if (!stills_run) { - // GPU FFT indexers and cuFFT plans are kept for reuse for the life of the broker, not of one run as - // in `rugnux` (FFTIndexerReuse, FFTPlanReuse): every pass of every collection builds the same pools, - // and making a plan takes a process-wide lock and took over a second on a card busy with the online - // analysis. Never destroyed - at exit the CUDA runtime may already be gone. - [[maybe_unused]] static const FFTIndexerReuse *fft_indexers = new FFTIndexerReuse; - [[maybe_unused]] static const FFTPlanReuse *fft_plans = new FFTPlanReuse; - logger.Info("rugnux live: starting {} forward threads; rugnux runs on {} threads while the images arrive, " - "planning against {:.1f} GB of host memory", nthreads, rugnux_nthreads, - static_cast(host_budget) / 1e9); - } + live_status(in_live_status) { + // GPU FFT indexers and cuFFT plans are kept for reuse for the life of the broker, not of one run as + // in `rugnux` (FFTIndexerReuse, FFTPlanReuse): every pass of every collection builds the same pools, + // and making a plan takes a process-wide lock and took over a second on a card busy with the online + // analysis. Never destroyed - at exit the CUDA runtime may already be gone. + [[maybe_unused]] static const FFTIndexerReuse *fft_indexers = new FFTIndexerReuse; + [[maybe_unused]] static const FFTPlanReuse *fft_plans = new FFTPlanReuse; + logger.Info("rugnux live: starting {} forward threads; rugnux runs on {} threads while the images arrive, " + "planning against {:.1f} GB of host memory", nthreads, rugnux_nthreads, + static_cast(host_budget) / 1e9); StartThreads(); } -JFJochReceiverRugnux::~JFJochReceiverRugnux() { +JFJochReceiverRugnuxRotation::~JFJochReceiverRugnuxRotation() { cancelled = true; reader.Ended(); { @@ -150,11 +139,7 @@ JFJochReceiverRugnux::~JFJochReceiverRugnux() { JoinThreads(); } -void JFJochReceiverRugnux::OnStart(std::chrono::steady_clock::time_point received) { - if (stills_run) { - StartStills(); - return; - } +void JFJochReceiverRugnuxRotation::OnStart(std::chrono::steady_clock::time_point received) { SendStartMessage(); // Rugnux reads the images back from the buffer, so it needs one slot per image: a slot taken @@ -175,13 +160,15 @@ void JFJochReceiverRugnux::OnStart(std::chrono::steady_clock::time_point receive StartRugnux(); } -void JFJochReceiverRugnux::FinishCollection() { - if (stills_run) { - FinishStills(); - return; - } +void JFJochReceiverRugnuxRotation::OnImage(uint32_t id, DataMessage &message, std::chrono::steady_clock::time_point received) { + live_status.ImageReceived(received); + CountImage(message); + StampImage(message, image_puller.GetCurrentFifoUtilization()); + ForwardImage(message, image_buffer.GetImageSlot()); +} + +void JFJochReceiverRugnuxRotation::FinishCollection() { reader.Ended(); - images_done = true; // The latency of the demonstration runs from the last image in (as the puller took it off the socket) // to every result in place. const auto last_image = live_status.LastImage(); @@ -238,164 +225,23 @@ void JFJochReceiverRugnux::FinishCollection() { current_status.SetStatus(GetStatus()); } -void JFJochReceiverRugnux::StartStills() { - const int64_t slots = image_buffer.GetStatus().total_slots; - stills_keep_free = std::min(stills_nthreads, slots / 2); - auto analysis = std::make_unique( - experiment, pixel_mask, MakeStartMessage(experiment, pixel_mask, *az_int_mapping), - indexing_thread_pool, stills_nthreads, - // What the pre-pass holds and fits on; the frames beyond it wait in the image buffer as well. - 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)); - }, - [this](DataMessage &m, const AzimuthalIntegrationProfile &p) { SendAnalysedImage(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 settings = GetSpotFindingSettings(); - settings.adaptive_threshold = true; - settings.min_pix_per_spot = std::nullopt; - return settings; - }, - // The start message did not go out (or the analysis could not be set up): nothing can be - // written, so the collection stops. - [this](const JFJochException &e) { Cancel(e); }, - &live_status, true, model, deadline_s); - std::unique_lock ul(rugnux_m); - stills = std::move(analysis); - if (cut) - stills->Cut("cancel"); -} - -void JFJochReceiverRugnux::FinishStills() { - images_done = true; - logger.Info("Receiving finished, {} images", images_collected.load()); - - // Where the start message never went out (or the analysis could not be set up), there is no end message - // either, and the error is the collection's result. - std::exception_ptr error; - try { - stills->Finish(); - } catch (const std::exception &) { - error = std::current_exception(); - } - - current_status.SetProgress(1.0); - current_status.SetStatus(GetStatus()); - - if (!error) - SendEndMessage(); - - end_time = std::chrono::system_clock::now(); - - CheckIfImagesSent(); - if (error) { - current_status.SetProgress({}); - current_status.SetStatus(GetStatus()); - std::rethrow_exception(error); - } - WriterFinalize(); - stills->Written(); - - current_status.SetProgress({}); - current_status.SetStatus(GetStatus()); -} - -void JFJochReceiverRugnux::HoldStillsImage(const DataMessage &message) { - // The analysis keeps per-image results by number, for the images the collection declares; the stream is not ours. - if (message.number < 0 || message.number >= experiment.GetImageNum()) { - logger.Error("Dropping image {} - the collection has {} images", message.number, experiment.GetImageNum()); - return; - } - CountImage(message); - const size_t size = message.image.GetCompressedSize(); - - // Copied into a slot of the image buffer, where it waits for the analysis; SendAnalysedImage writes it, with - // its analysis, to a second slot, which is sent. A frame that finds the buffer full, but for a slot kept - // free for each analysis thread to send with, is dropped, as any frame that finds the image buffer full. - auto held = std::make_shared(); - held->slot = image_buffer.GetImageSlot(stills_keep_free); - if (held->slot == nullptr) { - writer_queue_full = true; - return; - } - if (size > static_cast(experiment.GetImageBufferLocationSize())) { - logger.Error("Dropping image {} - {} B do not fit in buffer slot {} B", message.number, size, - experiment.GetImageBufferLocationSize()); - return; - } - memcpy(held->slot->GetImage(), message.image.GetCompressed(), size); - held->message = message; - held->message.image = CompressedImage(held->slot->GetImage(), size, message.image.GetWidth(), - message.image.GetHeight(), message.image.GetMode(), - message.image.GetCompressionAlgorithm(), message.image.GetChannel()); - StampImage(held->message, image_puller.GetCurrentFifoUtilization()); - stills->Hold(held); -} - -ZeroCopyReturnValue *JFJochReceiverRugnux::AnalysedImageSlot() { - // An analysed frame is not dropped because the writer lags behind the catch-up from a full buffer: the analysis - // waits for the slots being sent to come back, and the frames arriving meanwhile find the buffer full. A - // writer taking nothing for this long loses the frame. - constexpr auto WRITER_WAIT = std::chrono::seconds(2); - const auto t0 = std::chrono::steady_clock::now(); - auto loc = image_buffer.GetImageSlot(); - while (loc == nullptr && std::chrono::steady_clock::now() - t0 < WRITER_WAIT) { - std::this_thread::sleep_for(std::chrono::milliseconds(1)); - loc = image_buffer.GetImageSlot(); - } - return loc; -} - -void JFJochReceiverRugnux::SendAnalysedImage(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; - AddToAverages(message); - scan_result.Add(message); - - if (!serialmx_filter.ApplyFilter(message)) - ++images_skipped; - else - ForwardImage(message, AnalysedImageSlot()); - - 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 JFJochReceiverRugnux::StartRugnux() { +void JFJochReceiverRugnuxRotation::StartRugnux() { // A failure here costs the analysis only: the images go on to the writer as they would without it. try { // The start message as the writer gets it, so rugnux sees the dataset an HDF5 master states. std::vector start_cbor; image_buffer.GetStartMessage(start_cbor); reader.Open(CBORStream2Deserialize(start_cbor)->start_message.value()); - rugnux_done = std::async(std::launch::async, &JFJochReceiverRugnux::RugnuxThread, this); + rugnux_done = std::async(std::launch::async, &JFJochReceiverRugnuxRotation::RugnuxThread, this); rugnux_out = rugnux_published.get_future(); - probe_done = std::async(std::launch::async, &JFJochReceiverRugnux::ProbeThread, this); + probe_done = std::async(std::launch::async, &JFJochReceiverRugnuxRotation::ProbeThread, this); } catch (const std::exception &e) { logger.Error("rugnux live: cannot start - {}", e.what()); live_status.Finish(RugnuxLiveState::Failed, e.what()); } } -void JFJochReceiverRugnux::RugnuxThread() { +void JFJochReceiverRugnuxRotation::RugnuxThread() { bool published = false; try { const auto dataset = reader.GetDataset(); @@ -468,7 +314,7 @@ void JFJochReceiverRugnux::RugnuxThread() { rugnux_published.set_value(); } -void JFJochReceiverRugnux::ProbeThread() { +void JFJochReceiverRugnuxRotation::ProbeThread() { // Rugnux waits for half the sweep before it decides on a lattice. In the meantime the machine is // idle, so a small run of its own indexes the first 60 degrees and says at once whether there is // a crystal - quick feedback for the person at the beamline, above all where there is none. It @@ -515,12 +361,8 @@ void JFJochReceiverRugnux::ProbeThread() { logger.Debug("rugnux live: lattice probe - {}", live_status.Get().probe); } -void JFJochReceiverRugnux::AddToEndMessage(EndMessage &message) { +void JFJochReceiverRugnuxRotation::AddToEndMessage(EndMessage &message) { JFJochReceiverStream::AddToEndMessage(message); - if (stills) { - stills->FillEndMessage(message); - return; - } // A failure here costs the end message the analysis only: it goes out as the receiver made it. try { std::unique_lock ul(rugnux_m); @@ -574,18 +416,7 @@ void JFJochReceiverRugnux::AddToEndMessage(EndMessage &message) { } } -void JFJochReceiverRugnux::OnImage(uint32_t id, DataMessage &message, std::chrono::steady_clock::time_point received) { - if (stills_run) { - HoldStillsImage(message); - return; - } - live_status.ImageReceived(received); - CountImage(message); - StampImage(message, image_puller.GetCurrentFifoUtilization()); - ForwardImage(message, image_buffer.GetImageSlot()); -} - -void JFJochReceiverRugnux::Cancel(bool silent) { +void JFJochReceiverRugnuxRotation::Cancel(bool silent) { detector_finished = true; // The silent cancel is the detector reading idle at the end of every collection: rugnux goes on. if (silent) @@ -593,20 +424,16 @@ void JFJochReceiverRugnux::Cancel(bool silent) { // /cancel is the one way in for the beamline, which cannot tell the collection from its analysis: it // ends the collection while the images still arrive, as JFJochReceiverLite does, and in every case // cuts the analysis, so the end message goes out at once with what rugnux has established. - if (!images_done) + if (!end_message_received) JFJochReceiver::Cancel(silent); Cut("cancel"); } -void JFJochReceiverRugnux::Cut(const std::string &reason) { +void JFJochReceiverRugnuxRotation::Cut(const std::string &reason) { std::unique_lock ul(rugnux_m); if (cut || rugnux_result) return; cut = true; - if (stills) { - stills->Cut(reason); - return; - } reader.Ended(); // no image is waited for any more if (rugnux != nullptr) rugnux->Cancel(); @@ -616,13 +443,11 @@ void JFJochReceiverRugnux::Cut(const std::string &reason) { logger.Debug("rugnux live: analysis cut ({}) - the end message carries what is established", reason); } -void JFJochReceiverRugnux::SetSpotFindingSettings(const SpotFindingSettings &in_spot_finding_settings) { - // A rotation run takes rugnux's own spot finding; the frames of a serial run are analysed with these. - if (stills_run) - JFJochReceiver::SetSpotFindingSettings(in_spot_finding_settings); +void JFJochReceiverRugnuxRotation::SetSpotFindingSettings(const SpotFindingSettings &in_spot_finding_settings) { + // A rotation run takes rugnux's own spot finding } -JFJochReceiverOutput JFJochReceiverRugnux::GetFinalStatistics() const { +JFJochReceiverOutput JFJochReceiverRugnuxRotation::GetFinalStatistics() const { JFJochReceiverOutput ret = JFJochReceiver::GetFinalStatistics(); ret.analysis_warning = skipped_warning; // The receiver's own rotation indexer sees no image here: the lattice is rugnux's. diff --git a/receiver/JFJochReceiverRugnuxRotation.h b/receiver/JFJochReceiverRugnuxRotation.h new file mode 100644 index 000000000..721844583 --- /dev/null +++ b/receiver/JFJochReceiverRugnuxRotation.h @@ -0,0 +1,80 @@ +// SPDX-FileCopyrightText: 2026 Filip Leonarski, Paul Scherrer Institute +// SPDX-License-Identifier: GPL-3.0-only + +#pragma once + +#include +#include +#include +#include +#include + +#include "JFJochReceiverStream.h" +#include "RugnuxLiveStatus.h" +#include "../reader/JFJochImageBufferReader.h" +#include "../rugnux/Rugnux.h" + +// A rotation collection forwarded to the writer as JFJochReceiverLite does, but analysed by rugnux while it +// arrives instead of image by image: one rugnux thread reads the collection back from the image buffer. +// The end message waits for rugnux, and carries what it found (see AddToEndMessage). +class JFJochReceiverRugnuxRotation : public JFJochReceiverStream { + using JFJochReceiverStream::Cancel; + + JFJochImageBufferReader reader; + + const int rugnux_nthreads; + const size_t host_budget; // what the main run and the probe plan their host memory against + const std::shared_ptr model; + const double deadline_s; // after the last image, the analysis is cut (0 = never) + RugnuxLiveStatus &live_status; + std::string skipped_warning; // why the collection was not analysed, for the broker's status + + mutable std::mutex rugnux_m; + Rugnux *rugnux = nullptr; // while it runs, for Cut + Rugnux *probe = nullptr; // the same for the lattice probe + std::optional rugnux_result; + std::promise rugnux_published; // rugnux_result is in place, or the run has failed (RugnuxThread) + bool cut = false; + + // rugnux_out is ready once the run's result is in place, rugnux_done once the run is gone as well. + std::future rugnux_out; + std::future probe_done; + std::future rugnux_done; + + void OnStart(std::chrono::steady_clock::time_point received) override; + void OnImage(uint32_t id, DataMessage &message, std::chrono::steady_clock::time_point received) override; + void FinishCollection() override; + // Opens the reader on the collection and starts the rugnux and probe threads; a failure is the + // analysis's alone (live status Failed), never the collection's. + void StartRugnux(); + void RugnuxThread(); + void ProbeThread(); + // Stops the analysis where it is: rugnux returns what it has established (see Rugnux::Cancel), and + // the end message goes out with that. Reason: "deadline" or "cancel". + void Cut(const std::string &reason); + void AddToEndMessage(EndMessage &message) override; +public: + JFJochReceiverRugnuxRotation(const DiffractionExperiment &experiment, + const PixelMask &pixel_mask, + ImagePuller &image_puller, + ImagePusher &image_pusher, + Logger &logger, + int64_t forward_nthreads, + int rugnux_nthreads, + std::shared_ptr model, + double deadline_s, + RugnuxLiveStatus &live_status, + const SpotFindingSettings &spot_finding_settings, + JFJochReceiverCurrentStatus ¤t_status, + JFJochReceiverPlots &plots, + ImageBuffer &image_buffer, + ZMQPreviewSocket *zmq_preview_socket = nullptr, + ZMQMetadataSocket *zmq_metadata_socket = nullptr); + ~JFJochReceiverRugnuxRotation() override; + + void Cancel(bool silent) override; + + JFJochReceiverOutput GetFinalStatistics() const override; + + void SetSpotFindingSettings(const SpotFindingSettings &spot_finding_settings) override; +}; diff --git a/receiver/JFJochReceiverRugnuxStills.cpp b/receiver/JFJochReceiverRugnuxStills.cpp new file mode 100644 index 000000000..032a53a14 --- /dev/null +++ b/receiver/JFJochReceiverRugnuxStills.cpp @@ -0,0 +1,211 @@ +// SPDX-FileCopyrightText: 2026 Filip Leonarski, Paul Scherrer Institute +// SPDX-License-Identifier: GPL-3.0-only + +#include "JFJochReceiverRugnuxStills.h" + +#include +#include + +JFJochReceiverRugnuxStills::JFJochReceiverRugnuxStills(const DiffractionExperiment &in_experiment, + const PixelMask &in_pixel_mask, + ImagePuller &in_image_puller, + ImagePusher &in_image_pusher, + Logger &in_logger, + int64_t in_forward_nthreads, + std::shared_ptr in_model, + double in_deadline_s, + RugnuxLiveStatus &in_live_status, + const SpotFindingSettings &in_spot_finding_settings, + JFJochReceiverCurrentStatus &in_current_status, + JFJochReceiverPlots &in_plots, + ImageBuffer &in_image_buffer, + ZMQPreviewSocket *in_zmq_preview_socket, + ZMQMetadataSocket *in_zmq_metadata_socket, + IndexerThreadPool *in_indexing_thread_pool) + : JFJochReceiverStream(in_experiment, in_pixel_mask, in_image_puller, in_image_pusher, in_logger, + in_forward_nthreads, in_spot_finding_settings, in_current_status, in_plots, + in_image_buffer, in_zmq_preview_socket, in_zmq_metadata_socket, nullptr), + model(std::move(in_model)), + deadline_s(in_deadline_s), + live_status(in_live_status), + indexing_thread_pool(in_indexing_thread_pool), + stills_nthreads(std::min(16, in_forward_nthreads)) { + StartThreads(); +} + +JFJochReceiverRugnuxStills::~JFJochReceiverRugnuxStills() { + cancelled = true; + JoinThreads(); +} + +void JFJochReceiverRugnuxStills::OnStart(std::chrono::steady_clock::time_point received) { + const int64_t slots = image_buffer.GetStatus().total_slots; + stills_keep_free = std::min(stills_nthreads, slots / 2); + auto analysis = std::make_unique( + experiment, pixel_mask, MakeStartMessage(experiment, pixel_mask, *az_int_mapping), + indexing_thread_pool, stills_nthreads, + // What the pre-pass holds and fits on; the frames beyond it wait in the image buffer as well. + 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)); + }, + [this](DataMessage &m, const AzimuthalIntegrationProfile &p) { SendAnalysedImage(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 settings = GetSpotFindingSettings(); + settings.adaptive_threshold = true; + settings.min_pix_per_spot = std::nullopt; + return settings; + }, + // The start message did not go out (or the analysis could not be set up): nothing can be + // written, so the collection stops. + [this](const JFJochException &e) { Cancel(e); }, + &live_status, true, model, deadline_s); + std::unique_lock ul(stills_mutex); + stills = std::move(analysis); + if (cut) + stills->Cut("cancel"); +} + +void JFJochReceiverRugnuxStills::OnImage(uint32_t id, DataMessage &message, std::chrono::steady_clock::time_point received) { + // The analysis keeps per-image results by number, for the images the collection declares; the stream is not ours. + if (message.number < 0 || message.number >= experiment.GetImageNum()) { + logger.Error("Dropping image {} - the collection has {} images", message.number, experiment.GetImageNum()); + return; + } + CountImage(message); + const size_t size = message.image.GetCompressedSize(); + + // Copied into a slot of the image buffer, where it waits for the analysis; SendAnalysedImage writes it, with + // its analysis, to a second slot, which is sent. A frame that finds the buffer full, but for a slot kept + // free for each analysis thread to send with, is dropped, as any frame that finds the image buffer full. + auto held = std::make_shared(); + held->slot = image_buffer.GetImageSlot(stills_keep_free); + if (held->slot == nullptr) { + writer_queue_full = true; + return; + } + if (size > static_cast(experiment.GetImageBufferLocationSize())) { + logger.Error("Dropping image {} - {} B do not fit in buffer slot {} B", message.number, size, + experiment.GetImageBufferLocationSize()); + return; + } + memcpy(held->slot->GetImage(), message.image.GetCompressed(), size); + held->message = message; + held->message.image = CompressedImage(held->slot->GetImage(), size, message.image.GetWidth(), + message.image.GetHeight(), message.image.GetMode(), + message.image.GetCompressionAlgorithm(), message.image.GetChannel()); + StampImage(held->message, image_puller.GetCurrentFifoUtilization()); + stills->Hold(held); +} + +void JFJochReceiverRugnuxStills::FinishCollection() { + logger.Info("Receiving finished, {} images", images_collected.load()); + + // Where the start message never went out (or the analysis could not be set up), there is no end message + // either, and the error is the collection's result. + std::exception_ptr error; + try { + stills->Finish(); + } catch (const std::exception &) { + error = std::current_exception(); + } + + current_status.SetProgress(1.0); + current_status.SetStatus(GetStatus()); + + if (!error) + SendEndMessage(); + + end_time = std::chrono::system_clock::now(); + + CheckIfImagesSent(); + if (error) { + current_status.SetProgress({}); + current_status.SetStatus(GetStatus()); + std::rethrow_exception(error); + } + WriterFinalize(); + stills->Written(); + + current_status.SetProgress({}); + current_status.SetStatus(GetStatus()); +} + +ZeroCopyReturnValue *JFJochReceiverRugnuxStills::AnalysedImageSlot() { + // An analysed frame is not dropped because the writer lags behind the catch-up from a full buffer: the analysis + // waits for the slots being sent to come back, and the frames arriving meanwhile find the buffer full. A + // writer taking nothing for this long loses the frame. + constexpr auto WRITER_WAIT = std::chrono::seconds(2); + const auto t0 = std::chrono::steady_clock::now(); + auto loc = image_buffer.GetImageSlot(); + while (loc == nullptr && std::chrono::steady_clock::now() - t0 < WRITER_WAIT) { + std::this_thread::sleep_for(std::chrono::milliseconds(1)); + loc = image_buffer.GetImageSlot(); + } + return loc; +} + +void JFJochReceiverRugnuxStills::SendAnalysedImage(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; + AddToAverages(message); + scan_result.Add(message); + + if (!serialmx_filter.ApplyFilter(message)) + ++images_skipped; + else + ForwardImage(message, AnalysedImageSlot()); + + 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 JFJochReceiverRugnuxStills::AddToEndMessage(EndMessage &message) { + JFJochReceiverStream::AddToEndMessage(message); + if (stills) + stills->FillEndMessage(message); +} + +void JFJochReceiverRugnuxStills::Cancel(bool silent) { + detector_finished = true; + // The silent cancel is the detector reading idle at the end of every collection: rugnux goes on. + if (silent) + return; + // /cancel is the one way in for the beamline, which cannot tell the collection from its analysis: it + // ends the collection while the images still arrive, as JFJochReceiverLite does, and in every case + // cuts the analysis, so the end message goes out at once with what rugnux has established. + if (!end_message_received) + JFJochReceiver::Cancel(silent); + Cut("cancel"); +} + +void JFJochReceiverRugnuxStills::Cut(const std::string &reason) { + std::unique_lock ul(stills_mutex); + if (cut) + return; + cut = true; + if (stills) { + stills->Cut(reason); + return; + } + live_status.Cut(reason); + logger.Debug("rugnux live: analysis cut ({}) - the end message carries what is established", reason); +} diff --git a/receiver/JFJochReceiverRugnuxStills.h b/receiver/JFJochReceiverRugnuxStills.h new file mode 100644 index 000000000..23a9c9b45 --- /dev/null +++ b/receiver/JFJochReceiverRugnuxStills.h @@ -0,0 +1,59 @@ +// SPDX-FileCopyrightText: 2026 Filip Leonarski, Paul Scherrer Institute +// SPDX-License-Identifier: GPL-3.0-only + +#pragma once + +#include +#include +#include + +#include "JFJochReceiverStream.h" +#include "RugnuxLiveStatus.h" +#include "StillsGPUAnalysis.h" + +// A serial collection analysed frame by frame (StillsGPUAnalysis): each image the puller delivers waits in the +// image buffer for the pre-pass, is analysed at the geometry it decides, and goes to the writer with its analysis. +// The start message waits for the pre-pass, the end message carries the last merge and model step. +class JFJochReceiverRugnuxStills : public JFJochReceiverStream { + using JFJochReceiverStream::Cancel; + + const std::shared_ptr model; + const double deadline_s; // after the last image, the model step is cut (0 = never) + RugnuxLiveStatus &live_status; + + IndexerThreadPool *const indexing_thread_pool; + const int64_t stills_nthreads; + int64_t stills_keep_free = 0; // slots frames do not take, for the analysis threads to send with + + std::mutex stills_mutex; + bool cut = false; + std::unique_ptr stills; // under stills_mutex until the frames are in, for Cut + + void OnStart(std::chrono::steady_clock::time_point received) override; + void OnImage(uint32_t id, DataMessage &message, std::chrono::steady_clock::time_point received) override; + void FinishCollection() override; + ZeroCopyReturnValue *AnalysedImageSlot(); + void SendAnalysedImage(DataMessage &message, const AzimuthalIntegrationProfile &profile); + void Cut(const std::string &reason); + void AddToEndMessage(EndMessage &message) override; +public: + JFJochReceiverRugnuxStills(const DiffractionExperiment &experiment, + const PixelMask &pixel_mask, + ImagePuller &image_puller, + ImagePusher &image_pusher, + Logger &logger, + int64_t forward_nthreads, + std::shared_ptr model, + double deadline_s, + RugnuxLiveStatus &live_status, + const SpotFindingSettings &spot_finding_settings, + JFJochReceiverCurrentStatus ¤t_status, + JFJochReceiverPlots &plots, + ImageBuffer &image_buffer, + ZMQPreviewSocket *zmq_preview_socket = nullptr, + ZMQMetadataSocket *zmq_metadata_socket = nullptr, + IndexerThreadPool *indexing_thread_pool = nullptr); + ~JFJochReceiverRugnuxStills() override; + + void Cancel(bool silent) override; +}; diff --git a/receiver/JFJochReceiverService.cpp b/receiver/JFJochReceiverService.cpp index d3eb2cdea..033a3f7b9 100644 --- a/receiver/JFJochReceiverService.cpp +++ b/receiver/JFJochReceiverService.cpp @@ -2,7 +2,9 @@ // SPDX-License-Identifier: GPL-3.0-only #include "JFJochReceiverService.h" -#include "JFJochReceiverRugnux.h" +#include "JFJochReceiverRugnuxRotation.h" +#include "JFJochReceiverRugnuxStills.h" +#include "../rugnux/RugnuxDefaults.h" #include "../common/ThreadAffinity.h" #include "JFJochReceiverFPGA.h" #include "JFJochReceiverLite.h" @@ -117,39 +119,22 @@ void JFJochReceiverService::Start(const DiffractionExperiment &experiment, preview_image.Configure(experiment, pixel_mask); // Pedestal runs and host summation stay with JFJochReceiverFPGA - auto detector_type = experiment.GetDetectorType(); - if (fpga_image_puller && (detector_type != DetectorType::DECTRIS) && (experiment.GetImageNum() > 0)) { + bool stream_input = (experiment.GetDetectorType() == DetectorType::DECTRIS); + if (fpga_image_puller && !stream_input && (experiment.GetImageNum() > 0)) { if (experiment.IsCPUSummation()) logger.Warning("Summation above what the card can do needs JFJochReceiverFPGA: FPGA image puller not used"); else { fpga_puller = std::make_shared(experiment, pixel_mask, calibration, aq_devices, logger, nthreads_local, fpga_image_puller_slots); puller = fpga_puller; - detector_type = DetectorType::DECTRIS; // from here on, one stream receiver input like any other + stream_input = true; } } - switch (detector_type) { - case DetectorType::EIGER: - case DetectorType::JUNGFRAU: - if (experiment.IsPedestalRun()) { - receiver = std::make_unique(experiment, pixel_mask, - aq_devices, image_pusher, - logger, - spot_finding_settings, - receiver_status, - plots, - image_buffer, - zmq_preview_socket.get(), - zmq_metadata_socket.get(), - indexer_thread_pool.get()); - break; - } - receiver = std::make_unique(experiment, pixel_mask, - calibration, + if (!stream_input && experiment.IsPedestalRun()) { + receiver = std::make_unique(experiment, pixel_mask, aq_devices, image_pusher, logger, - nthreads_local, spot_finding_settings, receiver_status, plots, @@ -157,23 +142,38 @@ void JFJochReceiverService::Start(const DiffractionExperiment &experiment, zmq_preview_socket.get(), zmq_metadata_socket.get(), indexer_thread_pool.get()); - break; - case DetectorType::DECTRIS: - if (puller) - image_puller = puller; - else { - image_puller = std::make_shared( - experiment.GetDetectorSetup().GetDECTRISStream2Addr()); - } - // The dark frames of the mask sequence take Lite's mask path, with rugnux on as well. - if (rugnux_enable && experiment.GetDetectorMode() == DetectorMode::Standard) { - receiver = std::make_unique( - experiment, pixel_mask, *image_puller, image_pusher, logger, nthreads_local, - std::max(1, static_cast(AvailableCpus()) - rugnux_reserved_cpus), - rugnux_model, rugnux_deadline_s, rugnux_status, spot_finding_settings, receiver_status, plots, image_buffer, - zmq_preview_socket.get(), zmq_metadata_socket.get(), indexer_thread_pool.get()); - break; - } + } else if (!stream_input) { + receiver = std::make_unique(experiment, pixel_mask, + calibration, + aq_devices, image_pusher, + logger, + nthreads_local, + spot_finding_settings, + receiver_status, + plots, + image_buffer, + zmq_preview_socket.get(), + zmq_metadata_socket.get(), + indexer_thread_pool.get()); + } else { + if (puller) + image_puller = puller; + else + image_puller = std::make_shared(experiment.GetDetectorSetup().GetDECTRISStream2Addr()); + + // The dark frames of the mask sequence take Lite's mask path, with rugnux on as well. + if (rugnux_enable && (experiment.GetDetectorMode() == DetectorMode::Standard) && RugnuxRotationData(experiment)) + receiver = std::make_unique( + experiment, pixel_mask, *image_puller, image_pusher, logger, nthreads_local, + std::max(1, static_cast(AvailableCpus()) - rugnux_reserved_cpus), + rugnux_model, rugnux_deadline_s, rugnux_status, spot_finding_settings, receiver_status, plots, + image_buffer, zmq_preview_socket.get(), zmq_metadata_socket.get()); + else if (rugnux_enable && (experiment.GetDetectorMode() == DetectorMode::Standard)) + receiver = std::make_unique( + experiment, pixel_mask, *image_puller, image_pusher, logger, nthreads_local, + rugnux_model, rugnux_deadline_s, rugnux_status, spot_finding_settings, receiver_status, plots, + image_buffer, zmq_preview_socket.get(), zmq_metadata_socket.get(), indexer_thread_pool.get()); + else receiver = std::make_unique(experiment, pixel_mask, *image_puller, @@ -187,8 +187,6 @@ void JFJochReceiverService::Start(const DiffractionExperiment &experiment, zmq_preview_socket.get(), zmq_metadata_socket.get(), indexer_thread_pool.get()); - break; - } measurement = std::async(std::launch::async, &JFJochReceiverService::FinalizeMeasurement, this); state = ReceiverState::Running; diff --git a/receiver/JFJochReceiverService.h b/receiver/JFJochReceiverService.h index 9a875e7ba..7eabc44a9 100644 --- a/receiver/JFJochReceiverService.h +++ b/receiver/JFJochReceiverService.h @@ -27,7 +27,7 @@ class JFJochReceiverService { int64_t nthreads = 8; // Rotation collections from a DECTRIS detector analysed by rugnux while they arrive - // (JFJochReceiverRugnux), instead of image by image. Off unless asked for. + // (JFJochReceiverRugnuxRotation), instead of image by image. Off unless asked for. bool rugnux_enable = false; std::shared_ptr rugnux_model; int rugnux_reserved_cpus = 4; // left to the forward threads and the pusher diff --git a/receiver/RugnuxLiveStatus.h b/receiver/RugnuxLiveStatus.h index 5d98c2a70..b06de487f 100644 --- a/receiver/RugnuxLiveStatus.h +++ b/receiver/RugnuxLiveStatus.h @@ -110,7 +110,7 @@ struct RugnuxLiveSnapshot { double elapsed_s = 0.0; std::optional step_progress; // fraction of the current pass's images std::vector steps; - // The lattice probe on the first 60 degrees (JFJochReceiverRugnux::ProbeThread): an early answer + // The lattice probe on the first 60 degrees (JFJochReceiverRugnuxRotation::ProbeThread): an early answer // to "is there a crystal", seconds into the collection; the run's own answer comes later. std::string probe; std::optional probe_s; // when it answered diff --git a/tests/FPGAImagePullerTest.cpp b/tests/FPGAImagePullerTest.cpp index 2528828ba..a17a467a2 100644 --- a/tests/FPGAImagePullerTest.cpp +++ b/tests/FPGAImagePullerTest.cpp @@ -21,7 +21,7 @@ #include "../writer/HDF5Objects.h" #include "../writer/StreamWriter.h" #include "../receiver/FPGAImagePuller.h" -#include "../receiver/JFJochReceiverRugnux.h" +#include "../receiver/JFJochReceiverRugnuxRotation.h" #include "../receiver/JFJochReceiverService.h" #include "../rugnux/RugnuxDefaults.h" @@ -634,7 +634,7 @@ TEST_CASE("FPGAImagePuller_NotForPedestalAndHostSummation", "[FPGAImagePuller]") CHECK(out.status.images_collected == 2); } -// JFJochReceiverRugnux takes the same stream, with its own forwarding thread. +// JFJochReceiverRugnuxRotation takes the same stream, with its own forwarding thread. TEST_CASE("FPGAImagePuller_RugnuxReceiver", "[FPGAImagePuller]") { auto x = Base(DetJF(2), 6, 3); x.Conversion(); @@ -689,10 +689,10 @@ TEST_CASE("FPGAImagePuller_SaturatedAndErrorCountsFromTheCard", "[FPGAImagePulle SECTION("JFJochReceiverLite") { puller = Run(x, true, 2); } - SECTION("JFJochReceiverRugnux, stills") { + SECTION("JFJochReceiverRugnuxStills") { puller = Run(x, true, 2, true); } - SECTION("JFJochReceiverRugnux, rotation") { + SECTION("JFJochReceiverRugnuxRotation") { auto rotation = x; rotation.Goniometer(GoniometerAxis("omega", 0, 0.1, Coord(-1, 0, 0), std::nullopt)); puller = Run(rotation, true, 2, true); @@ -759,8 +759,8 @@ TEST_CASE("FPGAImagePuller_RugnuxReceiver_Summation", "[FPGAImagePuller]") { const JFCalibration calib = Calibration(x); FPGAImagePuller puller(x, mask, &calib, devices, logger, 2); - JFJochReceiverRugnux receiver(x, mask, puller, pusher, logger, 2, 2, nullptr, 30, live, Spots(), status, plots, - buffer); + JFJochReceiverRugnuxRotation receiver(x, mask, puller, pusher, logger, 2, 2, nullptr, 30, live, Spots(), status, + plots, buffer); receiver.StopReceiver(); puller.Stop(); CHECK(receiver.GetEfficiency() == 1.0f); diff --git a/tests/JFJochReceiverRugnuxTest.cpp b/tests/JFJochReceiverRugnuxTest.cpp index d0c591705..90c5dd599 100644 --- a/tests/JFJochReceiverRugnuxTest.cpp +++ b/tests/JFJochReceiverRugnuxTest.cpp @@ -10,7 +10,6 @@ #include "../image_analysis/structure_refinement/RugnuxModel.h" #include "../image_puller/TestImagePuller.h" #include "../image_pusher/ImagePusher.h" -#include "../receiver/JFJochReceiverRugnux.h" #include "../receiver/JFJochReceiverService.h" #include "../writer/HDF5Objects.h" diff --git a/tests/RugnuxResultFilesTest.cpp b/tests/RugnuxResultFilesTest.cpp index 13e66436b..b61576637 100644 --- a/tests/RugnuxResultFilesTest.cpp +++ b/tests/RugnuxResultFilesTest.cpp @@ -134,7 +134,7 @@ TEST_CASE("RugnuxResultFiles: rugnux's outputs -> END blocks -> CBOR -> writer, provenance.command_line = "rugnux -N 8 input_master.h5"; provenance.input_file = "/some/where/input_master.h5"; // The files as rugnux writes them from its outputs (RugnuxOutputFiles.cpp), and the same outputs as the - // broker puts them in the END message (JFJochReceiverRugnux::AddToEndMessage). + // broker puts them in the END message (JFJochReceiverRugnuxRotation::AddToEndMessage). EndMessage end{}; end.max_image_number = 100; end.merged = MergedMtzTable(reflections, TEST_CELL, experiment, provenance); @@ -369,7 +369,7 @@ TEST_CASE("RugnuxResultFiles: the refinement's model, maps and blobs reach the w .polar_contacts = 3, .metal_contacts = 1, .anomalous_sigma = 1.25}; { std::ofstream f(original + "_blobs.json"); f << DensityBlobsJson(DensityBlobsToMessage(blobs)); } - // As the broker puts them in the END message (JFJochReceiverRugnux::AddToEndMessage). + // As the broker puts them in the END message (JFJochReceiverRugnuxRotation::AddToEndMessage). EndMessage end{}; end.refined_model = PlacedModelFromStructure(TestModel(), TEST_CELL, sg.xhm()); end.refined_map_coefficients = maps;