query.cpp

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

//================================================================================
// This file is part of DCDB (DataCenter DataBase)
// Copyright (C) 2011-2018 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"
#include "sensordatastore.h"
#include "sensoroperations.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::genOutput(std::list<DCDB::SensorDataStoreReading> &results)
{
	int64_t prev = 0;
    uint64_t prevT = 0;
    
    int64_t dx = 0;
    uint64_t dt = 0;
    
    /* Print Header */
    std::cout << "Sensor,Time";
    if(valueFormat.printValue) {
        std::cout << ",Value";
        if(valueFormat.unit != "none")
            std::cout << " (" << valueFormat.unit << ")";
    }
    
    if(deltaFormat.printValue) {
        std::cout << ",Delta";
        if(deltaFormat.unit != "none")
            std::cout << " (" << deltaFormat.unit << ")";
    }
    if(deltaTFormat.printValue) {
        std::cout << ",DeltaT";
        if(deltaTFormat.unit != "none")
            std::cout << " (" << deltaTFormat.unit << ")";
    }
    if(derivativeFormat.printValue) {
        std::cout << ",Derivative";
        if(derivativeFormat.unit != "none")
            std::cout << " (" << derivativeFormat.unit << ")";
    }
    if(integralFormat.printValue) {
        std::cout << ",Integral";
        if(integralFormat.unit != "none")
            std::cout << " (" << integralFormat.unit << ")";
    }
    
    /*End of header */
    std::cout << std::endl;
    
	for (std::list<DCDB::SensorDataStoreReading>::iterator reading = results.begin(); reading != results.end(); reading++) {

		/* 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(valueFormat.printValue) {
            
            int64_t result = (*reading).value;
            
            if(valueFormat.scalingFactor != 1.0 || baseScalingFactor != 1.0) {
                if( DCDB::scale(&result, valueFormat.scalingFactor, baseScalingFactor) == DCDB::DCDB_OP_OVERFLOW)
                    notifyOverflow = true;
            }
            
            /* Convert the unit if requested */
            if (DCDB::UnitConv::fromString(valueFormat.unit) != baseUnit) {
                
                if (!DCDB::UnitConv::convert((*reading).value, baseUnit, DCDB::UnitConv::fromString(valueFormat.unit)))
                    std::cerr << "Warning, cannot convert units ("
                    << DCDB::UnitConv::toString(baseUnit) << " -> "
                    << valueFormat.unit << ")"
                    << std::endl;
            }

            std::cout << "," << result;
        }

		/* Print Delta */
        if(deltaFormat.printValue && reading != results.begin()) {
            
            int64_t result;
            int64_t current = (*reading).value;
            int64_t previous = prev;
            
            if(deltaFormat.scalingFactor != 1.0) {
                if( DCDB::scale(&current, deltaFormat.scalingFactor, baseScalingFactor) == DCDB::DCDB_OP_OVERFLOW)
                    notifyOverflow = true;
                if( DCDB::scale(&previous, deltaFormat.scalingFactor, baseScalingFactor) == DCDB::DCDB_OP_OVERFLOW)
                    notifyOverflow = true;
            }
            
            if( DCDB::delta(current, previous, &result) == DCDB::DCDB_OP_OVERFLOW )
                notifyOverflow = true;
            
            if (DCDB::UnitConv::fromString(deltaFormat.unit) != baseUnit) {
                if (!DCDB::UnitConv::convert(result, baseUnit, DCDB::UnitConv::fromString(deltaFormat.unit)))
                    std::cerr << "Warning, cannot convert units ("
                    << DCDB::UnitConv::toString(baseUnit) << " -> "
                    << deltaFormat.unit << ")"
                    << std::endl;
            }
            
            std::cout << "," << result;
        }
        
        /* Print Delta T */
        if(deltaTFormat.printValue && reading != results.begin()) {
        
            int64_t result;
            int64_t current = (*reading).timeStamp.getRaw();
            int64_t previous = prevT;
            
            if(deltaTFormat.scalingFactor != 1.0 || baseScalingFactor != 1.0) {
                if( DCDB::scale(&current, deltaTFormat.scalingFactor, baseScalingFactor) == DCDB::DCDB_OP_OVERFLOW)
                    notifyOverflow = true;
                if( DCDB::scale(&previous, deltaTFormat.scalingFactor, baseScalingFactor) == DCDB::DCDB_OP_OVERFLOW)
                    notifyOverflow = true;
            }
            if( DCDB::delta(current, prevT, &result) == DCDB::DCDB_OP_OVERFLOW)
                notifyOverflow = true;
        
            if (DCDB::UnitConv::fromString(deltaTFormat.unit) != baseUnit) {
                if (!DCDB::UnitConv::convert(result, baseUnit, DCDB::UnitConv::fromString(deltaTFormat.unit)))
                    std::cerr << "Warning, cannot convert units ("
                    << DCDB::UnitConv::toString(baseUnit) << " -> "
                    << deltaTFormat.unit << ")"
                    << std::endl;
            }
            
            std::cout << "," << result;
        
        }
        
        /* Print Derivative */
        if(derivativeFormat.printValue && reading != results.begin()) {
            
            int64_t result;
            int64_t current = (*reading).value;
            int64_t previous = prev;
            
            /* No need to check overflow for dt as the scaling only applies to dx */
            if(derivativeFormat.scalingFactor != 1.0 || baseScalingFactor != 1.0) {
                if( DCDB::scale(&current, derivativeFormat.scalingFactor, baseScalingFactor) == DCDB::DCDB_OP_OVERFLOW)
                    notifyOverflow = true;
                if( DCDB::scale(&previous, derivativeFormat.scalingFactor, baseScalingFactor) == DCDB::DCDB_OP_OVERFLOW)
                    notifyOverflow = true;
            }
            
            if( DCDB::derivative(current, previous, (*reading).timeStamp.getRaw(), prevT, &result) == DCDB::DCDB_OP_OVERFLOW)
                notifyOverflow = true;
            
            if (DCDB::UnitConv::fromString(derivativeFormat.unit) != baseUnit) {
                if (!DCDB::UnitConv::convert(result, baseUnit, DCDB::UnitConv::fromString(derivativeFormat.unit)))
                    std::cerr << "Warning, cannot convert units ("
                    << DCDB::UnitConv::toString(baseUnit) << " -> "
                    << derivativeFormat.unit << ")"
                    << std::endl;
            }
            
            std::cout << "," << result;
            
        }
        
        /* Print Integral */
        if(integralFormat.printValue && reading != results.begin()) {
            
            int64_t result;
            int64_t current = (*reading).value;
            int64_t previous = prev;

            /* Very unlikely that dt > dx in case of overflow, so we just scale dx */
            if(integralFormat.scalingFactor != 1.0 || baseScalingFactor != 1.0) {
                if( DCDB::scale(&current, integralFormat.scalingFactor, baseScalingFactor) == DCDB::DCDB_OP_OVERFLOW)
                    notifyOverflow = true;
                if( DCDB::scale(&previous, integralFormat.scalingFactor, baseScalingFactor) == DCDB::DCDB_OP_OVERFLOW)
                    notifyOverflow = true;
            }
            
            if( DCDB::integral(current, previous, (*reading).timeStamp.getRaw(), prevT, &result) == DCDB::DCDB_OP_OVERFLOW)
                notifyOverflow = true;
            
            if (DCDB::UnitConv::fromString(integralFormat.unit) != baseUnit) {
                if (!DCDB::UnitConv::convert(result, baseUnit, DCDB::UnitConv::fromString(integralFormat.unit)))
                    std::cerr << "Warning, cannot convert units ("
                    << DCDB::UnitConv::toString(baseUnit) << " -> "
                    << integralFormat.unit << ")"
                    << std::endl;
            }
            
            std::cout << "," << result;
        }
		
        std::cout << std::endl;
            
        prev = (*reading).value;
        prevT = (*reading).timeStamp.getRaw();
    }
}


void DCDBQuery::checkModifier(std::list<std::string>::iterator it, double* scalingFactor, std::string* unit) {
    
    std::string modifierStr = it->substr(it->find('/')+1, it->length());
    
    /* Check what type of modification is requested */
    //boost::regex e("\\.?[0-9]*", boost::regex::extended);
    boost::regex e("[0-9]*\.?[0-9]*", boost::regex::extended);
    if (boost::regex_match(modifierStr, e))
        
        sscanf(modifierStr.c_str(), "%lf", scalingFactor);
   
    else
        unit = &modifierStr;
}

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

  /* Iterate over list of sensors requested by the user */
  for (std::list<std::string>::iterator it = sensors.begin(); it != sensors.end(); it++) {
      
      scalingFactor = 1;
      unit = "none";
      
      /* Output format initialization */
      valueFormat = {false,1,"none"};
      deltaFormat = {false,1,"none"};
      deltaTFormat = {false,1,"none"};
      derivativeFormat = {false,1,"none"};
      integralFormat = {false,1,"none"};
      
      std::string modifierStr;
      boost::smatch match;
      baseUnit = DCDB::Unit_None;
      
      sensorName = *it;
      
      DCDB::PublicSensor sen;
      sensorConfig.getPublicSensorByName(sen, it->c_str());
      
      /* Base scaling factor and unit of the public sensor */
      baseUnit = DCDB::UnitConv::fromString(sen.unit);
      baseScalingFactor = sen.scaling_factor;
      
      /* Retrieve sensor object first */
      boost::regex functExp("^delta|^delta_t|^derivative|^integral", boost::regex::extended);
      if(boost::regex_search(*it, match, functExp))
          sensorName = it->substr(it->find('(') + 1, it->size() - it->find('(') - (it->size() - it->find(')')) - 1 );
      else if (it->find('/') != std::string::npos)
          sensorName = it->substr(0, it->find('/'));
      
      DCDB::Sensor sensor(connection, sensorName);
      
      /* Iterate over the list to detect all instances of the sensor with name "sensorName" */
      while(it != sensors.end()) {
          
          if(it->find(sensorName) != std::string::npos) {
              
              /* Check first if the sensor we're looking is part of a function ...*/
              if(boost::regex_search(*it, match, functExp)) {
                  
                  if(match[0].compare("delta")==0) {
                      
                      deltaFormat.printValue = true;
                      
                      /* ...and for each function, check if we need a different scaling factor or unit... */
                      if(it->find('/') != std::string::npos) {
                          checkModifier(it, &scalingFactor, &unit);
                          deltaFormat.scalingFactor = scalingFactor;
                          deltaFormat.unit = unit;
                      }
                  }
                  else if(match[0].compare("delta_t")==0) {
                      
                      deltaTFormat.printValue = true;
                      
                      if(it->find('/') != std::string::npos) {
                          checkModifier(it, &scalingFactor, &unit);
                          deltaTFormat.scalingFactor = scalingFactor;
                          deltaTFormat.unit = unit;
                      }
                  }
                  else if(match[0].compare("derivative")==0) {
                      
                      derivativeFormat.printValue = true;
                      
                      if(it->find('/') != std::string::npos) {
                          checkModifier(it, &scalingFactor, &unit);
                          derivativeFormat.scalingFactor = scalingFactor;
                          derivativeFormat.unit = unit;
                      }
                  }
                  else if(match[0].compare("integral")==0) {
                      
                      integralFormat.printValue = true;
                      
                      if(it->find('/') != std::string::npos) {
                          checkModifier(it, &scalingFactor, &unit);
                          integralFormat.scalingFactor = scalingFactor;
                          integralFormat.unit = unit;
                      }
                  }
              }
              
              /* ...otherwise, print just the sensor values and check if they are requested in a different scaling factor or unit... */
              else {
                  
                  valueFormat.printValue = true;
                  
                  if(it->find('/') != std::string::npos) {
                      
                      checkModifier(it, &scalingFactor, &unit);
                      valueFormat.scalingFactor = scalingFactor;
                      valueFormat.unit = unit;
                  }
              }
              /*...remove this instance of the sensor from the list */
              it = sensors.erase(it);
          }
          else
              it++;
      }
      
      std::list<DCDB::SensorDataStoreReading> results;
      sensor.query(results, start, end, DCDB::AGGREGATE_NONE);
      genOutput(results);
      if(notifyOverflow)
          std::cout << "There has been overflow on some functions. " << std::endl;
  }

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

}

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