From bd060dc4d0d05de7d04ce859f0809eeabe1272a9 Mon Sep 17 00:00:00 2001 From: Markus Rosenstihl Date: Tue, 20 Oct 2026 23:21:43 +0200 Subject: [PATCH] create base class MeasurementData for ADC_Result and Accumulation --- doc/manual.rst | 30 +++++++ src/damaris/data/ADC_Result.py | 3 +- src/damaris/data/Accumulation.py | 48 +---------- src/damaris/data/MeasurementData.py | 127 ++++++++++++++++++++++++++++ tests/test_ADC_Result.py | 15 ++++ tests/test_Accumulation.py | 8 ++ 6 files changed, 184 insertions(+), 47 deletions(-) create mode 100644 src/damaris/data/MeasurementData.py diff --git a/doc/manual.rst b/doc/manual.rst index c12ba35..c56581e 100644 --- a/doc/manual.rst +++ b/doc/manual.rst @@ -759,6 +759,36 @@ Example: - **get_ymax()** Returns maximum of y +- **get_ymean(start=None, stop=None)** + Returns the mean value of every channel. If *start* and *stop* are given, + they select a time interval in seconds. The calculation uses raw values + for **timesignal** and the public mean values for **accu**. + + Example: + + .. code-block:: python + + means = timesignal.get_ymean(start=1e-6, stop=4e-6) + means = accu.get_ymean() + +- **get_phase(pos=None)** + Returns the phase in degrees from the first two channels at the first + sample or at time *pos* in seconds. + + Example: + + .. code-block:: python + + phase = timesignal.get_phase(pos=1e-6) + +- **get_ymin_channels()/get_ymax_channels()** + Return NumPy arrays containing the minimum or maximum for each channel. + **get_ymin()** and **get_ymax()** return the corresponding global extrema. + +- **get_yrms()/get_ystd()** + Return NumPy arrays containing the RMS or population standard deviation + for each channel. + .. bibliography:: diff --git a/src/damaris/data/ADC_Result.py b/src/damaris/data/ADC_Result.py index 5f3e178..a56c839 100644 --- a/src/damaris/data/ADC_Result.py +++ b/src/damaris/data/ADC_Result.py @@ -6,6 +6,7 @@ from .Resultable import Resultable from .Drawable import Drawable from .Signalpath import Signalpath from .DamarisFFT import DamarisFFT +from .MeasurementData import MeasurementData import threading import numpy import sys @@ -27,7 +28,7 @@ _Description = dict[str, str] # # ############################################################################# -class ADC_Result(Resultable, Drawable, DamarisFFT, Signalpath): +class ADC_Result(MeasurementData, Resultable, Drawable, DamarisFFT, Signalpath): default_bit_depth = 14 """ Represents the result of an ADC, encapsulating data and metadata diff --git a/src/damaris/data/Accumulation.py b/src/damaris/data/Accumulation.py index b10aadd..acf8534 100644 --- a/src/damaris/data/Accumulation.py +++ b/src/damaris/data/Accumulation.py @@ -16,6 +16,7 @@ from .Drawable import Drawable from .DamarisFFT import DamarisFFT from .Signalpath import Signalpath from .ADC_Result import ADC_Result +from .MeasurementData import MeasurementData import sys import threading @@ -33,7 +34,7 @@ _IndexList = list[tuple[int, int]] _Description = dict[str, str] -class Accumulation(Errorable, Drawable, DamarisFFT, Signalpath): +class Accumulation(MeasurementData, Errorable, Drawable, DamarisFFT, Signalpath): """ Represents an accumulation of sampled data with various attributes and methods for data manipulation, analysis, and exporting. @@ -220,44 +221,6 @@ class Accumulation(Errorable, Drawable, DamarisFFT, Signalpath): self.lock.release() return tmp_y - def get_ymean(self, start: float | None = None, stop: float | None = None) -> numpy.ndarray: - """ - return mean data from start to stop (in time units). Defaults to mean of whole channel data. - - Example: - accu = Accumulation(...) - means = accu.get_ymean(start=1e-6, stop=4e-6) - - :param start: start of the data - :param stop: stop of the data - :return: means for each channel - """ - if start == None and stop == None: - means = [ch.mean() for ch in self.y] - else: - istart = int(self.get_sampling_rate()*start) - istop = int(self.get_sampling_rate()*stop) - means = [ch[istart:istop].mean() for ch in self.y] - return numpy.array(means) - - def get_phase(self, pos: float | None = None) -> float: - """ - Return phase angle in degrees at a given time position. - - The phase is calculated from the first two channels using arctan2(y1, y0) - and converted to degrees. If pos is None, the phase at the first sample is returned. - - Example: - accu = Accumulation(...) - phase = accu.get_phase(pos=1e-6) - :param pos: time position in seconds - :return: phase angle in degrees - """ - if pos is None: - ipos = 0 - else: - ipos = int(self.get_sampling_rate()*pos) - return numpy.arctan2(self.y[1][ipos], self.y[0][ipos])*180/numpy.pi def get_ymin(self) -> float|int: @@ -671,13 +634,6 @@ class Accumulation(Errorable, Drawable, DamarisFFT, Signalpath): # External interfaces ------------------------------------------------------------------ # Overloaded operators ----------------------------------------------------------------- - def __len__(self) -> int: - """ - return number of samples per channel, 0 if empty - """ - if len(self.y)>0: - return len(self.y[0]) - return 0 def __repr__(self) -> str: """Redefining repr(Accumulation)""" diff --git a/src/damaris/data/MeasurementData.py b/src/damaris/data/MeasurementData.py new file mode 100644 index 0000000..22689f3 --- /dev/null +++ b/src/damaris/data/MeasurementData.py @@ -0,0 +1,127 @@ +# -*- coding: iso-8859-1 -*- + +from __future__ import annotations + +import numpy + + +class MeasurementData: + """Common channel-series access and numerical analysis for measurements. + + Concrete result classes provide storage and ``get_ydata``; all calculations + in this class therefore use the public channel values rather than their + internal representation. + """ + + def contains_data(self) -> bool: + """Return whether the measurement contains samples.""" + raise NotImplementedError + + def get_ydata(self, channel: int = 0) -> numpy.ndarray: + """Return public values for one channel.""" + return self.y[channel] + + def get_number_of_channels(self) -> int: + """Return the number of data channels.""" + return len(self.y) + + def get_sampling_rate(self) -> float: + """Return the sampling frequency in hertz.""" + return self.sampling_rate + + def __len__(self) -> int: + """Return the number of samples per channel. + + Example: + samples = len(timesignal) + """ + if not self.get_number_of_channels(): + return 0 + return len(self.get_ydata(0)) + + def get_ymean(self, start: float | None = None, + stop: float | None = None) -> numpy.ndarray: + """Return the mean of every channel over a time interval. + + Example: + means = timesignal.get_ymean(start=1e-6, stop=4e-6) + """ + if not self.contains_data(): + return numpy.array([]) + if (start is None) != (stop is None): + raise ValueError("start and stop must be specified together") + if start is None: + values = [self.get_ydata(channel).mean() + for channel in range(self.get_number_of_channels())] + else: + istart = int(self.get_sampling_rate() * start) + istop = int(self.get_sampling_rate() * stop) + values = [self.get_ydata(channel)[istart:istop].mean() + for channel in range(self.get_number_of_channels())] + return numpy.asarray(values) + + def get_phase(self, pos: float | None = None) -> float: + """Return the phase in degrees from the first two channels. + + Example: + phase = timesignal.get_phase(pos=1e-6) + """ + if self.get_number_of_channels() < 2: + raise ValueError("phase requires at least two channels") + ipos = 0 if pos is None else int(self.get_sampling_rate() * pos) + return float(numpy.degrees(numpy.arctan2( + self.get_ydata(1)[ipos], self.get_ydata(0)[ipos]))) + + def get_ymin_channels(self) -> numpy.ndarray: + """Return the minimum value of each channel. + + Example: + minima = accu.get_ymin_channels() + """ + if not self.contains_data(): + return numpy.array([]) + return numpy.asarray([self.get_ydata(i).min() + for i in range(self.get_number_of_channels())]) + + def get_ymax_channels(self) -> numpy.ndarray: + """Return the maximum value of each channel. + + Example: + maxima = accu.get_ymax_channels() + """ + if not self.contains_data(): + return numpy.array([]) + return numpy.asarray([self.get_ydata(i).max() + for i in range(self.get_number_of_channels())]) + + def get_ymin(self) -> float | int: + """Return the global minimum across all channels.""" + values = self.get_ymin_channels() + return 0 if len(values) == 0 else values.min() + + def get_ymax(self) -> float | int: + """Return the global maximum across all channels.""" + values = self.get_ymax_channels() + return 0 if len(values) == 0 else values.max() + + def get_yrms(self) -> numpy.ndarray: + """Return the root-mean-square value of each channel. + + Example: + rms = timesignal.get_yrms() + """ + if not self.contains_data(): + return numpy.array([]) + return numpy.asarray([numpy.sqrt(numpy.mean(self.get_ydata(i) ** 2)) + for i in range(self.get_number_of_channels())]) + + def get_ystd(self) -> numpy.ndarray: + """Return the population standard deviation of each channel. + + Example: + deviation = accu.get_ystd() + """ + if not self.contains_data(): + return numpy.array([]) + return numpy.asarray([numpy.std(self.get_ydata(i)) + for i in range(self.get_number_of_channels())]) \ No newline at end of file diff --git a/tests/test_ADC_Result.py b/tests/test_ADC_Result.py index 3bda02b..de9216e 100644 --- a/tests/test_ADC_Result.py +++ b/tests/test_ADC_Result.py @@ -30,6 +30,21 @@ class TestADCResult(unittest.TestCase): 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]) diff --git a/tests/test_Accumulation.py b/tests/test_Accumulation.py index f7354c1..d5c3172 100644 --- a/tests/test_Accumulation.py +++ b/tests/test_Accumulation.py @@ -33,6 +33,14 @@ class TestAccumulation(unittest.TestCase): 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])