receiver: the indexer belongs to the two receivers that index

IndexAndRefine and the indexer pool move from the base JFJochReceiver into
JFJochReceiverLite and JFJochReceiverFPGA. JFJochReceiverStream no longer
takes the pool, and the rugnux receivers no longer pass it nullptr. The
consensus cell and the indexer-error report of the end message are added
by both through one base helper, AddIndexingToEndMessage. No change for
Lite and FPGA; the other receivers carried an indexer that never ran and
reported no cell from it.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01HVeDQsU5a8AJUdJPgWevAK
This commit is contained in:
leonarski_fandClaude Opus 5.5 committed 2026-11-08 11:10:00 +01:00
1 parent 7080109a4a
commit 97cfae1fd8
11 files changed
+47 -34

No files matched your search

+16 -16
View File
@@ -4,6 +4,7 @@
#include "JFJochReceiver.h" #include "JFJochReceiver.h"
#include "../common/time_utc.h" #include "../common/time_utc.h"
#include "../frame_serialize/CBORStream2Serializer.h" #include "../frame_serialize/CBORStream2Serializer.h"
#include "../image_analysis/IndexAndRefine.h"
JFJochReceiver::JFJochReceiver(const DiffractionExperiment &in_experiment, JFJochReceiver::JFJochReceiver(const DiffractionExperiment &in_experiment,
ImageBuffer &in_image_buffer, ImageBuffer &in_image_buffer,
@@ -14,8 +15,7 @@ JFJochReceiver::JFJochReceiver(const DiffractionExperiment &in_experiment,
Logger &logger, Logger &logger,
const PixelMask &in_pixel_mask, const PixelMask &in_pixel_mask,
ZMQPreviewSocket *in_zmq_preview_socket, ZMQPreviewSocket *in_zmq_preview_socket,
ZMQMetadataSocket *in_zmq_metadata_socket, ZMQMetadataSocket *in_zmq_metadata_socket)
IndexerThreadPool *indexing_thread_pool)
: logger(logger), : logger(logger),
experiment(in_experiment), experiment(in_experiment),
spot_finding_settings(spot_finding_settings), spot_finding_settings(spot_finding_settings),
@@ -27,9 +27,7 @@ JFJochReceiver::JFJochReceiver(const DiffractionExperiment &in_experiment,
plots(in_plots), plots(in_plots),
scan_result(in_experiment), scan_result(in_experiment),
serialmx_filter(in_experiment), serialmx_filter(in_experiment),
pixel_mask(in_pixel_mask), pixel_mask(in_pixel_mask) {
// retain_outcomes=false: the receivers index only, there is no scaling or merging at the end
indexer(experiment, indexing_thread_pool, false, /*real_time=*/true) {
logger.Debug("Initializing receiver"); logger.Debug("Initializing receiver");
// Ensure there is nothing running for now // Ensure there is nothing running for now
if (!image_buffer.Finalize(std::chrono::seconds(1))) if (!image_buffer.Finalize(std::chrono::seconds(1)))
@@ -265,17 +263,6 @@ void JFJochReceiver::SendEndMessage() {
message.az_int_result["dataset"] = plots.GetAzIntProfile(); message.az_int_result["dataset"] = plots.GetAzIntProfile();
message.unit_cell = indexer.GetConsensusUnitCell();
// The images are written regardless, but the indexing rate is then not a statement about the crystal
if (const int64_t error_count = indexer.GetIndexerErrorCount(); error_count > 0) {
try {
indexer.RethrowIndexerError();
} catch (const std::exception &e) {
logger.Error("Indexing failed {} times in this collection, first with: {}. Those attempts are "
"left out of the indexing rate", error_count, e.what());
}
}
for (int i = 0; i < adu_histogram_module.size(); i++) for (int i = 0; i < adu_histogram_module.size(); i++)
message.adu_histogram["module" + std::to_string(i)] = adu_histogram_module[i]->GetHistogram(); message.adu_histogram["module" + std::to_string(i)] = adu_histogram_module[i]->GetHistogram();
@@ -304,6 +291,19 @@ void JFJochReceiver::SendEndMessage() {
zmq_preview_socket->EndDataCollection(message); zmq_preview_socket->EndDataCollection(message);
} }
void JFJochReceiver::AddIndexingToEndMessage(EndMessage &message, const IndexAndRefine &indexer) {
message.unit_cell = indexer.GetConsensusUnitCell();
// The images are written regardless, but the indexing rate is then not a statement about the crystal
if (const int64_t error_count = indexer.GetIndexerErrorCount(); error_count > 0) {
try {
indexer.RethrowIndexerError();
} catch (const std::exception &e) {
logger.Error("Indexing failed {} times in this collection, first with: {}. Those attempts are "
"left out of the indexing rate", error_count, e.what());
}
}
}
JFJochReceiverOutput JFJochReceiver::GetFinalStatistics() const { JFJochReceiverOutput JFJochReceiver::GetFinalStatistics() const {
JFJochReceiverOutput ret; JFJochReceiverOutput ret;
ret.efficiency = GetEfficiency(); ret.efficiency = GetEfficiency();
+5 -6
View File
@@ -18,8 +18,8 @@
#include "LossyFilter.h" #include "LossyFilter.h"
#include "../common/MovingAverage.h" #include "../common/MovingAverage.h"
#include "../common/ScanResultGenerator.h" #include "../common/ScanResultGenerator.h"
#include "../image_analysis/indexing/IndexerThreadPool.h"
#include "../image_analysis/IndexAndRefine.h" class IndexAndRefine;
class JFJochReceiver { class JFJochReceiver {
protected: protected:
@@ -82,8 +82,6 @@ protected:
std::optional<uint64_t> max_delay; std::optional<uint64_t> max_delay;
std::mutex max_delay_mutex; std::mutex max_delay_mutex;
IndexAndRefine indexer;
std::string writer_error; std::string writer_error;
void UpdateMaxDelay(uint64_t delay); void UpdateMaxDelay(uint64_t delay);
@@ -107,6 +105,8 @@ protected:
// The last word on the end message: a receiver with an analysis of its own fills in what it found // The last word on the end message: a receiver with an analysis of its own fills in what it found
virtual void AddToEndMessage(EndMessage &message) {} virtual void AddToEndMessage(EndMessage &message) {}
// For the receivers that index: the consensus cell, and how often the indexer failed
void AddIndexingToEndMessage(EndMessage &message, const IndexAndRefine &indexer);
virtual uint64_t GetImagesDropped() const { return 0; } virtual uint64_t GetImagesDropped() const { return 0; }
virtual void Cancel(const JFJochException &e); virtual void Cancel(const JFJochException &e);
public: public:
@@ -119,8 +119,7 @@ public:
Logger &logger, Logger &logger,
const PixelMask &pixel_mask, const PixelMask &pixel_mask,
ZMQPreviewSocket *zmq_preview_socket = nullptr, ZMQPreviewSocket *zmq_preview_socket = nullptr,
ZMQMetadataSocket *zmq_metadata_socket = nullptr, ZMQMetadataSocket *zmq_metadata_socket = nullptr);
IndexerThreadPool *indexing_thread_pool = nullptr);
virtual ~JFJochReceiver(); virtual ~JFJochReceiver();
virtual void StopReceiver() = 0; virtual void StopReceiver() = 0;
+7 -2
View File
@@ -29,9 +29,10 @@ JFJochReceiverFPGA::JFJochReceiverFPGA(const DiffractionExperiment &in_experimen
in_logger, in_logger,
in_pixel_mask, in_pixel_mask,
in_zmq_preview_socket, in_zmq_preview_socket,
in_zmq_metadata_socket, in_zmq_metadata_socket),
indexing_thread_pool),
calibration(nullptr), calibration(nullptr),
// retain_outcomes=false: the receiver indexes only, there is no scaling or merging at the end
indexer(experiment, indexing_thread_pool, false, /*real_time=*/true),
acquisition_device(in_aq_device), acquisition_device(in_aq_device),
ndatastreams(experiment.GetDataStreamsNum()), ndatastreams(experiment.GetDataStreamsNum()),
frame_transformation_nthreads( frame_transformation_nthreads(
@@ -498,6 +499,10 @@ float JFJochReceiverFPGA::GetProgress() const {
return static_cast<float>(acquisition_device[0].Counters().GetSlowestFrameNumber()) / static_cast<float>(frames); return static_cast<float>(acquisition_device[0].Counters().GetSlowestFrameNumber()) / static_cast<float>(frames);
} }
void JFJochReceiverFPGA::AddToEndMessage(EndMessage &message) {
AddIndexingToEndMessage(message, indexer);
}
void JFJochReceiverFPGA::FinalizeMeasurement() { void JFJochReceiverFPGA::FinalizeMeasurement() {
for (auto &future: frame_transformation_futures) for (auto &future: frame_transformation_futures)
future.get(); future.get();
+3
View File
@@ -39,6 +39,8 @@
class JFJochReceiverFPGA : public JFJochReceiver { class JFJochReceiverFPGA : public JFJochReceiver {
const JFCalibration *calibration; const JFCalibration *calibration;
IndexAndRefine indexer;
std::vector<uint16_t> roi_map; std::vector<uint16_t> roi_map;
std::vector<std::future<void>> frame_transformation_futures; std::vector<std::future<void>> frame_transformation_futures;
@@ -70,6 +72,7 @@ class JFJochReceiverFPGA : public JFJochReceiver {
uint32_t threadid, uint32_t threadid,
ModuleSummation &summation); ModuleSummation &summation);
void Cancel(const JFJochException &e) override; void Cancel(const JFJochException &e) override;
void AddToEndMessage(EndMessage &message) override;
void FinalizeMeasurement(); void FinalizeMeasurement();
void SendCalibration(); void SendCalibration();
+8 -2
View File
@@ -52,8 +52,9 @@ JFJochReceiverLite::JFJochReceiverLite(const DiffractionExperiment &in_experimen
in_plots, in_plots,
in_image_buffer, in_image_buffer,
in_zmq_preview_socket, in_zmq_preview_socket,
in_zmq_metadata_socket, in_zmq_metadata_socket),
indexing_thread_pool), // retain_outcomes=false: the receiver indexes only, there is no scaling or merging at the end
indexer(experiment, indexing_thread_pool, false, /*real_time=*/true),
analysis(nthreads) { analysis(nthreads) {
StartThreads(); StartThreads();
} }
@@ -89,6 +90,11 @@ void JFJochReceiverLite::ImageThread(uint32_t id) {
logger.Debug("Thread {} finished", id); logger.Debug("Thread {} finished", id);
} }
void JFJochReceiverLite::AddToEndMessage(EndMessage &message) {
JFJochReceiverStream::AddToEndMessage(message);
AddIndexingToEndMessage(message, indexer);
}
void JFJochReceiverLite::OnImage(uint32_t id, DataMessage &data_msg, std::chrono::steady_clock::time_point received) { void JFJochReceiverLite::OnImage(uint32_t id, DataMessage &data_msg, std::chrono::steady_clock::time_point received) {
auto start = std::chrono::high_resolution_clock::now(); auto start = std::chrono::high_resolution_clock::now();
+2
View File
@@ -10,6 +10,7 @@
#include "../image_analysis/MXAnalysisWithoutFPGA.h" #include "../image_analysis/MXAnalysisWithoutFPGA.h"
class JFJochReceiverLite : public JFJochReceiverStream { class JFJochReceiverLite : public JFJochReceiverStream {
IndexAndRefine indexer;
std::vector<std::unique_ptr<MXAnalysisWithoutFPGA>> analysis; // per image thread std::vector<std::unique_ptr<MXAnalysisWithoutFPGA>> analysis; // per image thread
// Each thread builds a GPU analysis of its own // Each thread builds a GPU analysis of its own
@@ -17,6 +18,7 @@ class JFJochReceiverLite : public JFJochReceiverStream {
void ImageThread(uint32_t id) override; void ImageThread(uint32_t id) override;
void OnImage(uint32_t id, DataMessage &message, std::chrono::steady_clock::time_point received) override; void OnImage(uint32_t id, DataMessage &message, std::chrono::steady_clock::time_point received) override;
void AddToEndMessage(EndMessage &message) override;
public: public:
JFJochReceiverLite(const DiffractionExperiment& experiment, JFJochReceiverLite(const DiffractionExperiment& experiment,
const PixelMask &pixel_mask, const PixelMask &pixel_mask,
+1 -1
View File
@@ -100,7 +100,7 @@ JFJochReceiverRugnuxRotation::JFJochReceiverRugnuxRotation(const DiffractionExpe
ZMQMetadataSocket *in_zmq_metadata_socket) ZMQMetadataSocket *in_zmq_metadata_socket)
: JFJochReceiverStream(in_experiment, in_pixel_mask, in_image_puller, in_image_pusher, in_logger, : 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_forward_nthreads, in_spot_finding_settings, in_current_status, in_plots,
in_image_buffer, in_zmq_preview_socket, in_zmq_metadata_socket, nullptr), in_image_buffer, in_zmq_preview_socket, in_zmq_metadata_socket),
reader(in_image_buffer), reader(in_image_buffer),
rugnux_nthreads(in_rugnux_nthreads), rugnux_nthreads(in_rugnux_nthreads),
host_budget(RugnuxHostBudget(in_image_buffer)), host_budget(RugnuxHostBudget(in_image_buffer)),
+1 -1
View File
@@ -26,7 +26,7 @@ JFJochReceiverRugnuxStills::JFJochReceiverRugnuxStills(const DiffractionExperime
IndexerThreadPool *in_indexing_thread_pool) IndexerThreadPool *in_indexing_thread_pool)
: JFJochReceiverStream(in_experiment, in_pixel_mask, in_image_puller, in_image_pusher, in_logger, : 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_forward_nthreads, in_spot_finding_settings, in_current_status, in_plots,
in_image_buffer, in_zmq_preview_socket, in_zmq_metadata_socket, nullptr), in_image_buffer, in_zmq_preview_socket, in_zmq_metadata_socket),
model(std::move(in_model)), model(std::move(in_model)),
deadline_s(in_deadline_s), deadline_s(in_deadline_s),
live_status(in_live_status), live_status(in_live_status),
+1
View File
@@ -12,6 +12,7 @@
#include "FPGAImagePuller.h" #include "FPGAImagePuller.h"
#include "../preview/PreviewImage.h" #include "../preview/PreviewImage.h"
#include "../image_puller/ZMQImagePuller.h" #include "../image_puller/ZMQImagePuller.h"
#include "../image_analysis/indexing/IndexerThreadPool.h"
#include "../acquisition_device/AcquisitionDeviceGroup.h" #include "../acquisition_device/AcquisitionDeviceGroup.h"
#include "../preview/ZMQMetadataSocket.h" #include "../preview/ZMQMetadataSocket.h"
+2 -4
View File
@@ -44,8 +44,7 @@ JFJochReceiverStream::JFJochReceiverStream(const DiffractionExperiment &in_exper
JFJochReceiverPlots &in_plots, JFJochReceiverPlots &in_plots,
ImageBuffer &in_image_buffer, ImageBuffer &in_image_buffer,
ZMQPreviewSocket *in_zmq_preview_socket, ZMQPreviewSocket *in_zmq_preview_socket,
ZMQMetadataSocket *in_zmq_metadata_socket, ZMQMetadataSocket *in_zmq_metadata_socket)
IndexerThreadPool *indexing_thread_pool)
: JFJochReceiver(in_experiment, : JFJochReceiver(in_experiment,
in_image_buffer, in_image_buffer,
in_image_pusher, in_image_pusher,
@@ -55,8 +54,7 @@ JFJochReceiverStream::JFJochReceiverStream(const DiffractionExperiment &in_exper
in_logger, in_logger,
in_pixel_mask, in_pixel_mask,
in_zmq_preview_socket, in_zmq_preview_socket,
in_zmq_metadata_socket, in_zmq_metadata_socket),
indexing_thread_pool),
image_puller(in_image_puller), image_puller(in_image_puller),
nthreads(in_nthreads), nthreads(in_nthreads),
measurement_started(1), measurement_started(1),
+1 -2
View File
@@ -68,8 +68,7 @@ public:
JFJochReceiverPlots &plots, JFJochReceiverPlots &plots,
ImageBuffer &image_buffer, ImageBuffer &image_buffer,
ZMQPreviewSocket *zmq_preview_socket = nullptr, ZMQPreviewSocket *zmq_preview_socket = nullptr,
ZMQMetadataSocket *zmq_metadata_socket = nullptr, ZMQMetadataSocket *zmq_metadata_socket = nullptr);
IndexerThreadPool *indexing_thread_pool = nullptr);
void StopReceiver() override; void StopReceiver() override;
void Cancel(bool silent) override; void Cancel(bool silent) override;