[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

def format_unit(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, '')

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, 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:
                data = packet.payload.data
                unit_str = format_unit(packet.payload.unit)
                mqflags, mqflags_str = packet.payload.mq_flags, ""

                if sr.QuantityFlag.AC in mqflags:
                    mqflags_str = "AC"
                elif sr.QuantityFlag.DC in mqflags:
                    mqflags_str = "DC"

                for i in range(packet.payload.num_samples):
                    dev = "%s %s" % (device.vendor, device.model)
                    mag_str = "%f" % data[0][i]
                    val = ' '.join([mag_str, unit_str, mqflags_str])

                    self.measured.emit(dev, val)

            # 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.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 update(self, device, value):
        '''Updates the labels with new measurement values.'''

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