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
|
import csv
import os
import signal
from subprocess import Popen, PIPE
from tempfile import NamedTemporaryFile
from devlib.instrument import Instrument, CONTINUOUS, MeasurementsCsv
from devlib.exception import HostError
from devlib.host import PACKAGE_BIN_DIRECTORY
from devlib.utils.misc import which
INSTALL_INSTRUCTIONS="""
MonsoonInstrument requires the monsoon.py tool, available from AOSP:
https://android.googlesource.com/platform/cts/+/master/tools/utils/monsoon.py
Download this script and put it in your $PATH (or pass it as the monsoon_bin
parameter to MonsoonInstrument). `pip install python-gflags pyserial` to install
the dependencies.
"""
class MonsoonInstrument(Instrument):
"""Instrument for Monsoon Solutions power monitor
To use this instrument, you need to install the monsoon.py script available
from the Android Open Source Project. As of May 2017 this is under the CTS
repository:
https://android.googlesource.com/platform/cts/+/master/tools/utils/monsoon.py
Collects power measurements only, from a selection of two channels, the USB
passthrough channel and the main output channel.
:param target: Ignored
:param monsoon_bin: Path to monsoon.py executable. If not provided,
``$PATH`` is searched.
:param tty_device: TTY device to use to communicate with the Power
Monitor. If not provided, a sane default is used.
"""
mode = CONTINUOUS
def __init__(self, target, monsoon_bin=None, tty_device=None):
super(MonsoonInstrument, self).__init__(target)
self.monsoon_bin = monsoon_bin or which('monsoon.py')
if not self.monsoon_bin:
raise HostError(INSTALL_INSTRUCTIONS)
self.tty_device = tty_device
self.process = None
self.output = None
self.sample_rate_hz = 500
self.add_channel('output', 'power')
self.add_channel('USB', 'power')
def reset(self, sites=None, kinds=None, channels=None):
super(MonsoonInstrument, self).reset(sites, kinds)
def start(self):
if self.process:
self.process.kill()
os.system(self.monsoon_bin + ' --usbpassthrough off')
cmd = [self.monsoon_bin,
'--hz', str(self.sample_rate_hz),
'--samples', '-1', # -1 means sample indefinitely
'--includeusb']
if self.tty_device:
cmd += ['--device', self.tty_device]
self.logger.debug(' '.join(cmd))
self.buffer_file = NamedTemporaryFile(prefix='monsoon', delete=False)
self.process = Popen(cmd, stdout=self.buffer_file, stderr=PIPE)
def stop(self):
process = self.process
self.process = None
if not process:
raise RuntimeError('Monsoon script not started')
process.poll()
if process.returncode is not None:
stdout, stderr = process.communicate()
raise HostError(
'Monsoon script exited unexpectedly with exit code {}.\n'
'stdout:\n{}\nstderr:\n{}'.format(process.returncode,
stdout, stderr))
process.send_signal(signal.SIGINT)
stderr = process.stderr.read()
self.buffer_file.close()
with open(self.buffer_file.name) as f:
stdout = f.read()
os.remove(self.buffer_file.name)
self.buffer_file = None
self.output = (stdout, stderr)
os.system(self.monsoon_bin + ' --usbpassthrough on')
# Wait for USB connection to be restored
print ('waiting for usb connection to be back')
os.system('adb wait-for-device')
def get_data(self, outfile):
if self.process:
raise RuntimeError('`get_data` called before `stop`')
stdout, stderr = self.output
with open(outfile, 'wb') as f:
writer = csv.writer(f)
active_sites = [c.site for c in self.active_channels]
# Write column headers
row = []
if 'output' in active_sites:
row.append('output_power')
if 'USB' in active_sites:
row.append('USB_power')
writer.writerow(row)
# Write data
for line in stdout.splitlines():
# Each output line is a main_output, usb_output measurement pair.
# (If our user only requested one channel we still collect both,
# and just ignore one of them)
output, usb = line.split()
row = []
if 'output' in active_sites:
row.append(output)
if 'USB' in active_sites:
row.append(usb)
writer.writerow(row)
return MeasurementsCsv(outfile, self.active_channels, self.sample_rate_hz)
|
