Re #5051 Added functionality to DataProcessorAlg

Code/Mantid/Framework/API/inc/MantidAPI/DataProcessorAlgorithm.h

@@ -4,6 +4,7 @@
 #include "MantidKernel/System.h"
 #include "MantidAPI/Algorithm.h"
 #include "MantidAPI/ITableWorkspace.h"
+#include "MantidAPI/IEventWorkspace.h"
 #include "MantidKernel/PropertyManager.h"
 #include <vector>
 
@@ -50,14 +51,17 @@ protected:
   std::vector<std::string> splitInput(const std::string & input);
   void forwardProperties();
   boost::shared_ptr<Kernel::PropertyManager> getProcessProperties(const std::string &propertyManager);
+  /// MPI option. If false, we will use one job event if MPI is available
+  bool m_useMPI;
+  Workspace_sptr assemble(const std::string &partialWSName, const std::string &outputWSName);
+  void saveNexus(const std::string &outputWSName,
+        const std::string &outputFile);
 
 private:
   /// The name of the algorithm to invoke when loading data
   std::string m_loadAlg;
   /// The name of the algorithm to invoke when accumulating data chunks
   std::string m_accumulateAlg;
-  /// MPI option. If false, we will use one job event if MPI is available
-  bool m_useMPI;
 };
 
 } // namespace API

Code/Mantid/Framework/API/src/DataProcessorAlgorithm.cpp

@@ -39,7 +39,7 @@ namespace API
   }
 
   //----------------------------------------------------------------------------------------------
-  void DataProcessorAlgorithm::setLoadAlg(const std::string & alg)
+  void DataProcessorAlgorithm::setLoadAlg(const std::string &alg)
   {
     if (alg.empty())
       throw std::invalid_argument("Cannot set load algorithm to empty string");
@@ -64,11 +64,44 @@ namespace API
     throw std::runtime_error("DataProcessorAlgorithm::loadChunk is not implemented");
   }
 
+  Workspace_sptr DataProcessorAlgorithm::assemble(const std::string &partialWSName, const std::string &outputWSName)
+  {
+#ifdef MPI_BUILD
+    Workspace_sptr partialWS = AnalysisDataService::Instance().retrieve(partialWSName);
+    IAlgorithm_sptr gatherAlg = createSubAlgorithm("GatherWorkspaces");
+    gatherAlg->setAlwaysStoreInADS(true);
+    gatherAlg->setProperty("InputWorkspace", partialWS);
+    gatherAlg->setPropertyValue("OutputWorkspace", outputWSName);
+    gatherAlg->execute();
+#endif
+    Workspace_sptr outputWS = AnalysisDataService::Instance().retrieve(outputWSName);
+    return outputWS;
+  }
+
+  void DataProcessorAlgorithm::saveNexus(const std::string &outputWSName,
+      const std::string &outputFile)
+  {
+
+    bool saveOutput = true;
+#ifdef MPI_BUILD
+    if(boost::mpi::communicator().rank()>0) saveOutput = false;
+#endif
+
+    if (saveOutput && outputFile.size() > 0)
+    {
+      IAlgorithm_sptr saveAlg = createSubAlgorithm("SaveNexus");
+      saveAlg->setPropertyValue("Filename", outputFile);
+      saveAlg->setPropertyValue("InputWorkspace", outputWSName);
+      saveAlg->execute();
+    }
+  }
+
   /// Determine what kind of input data we have and load it
   //TODO: Chunking, MPI, etc...
   Workspace_sptr DataProcessorAlgorithm::load(const std::string &inputData)
   {
     Workspace_sptr inputWS;
+    std::string outputWSName = inputData;
 
     // First, check whether we have the name of an existing workspace
     if (AnalysisDataService::Instance().doesExist(inputData))
@@ -90,7 +123,6 @@ namespace API
       {
         IAlgorithm_sptr loadAlg = createSubAlgorithm(m_loadAlg);
         loadAlg->setProperty("Filename", foundFile);
-        loadAlg->setPropertyValue("OutputWorkspace", inputData);
         loadAlg->setAlwaysStoreInADS(true);
 
         // Set up MPI if available
@@ -102,19 +134,17 @@ namespace API
           m_useMPI = true;
           // The communicator containing all processes
           boost::mpi::communicator world;
-          // Create a new communicator that includes only those processes that have an input workspace
-          boost::mpi::communicator included = world.split(1);
-
-          loadAlg->setProperty("ChunkNumber", included.rank());
-          loadAlg->setProperty("TotalChunks", included.size());
+          g_log.notice() << "Chunk/Total: " << world.rank()+1 << "/" << world.size() << std::endl;
+          loadAlg->setPropertyValue("OutputWorkspace", outputWSName);
+          loadAlg->setProperty("ChunkNumber", world.rank()+1);
+          loadAlg->setProperty("TotalChunks", world.size());
         }
+#else
+        loadAlg->setPropertyValue("OutputWorkspace", outputWSName);
 #endif
-
-
-
-
         loadAlg->execute();
-        Workspace_sptr inputMatrixWS = AnalysisDataService::Instance().retrieve(inputData);
+
+        Workspace_sptr inputMatrixWS = AnalysisDataService::Instance().retrieve(outputWSName);
       }
     }
     return inputWS;

Code/Mantid/Framework/PythonAPI/PythonAlgorithms/WorkflowAlgorithms/EQSANSLiveReduce.py

@@ -30,7 +30,7 @@ class EQSANSLiveReduce(PythonAlgorithm):
                                       Description="Output workspace containing the reduced data")
         self.declareProperty("ReductionProcess", True,
                              Description="If true, both the reduction and the post-processing will be run")
-        self.declareProperty("PostProcess", True,
+        self.declareProperty("PostProcess", False,
                              Description="If true, I(q) will be computed from the input workspace")
         self.declareFileProperty("LogDataFile", "",
                                  FileAction.OptionalLoad, [".nxs"],

Code/Mantid/Framework/WorkflowAlgorithms/src/ReductionProcessor.cpp

@@ -5,7 +5,11 @@
 #include "MantidKernel/System.h"
 #include "MantidAPI/FileFinder.h"
 #include "MantidAPI/AlgorithmManager.h"
+#include "MantidAPI/FileProperty.h"
 #include <stdexcept>
+#ifdef MPI_BUILD
+#include <boost/mpi.hpp>
+#endif
 
 using namespace Mantid::Kernel;
 using namespace Mantid::API;
@@ -31,6 +35,8 @@ namespace WorkflowAlgorithms
     declareProperty("LoadAlgorithm", "LoadEventNexus");
     declareProperty("ProcessingAlgorithm", "");
     declareProperty(new WorkspaceProperty<>("OutputWorkspace","",Direction::Output));
+    declareProperty(new FileProperty("OutputFile", "", FileProperty::OptionalSave, ".nxs"),
+        "File path for the output nexus file");
   }
 
   void ReductionProcessor::exec()
@@ -44,9 +50,7 @@ namespace WorkflowAlgorithms
     Workspace_sptr inputWS = load(inputData);
 
     // Process the data
-    g_log.information() << "Starting to process " << inputData << std::endl;
-
-    const std::string outputWSName = getPropertyValue("OutputWorkspace");
+    std::string outputWSName = getPropertyValue("OutputWorkspace");
     const std::string procAlgName = getProperty("ProcessingAlgorithm");
 
     IAlgorithm_sptr procAlg = createSubAlgorithm(procAlgName);
@@ -54,11 +58,15 @@ namespace WorkflowAlgorithms
     procAlg->setAlwaysStoreInADS(true);
     procAlg->setPropertyValue("OutputWorkspace", outputWSName);
     procAlg->execute();
-
-    g_log.information() << "Done processing " << inputData << std::endl;
-
     Workspace_sptr outputWS = AnalysisDataService::Instance().retrieve(outputWSName);
+
+    // Assemble
+    outputWS = assemble(outputWSName, outputWSName);
     setProperty("OutputWorkspace", outputWS);
+
+    // Save as necessary
+    const std::string outputFile = getPropertyValue("OutputFile");
+    saveNexus(outputWSName, outputFile);
   }
 
 }