使用 PyGame 时非阻塞串行读取线

Question

我正在开发一个用 PyGame 编写的心率监测器，使用 Arduino 作为输入，这个游戏的想法是让你通过放松练习来尝试和控制你的心率，而不管是什么屏幕。

我希望能够在游戏本身中运行像video/mouse/keybord按钮捕获等的东西，同时在左上角显示心率并从arduino更新它它改变了。

arduino读取一个心率监测器，然后发布一个JSON字符串，格式如下：

{'heart_rate': 65,'state': 'running'}

"State" 可以是 'intialising'、'running'、'failed' 或 'stopped'.

之一

虽然我对 Python 非常熟悉，但我已经从 http://www.akeric.com/blog/?p=1237 中获取了代码以开始使用 PyGame，因为我以前没有在这里冒险过。

我遇到的问题是，当我尝试从串行端口读取时，它会锁定游戏。

我已经阅读了关于线程的文章，我认为我已经正确地实现了它，但是以下代码仍然阻塞：

"""
default.py
www.akeric.com - 2010-09-07
Default, base, initial setup for a pygame program.
In this case, black background, white circle follows mouse position, and
framerate is shown in the title-bar.
"""

#-------------------------------------------------------------------------------
# Imports & Inits
import sys
import serial
import json
import threading

import pygame
from pygame.locals import *
pygame.init()
ser = serial.Serial('/dev/ttyUSB0', 9600, timeout=0)


#-------------------------------------------------------------------------------
# Constants
VERSION = '1.0'
WIDTH = 800
HEIGHT = 600
FRAMERATE = 60
CURRENT_MESSAGE = {'heart_rate': 0}

#-------------------------------------------------------------------------------
# Screen Setup
screen = pygame.display.set_mode((WIDTH, HEIGHT))
bgCol = Color('black')
clock = pygame.time.Clock()
moduleName = __file__.split('\')[-1]

#-------------------------------------------------------------------------------
# Define helper functions, classes, etc...
def text_objects(text, font):
    textSurface = font.render(text, True, (255,255,255))
    return textSurface, textSurface.get_rect()

def message_display(text,x_pos=(WIDTH/2),y_pos=(HEIGHT/2)):
    largeText = pygame.font.Font('freesansbold.ttf',25)
    TextSurf, TextRect = text_objects(text, largeText)
    TextRect.center = (x_pos,y_pos)
    screen.blit(TextSurf, TextRect)

def spam():
    pos = pygame.mouse.get_pos()
    pygame.draw.circle(screen, Color('white'), pos, 32)

def handle_data(data):
    CURRENT_MESSAGE = json.loads(data)
    message_display("HR: %s" % CURRENT_MESSAGE['heart_rate'],50,20)

def read_from_port(ser):
        while True:
           reading = ser.readline().decode()
           if len(reading) > 0:
               print "Data Recieved"
               handle_data(reading)



#-------------------------------------------------------------------------------
# Main Program
def main():
    print "Running Python version: %s"%sys.version
    print "Running PyGame version: %s"%pygame.ver
    print "Running %s version: %s"%(moduleName, VERSION)
    looping = True

    # Main Loop-----------------------
    while looping:
        # Maintain our framerate, set caption, clear background:
        clock.tick(FRAMERATE)
        pygame.display.set_caption("%s - FPS: %.2f" %(moduleName,clock.get_fps()) )
        screen.fill(bgCol)
        spam()


        # Update our display:---------
        pygame.display.flip()

#-------------------------------------------------------------------------------
# Execution from shell\icon:
if __name__ == "__main__":
    # Make running from IDE work better:
    thread = threading.Thread(target=read_from_port, args=(ser,))
    thread.start()
    sys.exit(main())

任何人都可以帮助我了解我哪里出错了吗？

Answer 1

我对 Python 知之甚少，但我会测试看看是否有任何字符等待读取，然后只读取该数字而不是使用 readline。我想你想使用 in_waiting 和 read(size).

Readline 将阻塞，直到收到回车 return。

所以，我想是这样的：

def read_from_port(ser):
    while True:           
       if in_waiting > 0:
           reading = read(in_waiting)
           print "Data Recieved"
           handle_data(reading)

现在您需要连接并解析分隔符处的字符串，以确保已收到整个字符串。

Answer 2

我修好了！ :)

#!/usr/bin/env python

import pygame
from threading import Thread
import serial
import json
from pygame.color import Color
from pygame.locals import *

CURHR = 0

ser = serial.Serial('/dev/pts/3', 9600, timeout=0)
def worker():

   global CURHR
   while True:
     msg = ser.readline()
     if len(msg) > 0:
       print "Message Received: %s" % msg
       current_message = json.loads(msg)
       if current_message["state"] == "running":
         CURHR=current_message['heart_rate']

t = Thread(target=worker)
t.daemon = True
t.start()

pygame.init()

#-------------------------------------------------------------------------------
# Constants
VERSION = '1.0'
WIDTH = 800
HEIGHT = 600
FRAMERATE = 60

#-------------------------------------------------------------------------------
# Define helper functions, classes, etc...
def text_objects(text, font):
    textSurface = font.render(text, True, (255,255,255))
    return textSurface, textSurface.get_rect()

def message_display(text,x_pos=(WIDTH/2),y_pos=(HEIGHT/2)):
    largeText = pygame.font.Font('freesansbold.ttf',25)
    TextSurf, TextRect = text_objects(text, largeText)
    TextRect.center = (x_pos,y_pos)
    screen.blit(TextSurf, TextRect)

def spam():
    pos = pygame.mouse.get_pos()
    pygame.draw.circle(screen, Color('white'), pos, 32)

screen = pygame.display.set_mode((WIDTH,HEIGHT))
clock  = pygame.time.Clock()
font   = pygame.font.SysFont("consolas", 25, True)
count  = 0
pygame.display.set_caption("Test")

done = False
while not done:
    screen.fill(Color('black'))
    for event in pygame.event.get(): # User did something
        if event.type == pygame.QUIT: # If user clicked close
            done = True
        elif event.type == KEYDOWN:
            if event.key == K_ESCAPE:
                  done = True

    spam()
    hr_string = "HR: %s"  % CURHR
    hr_text = font.render(hr_string, True, Color('red'))
    screen.blit(hr_text, [25,10])
    clock.tick(20)
    pygame.display.flip()

诀窍是设置一个全局，然后让线程设置该全局的值。

然后在屏幕 "blitted" 时渲染全局值（是这样吗？！）并且实时更新值而不影响光标的移动。

使用 PyGame 时非阻塞串行读取线

Non blocking serial readline when using PyGame

python

linux

pygame

arduino

pyserial