Line data Source code
1 : /**
2 : * @file DAPHNEFrameProcessor.hpp DAPHNE specific Task based raw processor
3 : * implementation
4 : *
5 : * This is part of the DUNE DAQ , copyright 2020.
6 : * Licensing/copyright details are in the COPYING file that you should have
7 : * received with this code.
8 : */
9 :
10 : #include "fdreadoutlibs/daphne/DAPHNEFrameProcessor.hpp"
11 : #include "fddetdataformats/DAPHNEFrame.hpp"
12 : #include "trgdataformats/TriggerPrimitive.hpp"
13 :
14 : #include "confmodel/GeoId.hpp"
15 :
16 : #include <atomic>
17 : #include <functional>
18 : #include <memory>
19 : #include <string>
20 :
21 : using dunedaq::datahandlinglibs::logging::TLVL_BOOKKEEPING;
22 : using dunedaq::datahandlinglibs::logging::TLVL_FRAME_RECEIVED;
23 :
24 : DUNE_DAQ_TYPESTRING(dunedaq::trigger::TriggerPrimitiveTypeAdapter, "TriggerPrimitive")
25 0 : DUNE_DAQ_TYPESTRING(std::vector<dunedaq::trigger::TriggerPrimitiveTypeAdapter>, "TriggerPrimitiveVector")
26 :
27 : namespace dunedaq {
28 : namespace fdreadoutlibs {
29 :
30 : void
31 0 : DAPHNEFrameProcessor::conf(const appmodel::DataHandlerModule* conf)
32 : {
33 0 : TLOG() << "Looking for TP sink...";
34 :
35 0 : for (auto output : conf->get_outputs()) {
36 0 : TLOG() << "On outputs... (" << output->UID() << "," << output->get_data_type() << ")";
37 0 : try {
38 0 : if (output->get_data_type() == "TriggerPrimitiveVector") {
39 0 : TLOG() << "Found TP sink.";
40 0 : m_tp_sink = get_iom_sender<std::vector<trigger::TriggerPrimitiveTypeAdapter>>(output->UID());
41 0 : TLOG() << " SINK INITIALIZED for TriggerPrimitives with UID : " << output->UID();
42 : }
43 0 : } catch (const ers::Issue& excpt) {
44 0 : ers::error(datahandlinglibs::ResourceQueueError(ERS_HERE, "tp", "DefaultRequestHandlerModel", excpt));
45 0 : }
46 : }
47 :
48 0 : TLOG() << "Registering processing tasks...";
49 0 : inherited::add_preprocess_task(std::bind(&DAPHNEFrameProcessor::timestamp_check, this, std::placeholders::_1));
50 :
51 0 : auto dp = conf->get_module_configuration()->get_data_processor();
52 0 : if (dp == nullptr) {
53 : TLOG() << " PDS Data processor does not exist.";
54 : } else {
55 0 : auto proc_conf = dp->cast<appmodel::PDSRawDataProcessor>();
56 0 : if (proc_conf == nullptr) {
57 0 : TLOG() << "PDS RawDataProcessor does not exist.";
58 : } else {
59 0 : m_def_adc_intg_thresh = proc_conf->get_default_adc_intg_thresh();
60 :
61 0 : auto geo_id = conf->get_geo_id();
62 0 : if (geo_id != nullptr) {
63 0 : m_det_id = geo_id->get_detector_id();
64 0 : m_crate_id = geo_id->get_crate_id();
65 0 : m_slot_id = geo_id->get_slot_id();
66 0 : m_stream_id = geo_id->get_stream_id();
67 : }
68 :
69 0 : m_channel_map = dunedaq::detchannelmaps::make_pds_map(proc_conf->get_channel_map());
70 0 : const std::vector<unsigned int> channel_mask_vec = proc_conf->get_channel_mask();
71 :
72 0 : for (int chan = 0; chan < 48; chan++) { // 40 physical PDS channel 8 not. 0->7 contain light info, 8,9, additional
73 : // info. 10-17 light, 18,19 not etc...
74 0 : trgdataformats::channel_t off_channel = m_channel_map->get_offline_channel_from_det_crate_slot_stream_chan(
75 0 : m_det_id, m_crate_id, m_slot_id, m_stream_id, chan);
76 0 : if (std::find(channel_mask_vec.begin(), channel_mask_vec.end(), off_channel) != channel_mask_vec.end())
77 0 : m_channel_mask_set.insert(
78 : off_channel); // m_channel_mask will be a vector fille with random chanel which need to be masked.
79 : }
80 :
81 0 : if (m_post_processing_enabled) {
82 : // Extract TPs back as a pre-processing task, due to LatencyBuffer post-proc issues using SkipList.
83 0 : inherited::add_preprocess_task(std::bind(&DAPHNEFrameProcessor::extract_tps, this, std::placeholders::_1));
84 : }
85 0 : }
86 : }
87 :
88 0 : TLOG() << "Calling parent conf.";
89 0 : inherited::conf(conf);
90 0 : }
91 :
92 : void
93 0 : DAPHNEFrameProcessor::start(const appfwk::DAQModule::CommandData_t& args)
94 : {
95 : // Reset timestamp check
96 0 : m_previous_ts = 0;
97 0 : m_current_ts = 0;
98 0 : m_first_ts_missmatch = true;
99 0 : m_ts_error_ctr = 0;
100 :
101 : // Reset stats
102 0 : m_t0 = std::chrono::high_resolution_clock::now();
103 0 : m_num_new_tps.exchange(0);
104 :
105 0 : inherited::start(args);
106 0 : }
107 : void
108 0 : DAPHNEFrameProcessor::stop(const appfwk::DAQModule::CommandData_t& args)
109 : {
110 0 : inherited::stop(args);
111 0 : }
112 : /**
113 : * Pipeline Stage 1.: Check proper timestamp increments in DAPHNE frame
114 : * */
115 : void
116 0 : DAPHNEFrameProcessor::timestamp_check(frameptr fp)
117 : {
118 : // Let Source Emulator deal with this
119 : /*
120 : // If EMU data, emulate perfectly incrementing timestamp
121 : if (inherited::m_emulator_mode) { // emulate perfectly incrementing timestamp
122 : // RS warning : not fixed rate!
123 : if (m_first_ts_fake) {
124 : fp->fake_timestamps(m_previous_ts, 16);
125 : m_first_ts_fake = false;
126 : } else {
127 : fp->fake_timestamps(m_previous_ts + 192, 16);
128 : }
129 : }*/
130 :
131 : // FIXME: This is a temporary fix to avoid frames with unphysical timestamp set to the far future to interfere with
132 : // the operations of the LB.
133 : // These frames are effectively "corrupted" or "invalid frames" and hould be handled as such.
134 :
135 0 : auto df_ptr = reinterpret_cast<dunedaq::fddetdataformats::DAPHNEFrame*>(fp);
136 :
137 0 : for (size_t i = 0; i < fp->get_num_frames(); ++i) {
138 :
139 0 : if (df_ptr[i].get_timestamp() > 0xFFFFFFFFFFFF0000 || df_ptr[i].get_timestamp() < 0xFFFF) {
140 0 : ers::warning(PDSUnphysicalFrameTimestamp(ERS_HERE, df_ptr[i].get_timestamp(), df_ptr[i].get_channel(), i));
141 0 : df_ptr[i].set_timestamp(0);
142 : }
143 : }
144 :
145 : // Acquire timestamp
146 0 : m_current_ts = fp->get_timestamp();
147 0 : uint64_t k_clock_frequency = 62500000; // NOLINT(build/unsigned)
148 0 : TLOG_DEBUG(TLVL_FRAME_RECEIVED) << "Received DAPHNE frame timestamp value of " << m_current_ts << " ticks (..."
149 0 : << std::fixed << std::setprecision(8)
150 0 : << (static_cast<double>(m_current_ts % (k_clock_frequency * 1000)) /
151 0 : static_cast<double>(k_clock_frequency))
152 0 : << " sec)"; // NOLINT
153 :
154 0 : if (m_ts_error_ctr > 1000) {
155 0 : if (!m_problem_reported) {
156 0 : std::cout << "*** Data Integrity ERROR *** Timestamp continuity is completely broken! "
157 0 : << "Something is wrong with the FE source or with the configuration!\n";
158 0 : m_problem_reported = true;
159 : }
160 : }
161 :
162 0 : m_previous_ts = m_current_ts;
163 0 : m_last_processed_daq_ts = m_current_ts;
164 0 : }
165 :
166 : /**
167 : * Pipeline Stage 2.: Check DAPHNE headers for error flags
168 : * */
169 : void
170 0 : DAPHNEFrameProcessor::frame_error_check(frameptr /*fp*/)
171 : {
172 : // check error fields
173 0 : }
174 :
175 : void
176 0 : DAPHNEFrameProcessor::extract_tps(constframeptr fp)
177 : {
178 :
179 0 : if (!fp || fp == nullptr) {
180 : return;
181 : }
182 :
183 0 : auto nonconstframeptr = const_cast<frameptr>(fp);
184 0 : auto df_ptr = reinterpret_cast<dunedaq::fddetdataformats::DAPHNEFrame*>((uint8_t*)nonconstframeptr); // NOLINT
185 0 : std::vector<trigger::TriggerPrimitiveTypeAdapter> ttpp;
186 :
187 0 : for (size_t i = 0; i < fp->get_num_frames(); i++) {
188 0 : for (size_t j = 0; j < fddetdataformats::DAPHNEFrame::PeakDescriptorData::max_peaks; j++) {
189 0 : if (df_ptr[i].peaks_data.is_found(j)) {
190 0 : int ch = m_channel_map->get_offline_channel_from_det_crate_slot_stream_chan(df_ptr[i].daq_header.det_id,
191 0 : df_ptr[i].daq_header.crate_id,
192 0 : df_ptr[i].daq_header.slot_id,
193 0 : df_ptr[i].daq_header.link_id,
194 0 : df_ptr[i].get_channel());
195 0 : if (std::binary_search(m_channel_mask_set.begin(), m_channel_mask_set.end(), ch))
196 0 : continue;
197 0 : if (df_ptr[i].peaks_data.get_adc_integral(j) < m_def_adc_intg_thresh)
198 0 : continue;
199 :
200 0 : trigger::TriggerPrimitiveTypeAdapter tpa;
201 0 : tpa.tp = peak_to_tp(df_ptr[i], j); // this is the trigger primitive
202 : // check for timestamps that are due to frame timestamps ~ ts=0, and ignore these peaks
203 0 : if (tpa.tp.time_start > 0xFFFFFFFFFFFF0000 || tpa.tp.time_start < 0xFFFF) {
204 0 : ers::warning(PDSPeakIgnored(ERS_HERE, tpa.tp.time_start, tpa.tp.channel, i, j));
205 0 : continue;
206 : }
207 :
208 0 : tpa.tp.detid = df_ptr->daq_header.det_id;
209 0 : ttpp.push_back(tpa);
210 : }
211 : }
212 : }
213 :
214 0 : int num_new_tps = ttpp.size();
215 0 : if (num_new_tps > 0) {
216 :
217 0 : const auto s_ts_begin = ttpp.front().tp.time_start;
218 0 : const auto channel_begin = ttpp.front().tp.channel;
219 0 : const auto s_ts_end = ttpp.back().tp.time_start;
220 0 : const auto channel_end = ttpp.back().tp.channel;
221 :
222 0 : if (!m_tp_sink->try_send(std::move(ttpp), iomanager::Sender::s_no_block)) {
223 0 : ers::warning(FailedToSendTPVector(ERS_HERE, s_ts_begin, channel_begin, s_ts_end, channel_end));
224 0 : m_tps_send_failed += num_new_tps;
225 : } else {
226 0 : m_num_new_tps += num_new_tps;
227 : }
228 : }
229 :
230 0 : return;
231 0 : }
232 :
233 : dunedaq::trgdataformats::TriggerPrimitive
234 0 : DAPHNEFrameProcessor::peak_to_tp(dunedaq::fddetdataformats::DAPHNEFrame& frame, int i)
235 : {
236 0 : dunedaq::trgdataformats::TriggerPrimitive tp;
237 : // TODO: add check on peak presence
238 0 : tp.time_start = frame.get_timestamp() + frame.peaks_data.get_sample_start(i);
239 0 : tp.samples_to_peak = frame.peaks_data.get_sample_max(i);
240 0 : tp.samples_over_threshold = frame.peaks_data.get_samples_over_baseline(i);
241 : // FIXME : hard-coded channel map
242 : // WARNING: slot ids in DAPHNEs are all 0!
243 0 : tp.channel = m_channel_map->get_offline_channel_from_det_crate_slot_stream_chan(frame.daq_header.det_id,
244 0 : frame.daq_header.crate_id,
245 0 : frame.daq_header.slot_id,
246 0 : frame.daq_header.link_id,
247 0 : frame.get_channel());
248 0 : tp.adc_integral = frame.peaks_data.get_adc_integral(i);
249 0 : tp.adc_peak = frame.peaks_data.get_adc_max(i);
250 0 : tp.detid = dunedaq::trgdataformats::TypeDefaults::s_invalid_detid;
251 0 : return tp;
252 : }
253 :
254 : void
255 0 : DAPHNEFrameProcessor::generate_opmon_data()
256 : {
257 :
258 : // right now, just fill some basic tp info...
259 0 : if (m_post_processing_enabled) {
260 0 : auto now = std::chrono::high_resolution_clock::now();
261 0 : int num_new_tps = m_num_new_tps.exchange(0);
262 0 : int new_tps_suppressed_too_long = 0; // not relevant for PDS TPs
263 0 : int new_tps_send_failed = m_tps_send_failed.exchange(0);
264 0 : double seconds = std::chrono::duration_cast<std::chrono::microseconds>(now - m_t0).count() / 1000000.;
265 0 : TLOG_DEBUG(TLVL_BOOKKEEPING) << "TP rate: " << std::to_string(num_new_tps / seconds / 1000.) << " [kHz]";
266 0 : TLOG_DEBUG(TLVL_BOOKKEEPING) << "Total new TPs: " << num_new_tps;
267 :
268 0 : datahandlinglibs::opmon::HitFindingInfo tp_info;
269 0 : tp_info.set_rate_tp_hits(num_new_tps / seconds / 1000.);
270 :
271 0 : tp_info.set_num_tps_sent(num_new_tps);
272 0 : tp_info.set_num_tps_suppressed_too_long(new_tps_suppressed_too_long);
273 0 : tp_info.set_num_tps_send_failed(new_tps_send_failed);
274 :
275 0 : publish(std::move(tp_info));
276 :
277 0 : m_t0 = now;
278 0 : }
279 :
280 0 : inherited::generate_opmon_data();
281 0 : }
282 :
283 : } // namespace fdreadoutlibs
284 : } // namespace dunedaq
|
