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();
 
 
}