Skip to main content
 首页 » 编程设计

python之尝试通过蓝牙将串行数据从 arduino 传感器发送到 python

2024年11月24日43JustinYoung

***Python code:*** 
import serial 
import pandas as pd 
import time 
import re 
import xlrd 
from msvcrt import getch 
import numpy as np 
i = 0 
x = 0 
y = 0 
 
df = pd.read_excel(r'C:\Users\lynchfamily\Desktop\mlglovesdata.xls') 
# Read COM9 
# Read from COM10 as well 
# Readline() only works with a timeout (IMPORTANT) 
serHC = serial.Serial('COM9', 115200,timeout=.250,parity=serial.PARITY_NONE,rtscts=1) # This is the JY 
serRN = serial.Serial('COM10', 115200,timeout=.250,parity=serial.PARITY_NONE,rtscts=1) # This is the silvermate 
 
 
def serialin(): 
    # Sensor lists 
    sensor_names = list() 
    sensor_values = list() 
    global i 
    # Read a certain amount of bytes from serial and then continue 
    # Regular expressions for finding the proper data 
    while i < 6: 
        # print(i) for debugging 
        global serHC 
        global serRN 
        #searchObj = re.search(r'(A\d?\d*)?(\d*)?',serHC.read(4).decode(),re.I) 
        #searchObjRN = re.search(r'(A\d?\d*)?(\d*)?',serRN.read(4).decode(),re.I) 
        # Serial data stops while in loop 
        # The if statements keep the false values out of the program 
        #if searchObj.group(1): 
        sensor_names.append(serHC.read(2)) 
        #if searchObj.group(2): 
        sensor_values.append(serHC.read(2)) 
        #if searchObjRN.group(1): 
        sensor_names.append(serRN.read(2)) 
        #if searchObjRN.group(2): 
        sensor_values.append(serRN.read(2)) 
        i = i + 1 
    while 1: 
        # Get the key from the msvcrt module 
        key = getch().decode('ASCII') 
 
        # If key is pressed, do something 
        if key: 
            print(key) 
            # Zip them together 
            # Final 2D list 
            final_2d_list = zip(sensor_names,sensor_values) 
            print(list(sorted(final_2d_list))) 
            #vals = df.Dataframe([ 
            #df.append(vals) 
            #print(sorted_array_1stdim[r]) 
            #sensor_values = [0] * 10 
            # Thread for reading definition 
            break 
            # Fancy recursion 
    sensor_values.clear() 
    sensor_names.clear() 
    i = 0 
    serialin() 
serialin()

Arduino 代码:

// The device with green colored wires 
void setup() { 
 
  Serial.begin(115200); 
 
} 
 
void loop() { 
   // It won't work with the I2C while loop for some reason. Perhaps it is getting stuck up on it 
 
   Serial.print("A4"); 
   Serial.print(analogRead(0)); // Read the local analog signal 
   delay(5); 
   Serial.print("A5"); 
   Serial.print(analogRead(1)); // Read the local analog signal 
   delay(5); 
   Serial.print("A6"); 
   Serial.print(analogRead(2)); // Read the local analog signal 
   delay(5); 
   Serial.print("A7"); 
   Serial.print(analogRead(3)); // Read the local analog signal 
 
 
}

我正在尝试通过来自 sparkfun 的 bluetooth silver mate 和 HC-06 模块将来自传感器的模拟数据发送到 python。
我必须在每次延迟 5 秒后读取模拟数据，这样读数才不会冲突。
数据来自串行端口 COM9 和 COM10。我知道 python 中的 serial 可能会阻塞，这就是为什么我尝试先读取它，然后将其放入列表中的原因。 我也知道，一旦 serial 被通读，它似乎是非阻塞的。当我使用 serHC.readline() 和 serRN.readline() 时，我得到了我期望看到的东西。
但是，列表中的数据并未根据传感器的变化进行更新。我不得不承认 python 不是我的主要编程语言，所以这就是我寻求帮助的原因。 我想也许使用多个线程可能会起作用，但我无法在主线程中获取 serHC 和 serRN 变量。

任何帮助将不胜感激!!

请您参考如下方法:

正如您所发现的，不可能从串行端口顺序读取:一个端口上的阻塞读取意味着丢失了另一个端口上同时发送的数据。

使用基于线程的方法。

下面的草图应该足以开始:

import serial 
import time 
import re 
import threading 
 
BYTES_TO_READ = 6 
 
# read from serial port 
def read_from_serial(board, port): 
    print("reading from {}: port {}".format(board, port)) 
    payload = b'' 
    ser = serial.Serial(port, 115200,timeout=.250, parity=serial.PARITY_NONE, rtscts=1) 
    bytes_count = 0 
    while bytes_count < BYTES_TO_READ: 
        read_bytes = ser.read(2) 
 
        # sum number of bytes returned (not 2), you have set the timeout on serial port 
        # see https://pythonhosted.org/pyserial/pyserial_api.html#serial.Serial.read 
        bytes_count = bytes_count + len(read_bytes) 
        payload = payload + read_bytes 
 
    # here you have the bytes, do your logic 
    # ... 
    print("READ from {}: [{}]".format(board, payload)) 
    return 
 
 
def main(): 
 
    board = { 
        'JY': 'COM9', 
        'SILVER': 'COM10' 
    } 
 
    threads = [] 
    for b in board: 
 
        t = threading.Thread(target=read_from_serial, args=(b, board[b],)) 
        threads.append(t) 
        t.start() 
 
    # wait for all threads termination 
    for t in threads: 
        t.join() 
 
main()

学习线程:https://pymotw.com/3/threading/

定期从连续剧中读取

下面是每 TIME_PERIOD 秒阅读的草图。 围绕读取的无限 while 循环的一部分有一个带有嵌套 try/catch block 的“线程”循环 用于捕获串行通信问题并在 TIME_PERIOD 后重试连接。

就拿它作为一个开始的例子吧! 

import serial 
import time 
import re 
import threading 
 
BYTES_TO_READ = 6 
 
TIME_PERIOD = 5 
 
def read_message(board, port, handle): 
    payload = b'' 
    bytes_count = 0 
    while bytes_count < BYTES_TO_READ: 
        read_bytes = handle.read(2) 
        bytes_count = bytes_count + len(read_bytes) 
        payload = payload + read_bytes 
    # here you have the bytes, do your logic 
    # ... 
    print("READ from {}: [{}]".format(board, payload)) 
 
def serial_thread(board, port): 
    print("reading from {}: port {}".format(board, port)) 
 
    while True: 
        try: 
            handle = serial.Serial(port, 115200,timeout=.250, parity=serial.PARITY_NONE, rtscts=1) 
 
            while True: 
                read_message(board, port, handle) 
                time.sleep(TIME_PERIOD) 
        except Exception as e: 
            print("ERROR: {}".format(e)) 
            print("retrying in {} seconds".format(TIME_PERIOD)) 
            handle.close() 
            time.sleep(TIME_PERIOD) 
 
def main(): 
    board = { 
        'JY': '/dev/ttyUSB0', 
        'SILVER': '/dev/ttyACM0' 
    } 
    threads = [] 
    for b in board: 
        t = threading.Thread(target=serial_thread, args=(b, board[b],)) 
        threads.append(t) 
        t.start() 
    # wait for all threads termination 
    for t in threads: 
        t.join() 
 
main()


  文章来源：
  https://www.windname.com/post/ajfhij2b1.html
阅读延展