从 python 中的各种 USB 设备读取和储存各种数据

read and stock various data from various usb devices in python

我是 python 的初学者,我正在尝试从通过 USB 集线器连接到计算机的多个传感器(湿度、温度、压力传感器...)读取数据。我的主要目标是每五分钟记录一次这些传感器的不同值,然后将其存储起来进行分析。

我已经获得了所有 data sheets and manuals 传感器(来自 Hygrosens Instruments),我知道它们是如何工作的以及它们发送的数据类型。但我不知道如何阅读它们。以下是我使用 pyserial 进行的尝试。

import serial #import the serial library
from time import sleep #import the sleep command from the time library
import binascii

output_file = open('hygro.txt', 'w') #create a file and allow you to write in it only. The name of this file is hygro.txt

ser = serial.Serial("/dev/tty.usbserial-A400DUTI", 9600) #load into a variable 'ser' the information about the usb you are listening. /dev/tty.usbserial.... is the port after plugging in the hygrometer, 9600 is for bauds, it can be diminished
count = 0
while 1:
    read_byte = ser.read(size=1)

所以现在我想找到数据行的结尾,因为我需要的测量信息在以 'V' 开头的行中,如果我的数据 sheet传感器,它说一行以 结尾,所以我想一次读取一个字节并查找 '<',然后是 'c',然后是 'r',然后是 '>'。所以我想这样做:

while 1:
    read_byte = ser.read(size=8) #read a byte
    read_byte_hexa =binascii.hexlify(read_byte) #convert the byte into hexadecimal

    trad_hexa = int(read_byte_hexa , 16) #convert the hexadecimal into an int in purpose to compare it with another int
    trad_firstcrchar = int('3c' , 16) #convert the hexadecimal of the '<' into a int to compare it with the first byte    
    if (trad_hexa == trad_firstcrchar ): #compare the first byte with the '<'    
        read_byte = ser.read(size=1) #read the next byte (I am not sure if that really works)
        read_byte_hexa =binascii.hexlify(read_byte)# from now I am doing the same thing as before
        trad_hexa = int(read_byte_hexa , 16)
        trad_scdcrchar = int('63' , 16)
        print(trad_hexa, end='/')# this just show me if it gets in the condition
        print(trad_scdcrchar)    
        if (trad_hexa == trad_scdcrchar ):    
            read_byte = ser.read(size=1) #read the next byte 
            read_byte_hexa =binascii.hexlify(read_byte)
            trad_hexa = int(read_byte_hexa , 16)
            trad_thirdcrchar = int('72' , 16)
            print(trad_hexa, end='///')
            print(trad_thirdcrchar)    
            if (trad_hexa == trad_thirdcrchar ):    
                read_byte = ser.read(size=1) #read the next byte 
                read_byte_hexa =binascii.hexlify(read_byte)
                trad_hexa = int(read_byte_hexa , 16)
                trad_fourthcrchar = int('3e' , 16)
                print(trad_hexa, end='////')
                print(trad_fourthcrchar)    
                if (trad_hexa == trad_fourthcrchar ):    
                    print ('end of the line')

但我不确定它是否有效,我的意思是我认为它没有时间读取第二个,我正在读取的第二个字节,它不完全是第二个。所以这就是为什么我想使用缓冲区,但我真的不明白我该怎么做。我将寻找它,但如果有人知道更简单的方法来做我想做的事,我准备尝试一下! 谢谢

您的印象似乎是传感器通信协议的行尾字符是 4 个不同的字符:<cr>。然而,这里指的是 carriage return,通常表示为 <cr>,在许多编程语言中仅表示为 \r(尽管它看起来像 2 个字符,但它表示 只有一个个字符)。

您可以通过逐行读取来自传感器的数据来大大简化您的代码,因为协议是结构化的。以下内容可帮助您入门:

import time

def parse_info_line(line):
    # implement to your own liking
    logical_channel, physical_probe, hardware_id, crc = [line[index:index+2] for index in (1, 3, 5, 19)]
    serialno = line[7:19]
    return physical_probe

def parse_value_line(line):
    channel, crc = [line[ind:ind+2] for ind in (1,7)]
    encoded_temp = line[3:7]
    return twos_comp(int(encoded_temp, 16), 16)/100.

def twos_comp(val, bits):
    """compute the 2's compliment of int value `val`"""
    if (val & (1 << (bits - 1))) != 0: # if sign bit is set e.g., 8bit: 128-255
        val = val - (1 << bits)        # compute negative value
    return val                         # return positive value as is

def listen_on_serial(ser):
    ser.readline() # do nothing with the first line: you have no idea when you start listening to the data broadcast from the sensor
    while True:
        line = ser.readline()
        try:
            first_char = line[0]
        except IndexError:  # got no data from sensor
            break
        else:
            if first_char == '@':  # begins a new sensor record
                in_record = True
            elif first_char == '$':
                in_record = False
            elif first_char == 'I':
                parse_info_line(line)
            elif first_char == 'V':
                print(parse_value_line(line))
            else:
                print("Unexpected character at the start of the line:\n{}".format(line))
            time.sleep(2)

twos_comp 函数是 written by travc,如果您有足够的声誉并且打算使用他的代码(即使您不这样做,它仍然是一个很好的答案,我刚才投票了)。 listen_on_serial 也可以改进(许多 Python 程序员会识别开关结构并使用字典而不是 if... elif... elif... 来实现它),但这只是为了让您入门。

作为测试,以下代码摘录模拟传感器发送一些数据(以行分隔,使用回车 return 作为行尾标记),我从 pdf 中复制了这些数据你链接到 (FAQ_terminalfenster_E.pdf).

>>> import serial
>>> import io
>>> 
>>> ser = serial.serial_for_url('loop://', timeout=1)
>>> serio = io.TextIOWrapper(io.BufferedRWPair(ser, ser), newline='\r', line_buffering=True)
>>> serio.write(u'A1A0\r'  # simulation of starting to listen halfway between 2 records
...     '$\r'              # marks the end of the previous record
...     '@\r'              # marks the start of a new sensor record
...     'I0101010000000000001B\r'  # info about a sensor's probe
...     'V0109470D\r'              # data matching that probe
...     'I0202010000000000002B\r'  # other probe, same sensor
...     'V021BB55C\r')             # data corresponding with 2nd probe
73L
>>> 
>>> listen_on_serial(serio)
23.75
70.93
>>> 

注意是recommended by the pyserial docs to be using TextIOWrapper when the end-of-line character is not \n (the linefeed character), as was also answered here.