MergeMDFiles.cpp

Go to the documentation of this file.

// Mantid Repository : https://github.com/mantidproject/mantid
//
// Copyright © 2018 ISIS Rutherford Appleton Laboratory UKRI,
//   NScD Oak Ridge National Laboratory, European Spallation Source,
//   Institut Laue - Langevin & CSNS, Institute of High Energy Physics, CAS
// SPDX - License - Identifier: GPL - 3.0 +
#include "MantidMDAlgorithms/MergeMDFiles.h"
#include "MantidAPI/FileProperty.h"
#include "MantidAPI/MultipleFileProperty.h"
#include "MantidDataObjects/BoxControllerNeXusIO.h"
#include "MantidDataObjects/MDBoxBase.h"
#include "MantidDataObjects/MDEventFactory.h"
#include "MantidKernel/CPUTimer.h"
#include "MantidKernel/Strings.h"
#include "MantidKernel/System.h"
#include "MantidKernel/VectorHelper.h"
 
#include <Poco/File.h>
#include <boost/scoped_ptr.hpp>
 
using namespace Mantid::Kernel;
using namespace Mantid::API;
using namespace Mantid::DataObjects;
 
namespace Mantid::MDAlgorithms {
 
// Register the algorithm into the AlgorithmFactory
DECLARE_ALGORITHM(MergeMDFiles)
 
//----------------------------------------------------------------------------------------------
MergeMDFiles::MergeMDFiles()
    : m_nDims(0), m_MDEventType(), m_fileBasedTargetWS(false), m_Filenames(), m_EventLoader(), m_OutIWS(),
      m_totalEvents(0), m_totalLoaded(0), m_fileMutex(), m_statsMutex() {}
 
//----------------------------------------------------------------------------------------------
MergeMDFiles::~MergeMDFiles() { clearEventLoaders(); }
 
//----------------------------------------------------------------------------------------------
 
//----------------------------------------------------------------------------------------------
void MergeMDFiles::init() {
  std::vector<std::string> exts(1, ".nxs");
  declareProperty(std::make_unique<MultipleFileProperty>("Filenames", exts),
                  "Select several MDEventWorkspace NXS files to merge "
                  "together. Files must have common box structure.");
 
  declareProperty(std::make_unique<FileProperty>("OutputFilename", "", FileProperty::OptionalSave, exts),
                  "Choose a file to which to save the output workspace. \n"
                  "Optional: if specified, the workspace created will be file-backed. \n"
                  "If not, it will be created in memory.");
 
  declareProperty("Parallel", false,
                  "Run the loading tasks in parallel.\n"
                  "This can be faster but might use more memory.");
 
  declareProperty(std::make_unique<WorkspaceProperty<IMDEventWorkspace>>("OutputWorkspace", "", Direction::Output),
                  "An output MDEventWorkspace.");
}
 
//----------------------------------------------------------------------------------------------
void MergeMDFiles::loadBoxData() {
  this->progress(0.05, "Loading File Info");
  // Get plain box structure and box tree
  std::vector<API::IMDNode *> &Boxes = m_BoxStruct.getBoxes();
  std::vector<uint64_t> &targetEventIndexes = m_BoxStruct.getEventIndex();
  // clear the averages for target event indexes;
  targetEventIndexes.assign(targetEventIndexes.size(), 0);
 
  // Total number of events in ALL files.
  m_totalEvents = 0;
 
  m_fileComponentsStructure.resize(m_Filenames.size());
  m_EventLoader.assign(m_Filenames.size(), nullptr);
 
  try {
    for (size_t i = 0; i < m_Filenames.size(); i++) {
      // load box structure and the experimental info from each target
      // workspace.
      m_fileComponentsStructure[i].loadBoxStructure(m_Filenames[i], m_nDims, m_MDEventType, true, true);
      // export just loaded experiment info to the target workspace
      m_fileComponentsStructure[i].exportExperiment(m_OutIWS);
 
      // Check for consistency
      if (i > 0) {
        if (m_fileComponentsStructure[i].getEventIndex().size() != targetEventIndexes.size())
          throw std::runtime_error("Inconsistent number of boxes found in file " + m_Filenames[i] +
                                   ". Cannot merge these files. Did you generate them all with "
                                   "exactly the same box structure?");
      }
 
      // calculate total number of events per target cell, which will be
      size_t nBoxes = Boxes.size();
      for (size_t j = 0; j < nBoxes; j++) {
        size_t ID = Boxes[j]->getID();
        targetEventIndexes[2 * ID + 1] += m_fileComponentsStructure[i].getEventIndex()[2 * ID + 1];
        m_totalEvents += m_fileComponentsStructure[i].getEventIndex()[2 * ID + 1];
      }
 
      // Open the event data, track the total number of events
      auto bc = std::shared_ptr<API::BoxController>(new API::BoxController(static_cast<size_t>(m_nDims)));
      bc->fromXMLString(m_fileComponentsStructure[i].getBCXMLdescr());
 
      m_EventLoader[i] = new BoxControllerNeXusIO(bc.get());
      m_EventLoader[i]->setDataType(sizeof(coord_t), m_MDEventType);
      m_EventLoader[i]->openFile(m_Filenames[i], "r");
    }
  } catch (...) {
    // Close all open files in case of error
    clearEventLoaders();
    throw;
  }
 
  const std::vector<int> &boxType = m_BoxStruct.getBoxType();
  // calculate event positions in the target file.
  uint64_t eventsStart = 0;
  for (auto mdBox : Boxes) {
    mdBox->clear();
    size_t ID = mdBox->getID();
 
    // avoid grid boxes;
    if (boxType[ID] == 2)
      continue;
 
    uint64_t nEvents = targetEventIndexes[2 * ID + 1];
    targetEventIndexes[ID * 2] = eventsStart;
    if (m_fileBasedTargetWS)
      mdBox->setFileBacked(eventsStart, nEvents, false);
 
    eventsStart += nEvents;
  }
 
  g_log.notice() << m_totalEvents << " events in " << m_Filenames.size() << " files.\n";
}
 
uint64_t MergeMDFiles::loadEventsFromSubBoxes(API::IMDNode *TargetBox) {
  TargetBox->clear();
 
  uint64_t nBoxEvents(0);
  std::vector<size_t> numFileEvents(m_EventLoader.size());
 
  for (size_t iw = 0; iw < this->m_EventLoader.size(); iw++) {
    size_t ID = TargetBox->getID();
    numFileEvents[iw] = static_cast<size_t>(m_fileComponentsStructure[iw].getEventIndex()[2 * ID + 1]);
    nBoxEvents += numFileEvents[iw];
  }
 
  // At this point memory required is known, so it is reserved all in one go
  TargetBox->reserveMemoryForLoad(nBoxEvents);
 
  for (size_t iw = 0; iw < this->m_EventLoader.size(); iw++) {
    size_t ID = TargetBox->getID();
    uint64_t fileLocation = m_fileComponentsStructure[iw].getEventIndex()[2 * ID + 0];
    if (numFileEvents[iw] == 0)
      continue;
    TargetBox->loadAndAddFrom(m_EventLoader[iw], fileLocation, numFileEvents[iw]);
  }
 
  return nBoxEvents;
}
 
//----------------------------------------------------------------------------------------------
void MergeMDFiles::doExecByCloning(const Mantid::API::IMDEventWorkspace_sptr &ws, const std::string &outputFile) {
  m_OutIWS = ws;
  m_MDEventType = ws->getEventTypeName();
 
  // Run the tasks in parallel? TODO: enable
  // bool Parallel = this->getProperty("Parallel");
 
  // Fix the box controller settings in the output workspace so that it splits
  // normally
  BoxController_sptr bc = ws->getBoxController();
  // set up internal variables characterizing the workspace.
  m_nDims = static_cast<int>(bc->getNDims());
 
  // Fix the max depth to something bigger.
  bc->setMaxDepth(20);
  bc->setSplitThreshold(5000);
  auto saver = std::shared_ptr<API::IBoxControllerIO>(new DataObjects::BoxControllerNeXusIO(bc.get()));
  saver->setDataType(sizeof(coord_t), m_MDEventType);
  if (m_fileBasedTargetWS) {
    bc->setFileBacked(saver, outputFile);
    // Complete the file-back-end creation.
    g_log.notice() << "Setting cache to 400 MB write.\n";
    bc->getFileIO()->setWriteBufferSize(400000000 / m_OutIWS->sizeofEvent());
  }
 
  /*   else
     {
         saver->openFile(outputFile,"w");
     }*/
  // Init box structure used for memory/file space calculations
  m_BoxStruct.initFlatStructure(ws, outputFile);
 
  // First, load all the box data and experiment info and calculate file
  // positions of the target workspace
  this->loadBoxData();
 
  size_t numBoxes = m_BoxStruct.getNBoxes();
  // Progress report based on events processed.
  m_progress = std::make_unique<Progress>(this, 0.1, 0.9, size_t(numBoxes));
  m_progress->setNotifyStep(0.1);
 
  // For tracking progress
  // uint64_t m_totalEventsInTasks = 0;
 
  // Prepare thread pool
  CPUTimer overallTime;
 
  auto ts = new ThreadSchedulerFIFO();
  ThreadPool tp(ts);
 
  Kernel::DiskBuffer *DiskBuf(nullptr);
  if (m_fileBasedTargetWS) {
    DiskBuf = bc->getFileIO();
  }
 
  this->m_totalLoaded = 0;
  const std::vector<API::IMDNode *> &boxes = m_BoxStruct.getBoxes();
 
  for (size_t ib = 0; ib < numBoxes; ib++) {
    auto box = boxes[ib];
    if (!box->isBox())
      continue;
    // load all contributed events into current box;
    this->loadEventsFromSubBoxes(box);
 
    if (DiskBuf) {
      if (box->getDataInMemorySize() > 0) { // data position has been already pre-calculated
        box->getISaveable()->save();
        box->clearDataFromMemory();
        // Kernel::ISaveable *Saver = box->getISaveable();
        // DiskBuf->toWrite(Saver);
      }
    }
    // else
    //{   size_t ID = box->getID();
    //    uint64_t filePosition = targetEventIndexes[2*ID];
    //    box->saveAt(saver.get(), filePosition);
    //}
    //
    // if (!Parallel)
    //{
    //  // Run the task serially only
    //  task->run();
    //  delete task;
    //}
    // else
    //{
    //  // Enqueue to run in parallel (at the joinAll() call below).
    //  ts->push(task);
    //}
 
    m_progress->reportIncrement(ib, "Loading and merging box data");
  }
  if (DiskBuf) {
    DiskBuf->flushCache();
    bc->getFileIO()->flushData();
  }
  // tp.joinAll();
  g_log.information() << overallTime << " to do all the adding.\n";
 
  // Close any open file handle
  clearEventLoaders();
 
  // Finish things up
  this->finalizeOutput(outputFile);
}
 
//----------------------------------------------------------------------------------------------
void MergeMDFiles::finalizeOutput(const std::string &outputFile) {
  CPUTimer overallTime;
 
  this->progress(0.90, "Refreshing Cache");
  m_OutIWS->refreshCache();
 
  g_log.information() << overallTime << " to run refreshCache().\n";
 
  if (!outputFile.empty()) {
    g_log.notice() << "Starting SaveMD to update the file back-end.\n";
    // create or open WS group and put there additional information about WS and
    // its dimensions
    bool old_data_there;
    // clang-format off
    boost::scoped_ptr< ::NeXus::File> file(MDBoxFlatTree::createOrOpenMDWSgroup(
        outputFile, m_nDims, m_MDEventType, false, old_data_there));
    // clang-format on
    this->progress(0.94, "Saving ws history and dimensions");
    MDBoxFlatTree::saveWSGenericInfo(file.get(), m_OutIWS);
    // Save each ExperimentInfo to a spot in the file
    this->progress(0.98, "Saving experiment infos");
    MDBoxFlatTree::saveExperimentInfos(file.get(), m_OutIWS);
 
    file->closeGroup();
    file->close();
    // -------------- Save Box Structure  -------------------------------------
    // OK, we've filled these big arrays of data representing flat box
    // structure. Save them.
    progress(0.91, "Writing Box Data");
    m_progress->resetNumSteps(8, 0.92, 1.00);
 
    // Save box structure;
    m_BoxStruct.saveBoxStructure(outputFile);
 
    g_log.information() << overallTime << " to run SaveMD structure\n";
  }
}
 
//----------------------------------------------------------------------------------------------
void MergeMDFiles::exec() {
  // clear disk buffer which can remain from previous runs
  // the existence/ usage of the buffer indicates if the algorithm works with
  // file based or memory based target workspaces;
  // pDiskBuffer = NULL;
  MultipleFileProperty *multiFileProp = dynamic_cast<MultipleFileProperty *>(getPointerToProperty("Filenames"));
  if (!multiFileProp) {
    throw std::logic_error("Filenames property must have MultipleFileProperty type.");
  }
  m_Filenames = VectorHelper::flattenVector(multiFileProp->operator()());
  if (m_Filenames.empty())
    throw std::invalid_argument("Must specify at least one filename.");
  std::string firstFile = m_Filenames[0];
 
  std::string outputFile = getProperty("OutputFilename");
  m_fileBasedTargetWS = false;
  if (!outputFile.empty()) {
    m_fileBasedTargetWS = true;
    if (Poco::File(outputFile).exists())
      throw std::invalid_argument(" File " + outputFile +
                                  " already exists. Can not use existing file "
                                  "as the target to MergeMD files.\n" +
                                  " Use it as one of source files if you want to add MD data to it");
  }
 
  // Start by loading the first file but just the box structure, no events, and
  // not file-backed
  // m_BoxStruct.loadBoxStructure(firstFile,
  auto loader = createChildAlgorithm("LoadMD", 0.0, 0.05, false);
  loader->setPropertyValue("Filename", firstFile);
  loader->setProperty("MetadataOnly", false);
  loader->setProperty("BoxStructureOnly", true);
  loader->setProperty("FileBackEnd", false);
  loader->executeAsChildAlg();
  IMDWorkspace_sptr result = (loader->getProperty("OutputWorkspace"));
 
  auto firstWS = std::dynamic_pointer_cast<API::IMDEventWorkspace>(result);
  if (!firstWS)
    throw std::runtime_error("Can not load MDEventWorkspace from initial file " + firstFile);
 
  // do the job
  this->doExecByCloning(firstWS, outputFile);
 
  m_OutIWS->setFileNeedsUpdating(false);
 
  setProperty("OutputWorkspace", m_OutIWS);
}
void MergeMDFiles::clearEventLoaders() {
  for (auto &loader : m_EventLoader) {
    delete loader;
    loader = nullptr;
  }
}
 
} // namespace Mantid::MDAlgorithms