import os import subprocess import unittest from datetime import datetime import numpy as np from damaris.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) adc.set_nChannels(2) 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)]) self.assertEqual(adc.nChannels, 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_get_result_by_index_copies_nChannels_and_is_clipped(self): """Test that get_result_by_index properly copies nChannels and is_clipped.""" adc = self.create_adc_result() adc.index = [(0, 1), (1, 2)] adc.set_nChannels(2) adc.is_clipped = True sub_result = adc.get_result_by_index(1) self.assertEqual(sub_result.nChannels, 2) self.assertEqual(sub_result.is_clipped, True) def test_get_result_by_index_data_independence(self): """Test that get_result_by_index returns independent data (not shared views).""" adc = self.create_adc_result() adc.index = [(0, 1), (1, 2)] original_x = adc.x.copy() original_data = adc.y[0].copy() sub_result = adc.get_result_by_index(1) sub_result.y[0][0] = 9999 sub_result.x[0] = 999.0 self.assertEqual(sub_result.y[0][0], 9999) self.assertTrue(np.array_equal(adc.y[0], original_data)) self.assertTrue(np.array_equal(adc.x, original_x)) 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") test_dir = os.path.dirname(__file__) with open(os.path.join(test_dir, "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()): 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).astype("float32"), strict=True) # test float addition for i in range(adc.get_nChannels()): np.testing.assert_array_equal(adc.y[i] + 10., (y[i] + 10 + 10.).astype("float32"), strict=True) adc += 10. for i in range(adc.get_nChannels()): np.testing.assert_array_equal(adc.y[i], (y[i] + 10 + 10.).astype("float32"), strict=True) 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()): 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).astype("float32"), strict=True) # test float subtraction for i in range(adc.get_nChannels()): np.testing.assert_array_equal(adc.y[i] - 10., (y[i] - 10 - 10.).astype("float32"), strict=True) adc -= 10. for i in range(adc.get_nChannels()): np.testing.assert_array_equal(adc.y[i], (y[i] - 10 - 10.).astype("float32"), strict=True) 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()): 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).astype("float32"), strict=True) # test float multiplication for i in range(adc.get_nChannels()): np.testing.assert_array_equal(adc.y[i] * 10.0, (y[i] * 10 * 10.0).astype("float32"), strict=True) adc *= 10.0 for i in range(adc.get_nChannels()): np.testing.assert_array_equal(adc.y[i], (y[i] * 10 * 10.0).astype("float32"), strict=True) 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).astype("float32"), strict=True) 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).astype("float32"), strict=True) self.assertRaises(ValueError, adc.__floordiv__, 1.0) self.assertRaises(ValueError, adc.__rfloordiv__, 1.0) def test_operators_keep_channel_metadata(self): """operator results keep nChannels/is_clipped identical to the operand""" adc = self.create_adc_result() adc.is_clipped = True results = [adc + 1, adc - 1, adc * 2, adc ** 2, adc / 2., adc // 2, 10 + adc, 10 - adc, 2 * adc, 100. / adc, 100 // adc, -adc] for r in results: self.assertEqual(r.get_nChannels(), adc.get_nChannels()) self.assertEqual(r.is_clipped, adc.is_clipped) def test_hdf_roundtrip_keeps_nchannels(self): """read_from_hdf must restore the number of channels""" import tempfile import tables from damaris.data.ADC_Result import read_from_hdf adc = self.create_adc_result() with tempfile.TemporaryDirectory() as tmpdir: path = os.path.join(tmpdir, "adc.h5") with tables.open_file(path, "w") as h5: adc.write_to_hdf(h5, h5.root, "adc", "test") with tables.open_file(path, "r") as h5: adc2 = read_from_hdf(h5.root.adc) self.assertIsNotNone(adc2) self.assertTrue(adc2.contains_data()) self.assertEqual(adc2.get_nChannels(), adc.get_nChannels()) 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) def test_empty_state_types(self): """Empty ADC_Result must hold an ndarray x and a list-of-ndarray y.""" adc = ADC_Result() self.assertIsInstance(adc.x, np.ndarray) self.assertIsInstance(adc.y, list) self.assertEqual(adc.x.size, 0) self.assertEqual(adc.get_xmin(), 0) self.assertEqual(adc.get_xmax(), 0) self.assertEqual(adc.get_xminpos(), 0) self.assertEqual(adc.get_ymin(), 0) self.assertEqual(adc.get_ymax(), 0) self.assertEqual(adc.get_yminpos(), 0) def test_constructor_coerces_lists(self): """Plain lists passed to the constructor must become numpy arrays.""" x = [0.0, 1.0, 2.0] y = [[10, 20, 30], [15, 25, 35]] index = [(0, 2)] adc = ADC_Result(x, y, index, 1000.0, {"key": "value"}, 1, datetime(2023, 1, 1)) self.assertIsInstance(adc.x, np.ndarray) np.testing.assert_array_equal(adc.x, [0.0, 1.0, 2.0]) for ch in adc.y: self.assertIsInstance(ch, np.ndarray) np.testing.assert_array_equal(adc.y[0], [10, 20, 30]) np.testing.assert_array_equal(adc.y[1], [15, 25, 35]) if __name__ == "__main__": unittest.main()