Add native QCustomPlot renderer and shared plot adapters

python/set_devices/oscilloscope.py

@@ -0,0 +1,111 @@
+"""Software oscilloscope capture and sample measurements (timestamps in ms)."""
+from __future__ import annotations
+
+import math
+
+
+def time_label(milliseconds):
+    """Readable time without rounding nanosecond divisions to zero."""
+    magnitude = abs(milliseconds)
+    if magnitude < .001:
+        return f"{milliseconds * 1e6:.6g} нс"
+    if magnitude < 1:
+        return f"{milliseconds * 1000:.6g} мкс"
+    if magnitude < 1000:
+        return f"{milliseconds:.6g} мс"
+    return f"{milliseconds / 1000:.6g} с"
+
+
+def measurements(points):
+    values = [value for _, value in points if math.isfinite(value)]
+    if not values:
+        return None
+    low, high = min(values), max(values)
+    mean = math.fsum(value / len(values) for value in values)
+    scale = max(abs(low), abs(high))
+    rms = scale * math.sqrt(math.fsum((value / scale) ** 2 for value in values) / len(values)) if scale else 0.
+    # Hysteresis rejects small excursions around the midpoint. Frequency is
+    # only reported after two complete rising crossings, never from one edge.
+    middle = (low + high) / 2
+    hysteresis = (high - low) * .05
+    armed = False
+    crossings = []
+    previous = None
+    if high > low:
+        for stamp, value in points:
+            if value < middle - hysteresis:
+                armed = True
+            if armed and previous and previous[1] < middle <= value and stamp > previous[0]:
+                t0, v0 = previous
+                crossings.append(t0 + (stamp - t0) * (middle - v0) / (value - v0))
+                armed = False
+            previous = (stamp, value)
+    period = ((crossings[-1] - crossings[0]) / (len(crossings) - 1)
+              if len(crossings) >= 2 else None)
+    return dict(minimum=low, maximum=high, peak_to_peak=high - low, mean=mean,
+                rms=rms, frequency=1000 / period if period and period > 0 else None)
+
+
+class SingleCapture:
+    """Edge trigger with 20% pretrigger history and 80% posttrigger samples.
+
+    Samples are display-scaled; storage/recording remains owned by the caller.
+    Capture is bounded by the caller's point capacity for each channel.
+    """
+    def __init__(self):
+        self.cancel()
+
+    def cancel(self):
+        self.armed = False
+        self.trigger_time = None
+        self.previous = None
+        self.data = {}
+        self.truncated = False
+
+    def arm(self, channel, level, falling, span_ms, capacity):
+        self.cancel()
+        self.channel = channel
+        self.level = level
+        self.falling = falling
+        self.span_ms = max(.0001, min(60_000., span_ms))
+        self.capacity = capacity
+        self.armed = True
+
+    def feed(self, stamp, sample, history):
+        """Return a complete frame once; history must include this sample."""
+        if not self.armed:
+            return None
+        value = sample.get(self.channel)
+        if self.trigger_time is None:
+            previous = self.previous
+            if value is not None:
+                self.previous = (stamp, value)
+            if previous is None or value is None or stamp <= previous[0]:
+                return None
+            crossed = (previous[1] > self.level >= value if self.falling
+                       else previous[1] < self.level <= value)
+            if not crossed:
+                return None
+            t0, v0 = previous
+            self.trigger_time = t0 + (stamp - t0) * (self.level - v0) / (value - v0)
+            self.left = self.trigger_time - .2 * self.span_ms
+            self.right = self.trigger_time + .8 * self.span_ms
+            if callable(history):
+                history = history()
+            source = history.get(self.channel, ())
+            self.truncated = not source or source[0][0] - self.trigger_time > -.2 * self.span_ms
+            self.data = {key: [(t, v) for t, v in points
+                               if -.2 * self.span_ms <= t - self.trigger_time <= .8 * self.span_ms]
+                         for key, points in history.items()}
+        else:
+            for key, value in sample.items():
+                if -.2 * self.span_ms <= stamp - self.trigger_time <= .8 * self.span_ms:
+                    points = self.data.setdefault(key, [])
+                    if len(points) < self.capacity:
+                        points.append((stamp, value))
+                    else:
+                        self.truncated = True
+        if stamp - self.trigger_time >= .8 * self.span_ms:
+            self.armed = False
+            return self.data
+        return None