Public Access
587 lines
22 KiB
Python
587 lines
22 KiB
Python
import os
|
|
import subprocess
|
|
import threading
|
|
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_shared_measurement_statistics_use_accumulation_means(self):
|
|
accu = Accumulation()
|
|
accu += self.create_adc_result()
|
|
np.testing.assert_allclose(accu.get_ymean(), [20, 25])
|
|
np.testing.assert_allclose(accu.get_ymin_channels(), [10, 15])
|
|
np.testing.assert_allclose(accu.get_ymax_channels(), [30, 35])
|
|
self.assertAlmostEqual(accu.get_phase(), np.degrees(np.arctan2(15, 10)), places=6)
|
|
|
|
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)])
|
|
self.assertEqual(sub_result.common_descriptions, {"key": "value"})
|
|
|
|
def test_get_accu_by_index_copies_is_clipped(self):
|
|
"""Test that get_accu_by_index properly copies is_clipped."""
|
|
adc = self.create_adc_result()
|
|
adc.index = [(0, 1), (1, 2)]
|
|
accu = Accumulation()
|
|
accu += adc
|
|
accu.is_clipped = True
|
|
|
|
sub_result = accu.get_accu_by_index(1)
|
|
|
|
self.assertEqual(sub_result.is_clipped, True)
|
|
|
|
def test_get_accu_by_index_data_independence(self):
|
|
"""Test that get_accu_by_index returns independent data (not shared views)."""
|
|
adc = self.create_adc_result()
|
|
adc.index = [(0, 1), (1, 2)]
|
|
accu = Accumulation()
|
|
accu += adc
|
|
original_y0 = accu.y[0].copy()
|
|
|
|
sub_result = accu.get_accu_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(accu.y[0], original_y0))
|
|
# x array should also be independent
|
|
self.assertEqual(accu.x[1], 1.0)
|
|
|
|
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
|
|
accu -= 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)
|
|
|
|
def test_add_invalid_operand_raises(self):
|
|
"""accu + bogus must raise ValueError instead of returning None."""
|
|
adc = self.create_adc_result()
|
|
accu = Accumulation()
|
|
accu += adc
|
|
self.assertRaises(ValueError, accu.__add__, "foo")
|
|
self.assertRaises(ValueError, accu.__iadd__, "foo")
|
|
|
|
def test_neg_empty_raises(self):
|
|
"""-empty accumulation must raise ValueError."""
|
|
self.assertRaises(ValueError, Accumulation().__neg__)
|
|
|
|
def test_add_adc_result_does_not_modify_job_ids(self):
|
|
"""accu + adc must not insert the adc job id into accu.job_ids."""
|
|
accu = Accumulation()
|
|
accu += self.create_adc_result()
|
|
n_before = len(accu.job_ids)
|
|
adc2 = self.create_adc_result()
|
|
adc2.job_id = 12
|
|
r = accu + adc2
|
|
self.assertIn(12, r.job_ids)
|
|
self.assertNotIn(12, accu.job_ids)
|
|
self.assertEqual(len(accu.job_ids), n_before)
|
|
|
|
def test_empty_plus_accumulation_copies_data(self):
|
|
"""empty_accu + accu must copy arrays and metadata, not alias them."""
|
|
adc = self.create_adc_result()
|
|
accu = Accumulation()
|
|
accu += adc
|
|
accu.is_clipped = True
|
|
|
|
r = Accumulation() + accu
|
|
self.assertIsNot(r.y[0], accu.y[0])
|
|
self.assertIsNot(r.job_ids, accu.job_ids)
|
|
self.assertTrue(r.is_clipped)
|
|
# in-place operation on the result must leave the operand unchanged
|
|
r *= 2
|
|
np.testing.assert_array_equal(accu.y[0], np.array([10, 20, 30], dtype="float32"))
|
|
|
|
def test_empty_iadd_accumulation_without_common_descriptions(self):
|
|
"""An accumulation without common descriptions can be copied in-place."""
|
|
other = Accumulation(x=np.array([0.0, 1.0]),
|
|
y=[np.array([10.0, 20.0])],
|
|
n=1,
|
|
index=[(0, 1)],
|
|
sampl_freq=1000.0)
|
|
accu = Accumulation()
|
|
|
|
accu += other
|
|
|
|
self.assertIsNone(accu.common_descriptions)
|
|
np.testing.assert_array_equal(accu.y[0], other.y[0])
|
|
|
|
def test_accumulation_plus_accumulation_keeps_both_job_ids(self):
|
|
"""accu1 + accu2 must keep the provenance of both operands."""
|
|
adc1 = self.create_adc_result()
|
|
adc2 = self.create_adc_result()
|
|
adc2.job_id = 12
|
|
accu1 = Accumulation()
|
|
accu1 += adc1
|
|
accu2 = Accumulation()
|
|
accu2 += adc2
|
|
r = accu1 + accu2
|
|
self.assertEqual(set(r.job_ids.keys()), {11, 12})
|
|
self.assertEqual(set(accu1.job_ids.keys()), {11})
|
|
self.assertEqual(set(accu2.job_ids.keys()), {12})
|
|
|
|
def test_empty_error_plus_non_error_raises(self):
|
|
"""error accumulation + non-error accumulation must raise ValueError."""
|
|
accu = Accumulation()
|
|
accu += self.create_adc_result()
|
|
empty_err = Accumulation(error=True)
|
|
self.assertRaises(ValueError, empty_err.__add__, accu)
|
|
free = []
|
|
def check_lock():
|
|
if empty_err.lock.acquire(timeout=1):
|
|
free.append(True)
|
|
empty_err.lock.release()
|
|
t = threading.Thread(target=check_lock)
|
|
t.start()
|
|
t.join(3)
|
|
self.assertTrue(free, "lock leaked after ValueError in __add__")
|
|
|
|
def test_neg_keeps_y_square_as_list(self):
|
|
"""-accu must store y_square as a list of per-channel arrays."""
|
|
x = np.array([0.0, 1.0, 2.0])
|
|
y = [np.array([10, 20, 30], dtype="float32")]
|
|
y_square = [np.array([100.0, 400.0, 900.0], dtype="float32")]
|
|
accu = Accumulation(x=x, y=y, y_2=y_square, n=2, index=[(0, 2)], sampl_freq=1000.0, error=True)
|
|
|
|
r = -accu
|
|
|
|
self.assertIsInstance(r.y_square, list)
|
|
# sums of squares are invariant under negation
|
|
np.testing.assert_array_equal(r.get_ysquare(0), y_square[0])
|
|
|
|
def test_iadd_adc_after_ndarray_y_2(self):
|
|
"""empty accu with ndarray y_square += adc must not crash (was AttributeError)."""
|
|
y_square_2d = np.array([[100.0, 400.0, 900.0]], dtype="float32")
|
|
accu = Accumulation(y_2=y_square_2d, error=True)
|
|
|
|
accu += self.create_adc_result()
|
|
|
|
self.assertIsInstance(accu.y_square, list)
|
|
np.testing.assert_array_equal(accu.get_ydata(0), np.array([10, 20, 30], dtype="float32"))
|
|
np.testing.assert_array_equal(accu.get_ysquare(0), np.array([10, 20, 30], dtype="float32") ** 2)
|
|
|
|
def test_iadd_accumulation_after_ndarray_y_2(self):
|
|
"""empty accu with ndarray y_square += accu must not crash (was AttributeError)."""
|
|
full = Accumulation(error=True)
|
|
full += self.create_adc_result()
|
|
y_square_2d = np.array([[100.0, 400.0, 900.0]], dtype="float32")
|
|
accu = Accumulation(y_2=y_square_2d, error=True)
|
|
|
|
accu += full
|
|
|
|
self.assertIsInstance(accu.y_square, list)
|
|
np.testing.assert_array_equal(accu.get_ydata(0), np.array([10, 20, 30], dtype="float32"))
|
|
np.testing.assert_array_equal(accu.get_ysquare(0), np.array([10, 20, 30], dtype="float32") ** 2)
|
|
|
|
def test_constructor_normalizes_ndarray_y_2(self):
|
|
"""an ndarray passed as y_2 must be stored as a list of channel arrays."""
|
|
x = np.array([0.0, 1.0, 2.0])
|
|
y = [np.array([10, 20, 30], dtype="float32")]
|
|
y_square_2d = np.array([[100.0, 400.0, 900.0]], dtype="float32")
|
|
|
|
accu = Accumulation(x=x, y=y, y_2=y_square_2d, n=2, index=[(0, 2)], sampl_freq=1000.0, error=True)
|
|
|
|
self.assertIsInstance(accu.y_square, list)
|
|
np.testing.assert_array_equal(accu.get_ysquare(0), y_square_2d[0])
|
|
|
|
def test_add_accumulations_without_time_period(self):
|
|
"""accu + accu without time information must not crash."""
|
|
accu1 = Accumulation()
|
|
accu1 += self.create_adc_result()
|
|
accu2 = Accumulation()
|
|
accu2 += self.create_adc_result()
|
|
accu1.time_period = None
|
|
accu2.time_period = []
|
|
|
|
r = accu1 + accu2
|
|
|
|
self.assertIsNone(r.time_period)
|
|
|
|
def test_iadd_accumulation_without_time_period(self):
|
|
"""accu += accu without time information must not crash (was IndexError)."""
|
|
accu1 = Accumulation()
|
|
accu1 += self.create_adc_result()
|
|
accu2 = Accumulation()
|
|
accu2 += self.create_adc_result()
|
|
accu1.time_period = []
|
|
accu2.time_period = []
|
|
|
|
accu1 += accu2
|
|
|
|
self.assertIsNone(accu1.time_period)
|
|
|
|
def test_add_adc_without_time_period(self):
|
|
"""accu + adc without time information must not crash."""
|
|
accu = Accumulation()
|
|
accu += self.create_adc_result()
|
|
accu.time_period = []
|
|
|
|
r = accu + self.create_adc_result()
|
|
|
|
self.assertIsNone(r.time_period)
|
|
|
|
def test_time_period_merged_on_add(self):
|
|
"""accu + accu with both time periods set merges to min/max."""
|
|
accu1 = Accumulation()
|
|
accu1 += self.create_adc_result()
|
|
accu2 = Accumulation()
|
|
accu2 += self.create_adc_result()
|
|
accu1.time_period = [datetime(2023, 1, 3), datetime(2023, 1, 5)]
|
|
accu2.time_period = [datetime(2023, 1, 2), datetime(2023, 1, 7)]
|
|
|
|
r = accu1 + accu2
|
|
|
|
self.assertEqual(r.time_period, [datetime(2023, 1, 2), datetime(2023, 1, 7)])
|
|
|
|
def test_time_period_merged_with_adc_job_date(self):
|
|
"""accu + adc merges self.time_period with other.job_date."""
|
|
accu = Accumulation()
|
|
accu += self.create_adc_result()
|
|
accu.time_period = [datetime(2023, 1, 3), datetime(2023, 1, 5)]
|
|
adc = self.create_adc_result()
|
|
adc.job_date = datetime(2023, 1, 1)
|
|
|
|
r = accu + adc
|
|
|
|
self.assertEqual(r.time_period, [datetime(2023, 1, 1), datetime(2023, 1, 5)])
|
|
|
|
def test_iadd_adc_without_job_date_sets_time_period_none(self):
|
|
"""empty accu += adc without job_date gets time_period None."""
|
|
accu = Accumulation()
|
|
adc = self.create_adc_result()
|
|
adc.job_date = None
|
|
|
|
accu += adc
|
|
|
|
self.assertIsNone(accu.time_period)
|
|
|
|
def test_get_accu_by_index_without_time_period(self):
|
|
"""get_accu_by_index without time information must not crash."""
|
|
accu = Accumulation()
|
|
adc = self.create_adc_result()
|
|
adc.index = [(0, 1), (1, 2)]
|
|
accu += adc
|
|
accu.time_period = None
|
|
|
|
r = accu.get_accu_by_index(0)
|
|
|
|
self.assertIsNone(r.time_period)
|
|
|
|
def test_write_to_tecmag_without_time_period_raises(self):
|
|
"""write_to_tecmag without time information must raise ValueError."""
|
|
accu = Accumulation()
|
|
accu += self.create_adc_result()
|
|
accu.time_period = None
|
|
|
|
self.assertRaises(ValueError, accu.write_to_tecmag, "/tmp/damaris_tecmag_no_time.tnt")
|
|
self.assertTrue(accu.lock.acquire(timeout=1), "lock leaked after ValueError in write_to_tecmag")
|
|
accu.lock.release()
|
|
|
|
def test_empty_state_accessors(self):
|
|
"""Empty Accumulation accessors must return ndarrays, min/max helpers 0."""
|
|
accu = Accumulation()
|
|
self.assertIsInstance(accu.x, np.ndarray)
|
|
self.assertEqual(accu.x.size, 0)
|
|
self.assertIsInstance(accu.get_ydata(), np.ndarray)
|
|
self.assertEqual(accu.get_ydata().size, 0)
|
|
self.assertIsInstance(accu.get_yerr(0), np.ndarray)
|
|
self.assertEqual(accu.get_yerr(0).size, 0)
|
|
self.assertEqual(accu.get_xmin(), 0)
|
|
self.assertEqual(accu.get_xmax(), 0)
|
|
self.assertEqual(accu.get_ymin(), 0)
|
|
self.assertEqual(accu.get_ymax(), 0)
|
|
|
|
def test_get_yerr_without_statistics_on_empty(self):
|
|
"""get_yerr on an empty accumulation without statistics must not crash."""
|
|
accu = Accumulation(error=False)
|
|
err = accu.get_yerr(0)
|
|
self.assertIsInstance(err, np.ndarray)
|
|
self.assertEqual(err.size, 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]]
|
|
y_2 = [[100.0, 400.0, 900.0], [225.0, 625.0, 1225.0]]
|
|
accu = Accumulation(x=x, y=y, y_2=y_2, n=1,
|
|
index=[(0, 2)], sampl_freq=1000.0, error=True)
|
|
self.assertIsInstance(accu.x, np.ndarray)
|
|
np.testing.assert_array_equal(accu.x, [0.0, 1.0, 2.0])
|
|
for ch in accu.y:
|
|
self.assertIsInstance(ch, np.ndarray)
|
|
for ch in accu.y_square:
|
|
self.assertIsInstance(ch, np.ndarray)
|
|
np.testing.assert_array_equal(accu.y[0], [10, 20, 30])
|
|
np.testing.assert_array_equal(accu.y_square[0], [100.0, 400.0, 900.0])
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|