Jungfraujoch/tests/StreamWriterTest.cpp

// SPDX-FileCopyrightText: 2024 Filip Leonarski, Paul Scherrer Institute <filip.leonarski@psi.ch>
// SPDX-License-Identifier: GPL-3.0-only

#include <catch2/catch_all.hpp>
#include <filesystem>

#include "../writer/StreamWriter.h"
#include "../image_pusher/ZMQStream2Pusher.h"
#include "../receiver/JFJochReceiverService.h"
#include "../image_pusher/ZMQWriterNotificationPuller.h"

TEST_CASE("StreamWriterTest_ZMQ","[StreamWriter]") {
    RegisterHDF5Filter();

    Logger logger("StreamWriterTest_ZMQ");

    DiffractionExperiment x(DetectorGeometry(2));
    x.FilePrefix("subdir/StreamWriterTest").NumTriggers(1).ImagesPerTrigger(5)
            .UseInternalPacketGenerator(true).Mode(DetectorMode::Raw).PedestalG0Frames(0);

    PixelMask pixel_mask(x);

    JFModuleGainCalibration gain;
    AcquisitionDeviceGroup aq_devices;
    for (int i = 0; i < x.GetDataStreamsNum(); i++)
        aq_devices.AddHLSDevice(64);

    ZMQStream2Pusher pusher ({"ipc://*"});
    JFJochReceiverService fpga_receiver_service(aq_devices, logger, pusher);

    std::unique_ptr<StreamWriter> writer;
    REQUIRE(x.GetImageNum() == 5);
    auto pusher_addr = pusher.GetAddress();
    REQUIRE(pusher_addr.size() == 1);
    REQUIRE_NOTHROW(writer = std::make_unique<StreamWriter>(logger, pusher_addr[0]));
    CHECK (writer->GetStatistics().state == StreamWriterState::Idle);
    REQUIRE_NOTHROW(fpga_receiver_service.Start(x, pixel_mask, nullptr));

    REQUIRE_NOTHROW(writer->Run());

    REQUIRE_NOTHROW(fpga_receiver_service.Stop());
    REQUIRE(fpga_receiver_service.GetStatus()->images_collected == 5);
    REQUIRE(fpga_receiver_service.GetStatus()->images_sent == 5);
    CHECK(writer->GetStatistics().state == StreamWriterState::Idle);
    CHECK(writer->GetStatistics().processed_images == 5);
    CHECK(writer->GetStatistics().file_prefix == x.GetFilePrefix());

    // HDF5 file can be opened
    std::unique_ptr<HDF5ReadOnlyFile> file;
    REQUIRE_NOTHROW(file = std::make_unique<HDF5ReadOnlyFile>("subdir/StreamWriterTest_data_000001.h5"));
    std::unique_ptr<HDF5DataSet> dataset;
    REQUIRE_NOTHROW(dataset = std::make_unique<HDF5DataSet>(*file, "/entry/data/data"));
    std::unique_ptr<HDF5DataSpace> dataspace;
    REQUIRE_NOTHROW(dataspace = std::make_unique<HDF5DataSpace>(*dataset));

    REQUIRE(dataspace->GetNumOfDimensions() == 3);
    REQUIRE(dataspace->GetDimensions()[0] == 5);
    REQUIRE(dataspace->GetDimensions()[1] == RAW_MODULE_COLS);
    REQUIRE(dataspace->GetDimensions()[2] == 2*RAW_MODULE_LINES);

    REQUIRE(std::filesystem::remove("subdir/StreamWriterTest_master.h5"));
    REQUIRE(std::filesystem::remove("subdir/StreamWriterTest_data_000001.h5"));
    REQUIRE(std::filesystem::remove("subdir"));
}

TEST_CASE("StreamWriterTest_ZMQ_Update","[StreamWriter]") {
    RegisterHDF5Filter();

    Logger logger("StreamWriterTest_ZMQ_Update");

    DatasetSettings d;
    d.FilePrefix("subdir/StreamWriterTest2").NumTriggers(1).ImagesPerTrigger(5).RunName("run1").RunNumber(256);
    DiffractionExperiment x(DetectorGeometry(2));
    x.UseInternalPacketGenerator(true).Mode(DetectorMode::Raw).PedestalG0Frames(0).ImportDatasetSettings(d);
    PixelMask pixel_mask(x);

    JFModuleGainCalibration gain;
    AcquisitionDeviceGroup aq_devices;
    for (int i = 0; i < x.GetDataStreamsNum(); i++)
        aq_devices.AddHLSDevice(64);

    ZMQStream2Pusher pusher ({"ipc://*"});
    pusher.WriterNotificationSocket("ipc://*");
    JFJochReceiverService fpga_receiver_service(aq_devices, logger, pusher);

    std::unique_ptr<StreamWriter> writer;
    REQUIRE(x.GetImageNum() == 5);
    auto pusher_addr = pusher.GetAddress();
    REQUIRE(pusher_addr.size() == 1);
    REQUIRE_NOTHROW(writer = std::make_unique<StreamWriter>(logger, pusher_addr[0]));
    CHECK (writer->GetStatistics().state == StreamWriterState::Idle);
    REQUIRE_NOTHROW(fpga_receiver_service.Start(x, pixel_mask, nullptr));

    REQUIRE_NOTHROW(writer->Run());

    REQUIRE_NOTHROW(fpga_receiver_service.Stop());
    REQUIRE(fpga_receiver_service.GetStatus()->images_collected == 5);
    REQUIRE(fpga_receiver_service.GetStatus()->images_sent == 5);
    CHECK(writer->GetStatistics().state == StreamWriterState::Idle);
    CHECK(writer->GetStatistics().processed_images == 5);
    CHECK(writer->GetStatistics().file_prefix == x.GetFilePrefix());

    // HDF5 file can be opened
    std::unique_ptr<HDF5ReadOnlyFile> file;
    REQUIRE_NOTHROW(file = std::make_unique<HDF5ReadOnlyFile>("subdir/StreamWriterTest2_data_000001.h5"));
    std::unique_ptr<HDF5DataSet> dataset;
    REQUIRE_NOTHROW(dataset = std::make_unique<HDF5DataSet>(*file, "/entry/data/data"));
    std::unique_ptr<HDF5DataSpace> dataspace;
    REQUIRE_NOTHROW(dataspace = std::make_unique<HDF5DataSpace>(*dataset));

    REQUIRE(dataspace->GetNumOfDimensions() == 3);
    REQUIRE(dataspace->GetDimensions()[0] == 5);
    REQUIRE(dataspace->GetDimensions()[1] == RAW_MODULE_COLS);
    REQUIRE(dataspace->GetDimensions()[2] == 2*RAW_MODULE_LINES);

    REQUIRE(std::filesystem::remove("subdir/StreamWriterTest2_master.h5"));
    REQUIRE(std::filesystem::remove("subdir/StreamWriterTest2_data_000001.h5"));
    REQUIRE(std::filesystem::remove("subdir"));
}


TEST_CASE("StreamWriterTest_ZMQ_Update_NoNotification","[StreamWriter]") {
    // This tests simulates what happens if writer notification about writing end is missing
    // Expected end result: receiver ends with an exception

    RegisterHDF5Filter();

    Logger logger("StreamWriterTest_ZMQ_Update_NoNotification");

    DatasetSettings d;
    d.FilePrefix("subdir/StreamWriterTest3").NumTriggers(1).ImagesPerTrigger(5).RunName("run1").RunNumber(256);
    DiffractionExperiment x(DetectorGeometry(2));
    x.UseInternalPacketGenerator(true).Mode(DetectorMode::Raw).PedestalG0Frames(0).ImportDatasetSettings(d);
    PixelMask pixel_mask(x);

    JFModuleGainCalibration gain;
    AcquisitionDeviceGroup aq_devices;
    for (int i = 0; i < x.GetDataStreamsNum(); i++)
        aq_devices.AddHLSDevice(64);

    ZMQStream2Pusher pusher ({"ipc://*"});
    pusher.WriterNotificationSocket("ipc://*");
    JFJochReceiverService fpga_receiver_service(aq_devices, logger, pusher);

    std::unique_ptr<StreamWriter> writer;
    REQUIRE(x.GetImageNum() == 5);
    auto pusher_addr = pusher.GetAddress();
    REQUIRE(pusher_addr.size() == 1);

    REQUIRE_NOTHROW(writer = std::make_unique<StreamWriter>(logger, pusher_addr[0]));
    writer->DebugSkipWriteNotification(true);

    CHECK (writer->GetStatistics().state == StreamWriterState::Idle);
    REQUIRE_NOTHROW(fpga_receiver_service.Start(x, pixel_mask, nullptr));

    REQUIRE_NOTHROW(writer->Run());

    REQUIRE_THROWS(fpga_receiver_service.Stop());
    REQUIRE(fpga_receiver_service.GetStatus()->images_collected == 5);
    REQUIRE(fpga_receiver_service.GetStatus()->images_sent == 5);
    CHECK(writer->GetStatistics().state == StreamWriterState::Idle);
    CHECK(writer->GetStatistics().processed_images == 5);
    CHECK(writer->GetStatistics().file_prefix == x.GetFilePrefix());

    // HDF5 file can be opened
    std::unique_ptr<HDF5ReadOnlyFile> file;
    REQUIRE_NOTHROW(file = std::make_unique<HDF5ReadOnlyFile>("subdir/StreamWriterTest3_data_000001.h5"));
    std::unique_ptr<HDF5DataSet> dataset;
    REQUIRE_NOTHROW(dataset = std::make_unique<HDF5DataSet>(*file, "/entry/data/data"));
    std::unique_ptr<HDF5DataSpace> dataspace;
    REQUIRE_NOTHROW(dataspace = std::make_unique<HDF5DataSpace>(*dataset));

    REQUIRE(dataspace->GetNumOfDimensions() == 3);
    REQUIRE(dataspace->GetDimensions()[0] == 5);
    REQUIRE(dataspace->GetDimensions()[1] == RAW_MODULE_COLS);
    REQUIRE(dataspace->GetDimensions()[2] == 2*RAW_MODULE_LINES);

    REQUIRE(std::filesystem::remove("subdir/StreamWriterTest3_master.h5"));
    REQUIRE(std::filesystem::remove("subdir/StreamWriterTest3_data_000001.h5"));
    REQUIRE(std::filesystem::remove("subdir"));
}