Skip to content

Commit

Permalink
New tests for calculate_target
Browse files Browse the repository at this point in the history
  • Loading branch information
@hrobarts
hrobarts committed Oct 16, 2024
1 parent 69e4466 commit da5f977
Show file tree
Hide file tree
Showing 2 changed files with 118 additions and 77 deletions.
30 changes: 8 additions & 22 deletions Wrappers/Python/cil/processors/FluxNormaliser.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -128,7 +128,6 @@ def check_input(self, dataset):

return True


def _calculate_flux(self):
'''
Function to calculate flux from a region of interest in the data. If the
Expand Down Expand Up @@ -216,13 +215,16 @@ def _calculate_flux(self):
if 0 in self.flux:
raise ValueError('Flux value can\'t be 0, provide a different flux\
or region of interest with non-zero values')



def _calculate_target(self):
'''
Calculate the target value for the normalisation
'''
if isinstance(self.target, float):

if self.flux is None:
raise ValueError('Flux not found')

if isinstance(self.target, (int,float)):
self.target_value = self.target
elif isinstance(self.target, str):
if self.target == 'first':
Expand All @@ -236,11 +238,9 @@ def _calculate_target(self):
raise ValueError("Target string not recognised, found {}, expected 'first' or 'mean'"
.format(self.target))
else:
raise TypeError("Target must be string or float, found {}"
raise TypeError("Target must be string or a number, found {}"
.format(type(self.target)))



def preview_configuration(self, angle=None, channel=None, log=False):
'''
Preview the FluxNormalisation processor configuration for roi mode.
Expand Down Expand Up @@ -406,18 +406,4 @@ def process(self, out=None):
arr_proj *= self.target_value/f
out.array.flat[i*proj_size:(i+1)*proj_size] = arr_proj


# if 'angle' in data.dimension_labels:
# proj_axis = data.get_dimension_axis('angle')
# slice_proj = [slice(None)]*len(data.shape)
# slice_proj[proj_axis] = 0

# for i in range(len(data.geometry.angles)):
# if len(flux_size) > 0:
# f = self.flux[i]
# slice_proj[proj_axis] = i
# out.array[tuple(slice_proj)] = data.array[tuple(slice_proj)]*self.target_value/f
# else:
# out.array = data.array*self.target_value/f

return out
return out
165 changes: 110 additions & 55 deletions Wrappers/Python/test/test_DataProcessor.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -3160,6 +3160,59 @@ def test_calculate_flux(self):
with self.assertRaises(ValueError):
processor._calculate_flux()

def test_calculate_target(self):

# check target calculated with default method 'mean'
processor = FluxNormaliser(flux=1)
processor.set_input(self.data_cone)
processor._calculate_flux()
processor._calculate_target()
self.assertAlmostEqual(processor.target_value, 1)

processor = FluxNormaliser(flux=numpy.linspace(1,3,len(self.data_cone.geometry.angles)))
processor.set_input(self.data_cone)
processor._calculate_flux()
processor._calculate_target()
self.assertAlmostEqual(processor.target_value, 2)

# check target calculated with method 'first'
processor = FluxNormaliser(flux=1, target='first')
processor.set_input(self.data_cone)
processor._calculate_flux()
processor._calculate_target()
self.assertAlmostEqual(processor.target_value, 1)

processor = FluxNormaliser(flux=numpy.linspace(1,3,len(self.data_cone.geometry.angles)),
target='first')
processor.set_input(self.data_cone)
processor._calculate_flux()
processor._calculate_target()
self.assertAlmostEqual(processor.target_value, 1)

# check target calculated with float
processor = FluxNormaliser(flux=1,
target=55.0)
processor.set_input(self.data_cone)
processor._calculate_flux()
processor._calculate_target()
self.assertAlmostEqual(processor.target_value, 55.0)

# check error if target is an unrecognised string
processor = FluxNormaliser(flux=1,
target='string')
processor.set_input(self.data_cone)
processor._calculate_flux()
with self.assertRaises(ValueError):
processor._calculate_target()

# check error if target is not a string or floar
processor = FluxNormaliser(flux=1,
target={'string': 10})
processor.set_input(self.data_cone)
processor._calculate_flux()
with self.assertRaises(TypeError):
processor._calculate_target()

@patch("matplotlib.pyplot.figure")
def test_preview_configuration(self, mock_plot):
# Test error in preview configuration if there is no roi
Expand Down Expand Up @@ -3236,67 +3289,69 @@ def test_FluxNormaliser(self):
data_norm_test.array[a,:,:]*= norm_value
numpy.testing.assert_allclose(data_norm.array, data_norm_test.array, atol=1e-6)

# #Test roi with no norm_value
# roi = {'vertical':(0,10), 'horizontal':(0,10)}
# processor = FluxNormaliser(roi=roi)
# processor.set_input(self.data_cone)
# data_norm = processor.get_output()
# numpy.testing.assert_allclose(data_norm.array, self.data_cone.array)
# #Test roi with no target
roi = {'vertical':(0,10), 'horizontal':(0,10)}
processor = FluxNormaliser(roi=roi)
processor.set_input(self.data_cone)
data_norm = processor.get_output()
numpy.testing.assert_allclose(data_norm.array, self.data_cone.array)

# #Test roi with norm_value
# roi = {'vertical':(0,10), 'horizontal':(0,10)}
# processor = FluxNormaliser(roi=roi, norm_value=5)
# processor.set_input(self.data_cone)
# data_norm = processor.get_output()
# numpy.testing.assert_allclose(data_norm.array, 5*self.data_cone.array)
roi = {'vertical':(0,10), 'horizontal':(0,10)}
processor = FluxNormaliser(roi=roi, target=5.0)
processor.set_input(self.data_cone)
data_norm = processor.get_output()
numpy.testing.assert_allclose(data_norm.array, 5*self.data_cone.array)

# # Test roi with just one dimension
# roi = {'vertical':(0,2)}
# processor = FluxNormaliser(roi=roi, norm_value=5)
# processor.set_input(self.data_cone)
# data_norm = processor.get_output()
# numpy.testing.assert_allclose(data_norm.array, 5*self.data_cone.array)
roi = {'vertical':(0,2)}
processor = FluxNormaliser(roi=roi, target=5)
processor.set_input(self.data_cone)
data_norm = processor.get_output()
numpy.testing.assert_allclose(data_norm.array, 5*self.data_cone.array)

# test roi with different data shapes and different flux values per projection
# for data in [self.data_cone, self.data_parallel, self.data_multichannel,
# self.data_slice, self.data_reorder]:
# roi = {'horizontal':(25,40)}
# processor = FluxNormaliser(roi=roi, norm_value=5)
# processor.set_input(data)
# data_norm = processor.get_output()

# ax = data.get_dimension_axis('horizontal')
# slc = [slice(None)]*len(data.shape)
# slc[ax] = slice(25,40)
# axes=[ax]
# if 'vertical' in data.dimension_labels:
# axes.append(data.get_dimension_axis('vertical'))
# if 'channel' in data.dimension_labels:
# axes.append(data.get_dimension_axis('channel'))
# flux = numpy.mean(data.array[tuple(slc)], axis=tuple(axes))
# slice_proj = [slice(None)]*len(data.shape)
# proj_axis = data.get_dimension_axis('angle')
# data_norm_test = data.copy()
# for i in range(len(data.geometry.angles)):
# f = flux[i]
# slice_proj[proj_axis] = i
# with numpy.errstate(divide='ignore', invalid='ignore'):
# data_norm_test.array[tuple(slice_proj)] = 5/f*data.array[tuple(slice_proj)]
# numpy.testing.assert_allclose(data_norm.array, data_norm_test.array, atol=1e-6,
# err_msg='Flux Normaliser roi test failed with data shape: ' + str(data.shape) + ' and configuration:\n' + str(data.geometry.config.system))

# data = self.data_single_angle
# processor = FluxNormaliser(roi=roi, norm_value=5)
# processor.set_input(data)
# data_norm = processor.get_output()
# ax = data.get_dimension_axis('horizontal')
# slc = [slice(None)]*len(data.shape)
# slc[ax] = slice(25,40)
# axes=[ax,data.get_dimension_axis('vertical')]
# flux = numpy.mean(data.array[tuple(slc)], axis=tuple(axes))

# numpy.testing.assert_allclose(data_norm.array, 5/flux*data.array, atol=1e-6,
# err_msg='Flux Normaliser roi test failed with data shape: ' + str(data.shape) + ' and configuration:\n' + str(data.geometry.config.system))
for data in [ self.data_cone, self.data_parallel, self.data_multichannel,
self.data_slice, self.data_reorder]:
roi = {'horizontal':(25,40)}
processor = FluxNormaliser(roi=roi, target=5)
processor.set_input(data)
data_norm = processor.get_output()

ax = data.get_dimension_axis('horizontal')
slc = [slice(None)]*len(data.shape)
slc[ax] = slice(25,40)
axes=[ax]
if 'vertical' in data.dimension_labels:
axes.append(data.get_dimension_axis('vertical'))
flux = numpy.mean(data.array[tuple(slc)], axis=tuple(axes))
slice_proj = [slice(None)]*len(data.shape)
proj_axis = data.get_dimension_axis('angle')
data_norm_test = data.copy()
h_size = data.get_dimension_size('horizontal')
if 'vertical' in data.dimension_labels:
v_size = data.get_dimension_size('vertical')
else:
v_size = 1
proj_size = h_size*v_size
for i in range(len(data.geometry.angles)*data.geometry.channels):
data_norm_test.array.flat[i*proj_size:(i+1)*proj_size] /=flux.flat[i]
data_norm_test.array.flat[i*proj_size:(i+1)*proj_size] *=5
numpy.testing.assert_allclose(data_norm.array, data_norm_test.array, atol=1e-6,
err_msg='Flux Normaliser roi test failed with data shape: ' + str(data.shape) + ' and configuration:\n' + str(data.geometry.config.system))

data = self.data_single_angle
processor = FluxNormaliser(roi=roi, target=5)
processor.set_input(data)
data_norm = processor.get_output()
ax = data.get_dimension_axis('horizontal')
slc = [slice(None)]*len(data.shape)
slc[ax] = slice(25,40)
axes=[ax,data.get_dimension_axis('vertical')]
flux = numpy.mean(data.array[tuple(slc)], axis=tuple(axes))

numpy.testing.assert_allclose(data_norm.array, 5/flux*data.array, atol=1e-6,
err_msg='Flux Normaliser roi test failed with data shape: ' + str(data.shape) + ' and configuration:\n' + str(data.geometry.config.system))

if __name__ == "__main__":

Expand Down

0 comments on commit da5f977

Please sign in to comment.