AcousticDataFilter.cc

Go to the documentation of this file.

#include "AcousticDataFilter.h"
#ifdef DEBUG_ACOUSTICDATAFILTER
#include <stdlib.h>
#include <iostream>
#include <boost/date_time/posix_time/posix_time.hpp>
#include <Tools/KM3NeT_Debug.h>
#define DEBUG_MSG(str,lev) do { if (((uint8_t)lev)>=AcousticDataFilter::DEBUG_LEVEL) { DBG_MSG << boost::posix_time::microsec_clock::local_time() << " ACOUSTICDATAFILTER(" << (void*)this << "): "<< str << '\n'; } }while( false )
#else
#define DEBUG_MSG(str,lev) do { } while ( false )
#endif

/**
 * \author fsimeone
 */

class Parser
{
public:
    Parser(std::string d,char t)
    {
        std::istringstream iss(d);
        std::string token;

        while (std::getline(iss, token, t))
        {
            boost::algorithm::trim(token);
            size_t pos   = token.find('=');
                        std::string key=token.substr(0, pos);
                        std::string value=token.substr(pos+1);
                        boost::algorithm::trim(key);
                        boost::algorithm::trim(value);
            mmap.insert(std::pair<std::string,std::string>(key,value));
        }
    }

    std::vector<std::string> Find(std::string tag)
    {
        std::vector<std::string> r;
        std::pair <std::multimap<std::string,std::string>::iterator, std::multimap<std::string,std::string>::iterator> i;
        i = mmap.equal_range(tag);
        for (std::multimap<std::string,std::string>::iterator it=i.first; it!=i.second; ++it)
        {
          r.push_back(it->second);
        }

        return r;
    }

  std::string Find(std::string tag, size_t index, std::string value = "")
  {
    std::vector<std::string> result = Find(tag);

    if (result.size() > index)
      return result[index];
    else
      return value;
  }

private:
        std::multimap<std::string,std::string> mmap;
};

AcousticDataFilter::AcousticDataFilter( const std::string& address,\
                                                                                const std::string& port,\
                                                                                const std::string& name,\
                                                                    const std::string& server,\
                                                                                const std::string& wisdom,\
                                                                                const std::string& output,\
                                                                                const std::string& raw,\
                                                                    JLOGGER::JLogger*           logger,\
                                                                    const int          level):  KM3NETDAQ::JDAQClient(name,server,logger,level),\
                                                                                                server_acceptor_(server_io_service_),
                                                                                                strand_(trigger_io_service_)
{
        DEBUG_MSG("Constructor",0);

        replaceEvent(KM3NETDAQ::RC_CMD, clientTag(), ev_configure);

        address_=address;
        port_=port;
        exit_=false;
        running_=false;
        wisdom_file_=wisdom;
        toa_file_path_=output;
        debug_file_path_=raw;
        Start();
}

AcousticDataFilter::~AcousticDataFilter()
{
        DEBUG_MSG("Destructor",0);
        Stop();
}

void AcousticDataFilter::Start()
{
        DEBUG_MSG("Start",2);
        server_work_ = new boost::asio::io_service::work(server_io_service_);
    trigger_work_ = new boost::asio::io_service::work(trigger_io_service_);

        boost::asio::ip::tcp::resolver l_resolver(server_io_service_);
        boost::asio::ip::tcp::resolver::query l_query(address_,port_);
        boost::asio::ip::tcp::endpoint l_endpoint = *l_resolver.resolve(l_query);
        server_acceptor_.open(l_endpoint.protocol());
        server_acceptor_.set_option(boost::asio::ip::tcp::acceptor::reuse_address(true));
        server_acceptor_.bind(l_endpoint);
        server_acceptor_.listen();

    for (std::size_t i=0; i<SERVER_THREAD_POOL; ++i)    {       server_threads_.create_thread(boost::bind(&boost::asio::io_service::run,&server_io_service_));  }
    for (std::size_t i=0; i<TRIGGER_THREAD_POOL; ++i)   {       trigger_threads_.create_thread(boost::bind(&boost::asio::io_service::run,&trigger_io_service_));        }
        strand_.post(boost::bind(&AcousticDataFilter::Start_accept,this));
}

void AcousticDataFilter::Stop()
{
        DEBUG_MSG("Stop",2);

        server_acceptor_.close();
        delete server_work_;

        std::vector<pLink>::iterator i;
        for (i=links_.begin(); i!=links_.end(); ++i){ (*i)->Disconnect(); }
        server_io_service_.stop();
        server_threads_.join_all();
        server_io_service_.reset();

        delete trigger_work_;
        trigger_io_service_.stop();
        trigger_threads_.join_all();
        trigger_io_service_.reset();

        links_.clear();
        doms_.clear();
        probes_.clear();

        if (toa_file_.is_open()) toa_file_.close();
}

void AcousticDataFilter::Start_accept()
{
        DEBUG_MSG("Start accept",2);
        pLink l_link=pLink(new KM3NeT_Acoustic_Link(server_io_service_,doms_));
        links_.push_back(l_link);
        server_acceptor_.async_accept(l_link->Socket(), boost::bind(&AcousticDataFilter::Accept_completed, this, l_link, boost::asio::placeholders::error));
}

void AcousticDataFilter::Accept_completed(pLink link, const boost::system::error_code& error)
{
        DEBUG_MSG("Accept completed",2);

        if (!error)
        {
                link->Start_reading();
                strand_.post(boost::bind(&AcousticDataFilter::Start_accept,this));
        }
}

void AcousticDataFilter::actionInit(int length, const char* buffer)
{
        DEBUG_MSG("SM:Init",2);
        FILE* pfile;
        pfile=fopen(wisdom_file_.c_str(),"r");
        if (pfile!=NULL)
        {
#ifdef TOAFINDER_USE_FLOAT
                if (fftwf_import_wisdom_from_file(pfile)==1)
#else
                if (fftw_import_wisdom_from_file(pfile)==1)
#endif
                {
                        DEBUG_MSG("SM:Init wisdom imported",2);
                }
                fclose(pfile);
        }
}

void AcousticDataFilter::actionReset(int length, const char* buffer)
{
        DEBUG_MSG("SM:Reset",2);
    FILE* pfile;
    pfile=fopen(wisdom_file_.c_str(),"w");
    if (pfile!=NULL)
    {
DEBUG_MSG("SM:Reset wisdom saved",2);
#ifdef TOAFINDER_USE_FLOAT
                fftwf_export_wisdom_to_file(pfile);
#else
                fftw_export_wisdom_to_file(pfile);
#endif
                fclose(pfile);
        }
        links_.clear();
        doms_.clear();
        probes_.clear();
        if (toa_file_.is_open()) toa_file_.close();
}

void AcousticDataFilter::actionConfigure(int length, const char* buffer)
{
        DEBUG_MSG("SM:Configure",2);
        Parser dummy(std::string(buffer, length),';');
        DEBUG_MSG("SM:Cerco",2);

        std::string l_debug_file_path;
        std::string l_window;
        std::string l_overlap;
        std::vector<std::string> l_toa_path;
        std::vector<std::string> l_dbg_path;
        std::vector<std::string> l_probes;
        std::vector<std::string> l_dom_configuration;

        l_window            = dummy.Find("ADF_analysis_window_size", 0);
        l_overlap           = dummy.Find("ADF_analysis_window_overlap", 0);
        l_probes            = dummy.Find("ADF_waveform");
        l_dom_configuration = dummy.Find("ADF_DOM_configuration");

        l_toa_path=dummy.Find("ADF_TOA_path");
        if (l_toa_path.size()!=0) toa_file_path_=l_toa_path[0];
        l_dbg_path=dummy.Find("ADF_DBG_path");
        if (l_dbg_path.size()!=0) debug_file_path_=l_dbg_path[0];

        DEBUG_MSG("SM:Configure. win="<<l_window,2);
        DEBUG_MSG("SM:Configure. ovr="<<l_overlap,2);
        DEBUG_MSG("SM:Configure. TOA_path="<<toa_file_path_,2);
        DEBUG_MSG("SM:Configure. DBG_path="<<debug_file_path_,2);

        DEBUG_MSG("SM:Configure. n dom="<<l_dom_configuration.size(),2);
        DEBUG_MSG("SM:Configure. n pro="<<l_probes.size(),2);

        Create_probes(l_probes,atoi(l_window.c_str()));
        Create_DOMs(l_dom_configuration,debug_file_path_,atoi(l_window.c_str()),atoi(l_overlap.c_str()));



}

void AcousticDataFilter::actionStart(int length, const char* buffer)
{
        DEBUG_MSG("SM start",2);
        running_=true;

}

void AcousticDataFilter::Create_probes(std::vector<std::string> probes,std::size_t window)
{
        DEBUG_MSG("Create probes",1);

        for(size_t t=0;t<probes.size();++t)
        {
                std::istringstream l_iss(probes.at(t));
                std::string l_line;
                uint32_t l_id;
                uint32_t l_samples;
                real_type* l_buffer;

                std::getline(l_iss, l_line,' ');
                l_id=atoi(l_line.c_str());
                DEBUG_MSG("id probe="<<l_id,1);
                std::getline(l_iss, l_line,' ');
                l_samples=atoi(l_line.c_str());
                DEBUG_MSG("n samples="<<l_samples,1);
                l_buffer=(real_type*)malloc(l_samples*sizeof(real_type));
                for (uint32_t i=0; i<l_samples; ++i)
                {
                        std::getline(l_iss, l_line,' ');
                        real_type l_dummy=atof(l_line.c_str());
                        DEBUG_MSG("sample="<<l_dummy,0);
                        l_buffer[i]=l_dummy;
                }
                DEBUG_MSG("creating probes",1);
                pProbe l_probe = boost::make_shared<KM3::TOALIB::CTOAFinder::probeSignalT>(*(new KM3::TOALIB::CTOAFinder::probeSignalT(window)));
                DEBUG_MSG("copying data",1);
                l_probe->copyFromBuffer(l_buffer,l_samples);
                DEBUG_MSG("insert probe",1);
                probes_.insert(probe_entry(l_id,l_probe));
        }
        DEBUG_MSG("Probes creation finished",1);
}

void AcousticDataFilter::Create_DOMs(std::vector<std::string> dom_configuration,std::string& debug_file_path,std::size_t window,std::size_t overlap)
{
        DEBUG_MSG("Create DOM",1);

        for(size_t t=0;t<dom_configuration.size();++t)
        {
                std::istringstream l_iss(dom_configuration.at(t));
                std::string l_line;
                uint32_t l_dom_id;
                uint32_t l_probe_id;
                uint32_t l_threshold;
                uint32_t l_waveforms;
                uint32_t l_test;

                std::getline(l_iss, l_line,' ');
                l_dom_id=atoi(l_line.c_str());
                DEBUG_MSG("DOM id="<<l_dom_id,1);
                pDOM l_dom = boost::make_shared<KM3NeT_DOM>(*(new KM3NeT_DOM(   trigger_io_service_,\
                                                                                                                                                boost::bind(&AcousticDataFilter::Send_toa_async, this, _1),\
                                                                                                                                                debug_file_path,\
                                                                                                                                                window,\
                                                                                                                                                overlap)));
                doms_.insert(dom_entry(l_dom_id,l_dom));
                std::getline(l_iss, l_line,' ');
                l_waveforms=atoi(l_line.c_str());
                DEBUG_MSG("n waveforms="<<l_waveforms,1);
                for (uint32_t i=0; i<l_waveforms; ++i)
                {
                        std::getline(l_iss, l_line,' ');
                        l_probe_id=atoi(l_line.c_str());
                        std::getline(l_iss, l_line,' ');
                        l_threshold=atoi(l_line.c_str());
                        DEBUG_MSG("adding probe "<<l_probe_id<<" with threshold "<<l_threshold<<" to DOM "<<l_dom_id,1);
                        l_dom->Add_probe(probes_.find(l_probe_id)->second,l_probe_id,l_threshold);
                }
        std::getline(l_iss, l_line,' ');
        l_test=atoi(l_line.c_str());
                DEBUG_MSG("setting test id="<<l_test,1);
                l_dom->Set_test(l_test);
        }
        DEBUG_MSG("DOMs creation finished",1);
}

void AcousticDataFilter::Send_toa(ToA_Packet packet)
{
        if (!running_) return;

    DEBUG_MSG("Sent toa from ID="<<packet.Header.DOM_Identifier,3);

        if (!toa_file_.is_open()) Open_toa_file();

        if (toa_file_.tellp()>MAX_TOA_SIZE)
        {
                DEBUG_MSG("Toa file closed",1);
                toa_file_.close();
                Open_toa_file();
        }

        if (toa_file_.is_open()) toa_file_.write((char*)&packet,sizeof(ToA_Packet));

}

void AcousticDataFilter::Open_toa_file()
{
        std::string l_file=toa_file_path_;
        std::time_t seconds = std::time(0);
        l_file+="TOA_"+std::to_string((long long unsigned int)seconds)+".bin";
        toa_file_.open(l_file.c_str(),std::ofstream::binary);
    DEBUG_MSG("Toa file opened at "<<seconds,1);
}

void AcousticDataFilter::actionPause(int length, const char* buffer)
{
        DEBUG_MSG("SM: pause",2);
        running_=false;
}

void AcousticDataFilter::actionContinue(int length, const char* buffer)
{
        DEBUG_MSG("SM: continue",2);
        running_=true;
}

void AcousticDataFilter::actionStop(int length, const char* buffer)
{
        DEBUG_MSG("SM:Stop",2);
        links_.clear();
        doms_.clear();
        probes_.clear();
        if (toa_file_.is_open()) toa_file_.close();


}

void AcousticDataFilter::actionQuit(int length, const char* buffer)
{
        DEBUG_MSG("SM:Quit",2);
        links_.clear();
        doms_.clear();
        probes_.clear();
        if (toa_file_.is_open()) toa_file_.close();


}