[sigrok-meter.git] / sigrok-meter

#!/usr/bin/env python

##
## This file is part of the sigrok-meter project.
##
## Copyright (C) 2013 Uwe Hermann <uwe@hermann-uwe.de>
## Copyright (C) 2014 Jens Steinhauser <jens.steinhauser@gmail.com>
##
## This program is free software; you can redistribute it and/or modify
## it under the terms of the GNU General Public License as published by
## the Free Software Foundation; either version 2 of the License, or
## (at your option) any later version.
##
## This program is distributed in the hope that it will be useful,
## but WITHOUT ANY WARRANTY; without even the implied warranty of
## MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
## GNU General Public License for more details.
##
## You should have received a copy of the GNU General Public License
## along with this program; if not, write to the Free Software
## Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA  02110-1301 USA
##

import argparse
import datetime
import os.path
import PyQt4.QtCore as QtCore
import PyQt4.QtGui as QtGui
import re
import sigrok.core as sr
import sys
import textwrap

default_drivers = [('demo', {'analog_channels': 1})]
default_loglevel = sr.LogLevel.WARN

class SamplingThread(QtCore.QObject):
    '''A class that handles the reception of sigrok packets in the background.'''

    class Worker(QtCore.QObject):
        '''Helper class that does the actual work in another thread.'''

        '''Signal emitted when new data arrived.'''
        measured = QtCore.pyqtSignal(object)

        def __init__(self, drivers, loglevel):
            super(self.__class__, self).__init__()

            self.context = sr.Context_create()
            self.context.log_level = loglevel

            self.sr_pkg_version = self.context.package_version
            self.sr_lib_version = self.context.lib_version

            self.devices = []
            for name, options in drivers:
                try:
                    dr = self.context.drivers[name]
                    self.devices.append(dr.scan(**options)[0])
                except:
                    print('error getting device for driver "{}", skipping'.format(name))

            if not self.devices:
                print('no devices found')

        def start_sampling(self):
            self.session = self.context.create_session()
            for dev in self.devices:
                self.session.add_device(dev)
                dev.open()
            self.session.add_datafeed_callback(self.callback)
            self.session.start()
            self.session.run()

        def stop_sampling(self):
            self.session.stop()

        def callback(self, device, packet):
            if packet.type == sr.PacketType.ANALOG:
                dev = '{} {}'.format(device.vendor, device.model)

                # only send the most recent value
                mag = packet.payload.data[0][-1]

                self.measured.emit((dev, mag, packet.payload.unit,
                        packet.payload.mq_flags))

            # wait a short time so that in any case we don't flood the GUI
            # with new data (for example if the demo device is used)
            self.thread().msleep(100)

    # signal used to start the worker across threads
    _start_signal = QtCore.pyqtSignal()

    def __init__(self, drivers, loglevel):
        super(self.__class__, self).__init__()

        self.worker = self.Worker(drivers, loglevel)
        self.thread = QtCore.QThread()
        self.worker.moveToThread(self.thread)

        self._start_signal.connect(self.worker.start_sampling)

        self.measured = self.worker.measured

        self.thread.start()

    def start(self):
        '''Starts sampling'''
        self._start_signal.emit()

    def stop(self):
        '''Stops sampling and the background thread.'''
        self.worker.stop_sampling()
        self.thread.quit()
        # the timeout is needed when the demo device is used, because it
        # produces so much outstanding data that quitting takes a long time
        self.thread.wait(500)

    def sr_pkg_version(self):
        '''Returns the version number of the libsigrok package.'''
        return self.worker.sr_pkg_version

    def sr_lib_version(self):
        '''Returns the version number fo the libsigrok library.'''
        return self.worker.sr_lib_version

class SigrokMeter(QtGui.QMainWindow):
    '''The main window of the application.'''

    def __init__(self, thread):
        super(SigrokMeter, self).__init__()
        self.setup_ui()

        self.inf = float('inf')

        self.thread = thread
        self.thread.measured.connect(self.update, QtCore.Qt.QueuedConnection)
        self.thread.start()

    def setup_ui(self):
        self.setWindowTitle('sigrok-meter')
        self.setMinimumHeight(130)
        self.setMinimumWidth(260)

        p = os.path.abspath(os.path.dirname(__file__))
        p = os.path.join(p, 'sigrok-logo-notext.png')
        self.setWindowIcon(QtGui.QIcon(p))

        actionQuit = QtGui.QAction(self)
        actionQuit.setText('&Quit')
        actionQuit.setIcon(QtGui.QIcon.fromTheme('application-exit'))
        actionQuit.setShortcut('Ctrl+Q')
        actionQuit.triggered.connect(self.close)

        actionAbout = QtGui.QAction(self)
        actionAbout.setText('&About')
        actionAbout.setIcon(QtGui.QIcon.fromTheme('help-about'))
        actionAbout.triggered.connect(self.show_about)

        menubar = self.menuBar()
        menuFile = menubar.addMenu('&File')
        menuFile.addAction(actionQuit)
        menuHelp = menubar.addMenu('&Help')
        menuHelp.addAction(actionAbout)

        self.lblValue = QtGui.QLabel('waiting for data...')
        self.lblValue.setAlignment(QtCore.Qt.AlignCenter)
        font = self.lblValue.font()
        font.setPointSize(font.pointSize() * 1.7)
        font.setBold(True)
        self.lblValue.setFont(font)
        self.setCentralWidget(self.lblValue)
        self.centralWidget().setContentsMargins(0, 0, 0, 0)

        self.lblDevName = QtGui.QLabel()
        self.lblDevName.setToolTip('Name of used measurement device.')
        self.statusBar().insertWidget(0, self.lblDevName, 10)
        self.lblTime = QtGui.QLabel()
        self.lblTime.setToolTip('Time of the last measurement.')
        self.statusBar().insertWidget(1, self.lblTime)

        self.statusBar().setSizeGripEnabled(False)

    def show_about(self):
        text = textwrap.dedent('''\
            <div align="center">
                <b>sigrok-meter</b><br/>
                0.1.0<br/>
                Using libsigrok {} (lib version {}).<br/>
                <a href='http://www.sigrok.org'>
                         http://www.sigrok.org</a><br/>
                <br/>
                This program comes with ABSOLUTELY NO WARRANTY;<br/>
                for details visit
                <a href='http://www.gnu.org/licenses/gpl.html'>
                         http://www.gnu.org/licenses/gpl.html</a>
            </div>
        '''.format(self.thread.sr_pkg_version(), self.thread.sr_lib_version()))

        QtGui.QMessageBox.about(self, 'About sigrok-meter', text)

    def format_unit(self, u):
        units = {
            sr.Unit.VOLT:                   'V',
            sr.Unit.AMPERE:                 'A',
            sr.Unit.OHM:                   u'\u03A9',
            sr.Unit.FARAD:                  'F',
            sr.Unit.KELVIN:                 'K',
            sr.Unit.CELSIUS:               u'\u00B0C',
            sr.Unit.FAHRENHEIT:            u'\u00B0F',
            sr.Unit.HERTZ:                  'Hz',
            sr.Unit.PERCENTAGE:             '%',
          # sr.Unit.BOOLEAN
            sr.Unit.SECOND:                 's',
            sr.Unit.SIEMENS:                'S',
            sr.Unit.DECIBEL_MW:             'dBu',
            sr.Unit.DECIBEL_VOLT:           'dBV',
          # sr.Unit.UNITLESS
            sr.Unit.DECIBEL_SPL:            'dB',
          # sr.Unit.CONCENTRATION
            sr.Unit.REVOLUTIONS_PER_MINUTE: 'rpm',
            sr.Unit.VOLT_AMPERE:            'VA',
            sr.Unit.WATT:                   'W',
            sr.Unit.WATT_HOUR:              'Wh',
            sr.Unit.METER_SECOND:           'm/s',
            sr.Unit.HECTOPASCAL:            'hPa',
            sr.Unit.HUMIDITY_293K:          '%rF',
            sr.Unit.DEGREE:                u'\u00B0',
            sr.Unit.HENRY:                  'H'
        }

        return units.get(u, '')

    def format_mqflags(self, mqflags):
        if sr.QuantityFlag.AC in mqflags:
            s = 'AC'
        elif sr.QuantityFlag.DC in mqflags:
            s = 'DC'
        else:
            s = ''

        return s

    def format_mag(self, mag):
        if mag == self.inf:
            return u'\u221E'
        return '{:f}'.format(mag)

    def update(self, data):
        '''Updates the labels with new measurement values.'''

        device, mag, unit, mqflags = data

        unit_str = self.format_unit(unit)
        mqflags_str = self.format_mqflags(mqflags)
        mag_str = self.format_mag(mag)
        value = ' '.join([mag_str, unit_str, mqflags_str])

        n = datetime.datetime.now().time()
        now = '{:02}:{:02}:{:02}.{:03}'.format(
                n.hour, n.minute, n.second, n.microsecond / 1000)

        self.lblValue.setText(value)
        self.lblDevName.setText(device)
        self.lblTime.setText(now)

def parse_cli():
    parser = argparse.ArgumentParser(
        description='Simple sigrok GUI for multimeters and dataloggers.',
        epilog=textwrap.dedent('''\
            The DRIVER string is the same as for sigrok-cli(1).

            examples:

              %(prog)s --driver tecpel-dmm-8061-ser:conn=/dev/ttyUSB0

              %(prog)s --driver uni-t-ut61e:conn=1a86.e008
        '''),
        formatter_class=argparse.RawDescriptionHelpFormatter)

    parser.add_argument('-d', '--driver',
        action='append',
        help='The driver to use')
    parser.add_argument('-l', '--loglevel',
        type=int,
        help='Set loglevel (5 is most verbose)')
    args = parser.parse_args()

    result = {
        'drivers': default_drivers,
        'loglevel': default_loglevel
    }

    if args.driver:
        result['drivers'] = []
        for d in args.driver:
            m = re.match('(?P<name>[^:]+)(?P<opts>(:[^:=]+=[^:=]+)*)', d)
            if not m:
                sys.exit('error parsing option "{}"'.format(d))

            opts = m.group('opts').split(':')[1:]
            opts = [tuple(kv.split('=')) for kv in opts]
            opts = dict(opts)

            result['drivers'].append((m.group('name'), opts))

    if args.loglevel != None:
        try:
            result['loglevel'] = sr.LogLevel.get(args.loglevel)
        except:
            sys.exit('error: invalid log level')

    return result

if __name__ == '__main__':
    args = parse_cli()

    thread = SamplingThread(args['drivers'], args['loglevel'])

    app = QtGui.QApplication([])
    s = SigrokMeter(thread)
    s.show()

    r = app.exec_()
    thread.stop()
    sys.exit(r)
Commit	Line	Data
	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)