Gas station without pumps

2012 June 12

Improved super pulley code

In Homemade super pulley, I described the “super pulley” that my son built and that I wrote some Arduino code for.

After our preliminary experiments using the super pulley to record the initial parts of soda-bottle rocket flights, my son was dissatisfied with the minimal code that I had written.  It was difficult to do the cut and paste from the Arduino serial monitor to save the data, and entering the necessary metadata was bit difficult with the bright sun washing out the laptop screen.

He decided it would be worth his time to improve the code.  He made two major changes to the code: first, he implemented a Python program on the laptop to capture the Arduino output and save it (with metadata) to a file.  Second, he rewrote the Arduino code to be interrupt driven and to have a queue of output, so that we could record far more than 400 points, while still being able to handle bursts of very rapid data.

These programs got him into three new coding concepts for him: circular buffers for queues, interrupts, and multi-threading.  He researched and chose these techniques, then implemented the program without my assistance, although I did go over it with him at the end, editing for style and clarity of documentation.  In fact, I tried to talk him out of using multi-threading—I favor lighter-weight techniques like the Unix select() or poll() functions, but portability won for him.  There is one place where portability was lost, though—he couldn’t find documentation on how to look for USB devices from Python on a Windows machine, and just used glob('/dev/tty.usb*') to look for the Arduino, which should work for Macs, Linux, and other Unix clones. On Windows, it looks like you have to open one of the COM ports, but I’m not sure how you figure out which one.

The only non-standard Python library used is PySerial, for talking to the USB port.

He decided to use curses rather than PyGUI or a plain terminal window, because he wanted to do some display of the number of samples as they were read, but did not want the hassle of setting up a full GUI.  (He has used both curses and PyGUI previously in science fair projects, and used PyGUI for the underwater ROV control, where curses would not have been adequate.)

Here is the revised Arduino code:

// Smart Pulley Rocket Timer
// By Abe Karplus and Kevin Karplus
// To use, connect the output from the photogate of a smart pulley to
//     digital pin 2 on the Arduino.
// On initialization, the Arduino will print "Arduino ready." to serial.
// For each tick of the photogate, the Arduino prints three tab-separated
//     values on a line. The first is the time since the start, in
//     microseconds. The second is the amount of string unwound, in meters.
//     The third is the instantaneous speed of unwinding, in meters per second.
// The ticks are buffered so that they may be recorded faster than the
//     serial can output them. The buffer holds 400 ticks, maximum.
// Serial baudrate is 115200.
// The time and distance may be reset by sending the character 'R'.
// License: CC-BY-NC

#define BUFFER_SIZE (400) // size of tick buffer (limited by RAM)
#define TICKS_PER_REV (6) // number of holes in smart pulley
#define DIAMETER (0.044) // diameter of pulley sheave in meters

const float meter_per_tick = 3.14159265358979 * DIAMETER  / TICKS_PER_REV;
    // meters of string unwound per tick

// queue of times implemented as circular buffer
volatile unsigned long times[BUFFER_SIZE];
volatile unsigned int read_head  = 0; // position in buffer for removal
volatile unsigned int write_head = 0; // position in buffer for insertion
volatile bool wrote_last = 0;   // what was most recent operation on queue?
                                // 0 for read, 1 for write

// Next item to be read is at times[read_head].
// Next empty space to write into is at times[write_head].
// Number of entries in buffer is (write_head - read_head) mod BUFFER_SIZE.
//      Buffer is full if read_head==write_head and wrote_last.
//      Buffer is empty if  read_head==write_head and not wrote_last.

unsigned long count;        // how many ticks have been seen since reset
unsigned long old_time;     // if count>0, what was the time of the previous tick?
unsigned long first_time;   // what was the time for the first tick since reset?
// All times are in microseconds.

// print_time() prints one output line,
//      removing the corresponding time from the queue
//      and updating count, old_time, (and first_time when needed).
// It does nothing if the queue is empty.
void print_time(void)
    noInterrupts();     // don't add to the queue while reading from it
    if (read_head == write_head && !wrote_last)
    {   // buffer empty
    unsigned long datum = times[read_head++];   // time taken from queue.
    if (read_head>=BUFFER_SIZE)
    {   read_head = 0; // wrap around
    wrote_last = 0;
    interrupts();       // finished getting the data, re-enable adding

    if (count==0)
    {   // On first call, do no output, just initialize the global times.
        old_time = first_time = datum;
    Serial.print(datum - first_time); // time since start (usec)
    Serial.print(count++ * meter_per_tick, 3); // distance unwound (m)
    Serial.println(meter_per_tick*1e6 / (datum - old_time), 3); // speed (m/s)
    old_time = datum;

// get_time() is the interrupt handler
// It adds the time returned by micros() to the queue,
// unless the queue is full, in which case nothing is done.
void get_time(void)
    // record time immediately to avoid
    // fluctuation due to variable amount of code executed
    unsigned long now = micros();

    if (read_head == write_head && wrote_last)
    {   // buffer full
    times[write_head++] = now;
    if (write_head>=BUFFER_SIZE)
    {   // wrap_around
    wrote_last = 1;

// do_reset() empties the queue and sents the count to 0
void do_reset(void)
    noInterrupts();     // don't take interrupts in the middle of resetting
    read_head = write_head = 0;
    wrote_last = 0;
    count = 0;
    // let the user know that the Arduino is ready
    Serial.print(F("\nArduino ready.\n"));

// interrupt handler for data becoming available on the serial input
// Reset if an R is received
void serialEvent(void)
    if ( == 'R')

void setup(void)
    Serial.begin(115200);       // use the fastest serial connection available
    // now start looking for ticks on pin 2
    attachInterrupt(0, get_time, FALLING); // pin 2 interrupt

void loop(void)
{   // keep trying to print out what is in the queue.

And here is the Python code:

 #  Smart Pulley Rocket Timer
 #  By Abe Karplus
 #  12 June 2012
 #  Interfaces with the Arduino program rockt_timer.ino for water rocket
 #   timing using a smart pulley. To use, upload rocket_timer.ino to the
 #   Arduino and run `python` on the command line.
 #  This program uses the PySerial library from
 #   in addition to the standard libraries glob, curses, threading, and time.
 #  To find the Arduino, this program searches in the /dev directory;
 #   this will need to be changed for Windows systems.
 #  If supported by the terminal (such as xterm-256color), uses colors.
 #  License: CC-BY-NC

from serial import Serial, SerialException
from glob import glob
import curses
from threading import Thread, Event
from time import sleep

def cursor_vis(n):
    """Change the visibility of the cursor (same as curs_set),
    but ignore errors.
    except curses.error:

def send_reset(ard):
    """Send the Arduino a reset command character."""
    # discard any Arduino output lines that are waiting to be read


def wait_for_ready(ard):
    """Discard input until a line with 'Arduino ready.' appears.
    Needed to avoid leftover trash from Arduino bootloader.
    for line in ard:
        if 'Arduino ready.' in line:

def prepare_arduino():
    """Create a serial connection to an Arduino
    (assuming it is the unique device with /dev/tty.usb*)
    and return the Serial file-like object.
    # find Arduino port
    ports = []
    while len(ports) != 1:
        ports = glob('/dev/tty.usb*')
    port = ports[0]

    # create a Serial connection
    while True:
            arduino = Serial(port=port, baudrate=115200)
        except SerialException as e:
            # occurs upon physically connecting USB
            # so not really a problem.

    return arduino

def add_readings(scr, ard, rds, cond):
    """Infinite loop run in a daemon thread.
    When Event cond is set,
        Read data from the arduino Serial object ard.
        Store the data in list rds.
        Display len(rds) on the curses window scr.
    while True:
        line = next(ard).strip()
        if 'Arduino ready.' in line or not line:
            # Arduino has been reset
            rds[:] = []    # clear list in-place
            scr.move(3, 22)

        # display running count of reads
        scr.addstr(3, 22, str(len(rds)), curses.color_pair(4))

def prepare_screen(scr):
    """Initialize curses window scr.
    Set up color pairs and display initial message.
        curses.init_pair(1, 0, 15) # Black on White
        curses.init_pair(2, 0, 10) # Black on Green
        curses.init_pair(3, 0, 11) # Black on Yellow
        curses.init_pair(4, 0,  9) # Black on Salmon
        curses.init_pair(5, 0, 14) # Black on Cyan
    except curses.error:
        for n in range(1, 6):
            curses.init_pair(n, curses.COLOR_BLACK, curses.COLOR_WHITE)

    # use black on white text by default
    scr.bkgdset(ord(' '), curses.color_pair(1))

    cursor_vis(0)   # hide cursor

    # setup permanent header
    scr.addstr(1, 1, 'Welcome to the Rocket Timer smart pulley interface.', curses.A_BOLD)

    # setup message line
    scr.addstr(3, 1, 'Waiting for the Arduino to connect.', curses.color_pair(3))


def save_readings(scr, rds, d=dict(volume='1.3', pressure='3', water='0')):
    """Request metadata and file name, and save rds to file.
    Clear rds to become an empty list.
    d is a dictionary of default values for file metadata.
    # clear message and prompts
    scr.move(3, 1)

    # setup message
    scr.addstr(3, 1, 'Saving reads.', curses.color_pair(5))
    cursor_vis(1)    # make cursor visible
    curses.echo()    # let user see what they type

    scr.addstr(4, 1, 'Save to file: ')
    fname = scr.getstr().strip()    # request file name
    scr.move(4, 1)

    scr.addstr(4, 1, 'Bottle volume in liters (default {}): '.format(d['volume']))
    # use default if blank, or changes default
    d['volume'] = scr.getstr().strip() or d['volume']
    scr.move(4, 1)

    scr.addstr(4, 1, 'Gauge pressure in bar (default {}): '.format(d['pressure']))
    # use default if blank, or changes default
    d['pressure'] = scr.getstr().strip() or d['pressure']
    scr.move(4, 1)

    scr.addstr(4, 1, 'Water mass in grams (default {}): '.format(d['water']))
    # use default if blank, or changes default
    d['water'] = scr.getstr().strip() or d['water']
    scr.move(4, 1)

    # return curses to display only (no character echo and hidden cursor)

    header = '\n'.join(l.strip() for l in """\
    # {fname}
    # Bottle volume: {d[volume]} liters
    # Gauge pressure: {d[pressure]} bar
    # Water mass: {d[water]} grams
    # Each line represents one tick of the photogate.
    # `time` is the time in microseconds since the beginning.
    # `distance` is the amount of string unwound, in meters.
    # `speed` is the instantaneous speed in meters per second of the unwinding.
    with open(fname, 'w') as f:
        for ln in rds:
    rds[:] = []    # clear list in-place

def do_interface(scr, ard, rds, cond):
    """Parse user interface commands, using a state machine.
        scr is curses window
        ard is Arduino Serial object
        rds is list of reads
        cond is Event to control the add_readings daemon
    # interpretations of keys typed in response to prompts:
    transitions_from_ready = {' ':'recording', 'q':'quitting', 'r':'reset'}
    transitions_from_save_query = {'y':'save readings', 'n':'reset'}

    state = 'reset'
    while True:
        scr.move(2, 0)
        if state == 'reset':
            scr.addstr(3, 1, 'Waiting for Arduino to reset.', curses.color_pair(3))
            state = 'ready to record'
        elif state == 'ready to record':
            scr.addstr(3, 1, 'Ready to record.', curses.color_pair(2))
            scr.addstr(5, 2, 'Space to record, q to quit, r to reset:')
            key = chr(scr.getch()).lower()
            state = transitions_from_ready.get(key,state)
        elif state == 'recording':
            scr.addstr(3, 1, 'Recording. Readings: 0', curses.color_pair(4))
            scr.addstr(5, 2, 'Space to finish.')
            cond.set()     # turn on recording in daemon thread
            while chr(scr.getch()) != ' ':
            cond.clear()    # turn off recording
            state = 'save query'
        elif state == 'save query':
            scr.addstr(3, 1, 'Save {} readings? (y/n)'.format(len(rds)), curses.color_pair(5))
            key = chr(scr.getch()).lower()
            state = transitions_from_save_query.get(key,state)
        elif state == 'save readings':
            save_readings(scr, rds)
            state = 'reset'
        elif state == 'quitting':
            scr.addstr(3, 1, 'Quitting.', curses.color_pair(4))
            raise ValueError('Invalid state {}'.format(state))

def main(scr):
    """Set up screen and threads and start user interface.

    arduino = prepare_arduino()

    readings = []

    cond = Event()
    reader_thread = Thread(target=add_readings, name='reader', args=(scr, arduino, readings, cond))
    reader_thread.daemon = True    # will not stop main from exiting

    do_interface(scr, arduino, readings, cond)

print 'Done.'

The code here is released under Creative Commons license CC-BY-NC:


  1. […] them a simple recording program (perhaps I can get my son to write them a versatile one, based on the one he wrote for our super pulley recording).  The students could sell the Arduino used if they weren’t interested in doing anything […]

    Pingback by Changing teaching plans « Gas station without pumps — 2012 June 14 @ 00:40 | Reply

  2. […] is a low enough frequency (say around 1kHz), then the program could be almost identical to the interrupt-driven data logger that my son wrote for the homemade super pulley, but interpreting the interrupt times […]

    Pingback by More on electronics course design « Gas station without pumps — 2012 June 16 @ 09:57 | Reply

  3. […] tried using the Arduino (with a slight modification of the Superpulley code my son wrote) to time the lower […]

    Pingback by Building a function generator kit « Gas station without pumps — 2012 July 7 @ 22:35 | Reply

  4. […] use the Arduino memory as a queue and send the time stamps to a Python program on the laptop (see Improved super pulley code).  That program introduced him to three new concepts: circular buffers for queues, interrupts, and […]

    Pingback by Data logging software for circuits course working « Gas station without pumps — 2012 December 31 @ 16:27 | Reply

  5. […] water rockets, which I have not done since my son was taking home-school physics and we wrote the timing program for measuring the ascent of the rockets that later turned into PteroDAQ to go along with the homemade Lego […]

    Pingback by Soda bottle rockets used again | Gas station without pumps — 2016 May 12 @ 20:48 | Reply

RSS feed for comments on this post. TrackBack URI

Leave a Reply

Fill in your details below or click an icon to log in: Logo

You are commenting using your account. Log Out /  Change )

Google photo

You are commenting using your Google account. Log Out /  Change )

Twitter picture

You are commenting using your Twitter account. Log Out /  Change )

Facebook photo

You are commenting using your Facebook account. Log Out /  Change )

Connecting to %s

This site uses Akismet to reduce spam. Learn how your comment data is processed.

%d bloggers like this: