Tests: live stills through ZeroMQ and the writer, as the removed GPU-stills tests

The two end-to-end GPU-stills tests of the FPGA receiver went with live stills
moving to JFJochReceiverRugnuxStills. Their checks return on the new path, with a
real ZMQStream2Pusher and StreamWriter: the writer finishes without error, there is
no progress value after Stop, the elapsed time covers the last image plus its
latency to written, and the five copies of the same test frame index at 1.0. The
writer-down twin checks that a collection with no writer listening ends with the
error and the next one runs.

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

tests/JFJochReceiverRugnuxTest.cpp

@@ -9,13 +9,16 @@
 #include "../compression/JFJochCompressor.h"
 #include "../image_analysis/structure_refinement/RugnuxModel.h"
 #include "../image_puller/TestImagePuller.h"
+#include "../image_puller/ZMQImagePuller.h"
 #include "../image_pusher/ImagePusher.h"
+#include "../image_pusher/ZMQStream2Pusher.h"
 #include "../receiver/JFJochReceiverDarkMask.h"
 #include "../receiver/JFJochReceiverLite.h"
 #include "../receiver/JFJochReceiverRugnuxRotation.h"
 #include "../receiver/JFJochReceiverRugnuxStills.h"
 #include "../receiver/JFJochReceiverService.h"
 #include "../writer/HDF5Objects.h"
+#include "../writer/StreamWriter.h"
 
 namespace {
     // What the writer would be given, in order.
@@ -74,7 +77,8 @@ namespace {
     public:
         const std::shared_ptr<TestImagePuller> puller = std::make_shared<TestImagePuller>();
 
-        Stream(const DiffractionExperiment &x, int64_t nframes, bool blank = false)
+        // With same_frame >= 0, every frame is that frame of the test data.
+        Stream(const DiffractionExperiment &x, int64_t nframes, bool blank = false, int64_t same_frame = -1)
                 : width(x.GetXPixelsNum()), height(x.GetYPixelsNum()) {
             HDF5ReadOnlyFile data("../../tests/test_data/compression_benchmark.h5");
             HDF5DataSet dataset(data, "/entry/data/data");
@@ -82,7 +86,8 @@ namespace {
             std::vector<int16_t> image(file_space.GetDimensions()[1] * file_space.GetDimensions()[2]);
             JFJochBitShuffleCompressor compressor(CompressionAlgorithm::BSHUF_LZ4);
             for (int64_t i = 0; i < nframes; i++) {
-                const std::vector<hsize_t> start = {static_cast<hsize_t>(i % file_space.GetDimensions()[0]), 0, 0};
+                const int64_t frame = (same_frame >= 0) ? same_frame : i;
+                const std::vector<hsize_t> start = {static_cast<hsize_t>(frame % file_space.GetDimensions()[0]), 0, 0};
                 const std::vector<hsize_t> size = {1, file_space.GetDimensions()[1], file_space.GetDimensions()[2]};
                 if (blank)
                     std::fill(image.begin(), image.end(), 0);
@@ -439,6 +444,105 @@ TEST_CASE("JFJochReceiverRugnux_Stills_WriterDown", "[JFJochReceiverRugnux]") {
     CHECK(pusher.Events() == std::vector<std::string>{"start", "image", "end"});
 }
 
+TEST_CASE("JFJochReceiverRugnux_Stills_ZMQWriter", "[JFJochReceiverRugnux]") {
+    // As the removed GPU-stills test of the FPGA receiver: five copies of one frame, through ZeroMQ to the writer.
+    Logger logger(Catch::getResultCapture().getCurrentTestName());
+    RegisterHDF5Filter();
+
+    const int64_t images = 5;
+    DiffractionExperiment x = StillsExperiment(images);
+    x.FilePrefix("rugnux_stills_zmq").SetFileWriterFormat(FileWriterFormat::NXmxVDS).OverwriteExistingFiles(true);
+    PixelMask mask(x);
+
+    Stream stream(x, images, false, 4);
+    stream.Frames(0, images);
+    stream.End();
+
+    ZMQStream2Pusher pusher({"ipc://*"});
+    pusher.WriterNotificationSocket("ipc://*");
+
+    ZMQImagePuller puller(pusher.GetAddress()[0]);
+    StreamWriter writer(logger, puller);
+    auto writer_future = std::async(std::launch::async, &StreamWriter::Run, &writer);
+
+    AcquisitionDeviceGroup group;
+    JFJochReceiverService service(group, logger, pusher);
+    service.NumThreads(4);
+    service.Indexing(x.GetIndexingSettings());
+    service.RugnuxLive(true, nullptr, 4, 60);
+
+    REQUIRE(!service.GetProgress().has_value());
+
+    SpotFindingSettings settings = DiffractionExperiment::DefaultDataProcessingSettings();
+    settings.signal_to_noise_threshold = 2.5;
+    settings.photon_count_threshold = 5;
+    settings.min_pix_per_spot = 1;
+    settings.max_pix_per_spot = 200;
+    settings.high_resolution_limit = 2.0;
+    settings.low_resolution_limit = 50.0;
+    service.SetSpotFindingSettings(settings);
+
+    service.Start(x, mask, nullptr, stream.puller);
+    const auto out = service.Stop();
+
+    CHECK(out.efficiency == 1.0f);
+    REQUIRE(out.status.indexing_rate);
+    CHECK(out.status.indexing_rate.value() == 1.0f);
+    CHECK(out.status.images_sent == images);
+    CHECK(!out.status.cancelled);
+
+    // Done once everything is written, with the latency from the last image to then.
+    const auto live = service.GetRugnuxStatus();
+    CHECK(live.state == RugnuxLiveState::Done);
+    REQUIRE(live.last_image_to_written_s);
+    REQUIRE(live.last_image_s);
+    CHECK(live.elapsed_s >= live.last_image_s.value() + live.last_image_to_written_s.value() - 1e-6);
+
+    REQUIRE(!service.GetProgress().has_value());
+    REQUIRE_NOTHROW(writer_future.get());
+}
+
+TEST_CASE("JFJochReceiverRugnux_Stills_ZMQWriterDown", "[JFJochReceiverRugnux]") {
+    // With no writer listening the collection ends with the error, and the next one runs.
+    Logger logger(Catch::getResultCapture().getCurrentTestName());
+    RegisterHDF5Filter();
+
+    const int64_t images = 5;
+    DiffractionExperiment x = StillsExperiment(images);
+    x.FilePrefix("rugnux_stills_zmq_writer_down").SetFileWriterFormat(FileWriterFormat::NXmxVDS)
+            .OverwriteExistingFiles(true);
+    PixelMask mask(x);
+
+    ZMQStream2Pusher pusher({"ipc://*"});
+    pusher.WriterNotificationSocket("ipc://*");
+
+    AcquisitionDeviceGroup group;
+    JFJochReceiverService service(group, logger, pusher);
+    service.NumThreads(4);
+    service.Indexing(x.GetIndexingSettings());
+    service.RugnuxLive(true, nullptr, 4, 60);
+
+    Stream stream(x, images, false, 4);
+    stream.Frames(0, images);
+    stream.End();
+    service.Start(x, mask, nullptr, stream.puller);
+    REQUIRE_THROWS_AS(service.Stop(), JFJochException);
+    REQUIRE(!service.GetProgress().has_value());
+
+    ZMQImagePuller puller(pusher.GetAddress()[0]);
+    StreamWriter writer(logger, puller);
+    auto writer_future = std::async(std::launch::async, &StreamWriter::Run, &writer);
+
+    Stream next(x, images, false, 4);
+    next.Frames(0, images);
+    next.End();
+    service.Start(x, mask, nullptr, next.puller);
+    const auto out = service.Stop();
+    CHECK(out.status.images_sent == images);
+    CHECK(!out.status.cancelled);
+    REQUIRE_NOTHROW(writer_future.get());
+}
+
 TEST_CASE("JFJochReceiverRugnux_Rotation", "[JFJochReceiverRugnux]") {
     // A scan stays with rugnux's own run: the start message goes out at once, and there are no stills results.
     Logger logger(Catch::getResultCapture().getCurrentTestName());