epicsdb.cpp

Go to the documentation of this file.

#include "stdafx.h"
#include "Poco/SingletonHolder.h"
#include "Poco/Data/Common.h"
#include "Poco/Data/SessionPool.h"
#include "Poco/Data/MySQl/Connector.h"
#include "Poco/Data/RecordSet.h"

#include "selogger.h"
#include "epicsdb.h"
#include "icputils.h"

class EpicsSessionPool : public Poco::Data::SessionPool
{
public:
        EpicsSessionPool() : Poco::Data::SessionPool("MySQL", "host=localhost;port=3306;db=archive;user=report;password=$report;compress=true;auto-reconnect=true", 3, 32, 0)
        {
                Poco::Data::MySQL::Connector::registerConnector();
        }

        void customizeSession(Poco::Data::Session& session)
        {
//              session.setProperty("maxRetryAttempts", 1000000);
//              if (strlen(isisdb_per_session_sql) > 0)
//              {
//                      Poco::Data::Statement stmt(session);
//                      stmt << isisdb_per_session_sql;
//                      stmt.execute();
//                      poco_assert(stmt.done());
//              }
        }
};

static Poco::SingletonHolder<EpicsSessionPool> epics_session_pool;

//select eng_id from smpl_engine where name='inst_engine' 'block_engine'
//select grp_id from chan_grp where eng_id= and name='BLOCKS' 'INST'
//select channel_id,name,descr from channel where grp_id=

//select grp_id from chan_grp inner join smpl_eng on chan_grp.eng_id=smpl_eng.eng_id and smpl_eng.name='block_engine' and chan_grp.name='BLOCKS';


unsigned __stdcall run_epics_db(void* arg)
{
        using namespace Poco::Data;
        std::vector<std::string> channel_name, smpl_time;
        std::vector<float> float_val;
        std::vector<unsigned> nanosecs;
        unsigned nrows;
        epics_thread_data_t* icp_data = (epics_thread_data_t*)arg;
        bool in_retry = false;
//      strcpy(icp_data->lastread_iso, "2014-06-09T10:00:00");
        while(true)
        {
                try
                {
                        Session epics_session(epics_session_pool.get()->get());
                        in_retry = false;
                        while( epics_session.isConnected() )
                        {
                                smpl_time.clear();
                                channel_name.clear(); 
                                float_val.clear();
                                nanosecs.clear();
                                try
                                {
                                        Statement epics_stmt1(epics_session), epics_stmt2(epics_session), epics_stmt3(epics_session);
                                        int grp_id; // group id for where SECI like block PVs will go - we need to check this each time as it changhes when new blocks are loaded
                                        epics_stmt1 << "SELECT grp_id FROM chan_grp INNER JOIN smpl_eng ON chan_grp.eng_id=smpl_eng.eng_id AND smpl_eng.name='block_engine' AND chan_grp.name='BLOCKS'", into(grp_id);
                                        epics_stmt1.execute();
                                    // poco 1.4 doesn't support mysql TIMESTAMP field, hence DATE_FORMAT operator to convert to string
                                        // when EPICS archiver is restarted, we can get a stream of NULL values followed bythe  real values, hence "is not null"
                                        epics_stmt2 << "SELECT channel.name, DATE_FORMAT(smpl_time, '%Y-%m-%dT%H:%i:%s'), nanosecs, float_val FROM channel INNER JOIN sample ON (channel.channel_id=sample.channel_id) " <<
                                       " AND (float_val IS NOT NULL) AND (grp_id=" << grp_id << ") AND (smpl_time = '" << icp_data->lastread_iso << "') AND (nanosecs > " << icp_data->lastread_nano << ") ORDER BY smpl_time, nanosecs", into(channel_name), into(smpl_time), into(nanosecs), into(float_val), limit(500);
                                        while( !epics_stmt2.done() )
                                        {
                                                nrows = epics_stmt2.execute();
                                                Poco::Thread::sleep(1000);
                                        }
                                        epics_stmt3 << "SELECT channel.name, DATE_FORMAT(smpl_time, '%Y-%m-%dT%H:%i:%s'), nanosecs, float_val FROM channel INNER JOIN sample ON (channel.channel_id=sample.channel_id) " <<
                                                   "  AND (float_val IS NOT NULL) AND (grp_id=" << grp_id << ") AND (smpl_time > '" << icp_data->lastread_iso << "') ORDER BY smpl_time, nanosecs", into(channel_name), into(smpl_time), into(nanosecs), into(float_val), limit(500);
                                        while( !epics_stmt3.done() )
                                        {
                                                nrows = epics_stmt3.execute();
                                                Poco::Thread::sleep(1000);
                                        }
                                        if (smpl_time.size() > 0)
                                        {
                                                strcpy(icp_data->lastread_iso, smpl_time.back().c_str());
                                                icp_data->lastread_nano = nanosecs.back();
                                                // erase PV prefix - it should be e.g. IN:LARMOR:CS:SB prefix
                                                for(int i=0; i<channel_name.size(); ++i)
                                                {
                                                        int n = channel_name[i].find_last_of(':');  
                                                        if (n != std::string::npos)
                                                        {
                                                                channel_name[i].erase(0, n+1);
                                                        }
                                                }
                                                se_log_fvalues(-1, "EPICS", smpl_time, channel_name, float_val);
                                                std::string log_file = std::string("c:\\data\\") + icp_data->inst_name + padWithZeros(*(icp_data->run_number), 8) + ".log";
                                                std::fstream f;
                                                f.open(log_file, std::ios::app);
                                                if (f.good())
                                                {
                                                        for(int i=0; i<smpl_time.size(); ++i)
                                                        {
                                                                f << smpl_time[i] << "\t" << channel_name[i] << "\t" << float_val[i] << std::endl;
                                                        }
                                                        f.close();
                                                }
                                        }
                                }
                                catch(const std::exception& ex)
                                {
                                        std::cerr << "error executing statement " << ex.what() << std::endl;
                                }
                                Poco::Thread::sleep(5000);
                        }
                        std::cerr << "EPICS MySQL session has disconnected " << std::endl;
                }
                catch(const std::exception& ex)
                {
                        if (!in_retry)
                        {
                                in_retry = true;
                                std::cerr << "Error creating EPICS MySQL session: " << ex.what() << ", will attempt to retry every 30 seconds" << std::endl;
                        }
                }
                Poco::Thread::sleep(30000);
        }
}

#if 0
select eng_id from smpl_engine where name='inst_engine'
select grp_id from chan_grp where eng_id= and name=

select channel_id,name,descr from channel where grp_id=
smpl_mode

select smpl_time,num_val,float_val_str_val,str_val ,arrav_val  status_id=  severity_id=  datatype      where channel_id=

select unit from num_metadata where channel_id=

slect enum_val from enum_metadata where channel_id= and enum_nbr=

# Dump all values for all channels
SELECT channel.name, smpl_time, severity.name, status.name, float_val
   FROM channel, severity, status, sample 
   WHERE channel.channel_id = sample.channel_id AND
         severity.severity_id = sample.severity_id AND
         status.status_id = sample.status_id
   ORDER BY channel.name, smpl_time
   LIMIT 50;

# Same with INNER JOIN
SELECT channel.name AS channel,
       smpl_time,
       severity.name AS severity,
       status.name AS status,
       float_val
   FROM sample INNER JOIN channel INNER JOIN severity INNER JOIN status
   ON channel.channel_id = sample.channel_id AND
      severity.severity_id = sample.severity_id AND
      status.status_id = sample.status_id
   ORDER BY smpl_time
   LIMIT 50;
#endif