Concurrent functions running in separate process using pygame and multiprocessing

Question

Suppose we want to drive an autonomous car by predicting image labels from a previous set of images and labels collected (A Machine Learning application). For this task, the car is connected via bluetooth serial (rfcomm) to the Host Computer (A PC with *NIX) and the images are streamed directly from an Android phone using IP Webcam, meanwhile, the PC is running a program that links this two functions, displaying the captured images in a drawing environment created by pygame, and sending the instructions back to the car using serial.

At the moment, I've tried to implement those processes using the multiprocessing module, the seemed to work, but when I execute the client, the drawing function (if __name__ == '__main__') works after the getKeyPress() function ends.

The question is: It is possible to parallelize or synchronize the drawing fuinction enclosed within the if __name__ == '__main__' with the process declared in getKyPress(), such that the program works in two independent processes?

Here's the implemented code so far:

import urllib
import time
import os
import sys
import serial
import signal
import multiprocessing 
import numpy as np
import scipy
import scipy.io as sio
import matplotlib.image as mpimg
from pygame.locals import *


PORT = '/dev/rfcomm0'
SPEED = 115200

ser = serial.Serial(PORT)

status = False
move = None
targets = []
inputs = []
tic = False



def getKeyPress():
    import pygame
    pygame.init()
    global targets
    global status
    while not status:
          pygame.event.pump()
          keys = pygame.key.get_pressed()
          targets, status = processOutputs(targets, keys)
    targets = np.array(targets)
    targets = flattenMatrix(targets)
    sio.savemat('targets.mat', {'targets':targets})       


def rgb2gray(rgb):
    r, g, b = np.rollaxis(rgb[...,:3], axis = -1)
    return 0.299 * r + 0.587 * g + 0.114 * b

def processImages(inputX, inputs):
    inputX = flattenMatrix(inputX)
    if len(inputs) == 0:
       inputs = inputX
    elif inputs.shape[1] >= 1:
       inputs = np.hstack((inputs, inputX))
    return inputs

def flattenMatrix(mat):
    mat = mat.flatten(1)
    mat = mat.reshape((len(mat), 1))
    return mat

def send_command(val):
    connection = serial.Serial( PORT,
                                SPEED,
                                timeout=0,
                                stopbits=serial.STOPBITS_TWO
                                )
    connection.write(val)
    connection.close()

def processOutputs(targets, keys):
    global move
    global status
    global tic
    status = False
    keypress = ['K_p', 'K_UP', 'K_LEFT', 'K_DOWN', 'K_RIGHT']
    labels = [1, 2, 3, 4, 5]
    commands = ['p', 'w', 'r', 'j', 's']
    text = ['S', 'Up', 'Left', 'Down', 'Right']
    if keys[K_q]:
       status = True
       return targets, status            
    else:
       for i, j, k, g in zip(keypress, labels, commands, text):
           cmd = compile('cond = keys['+i+']', '<string>', 'exec')
           exec cmd
           if cond:
              move = g
              targets.append(j)
              send_command(k)
              break
    send_command('p')
    return targets, status


targetProcess = multiprocessing.Process(target=getKeyPress)
targetProcess.daemon = True
targetProcess.start()

if __name__ == '__main__': 
   import pygame
   pygame.init()
   w = 288
   h = 352
   size=(w,h)
   screen = pygame.display.set_mode(size)
   c = pygame.time.Clock() # create a clock object for timing
   pygame.display.set_caption('Driver')
   ubuntu = pygame.font.match_font('Ubuntu')
   font = pygame.font.Font(ubuntu, 13)
   inputs = []
   try:
     while not status:
           urllib.urlretrieve("http://192.168.0.10:8080/shot.jpg", "input.jpg")
           try:
             inputX = mpimg.imread('input.jpg')
           except IOError:
             status = True
           inputX = rgb2gray(inputX)/255
           out = inputX.copy()
           out = scipy.misc.imresize(out, (352, 288), interp='bicubic', mode=None)
           scipy.misc.imsave('input.png', out)
           inputs = processImages(inputX, inputs)
           print inputs.shape[1]
           img=pygame.image.load('input.png')
           screen.blit(img,(0,0))
           pygame.display.flip() 
           c.tick(1)
           if move != None:
              text = font.render(move, False, (255, 128, 255), (0, 0, 0))
              textRect = text.get_rect()
              textRect.centerx = 20 #screen.get_rect().centerx
              textRect.centery = 20 #screen.get_rect().centery
              screen.blit(text, textRect)
              pygame.display.update()
           if status:
              targetProcess.join()
              sio.savemat('inputs.mat', {'inputs':inputs})
   except KeyboardInterrupt:
     targetProcess.join()
     sio.savemat('inputs.mat', {'inputs':inputs})

   targetProcess.join()
   sio.savemat('inputs.mat', {'inputs':inputs})

Thanks in advance.

It sounds like you might want [threading, not multiprocessing?](http://stackoverflow.com/a/3046201/341744) It appears you are trying to load the image in a variable in one process without the other one having knowledge of it. Since you're writing to a `png`, you can open that in the rendering process/thread. — ninMonkey, Apr 04 '13 at 23:39
You seem to initialize pygame first and fork much later. I suspect that forking might make both copies inherit the shared state of SDL, as a threads would. I'd try to initialize pygame separately in both target functions, and not import it at all at the top level. — 9000, Apr 05 '13 at 03:48

score 1 · Answer 1 · answered Apr 10 '13 at 15:04

I would personally suggest writing this without using the multiprocessing module: it uses fork() which has unspecified effects with most complex libraries, like in this case pygame.

You should try to write this as two completely separate programs. It forces you to think about what data needs to go from one to the other, which is both a bad and a good thing (as it may clarify things). You can use some inter-process communication facility, like the stdin/stdout pipe; e.g. in one program (the "main" one) you start the other as a sub-process like this:

popen = subprocess.Popen([sys.executable, '-u', 'my_subproc.py'],
                         stdin=subprocess.PIPE, stdout=subprocess.PIPE)

(The -u is for unbuffered.)

Then read/write the data to popen.stdin/popen.stdout in the parent process, and to sys.stdin/sys.stdout in the subprocess. The simplest example would be if the two processes only need a synchronization signal, e.g. the parent process waits in a loop for the subprocess to say "next please". To do this the subprocess does print 'next please', and the parent process does popen.stdin.readline(). (The print goes to sys.stdin in the subprocess.)

Unrelated small note:

keypress = ['K_p', ...]
...
cmd = compile('cond = keys['+i+']', '<string>', 'exec')
exec cmd
if cond:

This looks like very heavy code to just do:

keypress = [K_p, ...]     # not strings, directly the values
...
if keys[i]:

score 0 · Accepted Answer · answered Apr 12 '13 at 10:27

My suggestion is to use separate threads.

#At the beginning
import threading

#Instead of def getKeyPress()
class getKeyPress(threading.Thread):
    def run(self):
        import pygame
        pygame.init()
        global targets
        global status
        while not status:
               pygame.event.pump()
               keys = pygame.key.get_pressed()
               targets, status = processOutputs(targets, keys)
        targets = np.array(targets)
        targets = flattenMatrix(targets)
        sio.savemat('targets.mat', {'targets':targets}) 

#Instead of 
#targetProcess = multiprocessing.Process(target=getKeyPress)
#targetProcess.daemon = True
#targetProcess.start()  
gkp = getKeyPress()
gkp.start()

An alternative would be creating two different scripts and using sockets to handle the inter-process communication.

Concurrent functions running in separate process using pygame and multiprocessing

2 Answers2