Receivers: one send block, one count, one stamp and one set of averages for the stream receivers

ForwardImage, CountImage, StampImage and AddToAverages move into JFJochReceiver, with SerializeImageOrDrop;
JFJochReceiverLite and JFJochReceiverRugnux call them instead of writing the same lines four times. The
dark mask keeps its store-only lines, and JFJochReceiverFPGA its own compress-into-slot block.

Two conventions had to be chosen. Compression ratio takes the stream's uncompressed size, as Lite did;
the stills path used pixels x image depth, the same number for any stream whose images are what its start
message says. Averages take only whole images, as stills and FPGA did: data_collection_efficiency is a
key of our own stream, a DECTRIS stream has no such field and counts as whole, so on DECTRIS nothing
changes and on the puller a partial image no longer enters the average.

The rotation forward's SetIndexed(false) becomes value_or(false): a forwarded image has no indexing result.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01HVeDQsU5a8AJUdJPgWevAK

receiver/JFJochReceiver.cpp

@@ -5,6 +5,7 @@
 #include "../common/CUDAWrapper.h"
 #include "../common/time_utc.h"
 #include "JFJochCompressor.h"
+#include "../frame_serialize/CBORStream2Serializer.h"
 
 JFJochReceiver::JFJochReceiver(const DiffractionExperiment &in_experiment,
                                ImageBuffer &in_image_buffer,
@@ -161,6 +162,101 @@ void JFJochReceiver::SaveStartMessageToImageBuffer(const StartMessage &msg) {
     image_buffer.SaveStartMessage(buffer);
 }
 
+// Serialize a processed image into its buffer slot. The slot holds the compressed
+// image plus the per-image CBOR metadata (spot list, reflection list, ...); if an
+// unusually rich frame makes that metadata too large to fit alongside the image,
+// drop just this frame (log + recycle the slot) instead of letting the serializer
+// throw, which would abort the whole data collection. Returns true when the image
+// was serialized and the slot is ready for the caller to commit.
+bool JFJochReceiver::SerializeImageOrDrop(CBORStream2Serializer &serializer, DataMessage &msg,
+                                          ZeroCopyReturnValue &loc, size_t slot_size, Logger &logger) {
+    try {
+        serializer.SerializeImage(msg);
+        return true;
+    } catch (const JFJochException &) {
+        // Confirm this is a slot-overflow (not some other CBOR error) by re-serializing
+        // the metadata alone with an empty image placeholder: that always fits when the
+        // image is what overflowed, and gives the exact metadata size for the log. If it
+        // throws too, it is not a simple overflow and the exception propagates.
+        const auto image = msg.image;
+        msg.image = CompressedImage(nullptr, 0, image.GetWidth(), image.GetHeight(),
+                                    image.GetMode(), image.GetCompressionAlgorithm(),
+                                    image.GetChannel());
+        CBORStream2Serializer meas(static_cast<uint8_t *>(loc.GetImage()), slot_size);
+        meas.SerializeImage(msg);
+        const size_t metadata_size = meas.GetImageAppendOffset();
+        msg.image = image;
+
+        logger.Error("Dropping image {} - CBOR metadata too large to store image: "
+                     "metadata {} B + image {} B do not fit in buffer slot {} B",
+                     msg.number, metadata_size, image.GetCompressedSize(), slot_size);
+        loc.release();
+        return false;
+    }
+}
+
+void JFJochReceiver::CountImage(const DataMessage &message) {
+    ++images_collected;
+    compressed_size += message.image.GetCompressedSize();
+    uncompressed_size += message.image.GetUncompressedSize();
+    UpdateMaxImageReceived(message.number);
+}
+
+void JFJochReceiver::StampImage(DataMessage &message, size_t fifo_delay) {
+    message.original_number = message.number;
+    message.user_data = experiment.GetImageAppendix();
+    message.run_number = experiment.GetRunNumber();
+    message.run_name = experiment.GetRunName();
+    message.receiver_buf_available = image_buffer.GetAvailSlots();
+    message.receiver_aq_dev_delay = fifo_delay;
+    // Uncompressed/compressed (e.g. 7x), as JFJochReceiverFPGA and JFJochReceiver::GetFinalStatistics
+    if (message.image.GetCompressedSize() > 0)
+        message.compression_ratio = static_cast<float>(message.image.GetUncompressedSize())
+                                    / static_cast<float>(message.image.GetCompressedSize());
+}
+
+void JFJochReceiver::AddToAverages(const DataMessage &message) {
+    // A stream that does not state it (DECTRIS) delivers whole images
+    if (message.image_collection_efficiency.value_or(1.0f) != 1.0f)
+        return;
+    saturated_pixels.Add(message.saturated_pixel_count);
+    error_pixels.Add(message.error_pixel_count);
+    if (message.roi.contains("beam")) {
+        roi_beam_npixel.Add(message.roi.at("beam").pixels);
+        roi_beam_sum.Add(message.roi.at("beam").sum);
+    }
+}
+
+bool JFJochReceiver::ForwardImage(DataMessage &message, ZeroCopyReturnValue *loc) {
+    if (loc == nullptr) {
+        writer_queue_full = true;
+        return false;
+    }
+    auto writer_buffer = (uint8_t *) loc->GetImage();
+    CBORStream2Serializer serializer(writer_buffer, experiment.GetImageBufferLocationSize());
+    if (!SerializeImageOrDrop(serializer, message, *loc, experiment.GetImageBufferLocationSize(), logger))
+        return false;
+    loc->SetImageNumber(message.number);
+    loc->SetImageSize(serializer.GetBufferSize());
+    loc->SetIndexed(message.indexing_result.value_or(false));
+    loc->ReadyToSend();
+
+    if (zmq_preview_socket != nullptr)
+        zmq_preview_socket->SendImage(writer_buffer, serializer.GetBufferSize());
+
+    if (zmq_metadata_socket != nullptr)
+        zmq_metadata_socket->AddDataMessage(message);
+
+    if (push_images_to_writer) {
+        // Only count images the pusher accepted; TCP refuses them on a broken/absent connection
+        if (image_pusher.SendImage(*loc))
+            ++images_sent;
+    } else
+        loc->release();
+    UpdateMaxImageSent(message.number);
+    return true;
+}
+
 void JFJochReceiver::SendEndMessage() {
     EndMessage message{};
     message.max_image_number = max_image_number_sent;

receiver/JFJochReceiver.h

@@ -100,6 +100,16 @@ protected:
     void SendStartMessage(StartMessage message);
     void SendEndMessage();
     void SaveStartMessageToImageBuffer(const StartMessage &msg);
+
+    static bool SerializeImageOrDrop(CBORStream2Serializer &serializer, DataMessage &msg,
+                                     ZeroCopyReturnValue &loc, size_t slot_size, Logger &logger);
+    void CountImage(const DataMessage &message);
+    // What the receiver states about an image from a stream
+    void StampImage(DataMessage &message, size_t fifo_delay);
+    void AddToAverages(const DataMessage &message);
+    // Serializes into the slot and sends it on; false if there is no slot or the image does not fit
+    bool ForwardImage(DataMessage &message, ZeroCopyReturnValue *loc);
+
     // The last word on the end message before it is sent: a receiver whose analysis is not the
     // per-image one fills in what it found.
     virtual void AddToEndMessage(EndMessage &message) {}

receiver/JFJochReceiverLite.cpp

@@ -7,39 +7,6 @@
 
 using namespace std::chrono_literals;
 
-// Serialize a processed image into its buffer slot. The slot holds the compressed
-// image plus the per-image CBOR metadata (spot list, reflection list, ...); if an
-// unusually rich frame makes that metadata too large to fit alongside the image,
-// drop just this frame (log + recycle the slot) instead of letting the serializer
-// throw, which would abort the whole data collection. Returns true when the image
-// was serialized and the slot is ready for the caller to commit.
-bool JFJochReceiverLite::SerializeImageOrDrop(CBORStream2Serializer &serializer, DataMessage &msg,
-                                              ZeroCopyReturnValue &loc, size_t slot_size, Logger &logger) {
-    try {
-        serializer.SerializeImage(msg);
-        return true;
-    } catch (const JFJochException &) {
-        // Confirm this is a slot-overflow (not some other CBOR error) by re-serializing
-        // the metadata alone with an empty image placeholder: that always fits when the
-        // image is what overflowed, and gives the exact metadata size for the log. If it
-        // throws too, it is not a simple overflow and the exception propagates.
-        const auto image = msg.image;
-        msg.image = CompressedImage(nullptr, 0, image.GetWidth(), image.GetHeight(),
-                                    image.GetMode(), image.GetCompressionAlgorithm(),
-                                    image.GetChannel());
-        CBORStream2Serializer meas(static_cast<uint8_t *>(loc.GetImage()), slot_size);
-        meas.SerializeImage(msg);
-        const size_t metadata_size = meas.GetImageAppendOffset();
-        msg.image = image;
-
-        logger.Error("Dropping image {} - CBOR metadata too large to store image: "
-                     "metadata {} B + image {} B do not fit in buffer slot {} B",
-                     msg.number, metadata_size, image.GetCompressedSize(), slot_size);
-        loc.release();
-        return false;
-    }
-}
-
 int64_t JFJochReceiverLite::NumberOfDataAnalysisThreads(int64_t requested_thread_number,
                                                         const DiffractionExperiment& in_experiment) {
     int64_t number_of_images = in_experiment.GetImageNum();
@@ -253,15 +220,11 @@ void JFJochReceiverLite::MaskThread(uint32_t id) {
                 end_message_received = true;
                 logger.Debug("Thread {} end message received in JFJochReceiverLite", id);
             } else if (msg->cbor->data_message) {
-                ++images_collected;
                 DataMessage data_msg = msg->cbor->data_message.value();
+                CountImage(data_msg);
 
                 dark_mask_analysis.AnalyzeImage(data_msg, buffer);
 
-                compressed_size += data_msg.image.GetCompressedSize();
-                uncompressed_size += data_msg.image.GetUncompressedSize();
-                UpdateMaxImageReceived(data_msg.number);
-
                 auto loc = image_buffer.GetImageSlot();
 
                 if (loc == nullptr)
@@ -327,17 +290,9 @@ void JFJochReceiverLite::DataAnalysisThread(uint32_t id) {
                 auto start = std::chrono::high_resolution_clock::now();
                 DataMessage data_msg = msg->cbor->data_message.value();
 
-                ++images_collected;
+                CountImage(data_msg);
                 collection_efficiency_sum += data_msg.image_collection_efficiency.value_or(1.0f);
 
-                auto compressed_size_img = data_msg.image.GetCompressedSize();
-                auto uncompressed_size_img = data_msg.image.GetUncompressedSize();
-
-                compressed_size += compressed_size_img;
-                uncompressed_size += uncompressed_size_img;
-
-                UpdateMaxImageReceived(data_msg.number);
-
                 auto image_start_time = std::chrono::high_resolution_clock::now();
 
                 AzimuthalIntegrationProfile profile(*az_int_mapping);
@@ -348,65 +303,16 @@ void JFJochReceiverLite::DataAnalysisThread(uint32_t id) {
                 std::chrono::duration<float> image_duration = image_end_time - image_start_time;
 
                 data_msg.processing_time_s = image_duration.count();
-                data_msg.original_number = data_msg.number;
-                data_msg.user_data = experiment.GetImageAppendix();
-                data_msg.run_number = experiment.GetRunNumber();
-                data_msg.run_name = experiment.GetRunName();
-                data_msg.receiver_buf_available = image_buffer.GetAvailSlots();
-                data_msg.receiver_aq_dev_delay = image_puller.GetCurrentFifoUtilization();
-
-                // Compression ratio is uncompressed/compressed (e.g. 7x), matching the FPGA path
-                // (JFJochReceiverFPGA) and the aggregate (JFJochReceiver::GetFinalStatistics) - not the
-                // reciprocal.
-                if (compressed_size_img > 0)
-                    data_msg.compression_ratio = static_cast<float>(uncompressed_size_img)
-                                                 / static_cast<float>(compressed_size_img);
-
-                saturated_pixels.Add(data_msg.saturated_pixel_count);
-                error_pixels.Add(data_msg.error_pixel_count);
-
-                if (data_msg.roi.contains("beam")) {
-                    roi_beam_npixel.Add(data_msg.roi["beam"].pixels);
-                    roi_beam_sum.Add(data_msg.roi["beam"].sum);
-                }
+                StampImage(data_msg, image_puller.GetCurrentFifoUtilization());
+                AddToAverages(data_msg);
 
                 plots.Add(data_msg, profile);
                 scan_result.Add(data_msg);
 
                 if (!serialmx_filter.ApplyFilter(data_msg))
                     ++images_skipped;
-                else {
-                    auto loc = image_buffer.GetImageSlot();
-
-                    if (loc == nullptr)
-                        writer_queue_full = true;
-                    else {
-                        auto writer_buffer = (uint8_t *) loc->GetImage();
-                        CBORStream2Serializer serializer(writer_buffer, experiment.GetImageBufferLocationSize());
-                        if (SerializeImageOrDrop(serializer, data_msg, *loc,
-                                                 experiment.GetImageBufferLocationSize(), logger)) {
-                            loc->SetImageNumber(data_msg.number);
-                            loc->SetImageSize(serializer.GetBufferSize());
-                            loc->SetIndexed(data_msg.indexing_result.value_or(false));
-                            loc->ReadyToSend();
-
-                            if (zmq_preview_socket != nullptr)
-                                zmq_preview_socket->SendImage(writer_buffer, serializer.GetBufferSize());
-
-                            if (zmq_metadata_socket != nullptr)
-                                zmq_metadata_socket->AddDataMessage(data_msg);
-
-                            if (push_images_to_writer) {
-                                // Only count images the pusher accepted; TCP refuses them on a broken/absent
-                                // connection. Images actually written are GetImagesWritten() (ACK-based).
-                                if (image_pusher.SendImage(*loc))
-                                    ++images_sent;
-                            } else
-                                loc->release();
-                            UpdateMaxImageSent(data_msg.number);
-                        }
-                    }
-                }
+                else
+                    ForwardImage(data_msg, image_buffer.GetImageSlot());
 
                 auto end = std::chrono::high_resolution_clock::now();
                 std::chrono::duration<float> duration_total = end - start;

receiver/JFJochReceiverLite.h

@@ -64,9 +64,6 @@ class JFJochReceiverLite : public JFJochReceiver {
     void MeasurementThread();
     void DataAnalysisThread(uint32_t id);
 public:
-    // Shared with JFJochReceiverRugnux, which forwards a collection the same way.
-    static bool SerializeImageOrDrop(CBORStream2Serializer &serializer, DataMessage &msg,
-                                     ZeroCopyReturnValue &loc, size_t slot_size, Logger &logger);
     // Takes what the detector's start message decides into the experiment.
     static void ApplyStartMessage(DiffractionExperiment &experiment, const StartMessage &message);
 

receiver/JFJochReceiverRugnux.cpp

@@ -13,7 +13,6 @@
 #include "../common/HostMemory.h"
 #include "../common/time_utc.h"
 #include "../frame_serialize/CBORStream2Deserializer.h"
-#include "../frame_serialize/CBORStream2Serializer.h"
 #include "../image_analysis/indexing/IndexerThreadPool.h"
 #include "../rugnux/RugnuxDefaults.h"
 #include "../rugnux/RugnuxLive.h"
@@ -377,11 +376,8 @@ void JFJochReceiverRugnux::HoldStillsImage(const DataMessage &message) {
         logger.Error("Dropping image {} - the collection has {} images", message.number, experiment.GetImageNum());
         return;
     }
-    ++images_collected;
+    CountImage(message);
     const size_t size = message.image.GetCompressedSize();
-    compressed_size += size;
-    uncompressed_size += message.image.GetUncompressedSize();
-    UpdateMaxImageReceived(message.number);
 
     // 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
@@ -402,11 +398,7 @@ void JFJochReceiverRugnux::HoldStillsImage(const DataMessage &message) {
     held->message.image = CompressedImage(held->slot->GetImage(), size, message.image.GetWidth(),
                                           message.image.GetHeight(), message.image.GetMode(),
                                           message.image.GetCompressionAlgorithm(), message.image.GetChannel());
-    held->message.original_number = message.number;
-    held->message.user_data = experiment.GetImageAppendix();
-    held->message.run_number = experiment.GetRunNumber();
-    held->message.run_name = experiment.GetRunName();
-    held->message.receiver_aq_dev_delay = image_puller.GetCurrentFifoUtilization();
+    StampImage(held->message, image_puller.GetCurrentFifoUtilization());
     stills->Hold(held);
 }
 
@@ -430,49 +422,13 @@ void JFJochReceiverRugnux::SendAnalysedImage(DataMessage &message, const Azimuth
         message.receiver_buf_available = status.available_slots;
         message.receiver_buf_in_preparation = status.preparation_slots;
         message.receiver_buf_in_sending = status.sending_slots;
-        if (message.image.GetCompressedSize() > 0)
-            message.compression_ratio = static_cast<float>(experiment.GetPixelsNum() * experiment.GetByteDepthImage())
-                                        / static_cast<float>(message.image.GetCompressedSize());
-
-        // A stream that does not state it (DECTRIS) delivers whole images.
-        if (message.image_collection_efficiency.value_or(1.0f) == 1.0f) {
-            saturated_pixels.Add(message.saturated_pixel_count);
-            error_pixels.Add(message.error_pixel_count);
-            if (message.roi.contains("beam")) {
-                roi_beam_npixel.Add(message.roi["beam"].pixels);
-                roi_beam_sum.Add(message.roi["beam"].sum);
-            }
-        }
-
+        AddToAverages(message);
         scan_result.Add(message);
 
         if (!serialmx_filter.ApplyFilter(message))
             ++images_skipped;
-        else if (auto loc = AnalysedImageSlot(); loc == nullptr)
-            writer_queue_full = true;
-        else {
-            auto writer_buffer = (uint8_t *) loc->GetImage();
-            CBORStream2Serializer serializer(writer_buffer, experiment.GetImageBufferLocationSize());
-            if (JFJochReceiverLite::SerializeImageOrDrop(serializer, message, *loc,
-                                                         experiment.GetImageBufferLocationSize(), logger)) {
-                loc->SetImageNumber(message.number);
-                loc->SetImageSize(serializer.GetBufferSize());
-                loc->SetIndexed(message.indexing_result.value_or(false));
-                loc->ReadyToSend();
-
-                if (zmq_preview_socket != nullptr)
-                    zmq_preview_socket->SendImage(writer_buffer, serializer.GetBufferSize());
-                if (zmq_metadata_socket != nullptr)
-                    zmq_metadata_socket->AddDataMessage(message);
-
-                if (push_images_to_writer) {
-                    if (image_pusher.SendImage(*loc))
-                        ++images_sent;
-                } else
-                    loc->release();
-                UpdateMaxImageSent(message.number);
-            }
-        }
+        else
+            ForwardImage(message, AnalysedImageSlot());
 
         plots.Add(message, profile);
         current_status.SetProgress(GetProgress());
@@ -699,49 +655,9 @@ void JFJochReceiverRugnux::ForwardThread(uint32_t id) {
             } else if (msg->cbor->data_message) {
                 live_status.ImageReceived(msg->received);
                 DataMessage data_msg = msg->cbor->data_message.value();
-                ++images_collected;
-
-                const auto compressed_size_img = data_msg.image.GetCompressedSize();
-                const auto uncompressed_size_img = data_msg.image.GetUncompressedSize();
-                compressed_size += compressed_size_img;
-                uncompressed_size += uncompressed_size_img;
-                UpdateMaxImageReceived(data_msg.number);
-
-                data_msg.original_number = data_msg.number;
-                data_msg.user_data = experiment.GetImageAppendix();
-                data_msg.run_number = experiment.GetRunNumber();
-                data_msg.run_name = experiment.GetRunName();
-                data_msg.receiver_buf_available = image_buffer.GetAvailSlots();
-                data_msg.receiver_aq_dev_delay = image_puller.GetCurrentFifoUtilization();
-                if (compressed_size_img > 0)
-                    data_msg.compression_ratio = static_cast<float>(uncompressed_size_img)
-                                                 / static_cast<float>(compressed_size_img);
-
-                auto loc = image_buffer.GetImageSlot();
-                if (loc == nullptr)
-                    writer_queue_full = true;
-                else {
-                    auto writer_buffer = (uint8_t *) loc->GetImage();
-                    CBORStream2Serializer serializer(writer_buffer, experiment.GetImageBufferLocationSize());
-                    if (JFJochReceiverLite::SerializeImageOrDrop(serializer, data_msg, *loc,
-                                                                 experiment.GetImageBufferLocationSize(), logger)) {
-                        loc->SetImageNumber(data_msg.number);
-                        loc->SetImageSize(serializer.GetBufferSize());
-                        loc->SetIndexed(false);
-                        loc->ReadyToSend();
-
-                        if (zmq_preview_socket != nullptr)
-                            zmq_preview_socket->SendImage(writer_buffer, serializer.GetBufferSize());
-                        if (zmq_metadata_socket != nullptr)
-                            zmq_metadata_socket->AddDataMessage(data_msg);
-                        if (push_images_to_writer) {
-                            if (image_pusher.SendImage(*loc))
-                                ++images_sent;
-                        } else
-                            loc->release();
-                        UpdateMaxImageSent(data_msg.number);
-                    }
-                }
+                CountImage(data_msg);
+                StampImage(data_msg, image_puller.GetCurrentFifoUtilization());
+                ForwardImage(data_msg, image_buffer.GetImageSlot());
             }
             UpdateMaxDelay(image_puller.GetCurrentFifoUtilization());
             current_status.SetProgress(GetProgress());