dunedaq::fdreadoutlibs::WIBEthFrameProcessor Class Reference

#include <WIBEthFrameProcessor.hpp>

Inheritance diagram for dunedaq::fdreadoutlibs::WIBEthFrameProcessor:

Collaboration diagram for dunedaq::fdreadoutlibs::WIBEthFrameProcessor:

Public Types
using	inherited = datahandlinglibs::TaskRawDataProcessorModel<types::DUNEWIBEthTypeAdapter>
using	frameptr = types::DUNEWIBEthTypeAdapter*
using	constframeptr = const types::DUNEWIBEthTypeAdapter*
using	wibframeptr = dunedaq::fddetdataformats::WIBEthFrame*
Public Types inherited from dunedaq::opmonlib::MonitorableObject
using	NodePtr = std::weak_ptr<MonitorableObject>
using	NewNodePtr = std::shared_ptr<MonitorableObject>
using	ElementId = std::string

Public Member Functions
	WIBEthFrameProcessor (std::unique_ptr< datahandlinglibs::FrameErrorRegistry > &error_registry, bool processing_enabled)
void	start (const nlohmann::json &args) override
	Start operation.
void	stop (const nlohmann::json &args) override
	Stop operation.
void	conf (const appmodel::DataHandlerModule *conf) override
	Set the emulator mode, if active, timestamps of processed packets are overwritten with new ones.
Public Member Functions inherited from dunedaq::datahandlinglibs::TaskRawDataProcessorModel< types::DUNEWIBEthTypeAdapter >
	TaskRawDataProcessorModel (std::unique_ptr< FrameErrorRegistry > &error_registry, bool post_processing_enabled)
	~TaskRawDataProcessorModel ()
void	scrap (const nlohmann::json &) override
	Unconfigure.
void	reset_last_daq_time ()
std::uint64_t	get_last_daq_time () override
	Get newest timestamp of last seen packet.
void	preprocess_item (types::DUNEWIBEthTypeAdapter *item) override
	Preprocess one element.
void	postprocess_item (const types::DUNEWIBEthTypeAdapter *item) override
	Postprocess one element.
void	add_preprocess_task (Task &&task)
void	add_postprocess_task (Task &&task)
void	invoke_all_preprocess_functions (types::DUNEWIBEthTypeAdapter *item)
void	launch_all_preprocess_functions (types::DUNEWIBEthTypeAdapter *item)
Public Member Functions inherited from dunedaq::datahandlinglibs::RawDataProcessorConcept< types::DUNEWIBEthTypeAdapter >
	RawDataProcessorConcept ()
	RawDataProcessorConcept (const RawDataProcessorConcept &)=delete
	RawDataProcessorConcept is not copy-constructible.
	RawDataProcessorConcept (RawDataProcessorConcept &&)=delete
	RawDataProcessorConcept is not move-constructible.
virtual	~RawDataProcessorConcept ()
RawDataProcessorConcept &	operator= (const RawDataProcessorConcept &)=delete
	RawDataProcessorConcept is not copy-assginable.
RawDataProcessorConcept &	operator= (RawDataProcessorConcept &&)=delete
	RawDataProcessorConcept is not move-assignable.
Public Member Functions inherited from dunedaq::opmonlib::MonitorableObject
	MonitorableObject (const MonitorableObject &)=delete
MonitorableObject &	operator= (const MonitorableObject &)=delete
	MonitorableObject (MonitorableObject &&)=delete
MonitorableObject &	operator= (MonitorableObject &&)=delete
virtual	~MonitorableObject ()=default
auto	get_opmon_id () const noexcept
auto	get_opmon_level () const noexcept

Protected Member Functions
virtual void	generate_opmon_data () override
void	publish_processor_metric_to_opmon ()
void	publish_processor_metric_to_opmon_with_aggregation ()
std::map< int16_t, std::map< std::string, std::tuple< float, int16_t, int16_t, float, dunedaq::trgdataformats::channel_t, dunedaq::trgdataformats::channel_t > > >	calculate_all_metric_summaries_across_planes (const std::unordered_map< dunedaq::trgdataformats::channel_t, std::vector< std::pair< std::string, int16_t > > > &metrics)
void	use_pattern_generator (frameptr fp)
void	sequence_check (frameptr fp)
void	timestamp_check (frameptr fp)
void	find_hits (constframeptr fp)
Protected Member Functions inherited from dunedaq::datahandlinglibs::TaskRawDataProcessorModel< types::DUNEWIBEthTypeAdapter >
void	run_post_processing_thread (std::function< void(const types::DUNEWIBEthTypeAdapter *)> &function, folly::ProducerConsumerQueue< const types::DUNEWIBEthTypeAdapter * > &queue)
Protected Member Functions inherited from dunedaq::opmonlib::MonitorableObject
	MonitorableObject ()=default
void	register_node (ElementId name, NewNodePtr)
void	publish (google::protobuf::Message &&, CustomOrigin &&co={}, OpMonLevel l=to_level(EntryOpMonLevel::kDefault)) const noexcept

Protected Attributes
dunedaq::daqdataformats::timestamp_t	m_previous_ts = 0
dunedaq::daqdataformats::timestamp_t	m_current_ts = 0
uint16_t	m_previous_seq_id = 0
uint16_t	m_current_seq_id = 0
dunedaq::daqdataformats::timestamp_t	m_pattern_generator_previous_ts = 0
dunedaq::daqdataformats::timestamp_t	m_pattern_generator_current_ts = 0
bool	m_first_ts_missmatch = true
bool	m_ts_problem_reported = false
bool	m_ts_error_state = false
std::atomic< uint64_t >	m_ts_error_ctr { 0 }
bool	m_first_seq_id_mismatch = true
bool	m_seq_id_problem_reported = false
bool	m_seq_id_error_state = false
std::atomic< uint64_t >	m_seq_id_error_ctr { 0 }
std::atomic< int16_t >	m_seq_id_min_jump { 0 }
std::atomic< int16_t >	m_seq_id_max_jump { 0 }
Protected Attributes inherited from dunedaq::datahandlinglibs::TaskRawDataProcessorModel< types::DUNEWIBEthTypeAdapter >
std::vector< std::function< void(types::DUNEWIBEthTypeAdapter *)> >	m_preprocess_functions
std::unique_ptr< FrameErrorRegistry > &	m_error_registry
bool	m_post_processing_enabled
std::atomic< bool >	m_run_marker
std::vector< std::function< void(const types::DUNEWIBEthTypeAdapter *)> >	m_post_process_functions
std::vector< std::unique_ptr< folly::ProducerConsumerQueue< const types::DUNEWIBEthTypeAdapter * > > >	m_items_to_postprocess_queues
std::vector< std::unique_ptr< utilities::ReusableThread > >	m_post_process_threads
size_t	m_postprocess_queue_sizes
daqdataformats::SourceID	m_sourceid
std::atomic< uint64_t >	m_last_processed_daq_ts

Private Attributes
bool	m_first_hit = true
bool	m_tpg_metric_collect_enabled {false}
uint32_t	m_metric_collect_opmon_period { 128 }
std::unique_ptr< tpglibs::TPGenerator >	m_tp_generator
std::vector< std::pair< std::string, nlohmann::json > >	m_tpg_configs
uint32_t	m_tp_max_width
std::set< unsigned int >	m_channel_mask_set
uint16_t	m_tpg_threshold_selected
std::map< uint, std::atomic< int > >	m_tp_channel_rate_map
size_t	m_num_msg = 0
size_t	m_num_push_fail = 0
std::atomic< int >	m_tpg_hits_count { 0 }
uint32_t	m_det_id
uint32_t	m_crate_id
uint32_t	m_slot_id
uint32_t	m_stream_id
bool	m_emulator_mode = false
std::shared_ptr< detchannelmaps::TPCChannelMap >	m_channel_map
std::vector< std::pair< trgdataformats::channel_t, int16_t > >	m_channel_plane_numbers
std::vector< trigger::TriggerPrimitiveTypeAdapter >	m_tpa_vectors [3]
std::shared_ptr< iomanager::SenderConcept< std::vector< trigger::TriggerPrimitiveTypeAdapter > > >	m_tp_sink [3]
std::shared_ptr< iomanager::SenderConcept< fddetdataformats::WIBEthFrame > >	m_err_frame_sink
daqdataformats::SourceID	m_sourceid
std::atomic< uint64_t >	m_new_hits { 0 }
std::atomic< uint64_t >	m_new_tps { 0 }
std::atomic< uint64_t >	m_tps_suppressed_too_long { 0 }
std::atomic< uint64_t >	m_tps_send_failed { 0 }
std::atomic< uint64_t >	m_frame_counter { 0 }
std::chrono::time_point< std::chrono::high_resolution_clock >	m_t0

Additional Inherited Members
Static Public Member Functions inherited from dunedaq::opmonlib::MonitorableObject
static bool	publishable_metric (OpMonLevel entry, OpMonLevel system) noexcept

Detailed Description

Definition at line 52 of file WIBEthFrameProcessor.hpp.

Member Typedef Documentation

constframeptr

using dunedaq::fdreadoutlibs::WIBEthFrameProcessor::constframeptr = const types::DUNEWIBEthTypeAdapter*

Definition at line 58 of file WIBEthFrameProcessor.hpp.

frameptr

using dunedaq::fdreadoutlibs::WIBEthFrameProcessor::frameptr = types::DUNEWIBEthTypeAdapter*

Definition at line 57 of file WIBEthFrameProcessor.hpp.

inherited

using dunedaq::fdreadoutlibs::WIBEthFrameProcessor::inherited = datahandlinglibs::TaskRawDataProcessorModel<types::DUNEWIBEthTypeAdapter>

Definition at line 56 of file WIBEthFrameProcessor.hpp.

wibframeptr

using dunedaq::fdreadoutlibs::WIBEthFrameProcessor::wibframeptr = dunedaq::fddetdataformats::WIBEthFrame*

Definition at line 59 of file WIBEthFrameProcessor.hpp.

Constructor & Destructor Documentation

WIBEthFrameProcessor()

dunedaq::fdreadoutlibs::WIBEthFrameProcessor::WIBEthFrameProcessor ( std::unique_ptr< datahandlinglibs::FrameErrorRegistry > & error_registry, bool processing_enabled )

explicit

Definition at line 32 of file WIBEthFrameProcessor.cpp.

  : TaskRawDataProcessorModel<types::DUNEWIBEthTypeAdapter>(error_registry, processing_enabled)
{
}

Member Function Documentation

calculate_all_metric_summaries_across_planes()

std::map< int16_t, std::map< std::string, std::tuple< float, int16_t, int16_t, float, dunedaq::trgdataformats::channel_t, dunedaq::trgdataformats::channel_t > > > dunedaq::fdreadoutlibs::WIBEthFrameProcessor::calculate_all_metric_summaries_across_planes ( const std::unordered_map< dunedaq::trgdataformats::channel_t, std::vector< std::pair< std::string, int16_t > > > & metrics )

protected

Optimized version that calculates all metric summaries across all planes in a single pass Returns a map of plane_number -> map of metric_name -> summary statistics

Definition at line 249 of file WIBEthFrameProcessor.cpp.

                                                                                                                                                                          {
    // Structure to hold all statistics: plane -> metric -> (mean, min, max, stddev, min_channel_id, max_channel_id)
    std::map<int16_t, std::map<std::string, std::tuple<float, int16_t, int16_t, float, dunedaq::trgdataformats::channel_t, dunedaq::trgdataformats::channel_t>>> all_stats;
    
    // Structure to accumulate statistics: plane -> metric -> (count, mean, M2, min, max, min_channel_id, max_channel_id)
    std::map<int16_t, std::map<std::string, std::tuple<size_t, double, double, int16_t, int16_t, dunedaq::trgdataformats::channel_t, dunedaq::trgdataformats::channel_t>>> accumulators;
    
    // Single pass through all metrics to collect data using Welford's online algorithm for variance
    for (const auto& [channel, vec] : metrics) {
        if (!m_channel_map) continue;
        
        int16_t plane = m_channel_map->get_plane_from_offline_channel(channel);
        
        for (const auto& [name, val] : vec) {
            auto& [count, mean, M2, min, max, min_channel_id, max_channel_id] = accumulators[plane][name];
            
            count++;
            
            if (count == 1 || val < min) {
                min = val;
                min_channel_id = channel;
            }
            if (count == 1 || val > max) {
                max = val;
                max_channel_id = channel;
            }
            
            // Welford's online algorithm for variance calculation
            if (count == 1) {
                // First value: initialize mean and M2
                mean = val;
                M2 = 0.0;
            } else {
                // Update mean and M2 using Welford's algorithm
                double delta = val - mean;
                mean += delta / count;
                double delta2 = val - mean;
                M2 += delta * delta2;
            }
        }
    }
    
    // Calculate final statistics from accumulated data
    for (const auto& [plane, metric_map] : accumulators) {
        for (const auto& [metric_name, acc_data] : metric_map) {
            const auto& [count, mean, M2, min, max, min_channel_id, max_channel_id] = acc_data;
            
            if (count == 0) continue;
            
            float stddev = 0.0f;
            
            // Calculate standard deviation using accumulated M2
            if (count > 1) {
                stddev = std::sqrt(M2 / (count - 1));
            }
            
            all_stats[plane][metric_name] = std::make_tuple(static_cast<float>(mean), min, max, stddev, min_channel_id, max_channel_id);
        }
    }
    
    return all_stats;
}

conf()

void dunedaq::fdreadoutlibs::WIBEthFrameProcessor::conf ( const appmodel::DataHandlerModule * conf )

overridevirtual

Set the emulator mode, if active, timestamps of processed packets are overwritten with new ones.

Reimplemented from dunedaq::datahandlinglibs::TaskRawDataProcessorModel< types::DUNEWIBEthTypeAdapter >.

Definition at line 83 of file WIBEthFrameProcessor.cpp.

{
  size_t idx = 0;
  for (auto output : conf->get_outputs()) {
    try {
      if (output->get_data_type() == "TriggerPrimitiveVector") {
         m_tp_sink[idx++] = get_iom_sender<std::vector<trigger::TriggerPrimitiveTypeAdapter>>(output->UID());
      }
    } catch (const ers::Issue& excpt) {
      ers::error(datahandlinglibs::ResourceQueueError(ERS_HERE, "tp", "DefaultRequestHandlerModel", excpt));
    }
  }
 
  m_sourceid.id = conf->get_source_id();
  m_sourceid.subsystem = types::DUNEWIBEthTypeAdapter::subsystem;
  auto geo_id = conf->get_geo_id();
  if (geo_id != nullptr) {
    m_det_id = geo_id->get_detector_id();
    m_crate_id = geo_id->get_crate_id();
    m_slot_id = geo_id->get_slot_id();
    m_stream_id = geo_id->get_stream_id();
  }
  m_emulator_mode = conf->get_emulation_mode();
 
  // Setup pre-processing pipeline
  if (!m_emulator_mode)
    inherited::add_preprocess_task(std::bind(&WIBEthFrameProcessor::sequence_check, this, std::placeholders::_1));
 
  inherited::add_preprocess_task(std::bind(&WIBEthFrameProcessor::timestamp_check, this, std::placeholders::_1));
 
  // Check it post-processing is active
  auto dp = conf->get_module_configuration()->get_data_processor();
  if (dp != nullptr) {
    auto proc_conf = dp->cast<appmodel::TPCRawDataProcessor>();
    if (proc_conf != nullptr && m_post_processing_enabled) {
      m_tp_generator = std::make_unique<tpglibs::TPGenerator>();
 
      // Set the minimum TP samples over threshold.
      auto conf_sot_minima = proc_conf->get_sot_minima();
      std::vector<uint16_t> sot_minima{conf_sot_minima->get_sot_minimum_plane0(),
                                       conf_sot_minima->get_sot_minimum_plane1(),
                                       conf_sot_minima->get_sot_minimum_plane2()};
      m_tp_generator->set_sot_minima(sot_minima);
 
      const std::vector<unsigned int> channel_mask_vec = proc_conf->get_channel_mask();
 
      std::vector<const appmodel::ProcessingStep*> processing_steps = proc_conf->get_processing_steps();
      for (auto step : processing_steps) {
        m_tpg_configs.push_back(std::make_pair(step->class_name(), step->to_json(false).back()));
      }
 
      // Setup post-processing pipeline
      m_channel_map = dunedaq::detchannelmaps::make_tpc_map(proc_conf->get_channel_map());
      for (int chan = 0; chan < 64; chan++) {
        trgdataformats::channel_t off_channel = m_channel_map->get_offline_channel_from_det_crate_slot_stream_chan(m_det_id, m_crate_id, m_slot_id, m_stream_id, chan);
        int16_t plane = m_channel_map->get_plane_from_offline_channel(off_channel);
        m_channel_plane_numbers.push_back(std::make_pair(off_channel, plane));
 
        // This processor only needs to handle some (maybe 0) of the masked channels.
        // Only get those relevant channels for the later check.
        if (std::find(channel_mask_vec.begin(), channel_mask_vec.end(), off_channel) != channel_mask_vec.end())
          m_channel_mask_set.insert(off_channel);
      }
 
      m_metric_collect_opmon_period = proc_conf->get_metric_collect_opmon_rate();
 
      // Let the TPG generator configure
 
      m_tp_generator->configure(m_tpg_configs, m_channel_plane_numbers, types::DUNEWIBEthTypeAdapter::samples_tick_difference);
      
      // After it sees the configs, it will set the metric collector enable state
      
      m_tpg_metric_collect_enabled = m_tp_generator->get_metric_collector_enable_state();
 
      inherited::add_postprocess_task(std::bind(&WIBEthFrameProcessor::find_hits, this, std::placeholders::_1));
    }
  }
  inherited::conf(conf);
}

find_hits()

void dunedaq::fdreadoutlibs::WIBEthFrameProcessor::find_hits ( constframeptr fp )

protected

Pipeline Stage 2.: Do software TPG

Definition at line 466 of file WIBEthFrameProcessor.cpp.

{
  size_t nhits = 0;
  if (!fp)
    return;
  auto wfptr = reinterpret_cast<dunedaq::fddetdataformats::WIBEthFrame*>((uint8_t*)fp); // NOLINT
 
  // Check that the system is properly configured from the first hit.
  if (m_first_hit) {
    if (wfptr->daq_header.crate_id != m_crate_id || wfptr->daq_header.slot_id != m_slot_id || wfptr->daq_header.stream_id != m_stream_id) {
      ers::error(LinkMisconfiguration(ERS_HERE, wfptr->daq_header.crate_id, wfptr->daq_header.slot_id, wfptr->daq_header.stream_id, m_crate_id, m_slot_id, m_stream_id));
    }
 
    m_first_hit = false;
  }
 
  std::vector<trgdataformats::TriggerPrimitive> tps = (*m_tp_generator)(wfptr);
  m_frame_counter.fetch_add(1, std::memory_order_relaxed);
  if (m_tpg_metric_collect_enabled && m_frame_counter.load(std::memory_order_relaxed) % m_metric_collect_opmon_period == 0) {
    m_tp_generator->signal_metric_collection();
  }
 
  for (const auto& tp : tps) {
    // If this TP is on a masked channel, skip it.
    if (std::binary_search(m_channel_mask_set.begin(), m_channel_mask_set.end(), tp.channel))
      continue;
    // Need to move into a type adapter.
    trigger::TriggerPrimitiveTypeAdapter tpa;
    tpa.tp = tp;
 
    tpa.tp.detid = m_det_id;  // Last missing piece.
    m_tpa_vectors[m_channel_map->get_plane_from_offline_channel(tp.channel)].push_back(tpa);
    m_tp_channel_rate_map[tp.channel]++;
  }
 
  if (m_frame_counter.load(std::memory_order_relaxed) % 100 == 0) { // FIXME: Hard-coding 100 for now. This should be defined elsewhere or configurable.
    for (int i = 0; i < 3; i++) {
      int new_tps = m_tpa_vectors[i].size();
      if (new_tps == 0) {
        continue;
      }
      const auto s_ts_begin = m_tpa_vectors[i].front().tp.time_start;
      const auto channel_begin = m_tpa_vectors[i].front().tp.channel;
      const auto s_ts_end = m_tpa_vectors[i].back().tp.time_start;
      const auto channel_end = m_tpa_vectors[i].back().tp.channel;      
      if (!m_tp_sink[i]->try_send(std::move(m_tpa_vectors[i]), iomanager::Sender::s_no_block)) {
        ers::warning(FailedToSendTPVector(ERS_HERE, s_ts_begin, channel_begin, s_ts_end, channel_end));
        m_tps_send_failed++;
      } else {
        m_new_tps += new_tps;
        nhits += new_tps;
      }
    }
  }
 
  m_tpg_hits_count += nhits;
  return;
}

generate_opmon_data()

void dunedaq::fdreadoutlibs::WIBEthFrameProcessor::generate_opmon_data ( )

overrideprotectedvirtual

Hook for customisable pubblication. The function can throw, exception will be caught by the monitoring thread

Reimplemented from dunedaq::datahandlinglibs::TaskRawDataProcessorModel< types::DUNEWIBEthTypeAdapter >.

Definition at line 164 of file WIBEthFrameProcessor.cpp.

{
   datahandlinglibs::opmon::FixedRateDataProcessorInfo info;
 
   info.set_num_seq_id_errors(m_seq_id_error_ctr.load());
   info.set_min_seq_id_jump(m_seq_id_min_jump.exchange(0));
   info.set_max_seq_id_jump(m_seq_id_max_jump.exchange(0));
 
   info.set_num_ts_errors(m_ts_error_ctr.load());
   
   publish(std::move(info));
 
   m_error_registry->log_registered_errors();
 
   if (m_post_processing_enabled) {
     auto now = std::chrono::high_resolution_clock::now();
     int new_hits = m_tpg_hits_count.exchange(0);
     int new_tps = m_new_tps.exchange(0);
     int new_tps_suppressed_too_long = m_tps_suppressed_too_long.exchange(0);
     int new_tps_send_failed = m_tps_send_failed.exchange(0);
     double seconds = std::chrono::duration_cast<std::chrono::microseconds>(now - m_t0).count() / 1000000.;
     TLOG_DEBUG(TLVL_BOOKKEEPING) << "Hit rate: " << std::to_string(new_hits / seconds / 1000.) << " [kHz]";
     TLOG_DEBUG(TLVL_BOOKKEEPING) << "Total new hits: " << new_hits << " new TPs: " << new_tps;
 
     datahandlinglibs::opmon::HitFindingInfo tp_info;
     tp_info.set_rate_tp_hits(new_hits / seconds / 1000.);
 
     tp_info.set_num_tps_sent(new_tps);
     tp_info.set_num_tps_suppressed_too_long(new_tps_suppressed_too_long);
     tp_info.set_num_tps_send_failed(new_tps_send_failed);
 
     publish(std::move(tp_info));
     // Find the channels with the top  TP rates
     // Create a vector of pairs to store the map elements
     std::vector<std::pair<uint, int>> channel_tp_rate_vec(m_tp_channel_rate_map.begin(), m_tp_channel_rate_map.end());
     // Sort the vector in descending order of the value of the pairs
     sort(channel_tp_rate_vec.begin(), channel_tp_rate_vec.end(), [](std::pair<uint, int>& a, std::pair<uint, int>& b) { return a.second > b.second; });
     // Add the metrics to opmon
     // For convenience we are selecting only the top 10 elements
     if (channel_tp_rate_vec.size() != 0) {
       int top_highest_values = 10;
       if (channel_tp_rate_vec.size() < 10) {
         top_highest_values = channel_tp_rate_vec.size();
       }
       //datahandlinglibs::opmon::TPChannelsInfo channels_info;
       for (int i = 0; i < top_highest_values; i++) {
         datahandlinglibs::opmon::TPChannelInfo tpc_info;
         tpc_info.set_number_of_tps(channel_tp_rate_vec[i].second);
          tpc_info.set_channel_id(channel_tp_rate_vec[i].first);
         publish(std::move(tpc_info), {{"channel", std::to_string(channel_tp_rate_vec[i].first)}});
       }
     }
 
     // Reset the counter in the channel rate map
     for (auto& el : m_tp_channel_rate_map) {
       el.second = 0;
     }
     m_t0 = now;
 
     if (m_tpg_metric_collect_enabled && m_tp_generator) {
       publish_processor_metric_to_opmon();
       publish_processor_metric_to_opmon_with_aggregation();
     }
   }
   
   inherited::generate_opmon_data();
 }

publish_processor_metric_to_opmon()

void dunedaq::fdreadoutlibs::WIBEthFrameProcessor::publish_processor_metric_to_opmon ( )

protected

Publishes collected processor metrics to opmon, currently called in generate_opmon_data()

Definition at line 233 of file WIBEthFrameProcessor.cpp.

                                                        {
  auto metrics = m_tp_generator->get_processor_metrics();
  for (const auto& [channel, vec] : metrics) {
    datahandlinglibs::opmon::TPGProcessorInfo tpg_proc_info;
    for (const auto& [name, val] : vec) {
      if (name == "m_pedestal") {
        tpg_proc_info.set_pedestal(val);
      } else if (name == "m_accum") {
        tpg_proc_info.set_accum(val);
      }
    }
    publish(std::move(tpg_proc_info), {{"channel", std::to_string(channel)}});
  }
}

publish_processor_metric_to_opmon_with_aggregation()

void dunedaq::fdreadoutlibs::WIBEthFrameProcessor::publish_processor_metric_to_opmon_with_aggregation ( )

protected

Publishes collected processor metrics to opmon, with aggregation of metrics to summary statistics across physical planes

Definition at line 313 of file WIBEthFrameProcessor.cpp.

                                                                         {
  auto metrics = m_tp_generator->get_processor_metrics();
  
  // Use optimized single-pass calculation for all metrics across all planes
  auto all_stats = calculate_all_metric_summaries_across_planes(metrics);
  
  // Publish all calculated statistics
  for (const auto& [plane, metric_map] : all_stats) {
    for (const auto& [metric_name, stats] : metric_map) {
      const auto& [mean, min, max, stddev, min_channel_id, max_channel_id] = stats;
      
      datahandlinglibs::opmon::TPGProcessorReducedInfo info;
      info.set_average(mean);
      info.set_max(max);
      info.set_min(min);
      info.set_standard_dev(stddev);
      info.set_max_channel_id(max_channel_id);
      info.set_min_channel_id(min_channel_id);
      publish(std::move(info), {{"plane", std::to_string(plane)}, {"metric", metric_name}});
    }
  }
}

sequence_check()

void dunedaq::fdreadoutlibs::WIBEthFrameProcessor::sequence_check ( frameptr fp )

protected

Pipeline Stage 1.: Check proper sequence id increments in DAQ Eth header

Pipeline Stage 1.: Check proper timestamp increments in WIB frame

Definition at line 341 of file WIBEthFrameProcessor.cpp.

{
  // FIXME: Make source emulator deal with this! Hard to do since source emu is templated...
  /* If EMU data, emulate perfectly incrementing timestamp
  if (m_emulator_mode) {  
    // uint64_t ts_next = m_previous_seq_id + 1; // NOLINT(build/unsigned)
    auto wf = reinterpret_cast<wibframeptr>(((uint8_t*)fp));            // NOLINT
    for (unsigned int i = 0; i < fp->get_num_frames(); ++i) {           // NOLINT(build/unsigned)
      //auto wfh = const_cast<dunedaq::fddetdataformats::WIBEthFrame*>(wf->header());
      wf->daq_header.crate_id = m_crate_id;
      wf->daq_header.slot_id = m_slot_id;
      wf->daq_header.stream_id = m_stream_id; 
      wf->daq_header.seq_id = (m_previous_seq_id+i) & 0xfff;
      wf++;
    }
  }
  */
          
  // Acquire timestamp
  auto wfptr = reinterpret_cast<dunedaq::fddetdataformats::WIBEthFrame*>(fp); // NOLINT
  m_current_seq_id = wfptr->daq_header.seq_id;
 
  // Check sequence id
  // Calculate the next sequence id (12 bits)
  uint16_t expected_seq_id = (m_previous_seq_id + fp->get_num_frames()) & 0xfff;
  int16_t delta_seq_id = m_current_seq_id-expected_seq_id;
  if ( delta_seq_id > 0x800) {
    delta_seq_id -= 0x1000;
  } else if ( delta_seq_id < -0x7ff) {
    delta_seq_id += 0x1000;
  }
 
  if (delta_seq_id == 0) {
    m_seq_id_error_state = false;
  } else {
    // uint16_t delta_seq_id = (m_current_seq_id-expected_seq_id);
    ++m_seq_id_error_ctr;
    m_seq_id_max_jump = std::max(delta_seq_id, m_seq_id_max_jump.load());
    m_seq_id_min_jump = std::min(delta_seq_id, m_seq_id_min_jump.load());
 
    if (m_first_seq_id_mismatch) { // log once
      TLOG_DEBUG(TLVL_BOOKKEEPING) << "First sequence id MISMATCH! -> | previous: " << std::to_string(m_previous_seq_id) << " current: " + std::to_string(m_current_seq_id);
      m_first_seq_id_mismatch = false;
    } else {
      if (!m_seq_id_error_state) {
        m_error_registry->add_error("Sequence ID jump", datahandlinglibs::FrameErrorRegistry::ErrorInterval(expected_seq_id, m_current_seq_id));
        m_seq_id_error_state = true;
      }
    }    
  }
 
  if (m_seq_id_error_ctr > 1000) {
    if (!m_seq_id_problem_reported) {
      TLOG() << "*** Data Integrity ERROR *** Sequence ID continuity is completely broken! "
             << "Something is wrong with the FE source or with the configuration!";
      m_seq_id_problem_reported = true;
    }
  }
 
  m_previous_seq_id = m_current_seq_id;
 
}

start()

void dunedaq::fdreadoutlibs::WIBEthFrameProcessor::start ( const nlohmann::json & )

overridevirtual

Start operation.

Reimplemented from dunedaq::datahandlinglibs::TaskRawDataProcessorModel< types::DUNEWIBEthTypeAdapter >.

Definition at line 38 of file WIBEthFrameProcessor.cpp.

{
  // Reset software TPG resources
  if (m_post_processing_enabled) {
    m_tps_suppressed_too_long = 0;
    m_tps_send_failed = 0;
  } 
 
  // Reset timestamp check
  m_previous_ts = 0;
  m_current_ts = 0;
  m_first_ts_missmatch = true;
  m_ts_problem_reported = false;
  m_ts_error_state = false;
  m_ts_error_ctr = 0;
 
  m_first_seq_id_mismatch = true;
  m_seq_id_problem_reported = false;
  m_seq_id_error_state = false;
  m_seq_id_error_ctr = 0;
 
 
  // Reset stats
  m_t0 = std::chrono::high_resolution_clock::now();
  m_new_hits = 0;
  m_new_tps = 0;
  m_tpg_hits_count.exchange(0);
  inherited::start(args);
}

stop()

void dunedaq::fdreadoutlibs::WIBEthFrameProcessor::stop ( const nlohmann::json & )

overridevirtual

Stop operation.

Reimplemented from dunedaq::datahandlinglibs::TaskRawDataProcessorModel< types::DUNEWIBEthTypeAdapter >.

Definition at line 69 of file WIBEthFrameProcessor.cpp.

{
  inherited::stop(args);
  if (m_post_processing_enabled) {
    if (m_tpg_metric_collect_enabled) {
      m_tp_generator->free_metric_collector();
    }
    // Clears the pipelines and resets with the given configs.
    m_tp_generator->set_metric_collector_enable_state(m_tpg_metric_collect_enabled);
    m_tp_generator->configure(m_tpg_configs, m_channel_plane_numbers, types::DUNEWIBEthTypeAdapter::samples_tick_difference);
  }
}

timestamp_check()

void dunedaq::fdreadoutlibs::WIBEthFrameProcessor::timestamp_check ( frameptr fp )

protected

Pipeline Stage 1.: Check proper timestamp increments in DAQ Eth header

Pipeline Stage 1.: Check proper timestamp increments in WIB frame

Definition at line 408 of file WIBEthFrameProcessor.cpp.

{
 
  uint16_t wibeth_tick_difference = types::DUNEWIBEthTypeAdapter::expected_tick_difference;
  uint16_t wibeth_frame_tick_difference = wibeth_tick_difference * fp->get_num_frames();
 
  // FIXME: let source emulator deal with this!
  /* If EMU data, emulate perfectly incrementing timestamp
  if (m_emulator_mode) {                                     // emulate perfectly incrementing timestamp
    uint64_t ts_next = m_previous_ts + wibeth_frame_tick_difference; // NOLINT(build/unsigned)
    auto wf = reinterpret_cast<wibframeptr>(((uint8_t*)fp));            // NOLINT
    for (unsigned int i = 0; i < fp->get_num_frames(); ++i) {           // NOLINT(build/unsigned)
      //auto wfh = const_cast<dunedaq::fddetdataformats::WIBEthFrame*>(wf->header());
      wf->daq_header.crate_id = m_crate_id;
      wf->daq_header.slot_id = m_slot_id;
      wf->daq_header.stream_id = m_stream_id; 
      wf->set_timestamp(ts_next);
      ts_next += wibeth_tick_difference;
      wf++;
    }
  }*/
 
  auto wfptr = reinterpret_cast<dunedaq::fddetdataformats::WIBEthFrame*>(fp); // NOLINT
  m_current_ts = wfptr->get_timestamp();
 
  // Check timestamp
  if (m_previous_ts > 0 &&
      m_current_ts - m_previous_ts != wibeth_frame_tick_difference) [[unlikely]] {
    ++m_ts_error_ctr;
    if (m_first_ts_missmatch) { // log once
      TLOG_DEBUG(TLVL_BOOKKEEPING) << "First timestamp MISMATCH! -> | previous: " << std::to_string(m_previous_ts) << " current: " + std::to_string(m_current_ts);
      m_first_ts_missmatch = false;
    } else {
      if (!m_ts_error_state) {
        m_error_registry->add_error("Timestamp jump", datahandlinglibs::FrameErrorRegistry::ErrorInterval(m_previous_ts + wibeth_frame_tick_difference, m_current_ts));
        m_ts_error_state = true;
      }
    }
  } else {
    m_ts_error_state = false;
  }
 
  if (m_ts_error_ctr > 1000) {
    if (!m_ts_problem_reported) {
      TLOG() << "*** Data Integrity ERROR *** Timestamp continuity is completely broken! "
             << "Something is wrong with the FE source or with the configuration!";
      m_ts_problem_reported = true;
    }
  }
 
  m_previous_ts = m_current_ts;
  m_last_processed_daq_ts = m_current_ts;
}

use_pattern_generator()

void dunedaq::fdreadoutlibs::WIBEthFrameProcessor::use_pattern_generator ( frameptr fp )

protected

Pipeline Stage 0: Pattern generator for hit finding in emulated mode

Member Data Documentation

m_channel_map

std::shared_ptr<detchannelmaps::TPCChannelMap> dunedaq::fdreadoutlibs::WIBEthFrameProcessor::m_channel_map

private

Definition at line 161 of file WIBEthFrameProcessor.hpp.

m_channel_mask_set

std::set<unsigned int> dunedaq::fdreadoutlibs::WIBEthFrameProcessor::m_channel_mask_set

private

Definition at line 145 of file WIBEthFrameProcessor.hpp.

m_channel_plane_numbers

std::vector<std::pair<trgdataformats::channel_t, int16_t> > dunedaq::fdreadoutlibs::WIBEthFrameProcessor::m_channel_plane_numbers

private

Definition at line 165 of file WIBEthFrameProcessor.hpp.

m_crate_id

uint32_t dunedaq::fdreadoutlibs::WIBEthFrameProcessor::m_crate_id

private

Definition at line 156 of file WIBEthFrameProcessor.hpp.

m_current_seq_id

uint16_t dunedaq::fdreadoutlibs::WIBEthFrameProcessor::m_current_seq_id = 0

protected

Definition at line 96 of file WIBEthFrameProcessor.hpp.

m_current_ts

dunedaq::daqdataformats::timestamp_t dunedaq::fdreadoutlibs::WIBEthFrameProcessor::m_current_ts = 0

protected

Definition at line 93 of file WIBEthFrameProcessor.hpp.

m_det_id

uint32_t dunedaq::fdreadoutlibs::WIBEthFrameProcessor::m_det_id

private

Definition at line 155 of file WIBEthFrameProcessor.hpp.

m_emulator_mode

bool dunedaq::fdreadoutlibs::WIBEthFrameProcessor::m_emulator_mode = false

private

Definition at line 159 of file WIBEthFrameProcessor.hpp.

m_err_frame_sink

std::shared_ptr<iomanager::SenderConcept<fddetdataformats::WIBEthFrame> > dunedaq::fdreadoutlibs::WIBEthFrameProcessor::m_err_frame_sink

private

Definition at line 169 of file WIBEthFrameProcessor.hpp.

m_first_hit

bool dunedaq::fdreadoutlibs::WIBEthFrameProcessor::m_first_hit = true

private

Definition at line 139 of file WIBEthFrameProcessor.hpp.

m_first_seq_id_mismatch

bool dunedaq::fdreadoutlibs::WIBEthFrameProcessor::m_first_seq_id_mismatch = true

protected

Definition at line 106 of file WIBEthFrameProcessor.hpp.

m_first_ts_missmatch

bool dunedaq::fdreadoutlibs::WIBEthFrameProcessor::m_first_ts_missmatch = true

protected

Definition at line 101 of file WIBEthFrameProcessor.hpp.

m_frame_counter

std::atomic<uint64_t> dunedaq::fdreadoutlibs::WIBEthFrameProcessor::m_frame_counter { 0 }

private

Definition at line 179 of file WIBEthFrameProcessor.hpp.

{ 0 };

m_metric_collect_opmon_period

uint32_t dunedaq::fdreadoutlibs::WIBEthFrameProcessor::m_metric_collect_opmon_period { 128 }

private

Definition at line 141 of file WIBEthFrameProcessor.hpp.

{ 128 };

m_new_hits

std::atomic<uint64_t> dunedaq::fdreadoutlibs::WIBEthFrameProcessor::m_new_hits { 0 }

private

Definition at line 175 of file WIBEthFrameProcessor.hpp.

{ 0 }; // NOLINT(build/unsigned)

m_new_tps

std::atomic<uint64_t> dunedaq::fdreadoutlibs::WIBEthFrameProcessor::m_new_tps { 0 }

private

Definition at line 176 of file WIBEthFrameProcessor.hpp.

{ 0 };  // NOLINT(build/unsigned)

m_num_msg

size_t dunedaq::fdreadoutlibs::WIBEthFrameProcessor::m_num_msg = 0

private

Definition at line 150 of file WIBEthFrameProcessor.hpp.

m_num_push_fail

size_t dunedaq::fdreadoutlibs::WIBEthFrameProcessor::m_num_push_fail = 0

private

Definition at line 151 of file WIBEthFrameProcessor.hpp.

m_pattern_generator_current_ts

dunedaq::daqdataformats::timestamp_t dunedaq::fdreadoutlibs::WIBEthFrameProcessor::m_pattern_generator_current_ts = 0

protected

Definition at line 99 of file WIBEthFrameProcessor.hpp.

m_pattern_generator_previous_ts

dunedaq::daqdataformats::timestamp_t dunedaq::fdreadoutlibs::WIBEthFrameProcessor::m_pattern_generator_previous_ts = 0

protected

Definition at line 98 of file WIBEthFrameProcessor.hpp.

m_previous_seq_id

uint16_t dunedaq::fdreadoutlibs::WIBEthFrameProcessor::m_previous_seq_id = 0

protected

Definition at line 95 of file WIBEthFrameProcessor.hpp.

m_previous_ts

dunedaq::daqdataformats::timestamp_t dunedaq::fdreadoutlibs::WIBEthFrameProcessor::m_previous_ts = 0

protected

Definition at line 92 of file WIBEthFrameProcessor.hpp.

m_seq_id_error_ctr

std::atomic<uint64_t> dunedaq::fdreadoutlibs::WIBEthFrameProcessor::m_seq_id_error_ctr { 0 }

protected

Definition at line 109 of file WIBEthFrameProcessor.hpp.

{ 0 };

m_seq_id_error_state

bool dunedaq::fdreadoutlibs::WIBEthFrameProcessor::m_seq_id_error_state = false

protected

Definition at line 108 of file WIBEthFrameProcessor.hpp.

m_seq_id_max_jump

std::atomic<int16_t> dunedaq::fdreadoutlibs::WIBEthFrameProcessor::m_seq_id_max_jump { 0 }

protected

Definition at line 111 of file WIBEthFrameProcessor.hpp.

{ 0 };

m_seq_id_min_jump

std::atomic<int16_t> dunedaq::fdreadoutlibs::WIBEthFrameProcessor::m_seq_id_min_jump { 0 }

protected

Definition at line 110 of file WIBEthFrameProcessor.hpp.

{ 0 };

m_seq_id_problem_reported

bool dunedaq::fdreadoutlibs::WIBEthFrameProcessor::m_seq_id_problem_reported = false

protected

Definition at line 107 of file WIBEthFrameProcessor.hpp.

m_slot_id

uint32_t dunedaq::fdreadoutlibs::WIBEthFrameProcessor::m_slot_id

private

Definition at line 157 of file WIBEthFrameProcessor.hpp.

m_sourceid

daqdataformats::SourceID dunedaq::fdreadoutlibs::WIBEthFrameProcessor::m_sourceid

private

Definition at line 173 of file WIBEthFrameProcessor.hpp.

m_stream_id

uint32_t dunedaq::fdreadoutlibs::WIBEthFrameProcessor::m_stream_id

private

Definition at line 158 of file WIBEthFrameProcessor.hpp.

m_t0

std::chrono::time_point<std::chrono::high_resolution_clock> dunedaq::fdreadoutlibs::WIBEthFrameProcessor::m_t0

private

Definition at line 181 of file WIBEthFrameProcessor.hpp.

m_tp_channel_rate_map

std::map<uint, std::atomic<int> > dunedaq::fdreadoutlibs::WIBEthFrameProcessor::m_tp_channel_rate_map

private

Definition at line 148 of file WIBEthFrameProcessor.hpp.

m_tp_generator

std::unique_ptr<tpglibs::TPGenerator> dunedaq::fdreadoutlibs::WIBEthFrameProcessor::m_tp_generator

private

Definition at line 142 of file WIBEthFrameProcessor.hpp.

m_tp_max_width

uint32_t dunedaq::fdreadoutlibs::WIBEthFrameProcessor::m_tp_max_width

private

Definition at line 144 of file WIBEthFrameProcessor.hpp.

m_tp_sink

std::shared_ptr<iomanager::SenderConcept<std::vector<trigger::TriggerPrimitiveTypeAdapter> > > dunedaq::fdreadoutlibs::WIBEthFrameProcessor::m_tp_sink[3]

private

Definition at line 168 of file WIBEthFrameProcessor.hpp.

m_tpa_vectors

std::vector<trigger::TriggerPrimitiveTypeAdapter> dunedaq::fdreadoutlibs::WIBEthFrameProcessor::m_tpa_vectors[3]

private

Definition at line 166 of file WIBEthFrameProcessor.hpp.

m_tpg_configs

std::vector<std::pair<std::string, nlohmann::json> > dunedaq::fdreadoutlibs::WIBEthFrameProcessor::m_tpg_configs

private

Definition at line 143 of file WIBEthFrameProcessor.hpp.

m_tpg_hits_count

std::atomic<int> dunedaq::fdreadoutlibs::WIBEthFrameProcessor::m_tpg_hits_count { 0 }

private

Definition at line 153 of file WIBEthFrameProcessor.hpp.

{ 0 };

m_tpg_metric_collect_enabled

bool dunedaq::fdreadoutlibs::WIBEthFrameProcessor::m_tpg_metric_collect_enabled {false}

private

Definition at line 140 of file WIBEthFrameProcessor.hpp.

{false};

m_tpg_threshold_selected

uint16_t dunedaq::fdreadoutlibs::WIBEthFrameProcessor::m_tpg_threshold_selected

private

Definition at line 146 of file WIBEthFrameProcessor.hpp.

m_tps_send_failed

std::atomic<uint64_t> dunedaq::fdreadoutlibs::WIBEthFrameProcessor::m_tps_send_failed { 0 }

private

Definition at line 178 of file WIBEthFrameProcessor.hpp.

{ 0 };

m_tps_suppressed_too_long

std::atomic<uint64_t> dunedaq::fdreadoutlibs::WIBEthFrameProcessor::m_tps_suppressed_too_long { 0 }

private

Definition at line 177 of file WIBEthFrameProcessor.hpp.

{ 0 };

m_ts_error_ctr

std::atomic<uint64_t> dunedaq::fdreadoutlibs::WIBEthFrameProcessor::m_ts_error_ctr { 0 }

protected

Definition at line 104 of file WIBEthFrameProcessor.hpp.

{ 0 };

m_ts_error_state

bool dunedaq::fdreadoutlibs::WIBEthFrameProcessor::m_ts_error_state = false

protected

Definition at line 103 of file WIBEthFrameProcessor.hpp.

m_ts_problem_reported

bool dunedaq::fdreadoutlibs::WIBEthFrameProcessor::m_ts_problem_reported = false

protected

Definition at line 102 of file WIBEthFrameProcessor.hpp.

---

The documentation for this class was generated from the following files:

• /github/workspace/dunedaq/sourcecode/fdreadoutlibs/include/fdreadoutlibs/wibeth/WIBEthFrameProcessor.hpp
• /github/workspace/dunedaq/sourcecode/fdreadoutlibs/src/wibeth/WIBEthFrameProcessor.cpp