TRestMySQLToAnalysisProcess.cxx

/*************************************************************************
 * This file is part of the REST software framework.                     *
 *                                                                       *
 * Copyright (C) 2016 GIFNA/TREX (University of Zaragoza)                *
 * For more information see http://gifna.unizar.es/trex                  *
 *                                                                       *
 * REST 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 3 of the License, or     *
 * (at your option) any later version.                                   *
 *                                                                       *
 * REST 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 a copy of the GNU General Public License along with   *
 * REST in $REST_PATH/LICENSE.                                           *
 * If not, see http://www.gnu.org/licenses/.                             *
 * For the list of contributors see $REST_PATH/CREDITS.                  *
 *************************************************************************/
 
#include "TRestMySQLToAnalysisProcess.h"
using namespace std;
 
#if defined USE_SQL
#include <mysql/mysql.h>
#endif
 
ClassImp(TRestMySQLToAnalysisProcess);
 
TRestMySQLToAnalysisProcess::TRestMySQLToAnalysisProcess() { Initialize(); }
 
TRestMySQLToAnalysisProcess::TRestMySQLToAnalysisProcess(const char* configFilename) {
    Initialize();
    LoadConfig(configFilename);
}
 
TRestMySQLToAnalysisProcess::~TRestMySQLToAnalysisProcess() {}
 
void TRestMySQLToAnalysisProcess::LoadDefaultConfig() {
    SetName(this->ClassName());
    SetTitle("Default config");
}
 
void TRestMySQLToAnalysisProcess::LoadConfig(const string& configFilename, const string& name) {
    if (LoadConfigFromFile(configFilename, name)) LoadDefaultConfig();
}
 
void TRestMySQLToAnalysisProcess::InitProcess() {
    fStartTimestamp = fRunInfo->GetStartTimestamp();
    fEndTimestamp = fRunInfo->GetEndTimestamp();
 
    if (fSQLVariables.size() > 0)
        FillDBArrays();
    else {
        RESTWarning
            << "TRestMySQLToAnalysisProcess::InitProcess. No data base field entries have been specified!"
            << RESTendl;
        RESTWarning << "This process will do nothing!" << RESTendl;
    }
}
 
void TRestMySQLToAnalysisProcess::Initialize() {
    SetSectionName(this->ClassName());
 
#ifndef USE_SQL
    RESTWarning << "TRestMySQLToAnalysisProcess. REST was compiled without mySQL support" << RESTendl;
    RESTWarning << "This process will not be funcional" << RESTendl;
#endif
}
 
TRestEvent* TRestMySQLToAnalysisProcess::ProcessEvent(TRestEvent* inputEvent) {
    fEvent = inputEvent;
 
#if defined USE_SQL
    RESTDebug << "TRestMySQLToAnalysisProcess. Ev ID : " << fEvent->GetID() << RESTendl;
    RESTDebug << "TRestMySQLToAnalysisProcess. Get timestamp : " << fEvent->GetTime() << RESTendl;
    for (int n = 0; n < fAnaTreeVariables.size(); n++)
        RESTDebug << "TRestMySQLToAnalysisProcess. Variable : " << fAnaTreeVariables[n]
                  << " value : " << GetDBValueAtTimestamp(n, fEvent->GetTime()) << RESTendl;
 
    if (GetVerboseLevel() >= TRestStringOutput::REST_Verbose_Level::REST_Extreme) GetChar();
 
    for (int n = 0; n < fAnaTreeVariables.size(); n++)
        SetObservableValue((string)fAnaTreeVariables[n], GetDBValueAtTimestamp(n, fEvent->GetTime()));
#else
    if (fCheckSQL) {
        RESTWarning << "REST was not linked to SQL libraries. Run cmake using -DREST_SQL=ON" << RESTendl;
        RESTWarning << "Clearing process metadata info." << RESTendl;
        RESTWarning << "Please, remove this process from the data chain or enable support for MySQL."
                    << RESTendl;
 
        fAnaTreeVariables.clear();
        fSQLVariables.clear();
        fDBServerName = "";
        fDBName = "";
        fDBUserName = "";
        fDBUserPass = "";
        fDBTable = "";
 
        fCheckSQL = false;
    }
#endif
 
    return fEvent;
}
 
void TRestMySQLToAnalysisProcess::InitFromConfigFile() {
    size_t pos = 0;
 
    fDBServerName = GetParameter("server", "");
    if (fDBServerName == "")
        RESTError << "TRestMySQLToAnalysisProcess. Database server name not found!" << RESTendl;
 
    fDBName = GetParameter("database", "");
    if (fDBName == "") RESTError << "TRestMySQLToAnalysisProcess. Database not found!" << RESTendl;
 
    fDBUserName = GetParameter("user", "");
    if (fDBUserName == "")
        RESTError << "TRestMySQLToAnalysisProcess. Database user name not found!" << RESTendl;
 
    fDBUserPass = GetParameter("password", "");
    if (fDBUserPass == "")
        RESTError << "TRestMySQLToAnalysisProcess. Database user password not found!" << RESTendl;
 
    fDBTable = GetParameter("table", "");
    if (fDBTable == "")
        RESTError << "TRestMySQLToAnalysisProcess. Database table name not found!" << RESTendl;
 
    string definition;
    while ((definition = GetKEYDefinition("dbEntry", pos)) != "") {
        TString sqlName = GetFieldValue("sqlName", definition);
        fSQLVariables.push_back((string)sqlName);
 
        TString anaName = GetFieldValue("anaName", definition);
        fAnaTreeVariables.push_back((string)anaName);
    }
}
 
void TRestMySQLToAnalysisProcess::PrintMetadata() {
    BeginPrintProcess();
 
    RESTMetadata << "SQL data extracted from:" << RESTendl;
    RESTMetadata << "- database : " << fDBName << RESTendl;
    RESTMetadata << "- table : " << fDBTable << RESTendl;
    RESTMetadata << " " << RESTendl;
 
    RESTMetadata << "List of variables added to the analysis tree" << RESTendl;
    RESTMetadata << " ------------------------------------------ " << RESTendl;
    for (unsigned int n = 0; n < fAnaTreeVariables.size(); n++) {
        RESTMetadata << " + SQL field : " << fSQLVariables[n] << RESTendl;
        RESTMetadata << "   - Tree name : " << fAnaTreeVariables[n] << RESTendl;
        RESTMetadata << "   - Min value : " << fMinValues[n] << RESTendl;
        RESTMetadata << "   - Max value : " << fMaxValues[n] << RESTendl;
        RESTMetadata << "  " << RESTendl;
    }
    EndPrintProcess();
}
 
void TRestMySQLToAnalysisProcess::FillDBArrays() {
#if defined USE_SQL
    MYSQL* conn = mysql_init(nullptr);
    if (conn == nullptr) {
        RESTError << "TRestMySQLToAnalysisProcess::InitProcess. mysql_init() failed" << RESTendl;
        exit(1);
    }
 
    if (!mysql_real_connect(conn, fDBServerName.c_str(), fDBUserName.c_str(), fDBUserPass.c_str(),
                            fDBName.c_str(), 0, nullptr, 0)) {
        RESTError << "TRestMySQLToAnalysisProcess::InitProcess. Connection to DB failed!" << RESTendl;
        RESTError << mysql_error(conn) << RESTendl;
        exit(1);
    }
 
    string sqlQuery = BuildQueryString();
    RESTDebug << sqlQuery << RESTendl;
    if (mysql_query(conn, sqlQuery.c_str())) {
        RESTError << "Error making query to SQL database" << RESTendl;
        RESTError << mysql_error(conn) << RESTendl;
        RESTError << "Query string : " << sqlQuery << RESTendl;
        exit(1);
    }
 
    MYSQL_RES* result = mysql_store_result(conn);
 
    if (result == nullptr) {
        RESTError << "Error getting result from SQL query" << RESTendl;
        RESTError << mysql_error(conn) << RESTendl;
        RESTError << "Query string : " << sqlQuery << RESTendl;
        exit(1);
    }
 
    int num_fields = mysql_num_fields(result);
    int num_rows = mysql_num_rows(result);
 
    // We double the average sampling of data in the given range
    fSampling = (fEndTimestamp - fStartTimestamp) / num_rows / 2;
 
    if (num_rows < 3) {
        this->SetError("No DB entries found for the run period!");
        fDataBaseExists = false;
        return;
    }
 
    // We register all the data inside a std::vector.
    MYSQL_ROW row;
    std::vector<std::vector<Double_t>> data;
 
    while ((row = mysql_fetch_row(result))) {
        std::vector<Double_t> dataRow;
        for (int i = 0; i < num_fields; i++) {
            string value;
            if (row[i])
                value = row[i];
            else
                value = "NULL";
 
            dataRow.push_back(StringToDouble(value));
        }
        data.push_back(dataRow);
    }
 
    fMinValues.clear();
    fMaxValues.clear();
    for (int j = 0; j < data.size(); j++) {
        if (j == 0)
            for (int n = 1; n < data.front().size(); n++) {
                fMinValues.push_back(data.front()[n]);
                fMaxValues.push_back(data.front()[n]);
            }
 
        for (int n = 1; n < data[j].size(); n++) {
            if (fMinValues[n - 1] > data[j][n]) fMinValues[n - 1] = data[j][n];
            if (fMaxValues[n - 1] < data[j][n]) fMaxValues[n - 1] = data[j][n];
        }
    }
 
    RESTDebug << "Raw data size " << data.size() << RESTendl;
 
    fDBdata.clear();
    cout.precision(10);
    // We finally fill the array with a fixed time bin (given by fSampling)
    Double_t timeNow = fStartTimestamp;
    Int_t dbEntry = 0;
    while (timeNow < fEndTimestamp) {
        while (timeNow > data[dbEntry][0] && dbEntry + 1 < data.size()) dbEntry++;
 
        std::vector<Double_t> dataBuff;
        if (timeNow < data.front()[0]) {
            for (int n = 1; n < data.front().size(); n++) dataBuff.push_back(data.front()[n]);
        } else if (timeNow > data.back()[0]) {
            for (int n = 1; n < data.back().size(); n++) dataBuff.push_back(data.back()[n]);
        } else {
            for (int n = 1; n < data[dbEntry].size(); n++) {
                double y2 = data[dbEntry][n];
                double y1 = data[dbEntry - 1][n];
 
                // Normalized field
                double x2 = data[dbEntry][0];
                double x1 = data[dbEntry - 1][0];
 
                double m = (y2 - y1) / (x2 - x1);
                double l = y1 - m * x1;
 
                dataBuff.push_back(m * timeNow + l);
            }
        }
 
        timeNow += fSampling;
        fDBdata.push_back(dataBuff);
    }
 
    RESTDebug << "Added entries : " << fDBdata.size() << RESTendl;
    if (GetVerboseLevel() >= TRestStringOutput::REST_Verbose_Level::REST_Debug)
        TRestTools::PrintTable(fDBdata, 0, 5);
 
    mysql_close(conn);
#else
    RESTWarning << "REST was not linked to SQL libraries. Run cmake using -DREST_SQL=ON" << RESTendl;
#endif
}
 
string TRestMySQLToAnalysisProcess::BuildQueryString() {
    string sqlQuery = "SELECT timestamp";
    for (unsigned int n = 0; n < fSQLVariables.size(); n++) {
        sqlQuery += ",";
        sqlQuery += fSQLVariables[n];
    }
    sqlQuery += " FROM " + fDBTable + " where timestamp between ";
 
    string startStr = Form("%10.0lf", fStartTimestamp);
    string endStr = Form("%10.0lf", fEndTimestamp);
 
    sqlQuery += startStr + " and " + endStr;
 
    return sqlQuery;
}
 
Double_t TRestMySQLToAnalysisProcess::GetDBValueAtTimestamp(Int_t index, Double_t timestamp) {
    if (!fDataBaseExists) return -1;
    Int_t bin = (Int_t)((timestamp - fStartTimestamp) / fSampling);
 
    if (bin < 0) return fDBdata.front()[index];
    if ((unsigned int)(bin + 1) >= fDBdata.size()) return fDBdata.back()[index];
 
    double y2 = fDBdata[bin + 1][index];
    double y1 = fDBdata[bin][index];
 
    double m = (y2 - y1) / fSampling;
    double l = y1 - m * (fStartTimestamp + bin * fSampling);
 
    return m * timestamp + l;
}