AcousticDataFilter.cc

Go to the documentation of this file.

#include "AcousticDataFilter.h"
#include "Jeep/JMessage.hh"
#include "parser.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
 */
 
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);
 
    for (auto i=0; i< l_probes.size(); i++) {
            DEBUG_MSG("SM:Configure. l_probes[" << i << "]=" << l_probes[i] << "\n",2);
    }
 
        int probe_errors = Create_probes(l_probes,atoi(l_window.c_str()));
    if (probe_errors) {
            ERROR(probe_errors << " errors encountered when reading probes from configuration\n");
    }
 
        int dom_errors = Create_DOMs(l_dom_configuration,debug_file_path_,atoi(l_window.c_str()),atoi(l_overlap.c_str()));
 
    if (dom_errors) {
            ERROR(dom_errors << " errors encountered when reading doms from configuration\n");
    }
 
}
 
void AcousticDataFilter::actionStart(int length, const char* buffer)
{
        DEBUG_MSG("SM start",2);
        running_=true;
 
}
 
int AcousticDataFilter::Create_probes(std::vector<std::string> probes,std::size_t window)
{
    DEBUG_MSG("Create probes",1);
 
    int nerrors{0};
 
    for(const auto& probeString: probes)
    {
        auto waveform = parseWaveformSamples<real_type>(probeString);
        if (waveform==std::nullopt) {
                ++nerrors;
                ERROR("could not parse probe line " << probeString << "\n");
                continue;
        }
        DEBUG_MSG("creating probe id "<<waveform->id,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(waveform->samples.data(),waveform->samples.size());
        DEBUG_MSG("insert probe id"<<waveform->id,1);
        probes_.insert(probe_entry(waveform->id,l_probe));
    }
    DEBUG_MSG("Probes creation finished",1);
    return nerrors;
}
 
int 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);
 
    int nerrors{0};
 
    for (const auto& line : dom_configuration) {
       auto config = parseDOMConfiguration(line);
 
       if (config==std::nullopt) {
           ++nerrors;
           ERROR("could not parse DOM configuration line " << line << "\n");
           continue;
       }
 
       const auto& [ dom_id, probe_configs, write_flag ] = config.value();
 
       DEBUG_MSG("DOM id="<<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(dom_id,l_dom));
 
       for (const auto& probe_config: probe_configs) {
          DEBUG_MSG("adding probe "<<probe_config.id<<" with threshold "<<probe_config.threshold<<" to DOM "<<dom_id,1);
          auto known_probe = probes_.find(probe_config.id);
          if (known_probe== probes_.end()) {
                 ERROR("probe id " << probe_config.id << " not found in the internal map ! Not adding it to our list of probes !" << "\n");
                 continue;
          }
          l_dom->Add_probe(known_probe->second,probe_config.id,probe_config.threshold);
       }
       DEBUG_MSG("setting test id="<<write_flag,1);
       l_dom->Set_test(write_flag);
    }
    DEBUG_MSG("DOMs creation finished",1);
    return nerrors;
}
 
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();
 
 
}