71 lines
2.3 KiB
Python
71 lines
2.3 KiB
Python
#!/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:'milliseconds', sampling_rate=DEFAULT_SAMPLING_RATE, driver='dreamsourcelab-dslogic', config=None, channel=0):
|
|
|
|
proc = subprocess.run(['sigrok-cli',
|
|
'--driver', driver,
|
|
'--time', f'{duration}ms',
|
|
'--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(1500)
|
|
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)
|
|
|