savu.plugins.plugin_datasets.PluginDatasets._set_in_dataset_names() - Code Metrics - DiamondLightSource/Savu - Measure and Improve Code Quality continuously with Scrutinizer

PluginDatasets._set_in_dataset_names() A
last analyzed 2022-08-17 15:44 UTC

↳ Parent: savu.plugins.plugin_datasets

Complexity

Conditions

Size

Total Lines	7
Code Lines	7

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	3
eloc	7
nop	2
dl	0
loc	7
rs	10
c	0
b	0
f	0

# Copyright 2014 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_datasets
   :platform: Unix
   :synopsis: Base class of plugin containing all dataset related functions

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

"""

import copy
import numpy as np

import savu.plugins.plugin_datasets_notes as notes
from savu.core.utils import docstring_parameter
from savu.data.data_structures.plugin_data import PluginData


class PluginDatasets(object):

    def __init__(self, *args, **kwargs):
        super(PluginDatasets, self).__init__()
        self.exp = None
        self.data_objs = {}
        self.variable_data_flag = False
        self._max_itemsize = 0

    def __get_data_objects(self, dtype):
        """ Get the data objects associated with the plugin from the experiment
        data index.

        :params str dtype: "in_data" or "out_data"
        :returns: A list of data objects with the names given in
            ``self.parameters``
        :rtype: list(Data)
        """
        data_list = self.parameters[dtype + 'sets']
        data_objs = []
        for data in data_list:
            data_obj = self.exp.index[dtype][data]
            data_objs.append(data_obj)
        return data_objs

    def _clone_datasets(self):
        for data_obj in list(self.exp.index['out_data'].values()):
            if data_obj.raw and data_obj.data:
                data_obj.raw.create_next_instance(data_obj)
#                data_obj.clone = True

    def _finalise_datasets(self):
        in_data, out_data = self.get_datasets()
        for data in in_data + out_data:
            data._finalise_patterns()

    def _finalise_plugin_datasets(self):
        if 'dawn_runner' in list(self.exp.meta_data.get_dictionary().keys()):
            return

        in_pData, out_pData = self.get_plugin_datasets()

        params = {}
        for pData in in_pData + out_pData:
            pData._set_meta_data()
            params[pData] = pData._get_plugin_data_size_params()

        max_bytes = 0
        for key, value in params.items():
            if value['transfer_bytes'] > max_bytes:
                max_data = key
                max_bytes = value['transfer_bytes']

        # set mft and mfp for the largest dataset
        max_data.plugin_data_transfer_setup()

        to_set = list(set(params.keys()).difference(set([max_data])))

        for pData in to_set:
            if params[pData]['total_frames'] == params[max_data]['total_frames']:
                pData.plugin_data_transfer_setup(copy=max_data)
            else:
                if pData.max_frames == 'multiple':
                    msg = "If a plugin reduces the number of frames, the " \
                        "number of frames cannot be 'multiple'."
                    raise Exception(msg)
                pData.plugin_data_transfer_setup(calc=max_data)

    def __set_in_datasets(self):
        """ Set the in_data objects.

        :returns: the in_datasets associated with the plugin.
        :rtype: list[Data]
        """
        return self.__get_data_objects('in_data')

    def __set_out_datasets(self):
        """ Set the out_data objects.

        If the out_datasets do not exist inside the experiment then create
        them.

        :returns: the out_datasets associated with the plugin.
        :rtype: list[Data]
        """
        try:
            out_data = self.__get_data_objects('out_data')
        except KeyError:
            out_data = []
            for data in self.parameters['out_datasets']:
                self.exp.create_data_object("out_data", data)
            out_data = self.__get_data_objects('out_data')
        for data in out_data:
            data.extra_dims = self.get_plugin_tools().get_extra_dims()
        return out_data

    def _get_plugin_data(self, data_list):
        """ Encapsulate a PluginData object in each dataset associated with
        the plugin.

        :params list(Data) data_list: A list of Data objects used in a plugin.
        :returns: A list of PluginData objects.
        :rtype: list(PluginData)
        """
        pData_list = []
        ptools = self.get_plugin_tools()
        used = set()
        unique_data_list = \
            [x for x in data_list if x not in used and (used.add(x) or True)]
        for data in unique_data_list:
            pData_list.append(PluginData(data, self))
            pData_list[-1].extra_dims = ptools.get_extra_dims()
            pData_list[-1].multi_params_dict = ptools.get_multi_params_dict()
        return pData_list

    def _set_plugin_dataset_names(self):
        """ Fill in empty in/out_dataset entries with default values.
        """
        params = self.parameters
        orig_in = copy.copy(params['in_datasets'])
        in_names = self._set_in_dataset_names(params)
        # case that an extra in_dataset is added in the plugin
        in_names = orig_in if len(orig_in) and \
            len(in_names) > len(orig_in) else in_names
        self._set_out_dataset_names(params, in_names)
        # update the entry in the process list
        data_dict = {'in_datasets': params['in_datasets'],
                      'out_datasets': params['out_datasets']}
        idx = self.exp.meta_data.get('nPlugin')
        self.exp.meta_data.plugin_list._update_datasets(idx, data_dict)

    def _set_in_dataset_names(self, params):
        dIn = params['in_datasets']
        dIn = dIn if isinstance(dIn, list) else [dIn]
        dIn = self.exp._set_all_datasets('in_data') if len(dIn) ==0 else dIn
        params['in_datasets'] = dIn
        nIn = self.nInput_datasets() # datasets many be added dynamically here
        return self.check_nDatasets(params['in_datasets'], nIn, 'in_data')

    def _set_out_dataset_names(self, params, in_names):
        dOut = params['out_datasets'] if 'out_datasets' in params.keys() else []
        dOut = dOut if isinstance(dOut, list) else [dOut]
        dOut = (copy.copy(in_names) if len(dOut) == 0 else dOut)
        clones = self.nClone_datasets()
        params['out_datasets'] = dOut
        nOut = self.nOutput_datasets()
        names = self.check_nDatasets(params['out_datasets'], nOut,
                                     "out_data", clones=clones)
        if clones:
            dOut.extend(['itr_clone' + str(i) for i in range(clones)])
    
        for i in range(len(names)):
            new = names[i].split('in_datasets')
            if len(new) == 2:
                names[i] = in_names[int(list(new[1])[1])]
        params["out_datasets"] = names
        return names

    def check_nDatasets(self, names, nSets, dtype, clones=0):
        nSets = len(self.parameters[dtype + 'sets']) if nSets=='var' else nSets
        if len(names) is not (nSets - clones):
            if nSets == 0:
                names = []
            else:
                msg = "ERROR: Broken plugin chain. \n Please name the %s %s " \
                "sets associated with the plugin %s in the process file." % \
                (str(nSets), dtype, self.name)
                raise Exception(msg)
        return names

    def _set_plugin_datasets(self):
        """ Populate ``self.parameters`` in/out_datasets and
        plugin_in/out_datasets with the relevant objects (Data or PluginData).
        """
        if not self.exp._get_dataset_names_complete():
            self._set_plugin_dataset_names()
        self.parameters['in_datasets'] = self.__set_in_datasets()
        self.parameters['out_datasets'] = self.__set_out_datasets()
        self.parameters['plugin_in_datasets'] = \
            self._get_plugin_data(self.parameters['in_datasets'])
        self.parameters['plugin_out_datasets'] = \
            self._get_plugin_data(self.parameters['out_datasets'])

    @docstring_parameter('PluginData', 'in')
    @docstring_parameter(notes.datasets_notes.__doc__)
    def get_plugin_in_datasets(self):
        """ {0} """
        return self.parameters['plugin_in_datasets']

    @docstring_parameter('PluginData', 'out')
    @docstring_parameter(notes.datasets_notes.__doc__)
    def get_plugin_out_datasets(self):
        """ {0} """
        return self.parameters['plugin_out_datasets']

    @docstring_parameter("PluginData")
    @docstring_parameter(notes.two_datasets_notes.__doc__)
    def get_plugin_datasets(self):
        """ {0} """
        return self.get_plugin_in_datasets(), self.get_plugin_out_datasets()

    @docstring_parameter("Data", "in")
    @docstring_parameter(notes.datasets_notes.__doc__)
    def get_in_datasets(self):
        """ {0} """
        return self.parameters['in_datasets']

    @docstring_parameter("Data", "out")
    @docstring_parameter(notes.datasets_notes.__doc__)
    def get_out_datasets(self):
        """ {0} """
        return self.parameters['out_datasets']

    @docstring_parameter("PluginData")
    @docstring_parameter(notes.two_datasets_notes.__doc__)
    def get_datasets(self):
        """ {0} """
        return self.get_in_datasets(), self.get_out_datasets()

    @docstring_parameter("in")
    @docstring_parameter(notes.mData_notes.__doc__)
    def get_in_meta_data(self):
        """ {0} """
        return self.__set_meta_data(self.parameters['in_datasets'], 'in_data')

    @docstring_parameter("out")
    @docstring_parameter(notes.mData_notes.__doc__)
    def get_out_meta_data(self):
        """ {0} """
        return self.__set_meta_data(self.parameters['out_datasets'],
                                    'out_data')

    def get(self):
        """ Get a list of meta_data objects associated with the
        in/out_datasets.

        :returns: All MetaData objects associated with out data objects.
        :rtype: list(MetaData(in_datasets)), list(MetaData(out_datasets))
        """
        return self.get_in_meta_data(), self.get_out_meta_data()

    def __set_meta_data(self, data_list, dtype):
        """ Append all MetaData objs associated with specified datasets to a
        list.

        :params list(Data) data_list:
        :returns: All MetaData objects associated with data objects in
            data_list
        :rtype: list(MetaData)
        """
        meta_data = []
        for data in data_list:
            meta_data.append(data.meta_data)
        return meta_data

    def _set_unknown_shape(self, data, key):
        try:
            return (len(data.meta_data.get(key)),)
        except KeyError:
            return (0,)
        


1			# Copyright 2014 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_datasets
17			:platform: Unix
18			:synopsis: Base class of plugin containing all dataset related functions
19
20			.. moduleauthor:: Nicola Wadeson <[email protected]>
21
22			"""
23
24			import copy
25			import numpy as np
26
27			import savu.plugins.plugin_datasets_notes as notes
28			from savu.core.utils import docstring_parameter
29			from savu.data.data_structures.plugin_data import PluginData
30
31
32			class PluginDatasets(object):
33
34			def __init__(self, args, *kwargs):
35			super(PluginDatasets, self).__init__()
36			self.exp = None
37			self.data_objs = {}
38			self.variable_data_flag = False
39			self._max_itemsize = 0
40
41			def __get_data_objects(self, dtype):
42			""" Get the data objects associated with the plugin from the experiment
43			data index.
44
45			:params str dtype: "in_data" or "out_data"
46			:returns: A list of data objects with the names given in
47			``self.parameters``
48			:rtype: list(Data)
49			"""
50			data_list = self.parameters[dtype + 'sets']
51			data_objs = []
52			for data in data_list:
53			data_obj = self.exp.index[dtype][data]
54			data_objs.append(data_obj)
55			return data_objs
56
57			def _clone_datasets(self):
58			for data_obj in list(self.exp.index['out_data'].values()):
59			if data_obj.raw and data_obj.data:
60			data_obj.raw.create_next_instance(data_obj)
61			# data_obj.clone = True
62
63			def _finalise_datasets(self):
64			in_data, out_data = self.get_datasets()
65			for data in in_data + out_data:
66			data._finalise_patterns()
67
68			def _finalise_plugin_datasets(self):
69			if 'dawn_runner' in list(self.exp.meta_data.get_dictionary().keys()):
70			return
71
72			in_pData, out_pData = self.get_plugin_datasets()
73
74			params = {}
75			for pData in in_pData + out_pData:
76			pData._set_meta_data()
77			params[pData] = pData._get_plugin_data_size_params()
78
79			max_bytes = 0
80			for key, value in params.items():
81			if value['transfer_bytes'] > max_bytes:
82			max_data = key
83			max_bytes = value['transfer_bytes']
84
85			# set mft and mfp for the largest dataset
86			max_data.plugin_data_transfer_setup()
			0 ignored issues – show introduced 2021-02-25 17:14 UTC by Report Bug Copy Issue Report The variable `max_data` does not seem to be defined for all execution paths. Loading history...
87			to_set = list(set(params.keys()).difference(set([max_data])))
88
89			for pData in to_set:
90			if params[pData]['total_frames'] == params[max_data]['total_frames']:
91			pData.plugin_data_transfer_setup(copy=max_data)
92			else:
93			if pData.max_frames == 'multiple':
94			msg = "If a plugin reduces the number of frames, the " \
95			"number of frames cannot be 'multiple'."
96			raise Exception(msg)
97			pData.plugin_data_transfer_setup(calc=max_data)
98
99			def __set_in_datasets(self):
100			""" Set the in_data objects.
101
102			:returns: the in_datasets associated with the plugin.
103			:rtype: list[Data]
104			"""
105			return self.__get_data_objects('in_data')
106
107			def __set_out_datasets(self):
108			""" Set the out_data objects.
109
110			If the out_datasets do not exist inside the experiment then create
111			them.
112
113			:returns: the out_datasets associated with the plugin.
114			:rtype: list[Data]
115			"""
116			try:
117			out_data = self.__get_data_objects('out_data')
118			except KeyError:
119			out_data = []
120			for data in self.parameters['out_datasets']:
121			self.exp.create_data_object("out_data", data)
122			out_data = self.__get_data_objects('out_data')
123			for data in out_data:
124			data.extra_dims = self.get_plugin_tools().get_extra_dims()
125			return out_data
126
127			def _get_plugin_data(self, data_list):
128			""" Encapsulate a PluginData object in each dataset associated with
129			the plugin.
130
131			:params list(Data) data_list: A list of Data objects used in a plugin.
132			:returns: A list of PluginData objects.
133			:rtype: list(PluginData)
134			"""
135			pData_list = []
136			ptools = self.get_plugin_tools()
137			used = set()
138			unique_data_list = \
139			[x for x in data_list if x not in used and (used.add(x) or True)]
140			for data in unique_data_list:
141			pData_list.append(PluginData(data, self))
142			pData_list[-1].extra_dims = ptools.get_extra_dims()
143			pData_list[-1].multi_params_dict = ptools.get_multi_params_dict()
144			return pData_list
145
146			def _set_plugin_dataset_names(self):
147			""" Fill in empty in/out_dataset entries with default values.
148			"""
149			params = self.parameters
150			orig_in = copy.copy(params['in_datasets'])
151			in_names = self._set_in_dataset_names(params)
152			# case that an extra in_dataset is added in the plugin
153			in_names = orig_in if len(orig_in) and \
154			len(in_names) > len(orig_in) else in_names
155			self._set_out_dataset_names(params, in_names)
156			# update the entry in the process list
157			data_dict = {'in_datasets': params['in_datasets'],
158			'out_datasets': params['out_datasets']}
159			idx = self.exp.meta_data.get('nPlugin')
160			self.exp.meta_data.plugin_list._update_datasets(idx, data_dict)
161
162			def _set_in_dataset_names(self, params):
163			dIn = params['in_datasets']
164			dIn = dIn if isinstance(dIn, list) else [dIn]
165			dIn = self.exp._set_all_datasets('in_data') if len(dIn) ==0 else dIn
166			params['in_datasets'] = dIn
167			nIn = self.nInput_datasets() # datasets many be added dynamically here
168			return self.check_nDatasets(params['in_datasets'], nIn, 'in_data')
169
170			def _set_out_dataset_names(self, params, in_names):
171			dOut = params['out_datasets'] if 'out_datasets' in params.keys() else []
172			dOut = dOut if isinstance(dOut, list) else [dOut]
173			dOut = (copy.copy(in_names) if len(dOut) == 0 else dOut)
174			clones = self.nClone_datasets()
175			params['out_datasets'] = dOut
176			nOut = self.nOutput_datasets()
177			names = self.check_nDatasets(params['out_datasets'], nOut,
178			"out_data", clones=clones)
179			if clones:
180			dOut.extend(['itr_clone' + str(i) for i in range(clones)])
181
182			for i in range(len(names)):
183			new = names[i].split('in_datasets')
184			if len(new) == 2:
185			names[i] = in_names[int(list(new[1])[1])]
186			params["out_datasets"] = names
187			return names
188
189			def check_nDatasets(self, names, nSets, dtype, clones=0):
190			nSets = len(self.parameters[dtype + 'sets']) if nSets=='var' else nSets
191			if len(names) is not (nSets - clones):
192			if nSets == 0:
193			names = []
194			else:
195			msg = "ERROR: Broken plugin chain. \n Please name the %s %s " \
196			"sets associated with the plugin %s in the process file." % \
197			(str(nSets), dtype, self.name)
198			raise Exception(msg)
199			return names
200
201			def _set_plugin_datasets(self):
202			""" Populate ``self.parameters`` in/out_datasets and
203			plugin_in/out_datasets with the relevant objects (Data or PluginData).
204			"""
205			if not self.exp._get_dataset_names_complete():
206			self._set_plugin_dataset_names()
207			self.parameters['in_datasets'] = self.__set_in_datasets()
208			self.parameters['out_datasets'] = self.__set_out_datasets()
209			self.parameters['plugin_in_datasets'] = \
210			self._get_plugin_data(self.parameters['in_datasets'])
211			self.parameters['plugin_out_datasets'] = \
212			self._get_plugin_data(self.parameters['out_datasets'])
213
214			@docstring_parameter('PluginData', 'in')
215			@docstring_parameter(notes.datasets_notes.__doc__)
216			def get_plugin_in_datasets(self):
217			""" {0} """
218			return self.parameters['plugin_in_datasets']
219
220			@docstring_parameter('PluginData', 'out')
221			@docstring_parameter(notes.datasets_notes.__doc__)
222			def get_plugin_out_datasets(self):
223			""" {0} """
224			return self.parameters['plugin_out_datasets']
225
226			@docstring_parameter("PluginData")
227			@docstring_parameter(notes.two_datasets_notes.__doc__)
228			def get_plugin_datasets(self):
229			""" {0} """
230			return self.get_plugin_in_datasets(), self.get_plugin_out_datasets()
231
232			@docstring_parameter("Data", "in")
233			@docstring_parameter(notes.datasets_notes.__doc__)
234			def get_in_datasets(self):
235			""" {0} """
236			return self.parameters['in_datasets']
237
238			@docstring_parameter("Data", "out")
239			@docstring_parameter(notes.datasets_notes.__doc__)
240			def get_out_datasets(self):
241			""" {0} """
242			return self.parameters['out_datasets']
243
244			@docstring_parameter("PluginData")
245			@docstring_parameter(notes.two_datasets_notes.__doc__)
246			def get_datasets(self):
247			""" {0} """
248			return self.get_in_datasets(), self.get_out_datasets()
249
250			@docstring_parameter("in")
251			@docstring_parameter(notes.mData_notes.__doc__)
252			def get_in_meta_data(self):
253			""" {0} """
254			return self.__set_meta_data(self.parameters['in_datasets'], 'in_data')
255
256			@docstring_parameter("out")
257			@docstring_parameter(notes.mData_notes.__doc__)
258			def get_out_meta_data(self):
259			""" {0} """
260			return self.__set_meta_data(self.parameters['out_datasets'],
261			'out_data')
262
263			def get(self):
264			""" Get a list of meta_data objects associated with the
265			in/out_datasets.
266
267			:returns: All MetaData objects associated with out data objects.
268			:rtype: list(MetaData(in_datasets)), list(MetaData(out_datasets))
269			"""
270			return self.get_in_meta_data(), self.get_out_meta_data()
271
272			def __set_meta_data(self, data_list, dtype):
273			""" Append all MetaData objs associated with specified datasets to a
274			list.
275
276			:params list(Data) data_list:
277			:returns: All MetaData objects associated with data objects in
278			data_list
279			:rtype: list(MetaData)
280			"""
281			meta_data = []
282			for data in data_list:
283			meta_data.append(data.meta_data)
284			return meta_data
285
286			def _set_unknown_shape(self, data, key):
287			try:
288			return (len(data.meta_data.get(key)),)
289			except KeyError:
290			return (0,)
291
292

DiamondLightSource / Savu

PluginDatasets._set_in_dataset_names() A last analyzed 2022-08-17 15:44 UTC

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like

PluginDatasets._set_in_dataset_names() A
last analyzed 2022-08-17 15:44 UTC