70 lines
2.9 KiB
Python
70 lines
2.9 KiB
Python
"""Streaming PWM demodulation; sample coordinates, no GUI dependencies."""
|
|
import math
|
|
|
|
|
|
class PwmChannel:
|
|
def __init__(self, samplerate, active_high=True, carrier_hz=0, smoothing_ms=0):
|
|
if not math.isfinite(samplerate) or samplerate <= 0:
|
|
raise ValueError('Samplerate must be positive.')
|
|
if not math.isfinite(carrier_hz) or carrier_hz < 0 or carrier_hz > samplerate / 2:
|
|
raise ValueError('Carrier must be 0 (automatic) or <= samplerate / 2.')
|
|
if not math.isfinite(smoothing_ms) or smoothing_ms < 0:
|
|
raise ValueError('Smoothing must be nonnegative.')
|
|
self.rate = samplerate
|
|
self.active_high = active_high
|
|
self.window = samplerate / carrier_hz if carrier_hz else None
|
|
self.tau = smoothing_ms * samplerate / 1000
|
|
self.last = self.level = self.start = self.fall = None
|
|
self.origin = None
|
|
self.window_index = 1
|
|
self.high = 0
|
|
self.previous = None
|
|
|
|
def next_boundary(self):
|
|
if self.window is None or self.origin is None:
|
|
return None
|
|
return round(self.origin + self.window_index * self.window)
|
|
|
|
def _emit(self, start, end, high):
|
|
duty = high / (end - start)
|
|
value = 2 * duty - 1
|
|
midpoint = (start + end) // 2
|
|
previous = self.previous
|
|
if previous is not None and self.tau:
|
|
alpha = -math.expm1(-(midpoint - previous[0]) / self.tau)
|
|
value = previous[1] + alpha * (value - previous[1])
|
|
self.previous = (midpoint, value)
|
|
return dict(start=start, end=end, duty=duty, value=value,
|
|
frequency=self.rate / (end - start), midpoint=midpoint,
|
|
previous=previous)
|
|
|
|
def update(self, sample, pin):
|
|
level = bool(pin) == self.active_high
|
|
if self.last is None:
|
|
self.last, self.level = sample, level
|
|
if self.window is not None:
|
|
self.origin = self.start = sample
|
|
return []
|
|
if sample < self.last:
|
|
raise ValueError('Samples must be monotonic.')
|
|
events = []
|
|
if self.window is not None:
|
|
boundary = self.next_boundary()
|
|
while boundary <= sample:
|
|
self.high += (boundary - self.last) * self.level
|
|
events.append(self._emit(self.start, boundary, self.high))
|
|
self.start = self.last = boundary
|
|
self.high = 0
|
|
self.window_index += 1
|
|
boundary = self.next_boundary()
|
|
self.high += (sample - self.last) * self.level
|
|
elif level != self.level:
|
|
if level:
|
|
if self.start is not None and self.fall is not None:
|
|
events.append(self._emit(self.start, sample, self.fall - self.start))
|
|
self.start, self.fall = sample, None
|
|
elif self.start is not None:
|
|
self.fall = sample
|
|
self.last, self.level = sample, level
|
|
return events
|