added tests for Accumulation and ADC_Result

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)
 
 

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()

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"
+      }
+   }
+}
+}

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)
 

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()