sensorcache.cpp

/*
 * SensorCache.cpp
 *
 *  Created on: 3 Aug 2016
 *      Author: ottmi
 */

#include "sensorcache.h"
#include <dcdb/timestamp.h>

#include <exception>
#include <iostream>
#include <iomanip>
#include <algorithm>

namespace DCDB {

SensorCache::SensorCache() {
	// TODO Auto-generated constructor stub

}

SensorCache::~SensorCache() {
	// TODO Auto-generated destructor stub
}

void SensorCache::storeSensor(SensorId sid, uint64_t ts, uint64_t val) {
	cacheEntry_t e;

	/* Remove the reserved bytes to leverage the standard find function */
	sid.setRsvd(0);
	sensorCache_t::iterator it = sensorCache.find(sid);
	if (it != sensorCache.end()) {
		e = it->second;
		if (e.front().timestamp+MAX_HISTORY_NS > ts) {
		  e.pop_front();
		}
	}

        sensorReading_t s = { val, ts };
        sensorCache[sid].push_back(s);
}

bool SensorCache::checkValid(cacheEntry_t &entry) {
  if (entry.size() > 2) {
    TimeStamp ts;
    uint64_t avg = 0;
    cacheEntry_t::iterator it = entry.begin();
    uint64_t prev = it->timestamp;
    for (it++; it != entry.end(); it++) {
      avg+= it->timestamp - prev;
      prev = it->timestamp;
    }
    avg/= (entry.size()-1);

    /*
     * A SID is outdated if it's older than 5x the average sampling period.
     */
    if ((ts.getRaw() - entry.back().timestamp) > 5 * avg) {
      return false;
    }
  }

  return true;
}

int64_t SensorCache::getAverage(cacheEntry_t &entry, uint64_t avg) {
  TimeStamp ts;

  if (entry.size() > 0) {
    //std::cout << ts.getRaw() - entry.back().timestamp << std::endl;
    if (ts.getRaw() - entry.back().timestamp > avg * NS_PER_S) {
      throw std::out_of_range("Sid outdated");
      return 0;
    }

    double sum = 0;
    cacheEntry_t::reverse_iterator it, prev;
    it = prev = entry.rbegin();
    it++;
    while ((it != entry.rend()) && ((ts.getRaw() - it->timestamp) <= avg * NS_PER_S)) {
      uint64_t deltaT = (prev->timestamp - it->timestamp);
      sum+= ((it->val + prev->val) / 2) * deltaT;
      //std::cout << "SensorCache::getAverage sum=" << sum << " deltaT=" <<deltaT << " it=(" << it->timestamp << "," <<it->val <<") prev=(" <<  prev->timestamp << "," << prev->val <<") " << (ts.getRaw() - it->timestamp) << std::endl;
      prev = it++;
    }

    //std::cout << "SensorCache::getAverage (" << prev->timestamp << "," <<prev->val <<") (" <<  entry.back().timestamp << "," << entry.back().val << ") sum=" << sum << " deltaT=" << entry.back().timestamp - prev->timestamp << std::endl;
    if (prev == entry.rbegin()) {
      return entry.back().val;
    } else {
      return sum/(entry.back().timestamp - prev->timestamp);
    }
  }
  throw std::invalid_argument("Sid not found");
  return 0;
}

uint64_t SensorCache::getSensor(SensorId sid, uint64_t avg) {
  /* Remove the reserved bytes to leverage the standard find function */
  sid.setRsvd(0);
  sensorCache_t::iterator it = sensorCache.find(sid);

  if (it == sensorCache.end()) {
    throw std::invalid_argument("Sid not found");
    return 0;
  }

  if (!checkValid(it->second))
  {
    throw std::out_of_range("Sid outdated");
    return 0;
  }

  if (avg) {
    return getAverage(it->second, avg);
  } else {
    return it->second.back().val;
  }
}

uint64_t SensorCache::getSensor(std::string topic, uint64_t avg) {
  topic.erase(std::remove(topic.begin(), topic.end(), '/'), topic.end());

  size_t wp = topic.find("*");
  if (wp == std::string::npos) {
    return getSensor(DCDB::SensorId(topic), avg);
  }

  int wl = 33 - topic.length();

  /* Create SensorIds with the lowest and highest values matching the wildcard */
  DCDB::SensorId sidLow(std::string(topic).replace(wp, 1, std::string(wl, '0')));
  DCDB::SensorId sidHi(std::string(topic).replace(wp, 1, std::string(wl, 'f')));
  DCDB::SensorId sidMask(std::string(wp, 'f') + std::string(wl, '0') + std::string(topic.length()-wp-1, 'f'));
  sidLow.setRsvd(0);
  sidHi.setRsvd(0);

  /* See whether there's a SensorId in the cache >= sidLow */
  sensorCache_t::iterator it = sensorCache.lower_bound(sidLow);
  sensorCache_t::iterator mostRecentSidIt = sensorCache.end();

  bool foundOne = false;
  /* Iterate over the cache until the current entry is > sidHi */
  while ((it != sensorCache.end()) && (it->first <= sidHi)) {
    if ((it->first & sidMask) == sidLow) {
      foundOne = true;
      /* We only return one value, even if multiple SensorIds would match.
       * At least make sure it's the most recent value
       */
      if (checkValid(it->second) && ((mostRecentSidIt == sensorCache.end()) || mostRecentSidIt->second.back().timestamp < it->second.back().timestamp)) {
        mostRecentSidIt = it;
      }
    }
    it++;
  }

  if (mostRecentSidIt == sensorCache.end()) {
    /* Check whether we actually found at least an outdated entry */
    if (foundOne) {
      throw std::out_of_range("Sid outdated");
    } else {
      throw std::invalid_argument("Sid not found");
    }
  }

  if (avg) {
    return getAverage(mostRecentSidIt->second, avg);
  } else {
    return mostRecentSidIt->second.back().val;
  }
}

void SensorCache::dump() {
  std::cout << "SensorCache Dump:" << std::endl;
  for (sensorCache_t::iterator sit = sensorCache.begin(); sit != sensorCache.end(); sit++) {
    std::cout << "  id=" << sit->first.toString() << " data=[";
    for (cacheEntry_t::iterator eit = sit->second.begin(); eit != sit->second.end(); eit++) {
      if (eit != sit->second.begin()) {
        std::cout << ",";
      }
      std::cout << "(" <<  eit->val << "," << eit->timestamp/NS_PER_S << "." << std::setfill ('0') << std::setw (9) << eit->timestamp%NS_PER_S << ")";
    }
    std::cout << "]" << std::endl;
  }


}

} /* namespace DCDB */