diff --git a/src/damaris/data/Accumulation.py b/src/damaris/data/Accumulation.py index 65eef09..a995a4a 100644 --- a/src/damaris/data/Accumulation.py +++ b/src/damaris/data/Accumulation.py @@ -617,15 +617,18 @@ class Accumulation(Errorable, Drawable, DamarisFFT, Signalpath): 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 = self.use_error) r.job_id = self.job_id + r.job_ids = self.job_ids.copy() + r.time_period = self.time_period[:] if self.time_period is not None else None + if self.common_descriptions is not None: + r.common_descriptions = self.common_descriptions.copy() self.lock.release() return r # ADC_Result elif isinstance(other, ADC_Result): - # Other empty (return) - # todo: this is seems to be bugy!!!! (Achim) + # Other empty (return copy of self) if not other.contains_data(): - return + return self + 0 # Self empty (copy ADC_Result) if not self.contains_data(): tmp_y = [] @@ -691,9 +694,9 @@ class Accumulation(Errorable, Drawable, DamarisFFT, Signalpath): # Accumulation elif isinstance(other, Accumulation): - # Other empty (return) + # Other empty (return copy of self) if not other.contains_data(): - return + return self + 0 # Self empty (copy) if not self.contains_data(): @@ -776,7 +779,141 @@ class Accumulation(Errorable, Drawable, DamarisFFT, Signalpath): def __rsub__(self, other): """Redefining other - self""" - return self.__neg__(self.__add__(-other)) + return (-self) + other + + + def __mul__(self, other): + """Redefining self * other (scalar)""" + if isinstance(other, (int, float)): + if not self.contains_data(): + raise ValueError("Accumulation: You cant multiply an empty accumulation") + self.lock.acquire() + tmp_y = [] + tmp_ysquare = [] + for i in range(self.get_number_of_channels()): + tmp_y.append(self.y[i] * other) + if self.uses_statistics(): + tmp_ysquare.append(self.y_square[i] * (other**2)) + + if self.uses_statistics(): + r = Accumulation(x = numpy.array(self.x, dtype="float32"), y = tmp_y, y_2 = tmp_ysquare, n = self.n, index = self.index, sampl_freq = self.sampling_rate, error = self.use_error) + 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 = self.use_error) + r.job_id = self.job_id + r.job_ids = self.job_ids.copy() + r.time_period = self.time_period[:] if self.time_period is not None else None + if self.common_descriptions is not None: + r.common_descriptions = self.common_descriptions.copy() + self.lock.release() + return r + else: + raise ValueError(f"ValueError: Cannot multiply \"{other.__class__}\" to Accumulation!") + + def __rmul__(self, other): + """Redefining other (scalar) * self""" + return self.__mul__(other) + + def __imul__(self, other): + """Redefining self *= other (scalar)""" + if isinstance(other, (int, float)): + if not self.contains_data(): + raise ValueError("Accumulation: You cant multiply an empty accumulation") + self.lock.acquire() + for i in range(self.get_number_of_channels()): + self.y[i] *= other + if self.uses_statistics(): + self.y_square[i] *= (other**2) + self.lock.release() + return self + else: + raise ValueError(f"ValueError: Cannot multiply \"{other.__class__}\" to Accumulation!") + + def __truediv__(self, other): + """Redefining self / other (scalar)""" + if isinstance(other, (int, float)): + if not self.contains_data(): + raise ValueError("Accumulation: You cant divide an empty accumulation") + self.lock.acquire() + tmp_y = [] + tmp_ysquare = [] + for i in range(self.get_number_of_channels()): + tmp_y.append(self.y[i] / other) + if self.uses_statistics(): + tmp_ysquare.append(self.y_square[i] / (other**2)) + + if self.uses_statistics(): + r = Accumulation(x = numpy.array(self.x, dtype="float32"), y = tmp_y, y_2 = tmp_ysquare, n = self.n, index = self.index, sampl_freq = self.sampling_rate, error = self.use_error) + 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 = self.use_error) + r.job_id = self.job_id + r.job_ids = self.job_ids.copy() + r.time_period = self.time_period[:] if self.time_period is not None else None + if self.common_descriptions is not None: + r.common_descriptions = self.common_descriptions.copy() + self.lock.release() + return r + else: + raise ValueError(f"ValueError: Cannot divide \"{other.__class__}\" to Accumulation!") + + def __itruediv__(self, other): + """Redefining self /= other (scalar)""" + if isinstance(other, (int, float)): + if not self.contains_data(): + raise ValueError("Accumulation: You cant divide an empty accumulation") + self.lock.acquire() + for i in range(self.get_number_of_channels()): + self.y[i] /= other + if self.uses_statistics(): + self.y_square[i] /= (other**2) + self.lock.release() + return self + else: + raise ValueError(f"ValueError: Cannot divide \"{other.__class__}\" to Accumulation!") + + def __floordiv__(self, other): + """Redefining self // other (scalar)""" + if isinstance(other, float): + raise ValueError("ValueError: Cannot use floor division (//) on floats! Use \"/\" instead of \"//\"! ") + + if isinstance(other, int): + if not self.contains_data(): + raise ValueError("Accumulation: You cant divide an empty accumulation") + self.lock.acquire() + tmp_y = [] + for i in range(self.get_number_of_channels()): + tmp_y.append(self.y[i] // other) + + # Note: Statistics (y_square) cannot be easily maintained with floor division + 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.job_id = self.job_id + r.job_ids = self.job_ids.copy() + r.time_period = self.time_period[:] if self.time_period is not None else None + if self.common_descriptions is not None: + r.common_descriptions = self.common_descriptions.copy() + self.lock.release() + return r + else: + raise ValueError(f"ValueError: Cannot divide \"{other.__class__}\" to Accumulation!") + + def __ifloordiv__(self, other): + """Redefining self //= other (scalar)""" + if isinstance(other, float): + raise ValueError("ValueError: Cannot use floor division (//) on floats! Use \"/\" instead of \"//\"! ") + + if isinstance(other, int): + if not self.contains_data(): + raise ValueError("Accumulation: You cant divide an empty accumulation") + self.lock.acquire() + for i in range(self.get_number_of_channels()): + self.y[i] //= other + if self.uses_statistics(): + # Invalidate statistics + self.y_square = [] + self.use_error = False + self.lock.release() + return self + else: + raise ValueError(f"ValueError: Cannot divide \"{other.__class__}\" to Accumulation!") def __iadd__(self, other): @@ -863,7 +1000,7 @@ class Accumulation(Errorable, Drawable, DamarisFFT, Signalpath): elif isinstance(other, Accumulation): # Other empty (return) if not other.contains_data(): - return + return self # Self empty (copy) if not self.contains_data(): diff --git a/tests/test_Accumulation.py b/tests/test_Accumulation.py index 7234e69..daef1a8 100644 --- a/tests/test_Accumulation.py +++ b/tests/test_Accumulation.py @@ -233,7 +233,6 @@ class TestAccumulation(unittest.TestCase): for i in range(adc.get_nChannels()): np.testing.assert_array_equal(accu.y[i], y[i] - 10 - 10.) - @unittest.skip("Not implmented") def test_operator_mul_scalar(self): """ Test the functionality of __mul and __rmul__ @@ -254,33 +253,35 @@ class TestAccumulation(unittest.TestCase): for i in range(adc.get_nChannels()): np.testing.assert_array_equal(accu.y[i], y[i] * 10 * 10.0) - @unittest.skip("Not implmented") def test_operator_truediv_scalar(self): """ - Test the functionality of __div__ and __rdiv__ + 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(adc.y[i]/10, y[i] / 10, strict=True) - adc /= 10 + 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(adc.y[i], y[i] / 10, strict=True) + np.testing.assert_array_equal(accu.y[i], y[i] / 10) - @unittest.skip("Not implmented") def test_operator_floordiv_scalar(self): """ - Test the functionality of __div__ and __rdiv__ + 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(adc.y[i]//10, y[i] // 10, strict=True) - adc //= 10 + 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(adc.y[i], y[i] // 10, strict=True) - self.assertRaises(ValueError, adc.__floordiv__, 1.0) - self.assertRaises(ValueError, adc.__rfloordiv__, 1.0) + 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) if __name__ == "__main__": unittest.main()