Test Failed
Pull Request — master (#878)
by
unknown
06:41
created

savu.plugins.stats.statistics   F

Complexity

Total Complexity 90

Size/Duplication

Total Lines 413
Duplicated Lines 0 %

Importance

Changes 0
Metric Value
eloc 302
dl 0
loc 413
rs 2
c 0
b 0
f 0
wmc 90

30 Methods

Rating   Name   Duplication   Size   Complexity  
A Statistics.__init__() 0 5 1
A Statistics.setup() 0 11 4
A Statistics.get_data_stats() 0 2 1
A Statistics._write_stats_to_file() 0 25 4
A Statistics.get_global_stats() 0 2 1
A Statistics.get_stats_from_dataset() 0 20 4
B Statistics._write_stats_to_file4() 0 21 6
B Statistics._set_pattern_info() 0 20 8
A Statistics._get_unpadded_slice_list() 0 15 3
A Statistics.get_stats_from_num() 0 21 4
A Statistics.set_stats_residuals() 0 5 1
A Statistics.calc_rmsd() 0 8 2
A Statistics._write_stats_to_file2() 0 14 3
A Statistics._calc_rss() 0 8 2
A Statistics.get_stats() 0 15 3
A Statistics._write_stats_to_file3() 0 14 4
A Statistics._array_to_dict() 0 5 2
A Statistics.calc_volume_stats() 0 4 1
A Statistics._rmsd_from_rss() 0 2 1
A Statistics._count() 0 3 1
A Statistics._post_chain() 0 5 2
A Statistics.calc_stats_residuals() 0 5 2
B Statistics._unpad_slice() 0 20 6
A Statistics._link_stats_to_datasets() 0 14 4
A Statistics.calc_slice_stats() 0 20 3
A Statistics._de_list() 0 7 3
A Statistics.set_slice_stats() 0 7 3
B Statistics.set_volume_stats() 0 36 6
A Statistics._setup_class() 0 20 4
A Statistics.get_volume_stats() 0 2 1

How to fix   Complexity   

Complexity

Complex classes like savu.plugins.stats.statistics often do a lot of different things. To break such a class down, we need to identify a cohesive component within that class. A common approach to find such a component is to look for fields/methods that share the same prefixes, or suffixes.

Once you have determined the fields that belong together, you can apply the Extract Class refactoring. If the component makes sense as a sub-class, Extract Subclass is also a candidate, and is often faster.

1
"""
2
.. module:: statistics
3
   :platform: Unix
4
   :synopsis: Contains and processes statistics information for each plugin.
5
6
.. moduleauthor::Jacob Williamson <[email protected]>
7
8
"""
9
10
from savu.plugins.savers.utils.hdf5_utils import Hdf5Utils
11
from savu.plugins.stats.stats_utils import StatsUtils
12
13
import h5py as h5
14
import numpy as np
15
import os
16
17
18
class Statistics(object):
19
    _pattern_list = ["SINOGRAM", "PROJECTION", "TANGENTOGRAM", "VOLUME_YZ", "VOLUME_XZ", "VOLUME_XY", "VOLUME_3D", "4D_SCAN", "SINOMOVIE"]
20
    no_stats_plugins = ["BasicOperations", "Mipmap"]
21
    _key_list = ["max", "min", "mean", "mean_std_dev", "median_std_dev", "RMSD"]
22
23
24
    def __init__(self):
25
        self.calc_stats = True
26
        self.stats = {'max': [], 'min': [], 'mean': [], 'std_dev': [], 'RSS': [], 'data_points': []}
27
        self.stats_before_processing = {'max': [], 'min': [], 'mean': [], 'std_dev': []}
28
        self.residuals = {'max': [], 'min': [], 'mean': [], 'std_dev': []}
29
30
    def setup(self, plugin_self):
31
        if plugin_self.name in Statistics.no_stats_plugins:
32
            self.calc_stats = False
33
        if self.calc_stats:
34
            self.plugin = plugin_self
35
            self.plugin_name = plugin_self.name
36
            self.pad_dims = []
37
            self._already_called = False
38
            self._set_pattern_info()
39
        if self.calc_stats:
40
            Statistics._any_stats = True
41
42
43
    @classmethod
44
    def _setup_class(cls, exp):
45
        """Sets up the statistics class for the whole plugin chain (only called once)"""
46
        cls._any_stats = False
47
        cls.count = 2
48
        cls.global_stats = {}
49
        cls.exp = exp
50
        n_plugins = len(exp.meta_data.plugin_list.plugin_list)
51
        for i in range(1, n_plugins + 1):
52
            cls.global_stats[i] = np.array([])
53
        cls.global_residuals = {}
54
        cls.plugin_numbers = {}
55
        cls.plugin_names = {}
56
57
        cls.path = exp.meta_data['out_path']
58
        if cls.path[-1] == '/':
59
            cls.path = cls.path[0:-1]
60
        cls.path = f"{cls.path}/stats"
61
        if not os.path.exists(cls.path):
62
            os.mkdir(cls.path)
63
64
    def set_slice_stats(self, slice, base_slice):
65
        slice_stats_before = self.calc_slice_stats(base_slice)
66
        slice_stats_after = self.calc_slice_stats(slice, base_slice)
67
        for key in list(self.stats_before_processing.keys()):
68
            self.stats_before_processing[key].append(slice_stats_before[key])
69
        for key in list(self.stats.keys()):
70
            self.stats[key].append(slice_stats_after[key])
71
72
    def calc_slice_stats(self, my_slice, base_slice=None):
73
        """Calculates and returns slice stats for the current slice.
74
75
        :param slice1: The slice whose stats are being calculated.
76
        """
77
        if my_slice is not None:
78
            slice_num = self.plugin.pcount
79
            my_slice = self._de_list(my_slice)
80
            my_slice = self._unpad_slice(my_slice)
81
            slice_stats = {'max': np.amax(my_slice).astype('float64'), 'min': np.amin(my_slice).astype('float64'),
82
                           'mean': np.mean(my_slice), 'std_dev': np.std(my_slice), 'data_points': my_slice.size}
83
            if base_slice is not None:
84
                base_slice = self._de_list(base_slice)
85
                base_slice = self._unpad_slice(base_slice)
86
                rss = self._calc_rss(my_slice, base_slice)
87
            else:
88
                rss = None
89
            slice_stats['RSS'] = rss
90
            return slice_stats
91
        return None
92
93
    def _calc_rss(self, array1, array2):  # residual sum of squares
94
        if array1.shape == array2.shape:
95
            residuals = np.subtract(array1, array2)
96
            rss = sum(value**2 for value in np.nditer(residuals))
97
        else:
98
            #print("Warning: cannot calculate RSS, arrays different sizes.")  # need to make this an actual warning
99
            rss = None
100
        return rss
101
102
    def _rmsd_from_rss(self, rss, n):
103
        return np.sqrt(rss/n)
104
105
    def calc_rmsd(self, array1, array2):
106
        if array1.shape == array2.shape:
107
            rss = self._calc_rss(array1, array2)
108
            rmsd = self._rmsd_from_rss(rss, array1.size)
109
        else:
110
            print("Warning: cannot calculate RMSD, arrays different sizes.")  # need to make this an actual warning
111
            rmsd = None
112
        return rmsd
113
114
    def calc_stats_residuals(self, stats_before, stats_after):
115
        residuals = {'max': None, 'min': None, 'mean': None, 'std_dev': None}
116
        for key in list(residuals.keys()):
117
            residuals[key] = stats_after[key] - stats_before[key]
118
        return residuals
119
120
    def set_stats_residuals(self, residuals):
121
        self.residuals['max'].append(residuals['max'])
122
        self.residuals['min'].append(residuals['min'])
123
        self.residuals['mean'].append(residuals['mean'])
124
        self.residuals['std_dev'].append(residuals['std_dev'])
125
126
    def calc_volume_stats(self, slice_stats):
127
        volume_stats = np.array([max(slice_stats['max']), min(slice_stats['min']), np.mean(slice_stats['mean']),
128
                        np.mean(slice_stats['std_dev']), np.median(slice_stats['std_dev'])])
129
        return volume_stats
130
131
    def set_volume_stats(self):
132
        """Calculates volume-wide statistics from slice stats, and updates class-wide arrays with these values.
133
        Links volume stats with the output dataset and writes slice stats to file.
134
        """
135
        p_num = Statistics.count
136
        name = self.plugin_name
137
        i = 2
138
        while name in list(Statistics.plugin_numbers.keys()):
139
            name = self.plugin_name + str(i)
140
            i += 1
141
142
        if len(self.stats['max']) != 0:
143
            stats_array = self.calc_volume_stats(self.stats)
144
            Statistics.global_residuals[p_num] = {}
145
            before_processing = self.calc_volume_stats(self.stats_before_processing)
146
            #for key in list(before_processing.keys()):
147
            #    Statistics.global_residuals[p_num][key] = Statistics.global_stats[p_num][key] - before_processing[key]
148
            if None not in self.stats['RSS']:
149
                total_rss = sum(self.stats['RSS'])
150
                n = sum(self.stats['data_points'])
151
                RMSD = self._rmsd_from_rss(total_rss, n)
152
                stats_array = np.append(stats_array, RMSD)
153
            #else:
154
            #    stats_array = np.append(stats_array[p_num], None)
155
            if len(Statistics.global_stats[p_num]) == 0:
156
                Statistics.global_stats[p_num] = stats_array
157
            else:
158
                Statistics.global_stats[p_num] = np.vstack([Statistics.global_stats[p_num], stats_array])
159
            Statistics.plugin_numbers[name] = p_num
160
            if p_num not in list(Statistics.plugin_names.keys()):
161
                Statistics.plugin_names[p_num] = name
162
            self._link_stats_to_datasets(Statistics.global_stats[Statistics.plugin_numbers[name]])
163
164
        slice_stats_array = np.array([self.stats['max'], self.stats['min'], self.stats['mean'], self.stats['std_dev']])
165
        self._write_stats_to_file3(p_num)
166
        self._already_called = True
167
168
    def get_stats(self, plugin_name, n=None, stat=None):
169
        """Returns stats associated with a certain plugin.
170
171
        :param plugin_name: name of the plugin whose associated stats are being fetched.
172
        :param n: In a case where there are multiple instances of **plugin_name** in the process list,
173
            specify the nth instance. Not specifying will select the first (or only) instance.
174
        :param stat: Specify the stat parameter you want to fetch, i.e 'max', 'mean', 'median_std_dev'.
175
            If left blank will return the whole dictionary of stats:
176
            {'max': , 'min': , 'mean': , 'mean_std_dev': , 'median_std_dev': }
177
        """
178
        name = plugin_name
179
        if n is not None and n not in (0, 1):
180
            name = name + str(n)
181
        p_num = Statistics.plugin_numbers[name]
182
        return self.get_stats_from_num(p_num, stat)
183
184
    def get_stats_from_num(self, p_num, stat=None, instance=0):
185
        """Returns stats associated with a certain plugin, given the plugin number (its place in the process list).
186
187
        :param p_num: Plugin  number of the plugin whose associated stats are being fetched.
188
            If p_num <= 0, it is relative to the plugin number of the current plugin being run.
189
            E.g current plugin number = 5, p_num = -2 --> will return stats of the third plugin.
190
        :param stat: Specify the stat parameter you want to fetch, i.e 'max', 'mean', 'median_std_dev'.
191
            If left blank will return the whole dictionary of stats:
192
            {'max': , 'min': , 'mean': , 'mean_std_dev': , 'median_std_dev': }
193
        """
194
        if p_num <= 0:
195
            p_num = Statistics.count + p_num
196
        if Statistics.global_stats[p_num].ndim == 1:
197
            stats_array = Statistics.global_stats[p_num]
198
        else:
199
            stats_array = Statistics.global_stats[p_num][instance]
200
        stats_dict = self._array_to_dict(stats_array)
201
        if stat is not None:
202
            return stats_dict[stat]
203
        else:
204
            return stats_dict
205
206
    def get_stats_from_dataset(self, dataset, stat=None, instance=None):
207
        """Returns stats associated with a dataset.
208
209
        :param dataset: The dataset whose associated stats are being fetched.
210
        :param stat: Specify the stat parameter you want to fetch, i.e 'max', 'mean', 'median_std_dev'.
211
            If left blank will return the whole dictionary of stats:
212
            {'max': , 'min': , 'mean': , 'mean_std_dev': , 'median_std_dev': }
213
        :param instance: In the (rare) case that there are multiple sets of stats associated with the dataset,
214
            specify which set to return.
215
216
        """
217
        key = "stats"
218
        stats = {}
219
        if instance is not None and instance not in (0, 1):
220
            key = key + str(instance)
221
        stats = dataset.meta_data.get(key)
222
        if stat is not None:
223
            return stats[stat]
224
        else:
225
            return stats
226
227
    def get_data_stats(self):
228
        return Statistics.data_stats
229
230
    def get_volume_stats(self):
231
        return Statistics.volume_stats
232
233
    def get_global_stats(self):
234
        return Statistics.global_stats
235
236
    def _array_to_dict(self, stats_array):
237
        stats_dict = {}
238
        for i, value in enumerate(stats_array):
239
            stats_dict[Statistics._key_list[i]] = value
240
        return stats_dict
241
242
    def _set_pattern_info(self):
243
        """Gathers information about the pattern of the data in the current plugin."""
244
        in_datasets, out_datasets = self.plugin.get_datasets()
245
        try:
246
            self.pattern = self.plugin.parameters['pattern']
247
            if self.pattern == None:
248
                raise KeyError
249
        except KeyError:
250
            if not out_datasets:
251
                self.pattern = None
252
            else:
253
                patterns = out_datasets[0].get_data_patterns()
254
                for pattern in patterns:
255
                    if 1 in patterns.get(pattern)["slice_dims"]:
256
                        self.pattern = pattern
257
                        break
258
        self.calc_stats = False
259
        for dataset in out_datasets:
260
            if bool(set(Statistics._pattern_list) & set(dataset.data_info.get("data_patterns"))):
261
                self.calc_stats = True
262
263
    def _link_stats_to_datasets(self, stats):
264
        """Links the volume wide statistics to the output dataset(s)"""
265
        out_dataset = self.plugin.get_out_datasets()[0]
266
        n_datasets = self.plugin.nOutput_datasets()
267
        stats_dict = self._array_to_dict(stats)
268
        i = 2
269
        group_name = "stats"
270
        #out_dataset.data_info.set([group_name], stats)
271
        if n_datasets == 1:
272
            while group_name in list(out_dataset.meta_data.get_dictionary().keys()):
273
                group_name = f"stats{i}"
274
                i += 1
275
            for key in list(stats_dict.keys()):
276
                out_dataset.meta_data.set([group_name, key], stats_dict[key])
277
278
    def _write_stats_to_file2(self, p_num):
279
        path = Statistics.path
280
        filename = f"{path}/stats.h5"
281
        stats = Statistics.global_stats[p_num]
282
        array_dim = stats.shape
283
        self.hdf5 = Hdf5Utils(self.plugin.exp)
284
        group_name = f"{p_num}-{self.plugin_name}-stats"
285
        with h5.File(filename, "a") as h5file:
286
            if group_name not in h5file:
287
                group = h5file.create_group(group_name, track_order=None)
288
                dataset = self.hdf5.create_dataset_nofill(group, "stats", array_dim, stats.dtype)
289
                dataset[::] = stats[::]
290
            else:
291
                group = h5file[group_name]
292
293
294
    @classmethod
295
    def _write_stats_to_file4(cls):
296
        path = cls.path
297
        filename = f"{path}/stats.h5"
298
        stats = cls.global_stats
299
        cls.hdf5 = Hdf5Utils(cls.exp)
300
        for i in range(5):
301
            array = np.array([])
302
            stat = cls._key_list[i]
303
            for key in list(stats.keys()):
304
                if len(stats[key]) != 0:
305
                    if stats[key].ndim == 1:
306
                        array = np.append(array, stats[key][i])
307
                    else:
308
                        array = np.append(array, stats[key][0][i])
309
            array_dim = array.shape
310
            group_name = f"all-{stat}"
311
            with h5.File(filename, "a") as h5file:
312
                group = h5file.create_group(group_name, track_order=None)
313
                dataset = cls.hdf5.create_dataset_nofill(group, stat, array_dim, array.dtype)
314
                dataset[::] = array[::]
315
316
    def _write_stats_to_file3(self, p_num):
317
        path = Statistics.path
318
        filename = f"{path}/stats.h5"
319
        stats = self.global_stats
320
        self.hdf5 = Hdf5Utils(self.exp)
321
        with h5.File(filename, "a") as h5file:
322
            group = h5file.require_group("stats")
323
            if stats[p_num].shape != (0,):
324
                if str(p_num) in list(group.keys()):
325
                    del group[str(p_num)]
326
                dataset = group.create_dataset(str(p_num), shape=stats[p_num].shape, dtype=stats[p_num].dtype)
327
                dataset[::] = stats[p_num][::]
328
                dataset.attrs.create("plugin_name", self.plugin_names[p_num])
329
                dataset.attrs.create("pattern", self.pattern)
330
331
332
    def _write_stats_to_file(self, slice_stats_array, p_num):
333
        """Writes slice statistics to a h5 file"""
334
        path = Statistics.path
335
        filename = f"{path}/stats_p{p_num}_{self.plugin_name}.h5"
336
        slice_stats_dim = (slice_stats_array.shape[1],)
337
        self.hdf5 = Hdf5Utils(self.plugin.exp)
338
        with h5.File(filename, "a") as h5file:
339
            i = 2
340
            group_name = "/stats"
341
            while group_name in h5file:
342
                group_name = f"/stats{i}"
343
                i += 1
344
            group = h5file.create_group(group_name, track_order=None)
345
            max_ds = self.hdf5.create_dataset_nofill(group, "max", slice_stats_dim, slice_stats_array.dtype)
346
            min_ds = self.hdf5.create_dataset_nofill(group, "min", slice_stats_dim, slice_stats_array.dtype)
347
            mean_ds = self.hdf5.create_dataset_nofill(group, "mean", slice_stats_dim, slice_stats_array.dtype)
348
            std_dev_ds = self.hdf5.create_dataset_nofill(group, "standard_deviation",
349
                                                         slice_stats_dim, slice_stats_array.dtype)
350
            if slice_stats_array.shape[0] == 5:
351
                rmsd_ds = self.hdf5.create_dataset_nofill(group, "RMSD", slice_stats_dim, slice_stats_array.dtype)
352
                rmsd_ds[::] = slice_stats_array[4]
353
            max_ds[::] = slice_stats_array[0]
354
            min_ds[::] = slice_stats_array[1]
355
            mean_ds[::] = slice_stats_array[2]
356
            std_dev_ds[::] = slice_stats_array[3]
357
358
    def _unpad_slice(self, slice1):
359
        """If data is padded in the slice dimension, removes this pad."""
360
        out_datasets = self.plugin.get_out_datasets()
361
        if len(out_datasets) == 1:
362
            out_dataset = out_datasets[0]
363
        else:
364
            for dataset in out_datasets:
365
                if self.pattern in list(dataset.data_info.get(["data_patterns"]).keys()):
366
                    out_dataset = dataset
367
                    break
368
        slice_dims = out_dataset.get_slice_dimensions()
0 ignored issues
show
introduced by
The variable out_dataset does not seem to be defined for all execution paths.
Loading history...
369
        if self.plugin.pcount == 0:
370
            self.slice_list, self.pad = self._get_unpadded_slice_list(slice1, slice_dims)
371
        if self.pad:
372
            #for slice_dim in slice_dims:
373
            slice_dim = slice_dims[0]
374
            temp_slice = np.swapaxes(slice1, 0, slice_dim)
375
            temp_slice = temp_slice[self.slice_list[slice_dim]]
376
            slice1 = np.swapaxes(temp_slice, 0, slice_dim)
377
        return slice1
378
379
    def _get_unpadded_slice_list(self, slice1, slice_dims):
380
        """Creates slice object(s) to un-pad slices in the slice dimension(s)."""
381
        slice_list = list(self.plugin.slice_list[0])
382
        pad = False
383
        if len(slice_list) == len(slice1.shape):
384
            #for i in slice_dims:
385
            i = slice_dims[0]
386
            slice_width = self.plugin.slice_list[0][i].stop - self.plugin.slice_list[0][i].start
387
            if slice_width != slice1.shape[i]:
388
                pad = True
389
                pad_width = (slice1.shape[i] - slice_width) // 2  # Assuming symmetrical padding
390
                slice_list[i] = slice(pad_width, pad_width + 1, 1)
391
            return tuple(slice_list), pad
392
        else:
393
            return self.plugin.slice_list[0], pad
394
395
    def _de_list(self, slice1):
396
        """If the slice is in a list, remove it from that list."""
397
        if type(slice1) == list:
398
            if len(slice1) != 0:
399
                slice1 = slice1[0]
400
                slice1 = self._de_list(slice1)
401
        return slice1
402
403
404
    @classmethod
405
    def _count(cls):
406
        cls.count += 1
407
408
    @classmethod
409
    def _post_chain(cls):
410
        if cls._any_stats:
411
            stats_utils = StatsUtils()
412
            stats_utils.generate_figures(f"{cls.path}/stats.h5", cls.path)
413