//================================================================================
// Name        : AggregatorOperator.cpp
// Author      : Alessio Netti
// Contact     : info@dcdb.it
// Copyright   : Leibniz Supercomputing Centre
// Description :
//================================================================================

//================================================================================
// This file is part of DCDB (DataCenter DataBase)
// Copyright (C) 2019-2019 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 "AggregatorOperator.h"

AggregatorOperator::AggregatorOperator(const std::string& name) : OperatorTemplate(name) { 
    _window = 0;
    _buffer = nullptr;
}

AggregatorOperator::AggregatorOperator(const AggregatorOperator& other) : OperatorTemplate(other) {
    _window = other._window;
    _buffer = nullptr;
}

AggregatorOperator::~AggregatorOperator() {
    if(_buffer)
        delete _buffer;
}

void AggregatorOperator::printConfig(LOG_LEVEL ll) {
    LOG_VAR(ll) << "            Window:          " << _window;
    OperatorTemplate<AggregatorSensorBase>::printConfig(ll);
}

void AggregatorOperator::compute(U_Ptr unit) {
    // Clearing the buffer, if already allocated
    if(_buffer) 
        _buffer->clear();
    size_t elCtr=0;
    for(const auto& in : unit->getInputs()) {
        // Getting the most recent values as specified in _window
        // Since we do not clear the internal buffer, all sensor readings will be accumulated in the same vector
        elCtr = _buffer==nullptr ? 0 : _buffer->size();
        _buffer = _queryEngine.querySensor(in->getName(), _window, 0, _buffer);
        if(!_buffer || _buffer->size()<=elCtr)
            throw std::runtime_error("Operator " + _name + ": cannot read from sensor " + in->getName() + "!");
    }
    compute_internal(unit, _buffer);
}

void AggregatorOperator::compute_internal(U_Ptr unit, vector<reading_t> *buffer) {
    _quantileSensors.clear();
    reading_t reading;
    AggregatorSensorBase::aggregationOps_t op;
    reading.timestamp = getTimestamp();
    // Performing the actual aggregation operation
    for(const auto& out : unit->getOutputs()) {
        op = out->getOperation();
        if(op!=AggregatorSensorBase::QTL) {
            switch (op) {
                case AggregatorSensorBase::SUM:
                    reading.value = computeSum(_buffer);
                    break;
                case AggregatorSensorBase::AVG:
                    reading.value = computeAvg(_buffer);
                    break;
                case AggregatorSensorBase::MIN:
                    reading.value = computeMin(_buffer);
                    break;
                case AggregatorSensorBase::MAX:
                    reading.value = computeMax(_buffer);
                    break;
                case AggregatorSensorBase::STD:
                    reading.value = computeStd(_buffer);
                    break;
                case AggregatorSensorBase::OBS:
                    reading.value = computeObs(_buffer);
                    break;
                default:
                    LOG(warning) << _name << ": Encountered unknown operation!";
                    reading.value = 0;
                    break;
            }
            out->storeReading(reading);
        } else
            _quantileSensors.push_back(out);
    }

    if(!_quantileSensors.empty()) {
        vector<int64_t> result = computeQuantiles(_buffer);
        for(unsigned idx=0; idx<result.size(); idx++) {
            reading.value = result[idx];
            _quantileSensors[idx]->storeReading(reading);
        }
    }
}

int64_t AggregatorOperator::computeObs(vector<reading_t> *buffer) {
    return buffer->size();
}

int64_t AggregatorOperator::computeSum(vector<reading_t> *buffer) {
    int64_t acc=0;
    for(const auto& v : *buffer)
            acc += v.value;
    return acc;
}

int64_t AggregatorOperator::computeAvg(vector<reading_t> *buffer) {
    int64_t acc=0, ctr=buffer->size();
    for(const auto& v : *buffer)
        acc += v.value;
    acc = ctr > 0 ? acc/ctr : acc;
    return acc;
}

int64_t AggregatorOperator::computeMax(vector<reading_t> *buffer) {
    int64_t acc=0;
    bool maxInit=false;
    for(const auto& v : *_buffer)
        if(v.value>acc || !maxInit) {
            acc = v.value;
            maxInit = true;
        }
    return acc;
}

int64_t AggregatorOperator::computeMin(vector<reading_t> *buffer) {
    int64_t acc=0;
    bool minInit=false;
    for(const auto& v : *_buffer)
        if(v.value<acc || !minInit) {
            acc = v.value;
            minInit = true;
        }
    return acc;
}

int64_t AggregatorOperator::computeStd(vector<reading_t> *buffer) {
    int64_t avg = computeAvg(buffer);
    int64_t acc=0, val=0, ctr=buffer->size();
    for(const auto& v : *buffer) {
        val = v.value - avg;
        acc += val*val;
    }
    acc = ctr > 0 ? sqrt(acc/ctr) : sqrt(acc);
    return acc;
}

vector<int64_t> AggregatorOperator::computeQuantiles(vector<reading_t> *buffer) {
    size_t idx, mod;
    vector<int64_t> result;
    // Sorting the sensor reading buffer to extract quantiles
    std::sort(buffer->begin(), buffer->end(), [ ](const reading_t& lhs, const reading_t& rhs) { return lhs.value < rhs.value; });
    for(const auto& q : _quantileSensors) {
        idx = (buffer->size() * q->getQuantile()) / 100;
        mod = (buffer->size() * q->getQuantile()) % 100;
        result.push_back((mod==0 || idx==buffer->size()-1) ? buffer->at(idx).value : (buffer->at(idx).value + buffer->at(idx+1).value)/2);
    }
    return result;
}