Public Access
362 lines
15 KiB
Python
362 lines
15 KiB
Python
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_shared_measurement_statistics(self):
|
|
adc = self.create_adc_result()
|
|
np.testing.assert_allclose(adc.get_ymean(), [20, 25])
|
|
np.testing.assert_allclose(adc.get_ymin_channels(), [10, 15])
|
|
np.testing.assert_allclose(adc.get_ymax_channels(), [30, 35])
|
|
np.testing.assert_allclose(adc.get_yrms(),
|
|
[np.sqrt(np.mean(np.array([10, 20, 30]) ** 2)),
|
|
np.sqrt(np.mean(np.array([15, 25, 35]) ** 2))])
|
|
np.testing.assert_allclose(adc.get_ystd(), [np.std([10, 20, 30]),
|
|
np.std([15, 25, 35])])
|
|
|
|
def test_shared_phase_uses_public_channel_values(self):
|
|
adc = self.create_adc_result()
|
|
self.assertAlmostEqual(adc.get_phase(), np.degrees(np.arctan2(15, 10)))
|
|
|
|
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()
|