Saving API terminal output to JSON file

Question

I am exporting data from a sensor. The command to download the logged data from the sensor is:

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

download_handler is a function from the sensor package:

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

When ran, this outputs the data in the terminal in a format as follows:

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

I would like to save this to a file. Given the output format, I'm attempting to save to a JSON file (eventually, I'd like to save it to CSV). However, assigning a variable returns 'None' with format NoneType and trying to write a file (see below) returns 'null' within the file:

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

This may have to do with the 'e.wait()' command not being called or a formatting issue. Any help is appreciated!

EDIT: The following solution was provided by @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__

EDIT: Full code as requested. Note that downloadData() and downloadFormatted() are attempting to export data two different ways.

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()

S.B · Answer

I didn't work with that package before, so if there is not an option to retrieve the string representation of the JSON response and it just prints it to the terminal, use the following:

Change the sys.stdout. It is where the output goes and by default sys.stdout is connected to the terminal:

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__

Another option is of course just running your python file and redirect the stdout using shell feature(here BASH):

python your_script.py > output.txt

This sets the standard output to the output.txt so it's no longer the terminal and you don't need to change (add) anything to your code.

If you also want to redirect the standard error(where the error messages go), you know what to change --> sys.stderr. You can set them to the same file.

and for second option use:

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

ndclt · Answer

It's hard to answer correctly as I can't run your code.

I guess that the printing is done by the mbl_mw_logging_download function. You can redirect the stdout into a file (official Python doc here).

In your case, it could be done like this:

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)

The written file is not a json but you can read it after for working on its content.

Saving API terminal output to JSON file

2 Answers