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
default_drivers = [('demo', {'analog_channels': 1})]
default_loglevel = sr.LogLevel.WARN
+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)')
+ parser.add_argument('--pyside',
+ action='store_true',
+ default=False,
+ help='Force use of PySide (default is to use PyQt4)')
+ args = parser.parse_args()
+
+ result = {
+ 'drivers': default_drivers,
+ 'loglevel': default_loglevel,
+ 'pyside': args.pyside
+ }
+
+ 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__':
+ # The command line parsing and import of the Qt modules is done here,
+ # so that the modules are imported before the classes below derive
+ # from classes therein. The rest of the main function is at the bottom.
+
+ args = parse_cli()
+
+ global qt_signal
+
+ if args['pyside']:
+ from PySide import QtCore, QtGui
+ qt_signal = QtCore.Signal
+ else:
+ try:
+ from PyQt4 import QtCore, QtGui
+ qt_signal = QtCore.pyqtSignal
+ except:
+ sys.stderr.write('import of PyQt4 failed, using PySide\n')
+ from PySide import QtCore, QtGui
+ qt_signal = QtCore.Signal
+
class SamplingThread(QtCore.QObject):
'''A class that handles the reception of sigrok packets in the background.'''
'''Helper class that does the actual work in another thread.'''
'''Signal emitted when new data arrived.'''
- measured = QtCore.pyqtSignal(object)
+ measured = qt_signal(object)
+
+ '''Signal emmited in case of an error.'''
+ error = qt_signal(str)
def __init__(self, drivers, loglevel):
super(self.__class__, self).__init__()
+ self.sampling = False
+ self.drivers = drivers
+
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:
+ def start_sampling(self):
+ devices = []
+ for name, options in self.drivers:
try:
dr = self.context.drivers[name]
- self.devices.append(dr.scan(**options)[0])
+ devices.append(dr.scan(**options)[0])
except:
- print('error getting device for driver "{}", skipping'.format(name))
-
- if not self.devices:
- print('no devices found')
+ self.error.emit(
+ 'Unable to get device for driver "{}".'.format(name))
+ return
- def start_sampling(self):
self.session = self.context.create_session()
- for dev in self.devices:
+ for dev in devices:
self.session.add_device(dev)
dev.open()
self.session.add_datafeed_callback(self.callback)
self.session.start()
+ self.sampling = True
self.session.run()
+ # If sampling is 'True' here, it means that 'stop_sampling()' was
+ # not called, therefore 'session.run()' ended too early, indicating
+ # an error.
+ if self.sampling:
+ self.error.emit('An error occured during the acquisition.')
+
def stop_sampling(self):
- self.session.stop()
+ if self.sampling:
+ self.sampling = False
+ self.session.stop()
def callback(self, device, packet):
if packet.type == sr.PacketType.ANALOG:
self.thread().msleep(100)
# signal used to start the worker across threads
- _start_signal = QtCore.pyqtSignal()
+ _start_signal = qt_signal()
def __init__(self, drivers, loglevel):
super(self.__class__, self).__init__()
self._start_signal.connect(self.worker.start_sampling)
+ # expose the signals of the worker
self.measured = self.worker.measured
+ self.error = self.worker.error
self.thread.start()
self.thread = thread
self.thread.measured.connect(self.update, QtCore.Qt.QueuedConnection)
+ self.thread.error.connect(self.error)
self.thread.start()
def setup_ui(self):
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
+ def error(self, msg):
+ '''Error handler for the sampling thread.'''
+ QtGui.QMessageBox.critical(self, 'Error', msg)
+ self.close()
if __name__ == '__main__':
- args = parse_cli()
-
thread = SamplingThread(args['drivers'], args['loglevel'])
app = QtGui.QApplication([])
projects / sigrok-meter.git / blobdiff