Python - 加速度计以 1 kHz 速率读取和写入 CSV 文件

Question

我正在尝试使用 MPU-6000 加速度计和 Raspberry Pi Zero W 来记录挡风玻璃中的振动数据。我对 Python 还很陌生，所以请耐心等待。

我编写了一个 python2 脚本，用于将 MPU-6000 配置为通过 I2C 进行通信，并将时钟配置为 400 kHz。 当加速度计寄存器中有新数据可用时，MPU-6000 会发出中断，该数据被读取、转换为 2 的补码，然后与时间戳一起写入 CSV 文件。加速度计的输出速率配置为 1 kHz。

我遇到的情况是，在对所有三个传感器轴进行采样时，脚本无法将所有数据点写入 CSV 文件。我得到了大约 650 个数据点 pr 轴 pr 秒，而不是 1000 个数据点 pr 轴 pr 秒。 我尝试过只编写一个轴，事实证明，每秒 1000 个数据点是成功的。我知道 MPU-6000 有一个可用的 FIFO 寄存器，我可能可以突发读取以获得 1000 个样本/秒，没有任何问题。问题是获取每个样本的时间戳，所以我还没有尝试实现从 FIFO 寄存器读取。

我很可能会在 Matlab 中完成大部分后处理，因此 python 脚本应该做的最重要的事情是以确定的速率将任何形式的传感器数据写入 CSV 文件，并带有时间戳。

有没有办法进一步改进我的 Python 脚本，以便我可以对所有三个轴进行采样并以 1 kHz 的速率写入 CSV 文件？

我的脚本的部分内容如下所示:

#!/usr/bin/python
import smbus
import math
import csv
import time
import sys
import datetime

# Register addresses
power_mgmt_1 = 0x6b
power_mgmt_2 = 0x6c
samlerate_divider = 0x19
accel_config = 0x1C
INT_Enable = 0x38



def read_byte(reg):
    return bus.read_byte_data(address, reg)

def read_word(reg):
    h = bus.read_byte_data(address, reg)
    l = bus.read_byte_data(address, reg+1)
    value = (h <<8)+l
    return value

def read_word_2c(reg):
    val = read_word(reg)

    if (val >= 0x8000):
        return -((65535 - val) + 1)
    else:
        return val



csvwriter = None

def csv_open():
    csvfile = open('accel-data.csv', 'a')

    csvwriter = csv.writer(csvfile)




def csv_write(timedelta, accelerometerx, accelerometery, accelerometerz):

    global csvwriter

    csvwriter.writerow([timedelta,  accelerometerx, accelerometery, 
    accelerometerz])



# I2C configs
bus = smbus.SMBus(1) 
address = 0x69       

#Power management configurations
bus.write_byte_data(address, power_mgmt_1, 0)
bus.write_byte_data(address, power_mgmt_2, 0x00)


#Configure sample-rate divider
bus.write_byte_data(address, 0x19, 0x07)


#Configure data ready interrupt:
bus.write_byte_data(address,INT_Enable, 0x01) 



#Opening csv file and getting ready for writing
csv_open()

csv_write('Time', 'X_Axis', 'Y_Axis', 'Z_Axis') 

print
print "Accelerometer"
print "---------------------"

print "Printing acccelerometer data: "

#starttime = datetime.datetime.now()

while True:


    data_interrupt_read =  bus.read_byte_data(address, 0x3A)

    if data_interrupt_read == 1:


        meas_time = datetime.datetime.now()
#       delta_time = meas_time - starttime


        accelerometer_xout = read_word_2c(0x3b)
        accelerometer_yout = read_word_2c(0x3d)
        accelerometer_zout = read_word_2c(0x3f)

#       accelerometer_xout = read_word(0x3b)
#       accelerometer_yout = read_word(0x3d)
#       accelerometer_zout = read_word(0x3f)

#       accelerometer_xout_scaled = accelerometer_xout / 16384.0
#       accelerometer_yout_scaled = accelerometer_yout / 16384.0
#       accelerometer_zout_scaled = accelerometer_zout / 16384.0



#       csv_write(meas_time, accelerometer_xout_scaled, 
        accelerometer_yout_scaled, accelerometer_zout_scaled)

        csv_write(meas_time, accelerometer_xout, accelerometer_yout, 
        accelerometer_zout) 


    continue

Answer 1

如果您尝试写入的数据是连续的，那么最好的方法是最大限度地减少写入所需的处理量，并最大限度地减少写入的数据量。为此，一个好的方法是将原始数据写入二进制格式的文件。每个数据字只需要写入 2 个字节。 datetime 对象可以转换为时间戳，需要 4 个字节。因此，您可以使用如下格式:

[4 byte timestamp][2 byte x][2 byte y][2 byte z]

Python 的 struct 库可用于将多个变量转换为可写入文件的单个二进制字符串。数据似乎已签名，如果是这种情况，您可以尝试按原样写入单词，然后使用支持签名值的内置库稍后将其读回。

例如，可以使用以下命令将原始数据写入二进制文件:

#!/usr/bin/python
import smbus
import math
import csv
import time
import sys
import datetime
import struct

# Register addresses
power_mgmt_1 = 0x6b
power_mgmt_2 = 0x6c
samlerate_divider = 0x19
accel_config = 0x1C
INT_Enable = 0x38


def read_byte(reg):
    return bus.read_byte_data(address, reg)


def read_word(reg):
    h = bus.read_byte_data(address, reg)
    l = bus.read_byte_data(address, reg+1)
    value = (h <<8)+l
    return value


# I2C configs
bus = smbus.SMBus(1) 
address = 0x69       

#Power management configurations
bus.write_byte_data(address, power_mgmt_1, 0)
bus.write_byte_data(address, power_mgmt_2, 0x00)

#Configure sample-rate divider
bus.write_byte_data(address, 0x19, 0x07)

#Configure data ready interrupt:
bus.write_byte_data(address, INT_Enable, 0x01) 


print
print "Accelerometer"
print "---------------------"

print "Printing accelerometer data: "

#starttime = datetime.datetime.now()
bin_format = 'L3H'

with open('accel-data.bin', 'ab') as f_output:
    while True:
        #data_interrupt_read =  bus.read_byte_data(address, 0x3A)
        data_interrupt_read = 1

        if data_interrupt_read == 1:
            meas_time = datetime.datetime.now()
            timestamp = time.mktime(meas_time.timetuple())

            accelerometer_xout = read_word(0x3b)
            accelerometer_yout = read_word(0x3d)
            accelerometer_zout = read_word(0x3f)

            f_output.write(struct.pack(bin_format, timestamp, accelerometer_xout, accelerometer_yout, accelerometer_zout))

稍后，您可以使用以下方法将二进制文件转换为 CSV 文件:

from datetime import datetime
import csv
import struct

bin_format = 'L3h'  # Read data as signed words
entry_size = struct.calcsize(bin_format)

with open('accel-data.bin', 'rb') as f_input, open('accel-data.csv', 'wb') as f_output:
    csv_output = csv.writer(f_output)
    csv_output.writerow(['Time', 'X_Axis', 'Y_Axis', 'Z_Axis'])

    while True:
        bin_entry = f_input.read(entry_size)

        if len(bin_entry) < entry_size:
            break

        entry = list(struct.unpack(bin_format, bin_entry))
        entry[0] = datetime.fromtimestamp(entry[0]).strftime('%Y-%m-%d  %H:%M:%S')
        csv_output.writerow(entry)

如果您的数据收集不连续，您可以使用线程。一个线程会将您的数据读入一个特殊的队列。另一个线程可以将队列中的项目读取到磁盘上。

如果是连续的，如果数据写入速度慢于读取速度，则此方法将失败。

看看特价Format characters用于告诉struct如何打包和解包二进制数据。