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

jobtbby3 于 2023-02-17 发布在其他

关注(0)|答案(2)|浏览(153)

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

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中）。然而，赋值一个变量返回'None'格式为NoneType，而尝试写一个文件（见下文）在文件中返回'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()

JSON

来源：https://stackoverflow.com/questions/73722463/saving-api-terminal-output-to-json-file

2条答案

按热度按时间

xqkwcwgp1#

我以前没有使用过这个包，所以如果没有检索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）重定向stdout：

python your_script.py > output.txt

这会将标准输出设置为output.txt，因此它不再是终端，并且您不需要更改（添加）代码中的任何内容。
如果您还想重定向标准错误（错误消息所在的位置），您知道要更改什么--〉sys.stderr。
对于第二选项用途：

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

赞(0）回复(0）举报 2023-02-17

c0vxltue2#

很难正确回答，因为我无法运行您的代码。
我猜打印是由mbl_mw_logging_download函数完成的，您可以将stdout重定向到一个文件中（这里是Python的官方文档）。
在您的情况下，可以这样做：

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，但您可以在处理其内容后读取它。

赞(0）回复(0）举报 2023-02-17

我来回答

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

2条答案

相关问题

热门标签

最新问答