savu.data.data_structures.plugin_data.PluginData.get_slice_dimension() - Code Metrics - Inspection of "scrutinizer config" - DiamondLightSource/Savu - Measure and Improve Code Quality continuously with Scrutinizer

Test Failed

Pull Request — master (#708)

by Daniil

created 2021-02-25 17:10 UTC

PluginData.get_slice_dimension() A

↳ Parent: savu.data.data_structures.plugin_data

Complexity

Conditions

Size

Total Lines	8
Code Lines	4

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	1
eloc	4
nop	1
dl	0
loc	8
rs	10
c	0
b	0
f	0

# Copyright 201i Diamond Light Source Ltd.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

"""
.. module:: plugin_data
   :platform: Unix
   :synopsis: Contains the PluginData class. Each Data set used in a plugin \
       has a PluginData object encapsulated within it, for the duration of a \
       plugin run.

.. moduleauthor:: Nicola Wadeson <[email protected]>

"""
import sys
import copy
import h5py
import logging
import numpy as np
from fractions import gcd

from savu.data.meta_data import MetaData
from savu.data.data_structures.data_add_ons import Padding


class PluginData(object):
    """ The PluginData class contains plugin specific information about a Data
    object for the duration of a plugin.  An instance of the class is
    encapsulated inside the Data object during the plugin run
    """

    def __init__(self, data_obj, plugin=None):
        self.data_obj = data_obj
        self._preview = None
        self.data_obj._set_plugin_data(self)
        self.meta_data = MetaData()
        self.padding = None
        self.pad_dict = None
        self.shape = None
        self.core_shape = None
        self.multi_params = {}
        self.extra_dims = []
        self._plugin = plugin
        self.fixed_dims = True
        self.split = None
        self.boundary_padding = None
        self.no_squeeze = False
        self.pre_tuning_shape = None
        self._frame_limit = None
        self._increase_rank = 0

    def _get_preview(self):
        return self._preview

    def get_total_frames(self):
        """ Get the total number of frames to process (all MPI processes).

        :returns: Number of frames
        :rtype: int
        """
        temp = 1
        slice_dir = \
            self.data_obj.get_data_patterns()[
                self.get_pattern_name()]["slice_dims"]
        for tslice in slice_dir:
            temp *= self.data_obj.get_shape()[tslice]
        return temp

    def __set_pattern(self, name, first_sdim=None):
        """ Set the pattern related information in the meta data dict.
        """
        pattern = self.data_obj.get_data_patterns()[name]
        self.meta_data.set("name", name)
        self.meta_data.set("core_dims", pattern['core_dims'])
        self.__set_slice_dimensions(first_sdim=first_sdim)

    def get_pattern_name(self):
        """ Get the pattern name.

        :returns: the pattern name
        :rtype: str
        """
        try:
            name = self.meta_data.get("name")
            return name
        except KeyError:
            raise Exception("The pattern name has not been set.")

    def get_pattern(self):
        """ Get the current pattern.

        :returns: dict of the pattern name against the pattern.
        :rtype: dict
        """
        pattern_name = self.get_pattern_name()
        return {pattern_name: self.data_obj.get_data_patterns()[pattern_name]}

    def __set_shape(self):
        """ Set the shape of the plugin data processing chunk.
        """
        core_dir = self.data_obj.get_core_dimensions()
        slice_dir = self.data_obj.get_slice_dimensions()
        dirs = list(set(core_dir + (slice_dir[0],)))
        slice_idx = dirs.index(slice_dir[0])
        dshape = self.data_obj.get_shape()
        shape = []
        for core in set(core_dir):
            shape.append(dshape[core])
        self.__set_core_shape(tuple(shape))

        mfp = self._get_max_frames_process()
        if mfp > 1 or self._get_no_squeeze():
            shape.insert(slice_idx, mfp)
        self.shape = tuple(shape)

    def _set_shape_transfer(self, slice_size):
        dshape = self.data_obj.get_shape()
        shape_before_tuning = self._get_shape_before_tuning()
        add = [1]*(len(dshape) - len(shape_before_tuning))
        slice_size = slice_size + add

        core_dir = self.data_obj.get_core_dimensions()
        slice_dir = self.data_obj.get_slice_dimensions()
        shape = [None]*len(dshape)
        for dim in core_dir:
            shape[dim] = dshape[dim]
        i = 0
        for dim in slice_dir:
            shape[dim] = slice_size[i]
            i += 1
        return tuple(shape)

    def __get_slice_size(self, mft):
        """ Calculate the number of frames transfer in each dimension given
            mft. """
        dshape = list(self.data_obj.get_shape())

        if 'fixed_dimensions' in list(self.meta_data.get_dictionary().keys()):
            fixed_dims = self.meta_data.get('fixed_dimensions')
            for d in fixed_dims:
                dshape[d] = 1

        dshape = [dshape[i] for i in self.meta_data.get('slice_dims')]
        size_list = [1]*len(dshape)
        i = 0

        while(mft > 1 and i < len(size_list)):
            size_list[i] = min(dshape[i], mft)
            mft //= np.prod(size_list) if np.prod(size_list) > 1 else 1
            i += 1

        self.meta_data.set('size_list', size_list)
        return size_list

    def set_bytes_per_frame(self):
        """ Return the size of a single frame in bytes. """
        nBytes = self.data_obj.get_itemsize()
        dims = list(self.get_pattern().values())[0]['core_dims']
        frame_shape = [self.data_obj.get_shape()[d] for d in dims]
        b_per_f = np.prod(frame_shape)*nBytes
        return frame_shape, b_per_f

    def get_shape(self):
        """ Get the shape of the data (without padding) that is passed to the
        plugin process_frames method.
        """
        return self.shape

    def _set_padded_shape(self):
        pass

    def get_padded_shape(self):
        """ Get the shape of the data (with padding) that is passed to the
        plugin process_frames method.
        """
        return self.shape

    def get_shape_transfer(self):
        """ Get the shape of the plugin data to be transferred each time.
        """
        return self.meta_data.get('transfer_shape')

    def __set_core_shape(self, shape):
        """ Set the core shape to hold only the shape of the core dimensions
        """
        self.core_shape = shape

    def get_core_shape(self):
        """ Get the shape of the core dimensions only.

        :returns: shape of core dimensions
        :rtype: tuple
        """
        return self.core_shape

    def _set_shape_before_tuning(self, shape):
        """ Set the shape of the full dataset used during each run of the \
        plugin (i.e. ignore extra dimensions due to parameter tuning). """
        self.pre_tuning_shape = shape

    def _get_shape_before_tuning(self):
        """ Return the shape of the full dataset used during each run of the \
        plugin (i.e. ignore extra dimensions due to parameter tuning). """
        return self.pre_tuning_shape if self.pre_tuning_shape else\
            self.data_obj.get_shape()

    def __check_dimensions(self, indices, core_dir, slice_dir, nDims):
        if len(indices) is not len(slice_dir):
            sys.exit("Incorrect number of indices specified when accessing "
                     "data.")

        if (len(core_dir)+len(slice_dir)) is not nDims:
            sys.exit("Incorrect number of data dimensions specified.")

    def __set_slice_dimensions(self, first_sdim=None):
        """ Set the slice dimensions in the pluginData meta data dictionary.\
        Reorder pattern slice_dims to ensure first_sdim is at the front.
        """
        pattern = self.data_obj.get_data_patterns()[self.get_pattern_name()]
        slice_dims = pattern['slice_dims']

        if first_sdim:
            slice_dims = list(slice_dims)
            first_sdim = \
                self.data_obj.get_data_dimension_by_axis_label(first_sdim)
            slice_dims.insert(0, slice_dims.pop(slice_dims.index(first_sdim)))
            pattern['slice_dims'] = tuple(slice_dims)

        self.meta_data.set('slice_dims', tuple(slice_dims))

    def get_slice_dimension(self):
        """
        Return the position of the slice dimension in relation to the data
        handed to the plugin.
        """
        core_dirs = self.data_obj.get_core_dimensions()
        slice_dir = self.data_obj.get_slice_dimensions()[0]
        return list(set(core_dirs + (slice_dir,))).index(slice_dir)

    def get_data_dimension_by_axis_label(self, label, contains=False):
        """
        Return the dimension of the data in the plugin that has the specified
        axis label.
        """
        label_dim = self.data_obj.get_data_dimension_by_axis_label(
                label, contains=contains)
        plugin_dims = self.data_obj.get_core_dimensions()
        if self._get_max_frames_process() > 1 or self.max_frames == 'multiple':
            plugin_dims += (self.get_slice_dimension(),)
        return list(set(plugin_dims)).index(label_dim)

    def set_slicing_order(self, order):  # should this function be deleted?
        """
        Reorder the slice dimensions.  The fastest changing slice dimension
        will always be the first one stated in the pattern key ``slice_dir``.
        The input param is a tuple stating the desired order of slicing
        dimensions relative to the current order.
        """
        slice_dirs = self.data_obj.get_slice_dimensions()
        if len(slice_dirs) < len(order):
            raise Exception("Incorrect number of dimensions specifed.")
        ordered = [slice_dirs[o] for o in order]
        remaining = [s for s in slice_dirs if s not in ordered]
        new_slice_dirs = tuple(ordered + remaining)
        self.get_pattern()['slice_dir'] = new_slice_dirs

    def get_core_dimensions(self):
        """
        Return the position of the core dimensions in relation to the data
        handed to the plugin.
        """
        core_dims = self.data_obj.get_core_dimensions()
        first_slice_dim = (self.data_obj.get_slice_dimensions()[0],)
        plugin_dims = np.sort(core_dims + first_slice_dim)
        return np.searchsorted(plugin_dims, np.sort(core_dims))

    def set_fixed_dimensions(self, dims, values):
        """ Fix a data direction to the index in values list.

        :param list(int) dims: Directions to fix
        :param list(int) value: Index of fixed directions
        """
        slice_dirs = self.data_obj.get_slice_dimensions()
        if set(dims).difference(set(slice_dirs)):
            raise Exception("You are trying to fix a direction that is not"
                            " a slicing direction")
        self.meta_data.set("fixed_dimensions", dims)
        self.meta_data.set("fixed_dimensions_values", values)
        self.__set_slice_dimensions()
        shape = list(self.data_obj.get_shape())
        for dim in dims:
            shape[dim] = 1
        self.data_obj.set_shape(tuple(shape))
        #self.__set_shape()

    def _get_fixed_dimensions(self):
        """ Get the fixed data directions and their indices

        :returns: Fixed directions and their associated values
        :rtype: list(list(int), list(int))
        """
        fixed = []
        values = []
        if 'fixed_dimensions' in self.meta_data.get_dictionary():
            fixed = self.meta_data.get("fixed_dimensions")
            values = self.meta_data.get("fixed_dimensions_values")
        return [fixed, values]

    def _get_data_slice_list(self, plist):
        """ Convert a plugin data slice list to a slice list for the whole
        dataset, i.e. add in any missing dimensions.
        """
        nDims = len(self.get_shape())
        all_dims = self.get_core_dimensions() + self.get_slice_dimension()
        extra_dims = all_dims[nDims:]
        dlist = list(plist)
        for i in extra_dims:
            dlist.insert(i, slice(None))
        return tuple(dlist)

    def _get_max_frames_process(self):
        """ Get the number of frames to process for each run of process_frames.

        If the number of frames is not divisible by the previewing ``chunk``
        value then amend the number of frames to gcd(frames, chunk)

        :returns: Number of frames to process
        :rtype: int
        """
        if self._plugin and self._plugin.chunk > 1:
            frame_chunk = self.meta_data.get("max_frames_process")
            chunk = self.data_obj.get_preview().get_starts_stops_steps(
                key='chunks')[self.get_slice_directions()[0]]
            self.meta_data.set('max_frames_process', gcd(frame_chunk, chunk))
        return self.meta_data.get("max_frames_process")

    def _get_max_frames_transfer(self):
        """ Get the number of frames to transfer for each run of
        process_frames. """
        return self.meta_data.get('max_frames_transfer')

    def _set_no_squeeze(self):
        self.no_squeeze = True

    def _get_no_squeeze(self):
        return self.no_squeeze
    
    def _set_rank_inc(self, n):
        """ Increase the rank of the array passed to the plugin by n.
        
        :param int n: Rank increment.
        """
        self._increase_rank = n
    
    def _get_rank_inc(self):
        """ Return the increased rank value
        
        :returns: Rank increment
        :rtype: int
        """
        return self._increase_rank

    def _set_meta_data(self):
        fixed, _ = self._get_fixed_dimensions()
        sdir = \
            [s for s in self.data_obj.get_slice_dimensions() if s not in fixed]
        shape = self.data_obj.get_shape()
        shape_before_tuning = self._get_shape_before_tuning()

        diff = len(shape) - len(shape_before_tuning)
        if diff:
            shape = shape_before_tuning
            sdir = sdir[:-diff]

        if 'fix_total_frames' in list(self.meta_data.get_dictionary().keys()):
            frames = self.meta_data.get('fix_total_frames')
        else:
            frames = np.prod([shape[d] for d in sdir])

        base_names = [p.__name__ for p in self._plugin.__class__.__bases__]
        processes = self.data_obj.exp.meta_data.get('processes')

        if 'GpuPlugin' in base_names:
            n_procs = len([n for n in processes if 'GPU' in n])
        else:
            n_procs = len(processes)

        # Fixing f_per_p to be just the first slice dimension for now due to
        # slow performance from HDF5 when not slicing multiple dimensions
        # concurrently
        #f_per_p = np.ceil(frames/n_procs)
        f_per_p = np.ceil(shape[sdir[0]]/n_procs)
        self.meta_data.set('shape', shape)
        self.meta_data.set('sdir', sdir)
        self.meta_data.set('total_frames', frames)
        self.meta_data.set('mpi_procs', n_procs)
        self.meta_data.set('frames_per_process', f_per_p)
        frame_shape, b_per_f = self.set_bytes_per_frame()
        self.meta_data.set('bytes_per_frame', b_per_f)
        self.meta_data.set('bytes_per_process', b_per_f*f_per_p)
        self.meta_data.set('frame_shape', frame_shape)

    def __log_max_frames(self, mft, mfp, check=True):
        logging.debug("Setting max frames transfer for plugin %s to %d" %
                      (self._plugin, mft))
        logging.debug("Setting max frames process for plugin %s to %d" %
                      (self._plugin, mfp))
        self.meta_data.set('max_frames_process', mfp)
        if check:
            self.__check_distribution(mft)
        # (((total_frames/mft)/mpi_procs) % 1)

    def __check_distribution(self, mft):
        warn_threshold = 0.85
        nprocs = self.meta_data.get('mpi_procs')
        nframes = self.meta_data.get('total_frames')
        temp = (((nframes/mft)/float(nprocs)) % 1)
        if temp != 0.0 and temp < warn_threshold:
            shape = self.meta_data.get('shape')
            sdir = self.meta_data.get('sdir')
            logging.warning('UNEVEN FRAME DISTRIBUTION: shape %s, nframes %s ' +
                         'sdir %s, nprocs %s', shape, nframes, sdir, nprocs)

    def _set_padding_dict(self):
        if self.padding and not isinstance(self.padding, Padding):
            self.pad_dict = copy.deepcopy(self.padding)
            self.padding = Padding(self)
            for key in list(self.pad_dict.keys()):
                getattr(self.padding, key)(self.pad_dict[key])

    def plugin_data_setup(self, pattern, nFrames, split=None, slice_axis=None,
                          getall=None):
        """ Setup the PluginData object.

        :param str pattern: A pattern name
        :param int nFrames: How many frames to process at a time.  Choose from
            'single', 'multiple', 'fixed_multiple' or an integer (an integer
            should only ever be passed in exceptional circumstances)
        :keyword str slice_axis: An axis label associated with the fastest
            changing (first) slice dimension.
        :keyword list[pattern, axis_label] getall: A list of two values.  If
        the requested pattern doesn't exist then use all of "axis_label"
        dimension of "pattern" as this is equivalent to one slice of the
        original pattern.
        """

        if pattern not in self.data_obj.get_data_patterns() and getall:
            pattern, nFrames = self.__set_getall_pattern(getall, nFrames)

        # slice_axis is first slice dimension
        self.__set_pattern(pattern, first_sdim=slice_axis)
        if isinstance(nFrames, list):
            nFrames, self._frame_limit = nFrames
        self.max_frames = nFrames
        self.split = split

    def __set_getall_pattern(self, getall, nFrames):
        """ Set framework changes required to get all of a pattern of lower
        rank.
        """
        pattern, slice_axis = getall
        dim = self.data_obj.get_data_dimension_by_axis_label(slice_axis)
        # ensure data remains the same shape when 'getall' dim has length 1
        self._set_no_squeeze()
        if nFrames == 'multiple' or (isinstance(nFrames, int) and nFrames > 1):
            self._set_rank_inc(1)
        nFrames = self.data_obj.get_shape()[dim]
        return pattern, nFrames

    def plugin_data_transfer_setup(self, copy=None, calc=None):
        """ Set up the plugin data transfer frame parameters.
        If copy=pData (another PluginData instance) then copy """
        chunks = \
            self.data_obj.get_preview().get_starts_stops_steps(key='chunks')

        if not copy and not calc:
            mft, mft_shape, mfp = self._calculate_max_frames()
        elif calc:
            max_mft = calc.meta_data.get('max_frames_transfer')
            max_mfp = calc.meta_data.get('max_frames_process')
            max_nProc = int(np.ceil(max_mft/float(max_mfp)))
            nProc = max_nProc
            mfp = 1 if self.max_frames == 'single' else self.max_frames
            mft = nProc*mfp
            mft_shape = self._set_shape_transfer(self.__get_slice_size(mft))
        elif copy:
            mft = copy._get_max_frames_transfer()
            mft_shape = self._set_shape_transfer(self.__get_slice_size(mft))
            mfp = copy._get_max_frames_process()

        self.__set_max_frames(mft, mft_shape, mfp)


        if self._plugin and mft \
                and (chunks[self.data_obj.get_slice_dimensions()[0]] % mft):
            self._plugin.chunk = True
        self.__set_shape()

    def _calculate_max_frames(self):
        nFrames = self.max_frames
        self.__perform_checks(nFrames)
        td = self.data_obj._get_transport_data()
        mft, size_list = td._calc_max_frames_transfer(nFrames)
        self.meta_data.set('size_list', size_list)
        mfp = td._calc_max_frames_process(nFrames)
        if mft:
            mft_shape = self._set_shape_transfer(list(size_list))
        return mft, mft_shape, mfp


    def __set_max_frames(self, mft, mft_shape, mfp):
        self.meta_data.set('max_frames_transfer', mft)
        self.meta_data.set('transfer_shape', mft_shape)
        self.meta_data.set('max_frames_process', mfp)
        self.__log_max_frames(mft, mfp)
        # Retain the shape if the first slice dimension has length 1
        if mfp == 1 and self.max_frames == 'multiple':
            self._set_no_squeeze()

    def _get_plugin_data_size_params(self):
        nBytes = self.data_obj.get_itemsize()
        frame_shape = self.meta_data.get('frame_shape')
        total_frames = self.meta_data.get('total_frames')
        tbytes = nBytes*np.prod(frame_shape)*total_frames

        params = {'nBytes': nBytes, 'frame_shape': frame_shape,
                  'total_frames': total_frames, 'transfer_bytes': tbytes}
        return params

    def __perform_checks(self, nFrames):
        options = ['single', 'multiple']
        if not np.issubdtype(type(nFrames), np.int64) and nFrames not in options:
            e_str = ("The value of nFrames is not recognised.  Please choose "
            + "from 'single' and 'multiple' (or an integer in exceptional "
            + "circumstances).")
            raise Exception(e_str)

    def get_frame_limit(self):
        return self._frame_limit

    def get_current_frame_idx(self):
        """ Returns the index of the frames currently being processed.
        """
        global_index = self._plugin.get_global_frame_index()
        count = self._plugin.get_process_frames_counter()
        mfp = self.meta_data.get('max_frames_process')
        start = global_index[count]*mfp
        index = np.arange(start, start + mfp)
        nFrames = self.get_total_frames()
        index[index >= nFrames] = nFrames - 1
        return index


1			# Copyright 201i Diamond Light Source Ltd.
2			#
3			# Licensed under the Apache License, Version 2.0 (the "License");
4			# you may not use this file except in compliance with the License.
5			# You may obtain a copy of the License at
6			#
7			# http://www.apache.org/licenses/LICENSE-2.0
8			#
9			# Unless required by applicable law or agreed to in writing, software
10			# distributed under the License is distributed on an "AS IS" BASIS,
11			# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12			# See the License for the specific language governing permissions and
13			# limitations under the License.
14
15			"""
16			.. module:: plugin_data
17			:platform: Unix
18			:synopsis: Contains the PluginData class. Each Data set used in a plugin \
19			has a PluginData object encapsulated within it, for the duration of a \
20			plugin run.
21
22			.. moduleauthor:: Nicola Wadeson <[email protected]>
23
24			"""
25			import sys
26			import copy
27			import h5py
28			import logging
29			import numpy as np
30			from fractions import gcd
31
32			from savu.data.meta_data import MetaData
33			from savu.data.data_structures.data_add_ons import Padding
34
35
36			class PluginData(object):
37			""" The PluginData class contains plugin specific information about a Data
38			object for the duration of a plugin. An instance of the class is
39			encapsulated inside the Data object during the plugin run
40			"""
41
42			def __init__(self, data_obj, plugin=None):
43			self.data_obj = data_obj
44			self._preview = None
45			self.data_obj._set_plugin_data(self)
46			self.meta_data = MetaData()
47			self.padding = None
48			self.pad_dict = None
49			self.shape = None
50			self.core_shape = None
51			self.multi_params = {}
52			self.extra_dims = []
53			self._plugin = plugin
54			self.fixed_dims = True
55			self.split = None
56			self.boundary_padding = None
57			self.no_squeeze = False
58			self.pre_tuning_shape = None
59			self._frame_limit = None
60			self._increase_rank = 0
61
62			def _get_preview(self):
63			return self._preview
64
65			def get_total_frames(self):
66			""" Get the total number of frames to process (all MPI processes).
67
68			:returns: Number of frames
69			:rtype: int
70			"""
71			temp = 1
72			slice_dir = \
73			self.data_obj.get_data_patterns()[
74			self.get_pattern_name()]["slice_dims"]
75			for tslice in slice_dir:
76			temp *= self.data_obj.get_shape()[tslice]
77			return temp
78
79			def __set_pattern(self, name, first_sdim=None):
80			""" Set the pattern related information in the meta data dict.
81			"""
82			pattern = self.data_obj.get_data_patterns()[name]
83			self.meta_data.set("name", name)
84			self.meta_data.set("core_dims", pattern['core_dims'])
85			self.__set_slice_dimensions(first_sdim=first_sdim)
86
87			def get_pattern_name(self):
88			""" Get the pattern name.
89
90			:returns: the pattern name
91			:rtype: str
92			"""
93			try:
94			name = self.meta_data.get("name")
95			return name
96			except KeyError:
97			raise Exception("The pattern name has not been set.")
98
99			def get_pattern(self):
100			""" Get the current pattern.
101
102			:returns: dict of the pattern name against the pattern.
103			:rtype: dict
104			"""
105			pattern_name = self.get_pattern_name()
106			return {pattern_name: self.data_obj.get_data_patterns()[pattern_name]}
107
108			def __set_shape(self):
109			""" Set the shape of the plugin data processing chunk.
110			"""
111			core_dir = self.data_obj.get_core_dimensions()
112			slice_dir = self.data_obj.get_slice_dimensions()
113			dirs = list(set(core_dir + (slice_dir[0],)))
114			slice_idx = dirs.index(slice_dir[0])
115			dshape = self.data_obj.get_shape()
116			shape = []
117			for core in set(core_dir):
118			shape.append(dshape[core])
119			self.__set_core_shape(tuple(shape))
120
121			mfp = self._get_max_frames_process()
122			if mfp > 1 or self._get_no_squeeze():
123			shape.insert(slice_idx, mfp)
124			self.shape = tuple(shape)
125
126			def _set_shape_transfer(self, slice_size):
127			dshape = self.data_obj.get_shape()
128			shape_before_tuning = self._get_shape_before_tuning()
129			add = [1]*(len(dshape) - len(shape_before_tuning))
130			slice_size = slice_size + add
131
132			core_dir = self.data_obj.get_core_dimensions()
133			slice_dir = self.data_obj.get_slice_dimensions()
134			shape = [None]*len(dshape)
135			for dim in core_dir:
136			shape[dim] = dshape[dim]
137			i = 0
138			for dim in slice_dir:
139			shape[dim] = slice_size[i]
140			i += 1
141			return tuple(shape)
142
143			def __get_slice_size(self, mft):
144			""" Calculate the number of frames transfer in each dimension given
145			mft. """
146			dshape = list(self.data_obj.get_shape())
147
148			if 'fixed_dimensions' in list(self.meta_data.get_dictionary().keys()):
149			fixed_dims = self.meta_data.get('fixed_dimensions')
150			for d in fixed_dims:
151			dshape[d] = 1
152
153			dshape = [dshape[i] for i in self.meta_data.get('slice_dims')]
154			size_list = [1]*len(dshape)
155			i = 0
156
157			while(mft > 1 and i < len(size_list)):
158			size_list[i] = min(dshape[i], mft)
159			mft //= np.prod(size_list) if np.prod(size_list) > 1 else 1
160			i += 1
161
162			self.meta_data.set('size_list', size_list)
163			return size_list
164
165			def set_bytes_per_frame(self):
166			""" Return the size of a single frame in bytes. """
167			nBytes = self.data_obj.get_itemsize()
168			dims = list(self.get_pattern().values())[0]['core_dims']
169			frame_shape = [self.data_obj.get_shape()[d] for d in dims]
170			b_per_f = np.prod(frame_shape)*nBytes
171			return frame_shape, b_per_f
172
173			def get_shape(self):
174			""" Get the shape of the data (without padding) that is passed to the
175			plugin process_frames method.
176			"""
177			return self.shape
178
179			def _set_padded_shape(self):
180			pass
181
182			def get_padded_shape(self):
183			""" Get the shape of the data (with padding) that is passed to the
184			plugin process_frames method.
185			"""
186			return self.shape
187
188			def get_shape_transfer(self):
189			""" Get the shape of the plugin data to be transferred each time.
190			"""
191			return self.meta_data.get('transfer_shape')
192
193			def __set_core_shape(self, shape):
194			""" Set the core shape to hold only the shape of the core dimensions
195			"""
196			self.core_shape = shape
197
198			def get_core_shape(self):
199			""" Get the shape of the core dimensions only.
200
201			:returns: shape of core dimensions
202			:rtype: tuple
203			"""
204			return self.core_shape
205
206			def _set_shape_before_tuning(self, shape):
207			""" Set the shape of the full dataset used during each run of the \
208			plugin (i.e. ignore extra dimensions due to parameter tuning). """
209			self.pre_tuning_shape = shape
210
211			def _get_shape_before_tuning(self):
212			""" Return the shape of the full dataset used during each run of the \
213			plugin (i.e. ignore extra dimensions due to parameter tuning). """
214			return self.pre_tuning_shape if self.pre_tuning_shape else\
215			self.data_obj.get_shape()
216
217			def __check_dimensions(self, indices, core_dir, slice_dir, nDims):
218			if len(indices) is not len(slice_dir):
219			sys.exit("Incorrect number of indices specified when accessing "
220			"data.")
221
222			if (len(core_dir)+len(slice_dir)) is not nDims:
223			sys.exit("Incorrect number of data dimensions specified.")
224
225			def __set_slice_dimensions(self, first_sdim=None):
226			""" Set the slice dimensions in the pluginData meta data dictionary.\
227			Reorder pattern slice_dims to ensure first_sdim is at the front.
228			"""
229			pattern = self.data_obj.get_data_patterns()[self.get_pattern_name()]
230			slice_dims = pattern['slice_dims']
231
232			if first_sdim:
233			slice_dims = list(slice_dims)
234			first_sdim = \
235			self.data_obj.get_data_dimension_by_axis_label(first_sdim)
236			slice_dims.insert(0, slice_dims.pop(slice_dims.index(first_sdim)))
237			pattern['slice_dims'] = tuple(slice_dims)
238
239			self.meta_data.set('slice_dims', tuple(slice_dims))
240
241			def get_slice_dimension(self):
242			"""
243			Return the position of the slice dimension in relation to the data
244			handed to the plugin.
245			"""
246			core_dirs = self.data_obj.get_core_dimensions()
247			slice_dir = self.data_obj.get_slice_dimensions()[0]
248			return list(set(core_dirs + (slice_dir,))).index(slice_dir)
249
250			def get_data_dimension_by_axis_label(self, label, contains=False):
251			"""
252			Return the dimension of the data in the plugin that has the specified
253			axis label.
254			"""
255			label_dim = self.data_obj.get_data_dimension_by_axis_label(
256			label, contains=contains)
257			plugin_dims = self.data_obj.get_core_dimensions()
258			if self._get_max_frames_process() > 1 or self.max_frames == 'multiple':
259			plugin_dims += (self.get_slice_dimension(),)
260			return list(set(plugin_dims)).index(label_dim)
261
262			def set_slicing_order(self, order): # should this function be deleted?
263			"""
264			Reorder the slice dimensions. The fastest changing slice dimension
265			will always be the first one stated in the pattern key ``slice_dir``.
266			The input param is a tuple stating the desired order of slicing
267			dimensions relative to the current order.
268			"""
269			slice_dirs = self.data_obj.get_slice_dimensions()
270			if len(slice_dirs) < len(order):
271			raise Exception("Incorrect number of dimensions specifed.")
272			ordered = [slice_dirs[o] for o in order]
273			remaining = [s for s in slice_dirs if s not in ordered]
274			new_slice_dirs = tuple(ordered + remaining)
275			self.get_pattern()['slice_dir'] = new_slice_dirs
276
277			def get_core_dimensions(self):
278			"""
279			Return the position of the core dimensions in relation to the data
280			handed to the plugin.
281			"""
282			core_dims = self.data_obj.get_core_dimensions()
283			first_slice_dim = (self.data_obj.get_slice_dimensions()[0],)
284			plugin_dims = np.sort(core_dims + first_slice_dim)
285			return np.searchsorted(plugin_dims, np.sort(core_dims))
286
287			def set_fixed_dimensions(self, dims, values):
288			""" Fix a data direction to the index in values list.
289
290			:param list(int) dims: Directions to fix
291			:param list(int) value: Index of fixed directions
292			"""
293			slice_dirs = self.data_obj.get_slice_dimensions()
294			if set(dims).difference(set(slice_dirs)):
295			raise Exception("You are trying to fix a direction that is not"
296			" a slicing direction")
297			self.meta_data.set("fixed_dimensions", dims)
298			self.meta_data.set("fixed_dimensions_values", values)
299			self.__set_slice_dimensions()
300			shape = list(self.data_obj.get_shape())
301			for dim in dims:
302			shape[dim] = 1
303			self.data_obj.set_shape(tuple(shape))
304			#self.__set_shape()
305
306			def _get_fixed_dimensions(self):
307			""" Get the fixed data directions and their indices
308
309			:returns: Fixed directions and their associated values
310			:rtype: list(list(int), list(int))
311			"""
312			fixed = []
313			values = []
314			if 'fixed_dimensions' in self.meta_data.get_dictionary():
315			fixed = self.meta_data.get("fixed_dimensions")
316			values = self.meta_data.get("fixed_dimensions_values")
317			return [fixed, values]
318
319			def _get_data_slice_list(self, plist):
320			""" Convert a plugin data slice list to a slice list for the whole
321			dataset, i.e. add in any missing dimensions.
322			"""
323			nDims = len(self.get_shape())
324			all_dims = self.get_core_dimensions() + self.get_slice_dimension()
325			extra_dims = all_dims[nDims:]
326			dlist = list(plist)
327			for i in extra_dims:
328			dlist.insert(i, slice(None))
329			return tuple(dlist)
330
331			def _get_max_frames_process(self):
332			""" Get the number of frames to process for each run of process_frames.
333
334			If the number of frames is not divisible by the previewing ``chunk``
335			value then amend the number of frames to gcd(frames, chunk)
336
337			:returns: Number of frames to process
338			:rtype: int
339			"""
340			if self._plugin and self._plugin.chunk > 1:
341			frame_chunk = self.meta_data.get("max_frames_process")
342			chunk = self.data_obj.get_preview().get_starts_stops_steps(
343			key='chunks')[self.get_slice_directions()[0]]
344			self.meta_data.set('max_frames_process', gcd(frame_chunk, chunk))
345			return self.meta_data.get("max_frames_process")
346
347			def _get_max_frames_transfer(self):
348			""" Get the number of frames to transfer for each run of
349			process_frames. """
350			return self.meta_data.get('max_frames_transfer')
351
352			def _set_no_squeeze(self):
353			self.no_squeeze = True
354
355			def _get_no_squeeze(self):
356			return self.no_squeeze
357
358			def _set_rank_inc(self, n):
359			""" Increase the rank of the array passed to the plugin by n.
360
361			:param int n: Rank increment.
362			"""
363			self._increase_rank = n
364
365			def _get_rank_inc(self):
366			""" Return the increased rank value
367
368			:returns: Rank increment
369			:rtype: int
370			"""
371			return self._increase_rank
372
373			def _set_meta_data(self):
374			fixed, _ = self._get_fixed_dimensions()
375			sdir = \
376			[s for s in self.data_obj.get_slice_dimensions() if s not in fixed]
377			shape = self.data_obj.get_shape()
378			shape_before_tuning = self._get_shape_before_tuning()
379
380			diff = len(shape) - len(shape_before_tuning)
381			if diff:
382			shape = shape_before_tuning
383			sdir = sdir[:-diff]
384
385			if 'fix_total_frames' in list(self.meta_data.get_dictionary().keys()):
386			frames = self.meta_data.get('fix_total_frames')
387			else:
388			frames = np.prod([shape[d] for d in sdir])
389
390			base_names = [p.__name__ for p in self._plugin.__class__.__bases__]
391			processes = self.data_obj.exp.meta_data.get('processes')
392
393			if 'GpuPlugin' in base_names:
394			n_procs = len([n for n in processes if 'GPU' in n])
395			else:
396			n_procs = len(processes)
397
398			# Fixing f_per_p to be just the first slice dimension for now due to
399			# slow performance from HDF5 when not slicing multiple dimensions
400			# concurrently
401			#f_per_p = np.ceil(frames/n_procs)
402			f_per_p = np.ceil(shape[sdir[0]]/n_procs)
403			self.meta_data.set('shape', shape)
404			self.meta_data.set('sdir', sdir)
405			self.meta_data.set('total_frames', frames)
406			self.meta_data.set('mpi_procs', n_procs)
407			self.meta_data.set('frames_per_process', f_per_p)
408			frame_shape, b_per_f = self.set_bytes_per_frame()
409			self.meta_data.set('bytes_per_frame', b_per_f)
410			self.meta_data.set('bytes_per_process', b_per_f*f_per_p)
411			self.meta_data.set('frame_shape', frame_shape)
412
413			def __log_max_frames(self, mft, mfp, check=True):
414			logging.debug("Setting max frames transfer for plugin %s to %d" %
415			(self._plugin, mft))
416			logging.debug("Setting max frames process for plugin %s to %d" %
417			(self._plugin, mfp))
418			self.meta_data.set('max_frames_process', mfp)
419			if check:
420			self.__check_distribution(mft)
421			# (((total_frames/mft)/mpi_procs) % 1)
422
423			def __check_distribution(self, mft):
424			warn_threshold = 0.85
425			nprocs = self.meta_data.get('mpi_procs')
426			nframes = self.meta_data.get('total_frames')
427			temp = (((nframes/mft)/float(nprocs)) % 1)
428			if temp != 0.0 and temp < warn_threshold:
429			shape = self.meta_data.get('shape')
430			sdir = self.meta_data.get('sdir')
431			logging.warning('UNEVEN FRAME DISTRIBUTION: shape %s, nframes %s ' +
432			'sdir %s, nprocs %s', shape, nframes, sdir, nprocs)
433
434			def _set_padding_dict(self):
435			if self.padding and not isinstance(self.padding, Padding):
436			self.pad_dict = copy.deepcopy(self.padding)
437			self.padding = Padding(self)
438			for key in list(self.pad_dict.keys()):
439			getattr(self.padding, key)(self.pad_dict[key])
440
441			def plugin_data_setup(self, pattern, nFrames, split=None, slice_axis=None,
442			getall=None):
443			""" Setup the PluginData object.
444
445			:param str pattern: A pattern name
446			:param int nFrames: How many frames to process at a time. Choose from
447			'single', 'multiple', 'fixed_multiple' or an integer (an integer
448			should only ever be passed in exceptional circumstances)
449			:keyword str slice_axis: An axis label associated with the fastest
450			changing (first) slice dimension.
451			:keyword list[pattern, axis_label] getall: A list of two values. If
452			the requested pattern doesn't exist then use all of "axis_label"
453			dimension of "pattern" as this is equivalent to one slice of the
454			original pattern.
455			"""
456
457			if pattern not in self.data_obj.get_data_patterns() and getall:
458			pattern, nFrames = self.__set_getall_pattern(getall, nFrames)
459
460			# slice_axis is first slice dimension
461			self.__set_pattern(pattern, first_sdim=slice_axis)
462			if isinstance(nFrames, list):
463			nFrames, self._frame_limit = nFrames
464			self.max_frames = nFrames
465			self.split = split
466
467			def __set_getall_pattern(self, getall, nFrames):
468			""" Set framework changes required to get all of a pattern of lower
469			rank.
470			"""
471			pattern, slice_axis = getall
472			dim = self.data_obj.get_data_dimension_by_axis_label(slice_axis)
473			# ensure data remains the same shape when 'getall' dim has length 1
474			self._set_no_squeeze()
475			if nFrames == 'multiple' or (isinstance(nFrames, int) and nFrames > 1):
476			self._set_rank_inc(1)
477			nFrames = self.data_obj.get_shape()[dim]
478			return pattern, nFrames
479
480			def plugin_data_transfer_setup(self, copy=None, calc=None):
481			""" Set up the plugin data transfer frame parameters.
482			If copy=pData (another PluginData instance) then copy """
483			chunks = \
484			self.data_obj.get_preview().get_starts_stops_steps(key='chunks')
485
486			if not copy and not calc:
487			mft, mft_shape, mfp = self._calculate_max_frames()
488			elif calc:
489			max_mft = calc.meta_data.get('max_frames_transfer')
490			max_mfp = calc.meta_data.get('max_frames_process')
491			max_nProc = int(np.ceil(max_mft/float(max_mfp)))
492			nProc = max_nProc
493			mfp = 1 if self.max_frames == 'single' else self.max_frames
494			mft = nProc*mfp
495			mft_shape = self._set_shape_transfer(self.__get_slice_size(mft))
496			elif copy:
497			mft = copy._get_max_frames_transfer()
498			mft_shape = self._set_shape_transfer(self.__get_slice_size(mft))
499			mfp = copy._get_max_frames_process()
500
501			self.__set_max_frames(mft, mft_shape, mfp)
			0 ignored issues – show introduced 2021-02-25 17:14 UTC by Report Bug Copy Issue Report The variable `mft_shape` does not seem to be defined for all execution paths. Loading history... introduced 2021-02-25 17:14 UTC by Report Bug Copy Issue Report The variable `mft` does not seem to be defined for all execution paths. Loading history... introduced 2021-02-25 17:14 UTC by Report Bug Copy Issue Report The variable `mfp` does not seem to be defined for all execution paths. Loading history...
502
503			if self._plugin and mft \
504			and (chunks[self.data_obj.get_slice_dimensions()[0]] % mft):
505			self._plugin.chunk = True
506			self.__set_shape()
507
508			def _calculate_max_frames(self):
509			nFrames = self.max_frames
510			self.__perform_checks(nFrames)
511			td = self.data_obj._get_transport_data()
512			mft, size_list = td._calc_max_frames_transfer(nFrames)
513			self.meta_data.set('size_list', size_list)
514			mfp = td._calc_max_frames_process(nFrames)
515			if mft:
516			mft_shape = self._set_shape_transfer(list(size_list))
517			return mft, mft_shape, mfp
			0 ignored issues – show introduced 2021-02-25 17:14 UTC by Report Bug Copy Issue Report The variable `mft_shape` does not seem to be defined in case `mft` on line `515` is `False`. Are you sure this can never be the case? Loading history...
518
519			def __set_max_frames(self, mft, mft_shape, mfp):
520			self.meta_data.set('max_frames_transfer', mft)
521			self.meta_data.set('transfer_shape', mft_shape)
522			self.meta_data.set('max_frames_process', mfp)
523			self.__log_max_frames(mft, mfp)
524			# Retain the shape if the first slice dimension has length 1
525			if mfp == 1 and self.max_frames == 'multiple':
526			self._set_no_squeeze()
527
528			def _get_plugin_data_size_params(self):
529			nBytes = self.data_obj.get_itemsize()
530			frame_shape = self.meta_data.get('frame_shape')
531			total_frames = self.meta_data.get('total_frames')
532			tbytes = nBytesnp.prod(frame_shape)total_frames
533
534			params = {'nBytes': nBytes, 'frame_shape': frame_shape,
535			'total_frames': total_frames, 'transfer_bytes': tbytes}
536			return params
537
538			def __perform_checks(self, nFrames):
539			options = ['single', 'multiple']
540			if not np.issubdtype(type(nFrames), np.int64) and nFrames not in options:
541			e_str = ("The value of nFrames is not recognised. Please choose "
542			+ "from 'single' and 'multiple' (or an integer in exceptional "
543			+ "circumstances).")
544			raise Exception(e_str)
545
546			def get_frame_limit(self):
547			return self._frame_limit
548
549			def get_current_frame_idx(self):
550			""" Returns the index of the frames currently being processed.
551			"""
552			global_index = self._plugin.get_global_frame_index()
553			count = self._plugin.get_process_frames_counter()
554			mfp = self.meta_data.get('max_frames_process')
555			start = global_index[count]*mfp
556			index = np.arange(start, start + mfp)
557			nFrames = self.get_total_frames()
558			index[index >= nFrames] = nFrames - 1
559			return index
560

DiamondLightSource / Savu

Pull Request — master (#708)

PluginData.get_slice_dimension() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like