SNSPowderReduction.py

# 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 +
#pylint: disable=invalid-name,no-init,too-many-lines
import os
from pathlib import Path
import mantid.simpleapi as api
from mantid.api import mtd, AlgorithmFactory, AnalysisDataService, DistributedDataProcessorAlgorithm, \
    FileAction, FileProperty, ITableWorkspaceProperty, MultipleFileProperty, PropertyMode, WorkspaceProperty, \
    ITableWorkspace, MatrixWorkspace
from mantid.kernel import (
    ConfigService, Direction, EnabledWhenProperty, FloatArrayProperty, FloatBoundedValidator, IntArrayBoundedValidator,
    IntArrayProperty, Property, PropertyCriterion, PropertyManagerDataService, StringListValidator)
from mantid.dataobjects import SplittersWorkspace  # SplittersWorkspace
from mantid.utils import absorptioncorrutils
if AlgorithmFactory.exists('GatherWorkspaces'):
    HAVE_MPI = True
    from mpi4py import MPI
    mpiRank = MPI.COMM_WORLD.Get_rank()
else:
    HAVE_MPI = False
    mpiRank = 0 # simplify if clauses


EVENT_WORKSPACE_ID = "EventWorkspace"
EXTENSIONS_NXS = ["_event.nxs", ".nxs.h5"]


def noRunSpecified(runs):
    """ Check whether any run is specified
    Purpose:
        Check whether any run, which is larger than 0, is specified in the input numpy array
    Requirements:
        Input is a numpy array
    Guarantees:
        A boolean is returned
    :param runs:
    :return:
    """
    # return True if runs is of size zero
    if len(runs) <= 0:
        return True

    # return True if there is one and only one run in runs and it is not greater than 0.
    if len(runs) == 1:
        # break off the instrument part
        value = int(runs[0].split('_')[-1])
        # -1 turns off the runnumber
        return value <= 0

    return False


def get_workspace(workspace_name):
    """
    Purpose: Get the reference of a workspace
    Requirements:
    1. workspace_name is a string
    Guarantees:
    The reference of the workspace with same is returned.
    :exception RuntimeError: a RuntimeError from Mantid will be thrown
    :param workspace_name:
    :return:
    """
    assert isinstance(workspace_name, str), 'Input workspace name {0} must be a string but not a {1}.' \
                                            ''.format(workspace_name, type(workspace_name))

    return AnalysisDataService.retrieve(workspace_name)


def is_event_workspace(workspace):
    if isinstance(workspace, str):
        return get_workspace(workspace).id() == EVENT_WORKSPACE_ID
    else:
        return workspace.id() == EVENT_WORKSPACE_ID


def allEventWorkspaces(*args):
    """
    Purpose:
        Check whether all the inputs are event workspace
    Requirements:
        Each element in the args is a string as name of workspace
    Guarantees:
        return boolean
    @param args:
    @return:
    """
    result = True

    args = [is_event_workspace(arg) for arg in args]
    for arg in args:
        result = result and arg

    return result


def getBasename(filename):
    if type(filename) == list:
        filename = filename[0]
    name = os.path.split(filename)[-1]
    for extension in EXTENSIONS_NXS:
        name = name.replace(extension, '')
    return name

#pylint: disable=too-many-instance-attributes


class SNSPowderReduction(DistributedDataProcessorAlgorithm):
    COMPRESS_TOL_TOF = .01
    _resampleX = None
    _binning = None
    _bin_in_dspace = None
    _instrument = None
    _filterBadPulses = None
    _removePromptPulseWidth = None
    _LRef = None
    _DIFCref = None
    _wavelengthMin = None
    _wavelengthMax = None
    _vanPeakFWHM = None
    _vanSmoothing = None
    _vanRadius = None
    _scaleFactor = None
    _outDir = None
    _outPrefix = None
    _outTypes = None
    _chunks = None
    _splittersWS = None
    _splitinfotablews = None
    _normalisebycurrent = None
    _lowResTOFoffset = None
    _focusPos = None
    _charTable = None
    iparmFile = None
    _info = None
    _absMethod = None
    _sampleFormula = None
    _massDensity = None
    _containerShape = None

    def category(self):
        return "Diffraction\\Reduction"

    def seeAlso(self):
        return [ "DiffractionFocussing","AlignAndFocusPowder" ]

    def name(self):
        return "SNSPowderReduction"

    def summary(self):
        """
        summary of the algorithm
        :return:
        """
        return "The algorithm used for reduction of powder diffraction data obtained on SNS instruments (e.g. PG3) "

    def PyInit(self):
        self.copyProperties('AlignAndFocusPowderFromFiles', ['Filename', 'PreserveEvents'])

        self.declareProperty("Sum", False,
                             "Sum the runs. Does nothing for characterization runs")
        self.declareProperty("PushDataPositive", "None",
                             StringListValidator(["None", "ResetToZero", "AddMinimum"]),
                             "Add a constant to the data that makes it positive over the whole range.")
        arrvalidator = IntArrayBoundedValidator(lower=-1)
        self.declareProperty(IntArrayProperty("BackgroundNumber", values=[0], validator=arrvalidator),
                             doc="If specified overrides value in CharacterizationRunsFile If -1 turns off correction.")
        self.declareProperty(IntArrayProperty("VanadiumNumber", values=[0], validator=arrvalidator),
                             doc="If specified overrides value in CharacterizationRunsFile. If -1 turns off correction.")
        self.declareProperty(IntArrayProperty("VanadiumBackgroundNumber", values=[0], validator=arrvalidator),
                             doc="If specified overrides value in CharacterizationRunsFile. If -1 turns off correction.")
        self.declareProperty(FileProperty(name="CalibrationFile",defaultValue="",action=FileAction.OptionalLoad,
                                          extensions=[".h5", ".hd5", ".hdf", ".cal"]))  # CalFileName
        self.declareProperty(FileProperty(name="GroupingFile",defaultValue="",action=FileAction.OptionalLoad,
                                          extensions=[".xml"]), "Overrides grouping from CalibrationFile")
        self.declareProperty(MultipleFileProperty(name="CharacterizationRunsFile",
                                                  action=FileAction.OptionalLoad,
                                                  extensions=["txt"]), "File with characterization runs denoted")
        self.declareProperty(FileProperty(name="ExpIniFilename", defaultValue="", action=FileAction.OptionalLoad,
                                          extensions=[".ini"]))
        self.copyProperties('AlignAndFocusPowderFromFiles',
                            ['LorentzCorrection', 'UnwrapRef', 'LowResRef', 'CropWavelengthMin', 'CropWavelengthMax',
                             'RemovePromptPulseWidth', 'MaxChunkSize'])
        self.declareProperty(FloatArrayProperty("Binning", values=[0., 0., 0.],
                                                direction=Direction.Input),
                             "Positive is linear bins, negative is logorithmic")  # Params
        self.copyProperties('AlignAndFocusPowderFromFiles', ['ResampleX'])
        self.declareProperty("BinInDspace", True,
                             "If all three bin parameters a specified, whether they are in dspace (true) or "
                             "time-of-flight (false)")  # DSpacing
        # section of vanadium run processing
        self.declareProperty("StripVanadiumPeaks", True,
                             "Subtract fitted vanadium peaks from the known positions.")
        self.declareProperty("VanadiumFWHM", 7, "Default=7")
        self.declareProperty("VanadiumPeakTol", 0.05,
                             "How far from the ideal position a vanadium peak can be during StripVanadiumPeaks. "
                             "Default=0.05, negative turns off")
        self.declareProperty("VanadiumSmoothParams", "20,2", "Default=20,2")
        self.declareProperty("VanadiumRadius", .3175, "Radius for CarpenterSampleCorrection")
        self.declareProperty("BackgroundSmoothParams", "", "Default=off, suggested 20,2")

        # filtering
        self.declareProperty("FilterBadPulses", 95.,  # different default value
                             doc="Filter out events measured while proton charge is more than 5% below average")
        self.declareProperty("ScaleData", defaultValue=1., validator=FloatBoundedValidator(lower=0., exclusive=True),
                             doc="Constant to multiply the data before writing out. This does not apply to "
                                 "PDFgetN files.")
        self.declareProperty("OffsetData", defaultValue=0., validator=FloatBoundedValidator(lower=0., exclusive=False),
                             doc="Constant to add to the data before writing out. This does not apply to "
                                 "PDFgetN files.")
        self.declareProperty("SaveAs", "gsas",
                             "List of all output file types. Allowed values are 'fullprof', 'gsas', 'nexus', "
                             "'pdfgetn', and 'topas'")
        self.declareProperty("OutputFilePrefix", "", "Overrides the default filename for the output file (Optional).")
        self.declareProperty(FileProperty(name="OutputDirectory", defaultValue="",action=FileAction.Directory))

        # Caching options
        self.copyProperties('AlignAndFocusPowderFromFiles', 'CacheDir')
        self.declareProperty('CleanCache', False, 'Remove all cache files within CacheDir')
        self.setPropertySettings('CleanCache', EnabledWhenProperty('CacheDir', PropertyCriterion.IsNotDefault))
        property_names = ('CacheDir', 'CleanCache')
        [self.setPropertyGroup(name, 'Caching') for name in property_names]

        self.declareProperty("FinalDataUnits", "dSpacing", StringListValidator(["dSpacing","MomentumTransfer"]))

        # absorption correction
        self.declareProperty("TypeOfCorrection", "None",
                             StringListValidator(["None", "SampleOnly", "SampleAndContainer", "FullPaalmanPings"]),
                             doc="Specifies the Absorption Correction terms to calculate, if any.")
        self.declareProperty("SampleFormula", "", doc="Chemical formula of the sample")
        self.declareProperty("MeasuredMassDensity", defaultValue=0.1,
                             validator=FloatBoundedValidator(lower=0., exclusive=True),
                             doc="Measured mass density of sample in g/cc")  # in g/cc, way to validate?
        self.declareProperty("SampleNumberDensity", defaultValue=Property.EMPTY_DBL,
                             doc="Number density of the sample in number of atoms per cubic Angstrom will be used instead of calculated")
        self.declareProperty("ContainerShape", defaultValue="PAC06", doc="Defines the container geometry")
        self.declareProperty("ContainerScaleFactor", defaultValue=1.0,
                             validator=FloatBoundedValidator(lower=0),
                             doc="Factor to scale the container data")
        self.copyProperties("AbsorptionCorrection", "ElementSize")
        self.declareProperty("NumWavelengthBins", defaultValue=1000,
                             doc="Number of wavelength bin to calculate the for absorption correction")

        workspace_prop = WorkspaceProperty('SplittersWorkspace', '', Direction.Input, PropertyMode.Optional)
        self.declareProperty(workspace_prop, "Splitters workspace for split event workspace.")
        # replaced for matrix workspace, SplittersWorkspace and table workspace
        # tableprop = ITableWorkspaceProperty("SplittersWorkspace", "", Direction.Input, PropertyMode.Optional)
        # self.declareProperty(tableprop, "Splitters workspace for split event workspace.")

        infotableprop = ITableWorkspaceProperty("SplitInformationWorkspace", "", Direction.Input, PropertyMode.Optional)
        self.declareProperty(infotableprop, "Name of table workspace containing information for splitters.")

        self.declareProperty("LowResolutionSpectraOffset", -1,
                             "If larger and equal to 0, then process low resolution TOF and offset is the spectra "
                             "number. Otherwise, ignored.")  # LowResolutionSpectraOffset

        self.declareProperty("NormalizeByCurrent", True, "Normalize by current")

        self.declareProperty("CompressTOFTolerance", 0.01, "Tolerance to compress events in TOF.")

        self.copyProperties('AlignAndFocusPowderFromFiles', ['FrequencyLogNames', 'WaveLengthLogNames'])

        return

    def validateInputs(self):
        issues = dict()

        # If doing absorption correction, make sure the sample formula is correct
        if self.getProperty("TypeOfCorrection").value != "None":
            if self.getProperty("SampleFormula").value == '':
                issues['SampleFormula'] = "A sample formula must be provided."

        # The provided cache directory does not exist
        cache_dir = self.getProperty('CacheDir').value  # absolute or relative path, as a string
        if bool(cache_dir) and Path(cache_dir).exists() is False:
            issues['CacheDir'] = f'Directory {cache_dir} does not exist'

        # We cannot clear the cache if property "CacheDir" has not been set
        if self.getProperty('CleanCache').value and not bool(self.getProperty('CacheDir').value):
            issues['CleanCache'] = f'Property "CacheDir" must be set in order to clean the cache'

        return issues

    #pylint: disable=too-many-locals,too-many-branches,too-many-statements
    def PyExec(self):  # noqa
        """ Main execution body
        """
        # get generic information
        self._loadCharacterizations()
        self._resampleX = self.getProperty("ResampleX").value
        if self._resampleX != 0.:
            self._binning = [0.]
        else:
            self._binning = self.getProperty("Binning").value
            if len(self._binning) != 1 and len(self._binning) != 3:
                raise RuntimeError("Can only specify (width) or (start,width,stop) for binning. Found %d values." % len(self._binning))
            if len(self._binning) == 3:
                if self._binning[0] == 0. and self._binning[1] == 0. and self._binning[2] == 0.:
                    raise RuntimeError("Failed to specify the binning")
        self._bin_in_dspace = self.getProperty("BinInDspace").value
        self._filterBadPulses = self.getProperty("FilterBadPulses").value
        self._removePromptPulseWidth = self.getProperty("RemovePromptPulseWidth").value
        self._lorentz = self.getProperty("LorentzCorrection").value
        self._LRef = self.getProperty("UnwrapRef").value
        self._DIFCref = self.getProperty("LowResRef").value
        self._wavelengthMin = self.getProperty("CropWavelengthMin").value
        self._wavelengthMax = self.getProperty("CropWavelengthMax").value
        self._vanPeakFWHM = self.getProperty("VanadiumFWHM").value
        self._vanSmoothing = self.getProperty("VanadiumSmoothParams").value
        self._vanRadius = self.getProperty("VanadiumRadius").value
        self.calib = self.getProperty("CalibrationFile").value
        self._scaleFactor = self.getProperty("ScaleData").value
        self._offsetFactor = self.getProperty("OffsetData").value
        self._outDir = self.getProperty("OutputDirectory").value
        # Caching options
        self._cache_dir = self.getProperty("CacheDir").value
        self._clean_cache = self.getProperty("CleanCache").value

        self._outPrefix = self.getProperty("OutputFilePrefix").value.strip()
        self._outTypes = self.getProperty("SaveAs").value.lower()
        self._absMethod = self.getProperty("TypeOfCorrection").value
        self._sampleFormula = self.getProperty("SampleFormula").value
        self._massDensity = self.getProperty("MeasuredMassDensity").value
        self._numberDensity = self.getProperty("SampleNumberDensity").value
        self._containerShape = self.getProperty("ContainerShape").value
        self._containerScaleFactor = self.getProperty("ContainerScaleFactor").value
        self._elementSize = self.getProperty("ElementSize").value
        self._num_wl_bins = self.getProperty("NumWavelengthBins").value

        samRuns = self._getLinearizedFilenames("Filename")
        self._determineInstrument(samRuns[0])

        preserveEvents = self.getProperty("PreserveEvents").value
        if HAVE_MPI and preserveEvents:
            self.log().warning("preserveEvents set to False for MPI tasks.")
            preserveEvents = False
        self._info = None
        self._chunks = self.getProperty("MaxChunkSize").value

        # define splitters workspace and filter wall time
        self._splittersWS = self.getProperty("SplittersWorkspace").value
        if self._splittersWS is not None:
            # user specifies splitters workspace
            self.log().information("SplittersWorkspace is %s" % (str(self._splittersWS)))
            if len(samRuns) != 1:
                # TODO/FUTURE - This should be supported
                raise RuntimeError("Reducing data with splitters cannot happen when there are more than 1 sample run.")
            # define range of wall-time to import data
            sample_time_filter_wall = self._get_time_filter_wall(self._splittersWS.name(), samRuns[0])
            self.log().information("The time filter wall is %s" %(str(sample_time_filter_wall)))
        else:
            sample_time_filter_wall = (0.0, 0.0)
            self.log().information("SplittersWorkspace is None, and thus there is NO time filter wall. ")

        self._splitinfotablews = self.getProperty("SplitInformationWorkspace").value

        self._normalisebycurrent = self.getProperty("NormalizeByCurrent").value

        # Tolerance for compress TOF event.  If given a negative value, then use default 0.01
        self.COMPRESS_TOL_TOF = float(self.getProperty("CompressTOFTolerance").value)
        if self.COMPRESS_TOL_TOF < -0.:
            self.COMPRESS_TOL_TOF = 0.01

        # Clean the cache directory if so requested
        if self._clean_cache:
            api.CleanFileCache(CacheDir=self._cache_dir, AgeInDays=0)

        # Process data
        # List stores the workspacename of all data workspaces that will be converted to d-spacing in the end.
        workspacelist = []
        samwksplist = []

        self._lowResTOFoffset = self.getProperty("LowResolutionSpectraOffset").value
        focuspos = self._focusPos
        if self._lowResTOFoffset >= 0:
            # Dealing with the parameters for editing instrument parameters
            if "PrimaryFlightPath" in focuspos:
                l1 = focuspos["PrimaryFlightPath"]
                if l1 > 0:
                    specids = focuspos['SpectrumIDs'][:]
                    l2s = focuspos['L2'][:]
                    polars = focuspos['Polar'][:]
                    phis = focuspos['Azimuthal'][:]

                    specids.extend(specids)
                    l2s.extend(l2s)
                    polars.extend(polars)
                    phis.extend(phis)

                    focuspos['SpectrumIDs'] = specids
                    focuspos['L2'] = l2s
                    focuspos['Polar'] = polars
                    focuspos['Azimuthal'] = phis
        # ENDIF

        # calculate absorption from first sample run
        metaws = None
        if self._absMethod != "None" and self._info is None:
            absName = '__{}_abs'.format(getBasename(samRuns[0]))
            api.Load(Filename=samRuns[0], OutputWorkspace=absName, MetaDataOnly=True)
            self._info = self._getinfo(absName)
            metaws = absName
        # NOTE: inconsistent naming among different methods
        #       -> adding more comments to help clarify
        a_sample, a_container = absorptioncorrutils.calculate_absorption_correction(
            samRuns[0],  # filename: File to be used for absorption correction
            self._absMethod,  # [None, SampleOnly, SampleAndContainer, FullPaalmanPings]
            self._info,  # PropertyManager of run characterizations
            self._sampleFormula,  # Material for absorption correction
            self._massDensity,  # Mass density of the sample
            self._numberDensity,  # Optional number density of sample to be added
            self._containerShape,  # Shape definition of container
            self._num_wl_bins,  # Number of bins: len(ws.readX(0))-1
            self._elementSize,  # Size of one side of the integration element cube in mm
            metaws,  # Optional workspace containing metadata
            self.getProperty("CacheDir").value,  # Cache dir for absoption correction workspace
            "SHA", # Use cache files named with sha rather than a filename prefix
        )

        if self.getProperty("Sum").value and len(samRuns) > 1:
            self.log().information('Ignoring value of "Sum" property')
            # Sum input sample runs and then do reduction
            if self._splittersWS is not None:
                raise NotImplementedError("Summing spectra and filtering events are not supported simultaneously.")

            sam_ws_name = self._focusAndSum(samRuns, preserveEvents=preserveEvents, absorptionWksp=a_sample)
            assert isinstance(sam_ws_name, str), 'Returned from _focusAndSum() must be a string but not' \
                                                 '%s. ' % str(type(sam_ws_name))

            workspacelist.append(sam_ws_name)
            samwksplist.append(sam_ws_name)
        # ENDIF (SUM)
        else:
            # Process each sample run
            for sam_run_number in samRuns:
                # first round of processing the sample
                self._info = None
                if sample_time_filter_wall[0] == 0. and sample_time_filter_wall[-1] == 0. \
                        and self._splittersWS is None:
                    returned = self._focusAndSum([sam_run_number], preserveEvents=preserveEvents, absorptionWksp=a_sample)
                else:
                    returned = self._focusChunks(sam_run_number, sample_time_filter_wall,
                                                 splitwksp=self._splittersWS,
                                                 preserveEvents=preserveEvents)

                if isinstance(returned, list):
                    # Returned with a list of workspaces
                    focusedwksplist = returned
                    for sam_ws_name in focusedwksplist:
                        assert isinstance(sam_ws_name, str), 'Impossible to have a non-string value in ' \
                                                             'returned focused workspaces\' names.'
                        samwksplist.append(sam_ws_name)
                        workspacelist.append(sam_ws_name)
                    # END-FOR
                else:
                    # returned as a single workspace
                    sam_ws_name = returned
                    assert isinstance(sam_ws_name, str)
                    samwksplist.append(sam_ws_name)
                    workspacelist.append(sam_ws_name)
                # ENDIF
            # END-FOR
        # ENDIF (Sum data or not)

        for (samRunIndex, sam_ws_name) in enumerate(samwksplist):
            assert isinstance(sam_ws_name, str), 'Assuming that samRun is a string. But it is %s' % str(type(sam_ws_name))
            if is_event_workspace(sam_ws_name):
                self.log().information('Sample Run %s:  starting number of events = %d.' % (
                    sam_ws_name, get_workspace(sam_ws_name).getNumberEvents()))

            # Get run information
            self._info = self._getinfo(sam_ws_name)

            # process the container
            can_run_numbers = self._info["container"].value
            can_run_numbers = ['%s_%d' % (self._instrument, value) for value in can_run_numbers]
            can_run_ws_name = self._process_container_runs(can_run_numbers, samRunIndex, preserveEvents, absorptionWksp=a_container)
            if can_run_ws_name is not None:
                workspacelist.append(can_run_ws_name)

            # process the vanadium run
            van_run_number_list = self._info["vanadium"].value
            van_run_number_list = ['%s_%d' % (self._instrument, value) for value in van_run_number_list]
            van_specified = not noRunSpecified(van_run_number_list)
            if van_specified:
                van_run_ws_name = self._process_vanadium_runs(van_run_number_list, samRunIndex)
                workspacelist.append(van_run_ws_name)
            else:
                van_run_ws_name = None

            # return if MPI is used and there is more than 1 processor
            if mpiRank > 0:
                return

            # return if there is no sample run
            # Note: sample run must exist in logic
            # VZ: Discuss with Pete
            # if sam_ws_name == 0:
            #   return

            # the final bit of math to remove container run and vanadium run
            if can_run_ws_name is not None and self._containerScaleFactor != 0:
                # must convert the sample to a matrix workspace if the can run isn't one
                if not allEventWorkspaces(can_run_ws_name, sam_ws_name):
                    api.ConvertToMatrixWorkspace(InputWorkspace=sam_ws_name,
                                                 OutputWorkspace=sam_ws_name)

                # remove container run
                api.RebinToWorkspace(WorkspaceToRebin=can_run_ws_name,
                                     WorkspaceToMatch=sam_ws_name,
                                     OutputWorkspace=can_run_ws_name)
                api.Scale(InputWorkspace=can_run_ws_name,
                          OutputWorkspace=can_run_ws_name,
                          Factor=self._containerScaleFactor)
                api.Minus(LHSWorkspace=sam_ws_name,
                          RHSWorkspace=can_run_ws_name,
                          OutputWorkspace=sam_ws_name)
                # compress event if the sample run workspace is EventWorkspace
                if is_event_workspace(sam_ws_name) and self.COMPRESS_TOL_TOF > 0.:
                    api.CompressEvents(InputWorkspace=sam_ws_name,
                                       OutputWorkspace=sam_ws_name,
                                       Tolerance=self.COMPRESS_TOL_TOF)  # 10ns
                # canRun = str(canRun)

            if van_run_ws_name is not None:
                # subtract vanadium run from sample run by division
                num_hist_sam = get_workspace(sam_ws_name).getNumberHistograms()
                num_hist_van = get_workspace(van_run_ws_name).getNumberHistograms()
                assert num_hist_sam == num_hist_van, \
                    'Number of histograms of sample %d is not equal to van %d.' % (num_hist_sam, num_hist_van)
                api.Divide(LHSWorkspace=sam_ws_name,
                           RHSWorkspace=van_run_ws_name,
                           OutputWorkspace=sam_ws_name)
                normalized = True
                van_run_ws = get_workspace(van_run_ws_name)
                sam_ws = get_workspace(sam_ws_name)
                sam_ws.getRun()['van_number'] = van_run_ws.getRun()['run_number'].value
                # van_run_number = str(van_run_number)
            else:
                normalized = False

            # Compress the event again
            if is_event_workspace(sam_ws_name) and self.COMPRESS_TOL_TOF > 0.:
                api.CompressEvents(InputWorkspace=sam_ws_name,
                                   OutputWorkspace=sam_ws_name,
                                   Tolerance=self.COMPRESS_TOL_TOF)  # 5ns/

            # write out the files
            # FIXME - need documentation for mpiRank
            if mpiRank == 0:
                if self._scaleFactor != 1.:
                    api.Scale(sam_ws_name, Factor=self._scaleFactor, OutputWorkspace=sam_ws_name)
                if self._offsetFactor != 0.:
                    api.ConvertToMatrixWorkspace(InputWorkspace=sam_ws_name,
                                                 OutputWorkspace=sam_ws_name)
                    api.Scale(sam_ws_name, Factor=self._offsetFactor, OutputWorkspace=sam_ws_name,
                              Operation='Add')
                # make sure there are no negative values - gsas hates them
                if self.getProperty("PushDataPositive").value != "None":
                    addMin = self.getProperty("PushDataPositive").value == "AddMinimum"
                    api.ResetNegatives(InputWorkspace=sam_ws_name,
                                       OutputWorkspace=sam_ws_name,
                                       AddMinimum=addMin,
                                       ResetValue=0.)

                self._save(sam_ws_name, self._info, normalized, False)
        # ENDFOR

        # convert everything into d-spacing as the final units
        if mpiRank == 0:
            workspacelist = set(workspacelist) # only do each workspace once
            for wksp in workspacelist:
                api.ConvertUnits(InputWorkspace=wksp, OutputWorkspace=wksp,
                                 Target=self.getProperty("FinalDataUnits").value)

                propertyName = "OutputWorkspace%s" % wksp
                if not self.existsProperty(propertyName):
                    self.declareProperty(WorkspaceProperty(propertyName, wksp, Direction.Output))
                self.setProperty(propertyName, wksp)

        return

    def _getLinearizedFilenames(self, propertyName):
        runnumbers = self.getProperty(propertyName).value
        linearizedRuns = []
        for item in runnumbers:
            if type(item) == list:
                linearizedRuns.extend(item)
            else:
                linearizedRuns.append(item)
        return linearizedRuns

    def _determineInstrument(self, filename):
        name = getBasename(filename)
        parts = name.split('_')
        self._instrument = ConfigService.getInstrument(parts[0]).shortName()  # only works for instruments without '_'

    def _loadCharacterizations(self):
        self._focusPos = {}
        charFilename = self.getProperty("CharacterizationRunsFile").value
        charFilename = ','.join(charFilename)
        expIniFilename = self.getProperty("ExpIniFilename").value

        self._charTable = ''
        if charFilename is None or len(charFilename) <= 0:
            self.iparmFile = None
            return

        self._charTable = 'characterizations'
        results = api.PDLoadCharacterizations(Filename=charFilename,
                                              ExpIniFilename=expIniFilename,
                                              OutputWorkspace=self._charTable)
        # export the characterizations table
        charTable = results[0]
        if not self.existsProperty("CharacterizationsTable"):
            self.declareProperty(ITableWorkspaceProperty("CharacterizationsTable", self._charTable, Direction.Output))
        self.setProperty("CharacterizationsTable", charTable)

        # get the focus positions from the properties
        self.iparmFile = results[1]
        self._focusPos['PrimaryFlightPath'] = results[2]
        self._focusPos['SpectrumIDs'] = results[3]
        self._focusPos['L2'] = results[4]
        self._focusPos['Polar'] = results[5]
        self._focusPos['Azimuthal'] = results[6]

    #pylint: disable=too-many-branches
    def _load_event_data(self, filename, filter_wall=None, out_ws_name=None, **chunk):
        """ Load data optionally by chunk strategy
        Purpose:
            Load a complete or partial run, filter bad pulses.
            Output workspace name is formed as
            - user specified (highest priority)
            - instrument-name_run-number_0 (no chunking)
            - instrument-name_run-number_X: X is either ChunkNumber of SpectrumMin
        Requirements:
            1. run number is integer and larger than 0
        Guarantees:
            A workspace is created with name described above
        :param filename:
        :param filter_wall:
        :param out_ws_name: name of output workspace specified by user. it will override the automatic name
        :param chunk:
        :return:
        """
        # Check requirements
        assert len(filename) > 0, "Input file '%s' does not exist" % filename
        assert (chunk is None) or isinstance(chunk, dict), 'Input chunk must be either a dictionary or None.'

        # get event file's base name
        base_name = getBasename(filename)

        # give out the default output workspace name
        if out_ws_name is None:
            if chunk:
                if "ChunkNumber" in chunk:
                    out_ws_name = base_name + "__chk%d" % (int(chunk["ChunkNumber"]))
                elif "SpectrumMin" in chunk:
                    seq_number = 1 + int(chunk["SpectrumMin"])/(int(chunk["SpectrumMax"])-int(chunk["SpectrumMin"]))
                    out_ws_name = base_name + "__chk%d" % seq_number
            else:
                out_ws_name = "%s_%d" % (base_name, 0)
        # END-IF

        # Specify the other chunk information including Precount, FilterByTimeStart and FilterByTimeStop.
        if filename.endswith("_event.nxs") or filename.endswith(".nxs.h5"):
            chunk["Precount"] = True
            if filter_wall is not None:
                if filter_wall[0] > 0.:
                    chunk["FilterByTimeStart"] = filter_wall[0]
                if filter_wall[1] > 0.:
                    chunk["FilterByTimeStop"] = filter_wall[1]

        # Call Mantid's Load algorithm to load complete or partial data
        api.Load(Filename=filename, OutputWorkspace=out_ws_name, **chunk)

        # Log output
        if is_event_workspace(out_ws_name):
            self.log().debug("Load run %s: number of events = %d. " % (os.path.split(filename)[-1],
                                                                       get_workspace(out_ws_name).getNumberEvents()))
        if HAVE_MPI:
            msg = "MPI Task = %s ;" % (str(mpiRank))
            if is_event_workspace(out_ws_name):
                msg += "Number Events = " + str(get_workspace(out_ws_name).getNumberEvents())
            self.log().debug(msg)

        # filter bad pulses
        if self._filterBadPulses > 0.:
            # record number of events of the original workspace
            if is_event_workspace(out_ws_name):
                # Event workspace: record original number of events
                num_original_events = get_workspace(out_ws_name).getNumberEvents()
            else:
                num_original_events = -1

            # filter bad pulse
            api.FilterBadPulses(InputWorkspace=out_ws_name, OutputWorkspace=out_ws_name,
                                LowerCutoff=self._filterBadPulses)

            if is_event_workspace(out_ws_name):
                # Event workspace
                message = "FilterBadPulses reduces number of events from %d to %d (under %s percent) " \
                          "of workspace %s." % (num_original_events, get_workspace(out_ws_name).getNumberEvents(),
                                                str(self._filterBadPulses), out_ws_name)
                self.log().information(message)
        # END-IF (filter bad pulse)

        return out_ws_name

    def _getStrategy(self, filename):
        """
        Get chunking strategy by calling mantid algorithm 'DetermineChunking'
        :param filename:
        :return: a list of dictionary.  Each dictionary is a row in table workspace
        """
        # Determine chunk strategy can search in archive.
        # Therefore this will fail: assert os.path.exists(file_name), 'NeXus file %s does not exist.' % file_name
        self.log().debug("[Fx116] Run file Name : %s,\t\tMax chunk size: %s" %
                         (filename, str(self._chunks)))
        chunks = api.DetermineChunking(Filename=filename, MaxChunkSize=self._chunks)

        strategy = []
        for row in chunks:
            strategy.append(row)

        # For table with no rows
        if len(strategy) == 0:
            strategy.append({})

        # delete chunks workspace
        chunks = str(chunks)
        mtd.remove(chunks)

        return strategy

    def __logChunkInfo(self, chunk):
        keys = sorted(chunk.keys())

        keys = [str(key) + "=" + str(chunk[key]) for key in keys]
        self.log().information("Working on chunk [" + ", ".join(keys) + "]")

    def checkInfoMatch(self, left, right):
        if (left["frequency"].value is not None) and (right["frequency"].value is not None) \
           and (abs(left["frequency"].value - right["frequency"].value)/left["frequency"].value > .05):
            raise RuntimeError("Cannot add incompatible frequencies (%f!=%f)"
                               % (left["frequency"].value, right["frequency"].value))
        if (left["wavelength"].value is not None) and (right["wavelength"].value is not None) \
                and abs(left["wavelength"].value - right["wavelength"].value)/left["wavelength"].value > .05:
            raise RuntimeError("Cannot add incompatible wavelengths (%f != %f)"
                               % (left["wavelength"].value, right["wavelength"].value))

    #pylint: disable=too-many-arguments
    def _focusAndSum(self, filenames, preserveEvents=True, final_name=None, absorptionWksp=''):
        """Load, sum, and focus data in chunks
        Purpose:
            Load, sum and focus data in chunks;
        Requirements:
            1. input run numbers are in numpy array or list
        Guarantees:
            The experimental runs are focused and summed together
        @param run_number_list:
        @param extension:
        @param preserveEvents:
        @param absorptionWksp: will be divided from the data at a per-pixel level
        @return: string as the summed workspace's name
        """
        if final_name is None:
            final_name = getBasename(filenames[0])

        # only pass in the characterizations if the values haven't already been determined
        characterizations = self._charTable
        if self._info is not None:
            characterizations = ''

        # put together a list of the other arguments
        otherArgs = self._focusPos.copy()
        # use the workspaces if they already exists, or pass the filenames down
        # this assumes that AlignAndFocusPowderFromFiles will give the workspaces the canonical names
        cal, grp, msk = [self._instrument + name for name in ['_cal', '_group', '_mask']]
        if mtd.doesExist(cal) and mtd.doesExist(grp) and mtd.doesExist(msk):
            otherArgs['CalibrationWorkspace'] = cal
            otherArgs['GroupingWorkspace'] = grp
            otherArgs['MaskWorkspace'] = msk
        else:
            otherArgs['CalFileName'] = self.calib
            otherArgs['GroupFilename'] = self.getProperty("GroupingFile").value

        api.AlignAndFocusPowderFromFiles(Filename=','.join(filenames),
                                         OutputWorkspace=final_name,
                                         AbsorptionWorkspace=absorptionWksp,
                                         MaxChunkSize=self._chunks,
                                         FilterBadPulses=self._filterBadPulses,
                                         Characterizations=characterizations,
                                         CacheDir=self._cache_dir,
                                         Params=self._binning,
                                         ResampleX=self._resampleX,
                                         Dspacing=self._bin_in_dspace,
                                         PreserveEvents=preserveEvents,
                                         RemovePromptPulseWidth=self._removePromptPulseWidth,
                                         CompressTolerance=self.COMPRESS_TOL_TOF,
                                         LorentzCorrection=self._lorentz,
                                         UnwrapRef=self._LRef,
                                         LowResRef=self._DIFCref,
                                         LowResSpectrumOffset=self._lowResTOFoffset,
                                         CropWavelengthMin=self._wavelengthMin,
                                         CropWavelengthMax=self._wavelengthMax,
                                         FrequencyLogNames=self.getProperty("FrequencyLogNames").value,
                                         WaveLengthLogNames=self.getProperty("WaveLengthLogNames").value,