TCProcessor.cpp

Go to the documentation of this file.

#include "trigger/TCProcessor.hpp" // NOLINT(build/include)
 
#include "iomanager/Sender.hpp"
#include "logging/Logging.hpp"
 
#include "datahandlinglibs/DataHandlingIssues.hpp"
#include "datahandlinglibs/FrameErrorRegistry.hpp"
#include "datahandlinglibs/ReadoutLogging.hpp"
#include "datahandlinglibs/models/IterableQueueModel.hpp"
#include "trigger/TCWrapper.hpp"
#include "triggeralgs/TriggerCandidate.hpp"
 
#include "appmodel/TCDataProcessor.hpp"
#include "appmodel/TriggerDataHandlerModule.hpp"
 
using dunedaq::datahandlinglibs::logging::TLVL_BOOKKEEPING;
using dunedaq::datahandlinglibs::logging::TLVL_TAKE_NOTE;
 
// THIS SHOULDN'T BE HERE!!!!! But it is necessary.....
DUNE_DAQ_TYPESTRING(dunedaq::trigger::TCWrapper, "TriggerCandidate")
 
namespace dunedaq {
namespace trigger {
 
TCProcessor::TCProcessor(std::unique_ptr<datahandlinglibs::FrameErrorRegistry>& error_registry,
                         bool post_processing_enabled)
  : datahandlinglibs::TaskRawDataProcessorModel<TCWrapper>(error_registry, post_processing_enabled)
{
}
 
TCProcessor::~TCProcessor() {}
 
void
TCProcessor::start(const appfwk::DAQModule::CommandData_t& args)
{
  TLOG_DEBUG(TLVL_ENTER_EXIT_METHODS) << "TCProcessor: Entering start() method";
 
  m_running_flag.store(true);
  m_send_trigger_decisions_thread = std::thread(&TCProcessor::send_trigger_decisions, this);
  pthread_setname_np(m_send_trigger_decisions_thread.native_handle(), "mlt-dec"); // TODO: originally mlt-trig-dec
 
  // Reset stats
  m_tds_created_count.store(0);
  m_tds_sent_count.store(0);
  m_tds_dropped_count.store(0);
  m_tds_failed_bitword_count.store(0);
  m_tds_cleared_count.store(0);
  // per TC
  m_tc_received_count.store(0);
  m_tds_created_tc_count.store(0);
  m_tds_sent_tc_count.store(0);
  m_tds_dropped_tc_count.store(0);
  m_tds_failed_bitword_tc_count.store(0);
  m_tds_cleared_tc_count.store(0);
  m_tc_ignored_count.store(0);
  inherited::start(args);
 
  TLOG_DEBUG(TLVL_ENTER_EXIT_METHODS) << "TCProcessor: Exiting start() method";
}
 
void
TCProcessor::stop(const appfwk::DAQModule::CommandData_t& args)
{
  TLOG_DEBUG(TLVL_ENTER_EXIT_METHODS) << "TCProcessor: Entering stop() method";
 
  inherited::stop(args);
  m_running_flag.store(false);
 
  // Make sure condition_variable knows we flipped running flag
  {
    std::lock_guard<std::mutex> lock(m_td_vector_mutex);
    m_cv.notify_all();
  }
 
  // Wait for the TD-sending thread to stop
  m_send_trigger_decisions_thread.join();
 
  // Drop all TDs in vectors at run stage change. Have to do this
  // after joining m_send_trigger_decisions_thread so we don't
  // concurrently access the vectors
  clear_td_vectors();
 
  print_opmon_stats();
 
  TLOG_DEBUG(TLVL_ENTER_EXIT_METHODS) << "TCProcessor: Exiting stop() method";
}
 
void
TCProcessor::conf(const appmodel::DataHandlerModule* cfg)
{
  TLOG_DEBUG(TLVL_ENTER_EXIT_METHODS) << "TCProcessor: Entering conf() method";
 
  auto mtrg = cfg->cast<appmodel::TriggerDataHandlerModule>();
  if (mtrg == nullptr) {
    throw(InvalidConfiguration(ERS_HERE, "Provided null TriggerDataHandlerModule configuration!"));
  }
  for (auto output : mtrg->get_outputs()) {
    try {
      if (output->get_data_type() == "TriggerDecision") {
        m_td_sink = get_iom_sender<dfmessages::TriggerDecision>(output->UID());
      }
    } catch (const ers::Issue& excpt) {
      ers::error(datahandlinglibs::ResourceQueueError(ERS_HERE, "td", "DefaultRequestHandlerModel", excpt));
    }
  }
 
  auto dp = mtrg->get_module_configuration()->get_data_processor();
  auto proc_conf = dp->cast<appmodel::TCDataProcessor>();
 
  // Add all Source IDs to mandatoy links for now...
  for (auto const& link : mtrg->get_mandatory_source_ids()) {
    m_mandatory_links.push_back(
      dfmessages::SourceID{ daqdataformats::SourceID::string_to_subsystem(link->get_subsystem()), link->get_sid() });
  }
  for (auto const& link : mtrg->get_enabled_source_ids()) {
    m_mandatory_links.push_back(
      dfmessages::SourceID{ daqdataformats::SourceID::string_to_subsystem(link->get_subsystem()), link->get_sid() });
  }
 
  // TODO: Group links!
  // m_group_links_data = conf->get_groups_links();
  parse_group_links(m_group_links_data);
  print_group_links();
  m_total_group_links = m_group_links.size();
  TLOG_DEBUG(3) << "Total group links: " << m_total_group_links;
 
  m_tc_merging = proc_conf->get_merge_overlapping_tcs();
  m_ignore_tc_pileup = proc_conf->get_ignore_overlapping_tcs();
  m_buffer_timeout = proc_conf->get_buffer_timeout();
  m_send_timed_out_tds = (m_ignore_tc_pileup) ? false : proc_conf->get_td_out_of_timeout();
  m_td_readout_limit = proc_conf->get_td_readout_limit();
  m_ignored_tc_types = proc_conf->get_ignore_tc();
  m_ignoring_tc_types = !m_ignored_tc_types.empty();
 
  // Trigger bitwords
  std::vector<const appmodel::TriggerBitword*> bitwords = proc_conf->get_trigger_bitwords();
  m_use_bitwords = !bitwords.empty();
  if (m_use_bitwords) {
    set_trigger_bitwords(bitwords);
    print_trigger_bitwords();
  }
  TLOG_DEBUG(3) << "Use bitwords: " << m_use_bitwords;
  TLOG_DEBUG(3) << "Allow merging: " << m_tc_merging;
  TLOG_DEBUG(3) << "Ignore pileup: " << m_ignore_tc_pileup;
  TLOG_DEBUG(3) << "Buffer timeout: " << m_buffer_timeout;
  TLOG_DEBUG(3) << "Should send timed out TDs: " << m_send_timed_out_tds;
  TLOG_DEBUG(3) << "TD readout limit: " << m_td_readout_limit;
 
  // ROI map
  m_roi_conf_data = proc_conf->get_roi_group_conf();
  m_use_roi_readout = !m_roi_conf_data.empty();
  if (m_use_roi_readout) {
    parse_roi_conf(m_roi_conf_data);
    print_roi_conf(m_roi_conf);
  }
  TLOG_DEBUG(3) << "Use ROI readout?: " << m_use_roi_readout;
 
  // Custom readout map
  m_readout_window_map_data = proc_conf->get_tc_readout_map();
  m_use_readout_map = !m_readout_window_map_data.empty();
  if (m_use_readout_map) {
    parse_readout_map(m_readout_window_map_data);
    print_readout_map(m_readout_window_map);
  }
  TLOG_DEBUG(3) << "Use readout map: " << m_use_readout_map;
 
  // Ignoring TC types
  TLOG_DEBUG(3) << "Ignoring TC types: " << m_ignoring_tc_types;
  if (m_ignoring_tc_types) {
    TLOG_DEBUG(3) << "TC types to ignore: ";
    for (std::vector<unsigned int>::iterator it = m_ignored_tc_types.begin(); it != m_ignored_tc_types.end();) {
      TLOG_DEBUG(3) << *it;
      ++it;
    }
  }
  m_latency_monitoring.store(dp->get_latency_monitoring());
  inherited::add_postprocess_task(std::bind(&TCProcessor::make_td, this, std::placeholders::_1));
 
  inherited::conf(mtrg);
 
  TLOG_DEBUG(TLVL_ENTER_EXIT_METHODS) << "TCProcessor: Exiting conf() method";
}
 
void
TCProcessor::scrap(const appfwk::DAQModule::CommandData_t& args)
{
  TLOG_DEBUG(TLVL_ENTER_EXIT_METHODS) << "TCProcessor: Entering scrap() method";
 
  m_mandatory_links.clear();
  m_group_links.clear();
  m_roi_conf.clear();
  m_roi_conf_data.clear();
  m_roi_conf_ids.clear();
  m_roi_conf_probs.clear();
  m_roi_conf_probs_c.clear();
  m_pending_tds.clear();
  m_readout_window_map_data.clear();
  m_readout_window_map.clear();
  m_ignored_tc_types.clear();
 
  m_td_sink.reset();
 
  m_group_links_data.clear();
 
  inherited::scrap(args);
 
  TLOG_DEBUG(TLVL_ENTER_EXIT_METHODS) << "TCProcessor: Exiting scrap() method";
}
 
void
TCProcessor::generate_opmon_data()
{
  opmon::TCProcessorInfo info;
 
  info.set_tds_created_count(m_tds_created_count.load());
  info.set_tds_sent_count(m_tds_sent_count.load());
  info.set_tds_dropped_count(m_tds_dropped_count.load());
  info.set_tds_failed_bitword_count(m_tds_failed_bitword_count.load());
  info.set_tds_cleared_count(m_tds_cleared_count.load());
  info.set_tc_received_count(m_tc_received_count.load());
  info.set_tc_ignored_count(m_tc_ignored_count.load());
  info.set_tds_created_tc_count(m_tds_created_tc_count.load());
  info.set_tds_sent_tc_count(m_tds_sent_tc_count.load());
  info.set_tds_dropped_tc_count(m_tds_dropped_tc_count.load());
  info.set_tds_failed_bitword_tc_count(m_tds_failed_bitword_tc_count.load());
  info.set_tds_cleared_tc_count(m_tds_cleared_tc_count.load());
 
  this->publish(std::move(info));
 
  if (m_latency_monitoring.load() && m_running_flag.load()) {
    opmon::TriggerLatency lat_info;
 
    lat_info.set_latency_in(m_latency_instance.get_latency_in());
    lat_info.set_latency_out(m_latency_instance.get_latency_out());
 
    this->publish(std::move(lat_info));
  }
}

void
TCProcessor::make_td(const TCWrapper* tcw)
{
 
  auto tc = tcw->candidate;
  if (m_latency_monitoring.load())
    m_latency_instance.update_latency_in(tc.time_start);
  m_tc_received_count++;
 
  if ((m_use_readout_map) && (m_readout_window_map.count(tc.type))) {
    TLOG_DEBUG(3) << "Got TC of type " << static_cast<int>(tc.type) << ", timestamp " << tc.time_candidate
                  << ", start/end " << tc.time_start << "/" << tc.time_end << ", readout start/end "
                  << tc.time_candidate - m_readout_window_map[tc.type].first << "/"
                  << tc.time_candidate + m_readout_window_map[tc.type].second;
  } else {
    TLOG_DEBUG(3) << "Got TC of type " << static_cast<int>(tc.type) << ", timestamp " << tc.time_candidate
                  << ", start/end " << tc.time_start << "/" << tc.time_end;
  }
 
  // Option to ignore TC types (if given by config)
  if (m_ignoring_tc_types == true && check_trigger_type_ignore(static_cast<unsigned int>(tc.type)) == true) {
    TLOG_DEBUG(3) << " Ignore TC type: " << static_cast<unsigned int>(tc.type);
    m_tc_ignored_count++;
 
    /*FIXME: comment out this block: if a TC is to be ignored it shall just be ignored!
      if (m_tc_merging) {
        // Still need to check for overlap with existing TD, if overlaps, include in the TD, but don't extend
        // readout
        std::lock_guard<std::mutex> lock(m_td_vector_mutex);
        add_tc_ignored(*tc);
      }
      */
  } else {
    std::lock_guard<std::mutex> lock(m_td_vector_mutex);
    add_tc(tc);
    m_cv.notify_one();
    TLOG_DEBUG(10) << "pending tds size: " << m_pending_tds.size();
  }
  m_last_processed_daq_ts = tc.time_start;
  return;
}
 
dfmessages::TriggerDecision
TCProcessor::create_decision(const PendingTD& pending_td)
{
  m_earliest_tc_index = get_earliest_tc_index(pending_td);
  TLOG_DEBUG(5) << "earliest TC index: " << m_earliest_tc_index;
 
  if (pending_td.contributing_tcs.size() > 1) {
    TLOG_DEBUG(5) << "!!! TD created from " << pending_td.contributing_tcs.size() << " TCs !!!";
  }
 
  dfmessages::TriggerDecision decision;
  decision.trigger_number = 0; // filled by MLT
  decision.run_number = 0;     // filled by MLT
  decision.trigger_timestamp = pending_td.contributing_tcs[m_earliest_tc_index].time_candidate;
  decision.readout_type = dfmessages::ReadoutType::kLocalized;
 
  TDBitset td_bitword = get_TD_bitword(pending_td);
  TLOG_DEBUG(5) << "[MLT] TD has bitword: " << td_bitword << " "
                << static_cast<dfmessages::trigger_type_t>(td_bitword.to_ulong());
  decision.trigger_type = static_cast<dfmessages::trigger_type_t>(td_bitword.to_ulong()); // m_trigger_type;
 
  // decision.trigger_type = 1; // m_trigger_type;
 
  TLOG_DEBUG(3) << ", TC detid: " << pending_td.contributing_tcs[m_earliest_tc_index].detid
                << ", TC type: " << static_cast<int>(pending_td.contributing_tcs[m_earliest_tc_index].type)
                << ", TC cont number: " << pending_td.contributing_tcs.size()
                << ", DECISION trigger type: " << decision.trigger_type
                << ", DECISION timestamp: " << decision.trigger_timestamp
                << ", request window begin: " << pending_td.readout_start
                << ", request window end: " << pending_td.readout_end;
 
  std::vector<dfmessages::ComponentRequest> requests =
    create_all_decision_requests(m_mandatory_links, pending_td.readout_start, pending_td.readout_end);
  add_requests_to_decision(decision, requests);
 
  if (!m_use_roi_readout) {
    for (const auto& [key, value] : m_group_links) {
      std::vector<dfmessages::ComponentRequest> group_requests =
        create_all_decision_requests(value, pending_td.readout_start, pending_td.readout_end);
      add_requests_to_decision(decision, group_requests);
    }
  } else { // using ROI readout
    roi_readout_make_requests(decision);
  }
 
  m_tds_created_count++;
  m_tds_created_tc_count += pending_td.contributing_tcs.size();
 
  return decision;
}
 
void
TCProcessor::send_trigger_decisions()
{
  // A unique lock that can be locked and unlocked
  std::unique_lock<std::mutex> lock(m_td_vector_mutex);
 
  while (m_running_flag) {
    // TODO: think about better implementation (notify?, something event driven)
    m_cv.wait_for(lock, std::chrono::microseconds(100));
    auto ready_tds = get_ready_tds(m_pending_tds);
    TLOG_DEBUG(10) << "ready tds: " << ready_tds.size() << ", updated pending tds: " << m_pending_tds.size();
 
    for (std::vector<PendingTD>::iterator it = ready_tds.begin(); it != ready_tds.end();) {
      call_tc_decision(*it);
      ++it;
    }
  }
}
 
void
TCProcessor::call_tc_decision(const TCProcessor::PendingTD& pending_td)
{
 
  if (m_use_bitwords) {
    // Check trigger bitwords
    TDBitset td_bitword = get_TD_bitword(pending_td);
    if (!check_trigger_bitwords(td_bitword)) {
      // Don't process further if the bitword check failed
      m_tds_failed_bitword_count++;
      m_tds_failed_bitword_tc_count += pending_td.contributing_tcs.size();
      return;
    }
  }
 
  dfmessages::TriggerDecision decision = create_decision(pending_td);
  auto tn = decision.trigger_number;
  auto td_ts = decision.trigger_timestamp;
 
  if (m_latency_monitoring.load())
    m_latency_instance.update_latency_out(pending_td.contributing_tcs.front().time_start);
  if (!m_td_sink->try_send(std::move(decision), iomanager::Sender::s_no_block)) {
    ers::warning(TDDropped(ERS_HERE, tn, td_ts));
    m_tds_dropped_count++;
    m_tds_dropped_tc_count += pending_td.contributing_tcs.size();
  } else {
    m_tds_sent_count++;
    m_tds_sent_tc_count += pending_td.contributing_tcs.size();
  }
}
 
void
TCProcessor::add_tc(const triggeralgs::TriggerCandidate tc)
{
  bool tc_dealt = false;
  int64_t tc_wallclock_arrived =
    std::chrono::duration_cast<std::chrono::milliseconds>(std::chrono::steady_clock::now().time_since_epoch()).count();
 
  if (m_tc_merging || m_ignore_tc_pileup) {
 
    for (std::vector<PendingTD>::iterator it = m_pending_tds.begin(); it != m_pending_tds.end();) {
      // Don't deal with TC here if there's no overlap
      if (!check_overlap(tc, *it)) {
        it++;
        continue;
      }
 
      // If overlap and ignoring, we drop the TC and flag it as dealt with.
      if (m_ignore_tc_pileup) {
        m_tds_dropped_tc_count++;
        tc_dealt = true;
        TLOG_DEBUG(3) << "TC overlapping with a previous TD, dropping!";
        break;
      }
 
      // If we're here, TC merging must be on, in which case we're actually
      // going to merge the TC into the TD.
      it->contributing_tcs.push_back(tc);
      if ((m_use_readout_map) && (m_readout_window_map.count(tc.type))) {
        TLOG_DEBUG(3) << "TC with start/end times " << tc.time_candidate - m_readout_window_map[tc.type].first << "/"
                      << tc.time_candidate + m_readout_window_map[tc.type].second
                      << " overlaps with pending TD with start/end times " << it->readout_start << "/"
                      << it->readout_end;
        it->readout_start = ((tc.time_candidate - m_readout_window_map[tc.type].first) >= it->readout_start)
                              ? it->readout_start
                              : (tc.time_candidate - m_readout_window_map[tc.type].first);
        it->readout_end = ((tc.time_candidate + m_readout_window_map[tc.type].second) >= it->readout_end)
                            ? (tc.time_candidate + m_readout_window_map[tc.type].second)
                            : it->readout_end;
      } else {
        TLOG_DEBUG(3) << "TC with start/end times " << tc.time_start << "/" << tc.time_end
                      << " overlaps with pending TD with start/end times " << it->readout_start << "/"
                      << it->readout_end;
        it->readout_start = (tc.time_start >= it->readout_start) ? it->readout_start : tc.time_start;
        it->readout_end = (tc.time_end >= it->readout_end) ? tc.time_end : it->readout_end;
      }
      it->walltime_expiration = tc_wallclock_arrived + m_buffer_timeout;
      tc_dealt = true;
      break;
    }
  }
 
  // Don't do anything else if we've already dealt with the TC
  if (tc_dealt) {
    return;
  }
 
  // Create a new TD out of the TC
  PendingTD td_candidate;
  td_candidate.contributing_tcs.push_back(tc);
  if ((m_use_readout_map) && (m_readout_window_map.count(tc.type))) {
    td_candidate.readout_start = tc.time_candidate - m_readout_window_map[tc.type].first;
    td_candidate.readout_end = tc.time_candidate + m_readout_window_map[tc.type].second;
  } else {
    td_candidate.readout_start = tc.time_start;
    td_candidate.readout_end = tc.time_end;
  }
  td_candidate.walltime_expiration = tc_wallclock_arrived + m_buffer_timeout;
  m_pending_tds.push_back(td_candidate);
}
 
void
TCProcessor::add_tc_ignored(const triggeralgs::TriggerCandidate tc)
{
  for (std::vector<PendingTD>::iterator it = m_pending_tds.begin(); it != m_pending_tds.end();) {
    if (check_overlap(tc, *it)) {
      if ((m_use_readout_map) && (m_readout_window_map.count(tc.type))) {
        TLOG_DEBUG(3) << "!Ignored! TC with start/end times " << tc.time_candidate - m_readout_window_map[tc.type].first
                      << "/" << tc.time_candidate + m_readout_window_map[tc.type].second
                      << " overlaps with pending TD with start/end times " << it->readout_start << "/"
                      << it->readout_end;
      } else {
        TLOG_DEBUG(3) << "!Ignored! TC with start/end times " << tc.time_start << "/" << tc.time_end
                      << " overlaps with pending TD with start/end times " << it->readout_start << "/"
                      << it->readout_end;
      }
      it->contributing_tcs.push_back(tc);
      break;
    }
    ++it;
  }
  return;
}
 
bool
TCProcessor::check_overlap(const triggeralgs::TriggerCandidate& tc, const PendingTD& pending_td)
{
  if ((m_use_readout_map) && (m_readout_window_map.count(tc.type))) {
    return !(((tc.time_candidate + m_readout_window_map[tc.type].second) < pending_td.readout_start) ||
             ((tc.time_candidate - m_readout_window_map[tc.type].first > pending_td.readout_end)));
  } else {
    return !((tc.time_end < pending_td.readout_start) || (tc.time_start > pending_td.readout_end));
  }
}
 
std::vector<TCProcessor::PendingTD>
TCProcessor::get_ready_tds(std::vector<PendingTD>& pending_tds)
{
  std::vector<PendingTD> return_tds;
  for (std::vector<PendingTD>::iterator it = pending_tds.begin(); it != pending_tds.end();) {
    auto timestamp_now =
      std::chrono::duration_cast<std::chrono::milliseconds>(std::chrono::steady_clock::now().time_since_epoch())
        .count();
    if (timestamp_now >= it->walltime_expiration) {
      return_tds.push_back(*it);
      it = pending_tds.erase(it);
    } else if (check_td_readout_length(*it)) { // Also pass on TDs with (too) long readout window
      return_tds.push_back(*it);
      it = pending_tds.erase(it);
    } else {
      ++it;
    }
  }
  return return_tds;
}
 
int
TCProcessor::get_earliest_tc_index(const PendingTD& pending_td)
{
  int earliest_tc_index = -1;
  triggeralgs::timestamp_t earliest_tc_time;
  for (int i = 0; i < static_cast<int>(pending_td.contributing_tcs.size()); i++) {
    if (earliest_tc_index == -1) {
      earliest_tc_time = pending_td.contributing_tcs[i].time_candidate;
      earliest_tc_index = i;
    } else {
      if (pending_td.contributing_tcs[i].time_candidate < earliest_tc_time) {
        earliest_tc_time = pending_td.contributing_tcs[i].time_candidate;
        earliest_tc_index = i;
      }
    }
  }
  return earliest_tc_index;
}
 
bool
TCProcessor::check_td_readout_length(const PendingTD& pending_td)
{
  bool td_too_long = false;
  if (static_cast<int64_t>(pending_td.readout_end - pending_td.readout_start) >= m_td_readout_limit) {
    td_too_long = true;
    TLOG_DEBUG(3) << "Too long readout window: " << (pending_td.readout_end - pending_td.readout_start)
                  << ", sending immediate TD!";
  }
  return td_too_long;
}
 
void
TCProcessor::clear_td_vectors()
{
  std::lock_guard<std::mutex> lock(m_td_vector_mutex);
  m_tds_cleared_count += m_pending_tds.size();
  // Use std::accumulate to sum up the sizes of all contributing_tcs vectors
  size_t tds_cleared_tc_count =
    std::accumulate(m_pending_tds.begin(), m_pending_tds.end(), 0, [](size_t sum, const PendingTD& ptd) {
      return sum + ptd.contributing_tcs.size();
    });
  m_tds_cleared_tc_count += tds_cleared_tc_count;
  m_pending_tds.clear();
}
bool
TCProcessor::check_trigger_type_ignore(unsigned int tc_type)
{
  bool ignore = false;
  for (std::vector<unsigned int>::iterator it = m_ignored_tc_types.begin(); it != m_ignored_tc_types.end();) {
    if (tc_type == *it) {
      ignore = true;
      break;
    }
    ++it;
  }
  return ignore;
}
 
void
TCProcessor::print_trigger_bitwords()
{
  TLOG_DEBUG(3) << "Configured trigger words:";
  for (const auto& bitword : m_trigger_bitwords) {
    TLOG_DEBUG(3) << bitword;
  }
}
 
bool
TCProcessor::check_trigger_bitwords(const TDBitset& td_bitword) const
{
  bool trigger_check = false;
  for (const auto& bitword : m_trigger_bitwords) {
    TLOG_DEBUG(15) << "TD word: " << td_bitword << ", bitword: " << bitword;
    trigger_check = ((td_bitword & bitword) == bitword);
    TLOG_DEBUG(15) << "&: " << (td_bitword & bitword);
    TLOG_DEBUG(15) << "trigger?: " << trigger_check;
    if (trigger_check == true)
      break;
  }
  return trigger_check;
}
 
void
TCProcessor::set_trigger_bitwords(const std::vector<const appmodel::TriggerBitword*>& _bitwords)
{
  for (const appmodel::TriggerBitword* bitword : _bitwords) {
    TDBitset temp_bitword;
 
    for (const std::string& tctype_str : bitword->get_bitword()) {
      TCType tc_type = static_cast<TCType>(dunedaq::trgdataformats::string_to_trigger_candidate_type(tctype_str));
 
      if (tc_type == TCType::kUnknown) {
        throw(InvalidConfiguration(ERS_HERE, "Provided an unknown/non-existent TC type as a trigger bitword!"));
      }
 
      temp_bitword.set(static_cast<uint64_t>(tc_type));
    }
 
    m_trigger_bitwords.push_back(temp_bitword);
  }
}
 
void
TCProcessor::parse_readout_map(const std::vector<const appmodel::TCReadoutMap*>& data)
{
  for (auto readout_type : data) {
    TCType tc_type =
      static_cast<TCType>(dunedaq::trgdataformats::string_to_trigger_candidate_type(readout_type->get_tc_type_name()));
 
    // Throw error if unknown TC type
    if (tc_type == TCType::kUnknown) {
      throw(InvalidConfiguration(ERS_HERE, "Provided an unknown TC type in the TCReadoutMap for the TCProcessor"));
    }
 
    m_readout_window_map[tc_type] = { readout_type->get_time_before(), readout_type->get_time_after() };
  }
  return;
}
void
TCProcessor::print_readout_map(std::map<TCType, std::pair<triggeralgs::timestamp_t, triggeralgs::timestamp_t>> map)
{
  TLOG_DEBUG(3) << "MLT TD Readout map:";
  for (auto const& [key, val] : map) {
    TLOG_DEBUG(3) << "Type: " << static_cast<int>(key) << ", before: " << val.first << ", after: " << val.second;
  }
  return;
}
 
void
TCProcessor::parse_group_links(const nlohmann::json& data)
{
  for (auto group : data) {
    const nlohmann::json& temp_links_data = group["links"];
    std::vector<dfmessages::SourceID> temp_links;
    for (auto link : temp_links_data) {
      temp_links.push_back(
        dfmessages::SourceID{ daqdataformats::SourceID::string_to_subsystem(link["subsystem"]), link["element"] });
    }
    m_group_links.insert({ group["group"], temp_links });
  }
  return;
}
 
void
TCProcessor::print_group_links()
{
  TLOG_DEBUG(3) << "MLT Group Links:";
  for (auto const& [key, val] : m_group_links) {
    TLOG_DEBUG(3) << "Group: " << key;
    for (auto const& link : val) {
      TLOG_DEBUG(3) << link;
    }
  }
  TLOG_DEBUG(3) << " ";
  return;
}
dfmessages::ComponentRequest
TCProcessor::create_request_for_link(dfmessages::SourceID link,
                                     triggeralgs::timestamp_t start,
                                     triggeralgs::timestamp_t end)
{
  dfmessages::ComponentRequest request;
  request.component = link;
  request.window_begin = start;
  request.window_end = end;
 
  TLOG_DEBUG(10) << "setting request start: " << request.window_begin;
  TLOG_DEBUG(10) << "setting request end: " << request.window_end;
 
  return request;
}
 
std::vector<dfmessages::ComponentRequest>
TCProcessor::create_all_decision_requests(std::vector<dfmessages::SourceID> links,
                                          triggeralgs::timestamp_t start,
                                          triggeralgs::timestamp_t end)
{
  std::vector<dfmessages::ComponentRequest> requests;
  for (auto link : links) {
    requests.push_back(create_request_for_link(link, start, end));
  }
  return requests;
}
 
void
TCProcessor::add_requests_to_decision(dfmessages::TriggerDecision& decision,
                                      std::vector<dfmessages::ComponentRequest> requests)
{
  for (auto request : requests) {
    decision.components.push_back(request);
  }
}
 
void
TCProcessor::parse_roi_conf(const std::vector<const appmodel::ROIGroupConf*>& data)
{
  int counter = 0;
  float run_sum = 0;
  for (auto group : data) {
    roi_group temp_roi_group;
    temp_roi_group.n_links = group->get_number_of_link_groups();
    temp_roi_group.prob = group->get_probability();
    temp_roi_group.time_window = group->get_time_window();
    temp_roi_group.mode = group->get_groups_selection_mode();
    m_roi_conf.insert({ counter, temp_roi_group });
    m_roi_conf_ids.push_back(counter);
    m_roi_conf_probs.push_back(group->get_probability());
    run_sum += static_cast<float>(group->get_probability());
    m_roi_conf_probs_c.push_back(run_sum);
    counter++;
  }
  return;
}
 
void
TCProcessor::print_roi_conf(std::map<int, roi_group> roi_conf)
{
  TLOG_DEBUG(3) << "ROI CONF";
  for (const auto& [key, value] : roi_conf) {
    TLOG_DEBUG(3) << "ID: " << key;
    TLOG_DEBUG(3) << "n links: " << value.n_links;
    TLOG_DEBUG(3) << "prob: " << value.prob;
    TLOG_DEBUG(3) << "time: " << value.time_window;
    TLOG_DEBUG(3) << "mode: " << value.mode;
  }
  TLOG_DEBUG(3) << " ";
  return;
}
 
float
TCProcessor::get_random_num_float(float limit)
{
  float rnd = (double)rand() / RAND_MAX;
  return rnd * (limit);
}
 
int
TCProcessor::pick_roi_group_conf()
{
  float rnd_num = get_random_num_float(m_roi_conf_probs_c.back());
  for (int i = 0; i < static_cast<int>(m_roi_conf_probs_c.size()); i++) {
    if (rnd_num < m_roi_conf_probs_c[i]) {
      return i;
    }
  }
  return -1;
}
 
int
TCProcessor::get_random_num_int()
{
  int range = m_total_group_links;
  int rnd = rand() % range;
  return rnd;
}
void
TCProcessor::roi_readout_make_requests(dfmessages::TriggerDecision& decision)
{
  // Get configuration at random (weighted)
  int group_pick = pick_roi_group_conf();
  if (group_pick != -1) {
    roi_group this_group = m_roi_conf[m_roi_conf_ids[group_pick]];
    std::vector<dfmessages::SourceID> links;
 
    // If mode is random, pick groups to request at random
    if (this_group.mode == "kRandom") {
      TLOG_DEBUG(10) << "RAND";
      std::set<int> groups;
      while (static_cast<int>(groups.size()) < this_group.n_links) {
        groups.insert(get_random_num_int());
      }
      for (auto r_id : groups) {
        links.insert(links.end(), m_group_links[r_id].begin(), m_group_links[r_id].end());
      }
      // Otherwise, read sequntially by IDs, starting at 0
    } else {
      TLOG_DEBUG(10) << "SEQ";
      int r_id = 0;
      while (r_id < this_group.n_links) {
        links.insert(links.end(), m_group_links[r_id].begin(), m_group_links[r_id].end());
        r_id++;
      }
    }
 
    TLOG_DEBUG(10) << "TD timestamp: " << decision.trigger_timestamp;
    TLOG_DEBUG(10) << "group window: " << this_group.time_window;
 
    // Once the components are prepared, create requests and append them to decision
    std::vector<dfmessages::ComponentRequest> requests = create_all_decision_requests(
      links, decision.trigger_timestamp - this_group.time_window, decision.trigger_timestamp + this_group.time_window);
    add_requests_to_decision(decision, requests);
    links.clear();
  }
  return;
}
 
TCProcessor::TDBitset
TCProcessor::get_TD_bitword(const PendingTD& ready_td) const
{
  // get only unique types
  std::vector<int> tc_types;
  for (auto tc : ready_td.contributing_tcs) {
    tc_types.push_back(static_cast<int>(tc.type));
  }
  tc_types.erase(std::unique(tc_types.begin(), tc_types.end()), tc_types.end());
 
  // form TD bitword
  TDBitset td_bitword;
  for (auto tc_type : tc_types) {
    td_bitword.set(tc_type);
  }
  return td_bitword;
}
 
void
TCProcessor::print_opmon_stats()
{
  TLOG() << "TCProcessor opmon counters summary:";
  TLOG() << "------------------------------";
  TLOG() << "TDs created: \t\t\t" << m_tds_created_count << " \t(" << m_tds_created_tc_count << " TCs)";
  TLOG() << "TDs sent: \t\t\t" << m_tds_sent_count << " \t(" << m_tds_sent_tc_count << " TCs)";
  TLOG() << "TDs dropped: \t\t\t" << m_tds_dropped_count << " \t(" << m_tds_dropped_tc_count << " TCs)";
  TLOG() << "TDs failed bitword check: \t" << m_tds_failed_bitword_count << " \t(" << m_tds_failed_bitword_tc_count
         << " TCs)";
  TLOG() << "TDs cleared: \t\t\t" << m_tds_cleared_count << " \t(" << m_tds_cleared_tc_count << " TCs)";
  TLOG() << "------------------------------";
  TLOG() << "TCs received: \t" << m_tc_received_count;
  TLOG() << "TCs ignored: \t" << m_tc_ignored_count;
  TLOG();
}
 
} // namespace fdreadoutlibs
} // namespace dunedaq