34 f_data_type_size( 1 ),
36 f_record_size( 4096 ),
40 f_center_freq( 50.e6 ),
41 f_freq_range( 100.e6 ),
55 scarab::dig_calib_params t_dig_params;
56 scarab::get_calib_params( f_bit_depth, f_data_type_size, f_v_offset, f_v_range,
true, &t_dig_params );
58 vector< unsigned > t_chan_vec;
59 f_stream_no = a_hw_ptr->header().AddStream(
"Psyllid - ROACH2",
60 f_acq_rate, f_record_size, f_sample_size, f_data_type_size,
61 monarch3::sDigitizedS, f_bit_depth, monarch3::sBitsAlignedLeft, &t_chan_vec );
64 for( std::vector< unsigned >::const_iterator it = t_chan_vec.begin(); it != t_chan_vec.end(); ++it )
66 a_hw_ptr->header().GetChannelHeaders()[ *it ].SetVoltageOffset( t_dig_params.v_offset );
67 a_hw_ptr->header().GetChannelHeaders()[ *it ].SetVoltageRange( t_dig_params.v_range );
68 a_hw_ptr->header().GetChannelHeaders()[ *it ].SetDACGain( t_dig_params.dac_gain );
69 a_hw_ptr->header().GetChannelHeaders()[ *it ].SetFrequencyMin( f_center_freq - 0.5 * f_freq_range );
70 a_hw_ptr->header().GetChannelHeaders()[ *it ].SetFrequencyRange( f_freq_range );
80 butterfly_house::get_instance()->register_writer(
this, f_file_num );
114 if( a_midge ) a_midge->throw_ex( std::current_exception() );
121 midge::enum_t t_trig_command = stream::s_none;
122 midge::enum_t t_time_command = stream::s_none;
126 while( ! is_canceled() )
128 t_time_command = in_stream< 0 >().
get();
129 if( t_time_command == stream::s_none )
continue;
130 if( t_time_command == stream::s_error )
136 LTRACE(
plog,
"Triggered writer reading stream 0 (time) at index " << in_stream< 0 >().get_current_index() );
138 if( t_time_command == stream::s_exit )
140 LDEBUG(
plog,
"Triggered writer is exiting due to time-stream command; no run in progress" );
145 if( t_time_command == stream::s_stop )
147 LDEBUG(
plog,
"Triggered writer received stop command on the time stream while already stopped; no action taken" );
151 if( t_time_command == stream::s_run )
153 LWARN(
plog,
"Triggered writer received run command on the time stream while stopped; no action taken" );
157 if( t_time_command == stream::s_start )
159 LDEBUG(
plog,
"Triggered writer received start command on the time stream; looking for start command on the trigger stream" );
162 while( ! is_canceled() )
164 t_trig_command = stream::s_none;
165 for(
unsigned i_attempt = 0; i_attempt < 10 && t_trig_command != stream::s_start; ++i_attempt )
167 t_trig_command = in_stream< 1 >().
get();
168 LTRACE(
plog,
"(attempt " << i_attempt <<
") Triggered writer reading stream 1 (trig) at index " << in_stream< 1 >().get_current_index() );
171 if( t_trig_command == stream::s_start )
173 LDEBUG(
plog,
"Time and trig commands match: start" );
174 LINFO(
plog,
"Starting a run" );
176 LDEBUG(
plog,
"Will start file with next data" );
189 throw midge::node_nonfatal_error() <<
"Egg writer received unexpected trig-stream command while waiting for start: " << t_trig_command;
195 LDEBUG(
plog,
"Breaking out of not-running exe loop" );
207 midge::enum_t t_trig_command = stream::s_none;
208 midge::enum_t t_time_command = stream::s_none;
213 uint64_t t_bytes_per_record = f_record_size * f_sample_size * f_data_type_size;
214 uint64_t t_record_length_nsec = llrint( (
double)(
PAYLOAD_SIZE / 2) / (
double)f_acq_rate * 1.e3 );
216 while( ! is_canceled() )
218 t_trig_command = in_stream< 1 >().
get();
219 LTRACE(
plog,
"Egg writer reading stream 1 (trig) at index " << in_stream< 1 >().get_current_index() );
221 if( t_trig_command == stream::s_none )
223 LDEBUG(
plog,
"Egg writer received none command on the trig stream while run is in progress");
224 t_time_command = in_stream< 0 >().
get();
225 LTRACE(
plog,
"Egg writer reading stream 0 (time) at index " << in_stream< 0 >().get_current_index() );
226 LDEBUG(
plog,
"Advancing time stream; time command matched trig command? trig command = " << t_trig_command <<
"; time command = " << t_time_command );
227 if( t_time_command != stream::s_none )
235 throw error() <<
"Trig command doesn't match time command: time command = " << t_time_command <<
"; trig command = " << t_trig_command;
239 if( t_trig_command == stream::s_error )
245 if( t_trig_command == stream::s_exit )
247 LDEBUG(
plog,
"Egg writer is exiting due to trig-stream command; run is in progress" );
257 if( t_trig_command == stream::s_stop )
259 LDEBUG(
plog,
"Egg writer received stop command on the trig stream while run is in progress" );
265 t_time_command = in_stream< 0 >().
get();
266 LTRACE(
plog,
"Egg writer reading stream 0 (time) at index " << in_stream< 0 >().get_current_index() );
267 LDEBUG(
plog,
"Advancing time stream; time command matched trig command? trig command = " << t_trig_command <<
"; time command = " << t_time_command );
268 LDEBUG(
plog,
"Breaking out of is-running exe loop" );
273 if( t_trig_command == stream::s_start )
275 LERROR(
plog,
"Egg writer received unexpected start command on the trig stream while running")
281 t_time_command = in_stream< 0 >().
get();
282 LTRACE(
plog,
"Egg writer reading stream 0 (time) at index " << in_stream< 0 >().get_current_index() );
283 LDEBUG(
plog,
"Advancing time stream; time command matched trig command? trig command = " << t_trig_command <<
"; time command = " << t_time_command );
284 throw midge::node_nonfatal_error() <<
"Egg writer received unexpected start command on the trig stream while running";
287 if( t_trig_command == stream::s_run )
289 t_time_command = in_stream< 0 >().
get();
290 LTRACE(
plog,
"Egg writer reading stream 0 (time) at index " << in_stream< 0 >().get_current_index() );
291 LTRACE(
plog,
"Advancing time stream; time command matched trig command? trig command = " << t_trig_command <<
"; time command = " << t_time_command );
292 if( t_time_command != stream::s_run )
299 if (t_time_command == stream::s_stop)
301 t_trig_command = in_stream< 1 >().
get();
302 LDEBUG(
plog,
"Egg writer received stop command on the time stream while run in progress.");
303 LDEBUG(
plog,
"Advancing trig stream; time command matched trig command? trig command = " << t_trig_command <<
"; time command = " << t_time_command );
304 LDEBUG(
plog,
"Breaking out of is-running exe loop" );
310 throw midge::node_nonfatal_error() <<
"Trig command doesn't match time command: time command = " << t_time_command <<
"; trig command = " << t_trig_command;
316 t_time_data = in_stream< 0 >().data();
317 t_trig_data = in_stream< 1 >().data();
321 LDEBUG(
plog,
"Handling first packet in run" );
338 uint64_t t_time_id = t_time_data->get_pkt_in_session();
339 uint64_t t_trig_id = t_trig_data->get_id();
340 LTRACE(
plog,
"Time id: <" << t_time_id <<
">; Trig id: <" << t_trig_id <<
">");
342 if( t_time_id != t_trig_id )
362 LERROR(
plog,
"Mismatch between time id <" << t_time_id <<
"> and trigger id <" << t_trig_id <<
">" );
363 throw midge::node_nonfatal_error() <<
"Unable to match time and trigger streams";
369 if( t_trig_data->get_flag() )
371 LTRACE(
plog,
"Triggered packet, id <" << t_trig_data->get_id() <<
">" );
375 LDEBUG(
plog,
"New event" );
379 throw midge::node_nonfatal_error() <<
"Unable to write record to file; record ID: " << t_time_id;
382 LTRACE(
plog,
"Packet written (" << t_time_id <<
")" );
386 LTRACE(
plog,
"Untriggered packet, id <" << t_trig_id <<
">" );
409 butterfly_house::get_instance()->unregister_writer(
this );
425 LDEBUG(
plog,
"Configuring triggered_writer with:\n" << a_config );
426 a_node->set_file_num( a_config.get_value(
"file-num", a_node->get_file_num() ) );
427 if( a_config.has(
"device" ) )
429 const scarab::param_node& t_dev_config = a_config[
"device"].as_node();
430 a_node->set_bit_depth( t_dev_config.get_value(
"bit-depth", a_node->get_bit_depth() ) );
431 a_node->set_data_type_size( t_dev_config.get_value(
"data-type-size", a_node->get_data_type_size() ) );
432 a_node->set_sample_size( t_dev_config.get_value(
"sample-size", a_node->get_sample_size() ) );
433 a_node->set_record_size( t_dev_config.get_value(
"record-size", a_node->get_record_size() ) );
434 a_node->set_acq_rate( t_dev_config.get_value(
"acq-rate", a_node->get_acq_rate() ) );
435 a_node->set_v_offset( t_dev_config.get_value(
"v-offset", a_node->get_v_offset() ) );
436 a_node->set_v_range( t_dev_config.get_value(
"v-range", a_node->get_v_range() ) );
438 a_node->set_center_freq( a_config.get_value(
"center-freq", a_node->get_center_freq() ) );
439 a_node->set_freq_range( a_config.get_value(
"freq-range", a_node->get_freq_range() ) );
445 LDEBUG(
plog,
"Dumping configuration for triggered_writer" );
446 a_config.add(
"file-num", a_node->get_file_num() );
447 scarab::param_node t_dev_node;
448 t_dev_node.add(
"bit-depth", a_node->get_bit_depth() );
449 t_dev_node.add(
"data-type-size", a_node->get_data_type_size() );
450 t_dev_node.add(
"sample-size", a_node->get_sample_size() );
451 t_dev_node.add(
"record-size", a_node->get_record_size() );
452 t_dev_node.add(
"acq-rate", a_node->get_acq_rate() );
453 t_dev_node.add(
"v-offset", a_node->get_v_offset() );
454 t_dev_node.add(
"v-range", a_node->get_v_range() );
455 a_config.add(
"device", t_dev_node );
456 a_config.add(
"center-freq", a_node->get_center_freq() );
457 a_config.add(
"freq-range", a_node->get_freq_range() );
std::shared_ptr< header_wrapper > header_wrap_ptr
Base class for all writers.
bool f_start_file_with_next_data
virtual void prepare_to_write(monarch_wrap_ptr a_mw_ptr, header_wrap_ptr a_hw_ptr)
static scarab::logger plog("batch_executor")
void exe_loop_is_running(exe_loop_context &a_ctx)
monarch_wrap_ptr f_monarch_ptr
virtual void do_dump_config(const triggered_writer *a_node, scarab::param_node &a_config) const
const int8_t * get_raw_array() const
void exe_loop_not_running(exe_loop_context &a_ctx)
virtual void initialize()
virtual ~triggered_writer()
virtual ~triggered_writer_binding()
triggered_writer_binding()
uint64_t f_first_pkt_in_run
std::shared_ptr< monarch_wrapper > monarch_wrap_ptr
REGISTER_NODE_AND_BUILDER(data_producer, "data-producer", data_producer_binding)
A consumer to that writes triggered time ROACH packets to an egg file.
LOGGER(plog, "egg_writer")
virtual void execute(midge::diptera *a_midge=nullptr)
stream_wrap_ptr f_swrap_ptr
monarch3::M3Record * f_record_ptr
virtual void do_apply_config(triggered_writer *a_node, const scarab::param_node &a_config) const