105 lines
3.7 KiB
C++
105 lines
3.7 KiB
C++
// Copyright (2019-2023) Paul Scherrer Institute
|
|
|
|
#include "../common/JFJochException.h"
|
|
#include "StreamWriter.h"
|
|
#include "HDF5Writer.h"
|
|
#include "HDF5NXmx.h"
|
|
#include "MakeDirectory.h"
|
|
|
|
StreamWriter::StreamWriter(ZMQContext &context, Logger &in_logger, const std::string &zmq_addr, const std::string& repub_address)
|
|
: image_puller(context, repub_address), logger(in_logger) {
|
|
image_puller.Connect(zmq_addr);
|
|
logger.Info("Connected via ZMQ to {}", zmq_addr);
|
|
}
|
|
|
|
void StreamWriter::CollectImages(std::vector<HDF5DataFileStatistics> &v) {
|
|
bool run = true;
|
|
while (run && (image_puller.GetFrameType() != CBORStream2Deserializer::Type::START))
|
|
run = image_puller.WaitForImage();
|
|
|
|
if (!run)
|
|
return;
|
|
|
|
StartMessage start_message = image_puller.GetStartMessage();
|
|
logger.Info("Starting writing for dataset {} of {} images", start_message.file_prefix, start_message.number_of_images);
|
|
state = StreamWriterState::Started;
|
|
processed_images = 0;
|
|
processed_image_size = 0;
|
|
file_prefix = start_message.file_prefix;
|
|
|
|
CheckPath(start_message.file_prefix);
|
|
MakeDirectory(start_message.file_prefix);
|
|
HDF5Writer writer(start_message);
|
|
|
|
bool first_image = true;
|
|
run = image_puller.WaitForImage();
|
|
while (run && (image_puller.GetFrameType() == CBORStream2Deserializer::Type::IMAGE)) {
|
|
if (first_image) {
|
|
state = StreamWriterState::Receiving;
|
|
start_time = std::chrono::system_clock::now();
|
|
first_image = false;
|
|
}
|
|
auto image_array = image_puller.GetDataMessage();
|
|
writer.Write(image_array);
|
|
processed_images++;
|
|
processed_image_size += image_array.image.size;
|
|
run = image_puller.WaitForImage();
|
|
}
|
|
|
|
if (image_puller.GetFrameType() == CBORStream2Deserializer::Type::END) {
|
|
EndMessage end_message = image_puller.GetEndMessage();
|
|
end_time = std::chrono::system_clock::now();
|
|
if (end_message.write_master_file)
|
|
HDF5Metadata::NXmx(start_message, end_message);
|
|
state = StreamWriterState::Idle;
|
|
}
|
|
|
|
writer.GetStatistics(v);
|
|
}
|
|
|
|
void StreamWriter::Cancel() {
|
|
logger.Info("Cancel requested");
|
|
image_puller.Abort();
|
|
}
|
|
|
|
StreamWriterOutput StreamWriter::Run() {
|
|
StreamWriterOutput ret;
|
|
try {
|
|
CollectImages(ret.data_file_stats);
|
|
} catch (JFJochException &e) {
|
|
// Error during collecting images will skip to end data collection
|
|
// End data collection will consume all images till the end
|
|
logger.ErrorException(e);
|
|
}
|
|
|
|
ret.image_puller_stats = GetStatistics();
|
|
logger.Info("Write task done. Images = {} Throughput = {:.0f} MB/s Frame rate = {:.0f} Hz",
|
|
ret.image_puller_stats.processed_images, ret.image_puller_stats.performance_MBs, ret.image_puller_stats.performance_Hz);
|
|
return ret;
|
|
}
|
|
|
|
StreamWriterStatistics StreamWriter::GetStatistics() const {
|
|
float perf_MBs = 0.0f, perf_Hz = 0.0f;
|
|
|
|
|
|
if ((state != StreamWriterState::Started) && (processed_images > 0)) {
|
|
int64_t time_us;
|
|
if (state == StreamWriterState::Idle)
|
|
time_us = std::chrono::duration_cast<std::chrono::microseconds>(end_time - start_time).count();
|
|
else
|
|
time_us = std::chrono::duration_cast<std::chrono::microseconds>(std::chrono::system_clock::now() - start_time).count();
|
|
|
|
// MByte/s ==> Byte/us
|
|
perf_MBs = static_cast<float>(processed_image_size) / static_cast<float>(time_us);
|
|
perf_Hz = static_cast<float>(processed_images) * 1e6f / static_cast<float>(time_us);
|
|
}
|
|
|
|
return {
|
|
.processed_images = processed_images,
|
|
.performance_MBs = perf_MBs,
|
|
.performance_Hz = perf_Hz,
|
|
.file_prefix = file_prefix,
|
|
.state = state
|
|
};
|
|
}
|