Merge branch 'development' of https://gitlab.lrz.de/dcdb/dcdb into development

analytics/operators/clustering/ClusteringOperator.cpp

@@ -177,10 +177,12 @@ void ClusteringOperator::compute(U_Ptr unit) {
 void ClusteringOperator::computeFeatureVector(U_Ptr unit, uint64_t offset) {
     _currentfVector = cv::Mat(1, unit->getInputs().size(), CV_32F);
     std::vector<ClusteringSBPtr>& inputs = unit->getInputs();
+    uint64_t endTs = getTimestamp() - offset;
+    uint64_t startTs = endTs - _aggregationWindow;
     for(size_t idx=0; idx<inputs.size(); idx++) {
         _mean=0;
         _buffer.clear();
-        if(!_queryEngine.querySensor(inputs[idx]->getName(), _aggregationWindow - offset, offset, _buffer) || _buffer.empty())
+        if(!_queryEngine.querySensor(inputs[idx]->getName(), startTs, endTs, _buffer, false) || _buffer.empty())
             throw std::runtime_error("Operator " + _name + ": cannot read from sensor " + inputs[idx]->getName() + "!");
         
         // Computing MEAN

analytics/operators/regressor/RegressorOperator.cpp

@@ -187,11 +187,13 @@ void RegressorOperator::computeFeatureVector(U_Ptr unit) {
         _currentfVector = new cv::Mat(1, unit->getInputs().size()*REG_NUMFEATURES, CV_32F);
     int64_t val;
     size_t qId, qMod, idx, fIdx;
+    uint64_t endTs = getTimestamp();
+    uint64_t startTs = endTs - _aggregationWindow;
     std::vector<RegressorSBPtr>& inputs = unit->getInputs();
     for(idx=0; idx<inputs.size(); idx++) {
         _mean=0; _std=0; _diffsum=0; _qtl25=0; _qtl75=0;
         _buffer.clear();
-        if(!_queryEngine.querySensor(inputs[idx]->getName(), _aggregationWindow, 0, _buffer) || _buffer.empty())
+        if(!_queryEngine.querySensor(inputs[idx]->getName(), startTs, endTs, _buffer, false) || _buffer.empty())
             throw std::runtime_error("Operator " + _name + ": cannot read from sensor " + inputs[idx]->getName() + "!");
         if (inputs[idx]->getTrainingTarget())
             _currentTarget = (float)_buffer.back().value;

analytics/operators/smoothing/SmoothingOperator.cpp

@@ -47,10 +47,11 @@ void SmoothingOperator::compute(U_Ptr unit) {
     // Clearing the buffer
     _buffer.clear();
     SmoothingSBPtr sIn=unit->getInputs()[0], sOut=unit->getOutputs()[0];
-    uint64_t startTs = sOut->getTimestamp() ? getTimestamp()-sOut->getTimestamp() : 0;
+    uint64_t endTs = getTimestamp();
+    uint64_t startTs = sOut->getTimestamp() ? sOut->getTimestamp() : endTs;
 
     // Throwing an error does not make sense here - the query will often fail depending on insert batching
-    if(!_queryEngine.querySensor(sIn->getName(), startTs, 0, _buffer, true))
+    if(!_queryEngine.querySensor(sIn->getName(), startTs, endTs, _buffer, false))
         return;
     
     for(const auto& v : _buffer) {