savu.plugins.segmentation.thresholding.otsu_thresh.OtsuThresh.__init__() - Code Metrics - DiamondLightSource/Savu - Measure and Improve Code Quality continuously with Scrutinizer

OtsuThresh.init() A
last analyzed 2022-08-17 15:44 UTC

↳ Parent: savu.plugins.segmentation.thresholding.otsu_thresh

Complexity

Conditions

Size

Total Lines	2
Code Lines	2

Duplication

Lines	0
Ratio	0 %

Importance

Changes

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

# Copyright 2019 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:: otsu_thresh
   :platform: Unix
   :synopsis: Segmentation by thresholding based on Otsu's method.
              Optionally calculate cropping values based on the segmented image

.. moduleauthor:: Jacob Williamson <[email protected]>
"""


from savu.plugins.plugin import Plugin
from savu.plugins.driver.cpu_plugin import CpuPlugin
from savu.plugins.utils import register_plugin

import itertools
import numpy as np
from skimage.filters import threshold_otsu
from PIL import Image
import h5py as h5
import os

@register_plugin
class OtsuThresh(Plugin, CpuPlugin):

    def __init__(self):
        super(OtsuThresh, self).__init__("OtsuThresh")

    def setup(self):

        in_dataset, out_dataset = self.get_datasets()
        in_pData, out_pData = self.get_plugin_datasets()
        in_pData[0].plugin_data_setup(self.parameters['pattern'], 'single')

        for i in range(len(out_dataset)):
            out_dataset[0].create_dataset(in_dataset[0], dtype=np.uint8)
            out_pData[0].plugin_data_setup(self.parameters['pattern'], 'single')

        self.cropping = self.parameters["cropping"]
        self.buffer = self.parameters["buffer"]
        self.directions = self.parameters["directions"]

        self.shape = in_dataset[0].data_info.get("shape")
        self.fully_right, self.fully_below = self.shape[2] - 1, self.shape[1] - 1
        self.volume_crop = {"left": 0, "above": 0, "right": self.fully_right, "below": self.fully_below}
        self.orig_edges = {"left": 0, "above": 0, "right": self.fully_right, "below": self.fully_below}
        self.gap_size = 20

    def pre_process(self):

        if "left" in self.directions:
            self.volume_crop["left"] = self.fully_right
        if "above" in self.directions:
            self.volume_crop["above"] = self.fully_below
        if "right" in self.directions:
            self.volume_crop["right"] = 0
        if "below" in self.directions:
            self.volume_crop["below"] = 0

    def process_frames(self, data):
        threshold = threshold_otsu(data[0])
        thresh_result = (data[0] > threshold) * 1
        if self.cropping:
            cropped_slice = self._crop(thresh_result, ["left", "above", "right", "below"])
            #if self.pcount % (self.shape[0]//10) == 0:
            #    self.__save_image(cropped_slice, f"cropped_slices/slice{self.pcount}-cropped")
        if self.exp.meta_data.get("pre_run"):
            return None
        else:
            return thresh_result

    def post_process(self):
        if self.cropping:
            preview = self._cropping_post_process()
            if self.exp.meta_data.get("pre_run"):
                self._write_preview_to_file(preview)

    def nInput_datasets(self):
        return 1

    def nOutput_datasets(self):
        if self.exp.meta_data.get("pre_run"):
            return 0
        else:
            return 1

    def _crop(self, binary_slice, directions, buffer=0):
        # For 3 dimensional volume and 2 dimensional slices

        total_crop = {"left": 0, "above": 0, "right": self.fully_right, "below": self.fully_below}
        total_counter = 0
        reset_counter = 0
        dir_cycle = itertools.cycle(directions)
        while reset_counter < 4:
            direction = next(dir_cycle)
            crops = {"left": 0, "above": 0, "right": binary_slice.shape[1], "below": binary_slice.shape[0]}
            if direction in ["above", "below"]:
                axis = 1
            elif direction in ["left", "right"]:
                axis = 0

            non_zeros_list = np.count_nonzero(binary_slice, axis=axis)

            previous = None
            fills = []
            gaps = []
            for i, non_zeros in enumerate(non_zeros_list):
                if non_zeros + buffer < binary_slice.shape[axis]:  # check if some zeros (indicating sample)
                    if previous != "fill":
                        fills.append([])
                    fills[-1].append(i)
                    previous = "fill"
                else:
                    if previous != "gap":
                        gaps.append([])
                    gaps[-1].append(i)
                    previous = "gap"

            #  find largest area of zeros (assume this is where sample is)
            largest_fill = []
            for fill in fills:
                if len(fill) > len(largest_fill):
                    largest_fill = fill

            #  find a reasonably sized gap closest to the sample
            if direction in ["left", "above"]:
                for gap in gaps:
                    if gap[-1] < largest_fill[0]:
                        if len(gap) > self.gap_size:
                            crops[direction] = gap[-1]
                            reset_counter = 0
                total_crop[direction] += crops[direction]

            elif direction in ["right", "below"]:
                for gap in gaps[::-1]:
                    if gap[0] > largest_fill[-1]:
                        if len(gap) > self.gap_size:
                            crops[direction] = gap[0] + 1
                            reset_counter = 0
                total_crop[direction] -= (binary_slice.shape[1 - axis] - crops[direction])

            #if self.pcount == 0:
            #    self.__save_image(binary_slice, f"{total_counter}-{reset_counter}-{direction}")

            reset_counter += 1
            total_counter += 1

            binary_slice = binary_slice[crops["above"]: crops["below"], crops["left"]: crops["right"]]

        for direction in self.directions:
            if direction in ["left", "above"]:
                if total_crop[direction] < self.volume_crop[direction]:
                    self.volume_crop[direction] = total_crop[direction]
            if direction in ["right", "below"]:
                if total_crop[direction] > self.volume_crop[direction]:
                    self.volume_crop[direction] = total_crop[direction]
        return binary_slice

    def _cropping_post_process(self):
        for direction in self.directions:
            if direction in ["left", "above"]:
                self.volume_crop[direction] = int(self.volume_crop[direction] - self.buffer)
                if self.volume_crop[direction] < self.orig_edges[direction]:
                    self.volume_crop[direction] = self.orig_edges[direction]
            if direction in ["right", "below"]:
                self.volume_crop[direction] = int(self.volume_crop[direction] + self.buffer)
                if self.volume_crop[direction] > self.orig_edges[direction]:
                    self.volume_crop[direction] = self.orig_edges[direction]
        preview = f":, {self.volume_crop['above']}:{self.volume_crop['below']}, {self.volume_crop['left']}:{self.volume_crop['right']}"
        self.exp.meta_data.set("pre_run_preview", preview)

        return preview

    def _write_preview_to_file(self, preview):
        if self.exp.meta_data.get("pre_run"):
            folder = self.exp.meta_data['out_path']
            fname = self.exp.meta_data.get('datafile_name') + '_pre_run.nxs'
            filename = os.path.join(folder, fname)
            comm = self.get_communicator()
            if comm.rank == 0:
                with h5.File(filename, "a") as h5file:
                    fsplit = self.exp.meta_data["data_path"].split("/")
                    fsplit[-1] = ""
                    stats_path = "/".join(fsplit)
                    preview_group = h5file.require_group(stats_path)
                    preview_group.create_dataset("preview", data=preview)

    def __save_image(self, binary_slice, name):
        #  just used for testing
        binary_slice = binary_slice.astype(np.uint8)*150
        im = Image.fromarray(binary_slice)

        im.save(f"/scratch/Savu/images/{name}.jpeg")

1			# Copyright 2019 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:: otsu_thresh
17			:platform: Unix
18			:synopsis: Segmentation by thresholding based on Otsu's method.
19			Optionally calculate cropping values based on the segmented image
20
21			.. moduleauthor:: Jacob Williamson <[email protected]>
22			"""
23
24
25			from savu.plugins.plugin import Plugin
26			from savu.plugins.driver.cpu_plugin import CpuPlugin
27			from savu.plugins.utils import register_plugin
28
29			import itertools
30			import numpy as np
31			from skimage.filters import threshold_otsu
32			from PIL import Image
33			import h5py as h5
34			import os
35
36			@register_plugin
37			class OtsuThresh(Plugin, CpuPlugin):
38
39			def __init__(self):
40			super(OtsuThresh, self).__init__("OtsuThresh")
41
42			def setup(self):
43
44			in_dataset, out_dataset = self.get_datasets()
45			in_pData, out_pData = self.get_plugin_datasets()
46			in_pData[0].plugin_data_setup(self.parameters['pattern'], 'single')
47
48			for i in range(len(out_dataset)):
49			out_dataset[0].create_dataset(in_dataset[0], dtype=np.uint8)
50			out_pData[0].plugin_data_setup(self.parameters['pattern'], 'single')
51
52			self.cropping = self.parameters["cropping"]
53			self.buffer = self.parameters["buffer"]
54			self.directions = self.parameters["directions"]
55
56			self.shape = in_dataset[0].data_info.get("shape")
57			self.fully_right, self.fully_below = self.shape[2] - 1, self.shape[1] - 1
58			self.volume_crop = {"left": 0, "above": 0, "right": self.fully_right, "below": self.fully_below}
59			self.orig_edges = {"left": 0, "above": 0, "right": self.fully_right, "below": self.fully_below}
60			self.gap_size = 20
61
62			def pre_process(self):
63
64			if "left" in self.directions:
65			self.volume_crop["left"] = self.fully_right
66			if "above" in self.directions:
67			self.volume_crop["above"] = self.fully_below
68			if "right" in self.directions:
69			self.volume_crop["right"] = 0
70			if "below" in self.directions:
71			self.volume_crop["below"] = 0
72
73			def process_frames(self, data):
74			threshold = threshold_otsu(data[0])
75			thresh_result = (data[0] > threshold) * 1
76			if self.cropping:
77			cropped_slice = self._crop(thresh_result, ["left", "above", "right", "below"])
78			#if self.pcount % (self.shape[0]//10) == 0:
79			# self.__save_image(cropped_slice, f"cropped_slices/slice{self.pcount}-cropped")
80			if self.exp.meta_data.get("pre_run"):
81			return None
82			else:
83			return thresh_result
84
85			def post_process(self):
86			if self.cropping:
87			preview = self._cropping_post_process()
88			if self.exp.meta_data.get("pre_run"):
89			self._write_preview_to_file(preview)
90
91			def nInput_datasets(self):
92			return 1
93
94			def nOutput_datasets(self):
95			if self.exp.meta_data.get("pre_run"):
96			return 0
97			else:
98			return 1
99
100			def _crop(self, binary_slice, directions, buffer=0):
101			# For 3 dimensional volume and 2 dimensional slices
102
103			total_crop = {"left": 0, "above": 0, "right": self.fully_right, "below": self.fully_below}
104			total_counter = 0
105			reset_counter = 0
106			dir_cycle = itertools.cycle(directions)
107			while reset_counter < 4:
108			direction = next(dir_cycle)
109			crops = {"left": 0, "above": 0, "right": binary_slice.shape[1], "below": binary_slice.shape[0]}
110			if direction in ["above", "below"]:
111			axis = 1
112			elif direction in ["left", "right"]:
113			axis = 0
114
115			non_zeros_list = np.count_nonzero(binary_slice, axis=axis)
			0 ignored issues – show introduced 2022-07-26 15:17 UTC by Report Bug Copy Issue Report The variable `axis` does not seem to be defined for all execution paths. Loading history...
116			previous = None
117			fills = []
118			gaps = []
119			for i, non_zeros in enumerate(non_zeros_list):
120			if non_zeros + buffer < binary_slice.shape[axis]: # check if some zeros (indicating sample)
121			if previous != "fill":
122			fills.append([])
123			fills[-1].append(i)
124			previous = "fill"
125			else:
126			if previous != "gap":
127			gaps.append([])
128			gaps[-1].append(i)
129			previous = "gap"
130
131			# find largest area of zeros (assume this is where sample is)
132			largest_fill = []
133			for fill in fills:
134			if len(fill) > len(largest_fill):
135			largest_fill = fill
136
137			# find a reasonably sized gap closest to the sample
138			if direction in ["left", "above"]:
139			for gap in gaps:
140			if gap[-1] < largest_fill[0]:
141			if len(gap) > self.gap_size:
142			crops[direction] = gap[-1]
143			reset_counter = 0
144			total_crop[direction] += crops[direction]
145
146			elif direction in ["right", "below"]:
147			for gap in gaps[::-1]:
148			if gap[0] > largest_fill[-1]:
149			if len(gap) > self.gap_size:
150			crops[direction] = gap[0] + 1
151			reset_counter = 0
152			total_crop[direction] -= (binary_slice.shape[1 - axis] - crops[direction])
153
154			#if self.pcount == 0:
155			# self.__save_image(binary_slice, f"{total_counter}-{reset_counter}-{direction}")
156
157			reset_counter += 1
158			total_counter += 1
159
160			binary_slice = binary_slice[crops["above"]: crops["below"], crops["left"]: crops["right"]]
161
162			for direction in self.directions:
163			if direction in ["left", "above"]:
164			if total_crop[direction] < self.volume_crop[direction]:
165			self.volume_crop[direction] = total_crop[direction]
166			if direction in ["right", "below"]:
167			if total_crop[direction] > self.volume_crop[direction]:
168			self.volume_crop[direction] = total_crop[direction]
169			return binary_slice
170
171			def _cropping_post_process(self):
172			for direction in self.directions:
173			if direction in ["left", "above"]:
174			self.volume_crop[direction] = int(self.volume_crop[direction] - self.buffer)
175			if self.volume_crop[direction] < self.orig_edges[direction]:
176			self.volume_crop[direction] = self.orig_edges[direction]
177			if direction in ["right", "below"]:
178			self.volume_crop[direction] = int(self.volume_crop[direction] + self.buffer)
179			if self.volume_crop[direction] > self.orig_edges[direction]:
180			self.volume_crop[direction] = self.orig_edges[direction]
181			preview = f":, {self.volume_crop['above']}:{self.volume_crop['below']}, {self.volume_crop['left']}:{self.volume_crop['right']}"
182			self.exp.meta_data.set("pre_run_preview", preview)
183
184			return preview
185
186			def _write_preview_to_file(self, preview):
187			if self.exp.meta_data.get("pre_run"):
188			folder = self.exp.meta_data['out_path']
189			fname = self.exp.meta_data.get('datafile_name') + '_pre_run.nxs'
190			filename = os.path.join(folder, fname)
191			comm = self.get_communicator()
192			if comm.rank == 0:
193			with h5.File(filename, "a") as h5file:
194			fsplit = self.exp.meta_data["data_path"].split("/")
195			fsplit[-1] = ""
196			stats_path = "/".join(fsplit)
197			preview_group = h5file.require_group(stats_path)
198			preview_group.create_dataset("preview", data=preview)
199
200			def __save_image(self, binary_slice, name):
201			# just used for testing
202			binary_slice = binary_slice.astype(np.uint8)*150
203			im = Image.fromarray(binary_slice)
204
205			im.save(f"/scratch/Savu/images/{name}.jpeg")

DiamondLightSource / Savu

OtsuThresh.__init__() A last analyzed 2022-08-17 15:44 UTC

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like

OtsuThresh.init() A
last analyzed 2022-08-17 15:44 UTC