eventbuilder::EventBuilder Class Reference

Public Member Functions
	EventBuilder (const std::string &name, framework::Process &proc)
	enableLogging ("EventBuilder") virtual ~EventBuilder()=default
void	configure (framework::config::Parameters &ps) override
	Callback for the EventProcessor to configure itself from the given set of parameters.
void	produce (framework::Event &event) override
	Process the event and put new data products into it.
Public Member Functions inherited from framework::Producer
	Producer (const std::string &name, Process &process)
	Class constructor.
virtual void	process (Event &event) final
	Processing an event for a Producer is calling produce.
Public Member Functions inherited from framework::EventProcessor
	DECLARE_FACTORY (EventProcessor, EventProcessor *, const std::string &, Process &)
	declare that we have a factory for this class
	EventProcessor (const std::string &name, Process &process)
	Class constructor.
virtual	~EventProcessor ()=default
	Class destructor.
virtual void	beforeNewRun (ldmx::RunHeader &run_header)
	Callback for Producers to add parameters to the run header before conditions are initialized.
virtual void	onNewRun (const ldmx::RunHeader &run_header)
	Callback for the EventProcessor to take any necessary action when the run being processed changes.
virtual void	onFileOpen (EventFile &event_file)
	Callback for the EventProcessor to take any necessary action when a new event input ROOT file is opened.
virtual void	onFileClose (EventFile &event_file)
	Callback for the EventProcessor to take any necessary action when a event input ROOT file is closed.
virtual void	onProcessStart ()
	Callback for the EventProcessor to take any necessary action when the processing of events starts, such as creating histograms.
virtual void	onProcessEnd ()
	Callback for the EventProcessor to take any necessary action when the processing of events finishes, such as calculating job-summary quantities.
template<class T >
const T &	getCondition (const std::string &condition_name)
	Access a conditions object for the current event.
TDirectory *	getHistoDirectory ()
	Access/create a directory in the histogram file for this event processor to create histograms and analysis tuples.
void	setStorageHint (framework::StorageControl::Hint hint)
	Mark the current event as having the given storage control hint from this module_.
void	setStorageHint (framework::StorageControl::Hint hint, const std::string &purposeString)
	Mark the current event as having the given storage control hint from this module and the given purpose string.
int	getLogFrequency () const
	Get the current logging frequency from the process.
int	getRunNumber () const
	Get the run number from the process.
std::string	getName () const
	Get the processor name.
void	createHistograms (const std::vector< framework::config::Parameters > &histos)
	Internal function which is used to create histograms passed from the python configuration @parma histos vector of Parameters that configure histograms to create.

Private Member Functions
PhysicsEventData	assemble_payload (const std::vector< DataFragment > &fragments)
void	write_event_binary (const PhysicsEventData &ev, const std::string &path="events.bin")

Private Attributes
bool	m_verbose_parse
unsigned int	m_event_id
FragmentBuffer	m_event_buffer
std::string	m_input_file
packing::utility::Reader	m_reader
std::string	m_output_name {"BuilderOutput"}
long long	m_coherence_window_ns {5000000}
int	m_min_subsystems {2}
std::chrono::steady_clock::time_point	m_start_time
std::chrono::steady_clock::time_point	m_event_start_time
uint64_t	m_total_bytes_read {0}
uint64_t	m_total_events_built {0}
unsigned long	m_events_since_last_report {0}
std::chrono::steady_clock::time_point	m_window_start_time
uint64_t	m_window_events_count {0}
uint64_t	m_window_bytes_read {0}

Static Private Attributes
static constexpr unsigned int	WINDOW_SIZE = 100

Additional Inherited Members
Protected Member Functions inherited from framework::EventProcessor
void	abortEvent ()
	Abort the event immediately.
Protected Attributes inherited from framework::EventProcessor
HistogramPool	histograms_
	helper object for making and filling histograms
NtupleManager &	ntuple_ {NtupleManager::getInstance()}
	Manager for any ntuples.
logging::logger	the_log_
	The logger for this EventProcessor.

Detailed Description

Definition at line 13 of file EventBuilder.h.

Constructor & Destructor Documentation

EventBuilder()

eventbuilder::EventBuilder::EventBuilder ( const std::string & name, framework::Process & proc )

inline

Definition at line 16 of file EventBuilder.h.

      : framework::Producer(name, proc), m_verbose_parse(false), m_event_id(0) {}

Member Function Documentation

assemble_payload()

PhysicsEventData EventBuilder::assemble_payload ( const std::vector< DataFragment > & fragments )

private

Definition at line 439 of file EventBuilder.cxx.

                                                                                        {
    PhysicsEventData event_data;
    if (fragments.empty()) return event_data;
    event_data.event_id = m_event_id;
    event_data.timestamp = fragments.front().header.timestamp;
    for (const auto &fragment : fragments) {
        GenericDataBlock g;
        g.subsystem_id = fragment.header.subsystem_id;
        g.timestamp_ns = fragment.header.timestamp;
        g.data = fragment.payload;
        if (fragment.trailer.checksum != 0) g.checksum = fragment.trailer.checksum;
        else g.checksum = crc32(fragment.payload);
        event_data.blocks.push_back(std::move(g));
        event_data.systems_readout.push_back(fragment.header.subsystem_id);
    }
    return event_data;
}

configure()

void EventBuilder::configure ( framework::config::Parameters & parameters )

overridevirtual

Callback for the EventProcessor to configure itself from the given set of parameters.

The parameters a processor has access to are the member variables of the python class in the sequence that has class_name equal to the EventProcessor class name.

For an example, look at MyProcessor.

Parameters

parameters

Parameters for configuration.

Reimplemented from framework::EventProcessor.

Definition at line 61 of file EventBuilder.cxx.

                                                          {
  if (ps.exists("verbose_parse")) {
    m_verbose_parse = ps.get<bool>("verbose_parse");
  }
  if (ps.exists("dat_file")) {
    m_input_file = ps.get<std::string>("dat_file");
  } else {
    const char* env_input = std::getenv("EVENTBUILDER_INPUT");
    if (env_input) m_input_file = env_input;
  }
  if (ps.exists("output_name")) m_output_name = ps.get<std::string>("output_name");
  if (ps.exists("coherence_window_ns")) {
    m_coherence_window_ns = ps.get<double>("coherence_window_ns");
  }
  if (ps.exists("min_subsystems")) {
    m_min_subsystems = ps.get<int>("min_subsystems");
  }
 
  ldmx_log(info) << "configure(): dat_file='" << m_input_file << "' output_name='" << m_output_name
                 << "' verbose_parse=" << (m_verbose_parse?"true":"false")
                 << " coherence_window_ns=" << m_coherence_window_ns
                 << " min_subsystems=" << m_min_subsystems;
  
  if (!m_input_file.empty()) {
    ldmx_log(info) << "configure(): opening file '" << m_input_file << "'";
    m_reader.open(m_input_file);
    if (!m_reader) {
      ldmx_log(error) << "failed to open input file: '" << m_input_file << "'";
    } else {
      ldmx_log(info) << "configure(): file opened successfully";
    }
  } else {
    ldmx_log(error) << "no input file specified";
  }
 
  // Initialize performance tracking
  m_start_time = std::chrono::steady_clock::now();
  m_total_bytes_read = 0;
  m_total_events_built = 0;
  m_events_since_last_report = 0;
  
  // Initialize windowed metrics for real-time DAQ monitoring
  m_window_start_time = m_start_time;
  m_window_events_count = 0;
  m_window_bytes_read = 0;
}

References framework::config::Parameters::exists(), framework::config::Parameters::get(), and packing::utility::Reader::open().

produce()

void EventBuilder::produce ( framework::Event & event )

overridevirtual

Process the event and put new data products into it.

Parameters

event

The Event to process.

Implements framework::Producer.

Definition at line 108 of file EventBuilder.cxx.

                                                {
    static int produce_call_count = 0;
    produce_call_count++;
    ldmx_log(debug) << "produce() called, count=" << produce_call_count;
    ldmx_log(debug) << "verbose_parse=" << (m_verbose_parse ? "true" : "false");
    if (m_verbose_parse || produce_call_count <= 3) {
        ldmx_log(info) << "produce() call #" << produce_call_count;
    }
 
    // Start timing for this event
    m_event_start_time = std::chrono::steady_clock::now();
 
    // Track errors encountered during event assembly
    uint32_t current_event_errors = 0;
    
    while (m_reader && !m_reader.eof()) {
        // Try to read a RogueFrameHeader
        packing::RogueFrameHeader frame_header;
        frame_header.read(m_reader);
        
        // Store location of end-of-frame for potential recovery
        const long long frame_end =
            static_cast<long long>(m_reader.tell()) + frame_header.size();
        
        // Check if this is a data frame (channel 0) and not YAML
        if (frame_header.channel() != 0 || frame_header.probablyYaml()) {
            // Skip this frame
            if (m_verbose_parse) ldmx_log(debug) << "skipping non-data frame (channel=" << frame_header.channel() << ")";
            m_reader.seek(frame_end);
            continue;
        }
        
        // We have a valid data frame - try to parse it
        DataFragment fragment{{}, {}, {.checksum = 0}};
        bool parsed_ok = false;
        
        // Try RoR format first
        packing::LDMXRoRHeader ror_header;
        int pos_before_ror = m_reader.tell();
        
        // Attempt to read as RoR header
        try {
            ror_header.read(m_reader);
            // Successfully parsed RoR header
            fragment.header.subsystem_id = static_cast<uint64_t>(ror_header.subsystem());
            fragment.header.contributor_id = static_cast<uint64_t>(ror_header.contributor());
            fragment.header.timestamp = ror_header.timestamp();
            
            if (m_verbose_parse) {
                ldmx_log(debug) << "parsed RoR header subsys=" << (int)ror_header.subsystem()
                                << " contrib=" << (int)ror_header.contributor() 
                                << " ts=" << ror_header.timestamp();
            }
            
            // Read remaining frame payload after RoR header
            std::vector<uint8_t> payload_data;
            long long payload_size = frame_end - m_reader.tell();
            if (payload_size > 0) {
                payload_data.resize(payload_size);
                m_reader.read(reinterpret_cast<char*>(payload_data.data()), payload_size);
            }
            fragment.payload = std::move(payload_data);
            parsed_ok = true;
        } catch (...) {
            // RoR parse failed, try packing subsystem format
            if (m_verbose_parse) ldmx_log(debug) << "RoR header parse failed, trying packing subsystem format";
            current_event_errors |= ldmx::EventSummary::ERROR_PARSE_FAILURE;
            m_reader.seek(pos_before_ror);
        }
        
        if (!parsed_ok) {
            // Try packing subsystem format
            packing::rawdatafile::SubsystemPacket pkt;
            try {
                pkt.read(m_reader);
                fragment.header.subsystem_id = static_cast<uint64_t>(pkt.id());
                fragment.header.contributor_id = 0;  // Not available in packing subsystem format
                fragment.header.timestamp = static_cast<uint64_t>(pkt.header()[1]) * 1000000000ULL;  // event number to ns
                
                // Convert packet data to payload bytes
                const auto& data = pkt.data();
                fragment.payload.resize(data.size() * 4);
                std::memcpy(fragment.payload.data(), reinterpret_cast<const char*>(data.data()), data.size() * 4);
                
                if (m_verbose_parse) {
                    ldmx_log(debug) << "parsed packing subsystem pkt subsys=" << pkt.id()
                                    << " data_size=" << data.size();
                }
                parsed_ok = true;
            } catch (...) {
                // Both parse attempts failed, skip this frame
                if (m_verbose_parse) ldmx_log(debug) << "failed to parse as either format, skipping frame";
                current_event_errors |= ldmx::EventSummary::ERROR_PARSE_FAILURE;
                m_reader.seek(frame_end);
                continue;
            }
        }
        
        if (!parsed_ok) {
            m_reader.seek(frame_end);
            continue;
        }
 
        if (m_verbose_parse) ldmx_log(debug) << "adding fragment subsys=" << fragment.header.subsystem_id
                                             << " ts=" << fragment.header.timestamp << " bytes=" << fragment.payload.size();
        
        // Check if this fragment is outside the coherence window of the current event batch
        // If so, it means we should finalize the previous event before adding this one
        long long fragment_ts = fragment.header.timestamp;
        long long buffer_ref_time = m_event_buffer.get_reference_time();
        
        if (buffer_ref_time > 0 && (fragment_ts < buffer_ref_time - m_coherence_window_ns || 
                                     fragment_ts > buffer_ref_time + m_coherence_window_ns)) {
            // This fragment is outside the current window - try to build the previous event
            std::vector<DataFragment> assembled_event_fragments;
            if (m_event_buffer.try_build_event(m_coherence_window_ns, m_min_subsystems, assembled_event_fragments) &&
                !assembled_event_fragments.empty()) {
                // Output the complete event and return
                ++m_event_id;
                // Add each subsystem's raw data as vector<uint8_t> directly
                uint64_t event_timestamp = 0;
                for (const auto &frag : assembled_event_fragments) {
                    std::string subsys_name = packing::LDMXRoRHeader::getSubsystemName(
                        static_cast<uint8_t>(frag.header.subsystem_id),
                        static_cast<uint8_t>(frag.header.contributor_id));
                    std::vector<uint8_t> payload_bytes(frag.payload.begin(), frag.payload.end());
                    event.add(subsys_name, payload_bytes);
                    if (event_timestamp == 0) {
                        event_timestamp = frag.header.timestamp;
                    }
                }
                event.getEventHeader().setIntParameter("RoR Timestamp", static_cast<long long>(event_timestamp));
                // Still create PhysicsEventData for binary output file
                PhysicsEventData final_event = assemble_payload(assembled_event_fragments);
                write_event_binary(final_event, "events.bin");
                ldmx_log(info) << "assembled event id=" << m_event_id << " timestamp=" << final_event.timestamp
                               << " fragments=" << assembled_event_fragments.size() << " systems=" << final_event.systems_readout.size();
                
                ldmx::EventSummary summary;
                summary.setEventNumber(m_event_id);
                summary.setTimestampNs(static_cast<uint64_t>(final_event.timestamp));
                std::set<uint64_t> unique_sys;
                uint64_t total_payload = 0;
                for (const auto &f : assembled_event_fragments) {
                    unique_sys.insert(f.header.subsystem_id);
                    total_payload += f.payload.size();
                }
                // Check for duplicate subsystems
                if (unique_sys.size() != assembled_event_fragments.size()) {
                    current_event_errors |= ldmx::EventSummary::ERROR_DUPLICATE_SUBSYSTEM;
                }
                summary.setNSystems(static_cast<uint32_t>(unique_sys.size()));
                summary.setSystemIds(std::vector<uint64_t>(unique_sys.begin(), unique_sys.end()));
                summary.setPayloadSize(total_payload);
                summary.setErrorFlags(current_event_errors);
                event.add("EventSummary", summary);
                
                // Update performance metrics
                m_total_bytes_read += total_payload;
                m_total_events_built++;
                m_events_since_last_report++;
                std::chrono::steady_clock::time_point now = std::chrono::steady_clock::now();
                std::chrono::milliseconds event_build_time_ms = std::chrono::duration_cast<std::chrono::milliseconds>(now - m_event_start_time);
                std::chrono::seconds total_elapsed_s = std::chrono::duration_cast<std::chrono::seconds>(now - m_start_time);
                
                double events_per_sec = (total_elapsed_s.count() > 0) ? static_cast<double>(m_total_events_built) / total_elapsed_s.count() : 0.0;
                double mb_per_sec = (total_elapsed_s.count() > 0) ? static_cast<double>(m_total_bytes_read) / (1024.0 * 1024.0) / total_elapsed_s.count() : 0.0;
                
                // Update windowed metrics
                m_window_events_count++;
                m_window_bytes_read += total_payload;
                std::chrono::seconds window_elapsed_s = std::chrono::duration_cast<std::chrono::seconds>(now - m_window_start_time);
                
                // Reset window if we've processed WINDOW_SIZE events
                double window_events_per_sec = 0.0;
                double window_mb_per_sec = 0.0;
                if (m_window_events_count >= WINDOW_SIZE) {
                    double window_time_sec = std::max(1.0, static_cast<double>(window_elapsed_s.count()));
                    window_events_per_sec = static_cast<double>(m_window_events_count) / window_time_sec;
                    window_mb_per_sec = static_cast<double>(m_window_bytes_read) / (1024.0 * 1024.0) / window_time_sec;
                    
                    // Reset window
                    m_window_start_time = now;
                    m_window_events_count = 0;
                    m_window_bytes_read = 0;
                }
                
                // Write performance metrics to CSV
                long long event_build_time_ms_val = event_build_time_ms.count();
                write_performance_metric(m_event_id, static_cast<double>(event_build_time_ms_val), 
                                        events_per_sec, mb_per_sec, 
                                        window_events_per_sec, window_mb_per_sec,
                                        m_total_events_built, m_total_bytes_read);
                
                if (m_verbose_parse || m_events_since_last_report % 100 == 0) {
                    ldmx_log(info) << "Performance: "
                                   << "total_events=" << m_total_events_built << ", "
                                   << "events_per_sec=" << std::fixed << std::setprecision(2) << events_per_sec << ", "
                                   << "mb_per_sec=" << std::fixed << std::setprecision(3) << mb_per_sec;
                }
                
                current_event_errors = 0;  // Reset for next event
                // The fragment that tripped the window-close belongs to the NEXT
                // event, not the one we just emitted. Buffer it here (the buffer is
                // empty post-build, so this also resets the reference time) instead
                // of dropping it -- otherwise every event after the first loses its
                // triggering subsystem.
                m_event_buffer.add_fragment(std::move(fragment));
                return;  // Return the event to framework
            }
        }
 
        // Add fragment to buffer (may start a new event batch if buffer was empty)
        m_event_buffer.add_fragment(std::move(fragment));
        
        if (m_verbose_parse) ldmx_log(debug) << "frame added to buffer, searching for more frames...";
    }
 
    // Reached EOF - flush any remaining events in the buffer
    if (m_verbose_parse) ldmx_log(debug) << "reached EOF, flushing remaining events";
    
    std::vector<DataFragment> assembled_event_fragments;
    while (m_event_buffer.try_build_event(m_coherence_window_ns, m_min_subsystems, assembled_event_fragments)) {
        if (assembled_event_fragments.empty()) break;
        
        // Mark truncated events (those flushed at EOF)
        current_event_errors |= ldmx::EventSummary::ERROR_TRUNCATED_EVENT;
        
        ++m_event_id;
        // Add each subsystem's raw data as vector<uint8_t> directly
        uint64_t event_timestamp = 0;
        for (const auto &frag : assembled_event_fragments) {
            std::string subsys_name = packing::LDMXRoRHeader::getSubsystemName(
                static_cast<uint8_t>(frag.header.subsystem_id),
                static_cast<uint8_t>(frag.header.contributor_id));
            std::vector<uint8_t> payload_bytes(frag.payload.begin(), frag.payload.end());
            event.add(subsys_name, payload_bytes);
            if (event_timestamp == 0) {
                event_timestamp = frag.header.timestamp;
            }
        }
        event.getEventHeader().setIntParameter("RoR Timestamp", static_cast<long long>(event_timestamp));
        // Still create PhysicsEventData for binary output file
        PhysicsEventData final_event = assemble_payload(assembled_event_fragments);
        write_event_binary(final_event, "events.bin");
        ldmx_log(info) << "assembled event id=" << m_event_id << " timestamp=" << final_event.timestamp
                       << " fragments=" << assembled_event_fragments.size() << " systems=" << final_event.systems_readout.size();
        
        ldmx::EventSummary summary;
        summary.setEventNumber(m_event_id);
        summary.setTimestampNs(static_cast<uint64_t>(final_event.timestamp));
        std::set<uint64_t> unique_sys;
        uint64_t total_payload = 0;
        for (const auto &f : assembled_event_fragments) {
            unique_sys.insert(f.header.subsystem_id);
            total_payload += f.payload.size();
        }
        // Check for duplicate subsystems
        if (unique_sys.size() != assembled_event_fragments.size()) {
            current_event_errors |= ldmx::EventSummary::ERROR_DUPLICATE_SUBSYSTEM;
        }
        summary.setNSystems(static_cast<uint32_t>(unique_sys.size()));
        summary.setSystemIds(std::vector<uint64_t>(unique_sys.begin(), unique_sys.end()));
        summary.setPayloadSize(total_payload);
        summary.setErrorFlags(current_event_errors);
        event.add("EventSummary", summary);
        
        // Update performance metrics
        m_total_bytes_read += total_payload;
        m_total_events_built++;
        m_events_since_last_report++;
        std::chrono::steady_clock::time_point now = std::chrono::steady_clock::now();
        std::chrono::milliseconds event_build_time_ms = std::chrono::duration_cast<std::chrono::milliseconds>(now - m_event_start_time);
        std::chrono::seconds total_elapsed_s = std::chrono::duration_cast<std::chrono::seconds>(now - m_start_time);
        
        double events_per_sec = (total_elapsed_s.count() > 0) ? static_cast<double>(m_total_events_built) / total_elapsed_s.count() : 0.0;
        double mb_per_sec = (total_elapsed_s.count() > 0) ? static_cast<double>(m_total_bytes_read) / (1024.0 * 1024.0) / total_elapsed_s.count() : 0.0;
        
        // Update windowed metrics
        m_window_events_count++;
        m_window_bytes_read += total_payload;
        std::chrono::seconds window_elapsed_s = std::chrono::duration_cast<std::chrono::seconds>(now - m_window_start_time);
        
        // Calculate window metrics (for final event, may not have full window)
        double window_events_per_sec = 0.0;
        double window_mb_per_sec = 0.0;
        if (m_window_events_count >= WINDOW_SIZE || window_elapsed_s.count() > 0) {
            double window_time_sec = std::max(1.0, static_cast<double>(window_elapsed_s.count()));
            window_events_per_sec = static_cast<double>(m_window_events_count) / window_time_sec;
            window_mb_per_sec = static_cast<double>(m_window_bytes_read) / (1024.0 * 1024.0) / window_time_sec;
        }
        
        // Write performance metrics to CSV
        long long event_build_time_ms_val = event_build_time_ms.count();
        write_performance_metric(m_event_id, static_cast<double>(event_build_time_ms_val), 
                                events_per_sec, mb_per_sec,
                                window_events_per_sec, window_mb_per_sec,
                                m_total_events_built, m_total_bytes_read);
        
        if (m_verbose_parse || m_events_since_last_report % 100 == 0) {
            ldmx_log(info) << "Performance: "
                           << "total_events=" << m_total_events_built << ", "
                           << "events_per_sec=" << std::fixed << std::setprecision(2) << events_per_sec << ", "
                           << "mb_per_sec=" << std::fixed << std::setprecision(3) << mb_per_sec;
        }
        
        current_event_errors = 0;  // Reset for next event
        
        assembled_event_fragments.clear();
        return;
    }
    
    // No more events - print final summary statistics
    std::chrono::steady_clock::time_point final_time = std::chrono::steady_clock::now();
    std::chrono::seconds total_time_s = std::chrono::duration_cast<std::chrono::seconds>(final_time - m_start_time);
    double final_events_per_sec = (total_time_s.count() > 0) ? static_cast<double>(m_total_events_built) / total_time_s.count() : 0.0;
    double final_mb_per_sec = (total_time_s.count() > 0) ? static_cast<double>(m_total_bytes_read) / (1024.0 * 1024.0) / total_time_s.count() : 0.0;
    
    ldmx_log(info) << "\n===== FINAL STATISTICS =====";
    ldmx_log(info) << "Total events built: " << m_total_events_built;
    ldmx_log(info) << "Total bytes read: " << m_total_bytes_read / (1024.0 * 1024.0) << " MB";
    ldmx_log(info) << "Total time: " << total_time_s.count() << " seconds";
    ldmx_log(info) << "Average throughput: "
                   << "events_per_sec=" << std::fixed << std::setprecision(2) << final_events_per_sec << ", "
                   << "mb_per_sec=" << std::fixed << std::setprecision(3) << final_mb_per_sec;
    ldmx_log(info) << "=============================\n";
    ldmx_log(info) << "Event building complete";
    
    abortEvent();
}

References framework::EventProcessor::abortEvent(), packing::RogueFrameHeader::channel(), packing::LDMXRoRHeader::contributor(), packing::rawdatafile::SubsystemPacket::data(), packing::utility::Reader::eof(), ldmx::EventSummary::ERROR_DUPLICATE_SUBSYSTEM, ldmx::EventSummary::ERROR_PARSE_FAILURE, ldmx::EventSummary::ERROR_TRUNCATED_EVENT, packing::LDMXRoRHeader::getSubsystemName(), packing::rawdatafile::SubsystemPacket::header(), packing::RogueFrameHeader::probablyYaml(), packing::LDMXRoRHeader::read(), packing::rawdatafile::SubsystemPacket::read(), packing::RogueFrameHeader::read(), packing::utility::Reader::read(), packing::utility::Reader::seek(), ldmx::EventSummary::setErrorFlags(), ldmx::EventSummary::setEventNumber(), ldmx::EventSummary::setNSystems(), ldmx::EventSummary::setPayloadSize(), ldmx::EventSummary::setSystemIds(), ldmx::EventSummary::setTimestampNs(), packing::RogueFrameHeader::size(), packing::LDMXRoRHeader::subsystem(), packing::utility::Reader::tell(), and packing::LDMXRoRHeader::timestamp().

write_event_binary()

void EventBuilder::write_event_binary ( const PhysicsEventData & ev, const std::string & path = "events.bin" )

private

Definition at line 457 of file EventBuilder.cxx.

                                                                                       {
    static std::mutex g_out_mutex;
    std::lock_guard<std::mutex> lg(g_out_mutex);
    std::ofstream ofs(path, std::ios::binary | std::ios::app);
    if (!ofs) return;
    uint64_t event_id_u = static_cast<uint64_t>(ev.event_id);
    uint64_t ts = static_cast<uint64_t>(ev.timestamp);
    uint32_t nblocks = static_cast<uint32_t>(ev.blocks.size());
    ofs.write(reinterpret_cast<const char*>(&event_id_u), sizeof(event_id_u));
    ofs.write(reinterpret_cast<const char*>(&ts), sizeof(ts));
    ofs.write(reinterpret_cast<const char*>(&nblocks), sizeof(nblocks));
    for (const auto &b : ev.blocks) {
        uint64_t sid = b.subsystem_id;
        uint64_t bts = b.timestamp_ns;
        uint32_t psz = static_cast<uint32_t>(b.data.size());
        uint32_t csum = b.checksum;
        ofs.write(reinterpret_cast<const char*>(&sid), sizeof(sid));
        ofs.write(reinterpret_cast<const char*>(&bts), sizeof(bts));
        ofs.write(reinterpret_cast<const char*>(&psz), sizeof(psz));
        ofs.write(reinterpret_cast<const char*>(&csum), sizeof(csum));
        if (psz) ofs.write(reinterpret_cast<const char*>(b.data.data()), static_cast<std::streamsize>(psz));
    }
    ofs.flush();
}

Member Data Documentation

m_coherence_window_ns

long long eventbuilder::EventBuilder::m_coherence_window_ns {5000000}

private

Definition at line 38 of file EventBuilder.h.

{5000000};  // 5 ms window for collecting fragments

m_event_buffer

FragmentBuffer eventbuilder::EventBuilder::m_event_buffer

private

Definition at line 34 of file EventBuilder.h.

m_event_id

unsigned int eventbuilder::EventBuilder::m_event_id

private

Definition at line 33 of file EventBuilder.h.

m_event_start_time

std::chrono::steady_clock::time_point eventbuilder::EventBuilder::m_event_start_time

private

Definition at line 45 of file EventBuilder.h.

m_events_since_last_report

unsigned long eventbuilder::EventBuilder::m_events_since_last_report {0}

private

Definition at line 48 of file EventBuilder.h.

{0};

m_input_file

std::string eventbuilder::EventBuilder::m_input_file

private

Definition at line 35 of file EventBuilder.h.

m_min_subsystems

int eventbuilder::EventBuilder::m_min_subsystems {2}

private

Definition at line 41 of file EventBuilder.h.

{2};

m_output_name

std::string eventbuilder::EventBuilder::m_output_name {"BuilderOutput"}

private

Definition at line 37 of file EventBuilder.h.

{"BuilderOutput"};

m_reader

packing::utility::Reader eventbuilder::EventBuilder::m_reader

private

Definition at line 36 of file EventBuilder.h.

m_start_time

std::chrono::steady_clock::time_point eventbuilder::EventBuilder::m_start_time

private

Definition at line 44 of file EventBuilder.h.

m_total_bytes_read

uint64_t eventbuilder::EventBuilder::m_total_bytes_read {0}

private

Definition at line 46 of file EventBuilder.h.

{0};

m_total_events_built

uint64_t eventbuilder::EventBuilder::m_total_events_built {0}

private

Definition at line 47 of file EventBuilder.h.

{0};

m_verbose_parse

bool eventbuilder::EventBuilder::m_verbose_parse

private

Definition at line 32 of file EventBuilder.h.

m_window_bytes_read

uint64_t eventbuilder::EventBuilder::m_window_bytes_read {0}

private

Definition at line 54 of file EventBuilder.h.

{0};

m_window_events_count

uint64_t eventbuilder::EventBuilder::m_window_events_count {0}

private

Definition at line 53 of file EventBuilder.h.

{0};

m_window_start_time

std::chrono::steady_clock::time_point eventbuilder::EventBuilder::m_window_start_time

private

Definition at line 52 of file EventBuilder.h.

WINDOW_SIZE

unsigned int eventbuilder::EventBuilder::WINDOW_SIZE = 100

staticconstexprprivate

Definition at line 51 of file EventBuilder.h.

---

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

• EventBuilder/include/EventBuilder/EventBuilder.h
• EventBuilder/src/EventBuilder/EventBuilder.cxx