如何将MetaWear传感器API终端输出保存至JSON/CSV文件？

问题背景 #

我正在从MetaWear传感器导出日志数据，核心下载代码如下：

e = Event()
libmetawear.mbl_mw_logging_download(d.board, 0, byref(download_handler))
e.wait()

其中download_handler的定义为：

fn_wrapper = FnVoid_VoidP_UInt_UInt(progress_update_handler)
download_handler = LogDownloadHandler(context = None, received_progress_update = fn_wrapper, received_unknown_entry = cast(None, FnVoid_VoidP_UByte_Long_UByteP_UByte), received_unhandled_entry = cast(None, FnVoid_VoidP_DataP))  

运行后终端会输出以下格式的数据：

{epoch: 1663184089978, value: {x : 0.177, y : -0.060, z : 0.993}}
{epoch: 1663184089988, value: {x : 0.179, y : -0.059, z : 0.993}}
{epoch: 1663184089998, value: {x : 0.179, y : -0.060, z : 0.995}}
{epoch: 1663184090009, value: {x : 0.180, y : -0.060, z : 0.994}}

当前问题 #

我需要将这些数据保存到文件（最终目标是CSV格式），但尝试保存为JSON时，发现libmetawear.mbl_mw_logging_download返回None，使用以下代码写入的文件内容为null：

with open('data.json', 'w') as fp:
    json.dump(libmetawear.mbl_mw_logging_download(d.board, 0, byref(download_handler)), fp)

怀疑问题与未调用e.wait()或数据格式有关，寻求解决方法。

编辑：已有解决方案 #

@S.B提供了通过重定向标准输出保存数据的方法：

import sys

def print_to_terminal():
    """模拟传感器输出"""
    print("{epoch: 1663184089978, value: {x : 0.177, y : -0.060, z : 0.993}}")
    print("{epoch: 1663184089988, value: {x : 0.179, y : -0.059, z : 0.993}}")
    print("{epoch: 1663184089998, value: {x : 0.179, y : -0.060, z : 0.995}}")
    print("{epoch: 1663184090009, value: {x : 0.180, y : -0.060, z : 0.994}}")


with open("output.txt", "w", encoding="utf-8") as f:
    sys.stdout = f

    # 调用输出数据的函数
    print_to_terminal()

    # 恢复标准输出
    sys.stdout = sys.__stdout__

编辑：完整代码 #

以下是包含两种导出尝试的完整代码，downloadData()和downloadFormatted()分别对应两种不同的实现逻辑：

from __future__ import print_function
from mbientlab.metawear import MetaWear, libmetawear, parse_value, create_voidp, create_voidp_int
from mbientlab.metawear.cbindings import *
from time import sleep
from threading import Event
import platform
import sys
import time
import csv

# 全局变量定义
acceleration = [ [], [], [] ]
elapsedTime = [0]

class FusionDevice:

  def __init__(self, MACaddress):
    self.device = MetaWear(address)
    self.samples = 0
    self.callback = FnVoid_VoidP_DataP(self.data_handler)
    self.initTime = 0
    self.thisEpoch = 0
    
  def data_handler(self, ctx, data):
    coordinates = parse_value(data)
    acceleration[0].append(coordinates.x*9.8)
    acceleration[1].append(coordinates.y*9.8)
    acceleration[2].append(coordinates.z*9.8)
    self.thisEpoch = data.contents.epoch
    
    # 初始化第一个样本的基准时间
    if(self.samples == 0):
      self.initTime = self.thisEpoch
      
    # 计算后续样本的相对时间
    else:
      elapsedTime.append(float(self.thisEpoch-self.initTime))
      
    self.samples += 1
    
  def connect(self):
    self.device.connect()
    
  def configure(self):
    libmetawear.mbl_mw_settings_set_connection_parameters(self.device.board, 7.5, 7.5, 0, 6000)
    libmetawear.mbl_mw_acc_set_odr(self.device.board, 12.5) # 设置采样率12.5Hz
    libmetawear.mbl_mw_acc_set_range(self.device.board, 16.0)
    libmetawear.mbl_mw_acc_write_acceleration_config(self.device.board)
    
  def startLogging(self):
    signal = libmetawear.mbl_mw_acc_get_acceleration_data_signal(self.device.board)
    logger = create_voidp(lambda fn: libmetawear.mbl_mw_datasignal_log(signal, None, fn), resource = "acc_logger")
    libmetawear.mbl_mw_logging_start(self.device.board, 0)
    libmetawear.mbl_mw_acc_enable_acceleration_sampling(self.device.board)
    libmetawear.mbl_mw_acc_start(self.device.board)
    
  def downloadData(self):
    e = Event()
    def progress_update_handler(context, entries_left, total_entries):
      if (entries_left == 0):
        e.set()
        
    signal = libmetawear.mbl_mw_acc_get_acceleration_data_signal(self.device.board)
    logger = create_voidp(lambda fn: libmetawear.mbl_mw_datasignal_log(signal, None, fn), resource = "acc_logger")
    fn_wrapper = FnVoid_VoidP_UInt_UInt(progress_update_handler)
    download_handler = LogDownloadHandler(context = None, received_progress_update = fn_wrapper, received_unknown_entry = cast(None, FnVoid_VoidP_UByte_Long_UByteP_UByte), received_unhandled_entry = cast(None, FnVoid_VoidP_DataP))
    callback = FnVoid_VoidP_DataP(lambda ctx, p: print("{epoch: %d, value: %s}" % (p.contents.epoch, parse_value(p))))
    libmetawear.mbl_mw_logger_subscribe(logger, None, callback)
    libmetawear.mbl_mw_logging_download(self.device.board, 0, byref(download_handler))
    
  def downloadFormatted(self):
    libmetawear.mbl_mw_logging_stop(self.device.board)
    libmetawear.mbl_mw_acc_disable_acceleration_sampling(self.device.board)
    signal = libmetawear.mbl_mw_acc_get_acceleration_data_signal(self.device.board)
    logger = create_voidp(lambda fn: libmetawear.mbl_mw_datasignal_log(signal, None, fn), resource = "acc_logger")
    libmetawear.mbl_mw_logger_subscribe(logger, None, self.callback)
    libmetawear.mbl_mw_logging_download(self.device.board, 0, byref(self.data_handler))
    
    with open('Acc.csv', mode ='w') as acc_file:
      acc_writer = csv.writer(acc_file, delimiter=',', quoting=csv.QUOTE_MINIMAL)
      acc_writer.writerow(['Time(ms)', 'X', 'Y', 'Z' ])
      for i in range(len(acceleration[0])):
        acc_writer.writerow([elapsedTime[i], acceleration[0][i], acceleration[1][i],acceleration[2][i]])
        
  def reset(self):
    try:
      libmetawear.mbl_mw_logging_stop(self.device.board)
      libmetawear.mbl_mw_logging_clear_entries(self.device.board)
      libmetawear.mbl_mw_macro_erase_all(self.device.board)
      libmetawear.mbl_mw_debug_reset_after_gc(self.device.board)
      libmetawear.mbl_mw_debug_disconnect(self.device.board)
    except Exception as e:
      return -1
    return 0
    

# 设备测试流程
address = 'F2:C2:66:D8:D3:EA'
deviceTest = FusionDevice(address)
deviceTest.connect()
deviceTest.configure()
deviceTest.startLogging()
sleep(5.0)
# deviceTest.downloadData()
deviceTest.downloadFormatted()
deviceTest.reset()

内容的提问来源于stack exchange，提问作者mkp