main.cc

Go to the documentation of this file.

/* vim: set sw=2 expandtab tw=80: */
 
#include <libents/controller/controller.h>
#include <libents/controller/modules/microsd.h>
#include <libents/proto/sensor.h>
#include <libents/storage/fifo.h>
#include <libents/user_config.h>
#include <libtock-sync/services/alarm.h>
#include <libtock/kernel/ipc.h>
#include <libtock/tock.h>
#include <stdio.h>
#include <ulog.h>
 
#include "lorawan.h"
#include "user_config.h"
 
#define LOG_LINE_SIZE 240
 
typedef struct {
  int bytes;
  int total;
  int failed;
  int meas;
  int heartbeats;
} upload_stats;
 
upload_stats stats = {};
 
// buffer to store measurements
static uint8_t meas_buffer[256] = {};
static uint8_t meas_buffer_length = 0;
 
// buffer for user config
static uint8_t uc_buffer[256] = {};
static uint8_t uc_buffer_length = 0;
 
static uint8_t cmd = 0;
 
// last pid command
static int last_pid = 0;
 
static bool has_data = false;
 
static bool network_ready = false;
 
static int timesync_retry_delay_ms = 10000;
 
static int upload_interval = 60000;
 
static const int userconfig_timeout_ms = 300 * 1000;
 
static void ipc_callback(int pid, int len, int buf, void* ud);
 
static int get_payload(uint8_t* buffer, int size);
 
void ulog_prefix_handler(ulog_event* ev, char* prefix, size_t prefix_size);
 
void ulog_prefix_handler(ulog_event* ev, char* prefix, size_t prefix_size) {
  (void)ev;
 
  snprintf(prefix, prefix_size, "Core\t");
}
 
static void microsd_log_output(ulog_event* ev, void* arg) {
  (void)arg;
 
  // Guard against recursion. If anything reached from ControllerMicroSDLog()
  // ever logs, that log would re-enter this handler and recurse forever.
  static bool busy = false;
  if (busy) {
    return;
  }
  busy = true;
 
  // static to keep 128 bytes off the stack, matching microlog's own examples.
  // Safe despite being shared state because the busy guard above prevents
  // re-entry.
  static char line[LOG_LINE_SIZE];
  if (ulog_event_to_cstr(ev, line, sizeof(line)) == ULOG_STATUS_OK) {
    ControllerMicroSDLog(line, "core");
  }
 
  busy = false;
}
 
int main(void) {
  // Setup logging level and prefix
  ulog_output_level_set_all(ULOG_LEVEL_TRACE);
  ulog_prefix_set_fn(ulog_prefix_handler);
  ulog_info("=== Core App Initialized ===");
 
  // Load bytes into userconfig buffer
  //
  // Yes I am casting uint8_t to a uint16_t and it could overwrite the data
  // buffer. I know at the time of writing this that the user config stays
  // under 256 bytes as defined by protobuf.
  UserConfigStatus uc_status =
      UserConfigBytes(uc_buffer, (uint16_t*)&uc_buffer_length);
  if (uc_status != USERCONFIG_OK) {
    ulog_error("Could not load user config.");
  }
 
  // Option to print bytes to the buffer
  // printf("uc_buffer[%u]:", uc_buffer_length);
  // for (uint8_t i = 0; i < uc_buffer_length; i++) {
  //  printf(" %02x", uc_buffer[i]);
  //}
  // printf("\n\n");
 
  // start service after connected
  ipc_register_service_callback("org.ents.core", ipc_callback, NULL);
 
  // Print warning when using TEST_USER_CONFIG
#ifdef TEST_USER_CONFIG
  ulog_warn("TEST_USER_CONFIG is enabled!\n");
#endif  // TEST_USER_CONFIG
 
  // Initialize controller interface
  ControllerInit();
 
  // Setup sd card output
  ulog_output_id sd_output =
      ulog_output_add(microsd_log_output, NULL, ULOG_LEVEL_INFO);
  if (sd_output == ULOG_OUTPUT_INVALID) {
    // Almost always means ULOG_BUILD_EXTRA_OUTPUTS was not set when microlog
    // was compiled, see embedded/external/microlog/Makefile.
    ulog_error("Could not register microSD log output");
    return 1;
  }
 
  // Get update configuration from server
  UserConfigUpdateFromServer();
 
  // Reset esp32 and start webserver
  ControllerDeviceReset();
  UserConfigStart(userconfig_timeout_ms);
 
  // return codes
  int ret = 0;
 
  // Initialize LoRaWAN
 
  ret = lorawan_init();
  if (ret < 0) {
    return ret;
  }
 
  ret = lorawan_join();
  if (ret < 0) {
    return ret;
  }
 
  while (1) {
    ret = lorawan_timesync();
    if (ret < 0) {
      ulog_info("Retrying in %d ms", timesync_retry_delay_ms);
      libtocksync_alarm_delay_ms(timesync_retry_delay_ms);
    } else {
      break;
    }
  }
 
  network_ready = true;
 
  while (1) {
    ulog_trace("main loop");
 
    ret = libtocksync_alarm_yield_for_with_timeout(&has_data, upload_interval);
    if (ret == 0) {
      ulog_info("Loop triggered with condition");
    } else if (ret == -1) {
      ulog_info("Loop triggered with timeout");
    } else {
      ulog_warn("Unknown behavior with yield_for_with_timeout (error: %d)",
                ret);
    }
 
    //
    // Save data on matched command
    //
 
    if (has_data && cmd == 2) {
      // print out bytes
      //  Get number of bytes in buffer
      ulog_info("Received %d bytes:", meas_buffer_length);
      // for (int i=1; i < meas_buffer_length; i++) {
      //   printf("%x ", meas_buffer[i]);
      // }
      // printf("\n");
 
      // store in buffer
      ret = fifo_put(meas_buffer, meas_buffer_length);
      if (ret < 0) {
        ulog_error("Could not store measurement in buffer");
      }
      stats.meas++;
 
      // indicate data has been processed and trigger client
      has_data = false;
      ipc_notify_client(last_pid);
 
      // skip uploads if received ipc call
      continue;
    }
 
    //
    // Upload data
    //
 
    uint16_t meas_in_buffer = fifo_buffer_len();
 
    if (meas_in_buffer > 1) {
      // batch into minium of 4 measurements
      while (meas_in_buffer > 1) {
        ulog_debug("Buffer has %d measurements", meas_in_buffer);
 
        // format payload
        uint8_t buffer[60] = {};
        int len = get_payload(buffer, sizeof(buffer));
        if (len != 0) {
          ulog_debug("Uploading %d bytes", len);
 
          stats.total++;
          ret = lorawan_upload(buffer, len);
          if (ret < 0) {
            stats.failed++;
            ulog_error("Could not upload with LoRaWAN (error: %d)", ret);
          } else {
            ulog_debug("Uploaded %d bytes with LoRaWAN.");
 
            stats.bytes += len;
          }
        }
 
        meas_in_buffer = fifo_buffer_len();
      }
    } else {
      stats.total++;
 
      ulog_debug("Sending heartbeat");
 
      ret = lorawan_heartbeat();
      if (ret < 0) {
        stats.failed++;
        ulog_error("Error sending heartbeat (error: %d)");
      } else {
        stats.heartbeats++;
        ulog_debug("Heartbeat sent");
      }
    }
 
    //
    // print stats
    //
    if (!(stats.total % 6)) {
      ulog_info(
          "total uploads: %d  failed uploads: %d  measurements: %d  bytes: "
          "%d  heartbeats: %d",
          stats.total, stats.failed, stats.meas, stats.bytes, stats.heartbeats);
    }
  }
}
 
static void ipc_callback(int pid, int len, int buf, void* ud) {
  (void)len;
  (void)ud;
 
  ulog_trace("ipc_callabck");
 
  uint8_t* buffer = (uint8_t*)buf;
 
  // payload format
  cmd = buffer[0];
  uint8_t* length = buffer + 1;
  uint8_t* data = buffer + 2;
 
  // Reply with userconfig when requested
  if (cmd == 1) {
    ulog_trace("user config command");
 
    // copy from user config to data
    memcpy(data, uc_buffer, uc_buffer_length);
    *length = uc_buffer_length;
 
    // trigger client
    ipc_notify_client(pid);
 
    // Store measurements into buffer
  } else if (cmd == 2) {
    ulog_trace("measurement command");
 
    // copy data to buffer
    meas_buffer_length = *length;
    memcpy(meas_buffer, data, *length);
 
    // store last pid so it can be triggered
    // basically hold sensors in an interrupt until we store the measurement
    last_pid = pid;
    has_data = true;
 
    // ready command
  } else if (cmd == 3) {
    ulog_trace("ready command");
 
    buffer[1] = (uint8_t)network_ready;
 
    // trigger client
    ipc_notify_client(pid);
 
    // Catch all other commands
  } else {
    ulog_error("IPC command %d not implemented.", buffer[0]);
  }
}
 
static int get_payload(uint8_t* buffer, int size) {
  ulog_trace("get_payload");
 
  // return codes
  int ret = 0;
 
  int len = 0;
 
  Metadata meta = {};
  SensorMeasurement meas[8] = {};
 
  uint16_t length = fifo_buffer_len();
 
  uint16_t i = 0;
  for (i = 0; i < length; i++) {
    ret = fifo_peek(i, buffer, (uint8_t*)&len);
    if (ret < 0) {
      ulog_error("Could not read from buffer (error: %d)", ret);
      continue;
    }
    ulog_debug("Read %d bytes from buffer", len);
 
    // decode measuremnet
    ret = DecodeSensorMeasurement(buffer, len, &meas[i]);
    if (ret < 0) {
      ulog_error("Could not decode measurement (error %d), malformed data?",
                 ret);
      // TODO: Removed failing measurement. Can't use drop, need to remove
      // specific index.
      continue;
    }
 
    ret = RepeatedSensorMeasurementsSize(meta, meas, i + 1, (size_t*)&len);
    if (ret < 0) {
      ulog_error("Could not find size of payload (error %d)", ret);
      continue;
    }
    ulog_debug("Size of repeated sensor measurements: %d", len);
 
    // early stop when length exceeds size of buffer
    if (len > size) {
      ulog_debug("Over sensors size limit of %d. Removing last measurement.",
                 size);
 
      i--;
      break;
    }
 
    PrintSensorMeasurement(&meas[i]);
  }
 
  ret = EncodeRepeatedSensorMeasurements(meta, meas, i + 1, buffer, size,
                                         (size_t*)&len);
  if (ret < 0) {
    ulog_error("Could not encode %d repeated measurements (error %d)", i, ret);
    return 0;
  }
 
  // Clear uploaded measurements
  // ulog_debug("%d measurements to drop");
  for (uint16_t j = 0; j < i; j++) {
    ret = fifo_drop();
    if (ret < 0) {
      ulog_error("Could not remove measurement index %d from buffer", j);
    }
  }
  return len;
}