python - 将加速度计数据存储在树莓派 SD 卡中-6ren

python - 将加速度计数据存储在树莓派 SD 卡中

转载作者：行者123 更新时间：2023-11-28 22:10:31

我正在从事一个项目，我的目标是存储来自电机的加速度数据。为此，我正在使用 adafruit mma8451 加速传感器。现在要存储 3 轴加速度数据，我使用的是 Raspberry pi。我正在尝试将数据存储在 CSV 文件中。现在我需要系统每秒至少存储 250 个样本。现在的问题是，树莓派存储的一秒可能是 230 个样本，下一秒是 210，再下一秒可能是 225。所以基本上，树莓派在这个速度下并不稳定。如果它每秒存储 230 个样本，那将永远不会成为问题。但是不接受不同数量的样本，因为我需要准确的采样频率来进一步分析信号。

为了检查这是否是将数据存储在 csv 文件中的问题，我尝试查看 pi 是否只能打印 250 次“Hello World”。看来，如果我使用 time.sleep() 命令使用计时器，它是不稳定的。现在我想展示我在我的项目中使用的代码

count=0
Path = 0

with open('/home/pi/TrialV10/VibrationsDevice1.csv', 'a+') as csvfile:
    sensorwriter = csv.writer(csvfile) #, delimiter=' ',
                            #quotechar='|', quoting=csv.QUOTE_MINIMAL)
    sensorwriter.writerow(['Datetime', 'Time', 'X_Accel (m/s^2)', 'Y_Accel (m/s^2)', 'Z_Accel (m/s^2)', 'Count', 'Path'])



    if MotoDetection == 1:
        sensor.MotionSetup()
        while True:
            sensor.Motion()
            time.sleep(0)  
            if sensor.Motion() == True:
                if count==0:
                    TimeStart = time.time()
                    Timer = 0                    
                    pass
                while True: 
                    x, y, z = sensor.acceleration
                    time_now = datetime.datetime.now().strftime("%Y-%m-%d")
                    TimePresent = time.time()
                    Timer = TimePresent - TimeStart
                    X = x #+ Calcx
                    Y = y #+ Calcy
                    Z = z #+ Calcz
                    count = count + 1                  
                    print('DateTime={0}  Time ={1} X={2:0.3f} m/s^2  Y:{3:0.3f} m/s^2  Z:{4:0.3f} m/s^2 count={5}'.format(time_now, Timer, X, Y, Z, count))
                    sensorwriter.writerow([time_now, Timer, X, Y, Z, count])
                    time.sleep(0.004) 
                    if Timer > TimingA:
                        exit()

最佳答案

执行此操作的一种方法是缓冲您的传感器读数，并让一个单独的线程将它们分组写入文件。例如:

import threading
class FileWriterThread(threading.Thread):
    def __init__(self,filename,header):
        threading.Thread.__init__(self)
        self.buffer = []
        self.filename = filename
        self.header = header
    def run(self):
        with open(self.filename,"a+") as f:
            csvwriter = csv.writer(f)
            csvwriter.writerow(self.header)
            while True:
                time.sleep(1) #Wait 1 second between writes to the file.
                writebuffer = []
                while len(self.buffer) > 0: #Clear the current buffer
                    writebuffer.append(self.buffer.pop(0))
                while len(writebuffer) > 0:
                    csvwriter.writerow(writebuffer.pop(0))
    def writerow(self,row):
        self.buffer.append(row)

要打开文件并启动线程，您可以运行:

sensorwriter = FileWriterThread(filename,header)
sensorwriter.start()

要存储要写入文件的行，您可以使用:

sensorwriter.writerow(row)

最后，要停止线程并保存文件，您可以使用:

sensorwriter.cancel()

请记住，您必须创建一个新线程才能重新启动它。实现到您当前的代码中，这将导致:

import threading
class FileWriterThread(threading.Thread):
    def __init__(self,filename,header):
        threading.Thread.__init__(self)
        self.buffer = []
        self.filename = filename
        self.header = header
    def run(self):
        with open(self.filename,"a+") as f:
            csvwriter = csv.writer(f)
            csvwriter.writerow(self.header)
            while True:
                time.sleep(1) #Wait 1 second between writes to the file.
                writebuffer = []
                while len(self.buffer) > 0: #Clear the current buffer
                    writebuffer.append(self.buffer.pop(0))
                while len(writebuffer) > 0:
                    csvwriter.writerow(writebuffer.pop(0))
    def writerow(self,row):
        self.buffer.append(row)

sensorwriter = FileWriterThread('/home/pi/TrialV10/VibrationsDevice1.csv',['Datetime', 'Time', 'X_Accel (m/s^2)', 'Y_Accel (m/s^2)', 'Z_Accel (m/s^2)', 'Count', 'Path'])
sensorwriter.start()

count=0
Path = 0

if MotoDetection == 1:
    sensor.MotionSetup()
    while True:
        sensor.Motion()
        time.sleep(0)  
        if sensor.Motion() == True:
            if count==0:
                TimeStart = time.time()
                Timer = 0                    
                pass
            while True: 
                x, y, z = sensor.acceleration
                time_now = datetime.datetime.now().strftime("%Y-%m-%d")
                TimePresent = time.time()
                Timer = TimePresent - TimeStart
                X = x #+ Calcx
                Y = y #+ Calcy
                Z = z #+ Calcz
                count = count + 1                  
                print('DateTime={0}  Time ={1} X={2:0.3f} m/s^2  Y:{3:0.3f} m/s^2  Z:{4:0.3f} m/s^2 count={5}'.format(time_now, Timer, X, Y, Z, count))
                sensorwriter.writerow([time_now, Timer, X, Y, Z, count])
                time.sleep(0.004) 
                if Timer > TimingA:
                    sensorwriter.cancel()
                    exit()

希望这对您有所帮助。

编辑:问题似乎与文件 I/O 无关，很可能与打印语句有关。将文本打印到控制台是不稳定的，这实际上是导致稳定性问题的原因。不幸的是，没有办法让它 100% 稳定，但是你可以通过替换来稍微稳定它

print('DateTime={0}  Time ={1} X={2:0.3f} m/s^2  Y:{3:0.3f} m/s^2  Z:{4:0.3f} m/s^2 count={5}'.format(time_now, Timer, X, Y, Z, count))

与

sys.stdout.write('DateTime={0}  Time ={1} X={2:0.3f} m/s^2  Y:{3:0.3f} m/s^2  Z:{4:0.3f} m/s^2 count={5}\n'.format(time_now, Timer, X, Y, Z, count))

因为 sys.stdout.write() 似乎比 print() 更稳定。请记住在程序开始时导入 sys。不幸的是，总会有一些不稳定，这取决于操作系统当时正在运行的其他任务。

尝试最大化稳定性的一种方法是尝试限制采样率。放上代码

import time
SAMPLERATE = 150 #Number of samples per second.
SAMPLEDELAY = 1/SAMPLERATE
loopwait = 0

在你的程序和代码的开始

while time.time() < loopwait:
    pass
loopwait = time.time()+SAMPLEDELAY

作为 while True 循环中的第一条指令。这使得最终代码:

import threading,time
class FileWriterThread(threading.Thread):
    def __init__(self,filename,header):
        threading.Thread.__init__(self)
        self.buffer = []
        self.filename = filename
        self.header = header
    def run(self):
        with open(self.filename,"a+") as f:
            csvwriter = csv.writer(f)
            csvwriter.writerow(self.header)
            while True:
                time.sleep(1) #Wait 1 second between writes to the file.
                writebuffer = []
                while len(self.buffer) > 0: #Clear the current buffer
                    writebuffer.append(self.buffer.pop(0))
                while len(writebuffer) > 0:
                    csvwriter.writerow(writebuffer.pop(0))
    def writerow(self,row):
        self.buffer.append(row)

sensorwriter = FileWriterThread('/home/pi/TrialV10/VibrationsDevice1.csv',['Datetime', 'Time', 'X_Accel (m/s^2)', 'Y_Accel (m/s^2)', 'Z_Accel (m/s^2)', 'Count', 'Path'])
sensorwriter.start()

SAMPLERATE = 150 #Number of samples per second
SAMPLEDELAY = 1/SAMPLERATE
loopwait = 0

count=0
Path = 0

if MotoDetection == 1:
    sensor.MotionSetup()
    while True:
        sensor.Motion()
        time.sleep(0)  
        if sensor.Motion() == True:
            if count==0:
                TimeStart = time.time()
                Timer = 0                    
                pass
            while True: 
                while time.time() < loopwait:
                    pass
                loopwait = time.time()+SAMPLEDELAY
                x, y, z = sensor.acceleration
                time_now = datetime.datetime.now().strftime("%Y-%m-%d")
                TimePresent = time.time()
                Timer = TimePresent - TimeStart
                X = x #+ Calcx
                Y = y #+ Calcy
                Z = z #+ Calcz
                count = count + 1                  
                sys.stdout.write('DateTime={0}  Time ={1} X={2:0.3f} m/s^2  Y:{3:0.3f} m/s^2  Z:{4:0.3f} m/s^2 count={5}\n'.format(time_now, Timer, X, Y, Z, count))
                sensorwriter.writerow([time_now, Timer, X, Y, Z, count])
                time.sleep(0.004) 
                if Timer > TimingA:
                    sensorwriter.cancel()
                    exit()