From 59a0afaf8e7620323b62764e4a775c1cdd0ab02c Mon Sep 17 00:00:00 2001 From: Filip Leonarski Date: Sun, 8 Nov 2026 09:53:49 +0100 Subject: [PATCH] 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 Claude-Session: https://claude.ai/code/session_01HVeDQsU5a8AJUdJPgWevAK --- tests/JFJochReceiverRugnuxTest.cpp | 108 ++++++++++++++++++++++++++++- 1 file changed, 106 insertions(+), 2 deletions(-) diff --git a/tests/JFJochReceiverRugnuxTest.cpp b/tests/JFJochReceiverRugnuxTest.cpp index 03655b798..08b67257d 100644 --- a/tests/JFJochReceiverRugnuxTest.cpp +++ b/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 puller = std::make_shared(); - 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 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 start = {static_cast(i % file_space.GetDimensions()[0]), 0, 0}; + const int64_t frame = (same_frame >= 0) ? same_frame : i; + const std::vector start = {static_cast(frame % file_space.GetDimensions()[0]), 0, 0}; const std::vector 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{"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());