text:
#define VERSION_MAJOR 5
#define VERSION_MINOR 0
#define VERSION_SUB 0

#define STR_HELPER(x) #x
#define STR(x) STR_HELPER(x)

#define BEACON_PORT 50000
#define SERVER_PORT 8080
#define SERVER_STRING                                                          \
  "FITSWebQL v" STR(VERSION_MAJOR) "." STR(VERSION_MINOR) "." STR(VERSION_SUB)
#define VERSION_STRING "SV2019-11-14.0"
#define WASM_STRING "WASM2019-02-08.1"

#include <zlib.h>

/* CHUNK is the size of the memory chunk used by the zlib routines. */

#define CHUNK 0x4000
#define windowBits 15
#define GZIP_ENCODING 16

/* The following macro calls a zlib routine and checks the return
   value. If the return value ("status") is not OK, it prints an error
   message and exits the program. Zlib's error statuses are all less
   than zero. */

#define CALL_ZLIB(x)                                                           \
  {                                                                            \
    int status;                                                                \
    status = x;                                                                \
    if (status < 0) {                                                          \
      fprintf(stderr, "%s:%d: %s returned a bad status of %d.\n", __FILE__,    \
              __LINE__, #x, status);                                           \
      /*exit(EXIT_FAILURE);*/                                                  \
    }                                                                          \
  }

#include <cstdint>
#include <fcntl.h>
#include <pwd.h>
#include <string.h>
#include <sys/mman.h>
#include <sys/stat.h>
#include <sys/types.h>
#include <unistd.h>

#if !defined(__APPLE__) || !defined(__MACH__)
#include <bsd/string.h>
#endif

static bool is_gzip(const char *filename) {
  int fd = open(filename, O_RDONLY);

  if (fd == -1)
    return false;

  bool ok = true;
  uint8_t header[10];

  // try to read the first 10 bytes
  ssize_t bytes_read = read(fd, header, 10);

  // test for magick numbers and the deflate compression type
  if (bytes_read == 10) {
    if (header[0] != 0x1f || header[1] != 0x8b || header[2] != 0x08)
      ok = false;
  } else
    ok = false;

  close(fd);

  return ok;
}

#include "fits.hpp"
#include "json.h"
#include "mongoose.h"

#include <atomic>
#include <iostream>
#include <mutex>
#include <set>
#include <shared_mutex>
#include <sstream>
#include <thread>

#include <boost/algorithm/string.hpp>
#include <boost/algorithm/string/predicate.hpp>

#include <curl/curl.h>
#include <sqlite3.h>

#ifndef LOCAL
#include <pgsql/libpq-fe.h>

#define FITSHOME "/home"
#define JVO_HOST "localhost"
#define JVO_USER "jvo"

PGconn *jvo_db_connect(std::string db) {
  PGconn *jvo_db = NULL;

  std::string conn_str =
      "dbname=" + db + " host=" + JVO_HOST + " user=" + JVO_USER;

  jvo_db = PQconnectdb(conn_str.c_str());

  if (PQstatus(jvo_db) != CONNECTION_OK) {
    fprintf(stderr, "PostgreSQL connection failed: %s\n",
            PQerrorMessage(jvo_db));
    PQfinish(jvo_db);
    jvo_db = NULL;
  } else
    printf("PostgreSQL connection successful.\n");

  return jvo_db;
}

std::string get_jvo_path(PGconn *jvo_db, std::string db, std::string table,
                         std::string data_id) {
  std::string path;

  std::string sql_str =
      "SELECT path FROM " + table + " WHERE data_id = '" + data_id + "';";

  PGresult *res = PQexec(jvo_db, sql_str.c_str());
  int status = PQresultStatus(res);

  if (PQresultStatus(res) == PGRES_TUPLES_OK) {
    path = std::string(FITSHOME) + "/" + db + "/";

    size_t pos = table.find(".");

    if (pos == std::string::npos)
      path += std::string((const char *)PQgetvalue(res, 0, 0));
    else
      path += boost::algorithm::to_upper_copy(table.substr(0, pos)) + "/" +
              std::string((const char *)PQgetvalue(res, 0, 0));
  }

  PQclear(res);

  return path;
}
#endif

sqlite3 *splat_db = NULL;
std::string home_dir;

std::unordered_map<std::string, std::shared_ptr<FITS>> DATASETS;
std::shared_mutex fits_mutex;

std::shared_ptr<FITS> get_dataset(std::string id) {
  std::shared_lock<std::shared_mutex> lock(fits_mutex);

  auto item = DATASETS.find(id);

  if (item == DATASETS.end())
    return nullptr;
  else
    return item->second;
}

void insert_dataset(std::string id, std::shared_ptr<FITS> fits) {
  std::lock_guard<std::shared_mutex> guard(fits_mutex);

  DATASETS.insert(std::pair(id, fits));
}

#ifdef CLUSTER
#include <czmq.h>

inline std::set<std::string> cluster;
inline std::shared_mutex cluster_mtx;

inline bool cluster_contains_node(std::string node) {
  std::shared_lock<std::shared_mutex> lock(cluster_mtx);

  if (cluster.find(node) == cluster.end())
    return false;
  else
    return true;
}

inline void cluster_insert_node(std::string node) {
  std::lock_guard<std::shared_mutex> guard(cluster_mtx);
  cluster.insert(node);
}

inline void cluster_erase_node(std::string node) {
  std::lock_guard<std::shared_mutex> guard(cluster_mtx);
  cluster.erase(node);
}

zactor_t *speaker = NULL;
zactor_t *listener = NULL;
std::thread beacon_thread;
std::atomic<bool> exiting(false);
#endif

/** Thread safe cout class
 * Exemple of use:
 *    PrintThread{} << "Hello world!" << std::endl;
 */
class PrintThread : public std::ostringstream {
public:
  PrintThread() = default;

  ~PrintThread() {
    std::lock_guard<std::mutex> guard(_mutexPrint);
    std::cout << this->str();
  }

private:
  static std::mutex _mutexPrint;
};

std::mutex PrintThread::_mutexPrint{};

#include <ipp.h>

void ipp_init() {
  const IppLibraryVersion *lib;
  IppStatus status;
  Ipp64u mask, emask;

  /* Init IPP library */
  ippInit();
  /* Get IPP library version info */
  lib = ippGetLibVersion();
  printf("%s %s\n", lib->Name, lib->Version);

  /* Get CPU features and features enabled with selected library level */
  /*status = ippGetCpuFeatures(&mask, 0);
  if (ippStsNoErr == status) {
    emask = ippGetEnabledCpuFeatures();
    printf("Features supported by CPU\tby IPP\n");
    printf("-----------------------------------------\n");
    printf("  ippCPUID_MMX        = ");
    printf("%c\t%c\t", (mask & ippCPUID_MMX) ? 'Y' : 'N',
           (emask & ippCPUID_MMX) ? 'Y' : 'N');
    printf("Intel(R) Architecture MMX technology supported\n");
    printf("  ippCPUID_SSE        = ");
    printf("%c\t%c\t", (mask & ippCPUID_SSE) ? 'Y' : 'N',
           (emask & ippCPUID_SSE) ? 'Y' : 'N');
    printf("Intel(R) Streaming SIMD Extensions\n");
    printf("  ippCPUID_SSE2       = ");
    printf("%c\t%c\t", (mask & ippCPUID_SSE2) ? 'Y' : 'N',
           (emask & ippCPUID_SSE2) ? 'Y' : 'N');
    printf("Intel(R) Streaming SIMD Extensions 2\n");
    printf("  ippCPUID_SSE3       = ");
    printf("%c\t%c\t", (mask & ippCPUID_SSE3) ? 'Y' : 'N',
           (emask & ippCPUID_SSE3) ? 'Y' : 'N');
    printf("Intel(R) Streaming SIMD Extensions 3\n");
    printf("  ippCPUID_SSSE3      = ");
    printf("%c\t%c\t", (mask & ippCPUID_SSSE3) ? 'Y' : 'N',
           (emask & ippCPUID_SSSE3) ? 'Y' : 'N');
    printf("Intel(R) Supplemental Streaming SIMD Extensions 3\n");
    printf("  ippCPUID_MOVBE      = ");
    printf("%c\t%c\t", (mask & ippCPUID_MOVBE) ? 'Y' : 'N',
           (emask & ippCPUID_MOVBE) ? 'Y' : 'N');
    printf("The processor supports MOVBE instruction\n");
    printf("  ippCPUID_SSE41      = ");
    printf("%c\t%c\t", (mask & ippCPUID_SSE41) ? 'Y' : 'N',
           (emask & ippCPUID_SSE41) ? 'Y' : 'N');
    printf("Intel(R) Streaming SIMD Extensions 4.1\n");
    printf("  ippCPUID_SSE42      = ");
    printf("%c\t%c\t", (mask & ippCPUID_SSE42) ? 'Y' : 'N',
           (emask & ippCPUID_SSE42) ? 'Y' : 'N');
    printf("Intel(R) Streaming SIMD Extensions 4.2\n");
    printf("  ippCPUID_AVX        = ");
    printf("%c\t%c\t", (mask & ippCPUID_AVX) ? 'Y' : 'N',
           (emask & ippCPUID_AVX) ? 'Y' : 'N');
    printf("Intel(R) Advanced Vector Extensions instruction set\n");
    printf("  ippAVX_ENABLEDBYOS  = ");
    printf("%c\t%c\t", (mask & ippAVX_ENABLEDBYOS) ? 'Y' : 'N',
           (emask & ippAVX_ENABLEDBYOS) ? 'Y' : 'N');
    printf("The operating system supports Intel(R) AVX\n");
    printf("  ippCPUID_AES        = ");
    printf("%c\t%c\t", (mask & ippCPUID_AES) ? 'Y' : 'N',
           (emask & ippCPUID_AES) ? 'Y' : 'N');
    printf("Intel(R) AES instruction\n");
    printf("  ippCPUID_SHA        = ");
    printf("%c\t%c\t", (mask & ippCPUID_SHA) ? 'Y' : 'N',
           (emask & ippCPUID_SHA) ? 'Y' : 'N');
    printf("Intel(R) SHA new instructions\n");
    printf("  ippCPUID_CLMUL      = ");
    printf("%c\t%c\t", (mask & ippCPUID_CLMUL) ? 'Y' : 'N',
           (emask & ippCPUID_CLMUL) ? 'Y' : 'N');
    printf("PCLMULQDQ instruction\n");
    printf("  ippCPUID_RDRAND     = ");
    printf("%c\t%c\t", (mask & ippCPUID_RDRAND) ? 'Y' : 'N',
           (emask & ippCPUID_RDRAND) ? 'Y' : 'N');
    printf("Read Random Number instructions\n");
    printf("  ippCPUID_F16C       = ");
    printf("%c\t%c\t", (mask & ippCPUID_F16C) ? 'Y' : 'N',
           (emask & ippCPUID_F16C) ? 'Y' : 'N');
    printf("Float16 instructions\n");
    printf("  ippCPUID_AVX2       = ");
    printf("%c\t%c\t", (mask & ippCPUID_AVX2) ? 'Y' : 'N',
           (emask & ippCPUID_AVX2) ? 'Y' : 'N');
    printf("Intel(R) Advanced Vector Extensions 2 instruction set\n");
    printf("  ippCPUID_AVX512F    = ");
    printf("%c\t%c\t", (mask & ippCPUID_AVX512F) ? 'Y' : 'N',
           (emask & ippCPUID_AVX512F) ? 'Y' : 'N');
    printf("Intel(R) Advanced Vector Extensions 3.1 instruction set\n");
    printf("  ippCPUID_AVX512CD   = ");
    printf("%c\t%c\t", (mask & ippCPUID_AVX512CD) ? 'Y' : 'N',
           (emask & ippCPUID_AVX512CD) ? 'Y' : 'N');
    printf("Intel(R) Advanced Vector Extensions CD (Conflict Detection) "
           "instruction set\n");
    printf("  ippCPUID_AVX512ER   = ");
    printf("%c\t%c\t", (mask & ippCPUID_AVX512ER) ? 'Y' : 'N',
           (emask & ippCPUID_AVX512ER) ? 'Y' : 'N');
    printf("Intel(R) Advanced Vector Extensions ER instruction set\n");
    printf("  ippCPUID_ADCOX      = ");
    printf("%c\t%c\t", (mask & ippCPUID_ADCOX) ? 'Y' : 'N',
           (emask & ippCPUID_ADCOX) ? 'Y' : 'N');
    printf("ADCX and ADOX instructions\n");
    printf("  ippCPUID_RDSEED     = ");
    printf("%c\t%c\t", (mask & ippCPUID_RDSEED) ? 'Y' : 'N',
           (emask & ippCPUID_RDSEED) ? 'Y' : 'N');
    printf("The RDSEED instruction\n");
    printf("  ippCPUID_PREFETCHW  = ");
    printf("%c\t%c\t", (mask & ippCPUID_PREFETCHW) ? 'Y' : 'N',
           (emask & ippCPUID_PREFETCHW) ? 'Y' : 'N');
    printf("The PREFETCHW instruction\n");
    printf("  ippCPUID_KNC        = ");
    printf("%c\t%c\t", (mask & ippCPUID_KNC) ? 'Y' : 'N',
           (emask & ippCPUID_KNC) ? 'Y' : 'N');
    printf("Intel(R) Xeon Phi(TM) Coprocessor instruction set\n");
  }*/
}

inline const char *check_null(const char *str) {
  if (str != NULL)
    return str;
  else
    return ""; //"\"\"" ;
};

static volatile sig_atomic_t s_received_signal = 0;
static const char *s_http_port = "8080";
static const int s_num_worker_threads = 4;
static unsigned long s_next_id = 0;
struct mg_mgr mgr;

static void signal_handler(int sig_num) {
  printf("Interrupt signal [%d] received.\n", sig_num);

  signal(sig_num, signal_handler);
  s_received_signal = sig_num;

#ifdef CLUSTER
  exiting = true;
#endif
}
static struct mg_serve_http_opts s_http_server_opts;
static sock_t sock[2];

static int is_websocket(const struct mg_connection *nc) {
  return nc->flags & MG_F_IS_WEBSOCKET;
}

void mg_http_send_error_keep_alive(struct mg_connection *nc, int code,
                                   const char *reason) {
  if (reason == NULL)
    reason = "N/A";

  mg_send_head(nc, code, strlen(reason), "Content-Type: text/plain");
  mg_send(nc, reason, strlen(reason));
}

struct molecules_request {
  double freq_start;
  double freq_end;
  bool compress;
};

// This info is passed to the worker thread
struct work_request {
  unsigned long
      conn_id; // needed to identify the connection where to send the reply
  // optionally, more data that could be required by worker
  char *dataId;
  struct molecules_request *req;
};

// This info is passed by the worker thread to mg_broadcast
struct work_result_err {
  unsigned long conn_id;
  int code;
};

// This info is passed by the worker thread to mg_broadcast
struct work_result_header {
  unsigned long conn_id;
  bool compress;
};

// This info is passed by the worker thread to mg_broadcast
struct work_result_chunk {
  unsigned long conn_id;
  char *chunk;
  size_t len;
};

static void on_error_complete(struct mg_connection *nc, int ev, void *ev_data) {
  (void)nc;
  (void)ev_data;

  if (ev == MG_EV_POLL) {
    struct work_result_err *res = (struct work_result_err *)ev_data;

    if (res != NULL) {
      printf("conn. id %zu, sending an error code %d; cross-check: %zu\n",
             res->conn_id, res->code, (unsigned long)nc->user_data);

      if (res->conn_id == (unsigned long)nc->user_data)
        mg_http_send_error_keep_alive(nc, res->code, NULL);
    }
  }
}

static void on_header_complete(struct mg_connection *nc, int ev,
                               void *ev_data) {
  (void)nc;
  (void)ev_data;

  if (ev == MG_EV_POLL) {
    struct work_result_header *res = (struct work_result_header *)ev_data;

    if (res != NULL) {
      printf("conn. id %zu, sending a chunked header, compression: %d; "
             "cross-check: %zu\n",
             res->conn_id, res->compress, (unsigned long)nc->user_data);

      if (res->conn_id == (unsigned long)nc->user_data) {
        if (res->compress)
          mg_send_head(nc, 200, -1,
                       "Content-Type: application/json\r\nCache-Control: "
                       "no-cache\r\nContent-Encoding: gzip");
        else
          mg_send_head(
              nc, 200, -1,
              "Content-Type: application/json\r\nCache-Control: no-cache");
      }
    }
  }
}

static void on_chunk_complete(struct mg_connection *nc, int ev, void *ev_data) {
  (void)nc;
  (void)ev_data;

  struct work_result_chunk *res = (struct work_result_chunk *)ev_data;

  if (ev == MG_EV_POLL) {
    if (res != NULL) {
      printf("conn. id %zu, sending a chunk of length %zu; cross-check: %zu\n",
             res->conn_id, res->len, (unsigned long)nc->user_data);

      if (res->conn_id == (unsigned long)nc->user_data)
        if (res->chunk != NULL) {
          mg_send_http_chunk(nc, res->chunk, res->len);
          free(res->chunk);
          res->chunk = NULL;
        } else
          mg_send_http_chunk(nc, "", 0);
    }
  }
}

void broadcast_error(unsigned long conn_id, int code) {
  struct work_result_err res = {conn_id, code};
  mg_broadcast(&mgr, on_error_complete, (void *)&res, sizeof(res));
}

void broadcast_header(unsigned long conn_id, bool compress) {
  struct work_result_header res = {conn_id, compress};
  mg_broadcast(&mgr, on_header_complete, (void *)&res, sizeof(res));
}

void broadcast_chunk(unsigned long conn_id, const char *data, size_t len) {
  char *chunk = NULL;

  if (data != NULL)
    chunk = (char *)malloc(len);

  struct work_result_chunk res = {conn_id, chunk, len};
  mg_broadcast(&mgr, on_chunk_complete, (void *)&res, sizeof(res));
}

struct MolecularStream {
  bool first;
  bool compress;
  unsigned long conn_id;
  z_stream z;
  unsigned char out[CHUNK];
  FILE *fp;
};

static int sqlite_callback(void *userp, int argc, char **argv,
                           char **azColName) {
  MolecularStream *stream = (MolecularStream *)userp;
  // static long counter = 0;
  // printf("sqlite_callback: %ld, argc: %d\n", counter++, argc);

  if (argc == 8) {
    /*printf("sqlite_callback::molecule:\t");
      for (int i = 0; i < argc; i++)
      printf("%s:%s\t", azColName[i], argv[i]);
      printf("\n");*/

    std::string json;

    if (stream->first) {
      stream->first = false;
      broadcast_header(stream->conn_id, stream->compress);

      json = "{\"molecules\" : [";
    } else
      json = ",";

    // json-encode a spectral line
    char *encoded;

    // species
    encoded = json_encode_string(check_null(argv[0]));
    json += "{\"species\" : " + std::string(check_null(encoded)) + ",";
    if (encoded != NULL)
      free(encoded);

    // name
    encoded = json_encode_string(check_null(argv[1]));
    json += "\"name\" : " + std::string(check_null(encoded)) + ",";
    if (encoded != NULL)
      free(encoded);

    // frequency
    json += "\"frequency\" : " + std::string(check_null(argv[2])) + ",";

    // quantum numbers
    encoded = json_encode_string(check_null(argv[3]));
    json += "\"quantum\" : " + std::string(check_null(encoded)) + ",";
    if (encoded != NULL)
      free(encoded);

    // cdms_intensity
    encoded = json_encode_string(check_null(argv[4]));
    json += "\"cdms\" : " + std::string(check_null(encoded)) + ",";
    if (encoded != NULL)
      free(encoded);

    // lovas_intensity
    encoded = json_encode_string(check_null(argv[5]));
    json += "\"lovas\" : " + std::string(check_null(encoded)) + ",";
    if (encoded != NULL)
      free(encoded);

    // E_L
    encoded = json_encode_string(check_null(argv[6]));
    json += "\"E_L\" : " + std::string(check_null(encoded)) + ",";
    if (encoded != NULL)
      free(encoded);

    // linelist
    encoded = json_encode_string(check_null(argv[7]));
    json += "\"list\" : " + std::string(check_null(encoded)) + "}";
    if (encoded != NULL)
      free(encoded);

    // printf("%s\n", json.c_str());

    if (stream->compress) {
      stream->z.avail_in = json.length();                // size of input
      stream->z.next_in = (unsigned char *)json.c_str(); // input char array

      do {
        stream->z.avail_out = CHUNK;      // size of output
        stream->z.next_out = stream->out; // output char array
        CALL_ZLIB(deflate(&stream->z, Z_NO_FLUSH));
        size_t have = CHUNK - stream->z.avail_out;

        if (have > 0) {
          // printf("ZLIB avail_out: %zu\n", have);
          if (stream->fp != NULL)
            fwrite((const char *)stream->out, sizeof(char), have, stream->fp);

          broadcast_chunk(stream->conn_id, (char *)stream->out, have);
        }
      } while (stream->z.avail_out == 0);
    } else
      broadcast_chunk(stream->conn_id, json.c_str(), json.length());
  }

  return 0;
}

void stream_molecules(unsigned long conn_id, double freq_start, double freq_end,
                      bool compress) {
  if (splat_db == NULL)
    return broadcast_error(conn_id, 500);

  char strSQL[256];
  int rc;
  char *zErrMsg = 0;

  snprintf(strSQL, 256,
           "SELECT * FROM lines WHERE frequency>=%f AND frequency<=%f;",
           freq_start, freq_end);
  printf("%s\n", strSQL);

  struct MolecularStream stream;
  stream.first = true;
  stream.compress = compress;
  stream.conn_id = conn_id;
  stream.fp = NULL; // fopen("molecules.txt.gz", "w");

  if (compress) {
    stream.z.zalloc = Z_NULL;
    stream.z.zfree = Z_NULL;
    stream.z.opaque = Z_NULL;
    stream.z.next_in = Z_NULL;
    stream.z.avail_in = 0;

    CALL_ZLIB(deflateInit2(&stream.z, Z_BEST_COMPRESSION, Z_DEFLATED,
                           windowBits | GZIP_ENCODING, 9, Z_DEFAULT_STRATEGY));
  }

  rc = sqlite3_exec(splat_db, strSQL, sqlite_callback, &stream, &zErrMsg);

  if (rc != SQLITE_OK) {
    fprintf(stderr, "SQL error: %s\n", zErrMsg);
    sqlite3_free(zErrMsg);
    return broadcast_error(conn_id, 500);
  }

  std::string chunk_data;

  if (stream.first)
    chunk_data = "{\"molecules\" : []}";
  else
    chunk_data = "]}";

  if (compress) {
    stream.z.avail_in = chunk_data.length();
    stream.z.next_in = (unsigned char *)chunk_data.c_str();

    do {
      stream.z.avail_out = CHUNK;     // size of output
      stream.z.next_out = stream.out; // output char array
      CALL_ZLIB(deflate(&stream.z, Z_FINISH));
      size_t have = CHUNK - stream.z.avail_out;

      if (have > 0) {
        // printf("Z_FINISH avail_out: %zu\n", have);
        if (stream.fp != NULL)
          fwrite((const char *)stream.out, sizeof(char), have, stream.fp);

        broadcast_chunk(conn_id, (char *)stream.out, have);
      }
    } while (stream.z.avail_out == 0);

    CALL_ZLIB(deflateEnd(&stream.z));

    if (stream.fp != NULL)
      fclose(stream.fp);
  } else
    broadcast_chunk(conn_id, chunk_data.c_str(), chunk_data.length());

  // end of chunked encoding
  broadcast_chunk(conn_id, NULL, 0);
}

void *worker_thread_proc(void *param) {
  struct mg_mgr *mgr = (struct mg_mgr *)param;
  struct work_request req = {0, NULL, NULL};

  while (s_received_signal == 0) {
    if (read(sock[1], &req, sizeof(req)) < 0) {
      // if(s_received_signal == 0)
      perror("Reading worker sock");
    } else {
      if (req.dataId != NULL) {
        printf("handling %s\n", req.dataId);

        // stream molecules
        if (req.req != NULL) {
          double freq_start = req.req->freq_start;
          double freq_end = req.req->freq_end;

          if (FPzero(freq_start) || FPzero(freq_end)) {
            // get the frequency range from the FITS header
            auto fits = get_dataset(req.dataId);

            if (fits == nullptr)
              broadcast_error(req.conn_id, 404);
            else {
              if (fits->has_error)
                broadcast_error(req.conn_id, 404);
              else {
                std::unique_lock<std::mutex> header_lck(fits->header_mtx);
                while (!fits->processed_header)
                  fits->header_cv.wait(header_lck);

                if (!fits->has_header)
                  broadcast_error(req.conn_id, 404);
                else {
                  if (fits->depth <= 1 || !fits->has_frequency)
                    broadcast_error(req.conn_id, 501);
                  else
                    // extract the freq. range
                    fits->get_frequency_range(freq_start, freq_end);
                }
              }
            }
          }

          // process the response
          PrintThread{} << "get_molecules(" << req.dataId << ", " << freq_start
                        << " GHz," << freq_end << " GHz)" << std::endl;

          if (!FPzero(freq_start) && !FPzero(freq_end))
            stream_molecules(req.conn_id, freq_start, freq_end, compress);
          else
            broadcast_error(req.conn_id, 501);
        }

        free(req.dataId);
        req.dataId = NULL;
      }

      if (req.req != NULL) {
        free(req.req);
        req.req = NULL;
      }
    }
  }

  printf("worker_thread_proc terminated.\n");
  return NULL;
}

#ifdef LOCAL
static void get_directory(struct mg_connection *nc, const char *dir) {
  printf("get_directory(%s)\n", check_null(dir));

  struct dirent **namelist = NULL;
  int i, n;

  n = scandir(dir, &namelist, 0, alphasort);

  std::ostringstream json;

  char *encoded = json_encode_string(check_null(dir));

  json << "{\"location\" : " << check_null(encoded) << ", \"contents\" : [";

  if (encoded != NULL)
    free(encoded);

  bool has_contents = false;

  if (n < 0) {
    perror("scandir");

    json << "]}";
  } else {
    for (i = 0; i < n; i++) {
      // printf("%s\n", namelist[i]->d_name);

      char pathname[1024];

      sprintf(pathname, "%s/%s", dir, check_null(namelist[i]->d_name));

      struct stat64 sbuf;

      int err = stat64(pathname, &sbuf);

      if (err == 0) {
        char last_modified[255];

        struct tm lm;
        localtime_r(&sbuf.st_mtime, &lm);
        strftime(last_modified, sizeof(last_modified) - 1,
                 "%a, %d %b %Y %H:%M:%S %Z", &lm);

        size_t filesize = sbuf.st_size;

        if (S_ISDIR(sbuf.st_mode) && namelist[i]->d_name[0] != '.') {
          char *encoded = json_encode_string(check_null(namelist[i]->d_name));

          json << "{\"type\" : \"dir\", \"name\" : " << check_null(encoded)
               << ", \"last_modified\" : \"" << last_modified << "\"},";
          has_contents = true;

          if (encoded != NULL)
            free(encoded);
        }

        if (S_ISREG(sbuf.st_mode)) {
          const std::string filename = std::string(namelist[i]->d_name);
          const std::string lower_filename =
              boost::algorithm::to_lower_copy(filename);

          // if(!strcasecmp(get_filename_ext(check_null(namelist[i]->d_name)),
          // "fits"))
          if (boost::algorithm::ends_with(lower_filename, ".fits") ||
              boost::algorithm::ends_with(lower_filename, ".fits.gz")) {
            char *encoded = json_encode_string(check_null(namelist[i]->d_name));

            json << "{\"type\" : \"file\", \"name\" : " << check_null(encoded)
                 << ", \"size\" : " << filesize << ", \"last_modified\" : \""
                 << last_modified << "\"},";
            has_contents = true;

            if (encoded != NULL)
              free(encoded);
          }
        }
      } else
        perror("stat64");

      free(namelist[i]);
    };

    // overwrite the the last ',' with a list closing character
    if (has_contents)
      json.seekp(-1, std::ios_base::end);

    json << "]}";
  };

  if (namelist != NULL)
    free(namelist);

  mg_send_head(nc, 200, json.tellp(),
               "Content-Type: application/json\r\nCache-Control: no-cache");
  mg_send(nc, json.str().c_str(), json.tellp());
}
#endif

static void http_fits_response(struct mg_connection *nc,
                               std::vector<std::string> datasets,
                               bool composite, bool has_fits) {
  std::string html =
      "<!DOCTYPE html>\n<html>\n<head>\n<meta charset=\"utf-8\">\n";
  html.append(
      "<link href=\"https://fonts.googleapis.com/css?family=Inconsolata\" "
      "rel=\"stylesheet\"/>\n");
  html.append(
      "<link href=\"https://fonts.googleapis.com/css?family=Material+Icons\" "
      "rel=\"stylesheet\"/>\n");
  html.append("<script src=\"https://d3js.org/d3.v5.min.js\"></script>\n");
  html.append("<script "
              "src=\"https://cdn.jsdelivr.net/gh/jvo203/fits_web_ql/htdocs/"
              "fitswebql/reconnecting-websocket.js\"></script>\n");
  html.append("<script "
              "src=\"//cdnjs.cloudflare.com/ajax/libs/numeral.js/2.0.6/"
              "numeral.min.js\"></script>\n");
  html.append("<script "
              "src=\"https://cdn.jsdelivr.net/gh/jvo203/fits_web_ql/htdocs/"
              "fitswebql/ra_dec_conversion.js\"></script>\n");
  html.append("<script "
              "src=\"https://cdn.jsdelivr.net/gh/jvo203/fits_web_ql/htdocs/"
              "fitswebql/sylvester.js\"></script>\n");
  html.append("<script "
              "src=\"https://cdn.jsdelivr.net/gh/jvo203/fits_web_ql/htdocs/"
              "fitswebql/shortcut.js\"></script>\n");
  html.append("<script "
              "src=\"https://cdn.jsdelivr.net/gh/jvo203/fits_web_ql/htdocs/"
              "fitswebql/colourmaps.js\"></script>\n");
  html.append("<script "
              "src=\"https://cdn.jsdelivr.net/gh/jvo203/fits_web_ql/htdocs/"
              "fitswebql/lz4.min.js\"></script>\n");
  html.append("<script "
              "src=\"https://cdn.jsdelivr.net/gh/jvo203/fits_web_ql/htdocs/"
              "fitswebql/marchingsquares-isocontours.min.js\"></script>\n");
  html.append("<script "
              "src=\"https://cdn.jsdelivr.net/gh/jvo203/fits_web_ql/htdocs/"
              "fitswebql/marchingsquares-isobands.min.js\"></script>\n");

  // hevc wasm decoder
  html.append("<script "
              "src=\"https://cdn.jsdelivr.net/gh/jvo203/fits_web_ql/htdocs/"
              "fitswebql/hevc_" WASM_STRING ".js\"></script>\n");
  html.append(R"(<script>
        Module.onRuntimeInitialized = async _ => {
            api = {
                hevc_init: Module.cwrap('hevc_init', '', []),
                hevc_destroy: Module.cwrap('hevc_destroy', '', []),
                hevc_decode_nal_unit: Module.cwrap('hevc_decode_nal_unit', 'number', ['number', 'number', 'number', 'number', 'number', 'number', 'number', 'string']),
            };
        };
    </script>)");

  // bootstrap
  html.append(
      "<meta name=\"viewport\" content=\"width=device-width, initial-scale=1, "
      "user-scalable=no, minimum-scale=1, maximum-scale=1\">\n");
  html.append("<link rel=\"stylesheet\" "
              "href=\"https://maxcdn.bootstrapcdn.com/bootstrap/3.3.7/css/"
              "bootstrap.min.css\">\n");
  html.append("<script "
              "src=\"https://ajax.googleapis.com/ajax/libs/jquery/3.1.1/"
              "jquery.min.js\"></script>\n");
  html.append("<script "
              "src=\"https://maxcdn.bootstrapcdn.com/bootstrap/3.3.7/js/"
              "bootstrap.min.js\"></script>\n");

  // FITSWebQL main JavaScript + CSS
  html.append("<script src=\"fitswebql.js?" VERSION_STRING "\"></script>\n");
  html.append("<link rel=\"stylesheet\" href=\"fitswebql.css?" VERSION_STRING
              "\"/>\n");

  // HTML content
  html.append("<title>FITSWebQL</title></head><body>\n");
  html.append("<div id='votable' style='width: 0; height: 0;' data-va_count='" +
              std::to_string(datasets.size()) + "' ");

  if (datasets.size() == 1)
    html.append("data-datasetId='" + datasets[0] + "' ");
  else {
    for (int i = 0; i < datasets.size(); i++)
      html.append("data-datasetId" + std::to_string(i + 1) + "='" +
                  datasets[i] + "' ");

    if (composite && datasets.size() <= 3)
      html.append("data-composite='1' ");
  }

  html.append("data-root-path='/" + std::string("fitswebql") +
              "/' data-server-version='" + VERSION_STRING +
              "' data-server-string='" + SERVER_STRING +
              "' data-server-mode='" + "SERVER" + "' data-has-fits='" +
              std::to_string(has_fits) + "'></div>\n");

#ifdef PRODUCTION
  html.append(R"(<script>
        var WS_SOCKET = 'wss://';
        </script>)");
#else
  html.append(R"(<script>
        var WS_SOCKET = 'ws://';
        </script>)");
#endif

  // the page entry point
  html.append(R"(<script>
        const golden_ratio = 1.6180339887;
        var ALMAWS = null ;
        var wsVideo = null ;
        var wsConn = null ;
        var firstTime = true ;
        var has_image = false ;
        var PROGRESS_VARIABLE = 0.0 ;
        var PROGRESS_INFO = '' ;
        var RESTFRQ = 0.0 ;
        var USER_SELFRQ = 0.0 ;
        var USER_DELTAV = 0.0 ;
        var ROOT_PATH = '/fitswebql/' ;
        var idleResize = -1;
        window.onresize = resizeMe;
        window.onbeforeunload = function() {
            if(wsConn != null)
            {
                for(let i=0;i<va_count;i++)
                    wsConn[i].close();
            }

            if(wsVideo != null)
                wsVideo.close();
        };
        mainRenderer();
    </script>)");

  html.append("</body></html>");

  mg_send_head(nc, 200, html.size(),
               "Content-Type: text/html\r\nCache-Control: no-cache");
  mg_send(nc, html.c_str(), html.size());
}

static void execute_fits(struct mg_connection *nc, const char *dir,
                         const char *ext, const char *db, const char *table,
                         std::vector<std::string> datasets, bool composite,
                         const char *flux) {
  bool has_fits = true;

#ifndef LOCAL
  PGconn *jvo_db = NULL;

  if (strcmp(db, "") != 0)
    jvo_db = jvo_db_connect(db);
#endif

  int va_count = datasets.size();

  for (auto const &data_id : datasets) {
    auto item = get_dataset(data_id);

    if (item == nullptr) {
      // set has_fits to false and load the FITS dataset
      has_fits = false;
      std::shared_ptr<FITS> fits(new FITS(data_id, flux));

      insert_dataset(data_id, fits);

      std::string path;

      if (strcmp(dir, "") != 0 && strcmp(ext, "") != 0)
        path = std::string(dir) + "/" + data_id + "." + std::string(ext);

#ifndef LOCAL
      if (jvo_db != NULL && strcmp(table, "") != 0)
        path = get_jvo_path(jvo_db, db, table, data_id);
#endif

      if (path != "") {
        bool is_compressed = is_gzip(path.c_str());

        // load FITS data in a separate thread
        std::thread(&FITS::from_path_zfp, fits, path, is_compressed,
                    std::string(flux), va_count)
            .detach();
      } else {
        // the last resort
        std::string url = std::string("http://") + JVO_FITS_SERVER +
                          ":8060/skynode/getDataForALMA.do?db=" + JVO_FITS_DB +
                          "&table=cube&data_id=" + data_id + "_00_00_00";

        // download FITS data from a URL in a separate thread
        std::thread(&FITS::from_url, fits, url, std::string(flux), va_count)
            .detach();
      }
    } else {
      has_fits = has_fits && item->has_data;
      item->update_timestamp();
    }
  }

#ifndef LOCAL
  if (jvo_db != NULL)
    PQfinish(jvo_db);
#endif

  PrintThread{} << "has_fits: " << has_fits << std::endl;

  return http_fits_response(nc, datasets, composite, has_fits);
}

static void get_spectrum(struct mg_connection *nc, std::shared_ptr<FITS> fits) {
  std::ostringstream json;

  fits->to_json(json);

  if (json.tellp() > 0) {
    mg_send_head(nc, 200, json.tellp(),
                 "Content-Type: application/json\r\nCache-Control: no-cache");
    mg_send(nc, json.str().c_str(), json.tellp());
  } else
    mg_http_send_error_keep_alive(nc, 501, "Not Implemented");
}

static void ev_handler(struct mg_connection *nc, int ev, void *ev_data) {
  (void)nc;
  (void)ev_data;

  switch (ev) {
  case MG_EV_ACCEPT:
    break;
  case MG_EV_WEBSOCKET_HANDSHAKE_REQUEST: {
    struct http_message *hm = (struct http_message *)ev_data;
    printf("WEBSOCKET URI:\t%.*s\n", (int)hm->uri.len, hm->uri.p);
    break;
  }
  case MG_EV_WEBSOCKET_FRAME: {
    struct websocket_message *wm = (struct websocket_message *)ev_data;

    if (wm->data != NULL && wm->size > 0) {
      printf("[WS]:\t%.*s\n", (int)wm->size, wm->data);
    }

    break;
  }
  case MG_EV_HTTP_REQUEST: {
    struct http_message *hm = (struct http_message *)ev_data;
    printf("URI:\t%.*s\n", (int)hm->uri.len, hm->uri.p);

#ifdef LOCAL
    // get_directory
    if (strnstr(hm->uri.p, "/get_directory", hm->uri.len) != NULL) {
      char dir[1024] = "";
      struct mg_str query = hm->query_string;

      if (query.len > 0) {
        printf("%.*s\n", (int)query.len, query.p);

        if (mg_get_http_var(&query, "dir", dir, sizeof(dir) - 1) > 0)
          printf("dir: "
                 "%s"
                 "\n",
                 dir);
      }

      // return a json with a directory listing
      if (strcmp(dir, "") == 0)
        return get_directory(nc, home_dir.c_str());
      else
        return get_directory(nc, dir);
    }
#endif

    if (strnstr(hm->uri.p, "/get_molecules", hm->uri.len) != NULL) {
      struct mg_str query = hm->query_string;

      if (query.len > 0) {
        printf("%.*s\n", (int)query.len, query.p);

        char datasetid[256] = "";
        char freq_start_str[256] = "";
        char freq_end_str[256] = "";
        double freq_start = 0.0;
        double freq_end = 0.0;
        bool compress = false;

        struct mg_str *hdr = mg_get_http_header(hm, "accept-encoding");

        if (hdr != NULL) {
          if (strnstr(hdr->p, "gzip", hdr->len) != NULL)
            compress = true;
        }

        PrintThread{} << "Accept-Encoding:"
                      << std::string_view(hdr->p, hdr->len)
                      << "; compression support " << (compress ? "" : "not ")
                      << "found." << std::endl;

        mg_get_http_var(&query, "datasetId", datasetid, sizeof(datasetid));

        if (mg_get_http_var(&query, "freq_start", freq_start_str,
                            sizeof(freq_start_str) - 1) > 0)
          freq_start = atof(freq_start_str) / 1.0E9; //[Hz -> GHz]

        if (mg_get_http_var(&query, "freq_end", freq_end_str,
                            sizeof(freq_end_str) - 1) > 0)
          freq_end = atof(freq_end_str) / 1.0E9; //[Hz -> GHz]

        nc->user_data = (void *)++s_next_id;
        printf("MG_EV_HTTP_REQUEST conn_id = %zu\n", s_next_id);

        struct molecules_request *m_req = (struct molecules_request *)malloc(
            sizeof(struct molecules_request));

        if (m_req != NULL) {
          m_req->freq_start = freq_start;
          m_req->freq_end = freq_end;
          m_req->compress = compress;

          struct work_request req = {(unsigned long)nc->user_data,
                                     strdup(datasetid), m_req};

          if (write(sock[0], &req, sizeof(req)) < 0)
            perror("Writing worker sock");
          else
            break;
        }
      }
    }

    if (strnstr(hm->uri.p, "/get_spectrum", hm->uri.len) != NULL) {
      struct mg_str query = hm->query_string;

      if (query.len > 0) {
        printf("%.*s\n", (int)query.len, query.p);

        char datasetid[256] = "";

        if (mg_get_http_var(&query, "datasetId", datasetid,
                            sizeof(datasetid) - 1) > 0) {
          auto fits = get_dataset(datasetid);

          if (fits == nullptr) {
            mg_http_send_error_keep_alive(nc, 404, "Not Found");
            break;
          } else {
            if (fits->has_error) {
              mg_http_send_error_keep_alive(nc, 404, "Not Found");
              break;
            } else {
              /*std::unique_lock<std::mutex> data_lock(
                                                                                fits->data_mtx);
                                                                while
                 (!fits->processed_data) fits->data_cv.wait(data_lock);*/

              if (!fits->has_data) {
                // mg_http_send_error_keep_alive(nc, 404, "Not Found");
                mg_http_send_error_keep_alive(nc, 202, "Accepted");
                break;
              } else
                return get_spectrum(nc, fits);
            }
          }
        }
      }
    }

    if (strnstr(hm->uri.p, "FITSWebQL.html", hm->uri.len) != NULL) {
      struct mg_str query = hm->query_string;

      if (query.len > 0) {
        printf("%.*s\n", (int)query.len, query.p);

        std::vector<std::string> datasets;
        char dir[1024] = "";
        char ext[256] = "";
        char db[256] = "";
        char table[256] = "";
        char flux[256] = "";
        char tmp[256] = "";
        bool composite = false;

#ifdef LOCAL
        char pattern[] = "filename";
#else
        char pattern[] = "datasetId";
#endif

        // first try to find the main pattern
        if (mg_get_http_var(&query, pattern, tmp, sizeof(tmp) - 1) > 0)
          datasets.push_back(std::string(tmp));
        else {
          // iterate through multiple patterns, starting with '1'
          int count = 0;
          strcat(pattern, std::to_string(++count).c_str());

          while (mg_get_http_var(&query, pattern, tmp, sizeof(tmp) - 1) > 0) {
            datasets.push_back(std::string(tmp));
            size_t len = strlen(pattern);
            pattern[len - 1] = '\0';
            strcat(pattern, std::to_string(++count).c_str());
          }
        }

        mg_get_http_var(&query, "dir", dir, sizeof(dir) - 1);
        mg_get_http_var(&query, "ext", ext, sizeof(ext) - 1);
        mg_get_http_var(&query, "db", db, sizeof(db) - 1);
        mg_get_http_var(&query, "table", table, sizeof(table) - 1);
        mg_get_http_var(&query, "flux", tmp, sizeof(tmp) - 1);

        // validate the flux value
        if (strcmp(tmp, "linear") == 0 || strcmp(tmp, "logistic") == 0 ||
            strcmp(tmp, "ratio") == 0 || strcmp(tmp, "square") == 0 ||
            strcmp(tmp, "legacy") == 0)
          strcpy(flux, tmp);

        mg_get_http_var(&query, "view", tmp, sizeof(tmp) - 1);
        if (strcmp(tmp, "composite") == 0)
          composite = true;

        // sane defaults
        if (strstr(db, "hsc") != NULL)
          strcpy(flux, "ratio");

        if (strstr(table, "fugin") != NULL)
          strcpy(flux, "logistic");

        /*PrintThread{} << "dir:" << dir << ", ext:" << ext
                                                                                << ", db:" << db << ", table:" << table
                                                                                << ", composite:" << composite
                                                                                << ", flux:" << flux << ", ";
                                      for (auto const &dataset : datasets)
                                        std::cout << dataset << " ";
                                      std::cout << std::endl;*/

        if (datasets.size() > 0)
          return execute_fits(nc, dir, ext, db, table, datasets, composite,
                              flux);
      }

      mg_http_send_error_keep_alive(nc, 404, "Not Found");
      break;
    }

    mg_serve_http(nc, hm, s_http_server_opts);
    break;
  }
  case MG_EV_CLOSE: {
    if (is_websocket(nc) && nc->user_data != NULL) {
      printf("closing a websocket connection for %s\n", (char *)nc->user_data);
    }
  }
  default:
    break;
  }
}

int main(void) {
#ifdef CLUSTER
  setenv("ZSYS_SIGHANDLER", "false", 1);
  // LAN cluster node auto-discovery
  beacon_thread = std::thread([]() {
    speaker = zactor_new(zbeacon, NULL);
    if (speaker == NULL)
      return;

    zstr_send(speaker, "VERBOSE");
    zsock_send(speaker, "si", "CONFIGURE", BEACON_PORT);
    char *my_hostname = zstr_recv(speaker);
    if (my_hostname != NULL) {
      const char *message = "JVO:>FITSWEBQL::ENTER";
      const int interval = 1000; //[ms]
      zsock_send(speaker, "sbi", "PUBLISH", message, strlen(message), interval);
    }

    listener = zactor_new(zbeacon, NULL);
    if (listener == NULL)
      return;

    zstr_send(listener, "VERBOSE");
    zsock_send(listener, "si", "CONFIGURE", BEACON_PORT);
    char *hostname = zstr_recv(listener);
    if (hostname != NULL)
      free(hostname);
    else
      return;

    zsock_send(listener, "sb", "SUBSCRIBE", "", 0);
    zsock_set_rcvtimeo(listener, 500);

    while (!exiting) {
      char *ipaddress = zstr_recv(listener);
      if (ipaddress != NULL) {
        zframe_t *content = zframe_recv(listener);
        std::string_view message = std::string_view(
            (const char *)zframe_data(content), zframe_size(content));

        // ENTER
        if (message.find("ENTER") != std::string::npos) {
          if (strcmp(my_hostname, ipaddress) != 0) {
            std::string node = std::string(ipaddress);

            if (!cluster_contains_node(node)) {
              PrintThread{} << "found a new peer @ " << ipaddress << ": "
                            << message << std::endl;
              cluster_insert_node(node);
            }
          }
        }

        // LEAVE
        if (message.find("LEAVE") != std::string::npos) {
          if (strcmp(my_hostname, ipaddress) != 0) {
            std::string node = std::string(ipaddress);

            if (cluster_contains_node(node)) {
              PrintThread{} << ipaddress << " is leaving: " << message
                            << std::endl;
              cluster_erase_node(node);
            }
          }
        }

        zframe_destroy(&content);
        zstr_free(&ipaddress);
      }
    }

    if (my_hostname != NULL)
      free(my_hostname);
  });
#endif

  struct mg_connection *nc;
  int i;

  if (mg_socketpair(sock, SOCK_STREAM) == 0) {
    perror("Opening socket pair");
    exit(1);
  }

  signal(SIGTERM, signal_handler);
  signal(SIGINT, signal_handler);

  ipp_init();
  curl_global_init(CURL_GLOBAL_ALL);

  int rc = sqlite3_open_v2("splatalogue_v3.db", &splat_db,
                           SQLITE_OPEN_READONLY | SQLITE_OPEN_FULLMUTEX, NULL);

  if (rc) {
    fprintf(stderr, "Can't open local splatalogue database: %s\n",
            sqlite3_errmsg(splat_db));
    sqlite3_close(splat_db);
    splat_db = NULL;
  }

  struct passwd *passwdEnt = getpwuid(getuid());
  home_dir = passwdEnt->pw_dir;

  mg_mgr_init(&mgr, NULL);

  nc = mg_bind(&mgr, s_http_port, ev_handler);
  if (nc == NULL) {
    printf("Failed to create listener\n");
    return 1;
  }

  mg_set_protocol_http_websocket(nc);
  s_http_server_opts.document_root = "htdocs_mg"; // Serve current directory
  s_http_server_opts.enable_directory_listing = "no";
#ifdef LOCAL
  s_http_server_opts.index_files = "local.html";
#else
  s_http_server_opts.index_files = "test.html";
#endif
  s_http_server_opts.custom_mime_types =
      ".txt=text/plain,.html=text/html,.js=application/javascript,.ico=image/"
      "x-icon,.png=image/png,.gif=image/gif,.webp=image/webp,.jpg=image/"
      "jpeg,.jpeg=image/jpeg,.bpg=image/bpg,.mp4=video/mp4,.hevc=video/"
      "hevc,.css=text/css,.pdf=application/pdf,.svg=image/"
      "svg+xml,.wasm=application/wasm";

  for (i = 0; i < s_num_worker_threads; i++) {
    mg_start_thread(worker_thread_proc, &mgr);
  }

  printf("FITSWebQL: started on port %s\n", s_http_port);
  while (s_received_signal == 0) {
    mg_mgr_poll(&mgr, 200);
  }

  mg_mgr_free(&mgr);

  // no need to call shutdown() as close will do it for us
  close(sock[0]);
  close(sock[1]);

  printf("FITSWebQL: clean shutdown completed\n");

  curl_global_cleanup();

  if (splat_db != NULL)
    sqlite3_close(splat_db);

#ifdef CLUSTER
  if (speaker != NULL) {
    zstr_sendx(speaker, "SILENCE", NULL);

    const char *message = "JVO:>FITSWEBQL::LEAVE";
    const int interval = 1000; //[ms]
    zsock_send(speaker, "sbi", "PUBLISH", message, strlen(message), interval);

    zstr_sendx(speaker, "SILENCE", NULL);
    zactor_destroy(&speaker);
  }

  if (listener != NULL) {
    zstr_sendx(listener, "UNSUBSCRIBE", NULL);
    beacon_thread.join();
    zactor_destroy(&listener);
  }
#endif

  return 0;
}

ext: cpp


text:
/*
 * MutableBooleanNetwork.hpp
 *
 *  Created on: Nov 18, 2010
 *      Author: stewie
 */

#ifndef MUTABLEBOOLEANNETWORK_HPP_
#define MUTABLEBOOLEANNETWORK_HPP_

#include <cassert>
#include <cstddef>
#include <iosfwd>
#include <vector>

#include <boost/graph/graph_selectors.hpp>
#include <boost/graph/adjacency_list.hpp>

#include "BooleanDynamics.hpp"
#include "BooleanFunction.hpp"

namespace bn {

class MutableBooleanNetwork : public BooleanDynamics {
private:
	class Updater;
	friend class Updater;

public:
	typedef std::vector<int> TruthTable;
	typedef boost::adjacency_list<boost::vecS, boost::vecS,
			boost::bidirectionalS, TruthTable> Network;

	using BooleanDynamics::update; // make update(size_t) visible

	/**
	 * Returns the number of nodes in the network.
	 * @return the number of nodes
	 */
	size_t size() const {
		return boost::num_vertices(net);
	}

	MutableBooleanNetwork* clone() const {
		return new MutableBooleanNetwork(*this);
	}

	Network& topology() {
		return net;
	}

	const Network& topology() const {
		return net;
	}

	/**
	 * Sets the state of this network.
	 * @param s the new state
	 */
	void setState(const State& s) {
		state = s;
		assert(state.size() == size());
	}

	/**
	 * Returns a reference to the current state of this network.
	 * @return the state of this network
	 */
	const State& getState() const {
		return state;
	}

	void update();

	BooleanFunction getFunction(const size_t i) const;

	std::vector<BooleanFunction> getFunctions() const;

	BooleanDynamics* simulate() {
		return new Updater(*this);
	}

	MutableBooleanNetwork& operator=(MutableBooleanNetwork other) {
		swap(other);
		return *this;
	}

	void swap(MutableBooleanNetwork& other) {
		using std::swap;
		swap(state, other.state);
		swap(net, other.net);
	}

	void update(State& s);

	static MutableBooleanNetwork makeNetwork(const char topologyFilename[],
			const char functionFilename[]);

	friend std::ostream& operator<<(std::ostream& out,
			const MutableBooleanNetwork& rbn);

protected:
	/**
	 * The current state of the netwotk.
	 */
	State state;

	Network net;

private:
	class Updater : public BooleanDynamics {
	public:
		using BooleanDynamics::update;

		Updater(MutableBooleanNetwork& bn) :
			bn(&bn) {
		}

		Updater* clone() const {
			return new Updater(*this);
		}

		const State& getState() const {
			return state;
		}

		void update();

		void setState(const State& s) {
			state = s;
		}

		void update(State& s);

		size_t size() const {
			return bn->size();
		}

		BooleanFunction getFunction(const size_t i) const {
			return bn->getFunction(i);
		}

	private:
		MutableBooleanNetwork* bn;

		State state;
	};
};

void swap(MutableBooleanNetwork& a, MutableBooleanNetwork& b);

} // namespace bn

#endif /* MUTABLEBOOLEANNETWORK_HPP_ */

ext: hpp

max_stars_repo_name: Markfrancisrogers/BooleanNetwork

max_issues_repo_name: Markfrancisrogers/BooleanNetwork

max_forks_repo_name: Markfrancisrogers/BooleanNetwork


text:
#ifndef JSON_QUERY_SELCTOR_TYPE_HPP
#define JSON_QUERY_SELCTOR_TYPE_HPP

#include <boost/optional.hpp>
#include <boost/variant.hpp>
#include <iostream>
#include <iterator>
#include <type_traits>
#include <utility>
#include <vector>

#include "../json/json.hpp"
#include "../utils.hpp"

namespace selectors {
using namespace json;
namespace ranges = std::ranges;

class ApplySelectorError : public std::exception {
    std::string message;

public:
    ApplySelectorError(const std::string& message) : message(message) {}

    const char* what() const noexcept override { return message.c_str(); }
};

class SelectorNode;

// Used to detect wrong parsing because the default constructor of the
// first variant is sometimes used.
class InvalidSelector {
public:
    static const char* name() { return "Invalid"; }
    friend std::ostream& operator<<(std::ostream& o,
                                    const InvalidSelector& /*unused*/) {
        return o << "InvalidSelector";
    }
};

/**
 * Selects everything.
 *
 * Can only be the first element of a RootSelector but can be followed by other
 * selectors.
 *
 * Can be applied to all json items.
 */
class AnyRootSelector {
public:
    static const char* name() { return "Any"; }

    friend std::ostream& operator<<(std::ostream& o,
                                    const AnyRootSelector& /*unused*/) {
        return o << "AnyRootSelector";
    }
};

/**
 * Selects a single key/property from an object.
 *
 * Identified by a string surrounded with `"`.
 *
 * Can only be applied to objects.
 *
 * E.g.
 *
 * ```json
 * {
 *   "key1": 1,
 *   "key2": 2,
 *   "key3": 3
 * }
 * ```
 *
 * with the selector `"key2"` turns into
 *
 * ```json
 * 2
 * ```
 */
class KeySelector {
    std::string key;

public:
    KeySelector() = default;
    KeySelector(std::string key) : key(std::move(key)) {}
    KeySelector(const KeySelector&) = default;
    KeySelector& operator=(const KeySelector&) = default;

    static const char* name() { return "Key"; }

    const std::string& get() const { return key; }

    friend std::ostream& operator<<(std::ostream& o, const KeySelector& self) {
        return o << "KeySelector(" << self.key << ")";
    }
};

/**
 * Selects an element from an array.
 *
 * Identified by `[` and `]` with a single number in between.
 *
 * Can only be applied to arrays.
 *
 * E.g.
 *
 * ```json
 * [ 1, 2, 3 ]
 * ```
 *
 * with the selector `[1]` turns into
 *
 * ```json
 * 2
 * ```
 */
class IndexSelector {
    int index;

public:
    IndexSelector() = default;
    IndexSelector(int index) : index(index) {}
    IndexSelector(const IndexSelector&) = default;
    IndexSelector& operator=(const IndexSelector&) = default;

    static const char* name() { return "Index"; }

    int get() const { return index; }

    friend std::ostream& operator<<(std::ostream& o,
                                    const IndexSelector& self) {
        return o << "IndexSelector(" << self.index << ")";
    }
};

/**
 * Selects a range from an array.
 *
 * Identified by `[` and `]` with a range in between. The range can have a start
 * and an end separated by `:`. If there is nothing between the brackets it is
 * the same as `[:]`.
 *
 * Can only be applied to arrays.
 *
 * E.g. `[1:2]`, `[4:]`, `[:3]`, `[:]`, `[]`
 *
 * E.g.
 *
 * ```json
 * [ 1, 2, 3, 4, 5 ]
 * ```
 *
 * with the range `[1:3]` turns into
 *
 * ```json
 * [ 2, 3, 4 ]
 * ```
 */
class RangeSelector {
    // uses boost::optional instead of std::optional because that is easier to
    // use with the parser
    boost::optional<int> start;
    boost::optional<int> end;

public:
    RangeSelector() = default;
    RangeSelector(boost::optional<int> start, boost::optional<int> end)
        : start(start), end(end) {}
    RangeSelector(const RangeSelector&) = default;
    RangeSelector& operator=(const RangeSelector&) = default;

    static const char* name() { return "Range"; }

    const boost::optional<int>& get_start() const { return start; }
    const boost::optional<int>& get_end() const { return end; }

    friend std::ostream& operator<<(std::ostream& o,
                                    const RangeSelector& self) {
        return o << "RangeSelector("
                 << (self.start.has_value() ? std::to_string(self.start.value())
                                            : "")
                 << ","
                 << (self.end.has_value() ? std::to_string(self.end.value())
                                          : "")
                 << ")";
    }
};

/**
 * Selects (multiple) properties/keys from an object.
 *
 * Identified by `{` and `}` with a list of keys in between.
 *
 * Can only be applied to objects.
 *
 * E.g.
 *
 * ```json
 * {
 *   "key1": 1,
 *   "key2": 2,
 *   "key3": 3
 * }
 * ```
 *
 * with the selector `{"key1", "key3"}` turns into
 *
 * ```json
 * {
 *   "key1": 1,
 *   "key3": 3
 * }
 * ```
 */
class PropertySelector {
    std::vector<std::string> keys;

public:
    PropertySelector() = default;
    PropertySelector(std::vector<std::string> keys) : keys(std::move(keys)) {}
    PropertySelector(const PropertySelector&) = default;
    PropertySelector& operator=(const PropertySelector&) = default;

    static const char* name() { return "Property"; }

    const std::vector<std::string>& get_keys() const { return keys; }

    friend std::ostream& operator<<(std::ostream& o,
                                    const PropertySelector& self) {
        o << "PropertySelector(";
        for (const auto& x : self.keys) {
            o << x << ",";
        }
        return o << ")";
    }
};

/**
 * Selector to filter arrays.
 *
 * Identified by `|` followed by a key.
 *
 * Only works on arrays. When applied it filters the array for objects
 * containing the given key and returns an array containing the associated
 * items of the keys.
 *
 * Can only be applied to arrays.
 *
 * E.g.
 *
 * ```json
 * [
 *   { "key": 1 },
 *   { "key": 2 },
 *   { "key": 3 }
 * ]
 * ```
 *
 * with the filter `|"key"` turns into
 *
 * ```json
 * [ 1, 2, 3 ]
 * ```
 */
class FilterSelector {
    // Having a nested KeySelector makes it easier to parse and use in
    // apply_selector.
    KeySelector filter;

public:
    FilterSelector() = default;
    FilterSelector(const KeySelector& filter) : filter(filter) {}
    FilterSelector(const FilterSelector&) = default;
    FilterSelector& operator=(const FilterSelector&) = default;

    static const char* name() { return "Filter"; }

    const KeySelector& get() const { return filter; }

    friend std::ostream& operator<<(std::ostream& o,
                                    const FilterSelector& self) {
        return o << "FilterSelector(" << self.filter << ")";
    }
};

/**
 * Selector to truncate json.
 *
 * Identified by `!`.
 *
 * When applied truncates the json at the current item. Outputting the item if
 * it is a number or a string and outputting an empty object or array for
 * objects and arrays.
 *
 * Can be applied to all json items.
 */
class TruncateSelector {
public:
    static const char* name() { return "Truncate"; }

    friend std::ostream& operator<<(std::ostream& o,
                                    const TruncateSelector& /*unused*/) {
        return o << "TruncateSelector";
    }
};

/**
 * Selector to flatten json arrays.
 *
 * Identified by the prefix `..`.
 *
 * When applied will flatten one level of arrays.
 *
 * E.g.
 *
 * ```json
 * [ [1, 2], [3, 4], [5, 6] ]
 * ```
 *
 * turns into
 *
 * ```json
 * [1, 2, 3, 4, 5, 6]
 * ```
 */
class FlattenSelector {
public:
    static const char* name() { return "Flatten"; }

    friend std::ostream& operator<<(std::ostream& o,
                                    const FlattenSelector& /*unused*/) {
        return o << "FlattenSelector";
    }
};

/**
 * Constraint on the type of a selector.
 *
 * All selector types should be listed here.
 */
// clang-format off
template <typename S>
concept is_selector =
    basically_same_as<S, InvalidSelector> ||
    basically_same_as<S, AnyRootSelector> ||
    basically_same_as<S, KeySelector> ||
    basically_same_as<S, IndexSelector> ||
    basically_same_as<S, RangeSelector> ||
    basically_same_as<S, PropertySelector> ||
    basically_same_as<S, FilterSelector> ||
    basically_same_as<S, TruncateSelector> ||
    basically_same_as<S, FlattenSelector>;
// clang-format on

/**
 * Constraint for the type of iterator used for selector chains.
 *
 * Requirements:
 *
 * - copyable: need to save a copy in case of multiple sub-json items
 *   (e.g. in ranges).
 * - incrementable
 * - dereferecing yields `const SelectorNode&`
 */
// clang-format off
template <typename I>
concept sel_iter =
    std::copyable<I> &&
    std::incrementable<I> &&
    requires(I i) {
        { *i } ->std::same_as<const SelectorNode&>;
    };
// clang-format on

/**
 * Unifying wrapper for all selector kinds.
 *
 * This avoids having to allocate each constructor separately on the heap.
 */
class SelectorNode {
    using InnerVariant =
        boost::variant<InvalidSelector, AnyRootSelector, KeySelector,
                       IndexSelector, RangeSelector, PropertySelector,
                       FilterSelector, TruncateSelector, FlattenSelector>;

public:
    explicit SelectorNode() = default;

    template <is_selector S> explicit SelectorNode(S inner) : inner(inner) {}

    const char* name() const {
        return boost::apply_visitor(
            [](const is_selector auto& x) { return x.name(); }, inner);
    }

    template <is_selector S> const S& as() const {
        return boost::get<S>(inner);
    }

    template <sel_iter Iterator>
    JsonNode apply(const JsonNode& json, Iterator next, Iterator end) const {
        return json.apply_visitor(
            [next, end](const is_json_item auto& item,
                        const is_selector auto& s) {
                return s.apply(item, next, end);
            },
            inner);
    }

    friend std::ostream& operator<<(std::ostream& o, const SelectorNode& self) {
        boost::apply_visitor([&o](is_selector auto val) { o << val; },
                             self.inner);
        return o;
    }

    // private:
    InnerVariant inner;
};

/**
 * Inserts all items from the second vector into the first one.
 */
template <typename T>
void extend_vec_with(std::vector<T>& vec, const std::vector<T>& extension) {
    vec.insert(vec.end(), extension.begin(), extension.end());
}

// these template functions make it a little easier to find and extend what
// selector handles what json item
//
// also: ideally I would put them into a separate header but it's to much
// hassle to resolve the "define before use" dependencies

template <sel_iter I>
JsonNode apply_selector(const FlattenSelector& /*unused*/, const JsonArray& arr,
                        I next, I end) {
    std::vector<JsonNode> flattened_array;

    for (const JsonNode& item : arr.get()) {
        // calculate sub result and flatten if result is an array
        const JsonNode result = apply_selector(item, next, end);
        result.apply_visitor(
            overloaded{[&flattened_array](const JsonArray& nested_array) {
                           extend_vec_with(flattened_array, nested_array.get());
                       },
                       [](const is_json_item auto& /*unused*/) {}});
    }

    return JsonNode(JsonArray(flattened_array));
}

template <sel_iter I>
JsonNode apply_selector(const TruncateSelector& /*unused*/,
                        const JsonObject& /*unused*/, I /*unused*/,
                        I /*unused*/) {
    return JsonNode(JsonObject());
}

template <sel_iter I>
JsonNode apply_selector(const TruncateSelector& /*unused*/,
                        const JsonArray& /*unused*/, I /*unused*/,
                        I /*unused*/) {
    return JsonNode(JsonArray());
}

template <sel_iter I>
JsonNode apply_selector(const TruncateSelector& /*unused*/,
                        const is_json_item auto& json, I /*unused*/,
                        I /*unused*/) {
    return JsonNode(json);
}

template <sel_iter I>
JsonNode apply_selector(const FilterSelector& s, const JsonArray& arr, I next,
                        I end) {
    std::vector<JsonNode> result;

    for (const JsonNode& item : arr.get()) {
        try {
            // only check JsonObjects and ignore all other items
            item.apply_visitor(overloaded{
                [&result, &next, &end, key = s.get()](const JsonObject& obj) {
                    result.push_back(apply_selector(key, obj, next, end));
                },
                [](const is_json_item auto& /*unused*/) {}});
        } catch (const ApplySelectorError&) {
            // note: this means th key was not found and thus we ignore the item
        }
    }

    return JsonNode(JsonArray(result));
}

template <sel_iter I>
JsonNode apply_selector(const PropertySelector& s, const JsonObject& obj,
                        I next, I end) {
    const std::vector<std::string>& keys = s.get_keys();
    // initialize with correct size so we don't need back_inserter
    std::vector<std::pair<std::string, JsonNode>> result{keys.size()};

    std::ranges::transform(
        keys, result.begin(), [&obj, next, end](const std::string& key) {
            return std::make_pair(key,
                                  apply_selector(obj.find(key), next, end));
        });

    return JsonNode(JsonObject(result));
}

template <sel_iter I>
JsonNode apply_selector(const RangeSelector& s, const JsonArray& array, I next,
                        I end) {
    const std::vector<JsonNode>& arr = array.get();

    // range start and end or default values
    const auto range_start = s.get_start().get_value_or(0);
    const auto range_end = s.get_end().get_value_or(arr.size() - 1) + 1;

    auto begin_it = arr.cbegin() + range_start;
    auto end_it = arr.cbegin() + range_end;

    const unsigned long num_items = range_end - range_start;

    std::vector<JsonNode> result{num_items};
    std::transform(begin_it, end_it, result.begin(),
                   [next, end](const JsonNode& item) {
                       return apply_selector(item, next, end);
                   });

    return {JsonArray{result}};
}

template <sel_iter I>
JsonNode apply_selector(const IndexSelector& s, const JsonArray& arr, I next,
                        I end) {
    return apply_selector(arr.at(s.get()), next, end);
}

template <sel_iter I>
JsonNode apply_selector(const KeySelector& s, const JsonObject& obj, I next,
                        I end) {
    try {
        return apply_selector(obj.find(s.get()), next, end);
    } catch (std::out_of_range&) {
        throw ApplySelectorError("Key \"" + s.get() +
                                 "\" was not found in json object");
    }
}

template <sel_iter I>
JsonNode apply_selector(const AnyRootSelector& /*unused*/,
                        const is_json_item auto& json, I next, I end) {
    return apply_selector(json, next, end);
}

template <sel_iter I>
JsonNode apply_selector(const is_selector auto& s, const is_json_item auto& j,
                        I /*unused*/, I /*unused*/) {
    throw ApplySelectorError(
        std::string("selector and json object don't match: ") + s.name() +
        ", " + j.name());
}

// entry point for applying the next selector
template <sel_iter I>
JsonNode apply_selector(const JsonNode& json, I next, I end) {
    if (next == end) {
        return JsonNode(json);
    }

    const SelectorNode& next_s = *next;
    next++;
    return json.apply_visitor(
        [&next, &end](is_json_item auto& item, is_selector auto& selector) {
            return apply_selector(selector, item, next, end);
        },
        next_s.inner);
}

/**
 * Contains a list of sequential selectors.
 *
 * The result of applying the selector is the result of applying all selectors
 * in sequential order.
 */
class RootSelector {
    std::vector<SelectorNode> inner;

public:
    RootSelector() = default;
    RootSelector(std::vector<SelectorNode> inner) : inner(std::move(inner)) {}

    const std::vector<SelectorNode>& get() const { return inner; }

    JsonNode apply(const JsonNode& json) const {
        return apply_selector(json, inner.cbegin(), inner.cend());
    }

    friend std::ostream& operator<<(std::ostream& o, const RootSelector& self) {
        o << "RootSelector {";
        for (const auto& x : self.inner) {
            o << x << ",";
        }
        return o << "}";
    }
};

/**
 * Selectors is a list of independent [RootSelectors](@ref RootSelector).
 *
 * If there is only one root selector the result of applying the selectors is
 * simply the result of that selector. If there is more than one root
 * selectors the result is a list of the results of those selectors.
 */
class Selectors {
    std::vector<RootSelector> selectors;

public:
    Selectors() = default;
    Selectors(std::vector<RootSelector> selectors)
        : selectors(std::move(selectors)) {}

    const std::vector<RootSelector>& get() const { return selectors; }
    std::vector<RootSelector>& get() { return selectors; }

    /**
     * Apply all the selectors to the given Json.
     *
     * Throws ApplySelectorError if one of the selectors can't be applied to
     * the json.
     */
    JsonNode apply(const JsonNode& json) const {
        if (selectors.empty()) {
            return JsonNode(JsonLiteral(JSON_NULL));
        } else if (selectors.size() == 1) {
            return selectors[0].apply(json);
        } else {
            auto apply = [&json](const RootSelector& selector) {
                return selector.apply(json);
            };
            std::vector<JsonNode> array;
            ranges::transform(selectors, std::back_inserter(array), apply);
            return JsonNode(JsonArray(array));
        }
    }

    friend std::ostream& operator<<(std::ostream& o, const Selectors& self) {
        o << '[';
        for (const auto& x : self.selectors) {
            o << x << ',';
        }
        return o << ']';
    }
};

} // namespace selectors

#endif

max_stars_repo_name: Lythenas/json-query

max_issues_repo_name: Lythenas/json-query

max_forks_repo_name: Lythenas/json-query


text:
#ifndef HUSKYSYSTEM_H
#define HUSKYSYSTEM_H

#include <Eigen/Dense>
#include <iostream>
#include <fstream>
#include <memory>
#include "ros/ros.h"

#include "estimator/body_estimator.hpp"
#include "system/husky_state.hpp"
#include "utils/husky_data.hpp"
#include "utils/joint_state.hpp"
#include "utils/imu.hpp"
#include "utils/velocity.hpp"

#include "communication/pose_publisher_node.hpp"

// Threading
#include <boost/thread/condition.hpp>
#include <boost/thread/mutex.hpp>
#include <boost/thread/thread.hpp>
#include <boost/circular_buffer.hpp>

// TODO: Singleton design pattern (there should only be one HuskySystem)
class HuskySystem {

    public:
        EIGEN_MAKE_ALIGNED_OPERATOR_NEW
        // Default Contructor
        HuskySystem(ros::NodeHandle* nh, husky_inekf::husky_data_t* husky_data_buffer);
        ~HuskySystem();

        // Step forward one iteration of the system
        void step();

    private:
        // ROS NodeHandle
        ros::NodeHandle* nh_;
        // ROS pose publisher
        PosePublisherNode pose_publisher_node_;
        // ROS timestamp
        ros::Time timestamp_;
        // Passive Time Synchronizer
        // PassiveTimeSync ts_;
        // Cassie's current state estimate
        husky_inekf::HuskyState state_;
        // Husky data queues
        husky_inekf::husky_data_t* husky_data_buffer_;
        // Invariant extended Kalman filter for estimating the robot's body state
        husky_inekf::BodyEstimator estimator_;
        // Most recent data packet
        husky_inekf::JointStateMeasurementPtr joint_state_packet_;
        husky_inekf::ImuMeasurementPtr imu_packet_;
        husky_inekf::VelocityMeasurementPtr wheel_velocity_packet_;
        husky_inekf::VelocityMeasurementPtr camera_velocity_packet_;
        husky_inekf::VelocityMeasurementPtr gps_velocity_packet_;

        bool enable_wheel_vel_;
        bool enable_camera_vel_;
        bool enable_gps_vel_;
        Eigen::Matrix<double,3,3> wheel_vel_cov_;
        Eigen::Matrix<double,3,3> camera_vel_cov_;
        Eigen::Matrix<double,3,3> gps_vel_cov_;

        // Update most recent packet to use
        bool updateNextIMU();
        bool updateNextJointState();
        bool updateNextWheelVelocity();
        bool updateNextCameraVelocity();
        bool updateNextGPSVelocity();

        // Publish output path
        void logPoseTxt(const husky_inekf::HuskyState& state_);
        // Output file
        std::string file_name_;
        std::string tum_file_name_;
        std::string vel_est_file_name_;     // the velocity estimated by the filter
        std::string bias_est_file_name_;
        std::string vel_input_file_name_;   // the velocity used in correction step
        std::string imu_file_name_;

        std::ofstream outfile_;
        std::ofstream tum_outfile_;
        std::ofstream vel_est_outfile_;
        std::ofstream bias_est_outfile_;
        std::ofstream vel_input_outfile_;
        std::ofstream imu_outfile_;

        // Publish path node enable flag
        bool enable_pose_publisher_;
        bool enable_pose_logger_;
        bool enable_debug_logger_;
        bool new_pose_ready_;
        int log_pose_skip_;
        int skip_count_;

        double last_imu_time_;
};

#endif // HUSKYSYSTEM_H


text:
//==================================================================================================
/*!
  @file

    @copyright 2016 NumScale SAS

  Distributed under the Boost Software License, Version 1.0.
  (See accompanying file LICENSE.md or copy at http://boost.org/LICENSE_1_0.txt)
*/
//==================================================================================================
#ifndef BOOST_SIMD_FUNCTION_REM_2PI_HPP_INCLUDED
#define BOOST_SIMD_FUNCTION_REM_2PI_HPP_INCLUDED

#if defined(DOXYGEN_ONLY)
namespace boost { namespace simd
{

  /*!
    @ingroup group-trigonometric
    This function object returns the compute the remainder modulo \f$2\pi\f$.

    @par Header <boost/simd/function/rem_2pi.hpp>

    @par Note

     - The result is in \f$[-\pi, \pi]\f$.

     - If the input is near \f$\pi\f$ the output can be \f$\pi\f$ or \f$-\pi\f$
      depending  on register disponibility if extended arithmetic is used.

    @par Example:

      @snippet rem_2pi.cpp rem_2pi

    @par Possible output:

      @snippet rem_2pi.txt rem_2pi

  **/
  IEEEValue rem_2pi(IEEEValue const& x);
} }
#endif

#include <boost/simd/function/scalar/rem_2pi.hpp>
#include <boost/simd/function/simd/rem_2pi.hpp>

#endif

max_stars_repo_name: SylvainCorlay/pythran

max_issues_repo_name: SylvainCorlay/pythran

max_forks_repo_name: SylvainCorlay/pythran


text:
#include <Eigen/Dense>

#include <ancse/config.hpp>
#include <ancse/cfl_condition.hpp>
#include <ancse/fvm_rate_of_change.hpp>
#include <ancse/snapshot_writer.hpp>
#include <ancse/time_loop.hpp>

static int n_vars = 1;

template<class F>
Eigen::MatrixXd ic(const F &f, const Grid &grid) {
    Eigen::MatrixXd u0(n_vars, grid.n_cells);
    for(int i = 0; i < grid.n_cells; ++i) {
        u0.col(i) = f(cell_center(grid, i));
    }

    return u0;
}

TimeLoop make_fvm(const nlohmann::json &config,
                  const Grid &grid,
                  std::shared_ptr<Model> &model)
{
    double t_end = config["t_end"];
    double cfl_number = config["cfl_number"];

    auto n_ghost = grid.n_ghost;
    auto n_cells = grid.n_cells;

    auto n_vars = model->get_nvars();

    auto simulation_time = std::make_shared<SimulationTime>(t_end);
    auto fvm_rate_of_change
            = make_fvm_rate_of_change(config, grid, model,
                                      simulation_time);
    auto boundary_condition
            = make_boundary_condition(n_ghost,
                                      config["boundary_condition"]);
    auto time_integrator = make_runge_kutta(config,
                                            fvm_rate_of_change,
                                            boundary_condition,
                                            n_vars, n_cells);
    auto cfl_condition = make_cfl_condition(grid, model, cfl_number);
    auto snapshot_writer = std::make_shared< JSONSnapshotWriter<FVM> >
            (grid, model, simulation_time,
             std::string(config["output_dir"]),
             std::string(config["output_file"]));

    return TimeLoop(simulation_time, time_integrator,
                    cfl_condition, snapshot_writer);
}

void shock_test(const nlohmann::json &config)
{
    std::shared_ptr<Model> model = std::make_shared<Burgers>();

    auto fn = [](double x) {
        Eigen::VectorXd u(n_vars);
        if (x <= 0.5) { // left state
            u(0) = 1;
        } else {        // right state
            u(0) = 0;
        }
        return u;
    };

    int n_ghost = config["n_ghost"];
    int n_cells = int(config["n_interior_cells"]) + n_ghost * 2;

    auto grid = Grid({0.0, 1.0}, n_cells, n_ghost);
    auto u0 = ic(fn, grid);

    auto fvm = make_fvm(config, grid, model);
    fvm(u0);
}

void rarefaction_test(const nlohmann::json &config)
{
    std::shared_ptr<Model> model = std::make_shared<Burgers>();

    auto fn = [](double x) {
        Eigen::VectorXd u(n_vars);
        if (x <= 0.5) { // left state
            u(0) = 0;
        } else {        // right state
            u(0) = 1;
        }
        return u;
    };

    int n_ghost = config["n_ghost"];
    int n_cells = int(config["n_interior_cells"]) + n_ghost * 2;

    auto grid = Grid({0.0, 1.0}, n_cells, n_ghost);
    auto u0 = ic(fn, grid);

    auto fvm = make_fvm(config, grid, model);
    fvm(u0);
}

int main(int argc, char* const argv[])
{
    nlohmann::json config;
    std::string fileName;
    if (argc == 2) {
        fileName = argv[1];
    } else {
        fileName = "../config.json";
    }
    config = get_config (fileName);

    std::string ic_key = config["initial_conditions"];
    if (ic_key == "shock") {
        shock_test(config);
    } else if (ic_key == "rarefaction") {
        rarefaction_test(config);
    }

    return 0;
}


text:
/*

Copyright (c) 2014, Project OSRM, Dennis Luxen, others
All rights reserved.

Redistribution and use in source and binary forms, with or without modification,
are permitted provided that the following conditions are met:

Redistributions of source code must retain the above copyright notice, this list
of conditions and the following disclaimer.
Redistributions in binary form must reproduce the above copyright notice, this
list of conditions and the following disclaimer in the documentation and/or
other materials provided with the distribution.

THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND
ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE FOR
ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES
(INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;
LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON
ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
(INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.

*/

#include "../../data_structures/binary_heap.hpp"
#include "../../typedefs.h"

#include <boost/test/unit_test.hpp>
#include <boost/test/test_case_template.hpp>
#include <boost/mpl/list.hpp>

#include <random>

BOOST_AUTO_TEST_SUITE(binary_heap)

struct TestData
{
    unsigned value;
};

typedef NodeID TestNodeID;
typedef int TestKey;
typedef int TestWeight;
typedef boost::mpl::list<ArrayStorage<TestNodeID, TestKey>,
                         MapStorage<TestNodeID, TestKey>,
                         UnorderedMapStorage<TestNodeID, TestKey>> storage_types;

template <unsigned NUM_ELEM> struct RandomDataFixture
{
    RandomDataFixture()
    {
        for (unsigned i = 0; i < NUM_ELEM; i++)
        {
            data.push_back(TestData{i * 3});
            weights.push_back((i + 1) * 100);
            ids.push_back(i);
            order.push_back(i);
        }

        // Choosen by a fair W20 dice roll
        std::mt19937 g(15);

        std::shuffle(order.begin(), order.end(), g);
    }

    std::vector<TestData> data;
    std::vector<TestWeight> weights;
    std::vector<TestNodeID> ids;
    std::vector<unsigned> order;
};

constexpr unsigned NUM_NODES = 100;

BOOST_FIXTURE_TEST_CASE_TEMPLATE(insert_test, T, storage_types, RandomDataFixture<NUM_NODES>)
{
    BinaryHeap<TestNodeID, TestKey, TestWeight, TestData, T> heap(NUM_NODES);

    TestWeight min_weight = std::numeric_limits<TestWeight>::max();
    TestNodeID min_id;

    for (unsigned idx : order)
    {
        BOOST_CHECK(!heap.WasInserted(ids[idx]));

        heap.Insert(ids[idx], weights[idx], data[idx]);

        BOOST_CHECK(heap.WasInserted(ids[idx]));

        if (weights[idx] < min_weight)
        {
            min_weight = weights[idx];
            min_id = ids[idx];
        }
        BOOST_CHECK_EQUAL(min_id, heap.Min());
    }

    for (auto id : ids)
    {
        const auto &d = heap.GetData(id);
        BOOST_CHECK_EQUAL(d.value, data[id].value);

        const auto &w = heap.GetKey(id);
        BOOST_CHECK_EQUAL(w, weights[id]);
    }
}

BOOST_FIXTURE_TEST_CASE_TEMPLATE(delete_min_test, T, storage_types, RandomDataFixture<NUM_NODES>)
{
    BinaryHeap<TestNodeID, TestKey, TestWeight, TestData, T> heap(NUM_NODES);

    for (unsigned idx : order)
    {
        heap.Insert(ids[idx], weights[idx], data[idx]);
    }

    for (auto id : ids)
    {
        BOOST_CHECK(!heap.WasRemoved(id));

        BOOST_CHECK_EQUAL(heap.Min(), id);
        BOOST_CHECK_EQUAL(id, heap.DeleteMin());
        if (id + 1 < NUM_NODES)
            BOOST_CHECK_EQUAL(heap.Min(), id + 1);

        BOOST_CHECK(heap.WasRemoved(id));
    }
}

BOOST_FIXTURE_TEST_CASE_TEMPLATE(delete_all_test, T, storage_types, RandomDataFixture<NUM_NODES>)
{
    BinaryHeap<TestNodeID, TestKey, TestWeight, TestData, T> heap(NUM_NODES);

    for (unsigned idx : order)
    {
        heap.Insert(ids[idx], weights[idx], data[idx]);
    }

    heap.DeleteAll();

    BOOST_CHECK(heap.Empty());
}

BOOST_FIXTURE_TEST_CASE_TEMPLATE(decrease_key_test, T, storage_types, RandomDataFixture<10>)
{
    BinaryHeap<TestNodeID, TestKey, TestWeight, TestData, T> heap(10);

    for (unsigned idx : order)
    {
        heap.Insert(ids[idx], weights[idx], data[idx]);
    }

    std::vector<TestNodeID> rids(ids);
    std::reverse(rids.begin(), rids.end());

    for (auto id : rids)
    {
        TestNodeID min_id = heap.Min();
        TestWeight min_weight = heap.GetKey(min_id);

        // decrease weight until we reach min weight
        while (weights[id] > min_weight)
        {
            heap.DecreaseKey(id, weights[id]);
            BOOST_CHECK_EQUAL(heap.Min(), min_id);
            weights[id]--;
        }

        // make weight smaller than min
        weights[id] -= 2;
        heap.DecreaseKey(id, weights[id]);
        BOOST_CHECK_EQUAL(heap.Min(), id);
    }
}

BOOST_AUTO_TEST_SUITE_END()

max_stars_repo_name: cypox/devacus-backend

max_issues_repo_name: antoinegiret/osrm-backend

max_forks_repo_name: antoinegiret/osrm-backend


text:
/*****************************************************************************
FILE: predict.cpp

PURPOSE: Contains functions for training the model and running predictions
using the model.  All methods are part of the PredictBurnedArea class.

PROJECT:  Land Satellites Data System Science Research and Development (LSRD)
at the USGS EROS

LICENSE TYPE:  NASA Open Source Agreement Version 1.3

HISTORY:
Date          Programmer       Reason
----------    ---------------  -------------------------------------
11/26/2012    Jodi Riegle      Original development
9/3/2013      Gail Schmidt     Modified to work in the ESPA environment
                               Modified to support saving the model and then
                               reload the model

NOTES:
*****************************************************************************/

#include <sstream>
#include <iostream>
#include <fstream>
#include <boost/date_time/posix_time/posix_time.hpp>

#include "predict.h"
#include "PredictBurnedArea.h"
#include "output.h"
#include "error.h"
#include <math.h>

using namespace boost::posix_time;
using namespace std;

/******************************************************************************
MODULE: loadModel (class PredictBurnedArea)

PURPOSE: Loads a previously trained and saved model.

RETURN VALUE:
Type = None
Value          Description
-----          -----------

HISTORY:
Date          Programmer       Reason
----------    ---------------  -------------------------------------
9/3/2013      Gail Schmidt     Original development

NOTES:
*****************************************************************************/
void PredictBurnedArea::loadModel ()
{
  	gbtrees.load (LOAD_MODEL_XML.c_str());
}

/******************************************************************************
MODULE: trainModel (class PredictBurnedArea)

PURPOSE: Trains the model using a gradient boosted regression tree with the
specified parameters from the configuration file.  The model will be saved,
if specified, to the user-specified output XML file.

RETURN VALUE:
Type = bool
Value          Description
-----          -----------
false          Error training the model
true           Successful training of the model

HISTORY:
Date          Programmer       Reason
----------    ---------------  -------------------------------------
11/26/2012    Jodi Riegle      Original development
9/3/2013      Gail Schmidt     Modified to work in the ESPA environment

NOTES:
  1. It's assumed the configuration parameters (class members) have already
     been initialized.
  2. The order of the inputs in the CSV model is documented in
     PredictBurnedArea.h and the number of inputs is identified in that file.
*****************************************************************************/
bool PredictBurnedArea::trainModel ()
{
    int nTrees = TREE_CNT;          /* number of trees in the ensemble */
    float shrink = SHRINKAGE;       /* shrinkage factor */
    int depth = MAX_DEPTH;          /* maximum depth of each tree */
    float subsample_fraction = SUBSAMPLE_FRACTION;  /* fraction of the training
                                       data to be used for training vs.
                                       testing */
    int ncsv_inputs = NCSV_INPUTS;  /* number of inputs per training sample in
                                       the CSV file, not including the response
                                       value */
    string predictFile = PREDICT_OUT; /* output text file for writing the
                                         information regarding the training
                                         parameters and errors */

    /* Identify the index in the training data of the response or
       classification values (0-based) */
    int response_idx = ncsv_inputs;

    /* Document the settings used for training */
    ofstream predictOut;
    predictOut.open (predictFile.c_str());
    predictOut << "Number of trees (weak count): " << nTrees << endl;
    predictOut << "Max tree depth: " << depth << endl;
    predictOut << "Shrinkage: " << shrink << endl;
    predictOut << "Subsample portion (training vs. testing): " <<
        subsample_fraction * 100 << "%" << endl;
    predictOut << "Response index (0-based) in training data: " <<
        response_idx << endl;

    cout << second_clock::local_time() << " ======Reading=====" << endl;

    /* Read in csv file containing the training data to CvMLData object */
    cvml.read_csv (CSV_FILE.c_str());

    /* Change the response value in the training dataset to indicate it
       represents a category/classification and indicate the portion of the
       data that will be used for training vs. testing */
    CvTrainTestSplit spl (subsample_fraction, true);
    cvml.set_response_idx (response_idx);
    cvml.change_var_type (response_idx, CV_VAR_CATEGORICAL);
    cvml.set_train_test_split (&spl);

    /* Train the model using the data read from the input CSV file and the
       specified training parameters */
    cout << second_clock::local_time() <<
        " ======Training Using CvMLData=====" << endl;
    predictOut << "Loss function type: DEVIANCE_LOSS (for classification)" <<
        endl;
    gbtrees.train (&cvml, CvGBTreesParams(CvGBTrees::DEVIANCE_LOSS, nTrees,
        shrink, subsample_fraction, depth, true), false);
    predictOut << "Train misclassification: " << gbtrees.calc_error (&cvml,
        CV_TRAIN_ERROR, 0) << "%" << endl;
    predictOut << "Test misclassification: " << gbtrees.calc_error (&cvml,
        CV_TEST_ERROR, 0) << "%" << endl;

    cout << second_clock::local_time() <<
        " ======Training Completed=====" << endl;
    predictOut.close();

    /* Save the model if specified */
    if (save_model) {
        gbtrees.save (SAVE_MODEL_XML.c_str());
    }

    return true;
}

/******************************************************************************
MODULE: predictModel (class PredictBurnedArea)

PURPOSE: Run the gradient boosted regression tree model predictions to
obtain the probability mappings for burned probabilities.

RETURN VALUE:
Type = bool
Value          Description
-----          -----------
false          Error running the model
true           Model ran successfully

HISTORY:
Date          Programmer       Reason
----------    ---------------  -------------------------------------
11/26/2012    Jodi Riegle      Original development
9/3/2013      Gail Schmidt     Modified to work in the ESPA environment
                               Modified to write probability mappings vs.
                               simple burn/unburned classifications
9/10/2013     Gail Schmidt     Modified to use the cfmask QA values which are
                               more accurate than the SR QA values
12/8/2013     Gail Schmidt     Backed out the use of the cfmask and returned
                               to using the LEDAPS SR QA values
12/8/2013     Gail Schmidt     Added support for adjacent cloud shadow as one
                               of the QA values
4/7/2014      Gail Schmidt     Using a single QA/mask band now which is int16
                               vs. the old uint8 masks

NOTES:
  1. It's assumed the model has already been trained and/or loaded.
*****************************************************************************/
bool PredictBurnedArea::predictModel
(
    int iline,            /* I: line to be processed (0-based) */
    Output_t *output      /* O: 'output' data structure where buf contains
                                the probability mapping values */
)
{
    int bnd;                     /* band/index looping variable */
    int season;                  /* season looping variable */
    int indx;                    /* indices looping variable */
    int sample_indx;             /* current sample index for stacking data */
    char errmsg[MAX_STR_LEN];    /* error message */
    cv::Mat sample (1,NCSV_INPUTS+1,CV_32FC1);
                                 /* cvMat to hold the stacks of prediction
                                    information for each of the samples; size
                                    of this must be the same size as the array
                                    of data set to the training module */

    /* Loop through the predicted matrix rows which currently represent
       the samples in the input image.  The columns represent each band. */
    for( int y = 0; y < predMat.rows; y++ ) {
        /* Add the surface reflectance and indices */
        sample.at<float>(0) = predMat.at<float>(y,PREDMAT_B1);
        sample.at<float>(1) = predMat.at<float>(y,PREDMAT_B2);
        sample.at<float>(2) = predMat.at<float>(y,PREDMAT_B3);
        sample.at<float>(3) = predMat.at<float>(y,PREDMAT_B4);
        sample.at<float>(4) = predMat.at<float>(y,PREDMAT_B5);
        sample.at<float>(5) = predMat.at<float>(y,PREDMAT_B7);
        sample.at<float>(6) = predMat.at<float>(y,PREDMAT_NDVI);
        sample.at<float>(7) = predMat.at<float>(y,PREDMAT_NDMI);
        sample.at<float>(8) = predMat.at<float>(y,PREDMAT_NBR);
        sample.at<float>(9) = predMat.at<float>(y,PREDMAT_NBR2);

        /* Add the last year seasonal summaries, and add them as a group of
           bands/indices per season. */
        sample_indx = PREDMAT_NBR2+1;
        for (season = 0; season < PBA_NSEASONS; season++) {
            for (bnd = 0; bnd < PBA_NBANDS; bnd++) {
                sample.at<float>(sample_indx++) =
                    lySummaryMat.at<float>(y,season*PBA_NBANDS+bnd);
            }
        }

        /* Add the last year annual maximums for the indices */
        for (indx = 0; indx < PBA_NINDXS; indx++)
            sample.at<float>(sample_indx++) = maxIndxMat.at<float>(y,indx);

        /* Add the deltas of the annual maximums for the indices */
        if (qaMat.at<short>(y) == INPUT_FILL_VALUE) { // fill
            for (indx = 0; indx < PBA_NINDXS; indx++)
                sample.at<float>(sample_indx++) = INPUT_FILL_VALUE;
        }
        else { // not fill
            for (indx = 0; indx < PBA_NINDXS; indx++) {
                sample.at<float>(sample_indx++) =
                    predMat.at<float>(y,PREDMAT_NDVI+indx) -
                    maxIndxMat.at<float>(y,indx);
            }
        }

        /* Validate that the stack is not larger than the sample matrix was
           set up for */
        if (sample_indx > NCSV_INPUTS) {
            sprintf (errmsg, "The number of bands stacked in this sample "
                "(%d) is greater than the defined matrix size (%d).",
            sample_indx, NCSV_INPUTS);
            RETURN_ERROR (errmsg, "predict_model", false);
        }

        /* If the current pixel isn't cloudy, water, or fill, then run the
           prediction for this pixel. If the pixel is cloud, shadow, or water,
           then set it to PBA_CLOUD_WATER. If the pixel is fill then set it to
           PBA_FILL. */
        if (qaMat.at<short>(y) == INPUT_FILL_VALUE)  /* fill pixel */
            output->buf[y] = PBA_FILL;
        else if (qaMat.at<short>(y) < 0)   /* cloudy, snow, or water pixel */
            output->buf[y] = PBA_CLOUD_WATER;
        else {  /* do the probability mapping for burned (class of 1) */
            float response = gbtrees.predict_prob (sample, 1);
            output->buf[y] = (int16) (response * 100.0 + 0.5);
        }
    }

    /* Write the line of probability mappings to the output file */
    PutOutputLine (output, iline);
    sample.release();

    return true;
}

max_stars_repo_name: heyyy/burned-area

max_issues_repo_name: heyyy/burned-area

max_forks_repo_name: heyyy/burned-area


text:
// Copyright (c) 2008-2018 LG Electronics, Inc.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
//
// SPDX-License-Identifier: Apache-2.0

//
// test client using uMediaServer API
//
// Listen Only clients may use MediaChangeListener independently
//

#include <iostream>
#include <string>
#include <iomanip>
#include <thread>
#include <pbnjson.h>
#include <pbnjson.hpp>
#include <UMSConnector.h>
#include <GenerateUniqueID.h>
#include <boost/lexical_cast.hpp>
#include <list>

using namespace std;
using namespace pbnjson;
using namespace uMediaServer;

static const char* UMS_TOOLS_CONNECTION_BASE_ID = "com.webos.ums-status.";
static const char* CMD_GETACTIVE_PIPELINES = "palm://com.webos.media/getActivePipelines";
UMSConnector* connector;
#define ALIGN_INDENT 30

// This pretty_print function comes from luna-service component.
#define INDENT_INCREMENT 4
static void pretty_print(jvalue_ref object, int first_indent, int indent)
{
	if (!object) {
		printf("%*s<NULL>", first_indent, "");
		return;
	}

	if (!jis_array(object) && !jis_object(object))
	{
		printf("%*s%s", first_indent, "", jvalue_tostring_simple(object));
	}
	else if (jis_array(object))
	{
		int len = jarray_size(object);
		int i;
		printf("%*s[", first_indent, "");
		bool first = true;
		for (i=0;i<len;i++) {
			if (first) {
				printf("\n");
				first = false;
			} else {
				printf(",\n");
			}
			pretty_print(jarray_get(object, i), indent + INDENT_INCREMENT, indent + INDENT_INCREMENT);
		}
		printf("\n%*s]", indent, "");
	}
	else if (jis_object(object))
	{
		printf("%*s{", first_indent, "");
		bool first = true;
		jobject_iter it;
		(void)jobject_iter_init(&it, object);/* TODO: handle appropriately */
		jobject_key_value keyval;
		while (jobject_iter_next(&it, &keyval)) {
			if (first) {
				printf("\n");
				first = false;
			} else {
				printf(",\n");
			}
			// FIXME: contents of key are not being escaped
			raw_buffer key = jstring_get_fast(keyval.key);
			printf("%*s\"%.*s\": ", indent+INDENT_INCREMENT, "", (int)key.m_len, key.m_str);
			pretty_print(keyval.value, 0, indent + INDENT_INCREMENT);
		}
		printf("\n%*s}", indent, "");
	}
	else
	{
		printf("%*s<unknown json type>", first_indent, "");
	}
}

void print_left(int width, const string& text)
{
	ios::fmtflags f( cout.flags() );
	cout << std::left << std::setw(width) << text;
	cout.flags( f );
}

void print_right(int width, const string& text)
{
	ios::fmtflags f( cout.flags() );
	cout << std::right << std::setw(width) << text;
	cout.flags( f );
}

bool cb_(UMSConnectorHandle* handle, UMSConnectorMessage* message, void* ctx)
{
	string msg_received = connector->getMessageText(message);

	JDomParser parser;
	typedef struct {
		int i;
		int64_t timestamp;
	} index_time_map_t;
	std::list<index_time_map_t> time_index_map;

	auto mru_to_lru = [&](const index_time_map_t a,
			const index_time_map_t b)->bool {
		return (a.timestamp > b.timestamp);
	};

	if (!parser.parse(msg_received, pbnjson::JSchema::AllSchema())) {
		cout << "JDomParse.parse Error." << msg_received << endl;
		return false;
	}

	JValue parsed = parser.getDom();

	// no pipeline
	if (!parsed.isArray()) {
		cout << "No pipelines" << endl;
		return true;
	}

	for (int i=0; i < parsed.arraySize(); ++i) {
		index_time_map_t map_entry;
		map_entry.i = i;
		map_entry.timestamp = parsed[i]["timestamp"].asNumber<int64_t>();
		time_index_map.push_back(map_entry);
	}

	time_index_map.sort(mru_to_lru);

	int k = 1;
	for (auto it : time_index_map)
	{
		string id = parsed[it.i]["id"].asString();
		string type = parsed[it.i]["type"].asString();
		bool is_managed = parsed[it.i]["is_managed"].asBool();
		bool is_foreground = parsed[it.i]["is_foreground"].asBool();
		bool is_focus = parsed[it.i]["is_focus"].asBool();
		bool selected_by_policy = parsed[it.i]["selected_by_policy"].asBool();
		JValue resource = parsed[it.i]["resource"];

		cout << endl << "[ Pipeline : " << k++ << " ]" << endl << endl;

		print_right(ALIGN_INDENT, "id : "); print_left(ALIGN_INDENT, id);
		cout << endl;
		print_right(ALIGN_INDENT, "type : "); print_left(ALIGN_INDENT, type);
		cout << endl;
		print_right(ALIGN_INDENT, "is_managed : "); print_left(ALIGN_INDENT, (is_managed ? "true" : "false"));
		cout << endl;
		print_right(ALIGN_INDENT, "is_foreground : "); print_left(ALIGN_INDENT, (is_foreground ? "true" : "false"));
		cout << endl;
		print_right(ALIGN_INDENT, "is_focus: "); print_left(ALIGN_INDENT, (is_focus? "true" : "false"));
		cout << endl;
		print_right(ALIGN_INDENT, "selected_by_policy : "); print_left(ALIGN_INDENT, (selected_by_policy ? "true" : "false"));
		cout << endl;

		// print resources
		print_right(ALIGN_INDENT, "resources : ");
		std::stringstream stream;
		if (resource.isArray()) {
			for (int j=0; j < resource.arraySize(); ++j) {
				stream << resource[j]["resource"].asString() << "[" << resource[j]["index"].asNumber<int32_t>() << "] ";
			}
			print_left(ALIGN_INDENT, stream.str());
		}
		cout << endl; cout << endl;

		if (is_managed) {
			string uri = parsed[it.i]["uri"].asString();
			int32_t pid = parsed[it.i]["pid"].asNumber<int32_t>();
			string processState = parsed[it.i]["processState"].asString();
			string mediaState = parsed[it.i]["mediaState"].asString();
			string appId = parsed[it.i]["appId"].asString();

			print_right(ALIGN_INDENT, "uri : "); print_left(ALIGN_INDENT, uri);
			cout << endl;
			print_right(ALIGN_INDENT, "pid : "); print_left(ALIGN_INDENT, boost::lexical_cast<string>(pid));
			cout << endl;
			print_right(ALIGN_INDENT, "processState : "); print_left(ALIGN_INDENT, processState);
			cout << endl;
			print_right(ALIGN_INDENT, "mediaState : "); print_left(ALIGN_INDENT, mediaState);
			cout << endl;
			print_right(ALIGN_INDENT, "appId : "); print_left(ALIGN_INDENT, appId);
			cout << endl;
		}
	}

	return true;
}

int main(int argc, char *argv[])
{
	string input = "";

	string uid = "";

	try {
		uid = GenerateUniqueID()();
	}
	catch (const std::exception& e) {
		cout << "Exception while calling GenerateUniqueID: " << e.what() << endl;
	}

	string connection_id = UMS_TOOLS_CONNECTION_BASE_ID + uid;
	connector = new UMSConnector(connection_id, NULL, NULL, UMS_CONNECTOR_PRIVATE_BUS);

	try {
		std::thread([&] {
			connector->wait();
		}).detach();
	}
	catch (const std::exception& e) {
		cout << "Exception while calling detach: " << e.what() << endl;
	}

	pbnjson::JValue args = pbnjson::Object();
	JGenerator serializer(NULL);
	string payload_serialized;

	if (!serializer.toString(args, pbnjson::JSchema::AllSchema(), payload_serialized)) {
		cout << "failed serializer.toString()" << endl;
		return 0;
	}

	cout << "Test program start. type \"q\" for quit" << endl << endl;

	while (true) {
		bool rv = connector->sendMessage(CMD_GETACTIVE_PIPELINES, payload_serialized, cb_, NULL /*ctx*/);
		if (false == rv) {
			cout << "msg send fail" << endl;
			continue;
		}

		try {
			getline(cin, input);
		}
		catch (const std::exception& e)
		{
			cout << "Exception caught calling getline: " << e.what() << endl;
		}

		if (input == "exit" || input == "quit" || input == "q") {
			connector->stop();
			break;
		}
		cin.clear();
	}

    cout << "Test program end" << endl;

    return 0;
}

max_stars_repo_name: webosce/umediaserver

max_issues_repo_name: webosce/umediaserver

max_forks_repo_name: webosce/umediaserver