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

receiver/JFJochReceiver.cpp

@@ -4,6 +4,7 @@
 #include "JFJochReceiver.h"
 #include "../common/time_utc.h"
 #include "../frame_serialize/CBORStream2Serializer.h"
+#include "../image_analysis/IndexAndRefine.h"
 
 JFJochReceiver::JFJochReceiver(const DiffractionExperiment &in_experiment,
                                ImageBuffer &in_image_buffer,
@@ -14,8 +15,7 @@ JFJochReceiver::JFJochReceiver(const DiffractionExperiment &in_experiment,
                                Logger &logger,
                                const PixelMask &in_pixel_mask,
                                ZMQPreviewSocket *in_zmq_preview_socket,
-                               ZMQMetadataSocket *in_zmq_metadata_socket,
+                               ZMQMetadataSocket *in_zmq_metadata_socket)
-                               IndexerThreadPool *indexing_thread_pool)
         : logger(logger),
           experiment(in_experiment),
           spot_finding_settings(spot_finding_settings),
@@ -27,9 +27,7 @@ JFJochReceiver::JFJochReceiver(const DiffractionExperiment &in_experiment,
           plots(in_plots),
           scan_result(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");
     // Ensure there is nothing running for now
     if (!image_buffer.Finalize(std::chrono::seconds(1)))
@@ -265,17 +263,6 @@ void JFJochReceiver::SendEndMessage() {
 
     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++)
         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);
 }
 
+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 ret;
     ret.efficiency = GetEfficiency();

receiver/JFJochReceiver.h

@@ -18,8 +18,8 @@
 #include "LossyFilter.h"
 #include "../common/MovingAverage.h"
 #include "../common/ScanResultGenerator.h"
-#include "../image_analysis/indexing/IndexerThreadPool.h"
+
-#include "../image_analysis/IndexAndRefine.h"
+class IndexAndRefine;
 
 class JFJochReceiver {
 protected:
@@ -82,8 +82,6 @@ protected:
     std::optional<uint64_t> max_delay;
     std::mutex max_delay_mutex;
 
-    IndexAndRefine indexer;
-
     std::string writer_error;
 
     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
     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 void Cancel(const JFJochException &e);
 public:
@@ -119,8 +119,7 @@ public:
                    Logger &logger,
                    const PixelMask &pixel_mask,
                    ZMQPreviewSocket *zmq_preview_socket = nullptr,
-                   ZMQMetadataSocket *zmq_metadata_socket = nullptr,
+                   ZMQMetadataSocket *zmq_metadata_socket = nullptr);
-                   IndexerThreadPool *indexing_thread_pool = nullptr);
     virtual ~JFJochReceiver();
 
     virtual void StopReceiver() = 0;

receiver/JFJochReceiverFPGA.cpp

@@ -29,9 +29,10 @@ JFJochReceiverFPGA::JFJochReceiverFPGA(const DiffractionExperiment &in_experimen
                      in_logger,
                      in_pixel_mask,
                      in_zmq_preview_socket,
-                     in_zmq_metadata_socket,
+                     in_zmq_metadata_socket),
-                     indexing_thread_pool),
       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),
       ndatastreams(experiment.GetDataStreamsNum()),
       frame_transformation_nthreads(
@@ -498,6 +499,10 @@ float JFJochReceiverFPGA::GetProgress() const {
     return static_cast<float>(acquisition_device[0].Counters().GetSlowestFrameNumber()) / static_cast<float>(frames);
 }
 
+void JFJochReceiverFPGA::AddToEndMessage(EndMessage &message) {
+    AddIndexingToEndMessage(message, indexer);
+}
+
 void JFJochReceiverFPGA::FinalizeMeasurement() {
     for (auto &future: frame_transformation_futures)
         future.get();

receiver/JFJochReceiverFPGA.h

@@ -39,6 +39,8 @@
 class JFJochReceiverFPGA : public JFJochReceiver  {
     const JFCalibration *calibration;
 
+    IndexAndRefine indexer;
+
     std::vector<uint16_t> roi_map;
 
     std::vector<std::future<void>> frame_transformation_futures;
@@ -70,6 +72,7 @@ class JFJochReceiverFPGA : public JFJochReceiver  {
                          uint32_t threadid,
                          ModuleSummation &summation);
     void Cancel(const JFJochException &e) override;
+    void AddToEndMessage(EndMessage &message) override;
     void FinalizeMeasurement();
 
     void SendCalibration();

receiver/JFJochReceiverLite.cpp

@@ -52,8 +52,9 @@ JFJochReceiverLite::JFJochReceiverLite(const DiffractionExperiment &in_experimen
                            in_plots,
                            in_image_buffer,
                            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) {
     StartThreads();
 }
@@ -89,6 +90,11 @@ void JFJochReceiverLite::ImageThread(uint32_t 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) {
     auto start = std::chrono::high_resolution_clock::now();
 

receiver/JFJochReceiverLite.h

@@ -10,6 +10,7 @@
 #include "../image_analysis/MXAnalysisWithoutFPGA.h"
 
 class JFJochReceiverLite : public JFJochReceiverStream {
+    IndexAndRefine indexer;
     std::vector<std::unique_ptr<MXAnalysisWithoutFPGA>> analysis;   // per image thread
 
     // Each thread builds a GPU analysis of its own
@@ -17,6 +18,7 @@ class JFJochReceiverLite : public JFJochReceiverStream {
 
     void ImageThread(uint32_t id) override;
     void OnImage(uint32_t id, DataMessage &message, std::chrono::steady_clock::time_point received) override;
+    void AddToEndMessage(EndMessage &message) override;
 public:
     JFJochReceiverLite(const DiffractionExperiment& experiment,
                        const PixelMask &pixel_mask,

receiver/JFJochReceiverRugnuxRotation.cpp

@@ -100,7 +100,7 @@ JFJochReceiverRugnuxRotation::JFJochReceiverRugnuxRotation(const DiffractionExpe
                                                            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),
+                           in_image_buffer, in_zmq_preview_socket, in_zmq_metadata_socket),
       reader(in_image_buffer),
       rugnux_nthreads(in_rugnux_nthreads),
       host_budget(RugnuxHostBudget(in_image_buffer)),

receiver/JFJochReceiverRugnuxStills.cpp

@@ -26,7 +26,7 @@ JFJochReceiverRugnuxStills::JFJochReceiverRugnuxStills(const DiffractionExperime
                                                          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),
+                           in_image_buffer, in_zmq_preview_socket, in_zmq_metadata_socket),
       model(std::move(in_model)),
       deadline_s(in_deadline_s),
       live_status(in_live_status),

receiver/JFJochReceiverService.h

@@ -12,6 +12,7 @@
 #include "FPGAImagePuller.h"
 #include "../preview/PreviewImage.h"
 #include "../image_puller/ZMQImagePuller.h"
+#include "../image_analysis/indexing/IndexerThreadPool.h"
 #include "../acquisition_device/AcquisitionDeviceGroup.h"
 
 #include "../preview/ZMQMetadataSocket.h"

receiver/JFJochReceiverStream.cpp

@@ -44,8 +44,7 @@ JFJochReceiverStream::JFJochReceiverStream(const DiffractionExperiment &in_exper
                                            JFJochReceiverPlots &in_plots,
                                            ImageBuffer &in_image_buffer,
                                            ZMQPreviewSocket *in_zmq_preview_socket,
-                                           ZMQMetadataSocket *in_zmq_metadata_socket,
+                                           ZMQMetadataSocket *in_zmq_metadata_socket)
-                                           IndexerThreadPool *indexing_thread_pool)
     : JFJochReceiver(in_experiment,
                      in_image_buffer,
                      in_image_pusher,
@@ -55,8 +54,7 @@ JFJochReceiverStream::JFJochReceiverStream(const DiffractionExperiment &in_exper
                      in_logger,
                      in_pixel_mask,
                      in_zmq_preview_socket,
-                     in_zmq_metadata_socket,
+                     in_zmq_metadata_socket),
-                     indexing_thread_pool),
       image_puller(in_image_puller),
       nthreads(in_nthreads),
       measurement_started(1),

receiver/JFJochReceiverStream.h

@@ -68,8 +68,7 @@ public:
                          JFJochReceiverPlots &plots,
                          ImageBuffer &image_buffer,
                          ZMQPreviewSocket *zmq_preview_socket = nullptr,
-                         ZMQMetadataSocket *zmq_metadata_socket = nullptr,
+                         ZMQMetadataSocket *zmq_metadata_socket = nullptr);
-                         IndexerThreadPool *indexing_thread_pool = nullptr);
 
     void StopReceiver() override;
     void Cancel(bool silent) override;