Particle.Publish


#1

Hello. I am currently using a Photon to publish real-time sensor data (humidity, temperature, pressure, audio FFT) to Particle and send the data through webhooks to AWS API Gateway. Are there any foreseen limitations to publishing a continuous stream of data leveraging the Particle Publish method? I am interested in this method because we have hit memory limitations using AWS IOT, TLS and other methods due to the size of our payload. Thanks.


#2

How about the consulting the docs about this?
https://docs.particle.io/reference/device-os/firmware/photon/#particle-publish-


#3

I am aware of the docs. I was more looking to hear how others are achieving this with larger payloads. Sorry If I wasn’t clear.


#4

Pretty specific here:

https://docs.particle.io/reference/device-os/firmware/argon/#particle-publish-

image


#5

How large are your payloads, mainly the Audio FFT ?
In a perfect world: Each publish can send 622 characters of data at 1 sec intervals.


#6

If these limits are too small - then perhaps you need to look another transport type such as MQTT?


#7

@shanevanj

I appreciate the suggestion. The initial goal is to sample 1-5 seconds of audio using FFT when a decibel threshold is crossed and then send the data using Particle.publish. I am fairly new to c++ programming. I have been reading various threads on the forums about FFT and different implementations. Just trying to figure out which one will work for my use case.

I tried using MQTT with TLS.The real challenge has been limitations in available memory and publish size.


#8

I use MQTT as well, though with small payloads. What size publish are you attempting? Particle and AWS IoT (MQTT) both have size limits. You still have options, including running your own MQTT broker or REST API within AWS if you must have large payloads.

With respect to what’s available within Particle or AWS, we’d need more details. What does your code look like? Is it possible to break up your transmissions into multiple parts?


#9

Code is below.

/*
 * Project PhotonFFT
 * Description:
 * Author:
 * Date:
 */

#include "adc_hal.h"
#include "gpio_hal.h"
#include "pinmap_hal.h"
#include "pinmap_impl.h"

#include "ArduinoJson.h"

// TLS certificates

#define AWSIOTROOTCA                                                   \

"-----END CERTIFICATE-----"
const char caPem[] = AWSIOTROOTCA;


#define CERTIFICATE                                                    \
"-----BEGIN CERTIFICATE----- \r\n"                                     \

"-----END RSA PRIVATE KEY----- "
const char keyPem[] = PRIVATEKEY;

#include "particle-BNO055.h"
#define sensor_BNO_BufferSize (128)  // 100=1 second, 128 is the next that devides into 256 evenly
Adafruit_BNO055 bno = Adafruit_BNO055(55);
struct BNOData {
  float Accel_X;
  float Accel_Y;
  float Accel_Z;
};

BNOData last_BNO_Data = {0,0,0};
BNOData sensor_BNO_RingBuffer[sensor_BNO_BufferSize];
uint8_t firstIndex_BNO_RingBuffer = 0; // these are 0-255 for the modulo 64 to work
uint8_t lastIndex_BNO_RingBuffer = 0;

#include "ArduinoJson.h"
#include "MQTT-TLS.h"

MQTT client("amazonaws.com", 8883, callback, 1024);
//MQTT client(".amazonaws.com", 8883, callback);

// Threshold Event Transmit trigger
float threshold_dB_Audio = 1000;  // Based on the "20 log input voltage" method

#include "photon_dmaadc.h"
#include "arduinoFFT.h"

// For BME280
//#include <Wire.h>
#include <Adafruit_Sensor.h>
#include <Adafruit_BME280.h>
#define SEALEVELPRESSURE_HPA (1013.25)
#define sensorBufferSize (64)  // 50=5 seconds, 64 is the next that devides into 256 evenly

Adafruit_BME280 bme; // I2C
struct BME280Data {
  float Temperature;
  float Humidity;
  float Pressure;
};
enum sensorType {Temperature, Humidity, Pressure};
struct sensorData {
  short sensorType;
  float sensorValue;
};

BME280Data lastSensorData = {0,0,0};
//sensorData sensorDataRingBuffer[sensorBufferSize];
BME280Data sensorDataRingBuffer[sensorBufferSize];
uint8_t firstIndexRingBuffer = 0; // these are 0-255 for the modulo 64 to work
uint8_t lastIndexRingBuffer = 0;

unsigned long last_BME_Poll = 0;
unsigned long last_BME_interval = 100;  // 100ms is 10Hz
unsigned long last_BNO_Poll = 0;
unsigned long last_BNO_interval = 10;  // 10ms is 100Hz
unsigned long last_time_display = 0;
unsigned long last_time_interval = 1000;
unsigned long current_time = 0;
// 

// SYSTEM_THREAD(ENABLED);
// SYSTEM_MODE(AUTOMATIC);

// 2048 is a good size for this buffer. This is the number of samples; the number of bytes is twice
// this, but only half the buffer is sent a time, and then each pair of samples is averaged, so
// this results in 1024 byte writes from TCP, which is optimal.
// This uses 4096 bytes of RAM, which is also reasonable.
const size_t SAMPLE_BUF_SIZE = 2048;  // 2048 is 1/16 second @ 32kHz

// This is the pin the microphone is connected to.
const int SAMPLE_PIN = A0;

// The audio sample rate. The minimum is probably 8000 for minimally acceptable audio quality.
// Not sure what the maximum rate is, but it's pretty high.
const long SAMPLE_RATE = 32000;
const uint16_t sampleCount = SAMPLE_BUF_SIZE / 2; 
const double samplingFrequency = 32000;

uint16_t samples[SAMPLE_BUF_SIZE];
ADCDMA adcDMA(A0, samples, SAMPLE_BUF_SIZE);
arduinoFFT FFT = arduinoFFT(); /* Create FFT object */

void setup() {
  WiFi.setHostname("photon");
	Serial.begin(115200);
    delay(4000);

  // Start up the BME280
  while (!bme.begin()) {
    Serial.println("Could not find a valid BME280 sensor, check wiring, address, sensor ID!");
    delay(1000);
	}

  while(!bno.begin()) {
      /* There was a problem detecting the BNO055 ... check your connections */
    Serial.print("Ooops, no BNO055 detected ... Check your wiring or I2C ADDR!");
    delay(1000);
  }

  // Calibrate Acceleration values, doesn't seem to do anything
  //https://learn.adafruit.com/adafruit-bno055-absolute-orientation-sensor/device-calibration
  //Serial.println("Calibration status values: 0=uncalibrated, 3=fully calibrated");

  //uint8_t system, gyro, accel, mag = 0;
  //bno.getCalibration(&system, &gyro, &accel, &mag);
  //while(accel < 3){
   // bno.getCalibration(&system, &gyro, &accel, &mag);
   // Serial.print(accel, DEC);
   // delay(100);
 // }
  // 

  if(connectToAWSIoT())
  {
    Serial.println("Connected to AWSIoT...");
  }
  else
  {
    Serial.println("Not Connected to AWSIoT...");
  }
  Time.zone(-4);
  //Serial.println(Time.format(TIME_FORMAT_ISO8601_FULL));
  Serial.println(Time.format(Time.now(), "%Y-%m-%d %H:%M:%S"));

  adcDMA.start(samplingFrequency);

  String serverName = ".amazonaws.com";
  IPAddress remoteIP(WiFi.resolve(serverName));
  int numberOfReceivedPackage = 0;
  delay(1000);
  Serial.println("Server name resolves to: " + String(remoteIP));
  Serial.println("Pinging " + serverName + " started...");
    
  numberOfReceivedPackage = WiFi.ping(remoteIP);
    
  Serial.print("Pinging ended ->");
  Serial.println(numberOfReceivedPackage);

  uint32_t freemem = System.freeMemory();
  Serial.print("Free memory: ");
  Serial.println(freemem);

  delay(10000);
}

double vReal[sampleCount];
double vImag[sampleCount];
#define SCL_INDEX 0x00
#define SCL_TIME 0x01
#define SCL_FREQUENCY 0x02
#define SCL_PLOT 0x03

void loop()
{

  bool readyData = false;
	current_time = millis(); // For synchronisation of envSensor read
/*
    if (DMA_GetFlagStatus(DMA2_Stream0, DMA_FLAG_HTIF0)) 
    {
        DMA_ClearFlag(DMA2_Stream0, DMA_FLAG_HTIF0);
        for(uint16_t idx = 0; idx < sampleCount; idx++)
        {
            vReal[idx] = samples[idx] / 2;
            vImag[idx] = 0.0;
            // Serial.println(vReal[idx]);
        }
        readyData = true;
    }

    if (DMA_GetFlagStatus(DMA2_Stream0, DMA_FLAG_TCIF0)) {
        DMA_ClearFlag(DMA2_Stream0, DMA_FLAG_TCIF0);
        for(uint16_t idx = 0; idx < sampleCount; idx++)
        {
            vReal[idx] = samples[idx + sampleCount] / 2;
            vImag[idx] = 0.0;
            // Serial.println(vReal[idx]);
        }
        readyData = true;
    }

    if(readyData)
    {

        // Serial.println("Data:");
        // PrintVector(vReal, sampleCount, SCL_TIME);
        FFT.Windowing(vReal, sampleCount, FFT_WIN_TYP_HAMMING, FFT_FORWARD);	// Weigh data 
        // Serial.println("Weighed data:");
        // PrintVector(vReal, sampleCount, SCL_TIME);
        

        FFT.Compute(vReal, vImag, sampleCount, FFT_FORWARD); // Compute FFT 
        // Serial.println("Computed Real values:");
        // PrintVector(vReal, sampleCount, SCL_INDEX);
        // Serial.println("Computed Imaginary values:");
        // PrintVector(vImag, sampleCount, SCL_INDEX);
        FFT.ComplexToMagnitude(vReal, vImag, sampleCount); // Compute magnitudes 
        // Serial.println("Computed magnitudes:");
         PrintVector(vReal, (sampleCount >> 1), SCL_FREQUENCY);

        double x = FFT.MajorPeak(vReal, sampleCount, samplingFrequency);
        
        // A "20 log rule" to compute dB from audio input voltage
        // Voltage gain (dB) = 20×log (Audio output voltage / Audio input voltage)
        //Serial.printf("Peak dB: %d", 20 * log(x));
         Serial.println(x, 6);
    }
*/



//Serial.println("here");
  EnvironmentalData();  // Refresh envSensor
  IMU_Data();
 // printValues();
}

void PrintVector(double *vData, uint16_t bufferSize, uint8_t scaleType)
{
  for (uint16_t i = 0; i < bufferSize; i++)
  {
    double abscissa;
    /* Print abscissa value */
    switch (scaleType)
    {
      case SCL_INDEX:
        abscissa = (i * 1.0);
	break;
      case SCL_TIME:
        abscissa = ((i * 1.0) / samplingFrequency);
	break;
      case SCL_FREQUENCY:
//        abscissa = ((i * 1.0 * samplingFrequency) / sampleCount);
	break;
    }
//    Serial.print(abscissa, 6);
 //   if(scaleType==SCL_FREQUENCY)
 //     Serial.print("Hz");
 //   Serial.print(" ");
 //   Serial.println(vData[i], 4);
  //  // Possible threshold Event detection
  //  if (vData[i] >= double(threshold_dB_Audio) ) 
  //  {
    //  Serial.print("Threshhold ");
    //  Serial.println(vData[i], 4);
   //   transmitEventData();
  //  }
 
    //
  }
  Serial.println();
}

// Reads BME280 for data, if data is new then the flag is set
void EnvironmentalData()
{//Serial.println("read data");
  if((unsigned long)(current_time - last_BME_Poll) >= last_BME_interval)
  {
    // This manages clock rollover and compensates for drift
    last_BME_Poll = current_time - ((unsigned long)(current_time - last_BME_Poll) - last_BME_interval);
   //Serial.println("read data");
    float T = bme.readTemperature();
    float P = bme.readPressure() / 100.0F;
    float H = bme.readHumidity();
   
    sensorDataRingBuffer[(firstIndexRingBuffer++)%sensorBufferSize] = {T,P,H}; 
    
  }
}

// Reads BNO055 for data, if data is new then the flag is set
void IMU_Data()
{//Serial.println("read data");
  if((unsigned long)(current_time - last_BNO_Poll) >= last_BNO_interval)
  {
    // This manages clock rollover and compensates for drift
    last_BNO_Poll = current_time - ((unsigned long)(current_time - last_BNO_Poll) - last_BNO_interval);
   //Serial.println("read data");
            
    /* Get a new sensor event */
    sensors_event_t event;
    bno.getEvent(&event);
 
    sensor_BNO_RingBuffer[(firstIndex_BNO_RingBuffer++)%sensor_BNO_BufferSize] = {event.acceleration.x, event.acceleration.y, event.acceleration.z};


  }
}

// Send last 5 seconds of environmental data on threshold detect
// This would be replaced with AWS transmit code
void transmitEventData() {
  u_int temp = firstIndexRingBuffer - 50;
  while (temp != firstIndexRingBuffer)
  {
    BME280Data t_data = sensorDataRingBuffer[(temp++)%sensorBufferSize];

        Serial.print(t_data.Temperature);
        Serial.println(" *C");
    

   
      Serial.print("Pressure = ");
      Serial.print(t_data.Pressure);
      Serial.println(" hPa");
      Serial.print("Pressure = ");
      Serial.print(t_data.Pressure*0.0295299);
      Serial.println(" inhg");
    
    
   
      Serial.print("Humidity = ");
      Serial.print(t_data.Humidity);
      Serial.println(" %");
    
  }
}

// Send last 5 seconds of environmental data on threshold detect
// This would be replaced with AWS transmit code
void transmit_IMU_Data() {
  u_int temp = firstIndex_BNO_RingBuffer - 100;
  while (temp != firstIndex_BNO_RingBuffer)
  {
    BNOData t_data = sensor_BNO_RingBuffer[(temp++)%sensorBufferSize];
    Serial.print("X: ");
    Serial.print(t_data.Accel_X, 4);
    Serial.print("\tY: ");
    Serial.print(t_data.Accel_Y, 4);
    Serial.print("\tZ: ");
    Serial.print(t_data.Accel_Z, 4);
    Serial.println("");
  }
}

// Display BME280 data if it is new data
void printValues() {
  // Serial.println("Display");
  if((unsigned long)(current_time - last_time_display) >= last_time_interval)
  {
    // This manages clock rollover and compensates for drift
    last_time_display = current_time - ((unsigned long)(current_time - last_time_display) - last_time_interval);
 
    transmitEventData();
    transmit_IMU_Data();
  }
}


bool connectToAWSIoT()
{
	// Random client ID to connect AWS IoT
	String clientID = String::format("mindfulhost_%04x", random(0xFFFF));
	//
	client.enableTls(caPem, sizeof(caPem), certPem, sizeof(certPem), keyPem, sizeof(keyPem));
  
	// connect to the server
  while (client.connect(clientID))
  {
    Serial.println("Connect Failed");
    delay(1000);
  }
	//Serial.println(client.connect(clientID));
	if (client.isConnected()) {
			Serial.println("client connected");

			return true;
	}
	return false;
}

// recieve message
void callback(char* topic, byte* payload, unsigned int length) {
    char p[length + 1];
    memcpy(p, payload, length);
    p[length] = NULL;
    String message(p);
    
    Serial.println(message);
    delay(1000);
}
/*
void updateThingShadow_DTC()
{
  
  //char payload[2048];
  DynamicJsonDocument jsonBuffer(4096);
  jsonBuffer["uuid"] = "Some User ID";
  jsonBuffer["timestamp"] = Time.format(Time.now(), "%Y-%m-%d %H:%M:%S");
//  JsonArray envSensor = jsonBuffer.createNestedArray("envSensor");
 // JsonObject decibel = envSensor.createNestedObject("decibel");
 // JsonArray vibration =  envSensor.createNestedArray("vibration");
 // JsonArray audio =  envSensor.createNestedArray("audio");
 // jsonBuffer["configuration"] = "configuration";
 // JsonObject name = configuration.createNestedObject("name");
  //JsonObject location = configuration.createNestedObject("location");
 // JsonObject audioThreshold = configuration.createNestedObject("audioThreshold");

  u_int temp = firstIndexRingBuffer - 50;
  while (temp != firstIndexRingBuffer)
  {
    BME280Data t_data = sensorDataRingBuffer[(temp++)%sensorBufferSize];
    //envSensor.add({t_data.Humidity, t_data.Temperature, t_data.Pressure});
  }

  temp = firstIndex_BNO_RingBuffer - 100;
  while (temp != firstIndex_BNO_RingBuffer)
  {
    BNOData t_data = sensor_BNO_RingBuffer[(temp++)%sensorBufferSize];
    //vibration.add({t_data.Accel_X,t_data.Accel_Y,t_data.Accel_Z});
  }

//	root.printTo(payload);
//  if(client.isConnected())
//    client.publish(shadowUpdateTopic.c_str(), payload);
}
*/

/*
void publishInfo()
{

  char buff[512];
  DynamicJsonBuffer jsonBuffer(1024);
	JsonObject& root = jsonBuffer.createObject();
  root["sensor_id"] = "This sensor ID";
	root["timestamp"] = Time.now();
  

  root["start_time"] = start_time; // timestamp
  root["end_time"] = Time.now(); // timestamp
  // Publishing mileage data while driving
  root["mileage"] = storage.mileage;
	root.printTo(buff); //

	//
	String pubTopic = String::format("connectedcar/trip/%s", currObd.VINNumber);
	client.publish(pubTopic.c_str(), buff);
  
  "ts":"2017-03-11T14:55:22Z",
  "vin":""WDDSJ4GB0EN141171",
  "state":"driving",
  "lng":"35.94",
  "lat":"-86.66".
  "accel_x_axis": "23423434",
  "accel_y_axis": "23423434",
  "accel_z_axis": "23423434"
  
}
*/

#10

I’m having a really hard time following that code. Is that all one source file, or multiple? There are #include statements scattered throughout, and some things in </> code blocks while others are not. It’s not clear what is source, what is output, and how this is structured. Lack of indentation also makes it harder to read. Could you describe in more detail what you are publishing, how it’s structured, and how much data you are pushing in each publish?


#11

This is currently one file. There is alot of room for cleanup. The main logic does this:

  1. reads temp, humidity, pressure
  2. audio which is converted to samples using FFT
  3. connects to aws iot using mqtt tls
  4. creates json object with delta values
  5. updates aws iot thing shadow with a json payload

it turns out to be 16700 bytes every 1/32 seconds with FFT


#12

You’re trying to publish 16700 bytes 32 times per second? Did I read that right? That’s ~4.275 megabits/second. The absolute limit for Cat-M1 communication is about 375 kilobits/second.