From 65611b98a03e45053ba26569960013ab41c0b613 Mon Sep 17 00:00:00 2001 From: Markus Rosenstihl Date: Thu, 19 Mar 2026 17:30:47 +0100 Subject: [PATCH] Implemnted tests for ADC_Result --- src/tests/test_ADC_Result.py | 263 +++++++++++++++++++++++++++++++++++ 1 file changed, 263 insertions(+) create mode 100644 src/tests/test_ADC_Result.py diff --git a/src/tests/test_ADC_Result.py b/src/tests/test_ADC_Result.py new file mode 100644 index 0000000..b38f0db --- /dev/null +++ b/src/tests/test_ADC_Result.py @@ -0,0 +1,263 @@ +import os +import subprocess +import unittest +from datetime import datetime + +import numpy as np +from src.data.ADC_Result import ADC_Result + + +class TestADCResult(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) + return adc + + def test_constructor_with_no_arguments(self): + """Test initializer with no arguments.""" + adc = ADC_Result() + self.assertFalse(adc.contains_data()) + self.assertEqual(adc.sampling_rate, 0) + self.assertEqual(len(adc.index), 0) + self.assertEqual(len(adc.x), 0) + self.assertEqual(len(adc.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])] + index = [(0, 2)] + sampl_freq = 1000.0 + desc = {"key1": "value1"} + job_id = 1 + job_date = datetime(2023, 1, 1) + + adc = ADC_Result(x, y, index, sampl_freq, desc, job_id, job_date) + + self.assertTrue(adc.contains_data()) + self.assertEqual(adc.sampling_rate, sampl_freq) + self.assertEqual(adc.x.all(), x.all()) + self.assertEqual(adc.y[0].all(), y[0].all()) + self.assertEqual(adc.description, desc) + self.assertEqual(adc.job_id, job_id) + self.assertEqual(adc.job_date, job_date) + self.assertTrue(np.array_equal(adc.x, np.array(x, dtype="float32"))) + self.assertTrue(np.array_equal(adc.y[0], np.array(y[0], dtype="int16"))) + self.assertTrue(np.array_equal(adc.y[1], np.array(y[1], dtype="int16"))) + + def test_create_data_space(self): + """Test creating a new data space.""" + adc = ADC_Result() + adc.create_data_space(channels=2, samples=3) + + self.assertTrue(adc.contains_data()) + self.assertEqual(len(adc.y), 2) + self.assertEqual(len(adc.x), 3) + self.assertTrue(np.array_equal(adc.x, np.zeros(3, dtype="float32"))) + self.assertTrue(np.array_equal(adc.y[0], np.zeros(3, dtype="int16"))) + self.assertEqual(adc.index, [(0, 2)]) + + def test_add_sample_space(self): + """Test adding sample space.""" + adc = ADC_Result() + adc.create_data_space(channels=1, samples=3) + adc.add_sample_space(2) + + self.assertEqual(len(adc.x), 5) + self.assertEqual(len(adc.y[0]), 5) + self.assertEqual(adc.index[-1], (3, 4)) + + def test_get_result_by_index(self): + """Test retrieving data by index.""" + adc = self.create_adc_result() + adc.index = [(0, 1), (1, 2)] + adc.job_id = 42 + sub_result = adc.get_result_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)]) + self.assertEqual(sub_result.description, {"key": "value"}) + self.assertEqual(sub_result.job_id, 42) + + def test_set_sampling_rate(self): + """Test updating the sampling rate.""" + adc = ADC_Result() + adc.set_sampling_rate(5000.0) + self.assertEqual(adc.get_sampling_rate(), 5000.0) + + def test_set_nChannels(self): + """Test setting the number of channels.""" + adc = ADC_Result() + adc.set_nChannels(5) + self.assertEqual(adc.get_nChannels(), 5) + + def test_write_to_csv(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) + output = StringIO() + + adc.write_to_csv(output, delimiter=",") + content = output.getvalue() + expected = ( + "# adc_result\n" + "# t y0 y1 ...\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_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) + # write out data + hdffile = tables.open_file("test.hdf5", mode="w") + adc.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/adc_data", "test.hdf5"], capture_output=True) + content = h5dump.stdout.decode("utf-8") + with open("h5dump1_adc_data.ascii", "r") as f: + expected = f.read() + self.assertEqual(content, expected) + os.unlink("test.hdf5") + + def test_operator_len(self): + """Test the functionality of __len__""" + adc = self.create_adc_result() + self.assertEqual(len(adc), 3) + + def test_operator_add_scalar(self): + """ + Test the functionality of __add__ and __radd__ + """ + adc = self.create_adc_result() + 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)) + adc += 10 + for i in range(adc.get_nChannels()): + self.assertIsNone(np.testing.assert_array_equal(adc.y[i], y[i] + 10)) + + # test float addition + for i in range(adc.get_nChannels()): + self.assertIsNone(np.testing.assert_array_equal(adc.y[i] + 10., y[i] + 10.)) + adc += 10. + for i in range(adc.get_nChannels()): + self.assertIsNone(np.testing.assert_array_equal(adc.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 + # test integer subtraction + for i in range(adc.get_nChannels()): + self.assertIsNone(np.testing.assert_array_equal(adc.y[i] - 10, y[i] - 10)) + adc -= 10 + for i in range(adc.get_nChannels()): + self.assertIsNone(np.testing.assert_array_equal(adc.y[i], y[i] - 10)) + + # test float subtraction + for i in range(adc.get_nChannels()): + self.assertIsNone(np.testing.assert_array_equal(adc.y[i] - 10., y[i] - 10.)) + adc -= 10. + for i in range(adc.get_nChannels()): + self.assertIsNone(np.testing.assert_array_equal(adc.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 + # test integer multiplication + for i in range(adc.get_nChannels()): + self.assertIsNone(np.testing.assert_array_equal(adc.y[i] * 10, y[i] * 10)) + adc *= 10 + for i in range(adc.get_nChannels()): + self.assertIsNone(np.testing.assert_array_equal(adc.y[i], y[i] * 10)) + + # 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)) + 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)) + + 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()): + self.assertIsNone(np.testing.assert_array_equal(adc.y[i]/10, y[i] / 10)) + adc /= 10 + for i in range(adc.get_nChannels()): + self.assertIsNone(np.testing.assert_array_equal(adc.y[i], y[i] / 10)) + + 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()): + self.assertIsNone(np.testing.assert_array_equal(adc.y[i]//10, y[i] // 10)) + adc //= 10 + for i in range(adc.get_nChannels()): + self.assertIsNone(np.testing.assert_array_equal(adc.y[i], y[i] // 10)) + self.assertRaises(ValueError, adc.__floordiv__, 1.0) + self.assertRaises(ValueError, adc.__rfloordiv__, 1.0) + + def test_operator_overload_adc_result_exception(self): + """ + Test the ValueError raised when ADC_Result is used as an operand in an arithmetic operation + """ + adc1 = self.create_adc_result() + adc2 = self.create_adc_result() + self.assertRaises(ValueError, adc1.__add__, adc2) + self.assertRaises(ValueError, adc1.__radd__, adc2) + self.assertRaises(ValueError, adc1.__sub__, adc2) + self.assertRaises(ValueError, adc1.__rsub__, adc2) + self.assertRaises(ValueError, adc1.__mul__, adc2) + self.assertRaises(ValueError, adc1.__rmul__, adc2) + self.assertRaises(ValueError, adc1.__truediv__, adc2) + self.assertRaises(ValueError, adc1.__rtruediv__, adc2) + self.assertRaises(ValueError, adc1.__floordiv__, adc2) + self.assertRaises(ValueError, adc1.__rfloordiv__, adc2) + self.assertRaises(ValueError, adc1.__pow__, adc2) + +if __name__ == "__main__": + unittest.main()