Raspberry PI 和 Arduino 串行通信与 Python

Question

我在 Raspberry Pi（Python 脚本）和 Arduino nano/uno 之间的串行通信有一些问题。我有两个设备都通过 USB 端口连接，当我直接从 arduino IDE 串行监视器发送命令时，arduino sketch 总是正确响应：

<Arduino is ready>
<MSG:GDI,RESPONSE:EMD-1707-1993,Time:7>
<MSG:GDI,RESPONSE:EMD-1707-1993,Time:23>
<MSG:GDI,RESPONSE:EMD-1707-1993,Time:26>
<MSG:GDI,RESPONSE:EMD-1707-1993,Time:30>
<MSG:GDI,RESPONSE:EMD-1707-1993,Time:34>
<MSG:GDI,RESPONSE:EMD-1707-1993,Time:38>

但是，当我运行 python 脚本和他发送相同的命令时，草图以随机方式响应。

树莓派终端：

pi@raspberrypi:~/test/raspberry $ python test.py 
Sending GDI command to device...
<GDI>
Traceback (most recent call last):
  File "test.py", line 47, in <module>
    response = read_from_device(serial_connection)
  File "test.py", line 15, in read_from_device
    while ord(current_char) != MSG_START_CHAR: 
TypeError: ord() expected a character, but string of length 0 found

Arduino 串行监视器：

<Arduino is ready>
MGIRSPi35

小米密码为：

Arduino 草图：

#include <EEPROM.h>

#define DEVICE_BAUD_RATE        9600

#define EEPROM_SIZE             1024

#define ID_PREFIX               "EMD-"

#define MSG_START_CHAR          '<'
#define MSG_END_CHAR            '>'

#define MSG_GET_DEVICE_ID       "GDI"

const byte buffSize = 40;
char inputBuffer[buffSize];
byte bytesRecvd = 0;
boolean readInProgress = false;
boolean newMsg = false;

char cmd[buffSize] = {0};
int pin = 0;
int value = 0;

unsigned long curMillis;

void setup() {
  if(getDeviceId() == "") {
    setDeviceId();
  }

  Serial.begin(DEVICE_BAUD_RATE);

  while(!Serial) {
    ;
  }

  Serial.println("<Arduino is ready>");
}

void loop() {
  curMillis = millis();
  readMsg();
  processCommand();
}

void readMsg() {
  if(Serial.available() > 0) {
    char x = Serial.read();

    if(x == MSG_END_CHAR) {
      readInProgress = false;
      newMsg = true;
      inputBuffer[bytesRecvd] = 0;
      strcpy(cmd, inputBuffer);
    }

    if(readInProgress) {
      inputBuffer[bytesRecvd] = x;
      bytesRecvd++;

      if(bytesRecvd == buffSize) {
        bytesRecvd = buffSize - 1;
      }
    }

    if(x == MSG_START_CHAR) { 
      bytesRecvd = 0; 
      readInProgress = true;
    }
  }
}

void processCommand() {
  if(strcmp(cmd, MSG_GET_DEVICE_ID) == 0) {
    sendMsg(getDeviceId());
  } else {
    sendMsg("Command Not Found");
  }
}

void sendMsg(String response) {
  if(newMsg) {
    newMsg = false;
    Serial.print("<MSG:");
    Serial.print(cmd);
    Serial.print(",RESPONSE:");
    Serial.print(response);
    Serial.print(",Time:");
    Serial.print(curMillis >> 9);
    Serial.println(">");
  }
}

String getDeviceId() {
  String id = "";

  for(int i=0; i<EEPROM_SIZE; i++) {
    int value = EEPROM.read(i);

    if(value == 0xFF) {
      return id;
    }

    id += char(value);
  }

  return id;
}

void setDeviceId() {
  randomSeed(analogRead(0));

  String id = ID_PREFIX + String(random(1000, 10000)) + "-" + String(random(1000, 10000));

  for(int i=0; i<EEPROM_SIZE; i++) {
    EEPROM.write(i, i<id.length() ? id.charAt(i) : 0xFF);
  }
}

Python 脚本：

#!/usr/bin/python
import os
import serial

MSG_START_CHAR              = '<'
MSG_END_CHAR                = '>'

MSG_GET_DEVICE_ID           = 'GDI'


def read_from_device(serial_connection):  
    response = ""
    current_char = "z"

    while ord(current_char) != MSG_START_CHAR: 
        current_char = serial_connection.read()

    while ord(current_char) != MSG_START_CHAR:
        if ord(current_char) != MSG_START_CHAR:
            response = response + current_char 

        current_char = serial_connection.read()

    return(response)

def write_to_device(serial_connection, msg):
    cmd = MSG_START_CHAR + msg + MSG_END_CHAR

    print(cmd)

    serial_connection.write(cmd)

with serial.Serial('/dev/ttyACM0', 9600, timeout=10) as serial_connection:
    waiting_for_reply = False

    if waiting_for_reply == False:
        print('Sending {0} command to device...'.format(MSG_GET_DEVICE_ID))

        write_to_device(serial_connection, MSG_GET_DEVICE_ID)

        waiting_for_reply = True

    if waiting_for_reply == True:
        while serial_connection.inWaiting() == 0:
            pass

        response = read_from_device(serial_connection)

        print('Reply Received: {0}'.format(response))

        waiting_for_reply = False

    serial_connection.close()

Answer 1

问题是您要将 int 与字符串进行比较。在函数 read_from_device 中，这一行：

    while ord(current_char) != MSG_START_CHAR:

current_char 是一个字符串。
ord(current_char) 是一个整数（实际上是单字符字符串的 ASCII 值）。
MSG_START_CHAR 是一个字符串。

因此比较是在 int 和 string 之间进行的。那将始终为 False，因此 while 循环永远不会退出。最终没有更多来自 Arduino 的字符，此时您将对空字符串执行 ord()。这是不允许的，所以你会在那个时候得到回溯。

您根本不需要 ord 函数。在 Python 中，您只需直接使用字符串。

试试这个：

def read_from_device(serial_connection):  
    while serial_connection.read() != MSG_START_CHAR:
        pass

    response = ""
    while True:
        current_char = serial_connection.read()
        if current_char == MSG_END_CHAR:
            break
        response = response + current_char 
    return response

有一件事我在这里不知道：Arduino 正在发送单字节字符。在 Python3 中（如果您正在使用它），所有字符都是 unicode。我不知道 Serial 库将如何处理。

Raspberry PI 和 Arduino 串行通信与 Python

Raspberry PI and Arduino Serial communication with Python

python

arduino

raspberry-pi

pyserial

serial-communication