From 27d05496ab9e81dfd370341594afb9f0949c8fca Mon Sep 17 00:00:00 2001 From: Markus Rosenstihl Date: Sun, 18 Oct 2026 15:11:58 +0200 Subject: [PATCH] fix y_square latent bug: keep it always a list of per-channel arrays Accumulation.__neg__ passed y_2 = numpy.array(self.y_square), storing a 2D ndarray. A later __iadd__ on an accumulation with ndarray y_square crashed with AttributeError at the y_square.append sites (992, 1064). - __neg__ now passes a per-channel copy (values unchanged: sums of squares are invariant under negation) - __init__ normalizes ndarray y_2 to a list, so external callers cannot reintroduce the bug - y_square annotated as _YChannels Clears the remaining pyrefly errors at Accumulation.py:992 and :1064 (74 -> 72). Unit tests reproduce both crashes and pass with the fix. Generated by Mistral Vibe. Co-Authored-By: Mistral Vibe --- src/damaris/data/Accumulation.py | 6 ++-- tests/test_Accumulation.py | 48 ++++++++++++++++++++++++++++++++ 2 files changed, 52 insertions(+), 2 deletions(-) diff --git a/src/damaris/data/Accumulation.py b/src/damaris/data/Accumulation.py index 6b2666b..d6fd85f 100644 --- a/src/damaris/data/Accumulation.py +++ b/src/damaris/data/Accumulation.py @@ -77,7 +77,8 @@ class Accumulation(Errorable, Drawable, DamarisFFT, Signalpath): self.use_error = error if self.uses_statistics(): if (y_2 is not None): - self.y_square = y_2 + # y_square is always kept as a list of per-channel arrays + self.y_square: _YChannels = y_2 if isinstance(y_2, list) else list(y_2) elif (y_2 is None) : self.y_square = [] else: @@ -1147,7 +1148,8 @@ class Accumulation(Errorable, Drawable, DamarisFFT, Signalpath): tmp_y.append(numpy.array(-self.y[i], dtype="float32")) if self.uses_statistics(): - r = Accumulation(x = numpy.array(self.x, dtype="float32"), y = tmp_y, y_2 = numpy.array(self.y_square), n = self.n, index = self.index, sampl_freq = self.sampling_rate, error = True) + tmp_y_square = [numpy.array(sq) for sq in self.y_square] + r = Accumulation(x = numpy.array(self.x, dtype="float32"), y = tmp_y, y_2 = tmp_y_square, n = self.n, index = self.index, sampl_freq = self.sampling_rate, error = True) else: r = Accumulation(x = numpy.array(self.x, dtype="float32"), y = tmp_y, n = self.n, index = self.index, sampl_freq = self.sampling_rate, error = False) self._copy_provenance(r) diff --git a/tests/test_Accumulation.py b/tests/test_Accumulation.py index c2de3ea..760aed7 100644 --- a/tests/test_Accumulation.py +++ b/tests/test_Accumulation.py @@ -382,5 +382,53 @@ class TestAccumulation(unittest.TestCase): 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]) + if __name__ == "__main__": unittest.main()