import os import subprocess import unittest from datetime import datetime import numpy as np from damaris.data.ADC_Result import ADC_Result from damaris.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_array_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") # Use path relative to project root test_dir = os.path.dirname(__file__) with open(os.path.join(test_dir, "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.) 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) def test_operator_truediv_scalar(self): """ Test the functionality of __truediv__ and __itruediv__ """ adc = self.create_adc_result() y = adc.y accu = Accumulation() accu += adc 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) def test_operator_floordiv_scalar(self): """ Test the functionality of __floordiv__ and __ifloordiv__ """ adc = self.create_adc_result() y = adc.y accu = Accumulation() accu += adc 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) self.assertRaises(ValueError, accu.__floordiv__, 1.0) self.assertRaises(ValueError, accu.__ifloordiv__, 1.0) if __name__ == "__main__": unittest.main()