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());