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
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
|
"""
Utilities for running benchmarks.
Classes:
SerialMonitor -- captures serial output for a specific amount of time
ShellMonitor -- captures UNIX program output for a specific amount of time
Functions:
get_monitor -- return Monitor class suitable for the selected multipass arch
get_counter_limits -- return arch-specific multipass counter limits (max value, max overflow)
"""
import re
import serial
import serial.threaded
import subprocess
import sys
import time
class SerialReader(serial.threaded.Protocol):
"""
Character- to line-wise data buffer for serial interfaces.
Reads in new data whenever it becomes available and exposes a line-based
interface to applications.
"""
def __init__(self, peek = False):
"""Create a new SerialReader object."""
self.peek = peek
self.recv_buf = ''
self.lines = []
def __call__(self):
return self
def data_received(self, data):
"""Append newly received serial data to the line buffer."""
try:
str_data = data.decode('UTF-8')
self.recv_buf += str_data
if '\n\r' in self.recv_buf:
lines = self.recv_buf.split('\n\r')
elif '\r\n' in self.recv_buf:
lines = self.recv_buf.split('\r\n')
else:
lines = []
if len(lines) > 1:
self.lines.extend(lines[:-1])
self.recv_buf = lines[-1]
if self.peek:
print('\n'.join(lines[:-1]))
except UnicodeDecodeError:
pass
#sys.stderr.write('UART output contains garbage: {data}\n'.format(data = data))
def get_lines(self) -> list:
"""
Return the latest batch of complete lines.
The return value is a list and may be empty.
Empties the internal line buffer to ensure that no line is returned twice.
"""
ret = self.lines
self.lines = []
return ret
def get_line(self) -> str:
"""
Return the latest complete line, or None.
Empties the entire internal line buffer to ensure that no line is returned twice.
"""
if len(self.lines):
ret = self.lines[-1]
self.lines = []
return ret
return None
class SerialMonitor:
"""SerialMonitor captures serial output for a specific amount of time."""
def __init__(self, port: str, baud: int, peek = False):
"""
Create a new SerialMonitor connected to port at the specified baud rate.
Communication uses no parity, no flow control, and one stop bit.
Data collection starts immediately.
"""
self.ser = serial.serial_for_url(port, do_not_open=True)
self.ser.baudrate = baud
self.ser.parity = 'N'
self.ser.rtscts = False
self.ser.xonxoff = False
try:
self.ser.open()
except serial.SerialException as e:
sys.stderr.write('Could not open serial port {}: {}\n'.format(self.ser.name, e))
sys.exit(1)
self.reader = SerialReader(peek = peek)
self.worker = serial.threaded.ReaderThread(self.ser, self.reader)
self.worker.start()
def run(self, timeout: int = 10) -> list:
"""
Collect serial output for timeout seconds and return a list of all output lines.
Blocks until data collection is complete.
"""
time.sleep(timeout)
return self.reader.get_lines()
def get_lines(self) ->list:
return self.reader.get_lines()
def close(self):
"""Close serial connection."""
self.worker.stop()
self.ser.close()
class ShellMonitor:
"""SerialMonitor runs a program and captures its output for a specific amount of time."""
def __init__(self, script: str, peek = None):
"""
Create a new ShellMonitor object.
Does not start execution and monitoring yet.
"""
self.script = script
self.peek = peek
def run(self, timeout: int = 4) -> list:
"""
Run program for timeout seconds and return a list of its stdout lines.
stderr and return status are discarded at the moment.
"""
if type(timeout) != int:
raise ValueError('timeout argument must be int')
res = subprocess.run(['timeout', '{:d}s'.format(timeout), self.script],
stdout = subprocess.PIPE, stderr = subprocess.PIPE,
universal_newlines = True)
if self.peek:
print(res.stdout)
return res.stdout.split('\n')
def monitor(self):
raise NotImplementedError
def close(self):
"""
Do nothing, successfully.
Intended for compatibility with SerialMonitor.
"""
pass
def build(arch, app, opts = []):
command = ['make', '-B', 'arch={}'.format(arch), 'app={}'.format(app)]
command.extend(opts)
res = subprocess.run(command, stdout = subprocess.PIPE, stderr = subprocess.PIPE,
universal_newlines = True)
if res.returncode != 0:
raise RuntimeError('Build failure: ' + res.stderr)
return command
def flash(arch, app, opts = []):
command = ['make', 'arch={}'.format(arch), 'app={}'.format(app), 'program']
command.extend(opts)
res = subprocess.run(command, stdout = subprocess.PIPE, stderr = subprocess.PIPE,
universal_newlines = True)
if res.returncode != 0:
raise RuntimeError('Flash failure')
return command
def get_info(arch, opts: list = []) -> list:
"""
Return multipass "make info" output.
Returns a list.
"""
command = ['make', 'arch={}'.format(arch), 'info']
command.extend(opts)
res = subprocess.run(command, stdout = subprocess.PIPE, stderr = subprocess.PIPE,
universal_newlines = True)
if res.returncode != 0:
raise RuntimeError('make info Failure')
return res.stdout.split('\n')
def get_monitor(arch: str, **kwargs) -> object:
"""Return a SerialMonitor or ShellMonitor."""
for line in get_info(arch):
if 'Monitor:' in line:
_, port, arg = line.split(' ')
if port == 'run':
return ShellMonitor(arg, **kwargs)
else:
return SerialMonitor(port, arg, **kwargs)
raise RuntimeError('Monitor failure')
def get_counter_limits(arch: str) -> tuple:
"""Return multipass max counter and max overflow value for arch."""
for line in get_info(arch):
match = re.match('Counter Overflow: ([^/]*)/(.*)', line)
if match:
overflow_value = int(match.group(1))
max_overflow = int(match.group(2))
return overflow_value, max_overflow
raise RuntimeError('Did not find Counter Overflow limits')
|
