#!/usr/bin/env python
|
|
# Copyright (C) 2014 The Android Open Source Project
|
#
|
# Licensed under the Apache License, Version 2.0 (the "License");
|
# you may not use this file except in compliance with the License.
|
# You may obtain a copy of the License at
|
#
|
# http://www.apache.org/licenses/LICENSE-2.0
|
#
|
# Unless required by applicable law or agreed to in writing, software
|
# distributed under the License is distributed on an "AS IS" BASIS,
|
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
# See the License for the specific language governing permissions and
|
# limitations under the License.
|
|
"""Interface for a USB-connected Monsoon power meter
|
(http://msoon.com/LabEquipment/PowerMonitor/).
|
This file requires gflags, which requires setuptools.
|
To install setuptools: sudo apt-get install python-setuptools
|
To install gflags, see http://code.google.com/p/python-gflags/
|
To install pyserial, see http://pyserial.sourceforge.net/
|
|
Example usages:
|
Set the voltage of the device 7536 to 4.0V
|
python monsoon.py --voltage=4.0 --serialno 7536
|
|
Get 5000hz data from device number 7536, with unlimited number of samples
|
python monsoon.py --samples -1 --hz 5000 --serialno 7536
|
|
Get 200Hz data for 5 seconds (1000 events) from default device
|
python monsoon.py --samples 100 --hz 200
|
|
Get unlimited 200Hz data from device attached at /dev/ttyACM0
|
python monsoon.py --samples -1 --hz 200 --device /dev/ttyACM0
|
|
Output columns for collection with --samples, separated by space:
|
|
TIMESTAMP OUTPUT OUTPUT_AVG USB USB_AVG
|
| | | |
|
| | | ` (if --includeusb and --avg)
|
| | ` (if --includeusb)
|
| ` (if --avg)
|
` (if --timestamp)
|
"""
|
|
import fcntl
|
import os
|
import select
|
import signal
|
import stat
|
import struct
|
import sys
|
import time
|
import collections
|
|
import gflags as flags # http://code.google.com/p/python-gflags/
|
|
import serial # http://pyserial.sourceforge.net/
|
|
FLAGS = flags.FLAGS
|
|
class Monsoon:
|
"""
|
Provides a simple class to use the power meter, e.g.
|
mon = monsoon.Monsoon()
|
mon.SetVoltage(3.7)
|
mon.StartDataCollection()
|
mydata = []
|
while len(mydata) < 1000:
|
mydata.extend(mon.CollectData())
|
mon.StopDataCollection()
|
"""
|
|
def __init__(self, device=None, serialno=None, wait=1):
|
"""
|
Establish a connection to a Monsoon.
|
By default, opens the first available port, waiting if none are ready.
|
A particular port can be specified with "device", or a particular Monsoon
|
can be specified with "serialno" (using the number printed on its back).
|
With wait=0, IOError is thrown if a device is not immediately available.
|
"""
|
|
self._coarse_ref = self._fine_ref = self._coarse_zero = self._fine_zero = 0
|
self._coarse_scale = self._fine_scale = 0
|
self._last_seq = 0
|
self.start_voltage = 0
|
|
if device:
|
self.ser = serial.Serial(device, timeout=1)
|
return
|
|
while True: # try all /dev/ttyACM* until we find one we can use
|
for dev in os.listdir("/dev"):
|
if not dev.startswith("ttyACM"): continue
|
tmpname = "/tmp/monsoon.%s.%s" % (os.uname()[0], dev)
|
self._tempfile = open(tmpname, "w")
|
try:
|
os.chmod(tmpname, 0666)
|
except OSError:
|
pass
|
try: # use a lockfile to ensure exclusive access
|
fcntl.lockf(self._tempfile, fcntl.LOCK_EX | fcntl.LOCK_NB)
|
except IOError as e:
|
print >>sys.stderr, "device %s is in use" % dev
|
continue
|
|
try: # try to open the device
|
self.ser = serial.Serial("/dev/%s" % dev, timeout=1)
|
self.StopDataCollection() # just in case
|
self._FlushInput() # discard stale input
|
status = self.GetStatus()
|
except Exception as e:
|
print >>sys.stderr, "error opening device %s: %s" % (dev, e)
|
continue
|
|
if not status:
|
print >>sys.stderr, "no response from device %s" % dev
|
elif serialno and status["serialNumber"] != serialno:
|
print >>sys.stderr, ("Note: another device serial #%d seen on %s" %
|
(status["serialNumber"], dev))
|
else:
|
self.start_voltage = status["voltage1"]
|
return
|
|
self._tempfile = None
|
if not wait: raise IOError("No device found")
|
print >>sys.stderr, "waiting for device..."
|
time.sleep(1)
|
|
|
def GetStatus(self):
|
""" Requests and waits for status. Returns status dictionary. """
|
|
# status packet format
|
STATUS_FORMAT = ">BBBhhhHhhhHBBBxBbHBHHHHBbbHHBBBbbbbbbbbbBH"
|
STATUS_FIELDS = [
|
"packetType", "firmwareVersion", "protocolVersion",
|
"mainFineCurrent", "usbFineCurrent", "auxFineCurrent", "voltage1",
|
"mainCoarseCurrent", "usbCoarseCurrent", "auxCoarseCurrent", "voltage2",
|
"outputVoltageSetting", "temperature", "status", "leds",
|
"mainFineResistor", "serialNumber", "sampleRate",
|
"dacCalLow", "dacCalHigh",
|
"powerUpCurrentLimit", "runTimeCurrentLimit", "powerUpTime",
|
"usbFineResistor", "auxFineResistor",
|
"initialUsbVoltage", "initialAuxVoltage",
|
"hardwareRevision", "temperatureLimit", "usbPassthroughMode",
|
"mainCoarseResistor", "usbCoarseResistor", "auxCoarseResistor",
|
"defMainFineResistor", "defUsbFineResistor", "defAuxFineResistor",
|
"defMainCoarseResistor", "defUsbCoarseResistor", "defAuxCoarseResistor",
|
"eventCode", "eventData", ]
|
|
self._SendStruct("BBB", 0x01, 0x00, 0x00)
|
while True: # Keep reading, discarding non-status packets
|
bytes = self._ReadPacket()
|
if not bytes: return None
|
if len(bytes) != struct.calcsize(STATUS_FORMAT) or bytes[0] != "\x10":
|
print >>sys.stderr, "wanted status, dropped type=0x%02x, len=%d" % (
|
ord(bytes[0]), len(bytes))
|
continue
|
|
status = dict(zip(STATUS_FIELDS, struct.unpack(STATUS_FORMAT, bytes)))
|
assert status["packetType"] == 0x10
|
for k in status.keys():
|
if k.endswith("VoltageSetting"):
|
status[k] = 2.0 + status[k] * 0.01
|
elif k.endswith("FineCurrent"):
|
pass # needs calibration data
|
elif k.endswith("CoarseCurrent"):
|
pass # needs calibration data
|
elif k.startswith("voltage") or k.endswith("Voltage"):
|
status[k] = status[k] * 0.000125
|
elif k.endswith("Resistor"):
|
status[k] = 0.05 + status[k] * 0.0001
|
if k.startswith("aux") or k.startswith("defAux"): status[k] += 0.05
|
elif k.endswith("CurrentLimit"):
|
status[k] = 8 * (1023 - status[k]) / 1023.0
|
return status
|
|
def RampVoltage(self, start, end):
|
v = start
|
if v < 3.0: v = 3.0 # protocol doesn't support lower than this
|
while (v < end):
|
self.SetVoltage(v)
|
v += .1
|
time.sleep(.1)
|
self.SetVoltage(end)
|
|
def SetVoltage(self, v):
|
""" Set the output voltage, 0 to disable. """
|
if v == 0:
|
self._SendStruct("BBB", 0x01, 0x01, 0x00)
|
else:
|
self._SendStruct("BBB", 0x01, 0x01, int((v - 2.0) * 100))
|
|
|
def SetMaxCurrent(self, i):
|
"""Set the max output current."""
|
assert i >= 0 and i <= 8
|
|
val = 1023 - int((i/8)*1023)
|
self._SendStruct("BBB", 0x01, 0x0a, val & 0xff)
|
self._SendStruct("BBB", 0x01, 0x0b, val >> 8)
|
|
def SetUsbPassthrough(self, val):
|
""" Set the USB passthrough mode: 0 = off, 1 = on, 2 = auto. """
|
self._SendStruct("BBB", 0x01, 0x10, val)
|
|
|
def StartDataCollection(self):
|
""" Tell the device to start collecting and sending measurement data. """
|
self._SendStruct("BBB", 0x01, 0x1b, 0x01) # Mystery command
|
self._SendStruct("BBBBBBB", 0x02, 0xff, 0xff, 0xff, 0xff, 0x03, 0xe8)
|
|
|
def StopDataCollection(self):
|
""" Tell the device to stop collecting measurement data. """
|
self._SendStruct("BB", 0x03, 0x00) # stop
|
|
|
def CollectData(self):
|
""" Return some current samples. Call StartDataCollection() first. """
|
while True: # loop until we get data or a timeout
|
bytes = self._ReadPacket()
|
if not bytes: return None
|
if len(bytes) < 4 + 8 + 1 or bytes[0] < "\x20" or bytes[0] > "\x2F":
|
print >>sys.stderr, "wanted data, dropped type=0x%02x, len=%d" % (
|
ord(bytes[0]), len(bytes))
|
continue
|
|
seq, type, x, y = struct.unpack("BBBB", bytes[:4])
|
data = [struct.unpack(">hhhh", bytes[x:x+8])
|
for x in range(4, len(bytes) - 8, 8)]
|
|
if self._last_seq and seq & 0xF != (self._last_seq + 1) & 0xF:
|
print >>sys.stderr, "data sequence skipped, lost packet?"
|
self._last_seq = seq
|
|
if type == 0:
|
if not self._coarse_scale or not self._fine_scale:
|
print >>sys.stderr, "waiting for calibration, dropped data packet"
|
continue
|
|
def scale(val):
|
if val & 1:
|
return ((val & ~1) - self._coarse_zero) * self._coarse_scale
|
else:
|
return (val - self._fine_zero) * self._fine_scale
|
|
out_main = []
|
out_usb = []
|
for main, usb, aux, voltage in data:
|
out_main.append(scale(main))
|
out_usb.append(scale(usb))
|
return (out_main, out_usb)
|
|
elif type == 1:
|
self._fine_zero = data[0][0]
|
self._coarse_zero = data[1][0]
|
# print >>sys.stderr, "zero calibration: fine 0x%04x, coarse 0x%04x" % (
|
# self._fine_zero, self._coarse_zero)
|
|
elif type == 2:
|
self._fine_ref = data[0][0]
|
self._coarse_ref = data[1][0]
|
# print >>sys.stderr, "ref calibration: fine 0x%04x, coarse 0x%04x" % (
|
# self._fine_ref, self._coarse_ref)
|
|
else:
|
print >>sys.stderr, "discarding data packet type=0x%02x" % type
|
continue
|
|
if self._coarse_ref != self._coarse_zero:
|
self._coarse_scale = 2.88 / (self._coarse_ref - self._coarse_zero)
|
if self._fine_ref != self._fine_zero:
|
self._fine_scale = 0.0332 / (self._fine_ref - self._fine_zero)
|
|
|
def _SendStruct(self, fmt, *args):
|
""" Pack a struct (without length or checksum) and send it. """
|
data = struct.pack(fmt, *args)
|
data_len = len(data) + 1
|
checksum = (data_len + sum(struct.unpack("B" * len(data), data))) % 256
|
out = struct.pack("B", data_len) + data + struct.pack("B", checksum)
|
self.ser.write(out)
|
|
|
def _ReadPacket(self):
|
""" Read a single data record as a string (without length or checksum). """
|
len_char = self.ser.read(1)
|
if not len_char:
|
print >>sys.stderr, "timeout reading from serial port"
|
return None
|
|
data_len = struct.unpack("B", len_char)
|
data_len = ord(len_char)
|
if not data_len: return ""
|
|
result = self.ser.read(data_len)
|
if len(result) != data_len: return None
|
body = result[:-1]
|
checksum = (data_len + sum(struct.unpack("B" * len(body), body))) % 256
|
if result[-1] != struct.pack("B", checksum):
|
print >>sys.stderr, "invalid checksum from serial port"
|
return None
|
return result[:-1]
|
|
def _FlushInput(self):
|
""" Flush all read data until no more available. """
|
self.ser.flush()
|
flushed = 0
|
while True:
|
ready_r, ready_w, ready_x = select.select([self.ser], [], [self.ser], 0)
|
if len(ready_x) > 0:
|
print >>sys.stderr, "exception from serial port"
|
return None
|
elif len(ready_r) > 0:
|
flushed += 1
|
self.ser.read(1) # This may cause underlying buffering.
|
self.ser.flush() # Flush the underlying buffer too.
|
else:
|
break
|
if flushed > 0:
|
print >>sys.stderr, "dropped >%d bytes" % flushed
|
|
def main(argv):
|
""" Simple command-line interface for Monsoon."""
|
useful_flags = ["voltage", "status", "usbpassthrough", "samples", "current"]
|
if not [f for f in useful_flags if FLAGS.get(f, None) is not None]:
|
print __doc__.strip()
|
print FLAGS.MainModuleHelp()
|
return
|
|
if FLAGS.includeusb:
|
num_channels = 2
|
else:
|
num_channels = 1
|
|
if FLAGS.avg and FLAGS.avg < 0:
|
print "--avg must be greater than 0"
|
return
|
|
mon = Monsoon(device=FLAGS.device, serialno=FLAGS.serialno)
|
|
if FLAGS.voltage is not None:
|
if FLAGS.ramp is not None:
|
mon.RampVoltage(mon.start_voltage, FLAGS.voltage)
|
else:
|
mon.SetVoltage(FLAGS.voltage)
|
|
if FLAGS.current is not None:
|
mon.SetMaxCurrent(FLAGS.current)
|
|
if FLAGS.status:
|
items = sorted(mon.GetStatus().items())
|
print "\n".join(["%s: %s" % item for item in items])
|
|
if FLAGS.usbpassthrough:
|
if FLAGS.usbpassthrough == 'off':
|
mon.SetUsbPassthrough(0)
|
elif FLAGS.usbpassthrough == 'on':
|
mon.SetUsbPassthrough(1)
|
elif FLAGS.usbpassthrough == 'auto':
|
mon.SetUsbPassthrough(2)
|
else:
|
sys.exit('bad passthrough flag: %s' % FLAGS.usbpassthrough)
|
|
if FLAGS.samples:
|
# Make sure state is normal
|
mon.StopDataCollection()
|
status = mon.GetStatus()
|
native_hz = status["sampleRate"] * 1000
|
|
# Collect and average samples as specified
|
mon.StartDataCollection()
|
|
# In case FLAGS.hz doesn't divide native_hz exactly, use this invariant:
|
# 'offset' = (consumed samples) * FLAGS.hz - (emitted samples) * native_hz
|
# This is the error accumulator in a variation of Bresenham's algorithm.
|
emitted = offset = 0
|
chan_buffers = tuple([] for _ in range(num_channels))
|
# past n samples for rolling average
|
history_deques = tuple(collections.deque() for _ in range(num_channels))
|
|
try:
|
last_flush = time.time()
|
while emitted < FLAGS.samples or FLAGS.samples == -1:
|
# The number of raw samples to consume before emitting the next output
|
need = (native_hz - offset + FLAGS.hz - 1) / FLAGS.hz
|
if need > len(chan_buffers[0]): # still need more input samples
|
chans_samples = mon.CollectData()
|
if not all(chans_samples): break
|
for chan_buffer, chan_samples in zip(chan_buffers, chans_samples):
|
chan_buffer.extend(chan_samples)
|
else:
|
# Have enough data, generate output samples.
|
# Adjust for consuming 'need' input samples.
|
offset += need * FLAGS.hz
|
while offset >= native_hz: # maybe multiple, if FLAGS.hz > native_hz
|
this_sample = [sum(chan[:need]) / need for chan in chan_buffers]
|
|
if FLAGS.timestamp: print int(time.time()),
|
|
if FLAGS.avg:
|
chan_avgs = []
|
for chan_deque, chan_sample in zip(history_deques, this_sample):
|
chan_deque.appendleft(chan_sample)
|
if len(chan_deque) > FLAGS.avg: chan_deque.pop()
|
chan_avgs.append(sum(chan_deque) / len(chan_deque))
|
# Interleave channel rolling avgs with latest channel data
|
data_to_print = [datum
|
for pair in zip(this_sample, chan_avgs)
|
for datum in pair]
|
else:
|
data_to_print = this_sample
|
|
fmt = ' '.join('%f' for _ in data_to_print)
|
print fmt % tuple(data_to_print)
|
|
sys.stdout.flush()
|
|
offset -= native_hz
|
emitted += 1 # adjust for emitting 1 output sample
|
chan_buffers = tuple(c[need:] for c in chan_buffers)
|
now = time.time()
|
if now - last_flush >= 0.99: # flush every second
|
sys.stdout.flush()
|
last_flush = now
|
except KeyboardInterrupt:
|
print >>sys.stderr, "interrupted"
|
|
mon.StopDataCollection()
|
|
|
if __name__ == '__main__':
|
# Define flags here to avoid conflicts with people who use us as a library
|
flags.DEFINE_boolean("status", None, "Print power meter status")
|
flags.DEFINE_integer("avg", None,
|
"Also report average over last n data points")
|
flags.DEFINE_float("voltage", None, "Set output voltage (0 for off)")
|
flags.DEFINE_float("current", None, "Set max output current")
|
flags.DEFINE_string("usbpassthrough", None, "USB control (on, off, auto)")
|
flags.DEFINE_integer("samples", None,
|
"Collect and print this many samples. "
|
"-1 means collect indefinitely.")
|
flags.DEFINE_integer("hz", 5000, "Print this many samples/sec")
|
flags.DEFINE_string("device", None,
|
"Path to the device in /dev/... (ex:/dev/ttyACM1)")
|
flags.DEFINE_integer("serialno", None, "Look for a device with this serial number")
|
flags.DEFINE_boolean("timestamp", None,
|
"Also print integer (seconds) timestamp on each line")
|
flags.DEFINE_boolean("ramp", True, "Gradually increase voltage")
|
flags.DEFINE_boolean("includeusb", False, "Include measurements from USB channel")
|
|
main(FLAGS(sys.argv))
|
