Python 3, threading and references

Creating a thread

I used threading to enable real-time graphing of data from sensors. One thread collected data from the sensors. The main thread ran the real time graph. I had a few problems getting started. It came down to my incorrect use of brackets when creating the thread.

When we create a thread using the threading library, we need to pass the target to the thread without using brackets. e.g.

thread = threading.Thread(target=ThreadTest)

not

thread = threading.Thread(target=ThreadTest())

Otherwise the target is created in the main thread, which is what we are trying to avoid. Without the brackets, we pass a reference to the target. With the brackets, we have already created the object. I think that this is analogous to passing a pointer in C, but stand to be corrected.

Example

In test1.py I call ThreadTest without using brackets. test_thread starts in the thread and allows test1.py to continue running.

In test2.py, I pass ThreadTest() as the target. In this case the thread does not allow test2.py to continue running.

test1.py

import threading
from thread_test import ThreadTest

thread = threading.Thread(target=ThreadTest)
thread.start()
print('not blocked')

test2.py

import threading
from thread_test import ThreadTest

thread = threading.Thread(target=ThreadTest())
thread.start()
print('not blocked')

test_thread.py

from time import sleep


class ThreadTest():
    def __init__(self):
        print('thread_test started')
        while True:
            sleep(1)
            print('test_thread')

output from test1.py:

thread_test started
not blocked
test_thread
test_thread
test_thread

output from test2.py:

thread_test started
test_thread
test_thread
test_thread

Running pytest when the test files are in a different directory to the source files

I had a battle to get my testing directory structure to work outside of an IDE. Please find my solution below. Tested on Windows 7 using python 3.6 and Linux Mint using python 3.4, running the code using the command line:

python -m pytest test_compress_files.py

The file I wrote to be tested is called compress_files.py in a directory named \src. The file containing tests to be run using pytest is called test_compress_files.py in a subdirectory \tests, so the full directory path is \src\tests. I needed to add a file called context.py to the \src\tests directory. This file is used in test_compress_files.py to enable access to compress_files.py in the directory above. The __init__.py files are empty.

Directory structure:

\src
__init__.py
compress_files.py

\src\tests
__init__.py
context.py
test_compress_files.py  

compress_files.py contains the script to be tested.

context.py:

import os
import sys
sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), '..')))

import compress_files  

The line:

sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__)

comes from the suggestion at the hitch hikers guide to python at:

http://docs.python-guide.org/en/latest/writing/structure/.

This adds the path of the directory above the /src/tests directory to sys.path, which in this case is /src.

test_compress_files.py:

import os
import pytest
from .context import compress_files
from compress_files import *

# tests start here
...

I put this up as an answer to a stackoverflow question here.

Sending parameters to a Jupyter Notebook cell using click

Using libraries such as click and optparse we can send parameters to Python scripts when we run them from the command line. For example, passing a parameter called count with a value of 2 to a script called hello.py:

hello.py --count=2

How can I replicate this functionality in a cell of a Jupyter notebook? I like to run the same code in the notebook so that I can easily copy it to a stand alone script. Using sys.argv to pass parameters to the main function seemed one way to go and works with optparse:

from optparse import OptionParser
import sys


def main():
    parser = OptionParser()
    parser.add_option('-f', '--fake',
                      default='False',
                help='Fake data')
    (options,args) = parser.parse_args()
    print('options:{} args: {}'.format(options, args))
    if options.fake:
        print('Fake detected')
    
def test_args():
    
    print('hello')
    
if __name__ == '__main__':

    sys.argv = ['--fake', 'True' '--help']
    main()

output:

options:{'fake': 'False'} args: ['True--help']

Fake detected

Click seems to be flavor of the month, but I kept on getting a screen full of errors when I tried to run click through a Jupyter notebook cell. If we consider the Click example code:

import click

@click.command()
@click.option('--count', default=1, help='Number of greetings.')
@click.option('--name', prompt='Your name',
            help='The person to greet.')
def hello(count, name):
    """Simple program that greets NAME for a total of COUNT times."""
    for x in range(count):
        click.echo('Hello %s!' % name)

if __name__ == '__main__':
    hello()

If this file is called hello.py, then running from a command line:

hello.py 'Max' --count=3 

Gives this output:

Hello Max!

Hello Max!

Hello Max!

But using the same sys.argv trick that works with optparse produces a screen full of errors when the same code is run from a Jupyter notebook cell. The solution is to put the %%python magic at the start of the cell:

%%python

import sys
import click

@click.command()
@click.option('--count', default=1, help='Number of greetings.')
@click.option('--name', prompt='Your name',
            help='The person to greet.')
def hello(count, name):
    """Simple program that greets NAME for a total of COUNT times."""
    with open('echo.txt', 'w') as fobj:
        for x in range(count):
            click.echo('Hello %s!' % name)

if __name__ == '__main__':
    # first element is the script name, use empty string instead
    sys.argv = ['', '--name', 'Max', '--count', '3']
    hello()

A small tip, but one which cost me an hour or two of pondering. Finally I asked the hive mind of Stackoverflow. Please see this stackoverflow solution.

how to configure the accelerometer range on the microbit using micropython

This article details how to set the range of sensitivity on the accelerometer on the microbit board using micropython and the i2c interface. I am using v1.7.9 of micropython for the microbit, the mu editor and linux mint v17.
 

After listening to Joe Finney talk about his role in developing the microbit board I realised I could use it for some of my hand gesture assistive technology work. The accelerometer on the microbit board is an MMA8653FC, data sheet here. There are programming notes for this chip here. The default range for this chip is +/-2g. This can be reconfigured to be +/-4g or +/-8g. For some of the students I work with on gesture recognition I need the higher ranges. So I entered the world of microbit i2c programming. I chose the micropython platform as python is always the ‘second best choice’ for any programming application. Actually, I’m a fan of using C for embedded hardware, but in this case using micropython looked to be fastest way of getting a solution. I used the simple mu editor. Long story short, it’s all about syntax. Thanks go to fizban for his example microbit code to interface a microbit with an lcd display using i2c. After reading this code I fixed the mistake(s) I’d been making. The documentation for the i2c microbit micropython is here.

Here’s my working code:

''' microbit i2c communications with onboard accelerometer '''
from microbit import *

ACCELEROMETER = 0x1d
ACC_2G = [0x0e, 0x00]
ACC_4G = [0x0e, 0x01]
ACC_8G = [0x0e, 0x02]
CTRL_REG1_STANDBY = [0x2a, 0x00]
CTRL_REG_1_ACTIVE = [0x2a, 0x01]
PL_THS_REG = [0x14] # returns b'\x84'
PL_BF_ZCOMP = [0x13] # returns b'\44' = 'D'
WHO_AM_I = [0x0d] # returns 0x5a=b'Z'
XYZ_DATA_CFG = [0x0e]

def command(c):
''' send command to accelerometer '''
i2c.write(ACCELEROMETER, bytearray(c))

def i2c_read_acc(register):
''' read accelerometer register '''
i2c.write(ACCELEROMETER, bytearray(register), repeat=True)
read_byte = i2c.read(ACCELEROMETER, 1)
print('read: {}'.format(read_byte))

def main_text():
''' send accelerometer data as a string '''
print('starting main')
counter = 0
while True:
x = accelerometer.get_x()
y = accelerometer.get_y()
z = accelerometer.get_z()
counter = counter + 1
print('{} {} {} {}'.format(counter, x, y, z))
sleep(250)

print("sending i2c commands...")
print('reading PL_BF_ZCOMP :')
print(i2c_read_acc(PL_BF_ZCOMP))
print('reading WHO_AM_I')
print(i2c_read_acc(WHO_AM_I))
# check the initial accelerometer range
print('reading XYZ_DATA_CFG:')
print(i2c_read_acc(XYZ_DATA_CFG))
# change the accelerometer range
command(CTRL_REG1_STANDBY)
command(ACC_4G)
command(CTRL_REG_1_ACTIVE)
print('commands sent')
# check the accelerometer range
print('reading XYZ_DATA_CFG:')
print(i2c_read_acc(XYZ_DATA_CFG))
display.show(Image.MEH)
# main_text()

output:

reading PL_BF_ZCOMP :
read: b'D'
None
reading WHO_AM_I
read: b'Z'
None
reading XYZ_DATA_CFG:
read: b'\x00'
None
commands sent
reading XYZ_DATA_CFG:
read: b'\x01'
None

The onboard accelerometer has an i2c address of 0x1d. There is a good article on how to scan for and verify this address here. I set the variable ACCELEROMETER to be this value in line 4 so that I could refer to it throughout the code without having to remember the hex value. Too many hex values flying around – I’d be bound to make a mistake if I didn’t give them names.

To send a command over i2c, as shown in line 18 of the example code, you need to address the target then send the commands as a bytearray. In this case the target is the accelerometer. Typically we send two bytes to the accelerometer. The first specifies the register we want to change, the second the value we want to write to this register. For example, to set the accelerometer’s range of sensitivity, we need to set the value of the register called XYZ_DATA_CFG to the value that corresponds with the range we are after. The address of this register is 0x0e. To set the +/4G range, we want to set this register to be 0x01. Now the variable I set in line 6 should make sense. Look in the data sheet linked above for more details. Before we can change this register we have to set CTRL_REG1 to be inactive by writing 0x00 to it. After changing the XYZ_DATA_CFG register we have to set CTRL_REG1 to be active again by writing 0x01 to it. This is detailed in the accelerometer application notes which I linked at the start of this article.

If you uncomment the last line, then the raw accelerometer values will stream out. The last column are the values for the z-axis of the accelerometer. Lay the board flat on the table. With the default +/-2g range you will see the z-axis values being around +1024 or -1024 depending on if the board is face up or down. This corresponds to +/-1g on the +/-2g range. Now that the board is set to +/-4g, the values for +/-1 g will be +/-512. The maximum and minimum value for the accelerometer stays as +/-2048, but it is now spread over +/-4g. Similarly, if you go crazy and set the range to be +/-8g, then you will see +/-256 for the z-axis value from the accelerometer for the board laying flat. As you would expect, you have to wave the board harder to get it to max out when you set the sensitivity to the higher ranges compared with the default +/-2g range.

So what about the PL_BF_ZCOMP and WHO_AM_I registers that I read from in lines 43 and 45? These are two read only directories. Reading the values stored in these is a sanity check that the chip is turned on and I have working code. I read the XYZ_DATA_CFG before and after setting it to verify that the sensitivity range has been set. Read up on these registers in the data sheet.

Look at line 23. The repeat=True flag has to be set. This clears the ‘message end’ flag in the write command. The default for this flag is False, which means that the i2c write command has a ‘message end’ flag at the end of it, which terminates the operation. As we want to read from the chip in line 24, we need to not set the ‘message end’ flag. Otherwise you will just read 0xff. Can you guess why? The data line is held high for i2c, so if there is nothing coming out of the chip you are trying to read from, you just read a bunch of ‘1s’. Line 24 means ‘read 1 byte from the device with address ACCELEROMETER’.

Where I initially came unstuck was by sending data as individual bytes, using e.g. b’\x0e’ followed by b’\x02′ to try and change the XYZ_DATA_CFG register. This looks to be valid for the Adafruit implementation of micropython, but I couldn’t get it work.

parsing and unpacking python3 serial data containing double backslashes

I lost a day of my life figuring out how to parse serial data sent as bytes from the BBC Microbit using micropython. The problem is that the data byte string appears with double backslash characters instead of single backslashes when read in over a serial interface.
Actual data:

b'ST\\x00\\x00\\x00\\xe0\\xeaE\\x00\\x00HB\\x00\\x00`\\xc3\\x00\\x00\\x10C\\x00\\x00t\\xc4EN'

What I wanted as data:

b'ST\x00\x00\x00\xe0\xeaE\x00\x00HB\x00\x00`\xc3\x00\x00\x10C\x00\x00t\xc4EN'

So how to convert from one misformed byte string to the clean one that python 3 would use?
I really went around in circles on this one. In the end I used a kludge. But it works. My life can now move on.
I convert the double slash byte to a string. Then I use the replace method to replace ‘\\’ with ‘\’. Then I use the literal_eval function to recast it as a byte. I am open to suggestions for a cleaner way of doing this!
Here’s some example code I used in a jupyter notebook session. test2 is the misformed byte string received over the serial interface and test3 is the cleaned byte that I can now unpack and extract the data from.

from struct import *
from ast import literal_eval
PACKER = ('2s5f2s')
test2=b'ST\\x00\\x00\\x00\\xe0\\xeaE\\x00\\x00HB\\x00\\x00`\\xc3\\x00\\x00\\x10C\\x00\\x00t\\xc4EN'
test3 = str(test2)
test3 = test3.replace('\\\\', '\\')
print('{}'.format(test3))
test3 = literal_eval(test3)
print(test3)
print(unpack(PACKER,test3))

output:

b'ST\x00\x00\x00\xe0\xeaE\x00\x00HB\x00\x00`\xc3\x00\x00\x10C\x00\x00t\xc4EN'
b'ST\x00\x00\x00\xe0\xeaE\x00\x00HB\x00\x00`\xc3\x00\x00\x10C\x00\x00t\xc4EN'
(b'ST', 7516.0, 50.0, -224.0, 144.0, -976.0, b'EN')

The data was produced from reading the accelerometer on a BBC Microbit board then using

struct.pack(PACKER,scan).

I am programming the boards using micropython.
The data is packed using the packer format:

PACKER = ('2s5f2s')

The transmitted scan is constructed using:

values = (START, counter, DELTA, x, y, z, END)
scan = struct.pack(packer, *values)

Where values contains a START and END string (‘ST’ and ‘EN’ respectively), a constant called DELTA which represents the time in between samples and the x, y and z readings from the accelerometer. So PACKER means ‘2 characters followed by 5 floats followed by 2 characters’.
I was being obstinate in sending bytes over the serial interface instead of a string. Why use bytes and not just send a text string? Using the pack and unpack enforces a structure to the data packets and reduces the amount of data needed to be transmitted compared with a string. Consider a number ‘2048’ sent using the packer function. This is coded as an ‘f’ meaning a float. This is 2 bytes long. Sending ‘2048’ as a string would require 4 bytes, one for each of ‘2’, ‘0’, ‘4’ and ‘8’.
If I encode the string ‘ST 7516.0 50.0 -224.0 144.0 -976.0 EN’ using packer ‘2s5f2s’, the message is 26 bytes. If I send it as a string, it will be 37 bytes. Please see the example code and its output below.

from struct import *
PACKER = ('2s5f2s')
test = 'ST 7516.0 50.0 -224.0 144.0 -976.0 EN'
test2 = (b'ST',7516.0,50.0,-224.0,144.0,-976.0,b'EN')
print('string length: {}'.format(len(test)))
packed_data = pack(PACKER,*test2)
print('packed length: {}'.format(len(packed_data)))
print('unpacked data: {}'.format(unpack(PACKER,packed_data)))

output:

string length: 37
packed length: 26
unpacked data: (b'ST', 7516.0, 50.0, -224.0, 144.0, -976.0, b'EN')

The second reason for using pack and unpack for data packed transmission over sending a stream is that this enforces error checking. If the data is corrupted while reading from the sensor, then an error will be raised during the pack process at the transmitter end. If the data packet is corrupted during transmission, an error will be raised during the unpack process at the receiving end. This can be caught using a try-except clause.

parsing and unpacking python3 serial data containing double backslashes

edit 11th October 2017: The ‘eval’ statements in the code shown below can be replaced with the safer ‘literal_eval’ from the ast class in the standard library. From the python docs: ‘Safely evaluate an expression node or a string containing a Python literal or container display. The string or node provided may only consist of the following Python literal structures: strings, bytes, numbers, tuples, lists, dicts, sets, booleans, and None.’

This can be used for safely evaluating strings containing Python values from untrusted sources without the need to parse the values oneself. It is not capable of evaluating arbitrarily complex expressions, for example involving operators or indexing.

I lost a day of my life figuring out how to parse serial data sent as bytes from the BBC Microbit using micropython. The problem is that the data byte string appears with double backslash characters instead of single backslashes when read in over a serial interface.
Actual data:

b'ST\\x00\\x00\\x00\\xe0\\xeaE\\x00\\x00HB\\x00\\x00`\\xc3\\x00\\x00\\x10C\\x00\\x00t\\xc4EN'

What I wanted as data:

b'ST\x00\x00\x00\xe0\xeaE\x00\x00HB\x00\x00`\xc3\x00\x00\x10C\x00\x00t\xc4EN'

So how to convert from one misformed byte string to the clean one that python 3 would use?
I really went around in circles on this one. In the end I used a kludge. But it works. My life can now move on.
I convert the double slash byte to a string. Then I use the replace method to replace ‘\\’ with ‘\’. Then I use the literal_eval function to recast it as a byte. I am open to suggestions for a cleaner way of doing this!
Here’s some example code I used in a jupyter notebook session. test2 is the misformed byte string received over the serial interface and test3 is the cleaned byte that I can now unpack and extract the data from.

from struct import *
from ast import literal_eval
PACKER = ('2s5f2s')
test2=b'ST\\x00\\x00\\x00\\xe0\\xeaE\\x00\\x00HB\\x00\\x00`\\xc3\\x00\\x00\\x10C\\x00\\x00t\\xc4EN'
test3 = str(test2)
test3 = test3.replace('\\\\', '\\')
print('{}'.format(test3))
test3 = literal_eval(test3)
print(test3)
print(unpack(PACKER,test3))

output:

b'ST\x00\x00\x00\xe0\xeaE\x00\x00HB\x00\x00`\xc3\x00\x00\x10C\x00\x00t\xc4EN'
b'ST\x00\x00\x00\xe0\xeaE\x00\x00HB\x00\x00`\xc3\x00\x00\x10C\x00\x00t\xc4EN'
(b'ST', 7516.0, 50.0, -224.0, 144.0, -976.0, b'EN')

The data was produced from reading the accelerometer on a BBC Microbit board then using struct.pack(PACKER,scan). I am programming the boards using micropython.
The data is packed using the packer format:

PACKER = ('2s5f2s')

The transmitted scan is constructed using:

values = (START, counter, DELTA, x, y, z, END)
scan = struct.pack(packer, *values)

Where values contains a START and END string (‘ST’ and ‘EN’ respectively), a constant called DELTA which represents the time in between samples and the x, y and z readings from the accelerometer. So PACKER means ‘2 characters followed by 5 floats followed by 2 characters’.
I was being obstinate in sending bytes over the serial interface instead of a string. Why use bytes and not just send a text string? Using the pack and unpack enforces a structure to the data packets and reduces the amount of data needed to be transmitted compared with a string. Consider a number ‘2048’ sent using the packer function. This is coded as an ‘f’ meaning a float. This is 2 bytes long. Sending ‘2048’ as a string would require 4 bytes, one for each of ‘2’, ‘0’, ‘4’ and ‘8’.
If I encode the string:

'ST 7516.0 50.0 -224.0 144.0 -976.0 EN'

using packer ‘2s5f2s’, the message is 26 bytes. If I send it as a string, it will be 37 bytes. Please see the example code and its output below.

from struct import *
PACKER = ('2s5f2s')
test = 'ST 7516.0 50.0 -224.0 144.0 -976.0 EN'
test2 = (b'ST',7516.0,50.0,-224.0,144.0,-976.0,b'EN')
print('string length: {}'.format(len(test)))
packed_data = pack(PACKER,*test2)
print('packed length: {}'.format(len(packed_data)))
print('unpacked data: {}'.format(unpack(PACKER,packed_data)))

output:

string length: 37
packed length: 26
unpacked data: (b'ST', 7516.0, 50.0, -224.0, 144.0, -976.0, b'EN')

The second reason for using pack and unpack for data packed transmission over sending a stream is that this enforces error checking. If the data is corrupted while reading from the sensor, then an error will be raised during the pack process at the transmitter end. If the data packet is corrupted during transmission, an error will be raised during the unpack process at the receiving end. This can be caught using a try-except clause.

EWMA filter example using pandas and python

This article gives an example of how to use an exponentially weighted moving average filter to remove noise from a data set using the pandas library in python 3. I am writing this as the syntax for the library function has changed. The syntax I had been using is shown in Connor Johnoson’s well explained example here.
I will give some example code, plot the data sets then explain the code. The pandas documentation for this function is here. Like a lot of pandas documentation it is thorough, but could do with some more worked examples. I hope this article will plug some of that gap.
Here’s the example code:

import matplotlib.pyplot as plt
import pandas as pd
import numpy as np
ewma = pd.Series.ewm

x = np.linspace(0, 2 * np.pi, 100)
y = 2 * np.sin(x) + 0.1 * np.random.normal(x)
df = pd.Series(y)
# take EWMA in both directions then average them
fwd = ewma(df,span=10).mean() # take EWMA in fwd direction
bwd = ewma(df[::-1],span=10).mean() # take EWMA in bwd direction
filtered = np.vstack(( fwd, bwd[::-1] )) # lump fwd and bwd together
filtered = np.mean(filtered, axis=0 ) # average
plt.title('filtered and raw data')
plt.plot(y, color = 'orange')
plt.plot(filtered, color='green')
plt.plot(fwd, color='red')
plt.plot(bwd, color='blue')
plt.xlabel('samples')
plt.ylabel('amplitude')
plt.show()

This produces the following plot. Orange line = noisy data set. Blue line = backwards filtered EWMA data set. Red line = forwards filtered EWMA data set. Green line = sum and average of the two EWMA data sets. This is the final filtered output.

EWMA fiiltered and raw data.

Let’s look at the example code. After importing the libraries I will need in lines 1-5, I create some example data. Line 6 creates 100 x values with values spaced evenly from 0 to 2 * pi. Line 7 creates 100 y-values from these 100 x-values. Each y value = 2*sin(x)+some noise. The noise is generated using the np.random.normal function. This noisy sine function is plotted in line 15 and can be seen as the jagged orange line on the plot.
Forwards and backwards EWMA filtered data sets are created in lines 10 and 11.
Line 10 starts with the first x-sample and the corresponding y-sample and works forwards and creates an EWMA filtered data set called fwd. This is plotted in line 17 as the red line.
Line 11 starts at the opposite end of the data set and works backwards to the first – this is the backwards EWMA filtered set, called bwd. This is plotted in line 18 as the blue line.
These two EWMA filtered data sets are added and averaged in lines 12-13. This data set is called filtered. This data set is plotted in line 16 as the green line.
If you look at the ewma functions in line 10 and 11, there is a parameter called span. This controls the width of the filter. The lag of the backwards EWMA data behind the final averaged filtered output is equal to this value. Similarly the forward EWMA data set has an offset forwards of the noisy data set equal to this value. Increasing the span increases the smoothing and the lag. Increasing the value will also reduce the peaks of the filtered data in relation to the unfiltered data. You need to try out different values.
My present application for this filter is removing jitter from accelerometer data. I have also used this filter to smooth signals from hydrophones.

Using pyzmq to communicate between GUIs and processes

Graphical user interfaces (GUIs) all want to be the main thread. They don’t play well together. Trying to run GUIs built with different libraries concurrently and get them to talk to one another took me a while to figure out. This article shows how I used the pyzmq library to communicate between two graphical user interfaces (GUIs). 

 
I am working on unique hand gesture recognition. One GUI represents a hand position. This is represented by a GUI built with pyqt with a few range sliders. The sliders will be used to represent pitch, roll and speed of motion in the final application. A second GUI represents the gesture recognition interface. For this example it is a simple label box set up in pyqtgraph. I used pyqtgraph as this is the tool kit I am using in my final application for real time data display from an accelerometer mounted on a hand. I based my pyzmq script on the examples here.
 
I played with the publisher subscriber (pubsub) examples. One of the nice things about the pubsub model is that if you send something from the publisher, even if there are no subscribers waiting for the message, nothing blocks or stalls your script. Pubsub is only one way communication, from the publisher to the subscriber. I opted instead to use the pair model. In this pattern, a socket is set up that allows an object at each end to send messages back and forwards.
 
Pyzmq comes with a partial implementation of the Tornado server. This is explained here. So you can set up an eventloop to trigger on poll events using ioloop. If you are already using a GUI, then odds on you have an events handler running in that GUI. Getting this event handling loops to play nicely with the Tornado server led me down the coding rabbit hole. So I opted to use the event handling loop set up by timer = QtCore.QTimer() in pyqtgraph to poll one end of the pyzmq pair socket that I set up. This is not aesthetic, but I can’t see a more reliable method. I am using this QTimer to enable animation of the sensor data that I am using for displaying hand position, so it is already running. Which ever method I use to set up receiving data from the hand posture GUI, at some point I have to decide to look at the data and use it. I thought about using the pyzmq.Queue structure, which is process safe. I could use this to automatically update a list in my sensor display GUI with new posture positions. This won’t be looked at until the QTimer triggers. So I may as well simplify things and look for the updated posture position in the QTimer handling method.
 
Here’s the code I use to generate the rangeslider GUI. This can be downloaded from: github. Most of this is boilerplate to produce the GUI. Lines 102-107 create the pyzmq pair socket. Note the try/except wrapper in lines 97-99 around the socket.send_string. This raises a zmq.error.Again exception if there is nothing to receive the message. Using the try/except wrapper allows the code to continue. The ‘flags=zmq.NOBLOCK’ stops the code from blocking if there is nothing at the other end of the socket to receive the message. This isn’t an issue with the pubsub model; a publisher doesn’t care if there is no subscriber around to receive the message, but the pair pattern will fail without a receiver unless you explicitly tell it not to block.
</div>
<div>

'''
Created on 10 Oct 2016

@author: matthew oppenheim
use pyzmq pair context for communication
'''

from multiprocessing import Process
from PyQt4 import QtGui, QtCore
from qrangeslider import QRangeSlider
import sys
import zmq
from zmq.eventloop import ioloop, zmqstream
from pubsub_zmq import PubZmq, SubZmq

class Example(QtGui.QWidget):

def __init__(self):
app = QtGui.QApplication(sys.argv)
super().__init__()
ioloop.install()
self.port = 5556
self.topic = "1"
self.initUI()
sys.exit(app.exec_())

def initUI(self):
self.range_duration = QRangeSlider()
self.range_duration.show()
self.range_duration.setFixedWidth(300)
self.range_duration.setFixedHeight(36)
self.range_duration.setMin(0)
self.range_duration.setMax(1000)
self.range_duration.setRange(200,800)
self.textbox = QtGui.QLineEdit()
self.set_duration_btn = QtGui.QPushButton("send duration")
self.set_duration_btn.clicked.connect(lambda: self.button_click('duration'))
self.set_duration_btn.setFixedWidth(100)
self.range_pitch = QRangeSlider()
self.range_pitch.show()
self.range_pitch.setFixedWidth(300)
self.range_pitch.setFixedHeight(36)
self.range_pitch.setMin(-80)
self.range_pitch.setMax(80)
self.range_pitch.setRange(-20, 20)
self.set_pitch_btn = QtGui.QPushButton("send pitch")
self.set_pitch_btn.setFixedWidth(100)
self.set_pitch_btn.clicked.connect(lambda: self.button_click('pitch'))
self.range_roll = QRangeSlider()
self.range_roll.show()
self.range_roll.setFixedWidth(300)
self.range_roll.setFixedHeight(36)
self.range_roll.setMin(-80)
self.range_roll.setMax(80)
self.range_roll.setRange(-20, 20)
self.set_roll_btn = QtGui.QPushButton("send roll")
self.set_roll_btn.setFixedWidth(100)
self.set_roll_btn.clicked.connect(lambda: self.button_click('roll'))

hbox_duration = QtGui.QHBoxLayout()
hbox_duration.addStretch(1)
hbox_duration.addWidget(self.range_duration)
hbox_duration.addWidget(self.set_duration_btn)

hbox_pitch = QtGui.QHBoxLayout()
hbox_pitch.addStretch(1)
hbox_pitch.addWidget(self.range_pitch)
hbox_pitch.addWidget(self.set_pitch_btn)

hbox_pitch = QtGui.QHBoxLayout()
hbox_pitch.addStretch(1)
hbox_pitch.addWidget(self.range_pitch)
hbox_pitch.addWidget(self.set_pitch_btn)

hbox_roll = QtGui.QHBoxLayout()
hbox_roll.addStretch(1)
hbox_roll.addWidget(self.range_roll)
hbox_roll.addWidget(self.set_roll_btn)

vbox = QtGui.QVBoxLayout()
vbox.addStretch(1)
vbox.addLayout(hbox_pitch)
vbox.addLayout(hbox_roll)
vbox.addLayout(hbox_duration)
vbox.addWidget(self.textbox)

self.setLayout(vbox)
self.setGeometry(300, 300, 300, 150)
self.setWindowTitle('rangesliders')
self.socket = self.create_socket(self.port)
self.show()

@QtCore.pyqtSlot()
def button_click(self, message):
''' handle button click event '''
self.textbox.setText('sent {}'.format(message))
try:
self.socket.send_string(message, flags=zmq.NOBLOCK)
except zmq.error.Again as e:
print('no receiver for the message: {}'.format(e))

def create_socket(self, port):
''' create a socket using pyzmq with PAIR context '''
context = zmq.Context()
socket = context.socket(zmq.PAIR)
socket.bind("tcp://*:%s" % port)
return socket

if __name__ == '__main__':
ex = Example()

Here’s the simple label box that I use to test out receiving messages:


'''
pyqtgraph layout with a pyzmq pair context
for testing pubsub messaging with pyzmq
Created on 14 Oct 2016
using qt timer and polling instead of the tornado loop in zmq
@author: matthew oppenheim
'''

import pyqtgraph as pg
from pyqtgraph.Qt import QtGui, QtCore
from pubsub_zmq import SubZmq
from multiprocessing import Process
import zmq
import sys
import time

FRAMES_PER_SECOND = 30

class PyqtgraphPair(QtGui.QWidget):
def __init__(self):
super().__init__()
port = '5556'
topic = '1'

QtGui.QWidget.__init__(self)
self.layout = QtGui.QVBoxLayout()
self.setLayout(self.layout)
self.label = QtGui.QLabel("test")
self.set_label("new label")
self.layout.addWidget(self.label)
self.socket = self.create_socket(port)

def create_socket(self, port):
'''
Constructor
'''
context = zmq.Context()
socket = context.socket(zmq.PAIR)
socket.connect('tcp://localhost:%s' % port)
return socket

def process_message(self, message):
''' process the subscriber's message '''
#topic, text = enumerate(a for a in message)
message = message[0].decode()
message = message.split()[1]
print('sub received {}'.format(message))
self.set_label(self.label, 'changed')
#label.setText(message)
if message == 'exit':
ioloop.IOLoop.instance().stop()
print('sub io loop stopped')

def set_label(self, text):
''' set the label to text '''
self.label.setText(text)

def timer_timeout(self):
''' handle the QTimer timeout '''
try:
msg = self.socket.recv(flags=zmq.NOBLOCK).decode()
print('message received {}'.format(msg))
self.set_label(msg)
except zmq.error.Again as e:
return


if __name__ == '__main__':
pg.mkQApp()
win = PyqtgraphPair()
win.show()
win.resize(200,200)
timer = QtCore.QTimer()
timer.timeout.connect(win.timer_timeout)
timer.start(1000/FRAMES_PER_SECOND)
#win.set_label('hello')
if (sys.flags.interactive != 1) or not hasattr(QtCore, 'PYQT_VERSION'):
QtGui.QApplication.instance().exec_()

Polling for a new message takes place in line 61. This has the same try/except wrapper as in the rangeslider example.

python – how to communicate between threads using pydispatcher

The pydispatcher module makes it straight forwards to communicate between different threads in the same process in python.

Why would I want to do this?

I am collecting and processing sensor data from an accelerometer and want to display this real-time. The interface has some controls to save the data and to change the sampling rate of the sensor. Naturally, I want to interact with the user interface without having to wait for the sensor data to be collected and processed. I also want the sensor to be continuously sampled, not having to wait for the real-time display to update.

I run the the graphical user interface (GUI) in one thread and use a separate thread to handle getting data from the sensor. This way the sensor is continuously sampled and the display remains responsive.

I use pydispatcher to send sensor measurements from the sensor thread the display thread. I also use pydispatcher to communicate from the display thread back to the sensor thread to control the rate that the sensor collects data or to stop data collection. So I have two way communication between the threads. I pass numpy arrays from the sensor thread to the display and send text from the display thread to the sensor thread. The text is then interpreted by the sensor thread to alter the sensor sampling rate, or stop sampling. Pydispatcher does not seem to mind what kind of data is sent as messages.

The application that I have described takes up quite a lot of code and is split over several classes. So I will present the code for a simpler example, which shows how to set up and apply pydispatcher and introduces some of the features that makes the library versatile.

Here is an example python 3 script that creates two threads and has them communicate. When the script is executed, as it will have the __name__ as __main__, so lines 46-50 will be the first to execute. A thread that instigates the Alice class is defined and created in lines 47-48 and a separate thread that instigates the Bob class is defined then started in lines 49-50.

In line 26 the alice_thread thread prints out a message ‘Alice is procrastinating’ every second.

In line 43 the bob_thread sends a message to the alice_thread every three seconds using a dispatcher. The alice_thread reacts to this dispatcher message by returning a message of her own to the bob_thread using a separate dispatcher.

If we look at line 15 in the Alice class, a dispatcher listener is set up:

dispatcher.connect(self.alice_dispatcher_receive, signal=BOB_SIGNAL, sender=BOB_SENDER)

This means that when a dispatcher.send statement with the signal BOB_SIGNAL and sender BOB_SENDER is executed anywhere else in the process, the method alice_dispatcher will be triggered so long as an instance of the Alice class has been created. In line 43, the Bob class sets up a dispatcher sender, which is designed to trigger the dispatcher listener in the Alice class described above.

dispatcher.send(message='message from Bob', signal=BOB_SIGNAL, sender=BOB_SENDER)

Having signal and sender names for each dispatcher listener and sender is a little confusing at first. Why do we have to define two identifiers for the dispatcher? Being able to define two identifiers allows us to group dispatchers from the same sender, using the sender identifier. Then we can have the same sender class sending different types of signal, for example data from different sensors, each one with the same sender identifier but each one with different signal identifier. This is verbose, but this verbosity makes for unambiguous easy to maintain code.

Lines 6-9 define the names of the signals and senders for Alice and Bob.

When the alice_thread receives a dispatch from the bob_thread thread, she replies with a dispatch sender of her own (line 21). The corresponding dispatch listener is defined in the Bob class in line 33.

''' demonstrate the pydispatch module '''
from pydispatch import dispatcher
import threading
import time

ALICE_SIGNAL='alice_signal'
ALICE_SENDER='alice_sender'
BOB_SIGNAL='bob_signal'
BOB_SENDER='bob_sender'

class Alice():
''' alice procrastinates and replies to bob'''
def __init__(self):
print('alice instantiated')
dispatcher.connect(self.alice_dispatcher_receive, signal=BOB_SIGNAL, sender=BOB_SENDER)
self.alice()

def alice_dispatcher_receive(self, message):
''' handle dispatcher'''
print('alice has received message: {}'.format(message))
dispatcher.send(message='thankyou from Alice', signal=ALICE_SIGNAL, sender=ALICE_SENDER)

def alice(self):
''' loop and wait '''
while(1):
print('Alice is procrastinating')
time.sleep(1)

class Bob():
''' bob contacts alice periodically '''
def __init__(self):
print('Bob instantiated')
dispatcher.connect(self.bob_dispatcher_receive, signal=ALICE_SIGNAL, sender=ALICE_SENDER)
self.bob()

def bob_dispatcher_receive(self, message):
''' handle dispatcher '''
print('bob has received message: {}'.format(message))

def bob(self):
''' loop and send messages using a dispatcher '''
while(1):
dispatcher.send(message='message from Bob', signal=BOB_SIGNAL, sender=BOB_SENDER)
time.sleep(3)

if __name__ == '__main__':
alice_thread = threading.Thread(target=Alice)
alice_thread.start()
bob_thread = threading.Thread(target=Bob)
bob_thread.start()
Output:
alice instantiated
Alice is procrastinating
Bob instantiated
alice has received message: message from Bob
bob has received message: thankyou from Alice
Alice is procrastinating
Alice is procrastinating
Alice is procrastinating
alice has received message: message from Bob
bob has received message: thankyou from Alice
Alice is procrastinating
Alice is procrastinating
alice has received message: message from Bob
bob has received message: thankyou from Alice
Alice is procrastinating
Alice is procrastinating
Alice is procrastinating
alice has received message: message from Bob
bob has received message: thankyou from Alice

To conclude. There are different ways to communicate between threads in python. I choose pydispatcher as the library allows me to write code that I can understand when I come back to it 6 months later and I don’t have to worry about the type of message that I am passing between the threads.