将 API 终端输出保存到 JSON 文件

Question

我正在从传感器导出数据。从传感器下载记录数据的命令是：

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}}

我想将其保存到文件中。鉴于输出格式，我试图保存到一个 JSON 文件（最终，我想将它保存到 CSV）。但是，分配变量返回格式为 NoneType 的“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():
    """Simulate the output"""
    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

    # calling your function
    print_to_terminal()

    # restoring it back to default
    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

#Definitions
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
    
    #First sample
    if(self.samples == 0):
      self.initTime = self.thisEpoch
      
    #Rest of samples
    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()

Answer 1

我之前没有使用过那个包，所以如果没有选项来检索 JSON 响应的字符串表示并且它只是将它打印到终端，请使用以下命令：

更改sys.stdout。这是输出的地方，默认情况下 sys.stdout 连接到终端：

import sys

def print_to_terminal():
    """Simulate the output"""
    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

    # calling your function
    print_to_terminal()

    # restoring it back to default
    sys.stdout = sys.__stdout__

另一种选择当然是运行您的 python 文件并使用 shell 功能（此处为 BASH）重定向标准输出：

python your_script.py > output.txt

这会将标准输出设置为 output.txt，因此它不再是终端，您无需更改（添加）任何代码。

如果您还想重定向标准错误（错误消息所在的位置），您知道要更改什么 --> sys.stderr。您可以将它们设置为同一个文件。

和第二个选项使用：

python your_script.py > output.txt 2>&1

Answer 2

很难正确回答，因为我无法运行您的代码。

我猜打印是由 mbl_mw_logging_download 函数完成的。您可以将标准输出重定向到文件 (official Python doc here)。

在您的情况下，可以这样做：

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

写入的文件不是 json，但您可以在处理其内容后阅读它。

Answer 3

你有没有让这个工作？