1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
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)
|
