diff --git a/src/data/ADC_Result.py b/src/data/ADC_Result.py index 29d49d3..f55d4bc 100644 --- a/src/data/ADC_Result.py +++ b/src/data/ADC_Result.py @@ -149,21 +149,21 @@ class ADC_Result(Resultable, Drawable, DamarisFFT, Signalpath): def get_sampling_rate(self): - "Returns the samplingfrequency" + """Returns the samplingfrequency""" return self.sampling_rate + 0 def set_sampling_rate(self, hz): - "Sets the samplingfrequency in hz" + """Sets the samplingfrequency in hz""" self.sampling_rate = float(hz) def get_nChannels(self): - "Gets the number of channels" + """Gets the number of channels""" return self.nChannels + 0 def set_nChannels(self, channels): - "Sets the number of channels" + """Sets the number of channels""" self.nChannels = int(channels) diff --git a/src/data/Accumulation.py b/src/data/Accumulation.py index 8a389bc..a2510d1 100644 --- a/src/data/Accumulation.py +++ b/src/data/Accumulation.py @@ -16,6 +16,8 @@ from .Drawable import Drawable from .DamarisFFT import DamarisFFT from .Signalpath import Signalpath +from data.ADC_Result import ADC_Result + import sys import threading import tables @@ -599,7 +601,7 @@ class Accumulation(Errorable, Drawable, DamarisFFT, Signalpath): return r # ADC_Result - elif str(other.__class__) == "damaris.data.ADC_Result.ADC_Result": + elif isinstance(other, ADC_Result): # Other empty (return) # todo: this is seems to be bugy!!!! (Achim) if not other.contains_data(): @@ -660,12 +662,11 @@ class Accumulation(Errorable, Drawable, DamarisFFT, Signalpath): for key in list(self.common_descriptions.keys()): if (key in other.description and self.common_descriptions[key]==other.description[key]): r.common_descriptions[key]=self.common_descriptions[key] - self.lock.release() return r # Accumulation - elif str(other.__class__) == "damaris.data.Accumulation.Accumulation": + elif isinstance(other, Accumulation): # Other empty (return) if not other.contains_data(): return @@ -759,7 +760,6 @@ class Accumulation(Errorable, Drawable, DamarisFFT, Signalpath): if not self.contains_data(): raise ValueError("Accumulation: You cant add integers/floats to an empty accumulation") else: - self.lock.acquire() for i in range(self.get_number_of_channels()): #Dont change errors and mean value @@ -767,16 +767,13 @@ class Accumulation(Errorable, Drawable, DamarisFFT, Signalpath): self.y_square[i] += (2*self.y[i]*other) + ((other**2)*self.n) self.y[i] += other*self.n self.lock.release() - return self # ADC_Result - elif type(other).__module__+"."+type(other).__name__ == "damaris.data.ADC_Result.ADC_Result": - + elif isinstance(other, ADC_Result): # Other empty (return) if not other.contains_data(): return self - # Self empty (copy) if not self.contains_data(): self.lock.acquire() @@ -835,7 +832,7 @@ class Accumulation(Errorable, Drawable, DamarisFFT, Signalpath): return self # Accumulation - elif type(other).__module__+"."+type(other).__name__ == "damaris.data.Accumulation.Accumulation": + elif isinstance(other, Accumulation): # Other empty (return) if not other.contains_data(): return @@ -869,14 +866,14 @@ class Accumulation(Errorable, Drawable, DamarisFFT, Signalpath): else: self.lock.acquire() if self.sampling_rate != other.get_sampling_rate(): - raise ValueError("Accumulation: You cant add accumulations with diffrent sampling-rates") + raise ValueError("Accumulation: You cant add accumulations with different sampling-rates") if len(self.y[0]) != len(other): - raise ValueError("Accumulation: You cant add accumulations with diffrent number of samples") + raise ValueError("Accumulation: You cant add accumulations with different number of samples") if len(self.y) != other.get_number_of_channels(): - raise ValueError("Accumulation: You cant add accumulations with diffrent number of channels") + raise ValueError("Accumulation: You cant add accumulations with different number of channels") for i in range(len(self.index)): if self.index[i] != other.get_index_bounds(i): - raise ValueError("Accumulation: You cant add accumulations with diffrent indexing") + raise ValueError("Accumulation: You cant add accumulations with different indexing") if self.uses_statistics() and not other.uses_statistics(): raise ValueError("Accumulation: You cant add non-error accumulations to accumulations with error") @@ -888,12 +885,21 @@ class Accumulation(Errorable, Drawable, DamarisFFT, Signalpath): self.n += other.n self.time_period=[min(self.time_period[0],other.time_period[0]), max(self.time_period[1],other.time_period[1])] - self.job_id = other.job_id # added by Oleg Petrov + self.job_id = other.job_id + # Removes mismatched common description keys if self.common_descriptions is not None and other.common_descriptions is not None: - for key in list(self.common_descriptions.keys()): - if not (key in other.description and - self.common_descriptions[key]==other.common_descriptions[key]): - del self.common_descriptions[key] + # Get all keys that exist in both dictionaries + common_keys = set(self.common_descriptions.keys()) & set(other.common_descriptions.keys()) + + # Find keys where values also match + matching_keys = { + key for key in common_keys + if self.common_descriptions[key] == other.common_descriptions[key] + } + + # Remove keys that don't match + for key in set(self.common_descriptions.keys()) - matching_keys: + del self.common_descriptions[key] self.set_title(self.__title_pattern % self.n) self.lock.release() diff --git a/src/tests/h5dump1_accu.ascii b/src/tests/h5dump1_accu.ascii new file mode 100644 index 0000000..4fbca43 --- /dev/null +++ b/src/tests/h5dump1_accu.ascii @@ -0,0 +1,46 @@ +HDF5 "testaccu.hdf5" { +DATASET "/name/accu_data" { + DATATYPE H5T_IEEE_F32LE + DATASPACE SIMPLE { ( 2, 4 ) / ( 2, 4 ) } + DATA { + (0,0): 10, 0, 15, 0, + (1,0): 20, 0, 25, 0 + } + ATTRIBUTE "CLASS" { + DATATYPE H5T_STRING { + STRSIZE 6; + STRPAD H5T_STR_NULLTERM; + CSET H5T_CSET_ASCII; + CTYPE H5T_C_S1; + } + DATASPACE SCALAR + DATA { + (0): "CARRAY" + } + } + ATTRIBUTE "TITLE" { + DATATYPE H5T_STRING { + STRSIZE 9; + STRPAD H5T_STR_NULLTERM; + CSET H5T_CSET_ASCII; + CTYPE H5T_C_S1; + } + DATASPACE SCALAR + DATA { + (0): "accu data" + } + } + ATTRIBUTE "VERSION" { + DATATYPE H5T_STRING { + STRSIZE 3; + STRPAD H5T_STR_NULLTERM; + CSET H5T_CSET_ASCII; + CTYPE H5T_C_S1; + } + DATASPACE SCALAR + DATA { + (0): "1.1" + } + } +} +} diff --git a/src/tests/test_ADC_Result.py b/src/tests/test_ADC_Result.py index 31ab277..1824739 100644 --- a/src/tests/test_ADC_Result.py +++ b/src/tests/test_ADC_Result.py @@ -18,6 +18,7 @@ class TestADCResult(unittest.TestCase): sampl_freq = 1000.0 job_id = 11 adc = ADC_Result(x, y, index, sampl_freq, desc, job_id, job_date) + adc.set_nChannels(2) return adc def test_constructor_with_no_arguments(self): @@ -162,37 +163,37 @@ class TestADCResult(unittest.TestCase): y = adc.y # test integer addition for i in range(adc.get_nChannels()): - self.assertIsNone(np.testing.assert_array_equal(adc.y[i]+10, y[i] + 10)) + np.testing.assert_array_equal(adc.y[i]+10, y[i] + 10, strict=True) adc += 10 for i in range(adc.get_nChannels()): - self.assertIsNone(np.testing.assert_array_equal(adc.y[i], y[i] + 10)) + np.testing.assert_array_equal(adc.y[i], y[i] + 10, strict=True) # test float addition for i in range(adc.get_nChannels()): - self.assertIsNone(np.testing.assert_array_equal(adc.y[i] + 10., y[i] + 10.)) + np.testing.assert_array_equal(adc.y[i] + 10., y[i] + 10., strict=True) adc += 10. for i in range(adc.get_nChannels()): - self.assertIsNone(np.testing.assert_array_equal(adc.y[i], y[i] + 10 + 10.)) + np.testing.assert_array_equal(adc.y[i], y[i] + 10 + 10., strict=True) def test_operator_sub_scalar(self): """ Test the functionality of __sub__ and __rsub__ """ adc = self.create_adc_result() - y = adc.y[0][1]=5 + y = adc.y # test integer subtraction for i in range(adc.get_nChannels()): - self.assertIsNone(np.testing.assert_array_equal(adc.y[i] - 10, y[i] - 10)) + np.testing.assert_array_equal(adc.y[i] - 10, y[i] - 10, strict=True) adc -= 10 for i in range(adc.get_nChannels()): - self.assertIsNone(np.testing.assert_array_equal(adc.y[i], y[i] - 10)) + np.testing.assert_array_equal(adc.y[i], y[i] - 10, strict=True) # test float subtraction for i in range(adc.get_nChannels()): - self.assertIsNone(np.testing.assert_array_equal(adc.y[i] - 10., y[i] - 10.)) + np.testing.assert_array_equal(adc.y[i] - 10., y[i] - 10., strict=True) adc -= 10. for i in range(adc.get_nChannels()): - self.assertIsNone(np.testing.assert_array_equal(adc.y[i], y[i] - 10 - 10.)) + np.testing.assert_array_equal(adc.y[i], y[i] - 10 - 10., strict=True) def test_operator_mul_scalar(self): """ @@ -203,17 +204,17 @@ class TestADCResult(unittest.TestCase): y = adc.y # test integer multiplication for i in range(adc.get_nChannels()): - self.assertIsNone(np.testing.assert_array_equal(adc.y[i] * 10, y[i] * 10)) + np.testing.assert_array_equal(adc.y[i] * 10, y[i] * 10, strict=True) adc *= 10 for i in range(adc.get_nChannels()): - self.assertIsNone(np.testing.assert_array_equal(adc.y[i], y[i] * 10)) + np.testing.assert_array_equal(adc.y[i], y[i] * 10, strict=True) # test float multiplication for i in range(adc.get_nChannels()): - self.assertIsNone(np.testing.assert_array_equal(adc.y[i] * 10.0, y[i] * 10.0)) + np.testing.assert_array_equal(adc.y[i] * 10.0, y[i] * 10.0, strict=True) adc *= 10.0 for i in range(adc.get_nChannels()): - self.assertIsNone(np.testing.assert_array_equal(adc.y[i], y[i] * 10 * 10.0)) + np.testing.assert_array_equal(adc.y[i], y[i] * 10 * 10.0, strict=True) def test_operator_truediv_scalar(self): """ @@ -222,10 +223,10 @@ class TestADCResult(unittest.TestCase): adc = self.create_adc_result() y = adc.y for i in range(adc.get_nChannels()): - self.assertIsNone(np.testing.assert_array_equal(adc.y[i]/10, y[i] / 10)) + np.testing.assert_array_equal(adc.y[i]/10, y[i] / 10, strict=True) adc /= 10 for i in range(adc.get_nChannels()): - self.assertIsNone(np.testing.assert_array_equal(adc.y[i], y[i] / 10)) + np.testing.assert_array_equal(adc.y[i], y[i] / 10, strict=True) def test_operator_floordiv_scalar(self): """ @@ -234,10 +235,10 @@ class TestADCResult(unittest.TestCase): adc = self.create_adc_result() y = adc.y for i in range(adc.get_nChannels()): - self.assertIsNone(np.testing.assert_array_equal(adc.y[i]//10, y[i] // 10)) + np.testing.assert_array_equal(adc.y[i]//10, y[i] // 10, strict=True) adc //= 10 for i in range(adc.get_nChannels()): - self.assertIsNone(np.testing.assert_array_equal(adc.y[i], y[i] // 10)) + np.testing.assert_array_equal(adc.y[i], y[i] // 10, strict=True) self.assertRaises(ValueError, adc.__floordiv__, 1.0) self.assertRaises(ValueError, adc.__rfloordiv__, 1.0) diff --git a/src/tests/test_Accumulation.py b/src/tests/test_Accumulation.py new file mode 100644 index 0000000..e094c80 --- /dev/null +++ b/src/tests/test_Accumulation.py @@ -0,0 +1,284 @@ +import os +import subprocess +import unittest +from datetime import datetime + +import numpy as np + +from data.ADC_Result import ADC_Result +from src.data.Accumulation import Accumulation + + +class TestAccumulation(unittest.TestCase): + @staticmethod + def create_adc_result(): + x = np.array([0.0, 1.0, 2.0]) + y = [np.array([10, 20, 30]), np.array([15, 25, 35])] + index = [(0, 1)] + job_date = datetime(2023, 1, 1) + desc = {"key": "value"} + sampl_freq = 1000.0 + job_id = 11 + adc = ADC_Result(x, y, index, sampl_freq, desc, job_id, job_date) + adc.set_nChannels(2) + return adc + + def test_constructor_with_no_arguments(self): + """Test initializer with no arguments.""" + accu = Accumulation() + self.assertFalse(accu.contains_data()) + self.assertEqual(accu.sampling_rate, 0) + self.assertEqual(len(accu.index), 0) + self.assertEqual(len(accu.x), 0) + self.assertEqual(len(accu.y), 0) + + def test_constructor_with_arguments(self): + """Test initializer with all arguments.""" + x = np.array([0.0, 1.0, 2.0]) + y = [np.array([10, 20, 30]), np.array([15, 25, 35])] + n = 1 + index = [(0, 2)] + sampl_freq = 1000.0 + + accu = Accumulation(x=x, + y=y, + y_2=None, + index=index, + sampl_freq=sampl_freq, + n=n) + + self.assertTrue(accu.contains_data()) + self.assertEqual(accu.sampling_rate, sampl_freq) + self.assertEqual(accu.x.all(), x.all()) + self.assertEqual(accu.y[0].all(), y[0].all()) + # TODO make testing strict with array types! + np.testing.assert_array_equal(accu.x, np.array(x, dtype="float32"), strict=False) + np.testing.assert_array_equal(accu.y[0], np.array(y[0], dtype="int16"), strict=False) + np.testing.assert_equal(accu.y[1], np.array(y[1], dtype="int16"), strict=False) + + def test_add_adc_result(self): + """Test adding sample space.""" + adc = self.create_adc_result() + accu = Accumulation() + accu += adc + self.assertEqual(len(adc.x), len(accu.x)) + self.assertEqual(len(adc.y[0]), len(accu.y[0])) + self.assertEqual(adc.index[-1], accu.index[-1]) + np.testing.assert_array_equal(accu.x, adc.x) + np.testing.assert_array_equal(accu.y[0], adc.y[0]) + + def test_add_accumulation(self): + """Test adding sample space.""" + adc1 = self.create_adc_result() + adc1.set_description("common", "somerandomtext") + adc1.set_description("same_variable", "somerandomtext_variable1") + adc1.set_description("variable1", "somerandomtext_variable1_1") + adc2 = self.create_adc_result() + adc2.set_description("common", "somerandomtext") + adc2.set_description("same_variable", "somerandomtext_variable2") + adc2.set_description("variable2", "somerandomtext_variable2_2") + accu1 = Accumulation() + accu1 += adc1 + print(accu1.common_descriptions) + accu2 = Accumulation() + accu2 += adc2 + accu2 += accu1 + + self.assertEqual(len(accu1.x), len(accu2.x)) + self.assertEqual(len(accu1.y[0]), len(accu2.y[0])) + self.assertEqual(accu1.index[-1], accu2.index[-1]) + self.assertEqual(accu1.common_descriptions["same_variable"], "somerandomtext_variable1") + self.assertEqual(accu2.common_descriptions["common"], "somerandomtext") + np.testing.assert_array_equal(accu2.x, accu1.x) + np.testing.assert_array_equal(accu2.y[0], accu1.y[0]*2) + + def test_get_accu_by_index(self): + """Test retrieving data by index.""" + accu = Accumulation() + adc = self.create_adc_result() + adc.index = [(0, 1), (1, 2)] + adc.job_id = 42 + accu += adc + sub_result = accu.get_accu_by_index(1) + self.assertEqual(sub_result.sampling_rate, 1000.0) + self.assertTrue(np.array_equal(sub_result.x, np.array([1.0, 2.0]))) + self.assertTrue(np.array_equal(sub_result.y[0], np.array([20, 30]))) + self.assertTrue(np.array_equal(sub_result.y[1], np.array([25, 35]))) + self.assertEqual(sub_result.index, [(0, 1)]) + # TODO Fix #13: self.assertEqual(sub_result.common_descriptions, {"key": "value"}) + + def test_write_to_csv_without_error(self): + """Test the functionality of writing to CSV.""" + from io import StringIO + x = np.array([0.0, 1.0]) + y = [np.array([10, 20]), np.array([15, 25])] + index = [(0, 1)] + job_date = datetime(2023, 1, 1) + desc = {"key": "value"} + sampl_freq = 1000.0 + job_id=9 + adc = ADC_Result(x, y, index, sampl_freq, desc, job_id, job_date) + accu = Accumulation() + accu += adc + output = StringIO() + + accu.write_to_csv(output, delimiter=",") + content = output.getvalue() + expected = ( + "# accumulation 1\n" + "# key : value\n" + "# t ch0_mean ch1_mean\n" + "0.000000e+00,1.000000e+01,1.500000e+01\n" + "1.000000e+00,2.000000e+01,2.500000e+01\n" + ) + self.assertEqual(content, expected) + + def test_write_to_csv_with_error(self): + """Test the functionality of writing to CSV.""" + from io import StringIO + x = np.array([0.0, 1.0]) + y = [np.array([10, 20]), np.array([15, 25])] + index = [(0, 1)] + job_date = datetime(2023, 1, 1) + desc = {"key": "value"} + sampl_freq = 1000.0 + job_id=9 + adc = ADC_Result(x, y, index, sampl_freq, desc, job_id, job_date) + accu = Accumulation(error=True) + accu += adc + output = StringIO() + + accu.write_to_csv(output, delimiter=",") + content = output.getvalue() + expected = ( + "# accumulation 1\n" + "# key : value\n" + "# t ch0_mean ch0_err ch1_mean ch1_err\n" + "0.000000e+00,1.000000e+01,0.000000e+00,1.500000e+01,0.000000e+00\n" + "1.000000e+00,2.000000e+01,0.000000e+00,2.500000e+01,0.000000e+00\n" + ) + self.assertEqual(content, expected) + + def test_write_to_hdf(self): + """Test the functionality of writing to HDF.""" + import tables + # do not change the values here, or you need to recreate the h5dump with new values + x = np.array([0.0, 1.0]) + y = [np.array([10, 20]), np.array([15, 25])] + index = [(0, 1)] + job_date = datetime(2023, 1, 1) + desc = {"key": "value"} + sampl_freq = 1000.0 + job_id = 10 + adc = ADC_Result(x, y, index, sampl_freq, desc, job_id, job_date) + accu = Accumulation() + accu += adc + # write out data + hdffile = tables.open_file("testaccu.hdf5", mode="w") + accu.write_to_hdf(hdffile, where="/", name="name", title="title", complib="zlib", complevel=3) + hdffile.close() + # read back data with h5dump utility (apt-get -y install hdf5-tools) + h5dump = subprocess.run(["h5dump", "-d", "/name/accu_data", "testaccu.hdf5"], capture_output=True) + content = h5dump.stdout.decode("utf-8") + with open("tests/h5dump1_accu.ascii", "r") as f: + expected = f.read() + self.assertEqual(content, expected) + os.unlink("testaccu.hdf5") + + def test_operator_len(self): + """Test the functionality of __len__""" + adc = self.create_adc_result() + accu = Accumulation() + accu += adc + self.assertEqual(len(accu), 3) + + def test_operator_add_scalar(self): + """ + Test the functionality of __add__ and __radd__ + """ + adc = self.create_adc_result() + accu = Accumulation() + accu += adc + y = adc.y + # test integer addition + for i in range(adc.get_nChannels()): + np.testing.assert_array_equal(accu.y[i]+10, y[i] + 10) + accu += 10 + for i in range(adc.get_nChannels()): + np.testing.assert_array_equal(accu.y[i], y[i] + 10) + + # test float addition + accu += 10. + for i in range(adc.get_nChannels()): + np.testing.assert_array_equal(accu.y[i], y[i] + 10 + 10.) + + def test_operator_sub_scalar(self): + """ + Test the functionality of __sub__ and __rsub__ + """ + adc = self.create_adc_result() + y = adc.y + accu = Accumulation() + accu += adc + + # test integer subtraction + accu -= 10 + for i in range(adc.get_nChannels()): + np.testing.assert_array_equal(accu.y[i], y[i] - 10) + + # test float subtraction + adc -= 10. + for i in range(adc.get_nChannels()): + np.testing.assert_array_equal(accu.y[i], y[i] - 10 - 10.) + + @unittest.skip("Not implmented") + def test_operator_mul_scalar(self): + """ + Test the functionality of __mul and __rmul__ + """ + + adc = self.create_adc_result() + y = adc.y + accu = Accumulation() + accu += adc + + # test integer multiplication + accu *= 10 + for i in range(adc.get_nChannels()): + np.testing.assert_array_equal(accu.y[i], y[i] * 10) + + # test float multiplication + accu *= 10.0 + for i in range(adc.get_nChannels()): + np.testing.assert_array_equal(accu.y[i], y[i] * 10 * 10.0) + + @unittest.skip("Not implmented") + def test_operator_truediv_scalar(self): + """ + Test the functionality of __div__ and __rdiv__ + """ + adc = self.create_adc_result() + y = adc.y + for i in range(adc.get_nChannels()): + np.testing.assert_array_equal(adc.y[i]/10, y[i] / 10, strict=True) + adc /= 10 + for i in range(adc.get_nChannels()): + np.testing.assert_array_equal(adc.y[i], y[i] / 10, strict=True) + + @unittest.skip("Not implmented") + def test_operator_floordiv_scalar(self): + """ + Test the functionality of __div__ and __rdiv__ + """ + adc = self.create_adc_result() + y = adc.y + for i in range(adc.get_nChannels()): + np.testing.assert_array_equal(adc.y[i]//10, y[i] // 10, strict=True) + adc //= 10 + for i in range(adc.get_nChannels()): + np.testing.assert_array_equal(adc.y[i], y[i] // 10, strict=True) + self.assertRaises(ValueError, adc.__floordiv__, 1.0) + self.assertRaises(ValueError, adc.__rfloordiv__, 1.0) + +if __name__ == "__main__": + unittest.main()