sigrok-meter

   1 #!/usr/bin/env python
   2
   3 ##
   4 ## This file is part of the sigrok-meter project.
   5 ##
   6 ## Copyright (C) 2013 Uwe Hermann <uwe@hermann-uwe.de>
   7 ## Copyright (C) 2014 Jens Steinhauser <jens.steinhauser@gmail.com>
   8 ##
   9 ## This program is free software; you can redistribute it and/or modify
  10 ## it under the terms of the GNU General Public License as published by
  11 ## the Free Software Foundation; either version 2 of the License, or
  12 ## (at your option) any later version.
  13 ##
  14 ## This program is distributed in the hope that it will be useful,
  15 ## but WITHOUT ANY WARRANTY; without even the implied warranty of
  16 ## MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  17 ## GNU General Public License for more details.
  18 ##
  19 ## You should have received a copy of the GNU General Public License
  20 ## along with this program; if not, write to the Free Software
  21 ## Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA  02110-1301 USA
  22 ##
  23
  24 import argparse
  25 import datetime
  26 import os.path
  27 import PyQt4.QtCore as QtCore
  28 import PyQt4.QtGui as QtGui
  29 import re
  30 import sigrok.core as sr
  31 import sys
  32 import textwrap
  33
  34 default_drivers = [('demo', {'analog_channels': 1})]
  35 default_loglevel = sr.LogLevel.WARN
  36
  37 class SamplingThread(QtCore.QObject):
  38     '''A class that handles the reception of sigrok packets in the background.'''
  39
  40     class Worker(QtCore.QObject):
  41         '''Helper class that does the actual work in another thread.'''
  42
  43         '''Signal emitted when new data arrived.'''
  44         measured = QtCore.pyqtSignal(object)
  45
  46         def __init__(self, drivers, loglevel):
  47             super(self.__class__, self).__init__()
  48
  49             self.context = sr.Context_create()
  50             self.context.log_level = loglevel
  51
  52             self.sr_pkg_version = self.context.package_version
  53             self.sr_lib_version = self.context.lib_version
  54
  55             self.devices = []
  56             for name, options in drivers:
  57                 try:
  58                     dr = self.context.drivers[name]
  59                     self.devices.append(dr.scan(**options)[0])
  60                 except:
  61                     print('error getting device for driver "{}", skipping'.format(name))
  62
  63             if not self.devices:
  64                 print('no devices found')
  65
  66         def start_sampling(self):
  67             self.session = self.context.create_session()
  68             for dev in self.devices:
  69                 self.session.add_device(dev)
  70                 dev.open()
  71             self.session.add_datafeed_callback(self.callback)
  72             self.session.start()
  73             self.session.run()
  74
  75         def stop_sampling(self):
  76             self.session.stop()
  77
  78         def callback(self, device, packet):
  79             if packet.type == sr.PacketType.ANALOG:
  80                 dev = '{} {}'.format(device.vendor, device.model)
  81
  82                 # only send the most recent value
  83                 mag = packet.payload.data[0][-1]
  84
  85                 self.measured.emit((dev, mag, packet.payload.unit,
  86                         packet.payload.mq_flags))
  87
  88             # wait a short time so that in any case we don't flood the GUI
  89             # with new data (for example if the demo device is used)
  90             self.thread().msleep(100)
  91
  92     # signal used to start the worker across threads
  93     _start_signal = QtCore.pyqtSignal()
  94
  95     def __init__(self, drivers, loglevel):
  96         super(self.__class__, self).__init__()
  97
  98         self.worker = self.Worker(drivers, loglevel)
  99         self.thread = QtCore.QThread()
 100         self.worker.moveToThread(self.thread)
 101
 102         self._start_signal.connect(self.worker.start_sampling)
 103
 104         self.measured = self.worker.measured
 105
 106         self.thread.start()
 107
 108     def start(self):
 109         '''Starts sampling'''
 110         self._start_signal.emit()
 111
 112     def stop(self):
 113         '''Stops sampling and the background thread.'''
 114         self.worker.stop_sampling()
 115         self.thread.quit()
 116         # the timeout is needed when the demo device is used, because it
 117         # produces so much outstanding data that quitting takes a long time
 118         self.thread.wait(500)
 119
 120     def sr_pkg_version(self):
 121         '''Returns the version number of the libsigrok package.'''
 122         return self.worker.sr_pkg_version
 123
 124     def sr_lib_version(self):
 125         '''Returns the version number fo the libsigrok library.'''
 126         return self.worker.sr_lib_version
 127
 128 class SigrokMeter(QtGui.QMainWindow):
 129     '''The main window of the application.'''
 130
 131     def __init__(self, thread):
 132         super(SigrokMeter, self).__init__()
 133         self.setup_ui()
 134
 135         self.inf = float('inf')
 136
 137         self.thread = thread
 138         self.thread.measured.connect(self.update, QtCore.Qt.QueuedConnection)
 139         self.thread.start()
 140
 141     def setup_ui(self):
 142         self.setWindowTitle('sigrok-meter')
 143         self.setMinimumHeight(130)
 144         self.setMinimumWidth(260)
 145
 146         p = os.path.abspath(os.path.dirname(__file__))
 147         p = os.path.join(p, 'sigrok-logo-notext.png')
 148         self.setWindowIcon(QtGui.QIcon(p))
 149
 150         actionQuit = QtGui.QAction(self)
 151         actionQuit.setText('&Quit')
 152         actionQuit.setIcon(QtGui.QIcon.fromTheme('application-exit'))
 153         actionQuit.setShortcut('Ctrl+Q')
 154         actionQuit.triggered.connect(self.close)
 155
 156         actionAbout = QtGui.QAction(self)
 157         actionAbout.setText('&About')
 158         actionAbout.setIcon(QtGui.QIcon.fromTheme('help-about'))
 159         actionAbout.triggered.connect(self.show_about)
 160
 161         menubar = self.menuBar()
 162         menuFile = menubar.addMenu('&File')
 163         menuFile.addAction(actionQuit)
 164         menuHelp = menubar.addMenu('&Help')
 165         menuHelp.addAction(actionAbout)
 166
 167         self.lblValue = QtGui.QLabel('waiting for data...')
 168         self.lblValue.setAlignment(QtCore.Qt.AlignCenter)
 169         font = self.lblValue.font()
 170         font.setPointSize(font.pointSize() * 1.7)
 171         font.setBold(True)
 172         self.lblValue.setFont(font)
 173         self.setCentralWidget(self.lblValue)
 174         self.centralWidget().setContentsMargins(0, 0, 0, 0)
 175
 176         self.lblDevName = QtGui.QLabel()
 177         self.lblDevName.setToolTip('Name of used measurement device.')
 178         self.statusBar().insertWidget(0, self.lblDevName, 10)
 179         self.lblTime = QtGui.QLabel()
 180         self.lblTime.setToolTip('Time of the last measurement.')
 181         self.statusBar().insertWidget(1, self.lblTime)
 182
 183         self.statusBar().setSizeGripEnabled(False)
 184
 185     def show_about(self):
 186         text = textwrap.dedent('''\
 187             <div align="center">
 188                 <b>sigrok-meter</b><br/>
 189                 0.1.0<br/>
 190                 Using libsigrok {} (lib version {}).<br/>
 191                 <a href='http://www.sigrok.org'>
 192                          http://www.sigrok.org</a><br/>
 193                 <br/>
 194                 This program comes with ABSOLUTELY NO WARRANTY;<br/>
 195                 for details visit
 196                 <a href='http://www.gnu.org/licenses/gpl.html'>
 197                          http://www.gnu.org/licenses/gpl.html</a>
 198             </div>
 199         '''.format(self.thread.sr_pkg_version(), self.thread.sr_lib_version()))
 200
 201         QtGui.QMessageBox.about(self, 'About sigrok-meter', text)
 202
 203     def format_unit(self, u):
 204         units = {
 205             sr.Unit.VOLT:                   'V',
 206             sr.Unit.AMPERE:                 'A',
 207             sr.Unit.OHM:                   u'\u03A9',
 208             sr.Unit.FARAD:                  'F',
 209             sr.Unit.KELVIN:                 'K',
 210             sr.Unit.CELSIUS:               u'\u00B0C',
 211             sr.Unit.FAHRENHEIT:            u'\u00B0F',
 212             sr.Unit.HERTZ:                  'Hz',
 213             sr.Unit.PERCENTAGE:             '%',
 214           # sr.Unit.BOOLEAN
 215             sr.Unit.SECOND:                 's',
 216             sr.Unit.SIEMENS:                'S',
 217             sr.Unit.DECIBEL_MW:             'dBu',
 218             sr.Unit.DECIBEL_VOLT:           'dBV',
 219           # sr.Unit.UNITLESS
 220             sr.Unit.DECIBEL_SPL:            'dB',
 221           # sr.Unit.CONCENTRATION
 222             sr.Unit.REVOLUTIONS_PER_MINUTE: 'rpm',
 223             sr.Unit.VOLT_AMPERE:            'VA',
 224             sr.Unit.WATT:                   'W',
 225             sr.Unit.WATT_HOUR:              'Wh',
 226             sr.Unit.METER_SECOND:           'm/s',
 227             sr.Unit.HECTOPASCAL:            'hPa',
 228             sr.Unit.HUMIDITY_293K:          '%rF',
 229             sr.Unit.DEGREE:                u'\u00B0',
 230             sr.Unit.HENRY:                  'H'
 231         }
 232
 233         return units.get(u, '')
 234
 235     def format_mqflags(self, mqflags):
 236         if sr.QuantityFlag.AC in mqflags:
 237             s = 'AC'
 238         elif sr.QuantityFlag.DC in mqflags:
 239             s = 'DC'
 240         else:
 241             s = ''
 242
 243         return s
 244
 245     def format_mag(self, mag):
 246         if mag == self.inf:
 247             return u'\u221E'
 248         return '{:f}'.format(mag)
 249
 250     def update(self, data):
 251         '''Updates the labels with new measurement values.'''
 252
 253         device, mag, unit, mqflags = data
 254
 255         unit_str = self.format_unit(unit)
 256         mqflags_str = self.format_mqflags(mqflags)
 257         mag_str = self.format_mag(mag)
 258         value = ' '.join([mag_str, unit_str, mqflags_str])
 259
 260         n = datetime.datetime.now().time()
 261         now = '{:02}:{:02}:{:02}.{:03}'.format(
 262                 n.hour, n.minute, n.second, n.microsecond / 1000)
 263
 264         self.lblValue.setText(value)
 265         self.lblDevName.setText(device)
 266         self.lblTime.setText(now)
 267
 268 def parse_cli():
 269     parser = argparse.ArgumentParser(
 270         description='Simple sigrok GUI for multimeters and dataloggers.',
 271         epilog=textwrap.dedent('''\
 272             The DRIVER string is the same as for sigrok-cli(1).
 273
 274             examples:
 275
 276               %(prog)s --driver tecpel-dmm-8061-ser:conn=/dev/ttyUSB0
 277
 278               %(prog)s --driver uni-t-ut61e:conn=1a86.e008
 279         '''),
 280         formatter_class=argparse.RawDescriptionHelpFormatter)
 281
 282     parser.add_argument('-d', '--driver',
 283         action='append',
 284         help='The driver to use')
 285     parser.add_argument('-l', '--loglevel',
 286         type=int,
 287         help='Set loglevel (5 is most verbose)')
 288     args = parser.parse_args()
 289
 290     result = {
 291         'drivers': default_drivers,
 292         'loglevel': default_loglevel
 293     }
 294
 295     if args.driver:
 296         result['drivers'] = []
 297         for d in args.driver:
 298             m = re.match('(?P<name>[^:]+)(?P<opts>(:[^:=]+=[^:=]+)*)', d)
 299             if not m:
 300                 sys.exit('error parsing option "{}"'.format(d))
 301
 302             opts = m.group('opts').split(':')[1:]
 303             opts = [tuple(kv.split('=')) for kv in opts]
 304             opts = dict(opts)
 305
 306             result['drivers'].append((m.group('name'), opts))
 307
 308     if args.loglevel != None:
 309         try:
 310             result['loglevel'] = sr.LogLevel.get(args.loglevel)
 311         except:
 312             sys.exit('error: invalid log level')
 313
 314     return result
 315
 316 if __name__ == '__main__':
 317     args = parse_cli()
 318
 319     thread = SamplingThread(args['drivers'], args['loglevel'])
 320
 321     app = QtGui.QApplication([])
 322     s = SigrokMeter(thread)
 323     s.show()
 324
 325     r = app.exec_()
 326     thread.stop()
 327     sys.exit(r)