Re #5051 Starting to fill in DataProcessorAlgorithm

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 "MantidKernel/PropertyManager.h"
 #include <vector>
 
 namespace Mantid
@@ -45,15 +46,18 @@ protected:
   void setAccumAlg(const std::string & alg);
   ITableWorkspace_sptr determineChunk();
   void loadChunk();
-  void load();
+  Workspace_sptr load(const std::string &inputData);
   std::vector<std::string> splitInput(const std::string & input);
   void forwardProperties();
+  boost::shared_ptr<Kernel::PropertyManager> getProcessProperties(const std::string &propertyManager);
 
 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

 #include "MantidAPI/DataProcessorAlgorithm.h"
+#include "MantidAPI/AlgorithmProperty.h"
+#include "MantidAPI/AnalysisDataService.h"
+#include "MantidAPI/IEventWorkspace.h"
 #include "MantidKernel/System.h"
+#include "MantidAPI/FileFinder.h"
+#include "MantidAPI/AlgorithmManager.h"
+#include "MantidAPI/PropertyManagerDataService.h"
+#include "MantidKernel/PropertyManager.h"
 #include <stdexcept>
+#ifdef MPI_BUILD
+#include <boost/mpi.hpp>
+#endif
 
 using namespace Mantid::Kernel;
 using namespace Mantid::API;
@@ -18,6 +28,7 @@ namespace API
   {
     m_loadAlg = "Load";
     m_accumulateAlg = "Plus";
+    m_useMPI = false;
   }
     
   //----------------------------------------------------------------------------------------------
@@ -49,12 +60,83 @@ namespace API
 
   void DataProcessorAlgorithm::loadChunk()
   {
+
     throw std::runtime_error("DataProcessorAlgorithm::loadChunk is not implemented");
   }
 
-  void DataProcessorAlgorithm::load()
+  /// 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;
+
+    // First, check whether we have the name of an existing workspace
+    if (AnalysisDataService::Instance().doesExist(inputData))
+    {
+      Workspace_sptr inputWS = AnalysisDataService::Instance().retrieve(inputData);
+    }
+    else
+    {
+      std::string foundFile = FileFinder::Instance().getFullPath(inputData);
+      if (foundFile.empty())
+      {
+        // Get facility extensions
+        FacilityInfo facilityInfo = ConfigService::Instance().getFacility();
+        const std::vector<std::string>  facilityExts = facilityInfo.extensions();
+        foundFile = FileFinder::Instance().findRun(inputData, facilityExts);
+      }
+
+      if (!foundFile.empty())
+      {
+        IAlgorithm_sptr loadAlg = createSubAlgorithm(m_loadAlg);
+        loadAlg->setProperty("Filename", foundFile);
+        loadAlg->setPropertyValue("OutputWorkspace", inputData);
+        loadAlg->setAlwaysStoreInADS(true);
+
+        // Set up MPI if available
+#ifdef MPI_BUILD
+        // First, check whether the loader allows use to chunk the data
+        if (loadAlg->existsProperty("ChunkNumber")
+            && loadAlg->existsProperty("TotalChunks"))
+        {
+          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());
+        }
+#endif
+
+
+
+
+        loadAlg->execute();
+        Workspace_sptr inputMatrixWS = AnalysisDataService::Instance().retrieve(inputData);
+      }
+    }
+    return inputWS;
+  }
+
+  /// Get the property manager object of a given name from the property manager
+  /// data service, or create a new one.
+  boost::shared_ptr<PropertyManager>
+  DataProcessorAlgorithm::getProcessProperties(const std::string &propertyManager)
   {
-    throw std::runtime_error("DataProcessorAlgorithm::load is not implemented");
+    boost::shared_ptr<PropertyManager> processProperties;
+    if (PropertyManagerDataService::Instance().doesExist(propertyManager))
+    {
+      processProperties = PropertyManagerDataService::Instance().retrieve(propertyManager);
+    }
+    else
+    {
+      g_log.notice() << "Could not find property manager" << std::endl;
+      processProperties = boost::make_shared<PropertyManager>();
+      PropertyManagerDataService::Instance().addOrReplace(propertyManager, processProperties);
+    }
+    return processProperties;
   }
 
   std::vector<std::string> DataProcessorAlgorithm::splitInput(const std::string & input)

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

@@ -22,13 +22,15 @@ class EQSANSLiveReduce(PythonAlgorithm):
         return "EQSANSLiveReduce"
 
     def PyInit(self):
-        self.declareProperty("InputWorkspace", "",
-                             Direction=Direction.Input,
-                             Description="Workspace to be reduced")
+        self.declareWorkspaceProperty("InputWorkspace", "",
+                                      Direction=Direction.Input,
+                                      Description="Workspace to be reduced")
         self.declareWorkspaceProperty("OutputWorkspace", "",
                                       Direction=Direction.Output,
                                       Description="Output workspace containing the reduced data")
-        self.declareProperty("PostProcess", False,
+        self.declareProperty("ReductionProcess", True,
+                             Description="If true, both the reduction and the post-processing will be run")
+        self.declareProperty("PostProcess", True,
                              Description="If true, I(q) will be computed from the input workspace")
         self.declareFileProperty("LogDataFile", "",
                                  FileAction.OptionalLoad, [".nxs"],
@@ -36,12 +38,31 @@ class EQSANSLiveReduce(PythonAlgorithm):
 
     def PyExec(self):
         workspace = self.getPropertyValue("InputWorkspace")
+        output_workspace = self.getPropertyValue("OutputWorkspace")
+        reduction_process = self.getProperty("ReductionProcess")
         post_process = self.getProperty("PostProcess")
         
-        if post_process:
-            return self._post_process(workspace)
-
         # Reduce the data
+        if reduction_process:
+            self._reduction_process(workspace)
+
+        # Post-processing        
+        if post_process:
+            self._post_process(workspace)
+            RenameWorkspace(InputWorkspace=workspace+'_Iq', OutputWorkspace=output_workspace)
+            self._setWorkspaceProperty("OutputWorkspace", mtd[output_workspace]._getHeldObject())
+        elif reduction_process:
+            if workspace != output_workspace:
+                RenameWorkspace(InputWorkspace=workspace, OutputWorkspace=output_workspace)
+            self._setWorkspaceProperty("OutputWorkspace", mtd[output_workspace]._getHeldObject())
+        else:
+            mtd.sendErrorMessage("EQSANSLiveReduce has both the ReductionProcess and PostProcess properties set to FALSE")
+    
+    def _reduction_process(self, workspace):
+        """
+            Perform the main reduction process
+            @param workspace: workspace to reduce
+        """
         file_path = self.getProperty("LogDataFile")
         if os.path.isfile(file_path):
             LoadNexusLogs(Workspace=workspace, Filename=file_path,
@@ -53,7 +74,7 @@ class EQSANSLiveReduce(PythonAlgorithm):
         import reduction.instruments.sans.sns_command_interface as cmd 
         cmd.AppendDataFile([workspace])
         cmd.Reduce1D()
-    
+        
     def _post_process(self, workspace):
         """
             Post processing. Compute I(q) on the final reduced workspace
@@ -63,6 +84,6 @@ class EQSANSLiveReduce(PythonAlgorithm):
         from reduction.instruments.sans.sns_reduction_steps import AzimuthalAverageByFrame
         averager = AzimuthalAverageByFrame()
         averager.execute(cmd.ReductionSingleton(), workspace)
-        self.setPropertyValue("OutputWorkspace", workspace+'_Iq')
+        #self.setPropertyValue("OutputWorkspace", workspace+'_Iq')
                 
 mtd.registerPyAlgorithm(EQSANSLiveReduce())