CollectorClass.h

Go to the documentation of this file.

// Copyright (C) 2026 European Spallation Source, ERIC. See LICENSE file
//===----------------------------------------------------------------------===//
//===----------------------------------------------------------------------===//
#pragma once
 
#include <filesystem>
#include <cstdlib>
#include <sstream>
#include <string>
#include <iomanip>
#include <utility>
#include <optional>
#include <random>
#include <map>
#include <set>
 
#include "Structs.h"
#include "Readout.h"
#include "enums.h"
#include "hdf_interface.h"
#include "version.hpp"
#include "efu_time.h"
#include "TypeDescriptionParser.h"
 
 
#ifdef WIN32
// Export symbols if compile flags "READOUT_SHARED" and "READOUT_EXPORT" are set on Windows.
    #ifdef READOUT_SHARED
        #ifdef READOUT_EXPORT
            #define RL_API __declspec(dllexport)
        #else
            #define RL_API __declspec(dllimport)
        #endif
    #else
        // Disable definition if linking statically.
        #define RL_API
    #endif
#else
// Disable definition for non-Win32 systems.
#define RL_API
#endif
 
RL_API int validate_collector_file(const std::string & filename);
RL_API int validate_collector_file_impl(const HighFive::File & file, const std::string & filename);
 
RL_API bool warn_about_another_build(const HighFive::File & file, const std::string & filename);
 
RL_API void ensure_file_attributes(HighFive::File & file);
RL_API void ensure_collector_group_attributes(HighFive::Group & group);
RL_API void ensure_parameter_group_attributes(HighFive::Group & group);
 
RL_API std::string validate_file_attributes(const HighFive::File & file);
RL_API std::string validate_collector_group(const HighFive::Group & group);
RL_API std::string validate_collector_root(const HighFive::Group & group) ;
 
RL_API bool validate_collector_files(
  const std::vector<std::string> & in_filenames,
  std::set<std::string> & datasets,
  std::map<std::string, size_t> & sizes,
  std::set<std::string> & valid_files,
  std::string & collector_name,
  std::set<std::string> & parameters
  );
 
RL_API bool append_collector_files(const std::string & out_filename, const std::vector<std::string> & in_filenames, bool reset_datasets);
 
RL_API bool concatenate_collector_files(const std::string & out_filename, const std::vector<std::string> & in_filenames);
 
RL_API bool merge_collector_files(const std::string & out_filename, const std::vector<std::string> & in_filenames, bool remove_after_merge);
 
RL_API bool combine_collector_files(const std::string & out_filename, const std::vector<std::string> & in_filenames);
 
RL_API void merge_collector_datasets(const std::string & out_filename, const std::vector<std::string> & in_filenames, const std::string & which_dataset, bool remove_after_merge);
 
RL_API void copy_collector_parameters(const std::string & out_filename, const std::vector<std::string> & in_filenames);
 
// A singleton object to hold the current runtime's output file for the collector
class CollectorSink {
protected:
  CollectorSink() = default;
 
  std::set<std::string> users_;
  bool appending_{false};
  size_t points_before_{0};
  std::optional<std::string> filename_{std::nullopt};
  std::optional<HighFive::File> file_{std::nullopt};
  std::optional<HighFive::Group> collector_, parameters_{std::nullopt};
 
public:
  static const std::string & type_attribute() {
    static const std::string name{"type"};
    return name;
  }
  static const std::string & collector_group_type() {
    static const std::string name{"Readouts"};
    return name;
  }
  static const std::string & readout_dataset_name() {
    static const std::string name{"readouts"};
    return name;
  }
  static const std::string & cue_dataset_name() {
    static const std::string name{"cues"};
    return name;
  }
  static const std::string & weight_dataset_name() {
    static const std::string name{"weights"};
    return name;
  }
  static const std::string & normalization_dataset_name() {
    static const std::string name{"normalizations"};
    return name;
  }
 
  static const std::string & detector_attribute_name() {
    static const std::string name{"detector"};
    return name;
  }
  static const std::string & efu_address_attribute_name() {
    static const std::string name{"efu_address"};
    return name;
  }
  static const std::string & efu_port_attribute_name() {
    static const std::string name{"efu_port"};
    return name;
  }
 
  static const std::string & parameter_group_type() {
    static const std::string name{"Parameters"};
    return name;
  }
  static const std::string & parameter_group_name() {
    static const std::string name{"parameters"};
    return name;
  }
  static const std::string & parameter_unit_attribute_name() {
    static const std::string name{"unit"};
    return name;
  }
  static const std::string & parameter_description_attribute_name() {
    static const std::string name{"description"};
    return name;
  }
  static const std::string & program_attribute_name() {
    static const std::string name{"program"};
    return name;
  }
  static const std::string & version_attribute_name() {
    static const std::string name{"version"};
    return name;
  }
  static const std::string & revision_attribute_name() {
    static const std::string name{"revision"};
    return name;
  }
  static const std::string & program_attribute_value() {
    static const std::string name{"libreadout"};
    return name;
  }
  static const std::string & version_attribute_value() {
    static const std::string name{reinterpret_cast<const char *>(libreadout::version::version_number)};
    return name;
  }
  static const std::string & revision_attribute_value() {
    static const std::string name{reinterpret_cast<const char *>(libreadout::version::git_revision)};
    return name;
  }
 
  CollectorSink(CollectorSink &other) = delete;
  void operator=(const CollectorSink &) = delete;
  // Defined in CollectorClass.cpp and exported: the singleton must live in
  // libreadout so that every module of a process (the library itself, any C++
  // consumer executing these header-inline methods) shares ONE sink. A
  // header-inline definition would give each Windows module its own copy.
  RL_API static CollectorSink * instance();
  static void destroy() {
    instance()->teardown();
  }
 
  bool is_setup() const { return filename_.has_value(); }
  [[nodiscard]] std::string current_filename() const { return filename_.value_or(""); }
  [[nodiscard]] size_t user_count() const { return users_.size(); }
 
  static bool allow_existing_file() {
    const char * const flag = std::getenv("READOUT_ALLOW_EXISTING_FILE");
    return flag != nullptr && flag[0] != '\0' && flag[0] != '0';
  }
 
  void setup(const std::string& filename) {
    const bool existing = std::filesystem::exists(filename);
    if (existing && !allow_existing_file()) {
      std::stringstream message;
      message << "Collector file '" << filename << "' already exists.\n"
              << "  A second run into one file merges its rays into the first run's scan\n"
              << "  point, leaving one set of parameter values standing for all of them.\n"
              << "  Write to a new output directory (--dir), or merge files deliberately\n"
              << "  with readout-combine. Set READOUT_ALLOW_EXISTING_FILE=1 to add to it\n"
              << "  anyway.";
      throw std::runtime_error(message.str());
    }
    filename_ = filename;
    try {
      // If the file doesn't exist, create it, otherwise open it for read/write so we can add to it
      file_ = HighFive::File(filename, HighFive::File::OpenOrCreate);
    } catch (HighFive::Exception & ex) {
      std::cout << "Error opening file " << filename << " for writing:\n" << ex.what();
      throw;
    }
    collector_ = file_->getGroup("/");
    ensure_file_attributes(*file_);
    // Adding to a file that already holds points means adding a *point*, not more rays
    // to the last one: this run had its own parameter values and its own ray count, and
    // folding them into a previous run's point loses both.
    appending_ = existing;
    points_before_ = appending_ ? count_points() : 0;
  }
 
  [[nodiscard]] size_t count_points() const {
    if (!collector_.has_value()) {
      return 0;
    }
    for (const auto & name : collector_->listObjectNames()) {
      if (name == parameter_group_name()) {
        continue;
      }
      if (collector_->getObjectType(name) != HighFive::ObjectType::Group) {
        continue;
      }
      if (auto group = collector_->getGroup(name); group.exist(cue_dataset_name())) {
        return group.getDataSet(cue_dataset_name()).getDimensions().back();
      }
    }
    return 0;
  }
 
  template<class T>
  static void extend_points(HighFive::DataSet ds, const size_t length, const T value) {
    const auto current = ds.getDimensions().back();
    if (current >= length) {
      return;
    }
    ds.resize({length});
    for (size_t i = current; i < length; ++i) {
      ds.select({i}, {1}).write(value);
    }
  }
 
  void teardown() {
    if (users_.empty()) {
      if (file_.has_value()) {
        file_->flush();
        file_ = std::nullopt;
      }
      appending_ = false;
      points_before_ = 0;
      filename_ = std::nullopt;
      collector_ = std::nullopt;
      parameters_ = std::nullopt;
    }
  }
 
 
  std::optional<HighFive::Group> getCollector(
    const std::string& name,
    const DetectorType& detector,
    const ReadoutType& readout
    ) {
    return getCollector(name, hdf_compound_type(readout), detector);
  }
 
  std::optional<HighFive::Group> getCollector(
    const std::string& name,
    const HighFive::DataType& datatype,
    const std::optional<DetectorType>& detector = std::nullopt,
    const std::optional<std::string>& description = std::nullopt
    ) {
    using namespace HighFive;
    if (!file_.has_value()) {
      std::cerr << "File not initialized, cannot get dataset!" << std::endl;
      return std::nullopt;
    }
    if (!collector_.has_value()) {
      collector_ = file_->getGroup("/");
    }
    if (!collector_->exist(name)) {
      auto group = collector_->createGroup(name);
      group.createAttribute<std::string>(type_attribute(), collector_group_type());
      if (detector.has_value()) {
        group.createAttribute<std::string>(detector_attribute_name(), detectorType_name(detector.value()));
      }
 
      // but should chunk file operations to avoid too much disk IO?
      DataSetCreateProps props;
      props.add(Chunking(std::vector<hsize_t>{100}));
 
      // create the readouts dataset, set to an empty vector []
      auto ds = group.createDataSet(readout_dataset_name(), DataSpace({0}, {DataSpace::UNLIMITED}), datatype, props);
      if (description.has_value()) {
        ds.createAttribute<std::string>(parameter_description_attribute_name(), description.value());
      }
      // One element per point, the last of which is the point being collected now.
      // A group first seen partway through a file's life is padded with empty points,
      // so every collector group carries the same number as the file does.
      const auto points = points_before_ + 1;
      const auto ws = group.createDataSet(weight_dataset_name(), DataSpace({points}, {DataSpace::UNLIMITED}), AtomicType<double>(), props);
      for (size_t i = 0; i < points; ++i) ws.select({i},{1}).write(0.);
      const auto cs = group.createDataSet(cue_dataset_name(), DataSpace({points}, {DataSpace::UNLIMITED}), AtomicType<uint32_t>(), props);
      for (size_t i = 0; i < points; ++i) cs.select({i},{1}).write(0);
      const auto ns = group.createDataSet(normalization_dataset_name(), DataSpace({points}, {DataSpace::UNLIMITED}), AtomicType<uint64_t>(), props);
      for (size_t i = 0; i < points; ++i) ns.select({i},{1}).write(0);
 
    } else if (appending_) {
      // The group is already here from an earlier run, so open a new point on it: a
      // fresh weight and normalization to accumulate into, and a cue seeded with the
      // records already stored, which is where this point would end if it stored none.
      auto group = collector_->getGroup(name);
      const auto points = points_before_ + 1;
      const auto cds = group.getDataSet(cue_dataset_name());
      const auto stored = cds.getDimensions().back()
          ? cds.select({cds.getDimensions().back() - 1}, {1}).read<uint32_t>() : 0u;
      extend_points(group.getDataSet(weight_dataset_name()), points, 0.);
      extend_points(cds, points, stored);
      extend_points(group.getDataSet(normalization_dataset_name()), points,
                    static_cast<uint64_t>(0));
    }
    // Outside the branch above: a collector whose group is already in the file is still
    // a user of it. Registering only new groups meant that re-opening a file left the
    // sink with no users at all -- so the first-user gate never fired, ~Collector's
    // removeCollector found nothing to remove, and teardown could close the file while
    // another collector was still writing.
    users_.insert(name);
    // TODO verify that the group has the right components?
    return collector_->getGroup(name);
  }
 
  [[nodiscard]] auto removeCollector(const std::string& name) {
    return users_.erase(name);
  }
 
  template<class T>
  void addParameter(
    const std::string& name,
    T value,
    const std::optional<std::string> & unit = std::nullopt,
    const std::optional<std::string> & description = std::nullopt
    ) {
    using namespace HighFive;
    if (!file_.has_value()) {
      std::cerr << "File not initialized, cannot add parameter!" << std::endl;
      return;
    }
    if (!parameters_.has_value()) {
      // Opened rather than created when it is already there, as getCollector does for
      // collector groups: HighFive throws on createGroup for a name that exists, and
      // the per-parameter datasets below already handle being written twice.
      const auto & group_name = parameter_group_name();
      parameters_ = collector_->exist(group_name) ? collector_->getGroup(group_name)
                                                  : collector_->createGroup(group_name);
      ensure_parameter_group_attributes(parameters_.value());
    }
    if (!parameters_->exist(name)) {
      DataSetCreateProps props;
      props.add(Chunking(std::vector<hsize_t>{100}));
      // One value per point. A parameter first recorded partway through a file's life
      // is padded, so its length still matches the number of points.
      const auto points = points_before_ + 1;
      auto ds = parameters_->createDataSet(name, DataSpace({points}, {DataSpace::UNLIMITED}), create_datatype<T>(), props);
      ds.select({points - 1},{1}).write(value);
      if (unit.has_value()) {
        ds.createAttribute(parameter_unit_attribute_name(), unit.value());
      }
      if (description.has_value()) {
        ds.createAttribute(parameter_description_attribute_name(), description.value());
      }
    } else {
      // verify that the existing dataset matches the new value's type
      auto existing_ds = parameters_->getDataSet(name);
      if (existing_ds.getDataType() != create_datatype<T>()) {
        std::cerr << "Parameter " << name << " already exists with a different type!" << std::endl;
      }
      if (const auto n = parameter_unit_attribute_name();
        unit.has_value()
        && (!existing_ds.hasAttribute(n) || existing_ds.getAttribute(n).read<std::string>() != unit.value())
        ) {
        std::cerr << "Parameter " << name << " already exists with a different unit!" << std::endl;
      }
      if (const auto n = parameter_description_attribute_name();
        description.has_value()
        && (!existing_ds.hasAttribute(n) || existing_ds.getAttribute(n).read<std::string>() != description.value())
        ) {
        std::cerr << "Parameter " << name << " already exists with a different description!" << std::endl;
      }
      if (appending_) {
        // This run is a new point, so the value belongs beside the earlier ones rather
        // than on top of them -- the whole reason a parameter dataset is per-point.
        extend_points(existing_ds, points_before_ + 1, value);
        existing_ds.select({points_before_},{1}).write(value);
      } else if (existing_ds.select({existing_ds.getDimensions().back()-1},{1}).read<T>() != value) {
        // Written to the current point rather than over the whole dataset, which would
        // overwrite every earlier point's value with this one.
        std::cerr << "Parameter " << name << " already exists with a different value! Overwriting." << std::endl;
        existing_ds.select({existing_ds.getDimensions().back()-1},{1}).write(value);
      }
    }
  }
};
 
 
class Collector {
  std::string name_;
  std::optional<HighFive::Group> group_;
  std::optional<HighFive::DataSet> dataset_;
  double weight_{0};
  uint64_t normalization_{0};
  std::optional<DetectorType> detector_{DetectorType::Reserved};
  std::optional<ReadoutType> readout_{ReadoutType::CAEN};
  std::optional<HighFive::DataType> datatype_{std::nullopt};
  size_t record_size_{0};
 
public:
 
  explicit Collector(
    const std::string &filename,
    std::string name,
    const int Type=0x34,
    const uint64_t normalization=0
  ): name_{std::move(name)}, normalization_{normalization}, detector_{detectorType_from_int(Type)}, readout_{readoutType_from_detectorType(detector_.value())} {
    datatype_ = hdf_compound_type(readout_.value());
    record_size_ = datatype_->getSize();
    const auto sink = CollectorSink::instance();
    if (!sink->is_setup()) {
      sink->setup(filename);
    }
    group_ = sink->getCollector(name_, detector_.value(), readout_.value());
    dataset_ = group_->getDataSet(CollectorSink::readout_dataset_name());
  }
 
  explicit Collector(
    const std::string &filename,
    std::string name,
    const std::string &type_description,
    const uint64_t normalization=0,
    const int Type=0
  ): name_{std::move(name)}, normalization_{normalization},
     detector_{Type > 0 ? std::optional(detectorType_from_int(Type)) : std::nullopt},
     readout_{std::nullopt} {
    const auto schema = parse_type_description(type_description);
    datatype_ = build_hdf5_compound_type(schema);
    record_size_ = schema.total_size;
    const auto sink = CollectorSink::instance();
    if (!sink->is_setup()) {
      sink->setup(filename);
    }
    group_ = sink->getCollector(name_, datatype_.value(), detector_, type_description);
    dataset_ = group_->getDataSet(CollectorSink::readout_dataset_name());
  }
 
  [[nodiscard]] size_t record_size() const { return record_size_; }
 
  void addRecord(const double weight, const void * data) {
    if (!dataset_.has_value() || !datatype_.has_value()) {
      std::cerr << "Dataset not initialized, cannot save record!" << std::endl;
      return;
    }
    weight_ += weight;
    auto & d = dataset_.value();
    const auto pos = d.getDimensions().back();
    d.resize({pos + 1});
    d.select({pos}, {1}).write_raw(static_cast<const uint8_t *>(data), datatype_.value());
  }
 
  // Add a readout with time and weight information to the writer's storage
  void addReadout(const uint8_t Ring, const uint8_t FEN, const double tof, const double weight, const void * data) {
    if (!readout_.has_value()) {
      throw std::runtime_error("addReadout requires a typed Collector; use addRecord with a description-based Collector");
    }
    weight_ += weight;
    switch (readout_.value()) {
      case ReadoutType::CAEN: {
        const CAEN_event event{Ring, FEN, tof, weight, static_cast<const CAEN_readout_t*>(data)};
        return saveReadout(event);
      }
      case ReadoutType::CDT: {
        const CDT_event event{Ring, FEN, tof, weight, static_cast<const CDT_readout_t*>(data)};
        return saveReadout(event);
      }
      case ReadoutType::VMM3: {
        const VMM3_event event{Ring, FEN, tof, weight, static_cast<const VMM3_readout_t*>(data)};
        return saveReadout(event);
      }
      case ReadoutType::BM0: {
        const BM0_event event{Ring, FEN, tof, weight, static_cast<const BM0_readout_t*>(data)};
        return saveReadout(event);
      }
      case ReadoutType::BM2: {
        const BM2_event event{Ring, FEN, tof, weight, static_cast<const BM2_readout_t*>(data)};
        return saveReadout(event);
      }
      case ReadoutType::BMI: {
        const BMI_event event{Ring, FEN, tof, weight, static_cast<const BMI_readout_t*>(data)};
        return saveReadout(event);
      }
      default:
        throw std::runtime_error("Saving this readout type is not implemented yet!");
    }
  }
 
  template<class ... Args> void addParameter(Args && ... args) {
    CollectorSink::instance()->addParameter(std::forward<Args>(args)...);
  }
 
  void setEFU(const std::string & address, int port) {
    if (!group_.has_value() || address.empty() || port <= 0) return;
    if (!group_->hasAttribute(CollectorSink::efu_address_attribute_name())) {
      group_->createAttribute<std::string>(CollectorSink::efu_address_attribute_name(), address);
    }
    if (!group_->hasAttribute(CollectorSink::efu_port_attribute_name())) {
      group_->createAttribute<int>(CollectorSink::efu_port_attribute_name(), port);
    }
  }
 
  ~Collector() {
    // Add the accumulated weight to the existing dataset value -- which was initialized to [0.]
    const auto wds = group_->getDataSet(CollectorSink::weight_dataset_name());
    auto selection = wds.select({wds.getDimensions().back() - 1}, {1});
    selection.write(weight_ + selection.read<double>());
    // Record the number of readouts written -- where this points' readouts *end* (was initialized to [0u])
    if (dataset_.has_value()) {
      const auto count = dataset_->getDimensions().back();
      const auto cds = group_->getDataSet(CollectorSink::cue_dataset_name());
      auto cs = cds.select({cds.getDimensions().back()-1}, {1});
      cs.write<uint32_t>(static_cast<uint32_t>(count));
    }
    // Add the normalization to the existing dataset value -- which was initialized to [0]
    const auto nds = group_->getDataSet(CollectorSink::normalization_dataset_name());
    auto ns = nds.select({nds.getDimensions().back() - 1}, {1});
    ns.write(normalization_ + ns.read<uint64_t>());
 
    const auto sink = CollectorSink::instance();
    if (const auto count = sink->removeCollector(name_); count != 1) {
      std::cerr << "Warning: removed collector " << name_ << " from sink " << count << " times, but expected to remove exactly one." << std::endl;
    }
    // try closing the sink file (internally it checks if any collectors are still using it):
    sink->teardown();
  }
 
private:
  template<class T>
  void saveReadout(T data) {
    //TODO Consider buffering this internally and only writing when the buffer is full (or closing the file)
    if (!dataset_.has_value() || !datatype_.has_value()) {
      std::cerr << "Dataset not initialized, cannot save readout!" << std::endl;
      return;
    }
    auto & d = dataset_.value();
    // the dataset should be 1-D ... hopefully that's true
    auto pos = d.getDimensions().back();
    auto size = pos + 1;
    d.resize({size});
    d.select({pos}, {1}).write_raw(reinterpret_cast<const uint8_t *>(&data), datatype_.value());
  }
};
 
 
std::string filename_for_collector(const std::string & basepath, const std::string & basename, const std::string & extension = "h5");
std::string filename_for_collector_node(const std::string & basepath, const std::string & basename, int node, int nodes);