create base class MeasurementData for ADC_Result and Accumulation

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())])