query.cpp

//================================================================================
// Name        : query.cpp
// Author      : Axel Auweter
// Copyright   : Leibniz Supercomputing Centre
// Description : Implementation of query class of dcdbquery
//================================================================================

//================================================================================
// This file is part of DCDB (DataCenter DataBase)
// Copyright (C) 2011-2016 Leibniz Supercomputing Centre
//
// This program is free software; you can redistribute it and/or
// modify it under the terms of the GNU General Public License
// as published by the Free Software Foundation; either version 2
// of the License, or (at your option) any later version.
//
// This program is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
// GNU General Public License for more details.
//
// You should have received a copy of the GNU General Public License
// along with this program; if not, write to the Free Software
// Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301, USA.
//================================================================================

#include <iostream>
#include <list>
#include <string>
#include <algorithm>

#include <cstdlib>
#include <cinttypes>

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

#include "dcdbendian.h"
#include "query.h"

void DCDBQuery::setLocalTimeEnabled(bool enable) {
  useLocalTime = enable;
}

bool DCDBQuery::getLocalTimeEnabled() {
  return useLocalTime;
}

void DCDBQuery::setRawOutputEnabled(bool enable) {
  useRawOutput = enable;
}

bool DCDBQuery::getRawOutputEnabled() {
  return useRawOutput;
}

void DCDBQuery::setFloatOutputEnabled(bool enable) {
  useFloatOutput = enable;
}

bool DCDBQuery::getFloatOutputEnabled() {
  return useFloatOutput;
}

void DCDBQuery::genOutput(DCDB::SensorDataStoreReading& reading)
{
  double fvalue;
  int64_t ivalue;

  /* Assign the reading to local variable */
  if (useFloatOutput) {
      fvalue = reading.value;
  }
  else {
      ivalue = reading.value;
  }

  /* Convert the unit if requested */
  if (unitConvert) {
      if (useFloatOutput) {
          if (!DCDB::UnitConv::convert(fvalue, baseUnit, targetUnit)) {
              std::cerr << "Warning, cannot convert units ("
                  << DCDB::UnitConv::toString(baseUnit) << " -> "
                  << DCDB::UnitConv::toString(targetUnit) << ")" << std::endl;
              unitConvert = false;
          }
      }
      else {
          if (!DCDB::UnitConv::convert(ivalue, baseUnit, targetUnit)) {
              std::cerr << "Warning, cannot convert units ("
                  << DCDB::UnitConv::toString(baseUnit) << " -> "
                  << DCDB::UnitConv::toString(targetUnit) << ")" << std::endl;
              unitConvert = false;
          }
      }
  }

  /* Scale the value if requested */
  if (scale) {
      if (useFloatOutput) {
          fvalue *= scalingFactor;
      }
      else {
          ivalue *= scalingFactor;
      }
  }

  /* Print the sensor's public name */
  std::cout << sensorName << ",";

  /* Print the time stamp */
  if (useLocalTime) {
      reading.timeStamp.convertToLocal();
  }
  if (useRawOutput) {
      std::cout << reading.timeStamp.getRaw() << ",";
  }
  else {
      std::cout << reading.timeStamp.getString() << ",";
  }

  /* Print the sensor value */
  if (useFloatOutput) {
      std::cout << fvalue << std::endl;
  }
  else {
      std::cout << ivalue << std::endl;
  }
}

void DCDBQuery::queryCallback(DCDB::SensorDataStoreReading& reading, void* userData)
{
  DCDBQuery* self = (DCDBQuery*)userData;

  /* Generate output for the reading */
  self->genOutput(reading);
}

void DCDBQuery::doQuery(const char* hostname, std::list<std::string> sensors, DCDB::TimeStamp start, DCDB::TimeStamp end)
{
  /* Create a new connection to the database */
  connection = new DCDB::Connection();
  connection->setHostname(hostname);
  if (!connection->connect()) {
      std::cout << "Cannot connect to database." << std::endl;
      exit(EXIT_FAILURE);
  }

  /* Initialize the SensorConfig interface */
  DCDB::SensorConfig sensorConfig(connection);
  DCDB::SensorDataStore sensorDataStore(connection);

  /* Print the CSV header */
  std::cout << "Sensor,Time,Value" << std::endl;

  /* Iterate over list of sensors requested by the user */
  for (std::list<std::string>::iterator it = sensors.begin(); it != sensors.end(); it++) {
      unitConvert = false;
      scale = false;
      scalingFactor = 1;
      std::string modifierStr;
      baseUnit = DCDB::Unit_None;
      targetUnit = DCDB::Unit_None;

      /* Check if the sensor was requested in a different unit or with scaling factor */
      if (it->find('/') != std::string::npos) {
          modifierStr = it->substr(it->find('/')+1, it->length());

          /* Remove the modifier from the string */
          *it = it->substr(0, it->find('/'));

          /* Check what type of modificatino is requested */
          boost::regex e("\\.?[0-9]*", boost::regex::extended);
          if (boost::regex_match(modifierStr, e)) {
              scale = true;
              sscanf(modifierStr.c_str(), "%lf", &scalingFactor);
          }
          else {
              unitConvert = true;
              targetUnit = DCDB::UnitConv::fromString(modifierStr);

              DCDB::PublicSensor sen;
              sensorConfig.getPublicSensorByName(sen, it->c_str());
              baseUnit = DCDB::UnitConv::fromString(sen.unit);
          }
      }

      /* Lookup the sensor in the published sensors table */
      DCDB::PublicSensor publicSensor;
      switch (sensorConfig.getPublicSensorByName(publicSensor, it->c_str())) {
      case DCDB::SC_OK:
        break;
      case DCDB::SC_INVALIDSESSION:
        std::cout << "Invalid session." << std::endl;
        return;
      case DCDB::SC_UNKNOWNSENSOR:
        std::cout << "Unknown sensor: " << *it << std::endl;
        return;
      default:
        std::cout << "Unknown error." << std::endl;
        return;
      }

      /* If this is a virtual sensor... */
      if (publicSensor.is_virtual) {
          sensorName = *it;
          DCDB::VSensor vSen(connection, publicSensor);
          vSen.queryCB(queryCallback, this, start, end);
      }
      else {
          /* Expand the pattern into a list of existing sensors in the time range */
          std::list<DCDB::SensorId> sensorIds;
          switch (sensorConfig.getSensorListForPattern(sensorIds, publicSensor.pattern, start, end)) {
          case DCDB::SC_OK:
            break;
          case DCDB::SC_INVALIDPATTERN:
            std::cout << "Invalid pattern." << std::endl;
            return;
          default:
            std::cout << "Unknown error." << std::endl;
            return;
          }

          /* Iterate over the expanded list of sensorIds and output the results in CSV format */
          for (std::list<DCDB::SensorId>::iterator sit = sensorIds.begin(); sit != sensorIds.end(); sit++) {
              sensorName = *it;
              sensorDataStore.queryCB(queryCallback, this, *sit, start, end);
          }
      }
  }

  /*
   * Clean up
   */
  connection->disconnect();
  delete connection;

}

DCDBQuery::DCDBQuery()
{
  connection = nullptr;
  useLocalTime = false;
  useRawOutput = false;
  useFloatOutput = false;
}