#!/usr/bin/env python3 """A PyQt5 GUI providing a plot window for visualizing numeric output from a CircuitPython script. """ ################################################################ # Written in 2018-2021 by Garth Zeglin # To the extent possible under law, the author has dedicated all copyright # and related and neighboring rights to this software to the public domain # worldwide. This software is distributed without any warranty. # You should have received a copy of the CC0 Public Domain Dedication along with this software. # If not, see . ################################################################ # standard Python libraries import os, sys, struct, time, logging, functools, queue, signal, getpass import numpy as np # documentation: https://doc.qt.io/qt-5/index.html # documentation: https://www.riverbankcomputing.com/static/Docs/PyQt5/index.html from PyQt5 import QtCore, QtGui, QtWidgets, QtSerialPort # documentation on pyqtgraph: https://pyqtgraph.org/documentation/plotting.html import pyqtgraph # configure logging output log = logging.getLogger('main') log.setLevel(logging.INFO) ################################################################ class QtPicoSerial(QtCore.QObject): """Class to manage a serial connection to a CircuitPython sketch using Qt QSerialPort object for data transport. The data protocol is based on lines of text.""" # class variable with Qt signal used to communicate between background thread and serial port thread _threadedWrite = QtCore.pyqtSignal(bytes, name='threadedWrite') def __init__(self, main): super(QtPicoSerial,self).__init__() self._portname = None self._buffer = b'' self._port = None self.log = logging.getLogger('pico') self.log.setLevel(logging.INFO) # self.log.setLevel(logging.DEBUG) self.main = main return def is_open(self): return self._port is not None def available_ports(self): """Return a list of names of available serial ports.""" return [port.portName() for port in QtSerialPort.QSerialPortInfo.availablePorts()] def set_port(self, name): self._portname = name def open(self): """Open the serial port and initialize communications. If the port is already open, this will close it first. If the current name is None, this will not open anything. Returns True if the port is open, else False.""" if self._port is not None: self.close() if self._portname is None: self.log.debug("No port name provided so not opening port.") return False self._port = QtSerialPort.QSerialPort() self._port.setBaudRate(115200) self._port.setPortName(self._portname) # open the serial port, which should also reset the Pico if self._port.open(QtCore.QIODevice.ReadWrite): self.log.info("Opened serial port %s", self._port.portName()) # always process data as it becomes available self._port.readyRead.connect(self.read_input) # initialize the slot used to receive data from background threads self._threadedWrite.connect(self._data_send) return True else: # Error codes: https://doc.qt.io/qt-5/qserialport.html#SerialPortError-enum errcode = self._port.error() if errcode == QtSerialPort.QSerialPort.PermissionError: self.log.warning("Failed to open serial port %s with a QSerialPort PermissionError, which could involve an already running control process, a stale lock file, or dialout group permissions.", self._port.portName()) else: self.log.warning("Failed to open serial port %s with a QSerialPort error code %d.", self._port.portName(), errcode) self._port = None return False def set_and_open_port(self, name): self.set_port(name) self.open() def close(self): """Shut down the serial connection to the Pico.""" if self._port is not None: self.log.info("Closing serial port %s", self._port.portName()) self._port.close() self._port = None return def write(self, data): if self._port is not None: self._port.write(data) else: self.log.debug("Serial port not open during write.") @QtCore.pyqtSlot(bytes) def _data_send(self, data): """Slot to receive serial data on the main thread.""" self.write(data) def thread_safe_write(self, data): """Function to receive data to transmit from a background thread, then send it as a signal to a slot on the main thread.""" self._threadedWrite.emit(data) def read_input(self): # Read as much input as available; callback from Qt event loop. data = self._port.readAll() if len(data) > 0: self.data_received(data) return def _parse_serial_input(self, data): # parse a single line of status input provided as a bytestring tokens = data.split() self.log.debug("Received serial data: %s", tokens) self.main.pico_data_received(data) def data_received(self, data): # Manage the possibility of partial reads by appending new data to any previously received partial line. # The data arrives as a PyQT5.QtCore.QByteArray. self._buffer += bytes(data) # Process all complete newline-terminated lines. while b'\n' in self._buffer: first, self._buffer = self._buffer.split(b'\n', 1) first = first.rstrip() self._parse_serial_input(first) def send(self, string): self.log.debug("Sending to serial port: %s", string) self.write(string.encode()+b'\n') return ################################################################ class MainGUI(QtWidgets.QMainWindow): """A custom main window which provides all GUI controls. Requires a delegate main application object to handle user requests.""" def __init__(self, main, *args, **kwargs): super(MainGUI,self).__init__() # save the main object for delegating GUI events self.main = main # create the GUI elements self.console_queue = queue.Queue() self.setupUi() self._handler = None self.enable_console_logging() # finish initialization self.show() # manage the console output across threads self.console_timer = QtCore.QTimer() self.console_timer.timeout.connect(self._poll_console_queue) self.console_timer.start(50) # units are milliseconds return # ------------------------------------------------------------------------------------------------ def setupUi(self): self.setWindowTitle("Pico Plotter") self.resize(600, 800) self.centralwidget = QtWidgets.QWidget(self) self.setCentralWidget(self.centralwidget) self.verticalLayout = QtWidgets.QVBoxLayout(self.centralwidget) self.verticalLayout.setContentsMargins(-1, -1, -1, 9) # left, top, right, bottom # generate fields for configuring the Pico serial port hbox = QtWidgets.QHBoxLayout() self.verticalLayout.addLayout(hbox) hbox.addWidget(QtWidgets.QLabel("Pico serial port:")) self.portSelector = QtWidgets.QComboBox() hbox.addWidget(self.portSelector) self.update_port_selector() self.portSelector.activated['QString'].connect(self.pico_port_selected) rescan = QtWidgets.QPushButton('Rescan Serial Ports') rescan.pressed.connect(self.update_port_selector) hbox.addWidget(rescan) # Pico connection indicator and connect/disconnect buttons hbox = QtWidgets.QHBoxLayout() self.verticalLayout.addLayout(hbox) self.pico_connected = QtWidgets.QLabel() self.pico_connected.setLineWidth(3) self.pico_connected.setFrameStyle(QtWidgets.QFrame.Box) self.pico_connected.setAlignment(QtCore.Qt.AlignCenter) sizePolicy = QtWidgets.QSizePolicy(QtWidgets.QSizePolicy.Fixed, QtWidgets.QSizePolicy.Fixed) self.pico_connected.setSizePolicy(sizePolicy) self.set_pico_connected_state(False) hbox.addWidget(self.pico_connected) connect = QtWidgets.QPushButton('Connect') connect.pressed.connect(self.main.connect_to_pico) hbox.addWidget(connect) disconnect = QtWidgets.QPushButton('Disconnect') disconnect.pressed.connect(self.main.disconnect_from_pico) hbox.addWidget(disconnect) # text area for displaying both internal and received messages self.consoleOutput = QtWidgets.QPlainTextEdit() self.consoleOutput.setHorizontalScrollBarPolicy(QtCore.Qt.ScrollBarAsNeeded) self.verticalLayout.addWidget(self.consoleOutput) # main plotting area self.verticalLayout.addWidget(self.create_plot_widget()) # set up the status bar which appears at the bottom of the window self.statusbar = QtWidgets.QStatusBar(self) self.setStatusBar(self.statusbar) # set up the main menu self.menubar = QtWidgets.QMenuBar(self) self.menubar.setGeometry(QtCore.QRect(0, 0, 500, 22)) self.menubar.setNativeMenuBar(False) self.menubar.setObjectName("menubar") self.menuTitle = QtWidgets.QMenu(self.menubar) self.setMenuBar(self.menubar) self.actionQuit = QtWidgets.QAction(self) self.menuTitle.addAction(self.actionQuit) self.menubar.addAction(self.menuTitle.menuAction()) self.menuTitle.setTitle("File") self.actionQuit.setText("Quit") self.actionQuit.setShortcut("Ctrl+Q") self.actionQuit.triggered.connect(self.quitSelected) return # --- logging to screen ------------------------------------------------------------- def enable_console_logging(self): # get the root logger to receive all logging traffic logger = logging.getLogger() # create a logging handler which writes to the console window via self.write handler = logging.StreamHandler(self) handler.setFormatter(logging.Formatter('%(levelname)s:%(name)s: %(message)s')) logger.addHandler(handler) # logger.setLevel(logging.NOTSET) logger.setLevel(logging.DEBUG) handler.setLevel(logging.DEBUG) self._handler = handler log.info("Enabled logging in console window.") return def disable_console_logging(self): if self._handler is not None: logging.getLogger().removeHandler(self._handler) self._handler = None # --- window and qt event processing ------------------------------------------------------------- def show_status(self, string): self.statusbar.showMessage(string) def _poll_console_queue(self): """Write any queued console text to the console text area from the main thread.""" while not self.console_queue.empty(): string = str(self.console_queue.get()) stripped = string.rstrip() if stripped != "": self.consoleOutput.appendPlainText(stripped) return def write(self, string): """Write output to the console text area in a thread-safe way. Qt only allows calls from the main thread, but the service routines run on separate threads.""" self.console_queue.put(string) return def quitSelected(self): self.write("User selected quit.") self.close() def closeEvent(self, event): self.write("Received window close event.") self.main.app_is_exiting() self.disable_console_logging() super(MainGUI,self).closeEvent(event) def set_pico_connected_state(self, flag): if flag is True: self.pico_connected.setText(" Connected ") self.pico_connected.setStyleSheet("color: white; background-color: green;") else: self.pico_connected.setText(" Not Connected ") self.pico_connected.setStyleSheet("color: white; background-color: blue;") def update_port_selector(self): self.portSelector.clear() self.portSelector.addItem("") for port in QtSerialPort.QSerialPortInfo.availablePorts(): self.portSelector.insertItem(0, port.portName()) self.portSelector.setCurrentText(self.main.portname) # --- GUI widget event processing ---------------------------------------------------------------------- def pico_port_selected(self, name): self.write("Pico port selected: %s" % name) self.main.set_pico_port(name) def user_command_entered(self): payload = self.user_command.text() self.main.user_command(payload) self.user_command.clear() # --- plot widget management --------------------------------------------------------------------------- def create_plot_widget(self): """Create the plotter widget and associated data. Returns a PlotWidget.""" self.dataPlot = pyqtgraph.PlotWidget() self.plotLegend = pyqtgraph.LegendItem(offset=(-50, 50)) self.plotLegend.setParentItem(self.dataPlot.getPlotItem()) self.dataPlot.setTitle(title='Pico Data') self.dataPlot.setLabel(axis='bottom', text='samples') self.dataPlot.setXRange(0, 200) # self.dataPlot.enableAutoRange() # create a shared X axis sample index array self.plot_t_data = np.arange(100) # create lists to hold channel data self.plot_y_data = [] # list of numpy arrays self.plot_curves = [] # list of PlotCurveItem return self.dataPlot def add_plot_channel(self): samples = len(self.plot_t_data) ydata = np.zeros(samples) c_number = len(self.plot_curves) color = pyqtgraph.intColor(c_number) item = pyqtgraph.PlotCurveItem(x=self.plot_t_data, y=ydata, pen=pyqtgraph.mkPen(color=color, width=2)) self.plot_curves.append(item) self.plot_y_data.append(ydata) self.dataPlot.addItem(item) self.plotLegend.addItem(item, str(c_number+1)) def add_plot_data(self, vector): """Add a new vector of sample data to the plot. Creates a new plot channel if needed. Shifts previous data to keep a constant number of plotted samples.""" needed = len(vector) - len(self.plot_curves) if needed > 0: for i in range(needed): self.add_plot_channel() # shift previous data down. last_sample_index = self.plot_t_data[-1] self.plot_t_data[0:-1] = self.plot_t_data[1:] for data in self.plot_y_data: data[0:-1] = data[1:] # add a new column at the end self.plot_t_data[-1] = last_sample_index+1 for data, sample in zip(self.plot_y_data, vector): data[-1] = sample # update the widget data for y_data, curve in zip(self.plot_y_data, self.plot_curves): curve.setData(x=self.plot_t_data, y=y_data) self.dataPlot.autoRange() ################################################################ class MainApp(object): """Main application object holding any non-GUI related state.""" def __init__(self): # Attach a handler to the keyboard interrupt (control-C). signal.signal(signal.SIGINT, self._sigint_handler) # load any available persistent application settings QtCore.QCoreApplication.setOrganizationName("IDeATe") QtCore.QCoreApplication.setOrganizationDomain("ideate.cmu.edu") QtCore.QCoreApplication.setApplicationName('pico_plotter') self.settings = QtCore.QSettings() # uncomment to restore 'factory defaults' # self.settings.clear() # Pico serial port name self.portname = self.settings.value('pico_port', '') self.payload = '' # create the interface window self.window = MainGUI(self) # Initialize the Pico interface system. self.pico = QtPicoSerial(self) self.pico.set_port(self.portname) self.window.show_status("Disconnected.") return ################################################################ def app_is_exiting(self): self.pico.close() def _sigint_handler(self, signal, frame): print("Keyboard interrupt caught, running close handlers...") self.app_is_exiting() sys.exit(0) ################################################################ def set_pico_port(self, name): self.settings.setValue('pico_port', name) self.portname = name self.pico.set_port(name) return def connect_to_pico(self): success = self.pico.open() self.window.set_pico_connected_state(success) self.window.show_status("Connected." if success else "Disconnected.") return def disconnect_from_pico(self): self.pico.close() self.window.set_pico_connected_state(False) self.window.show_status("Disconnected.") return def pico_data_received(self, payload): """Process a message from the Pico.""" log.info("Received: %s", payload) s = payload.decode() # convert from bytes to a string if s[0] == '(': tokens = s[1:].rstrip().rstrip(')').split(',') numbers = [] try: for s in tokens: numbers.append(float(s)) except ValueError: pass # stop on any conversion error self.window.add_plot_data(numbers) def user_command(self, payload): """Process a message from the user command line.""" log.debug("Received user command: %s", payload) self.pico.write(payload.encode()) ################################################################ def main(): # Optionally add an additional root log handler to stream messages to the terminal console. if False: console_handler = logging.StreamHandler() console_handler.setLevel(logging.DEBUG) console_handler.setFormatter(logging.Formatter('%(levelname)s:%(name)s: %(message)s')) logging.getLogger().addHandler(console_handler) # initialize the Qt system itself app = QtWidgets.QApplication(sys.argv) # create the main application controller main = MainApp() # run the event loop until the user is done log.info("Starting event loop.") sys.exit(app.exec_()) ################################################################ # Main script follows. This sequence is executed when the script is initiated from the command line. if __name__ == "__main__": main()