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 <vibe@mistral.ai>
This commit is contained in:
markusro committed 2026-10-18 15:11:58 +02:00
1 parent c797aeb387
commit 27d05496ab
2 files changed
+52 -2

No files matched your search

+4 -2
View File
@@ -77,7 +77,8 @@ class Accumulation(Errorable, Drawable, DamarisFFT, Signalpath):
self.use_error = error self.use_error = error
if self.uses_statistics(): if self.uses_statistics():
if (y_2 is not None): 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) : elif (y_2 is None) :
self.y_square = [] self.y_square = []
else: else:
@@ -1147,7 +1148,8 @@ class Accumulation(Errorable, Drawable, DamarisFFT, Signalpath):
tmp_y.append(numpy.array(-self.y[i], dtype="float32")) tmp_y.append(numpy.array(-self.y[i], dtype="float32"))
if self.uses_statistics(): 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: 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) 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) self._copy_provenance(r)
+48
View File
@@ -382,5 +382,53 @@ class TestAccumulation(unittest.TestCase):
t.join(3) t.join(3)
self.assertTrue(free, "lock leaked after ValueError in __add__") 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__": if __name__ == "__main__":
unittest.main() unittest.main()