Jungfraujoch/image_pusher/HDF5FilePusher.cpp

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

#include "HDF5FilePusher.h"
#include "../frame_serialize/CBORStream2Deserializer.h"


void HDF5FilePusher::StartDataCollection(StartMessage &message) {
    if (writer)
        throw JFJochException(JFJochExceptionCategory::WrongDAQState, "Image pusher is already writing images");
    writer = std::make_unique<FileWriter>(message);
    writer_future = std::async(std::launch::async, &HDF5FilePusher::WriterThread, this);
}

bool HDF5FilePusher::EndDataCollection(const EndMessage &message) {
    if (writer_future.valid()) {
        writer_queue.PutBlocking({.end = true});
        writer_future.get();
    } else
        throw JFJochException(JFJochExceptionCategory::WrongDAQState, "Image pusher wasn't writing files");

    if (!writer)
        throw JFJochException(JFJochExceptionCategory::WrongDAQState, "Image pusher not ready for writing images");
    writer->WriteHDF5(message);
    writer->Finalize();
    writer.reset();

    return true;
}

bool HDF5FilePusher::SendImage(const uint8_t *image_data, size_t image_size, int64_t image_number) {
    if (!writer)
        throw JFJochException(JFJochExceptionCategory::WrongDAQState, "Image pusher not ready for sending");

    auto deserialized = CBORStream2Deserialize(image_data, image_size);
    if (deserialized->data_message)
        writer->Write(*deserialized->data_message);
    else
        throw JFJochException(JFJochExceptionCategory::InputParameterInvalid,
                              "HDF5FilePusher::SendImage accepts only data image");
    return true;
}

void HDF5FilePusher::SendImage(ZeroCopyReturnValue &z) {
    writer_queue.PutBlocking(ImagePusherQueueElement{
        .image_data = (uint8_t *) z.GetImage(),
        .z = &z,
        .end = false
    });
}

bool HDF5FilePusher::SendCalibration(const CompressedImage &message) {
    if (!writer)
        throw JFJochException(JFJochExceptionCategory::WrongDAQState, "Image pusher not ready for sending");

    writer->WriteHDF5(message);
    return true;
}

void HDF5FilePusher::WriterThread() {
    ImagePusherQueueElement e = writer_queue.GetBlocking();
    while (!e.end) {
        SendImage(e.image_data, e.z->GetImageSize(), e.z->GetImageNumber());
        e.z->release();
        e = writer_queue.GetBlocking();
    }
}

std::string HDF5FilePusher::PrintSetup() const {
    std::filesystem::path currentPath = std::filesystem::current_path();
    return "HDF5FilePusher: Images are written directly to file in base directory " + currentPath.string();
}