Добавить общие блоки Zynq, Ethernet DSView и декодер ШИМ
This commit is contained in:
@@ -0,0 +1,69 @@
|
||||
"""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
|
||||
Reference in New Issue
Block a user