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
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
|
import subprocess
import time
from dfatool.lennart.SigrokInterface import SigrokInterface
import logging
logger = logging.getLogger(__name__)
class SigrokCLIInterface(SigrokInterface):
def __init__(
self,
bin_temp_file="temp/out.bin",
sample_rate=100_000,
fake=False,
):
"""
creates SigrokCLIInterface object. Uses the CLI Interface (Command: sigrok-cli)
:param bin_temp_file: temporary file for binary output
:param sample_rate: The sample rate of the Logic analyzer
:param fake: if it should use existing data
"""
super(SigrokCLIInterface, self).__init__(sample_rate)
self.fake = fake
self.bin_temp_file = bin_temp_file
self.sigrok_cli_thread = None
def forceStopMeasure(self):
"""
Force stopping measure, sometimes needs pkill for killing definitly
:return: None
"""
self.sigrok_cli_thread.terminate()
try:
self.sigrok_cli_thread.wait(timeout=10)
except subprocess.TimeoutExpired:
logger.warning("sigrok-cli has not stopped. Killing it.")
self.sigrok_cli_thread.kill()
self.sigrok_cli_thread.communicate()
self.runOpenAnalyze()
def runMeasure(self):
"""
starts the measurement, with waiting for done
"""
if not self.fake:
self.runMeasureAsynchronous()
self.waitForAsynchronousMeasure()
def runMeasureAsynchronous(self):
"""
starts the measurement, not waiting for done
"""
shellcommand = [
"sigrok-cli",
"--output-file",
self.bin_temp_file,
"--output-format",
"binary",
"--continuous",
"-d",
self.driver,
"--config",
f"samplerate={self.sample_rate} Hz",
]
self.sigrok_cli_thread = subprocess.Popen(shellcommand)
def waitForAsynchronousMeasure(self):
"""
Wait until is command call is done
"""
if not self.fake:
self.sigrok_cli_thread.wait()
self.runOpenAnalyze()
def runOpenAnalyze(self):
"""
Opens the generated binary file and parses it byte by byte
"""
in_file = open(self.bin_temp_file, "rb") # opening for [r]eading as [b]inary
data = in_file.read() # if you only wanted to read 512 bytes, do .read(512)
in_file.close()
for x in data:
self.analyzeData(x)
|
