Jpp
 All Classes Namespaces Files Functions Variables Typedefs Enumerations Enumerator Friends Macros Pages
AcousticDataFilter.cpp
Go to the documentation of this file.
1 #include "AcousticDataFilter.h"
2 #ifdef DEBUG_ACOUSTICDATAFILTER
3 #include <stdlib.h>
4 #include <iostream>
5 #include <boost/date_time/posix_time/posix_time.hpp>
6 #include <Tools/KM3NeT_Debug.h>
7 #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 )
8 #else
9 #define DEBUG_MSG(str,lev) do { } while ( false )
10 #endif
11 
12 /**
13  * \author fsimeone
14  */
15 
16 class Parser
17 {
18 public:
19  Parser(std::string d,char t)
20  {
21  std::istringstream iss(d);
22  std::string token;
23 
24  while (std::getline(iss, token, t))
25  {
27  size_t pos = token.find('=');
28  std::string key=token.substr(0, pos);
29  std::string value=token.substr(pos+1);
32  mmap.insert(std::pair<std::string,std::string>(key,value));
33  }
34  }
35 
36  std::vector<std::string> Find(std::string tag)
37  {
40  i = mmap.equal_range(tag);
41  for (std::multimap<std::string,std::string>::iterator it=i.first; it!=i.second; ++it)
42  {
43  r.push_back(it->second);
44  }
45 
46  return r;
47  }
48 
49  std::string Find(std::string tag, size_t index, std::string value = "")
50  {
52 
53  if (result.size() > index)
54  return result[index];
55  else
56  return value;
57  }
58 
59 private:
61 };
62 
63 AcousticDataFilter::AcousticDataFilter( const std::string& address,\
64  const std::string& port,\
65  const std::string& name,\
66  const std::string& server,\
67  const std::string& wisdom,\
68  const std::string& output,\
69  const std::string& raw,\
70  JLOGGER::JLogger* logger,\
71  const int level): KM3NETDAQ::JDAQClient(name,server,logger,level),\
72  strand_(trigger_io_service_),\
73  server_acceptor_(server_io_service_)
74 {
75  DEBUG_MSG("Constructor",0);
76 
78 
79  address_=address;
80  port_=port;
81  exit_=false;
82  running_=false;
83  wisdom_file_=wisdom;
84  toa_file_path_=output;
85  debug_file_path_=raw;
86  Start();
87 }
88 
90 {
91  DEBUG_MSG("Destructor",0);
92  Stop();
93 }
94 
96 {
97  DEBUG_MSG("Start",2);
98  server_work_ = new boost::asio::io_service::work(server_io_service_);
99  trigger_work_ = new boost::asio::io_service::work(trigger_io_service_);
100 
101  boost::asio::ip::tcp::resolver l_resolver(server_io_service_);
102  boost::asio::ip::tcp::resolver::query l_query(address_,port_);
103  boost::asio::ip::tcp::endpoint l_endpoint = *l_resolver.resolve(l_query);
104  server_acceptor_.open(l_endpoint.protocol());
105  server_acceptor_.set_option(boost::asio::ip::tcp::acceptor::reuse_address(true));
106  server_acceptor_.bind(l_endpoint);
107  server_acceptor_.listen();
108 
109  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_)); }
110  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_)); }
111 }
112 
114 {
115  DEBUG_MSG("Stop",2);
116 
117  server_acceptor_.close();
118  delete server_work_;
119 
121  for (i=links_.begin(); i!=links_.end(); ++i){ (*i)->Disconnect(); }
122  server_io_service_.stop();
123  server_threads_.join_all();
124  server_io_service_.reset();
125 
126  delete trigger_work_;
127  trigger_io_service_.stop();
128  trigger_threads_.join_all();
129  trigger_io_service_.reset();
130 
131  links_.clear();
132  doms_.clear();
133  probes_.clear();
134 
135  if (toa_file_.is_open()) toa_file_.close();
136 }
137 
139 {
140  DEBUG_MSG("Start accept",2);
142  links_.push_back(l_link);
143  server_acceptor_.async_accept(l_link->Socket(), boost::bind(&AcousticDataFilter::Accept_completed, this, l_link, boost::asio::placeholders::error));
144 }
145 
146 void AcousticDataFilter::Accept_completed(pLink link, const boost::system::error_code& error)
147 {
148  DEBUG_MSG("Accept completed",2);
149 
150  if (!error)
151  {
152  link->Start_reading();
153  strand_.post(boost::bind(&AcousticDataFilter::Start_accept,this));
154  }
155 }
156 
157 void AcousticDataFilter::actionInit(int length, const char* buffer)
158 {
159  DEBUG_MSG("SM:Init",2);
160  FILE* pfile;
161  pfile=fopen(wisdom_file_.c_str(),"r");
162  if (pfile!=NULL)
163  {
164  if (fftwf_import_wisdom_from_file(pfile)==1)
165  {
166  DEBUG_MSG("SM:Init wisdom imported",2);
167  }
168  fclose(pfile);
169  }
170 }
171 
172 void AcousticDataFilter::actionReset(int length, const char* buffer)
173 {
174  DEBUG_MSG("SM:Reset",2);
175  FILE* pfile;
176  pfile=fopen(wisdom_file_.c_str(),"w");
177  if (pfile!=NULL)
178  {
179 DEBUG_MSG("SM:Reset wisdom saved",2);
180  fftwf_export_wisdom_to_file(pfile);
181  fclose(pfile);
182  }
183  links_.clear();
184  doms_.clear();
185  probes_.clear();
186  if (toa_file_.is_open()) toa_file_.close();
187 }
188 
189 void AcousticDataFilter::actionConfigure(int length, const char* buffer)
190 {
191  DEBUG_MSG("SM:Configure",2);
192  Parser dummy(std::string(buffer, length),';');
193  DEBUG_MSG("SM:Cerco",2);
194 
195  std::string l_debug_file_path;
196  std::string l_window;
197  std::string l_overlap;
198  std::vector<std::string> l_toa_path;
199  std::vector<std::string> l_dbg_path;
200  std::vector<std::string> l_probes;
201  std::vector<std::string> l_dom_configuration;
202 
203  l_window = dummy.Find("ADF_analysis_window_size", 0);
204  l_overlap = dummy.Find("ADF_analysis_window_overlap", 0);
205  l_probes = dummy.Find("ADF_waveform");
206  l_dom_configuration = dummy.Find("ADF_DOM_configuration");
207 
208  l_toa_path=dummy.Find("ADF_TOA_path");
209  if (l_toa_path.size()!=0) toa_file_path_=l_toa_path[0];
210  l_dbg_path=dummy.Find("ADF_DBG_path");
211  if (l_dbg_path.size()!=0) debug_file_path_=l_dbg_path[0];
212 
213  DEBUG_MSG("SM:Configure. win="<<l_window,2);
214  DEBUG_MSG("SM:Configure. ovr="<<l_overlap,2);
215  DEBUG_MSG("SM:Configure. TOA_path="<<toa_file_path_,2);
216  DEBUG_MSG("SM:Configure. DBG_path="<<debug_file_path_,2);
217 
218  DEBUG_MSG("SM:Configure. n dom="<<l_dom_configuration.size(),2);
219  DEBUG_MSG("SM:Configure. n pro="<<l_probes.size(),2);
220 
221  Create_probes(l_probes,atoi(l_window.c_str()));
222  Create_DOMs(l_dom_configuration,debug_file_path_,atoi(l_window.c_str()),atoi(l_overlap.c_str()));
223 
224 
225 
226 }
227 
228 void AcousticDataFilter::actionStart(int length, const char* buffer)
229 {
230  DEBUG_MSG("SM start",2);
231  running_=true;
232  strand_.post(boost::bind(&AcousticDataFilter::Start_accept,this));
233 }
234 
236 {
237  DEBUG_MSG("Create probes",1);
238 
239  for(int t=0;t<probes.size();++t)
240  {
241  std::istringstream l_iss(probes.at(t));
242  std::string l_line;
243  uint32_t l_id;
244  uint32_t l_samples;
245  real_type* l_buffer;
246 
247  std::getline(l_iss, l_line,' ');
248  l_id=atoi(l_line.c_str());
249  DEBUG_MSG("id probe="<<l_id,1);
250  std::getline(l_iss, l_line,' ');
251  l_samples=atoi(l_line.c_str());
252  DEBUG_MSG("n samples="<<l_samples,1);
253  l_buffer=(real_type*)malloc(l_samples*sizeof(real_type));
254  for (uint32_t i=0; i<l_samples; ++i)
255  {
256  std::getline(l_iss, l_line,' ');
257  real_type l_dummy=atof(l_line.c_str());
258  DEBUG_MSG("sample="<<l_dummy,0);
259  l_buffer[i]=l_dummy;
260  }
261  DEBUG_MSG("creating probes",1);
262  pProbe l_probe = boost::make_shared<KM3::TOALIB::CTOAFinder::probeSignalT>(*(new KM3::TOALIB::CTOAFinder::probeSignalT(window)));
263  DEBUG_MSG("copying data",1);
264  l_probe->copyFromBuffer(l_buffer,l_samples);
265  DEBUG_MSG("insert probe",1);
266  probes_.insert(probe_entry(l_id,l_probe));
267  }
268  DEBUG_MSG("Probes creation finished",1);
269 }
270 
271 void AcousticDataFilter::Create_DOMs(std::vector<std::string> dom_configuration,std::string& debug_file_path,std::size_t window,std::size_t overlap)
272 {
273  DEBUG_MSG("Create DOM",1);
274 
275  for(int t=0;t<dom_configuration.size();++t)
276  {
277  std::istringstream l_iss(dom_configuration.at(t));
278  std::string l_line;
279  uint32_t l_dom_id;
280  uint32_t l_probe_id;
281  uint32_t l_threshold;
282  uint32_t l_waveforms;
283  uint32_t l_test;
284 
285  std::getline(l_iss, l_line,' ');
286  l_dom_id=atoi(l_line.c_str());
287  DEBUG_MSG("DOM id="<<l_dom_id,1);
288  pDOM l_dom = boost::make_shared<KM3NeT_DOM>(*(new KM3NeT_DOM( trigger_io_service_,\
289  boost::bind(&AcousticDataFilter::Send_toa_async, this, _1),\
290  debug_file_path,\
291  window,\
292  overlap)));
293  doms_.insert(dom_entry(l_dom_id,l_dom));
294  std::getline(l_iss, l_line,' ');
295  l_waveforms=atoi(l_line.c_str());
296  DEBUG_MSG("n waveforms="<<l_waveforms,1);
297  for (uint32_t i=0; i<l_waveforms; ++i)
298  {
299  std::getline(l_iss, l_line,' ');
300  l_probe_id=atoi(l_line.c_str());
301  std::getline(l_iss, l_line,' ');
302  l_threshold=atoi(l_line.c_str());
303  DEBUG_MSG("adding probe "<<l_probe_id<<" with threshold "<<l_threshold<<" to DOM "<<l_dom_id,1);
304  l_dom->Add_probe(probes_.find(l_probe_id)->second,l_probe_id,l_threshold);
305  }
306  std::getline(l_iss, l_line,' ');
307  l_test=atoi(l_line.c_str());
308  DEBUG_MSG("setting test id="<<l_test,1);
309  l_dom->Set_test(l_test);
310  }
311  DEBUG_MSG("DOMs creation finished",1);
312 }
313 
315 {
316  if (!running_) return;
317 
318  DEBUG_MSG("Sent toa from ID="<<packet.Header.DOM_Identifier,3);
319 
320  if (!toa_file_.is_open()) Open_toa_file();
321 
322  if (toa_file_.tellp()>MAX_TOA_SIZE)
323  {
324  DEBUG_MSG("Toa file closed",1);
325  toa_file_.close();
326  Open_toa_file();
327  }
328 
329  if (toa_file_.is_open()) toa_file_.write((char*)&packet,sizeof(ToA_Packet));
330 
331 }
332 
334 {
335  std::string l_file=toa_file_path_;
336  std::time_t seconds = std::time(0);
337  l_file+="TOA_"+std::to_string((long long unsigned int)seconds)+".bin";
338  toa_file_.open(l_file.c_str(),std::ofstream::binary);
339  DEBUG_MSG("Toa file opened at "<<seconds,1);
340 }
341 
342 void AcousticDataFilter::actionPause(int length, const char* buffer)
343 {
344  DEBUG_MSG("SM: pause",2);
345  running_=false;
346 }
347 
348 void AcousticDataFilter::actionContinue(int length, const char* buffer)
349 {
350  DEBUG_MSG("SM: continue",2);
351  running_=true;
352 }
353 
354 void AcousticDataFilter::actionStop(int length, const char* buffer)
355 {
356  DEBUG_MSG("SM:Stop",2);
357 
358 }
359 
360 void AcousticDataFilter::actionQuit(int length, const char* buffer)
361 {
362  DEBUG_MSG("SM:Quit",2);
363 
364 }
365 
virtual void actionReset(int length, const char *buffer)
boost::atomic< bool > exit_
void Send_toa_async(ToA_Packet packet)
std::string Find(std::string tag, size_t index, std::string value="")
JDAQStateMachine::ev_configure_event ev_configure
virtual void actionInit(int length, const char *buffer)
boost::asio::io_service trigger_io_service_
std::map< std::size_t, pProbe > probes_
boost::asio::io_service::strand strand_
void Create_probes(std::vector< std::string > probes, std::size_t window)
boost::asio::io_service::work * server_work_
boost::thread_group trigger_threads_
Interface for logging messages.
Definition: JLogger.hh:22
void replaceEvent(const JTag &oldTag, const JTag &newTag, JDAQEvent_t &event)
Replace tag of given event in event table.
Definition: JDAQClient.hh:492
data_type r[M+1]
Definition: JPolint.hh:742
std::string trim(const std::string &buffer)
Trim string.
Definition: JLangToolkit.hh:79
virtual void actionQuit(int length, const char *buffer)
virtual void actionConfigure(int length, const char *buffer)
boost::shared_ptr< KM3NeT_DOM > pDOM
Definition: KM3NeT_DOM.h:65
do FILE
Definition: JFilter.sh:49
Parser(std::string d, char t)
return result
Definition: JPolint.hh:727
const JNET::JTag & clientTag() const
std::vector< pLink > links_
std::istream & getline(std::istream &in, JString &object)
Read string from input stream until end of line.
Definition: JString.hh:478
virtual void actionStop(int length, const char *buffer)
std::multimap< std::string, std::string > mmap
boost::atomic< bool > running_
#define MAX_TOA_SIZE
then echo n User name
Definition: JCookie.sh:45
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)
std::map< std::size_t, pDOM > doms_
then JMuonMCEvt f $INPUT_FILE o $INTERMEDIATE_FILE d
Definition: JMuonPath.sh:45
void Accept_completed(pLink link, const boost::system::error_code &error)
#define TRIGGER_THREAD_POOL
boost::thread_group server_threads_
boost::asio::ip::tcp::acceptor server_acceptor_
virtual void actionStart(int length, const char *buffer)
boost::asio::io_service server_io_service_
std::string to_string(const T &value)
Convert value to string.
void Send_toa(ToA_Packet packet)
std::string debug_file_path_
#define SERVER_THREAD_POOL
virtual void actionContinue(int length, const char *buffer)
boost::asio::io_service::work * trigger_work_
static const JNET::JTag RC_CMD
Definition: JDAQTags.hh:44
bool overlap(const JRange< T, JComparator_t > &first, const JRange< T, JComparator_t > &second)
Test overlap between ranges.
Definition: JRange.hh:651
void Create_DOMs(std::vector< std::string > doms, std::string &debug_file_path, std::size_t window, std::size_t overlap)
boost::shared_ptr< KM3::TOALIB::CTOAFinder::probeSignalT > pProbe
DAQ_Common_Header Header
std::vector< std::string > Find(std::string tag)
std::pair< std::size_t, pProbe > probe_entry
#define DEBUG_MSG(str, lev)
std::pair< std::size_t, pDOM > dom_entry
virtual void actionPause(int length, const char *buffer)