Add sigrok frequency meter

1 files changed, 71 insertions, 0 deletions

diff --git a/prototype/fw/freqmeter.py b/prototype/fw/freqmeter.py
new file mode 100644
index 0000000..54c1655
--- /dev/null
+++ b/prototype/fw/freqmeter.py
@@ -0,0 +1,71 @@
+#!/usr/bin/env python3
+
+import csv
+import subprocess
+import io
+import itertools
+import warnings
+import statistics
+import time
+
+count = lambda le_iter: sum(1 for _ in le_iter)
+
+DEFAULT_SAMPLING_RATE = 1e6 # sps
+
+def sigrok_capture(duration:'seconds'=1, sampling_rate=DEFAULT_SAMPLING_RATE, driver='dreamsourcelab-dslogic', config=None, channel=0):
+
+    proc = subprocess.run(['sigrok-cli',
+        '--driver', driver,
+        '--time', f'{duration}s',
+        '--config', (f'{config},' if config else '') + f'samplerate={int(sampling_rate/1e3)}k',
+        '--channels', str(channel),
+        '--output-format', 'csv'], check=True, stdout=subprocess.PIPE)
+
+    lines = proc.stdout.splitlines()
+    return lines[lines.index(b'logic')+1:]
+
+def debounce(lines, sampling_rate=DEFAULT_SAMPLING_RATE, debounce_interval=1e-3):
+    debounce_len = debounce_interval * sampling_rate
+
+    cumline = None
+    cumsum = None
+    for line, group in itertools.groupby(lines):
+        group_len = count(group)
+
+        if cumline is None:
+            cumline, cumsum = line, group_len
+
+        if group_len < debounce_len:
+            cumsum += group_len
+
+        else:
+            yield bool(int(cumline.decode())), cumsum
+            cumline, cumsum = line, group_len
+
+    yield bool(int(cumline.decode())), cumsum
+
+def calc_frequency(intervals, sampling_rate=DEFAULT_SAMPLING_RATE):
+    # make sure intervals alternate true/false
+    if not all( a != b for a, b in zip(intervals[0::2], intervals[1::2])):
+        raise ValueError('Intervals do not alternate!')
+
+    sums = [ DEFAULT_SAMPLING_RATE / (a[1] + b[1]) for a, b in zip(intervals[0::2], intervals[1::2]) ]
+    return statistics.mean(sums), statistics.stdev(sums)
+
+if __name__ == '__main__':
+    while True:
+        capture = sigrok_capture()
+        intervals = list(debounce(capture))
+
+        intervals = intervals[2:-1] # ignore partial first and last intervals
+
+        # Ignore last interval if we have an uneven number of intervals
+        if intervals[-1] == intervals[0]:
+            intervals = intervals[:-1]
+
+        try:
+            mean, stdev = calc_frequency(intervals)
+            print(f'\033[38;5;244m{time.strftime("%H:%M:%S")} \033[93m{mean*1e3:> 9.3f} \033[38;5;244m± {stdev*1e3:> 8.3f} mHz\033[0m')
+        except ValueError as e:
+            warnings.warn(*e.args)
+