liftof_lib/
master_trigger.rs

1//! MasterTriggerBoard communications
2//!
3//! The MTB (MasterTriggerBoard) is currently
4//! (Jan 2023) connected to the ethernet 
5//! via UDP sockets and sends out its 
6//! own datapackets per each triggered 
7//! event.
8//!
9//! The packet format contains the event id
10//! as well as number of hits and a mask 
11//! which encodes the hit channels.
12//!
13//! The data is encoded in IPBus packets.
14//! [see docs here](https://ipbus.web.cern.ch/doc/user/html/)
15//! 
16
17pub mod control;
18pub mod registers;
19
20use std::sync::{
21  Arc,
22  Mutex,
23};
24
25use std::time::{
26  Duration,
27  Instant
28};
29
30use std::thread;
31use crossbeam_channel::Sender;
32use serde_json::json;
33
34use tof_dataclasses::packets::TofPacket;
35use tof_dataclasses::monitoring::MtbMoniData;
36use tof_dataclasses::events::MasterTriggerEvent;
37use tof_dataclasses::events::master_trigger::TriggerType;
38use tof_dataclasses::version::ProtocolVersion;
39use tof_dataclasses::errors::{
40  MasterTriggerError
41};
42
43use tof_dataclasses::ipbus::IPBus;
44
45use tof_dataclasses::heartbeats::MTBHeartbeat;
46use tof_dataclasses::serialization::Packable;
47
48use crate::thread_control::ThreadControl;
49
50// make this public to not brake liftof-cc
51pub use crate::settings::MTBSettings;
52
53use control::*;
54use registers::*;
55
56/// helper function to parse output for TofBot
57fn remove_from_word(s: String, word: &str) -> String {
58  if let Some(index) = s.find(word) {
59    // Keep everything up to the found index (not including the word itself)
60    s[..index].to_string()
61  } else {
62    // If the word isn't found, return the original string
63    s
64  }
65}
66
67/// In case we get a broken DAQ package, 
68/// make sure we at least read it until the next 
69/// footer
70fn read_until_footer(bus : &mut IPBus) 
71  -> Result<Vec<u32>, MasterTriggerError> {
72  let mut data = Vec::<u32>::new();
73  loop {
74    let val = bus.read(0x11)?;
75    if val != 0xAAAAAAAA {
76        data.push(val);
77    }
78    if (val == 0x55555555) || (val == 0xAAAAAAAA) {
79      break;
80    }
81  }
82  Ok(data)
83}
84
85
86/// Read the complete event of the MTB
87///
88/// FIXME - this can get extended to read 
89/// multiple events at once. 
90/// For that, we just have to query the
91/// event size register multiple times.
92///
93/// <div class="warning"> Blocki until a UDP timeout error occurs or a non-zero result for MT.EVENT_QUEUE.SIZE register has been obtained.</div>
94///
95/// # Arguments
96///
97/// * bus       : connected IPBus for UDP comms
98pub fn get_event(bus                     : &mut IPBus)
99  -> Option<Result<MasterTriggerEvent, MasterTriggerError>> {
100  let mut mte = MasterTriggerEvent::new();
101  let n_daq_words_fixed = 9u32;
102  let mut data          : Vec<u32>;
103  loop {
104    // this register tells us how many times we can read out 
105    // the DAQ data register. An event has at least 12 fields.
106    // This is for an event with 1 LTB.
107    // (see gitlab docs https://gitlab.com/ucla-gaps-tof/firmware)
108    // so we definitly wait until we have at least 12. If so 
109    // we read out the rest later.
110    // If this reutrns an error, we quit right away.
111    // FIXME: There might be a tiny bug in this register, 
112    // where it is sometimes incorrect 
113    // (https://gitlab.com/ucla-gaps-tof/firmware/-/issues/69) - nice!
114    if EVQ_SIZE.get(bus).ok()? < 12 {
115      return None;
116    } else {
117      break;
118    }
119  }
120  // we read until we get ltb information, after that 
121  match bus.read_multiple(0x11, n_daq_words_fixed as usize, false) {
122    Ok(_data) => {
123      data = _data;
124    }
125    Err(err) => {
126      return Some(Err(err.into()));
127    }
128  }
129  if data.len() < 9 {
130    // something inconsistent happened here. We were requesting more 
131    // words than we got, that is bad
132    error!("Got MTB data, but the package ends before we get LTB information!");
133    warn!("Resetting master trigger DAQ");
134    match reset_daq(bus) {//, &mt_address) {
135      Err(err) => error!("Can not reset DAQ, error {err}"),
136      Ok(_)    => ()
137    }
138    return Some(Err(MasterTriggerError::DataTooShort));
139  }
140  let n_ltb = data[8].count_ones(); 
141  // in case of an odd number of ltbs, 
142  // there are some padding bytes
143  let odd   = n_ltb % 2 != 0;
144  let n_daq_words_flex : usize;
145  let n_hit_words      : usize;
146  // get hit fields
147  if odd {
148    n_hit_words = (n_ltb as usize + 1)/2;
149  } else {
150    n_hit_words = n_ltb as usize/2;
151  }
152  // the variable size part of the DAQ event
153  n_daq_words_flex = n_hit_words + 2; // crc + footer
154  let mut data_flex : Vec<u32>;
155  match bus.read_multiple(0x11, n_daq_words_flex, false) {
156    Ok(_data) => {
157      data_flex = _data;
158    }
159    Err(err) => { 
160      return Some(Err(err.into()));
161    }
162  }
163  data.append(&mut data_flex);
164  if data[0] != 0xAAAAAAAA {
165    error!("Got MTB data, but the header is incorrect {:x}", data[0]);
166    return Some(Err(MasterTriggerError::PackageHeaderIncorrect));
167  }
168 
169  let n_daq_words = n_daq_words_fixed as usize + n_daq_words_flex;
170  let foot_pos    = n_daq_words_fixed as usize + n_daq_words_flex - 1;
171  if data.len() != foot_pos + 1{
172    error!("Somehow the MTB DATA are misaligned! {}, {}", data.len(), foot_pos);
173    return Some(Err(MasterTriggerError::DataTooShort));
174  }
175  if data[foot_pos] != 0x55555555 {
176    error!("Got MTB data, but the footer is incorrect {:x}", data[foot_pos]);
177    if data[foot_pos] == 0xAAAAAAAA {
178      error!("Found next header, the package is TOO LONG! Attempt to fix for this event, but the next is LOST!");
179      info!("If we want to fix this, this whole mechanism needs a refactor and needs to fetch more thatn a single event at a time!");
180      // kill the lost event
181      read_until_footer(bus).ok()?;
182      // salvage from this event what is possible
183      data.pop();
184    } else {
185      // we try to recover!
186      let mut rest = read_until_footer(bus).ok()?;
187      data.append(&mut rest);
188      if data.len() != n_daq_words as usize + 1 {
189        error!("We tried to recover the event, however, that failed! Expected size of the packet {}, actual size {}", n_daq_words, data.len());
190        // get some debugging information to understand why this 
191        // happened
192        println!("-------------- DEBUG -------------------");
193        println!("N LTBs {} ({})", data[8].count_ones(), data[8]);
194        for k in data {
195          println!("-- {:x} ({})", k,k);
196        }
197        println!("--------------------");
198        return Some(Err(MasterTriggerError::PackageFooterIncorrect));
199      } else {
200        info!("Event recoveered!");
201      }
202    }
203  }
204
205  // ---------- FIll the MTBEvent now
206  mte.event_id       = data[1];
207  mte.timestamp      = data[2];
208  mte.tiu_timestamp  = data[3];
209  mte.tiu_gps32      = data[4];
210  mte.tiu_gps16      =  (data[5] & 0x0000ffff) as u16;
211  mte.trigger_source = ((data[5] & 0xffff0000) >> 16) as u16;
212  //mte.get_trigger_sources();
213  let rbmask = (data[7] as u64) << 32 | data[6] as u64; 
214  mte.mtb_link_mask  = rbmask;
215  mte.dsi_j_mask     = data[8];
216  for k in 9..9 + n_hit_words {
217    let ltb_hits = data[k as usize];
218    // split them up
219    let first  =  (ltb_hits & 0x0000ffff) as u16;
220    let second = ((ltb_hits & 0xffff0000) >> 16) as u16;
221    mte.channel_mask.push(first);
222    // if this is the last hit word, only push 
223    // it in case n_ltb is odd
224    if k == ( 9 + n_hit_words) {
225      if !odd {
226         mte.channel_mask.push(second);  
227      }
228    } else {
229      mte.channel_mask.push(second);
230    }
231  }
232  Some(Ok(mte))
233}
234
235/// Gather monitoring data from the Mtb
236///
237/// ISSUES - some values are always 0
238pub fn get_mtbmonidata(bus : &mut IPBus) 
239  -> Result<MtbMoniData, MasterTriggerError> {
240  let mut moni = MtbMoniData::new();
241  let data = bus.read_multiple(0x120, 4, true)?;
242  if data.len() < 4 {
243    return Err(MasterTriggerError::BrokenPackage);
244  }
245  let tiu_busy_len    = TIU_BUSY_LENGTH.get(bus)?;
246  let tiu_aux_link    = (TIU_USE_AUX_LINK.get(bus)? != 0) as u8;
247  let tiu_emu_mode    = (TIU_EMULATION_MODE.get(bus)? != 0) as u8;
248  let aggr_tiu        = TIU_LT_AND_RB_MULT.get(bus)?;
249  let tiu_link_bad    = (aggr_tiu & 0x1) as u8;
250  let tiu_busy_stuck  = ((aggr_tiu & 0x2) >> 1) as u8;
251  let tiu_busy_ign    = ((aggr_tiu & 0x4) >> 2) as u8;
252  let mut tiu_status  = 0u8;
253  tiu_status          = tiu_status | (tiu_emu_mode);
254  tiu_status          = tiu_status | (tiu_aux_link << 1);
255  tiu_status          = tiu_status | ((tiu_link_bad as u8) << 2);
256  tiu_status          = tiu_status | (tiu_busy_stuck << 3);
257  tiu_status          = tiu_status | (tiu_busy_ign << 4);
258  let daq_queue_len   = EVQ_NUM_EVENTS.get(bus)? as u16;
259  moni.tiu_status     = tiu_status;
260  moni.tiu_busy_len   = tiu_busy_len;
261  moni.daq_queue_len  = daq_queue_len;
262  // sensors are 12 bit
263  let first_word     = 0x00000fff;
264  let second_word    = 0x0fff0000;
265  moni.temp          = ( data[2] & first_word  ) as u16;  
266  moni.vccint        = ((data[2] & second_word ) >> 16) as u16;  
267  moni.vccaux        = ( data[3] & first_word  ) as u16;  
268  moni.vccbram       = ((data[3] & second_word ) >> 16) as u16;  
269 
270  let rate           = bus.read_multiple(0x17, 2, true)?;
271  // FIXME - technically, the rate is 24bit, however, we just
272  // read out 16 here (if the rate is beyond ~65kHz, we don't need 
273  // to know with precision
274  let mask           = 0x0000ffff;
275  moni.rate          = (rate[0] & mask) as u16;
276  moni.lost_rate     = (rate[1] & mask) as u16;
277  let rb_lost_rate  = RB_LOST_TRIGGER_RATE.get(bus)?;
278  if rb_lost_rate > 255 {
279    moni.rb_lost_rate = 255;
280  } else {
281    moni.rb_lost_rate = rb_lost_rate as u8;
282  }
283  Ok(moni)
284}
285
286/// Configure the MTB according to lifot settings.
287/// If the settings have a non-zero prescale for 
288/// any of the triggers, this will cause the 
289/// MTB to start triggering 
290/// (if it hasn't been triggering before)
291///
292/// CHANGELOG - in previous versions, this reset the 
293///             MTB DAQ multiple times, this is not 
294///             ncessary and caused more issues than
295///             actually fixed something, so these got 
296///             removed.
297///
298/// # Arguments:
299///   * bus        : IPBus connected to the MTB (UDP)
300///   * settings   : configure the MTB according
301///                  to these settings 
302pub fn configure_mtb(bus : &mut IPBus,
303                     settings   : &MTBSettings) -> Result<(), MasterTriggerError> {
304  let trace_suppression = settings.trace_suppression;
305  match set_trace_suppression(bus, trace_suppression) {
306    Err(err) => error!("Unable to set trace suppression mode! {err}"),
307    Ok(_)    => {
308      if trace_suppression {
309        println!("==> Setting MTB to trace suppression mode!");
310      } else {
311        println!("==> Setting MTB to ALL_RB_READOUT mode!");
312        warn!("Reading out all events from all RBs! Data might be very large!");
313      }
314    }
315  }
316
317  let tiu_ignore_busy    = settings.tiu_ignore_busy;
318  match TIU_BUSY_IGNORE.set(bus, tiu_ignore_busy as u32) {
319    Err(err) => error!("Unable to change tiu busy ignore settint! {err}"),
320    Ok(_)    => {
321      if tiu_ignore_busy {
322        warn!("Ignoring TIU since tiu_busy_ignore is set in the config file!");
323        println!("==> Ignroing TIU since tiu_busy_ignore is set in the config file!");
324      }
325    }
326  }
327
328  info!("Settting rb integration window!");
329  let int_wind = settings.rb_int_window;
330  match set_rb_int_window(bus, int_wind) {
331    Err(err) => error!("Unable to set rb integration window! {err}"),
332    Ok(_)    => {
333      info!("rb integration window set to {}", int_wind); 
334    } 
335  }
336
337  match unset_all_triggers(bus) {
338    Err(err) => error!("Unable to undo previous trigger settings! {err}"),
339    Ok(_)    => ()
340  }
341  match settings.trigger_type {
342    TriggerType::Poisson => {
343      match set_poisson_trigger(bus,settings.poisson_trigger_rate) {
344        Err(err) => error!("Unable to set the POISSON trigger! {err}"),
345        Ok(_)    => ()
346      }
347    }
348    TriggerType::Any     => {
349      match set_any_trigger(bus,settings.trigger_prescale) {
350        Err(err) => error!("Unable to set the ANY trigger! {err}"),
351        Ok(_)    => ()
352      }
353    }
354    TriggerType::Track   => {
355      match set_track_trigger(bus, settings.trigger_prescale) {
356        Err(err) => error!("Unable to set the TRACK trigger! {err}"),
357        Ok(_)    => ()
358      }
359    }
360    TriggerType::TrackCentral   => {
361      match set_central_track_trigger(bus, settings.trigger_prescale) {
362        Err(err) => error!("Unable to set the CENTRAL TRACK trigger! {err}"),
363        Ok(_)    => ()
364      }
365    }
366    TriggerType::TrackUmbCentral  => {
367      match set_track_umb_central_trigger(bus, settings.trigger_prescale) {
368        Err(err) => error!("Unable to set the TRACK UMB CENTRAL trigger! {err}"),
369        Ok(_)   => ()
370      }
371    }
372    TriggerType::Gaps    => {
373      match set_gaps_trigger(bus, settings.gaps_trigger_use_beta) {
374        Err(err) => error!("Unable to set the GAPS trigger! {err}"),
375        Ok(_)    => ()
376      }
377    }
378    TriggerType::Gaps633    => {
379      match set_gaps633_trigger(bus, settings.gaps_trigger_use_beta) {
380        Err(err) => error!("Unable to set the GAPS trigger! {err}"),
381        Ok(_)    => ()
382      }
383    }
384    TriggerType::Gaps422    => {
385      match set_gaps422_trigger(bus, settings.gaps_trigger_use_beta) {
386        Err(err) => error!("Unable to set the GAPS trigger! {err}"),
387        Ok(_)    => ()
388      }
389    }
390    TriggerType::Gaps211    => {
391      match set_gaps211_trigger(bus, settings.gaps_trigger_use_beta) {
392        Err(err) => error!("Unable to set the GAPS trigger! {err}"),
393        Ok(_)    => ()
394      }
395    }
396    TriggerType::UmbCube => {
397      match set_umbcube_trigger(bus) {
398        Err(err) => error!("Unable to set UmbCube trigger! {err}"),
399        Ok(_)    => ()
400      }
401    }
402    TriggerType::UmbCubeZ => {
403      match set_umbcubez_trigger(bus) {
404        Err(err) => error!("Unable to set UmbCubeZ trigger! {err}"),
405        Ok(_)    => ()
406      }
407    }
408    TriggerType::UmbCorCube => {
409      match set_umbcorcube_trigger(bus) {
410        Err(err) => error!("Unable to set UmbCorCube trigger! {err}"),
411        Ok(_)    => ()
412      }
413    }
414    TriggerType::CorCubeSide => {
415      match set_corcubeside_trigger(bus) {
416        Err(err) => error!("Unable to set CorCubeSide trigger! {err}"),
417        Ok(_)    => ()
418      }
419    }
420    TriggerType::Umb3Cube => {
421      match set_umb3cube_trigger(bus) {
422        Err(err) => error!("Unable to set Umb3Cube trigger! {err}"), 
423        Ok(_)    => ()
424      }
425    }
426    TriggerType::Unknown => {
427      println!("== ==> Not setting any trigger condition. You can set it through pico_hal.py");
428      warn!("Trigger condition undefined! Not setting anything!");
429      error!("Trigger conditions unknown!");
430    }
431    _ => {
432      error!("Trigger type {} not covered!", settings.trigger_type);
433      println!("= => Not setting any trigger condition. You can set it through pico_hal.py");
434      warn!("Trigger condition undefined! Not setting anything!");
435      error!("Trigger conditions unknown!");
436    }
437  }
438    
439  // combo trigger - still named "global_trigger" in settings 
440  // mistakenly.
441  // FIXME
442  if settings.use_combo_trigger {
443    let global_prescale = settings.global_trigger_prescale;
444    let prescale_val    = (u32::MAX as f32 * global_prescale as f32).floor() as u32;
445    println!("=> Setting an additonal trigger - using combo mode. Using prescale of {}", prescale_val as f32 / u32::MAX as f32);
446    // FIXME - the "global" is wrong. We need to rename this at some point
447    match settings.global_trigger_type {
448      TriggerType::Any             => {
449        match ANY_TRIG_PRESCALE.set(bus, prescale_val) {
450          Ok(_)    => (),
451          Err(err) => error!("Settting the prescale {} for the any trigger failed! {err}", prescale_val) 
452        }
453      }
454      TriggerType::Track           => {
455        match TRACK_TRIG_PRESCALE.set(bus, prescale_val) {
456          Ok(_)    => (),
457          Err(err) => error!("Settting the prescale {} for the any trigger failed! {err}", prescale_val) 
458        }
459      }
460      TriggerType::TrackCentral    => {
461        match TRACK_CENTRAL_PRESCALE.set(bus, prescale_val) {
462          Ok(_)    => (),
463          Err(err) => error!("Settting the prescale {} for the track central trigger failed! {err}", prescale_val) 
464        }
465      }
466      TriggerType::TrackUmbCentral => {
467        match TRACK_UMB_CENTRAL_PRESCALE.set(bus, prescale_val) {
468          Ok(_)    => (),
469          Err(err) => error!("Settting the prescale {} for the track umb central trigger failed! {err}", prescale_val) 
470        }
471      }
472      _ => {
473        error!("Unable to set {} as a global trigger type!", settings.global_trigger_type);
474      }
475    }
476  }
477  Ok(())
478}
479
480/// Communications with the master trigger over Udp
481///
482/// The master trigger can send packets over the network.
483/// These packets contain timestamps as well as the 
484/// eventid and a hitmaks to identify which LTBs have
485/// participated in the trigger.
486/// The packet format is described
487/// [here](https://gitlab.com/ucla-gaps-tof/firmware/-/tree/develop/)
488///
489/// # Arguments
490///
491/// * mt_address        : Udp address of the MasterTriggerBoard
492///
493/// * mt_sender         : push retrieved MasterTriggerEvents to 
494///                       this channel
495/// * mtb_timeout_sec   : reconnect to mtb when we don't
496///                       see events in mtb_timeout seconds.
497///
498/// * verbose           : Print "heartbeat" output 
499///
500pub fn master_trigger(mt_address     : &str,
501                      mt_sender      : &Sender<MasterTriggerEvent>,
502                      moni_sender    : &Sender<TofPacket>, 
503                      thread_control : Arc<Mutex<ThreadControl>>,
504                      verbose        : bool) {
505  let mut bus            : IPBus;
506  let mut heartbeat      = MTBHeartbeat::new();
507  let mut mtb_timeout    = Instant::now();
508  let mut moni_interval  = Instant::now();
509  let mut tc_timer       = Instant::now();
510  
511  let mut settings       : MTBSettings;
512  let mut cali_active    : bool;
513  let mut holdoff        : bool;
514  loop {
515    match thread_control.lock() {
516      Ok(tc) => {
517        settings    = tc.liftof_settings.mtb_settings.clone();  
518        cali_active = tc.calibration_active; 
519        holdoff     = tc.holdoff_mtb_thread;
520      }
521      Err(err) => {
522        error!("Can't acquire lock for ThreadControl! Unable to set calibration mode! {err}");
523        return;
524      }
525    }
526
527    if holdoff || cali_active {
528      thread::sleep(Duration::from_secs(5));
529    } else {
530      if !holdoff {
531        println!("=> Docking clamp released!");
532      }
533      break;
534    }
535    if moni_interval.elapsed().as_secs() > settings.mtb_moni_interval {
536      match IPBus::new(mt_address) {
537        Err(err) => {
538          debug!("Can't connect to MTB, will try again in 10 ms! {err}");
539          continue;
540        }
541        Ok(mut moni_bus) => {
542          match get_mtbmonidata(&mut moni_bus) { 
543            Err(err) => {
544              error!("Can not get MtbMoniData! {err}");
545            },
546            Ok(moni) => {
547              let tp = moni.pack();
548              match moni_sender.send(tp) {
549                Err(err) => {
550                  error!("Can not send MtbMoniData over channel! {err}");
551                },
552                Ok(_) => ()
553              }
554            }
555          }
556        }
557      }
558      moni_interval = Instant::now();
559    }
560  } 
561  let mtb_timeout_sec    = settings.mtb_timeout_sec;
562  let mtb_moni_interval  = settings.mtb_moni_interval;
563  
564  // verbose, debugging
565  let mut last_event_id       = 0u32;
566  let mut first               = true;
567  let mut slack_cadence       = 5; // send only one slack message 
568                              // every 5 times we send moni data
569  let mut evq_num_events      = 0u64;
570  let mut n_iter_loop         = 0u64;
571  let mut hb_timer            = Instant::now();
572  let hb_interval             = Duration::from_secs(settings.hb_send_interval as u64);
573  let mut n_non_recoverable   = 0usize; // events which could not be recovered. We 
574                                        // can use this to reset the DAQ at a certain 
575                                        // point
576  let connection_timeout = Instant::now(); 
577  loop { 
578    match IPBus::new(mt_address) {
579      Err(err) => {
580        debug!("Can't connect to MTB, will try again in 10 ms! {err}");
581        thread::sleep(Duration::from_millis(10));
582      }
583      Ok(_bus) => {
584        bus = _bus;
585        break
586      }
587    }
588    if connection_timeout.elapsed().as_secs() > 10 {
589      error!("Unable to connect to MTB after 10 seconds!");
590      match thread_control.lock() {
591        Ok(mut tc) => {
592          tc.thread_master_trg_active = false;
593        }
594        Err(err) => {
595          error!("Can't acquire lock for ThreadControl! Unable to set calibration mode! {err}");
596        },
597      }
598      return;
599    }
600  }
601  
602  debug!("Resetting master trigger DAQ");
603  // We'll reset the pid as well
604  bus.pid = 0;
605  match bus.realign_packet_id() {
606    Err(err) => error!("Can not realign packet ID! {err}"),
607    Ok(_)    => ()
608  }
609  
610  match reset_daq(&mut bus) {//, &mt_address) {
611    Err(err) => error!("Can not reset DAQ! {err}"),
612    Ok(_)    => ()
613  }
614  
615  match EVENT_CNT_RESET.set(&mut bus, 1) {
616    Err(err) => error!("Unable to reset event counter! {err}"),
617    Ok(_)    => println!("=> Event counter reset!")
618  }
619
620  match configure_mtb(&mut bus, &settings) {
621    Err(err) => error!("Configuring the MTB failed! {err}"),
622    Ok(())   => ()
623  }
624 
625  let mut preload_cache = 1000; // preload the cache when we are starting 
626  loop {
627    // Check thread control and what to do
628    // Deactivate this for now
629    if tc_timer.elapsed().as_secs_f32() > 2.5 {
630      match thread_control.try_lock() {
631        Ok(mut tc) => {
632          if tc.stop_flag || tc.sigint_recvd {
633            tc.end_all_rb_threads = true;
634            break;
635          }
636        
637        },
638        Err(err) => {
639          error!("Can't acquire lock for ThreadControl! Unable to set calibration mode! {err}");
640        },
641      }
642      tc_timer = Instant::now();
643    }
644    // This is a recovery mechanism. In case we don't see an event
645    // for mtb_timeout_sec, we attempt to reconnect to the MTB
646    if mtb_timeout.elapsed().as_secs() > mtb_timeout_sec {
647      if mtb_timeout.elapsed().as_secs() > mtb_timeout_sec {
648        info!("reconnection timer elapsed");
649      } else {
650        info!("reconnection requested");
651      }
652      match IPBus::new(mt_address) {
653        Err(err) => {
654          error!("Can't connect to MTB! {err}");
655          continue; // try again
656        }
657        Ok(_bus) => {
658          bus = _bus;
659          //thread::sleep(Duration::from_micros(1000));
660          debug!("Resetting master trigger DAQ");
661          // We'll reset the pid as well
662          bus.pid = 0;
663          match bus.realign_packet_id() {
664            Err(err) => error!("Can not realign packet ID! {err}"),
665            Ok(_)    => ()
666          }
667          match reset_daq(&mut bus) {//, &mt_address) {
668            Err(err) => error!("Can not reset DAQ! {err}"),
669            Ok(_)    => ()
670          }
671        }
672      }
673      mtb_timeout    = Instant::now();
674    }
675    if moni_interval.elapsed().as_secs() > mtb_moni_interval || first {
676      if first {
677        first = false;
678      }
679      match get_mtbmonidata(&mut bus) { 
680        Err(err) => {
681          error!("Can not get MtbMoniData! {err}");
682        },
683        Ok(_moni) => {
684          if settings.tofbot_webhook != String::from("")  {
685            let url  = &settings.tofbot_webhook;
686            let message = format!("\u{1F916}\u{1F680}\u{1F388} [LIFTOF (Bot)]\n rate - {}[Hz]\n {}", _moni.rate, settings);
687            let clean_message = remove_from_word(message, "tofbot_webhook");
688            let data = json!({
689              "text" : clean_message
690            });
691            match serde_json::to_string(&data) {
692              Ok(data_string) => {
693                if slack_cadence == 0 {
694                  match ureq::post(url)
695                      .set("Content-Type", "application/json")
696                      .send_string(&data_string) {
697                    Err(err) => { 
698                      error!("Unable to send {} to TofBot! {err}", data_string);
699                    }
700                    Ok(response) => {
701                      match response.into_string() {
702                        Err(err) => {
703                          error!("Not able to read response! {err}");
704                        }
705                        Ok(body) => {
706                          if verbose {
707                            println!("[master_trigger] - TofBot responded with {}", body);
708                          }
709                        }
710                      }
711                    }
712                  }
713                } else {
714                  slack_cadence -= 1;
715                }
716                if slack_cadence == 0 {
717                  slack_cadence = 5;
718                }
719              }
720              Err(err) => {
721                error!("Can not convert .json to string! {err}");
722              }
723            }
724          }
725          let tp = TofPacket::from(&_moni);
726          match moni_sender.send(tp) {
727            Err(err) => {
728              error!("Can not send MtbMoniData over channel! {err}");
729            },
730            Ok(_) => ()
731          }
732        }
733      }
734      moni_interval = Instant::now();
735    }
736    
737    match get_event(&mut bus){ //,
738      None     => {
739      }
740      Some(Err(err)) => {
741        match err {
742          MasterTriggerError::PackageFooterIncorrect
743          | MasterTriggerError::PackageHeaderIncorrect 
744          | MasterTriggerError::DataTooShort
745          | MasterTriggerError::BrokenPackage => {
746            // in case we can't recover an event for x times, let's reset the DAQ
747            // not sure if 10 is a good number
748            if n_non_recoverable == 100 {
749              error!("We have seen {} non-recoverable events, let's reset the DAQ!", n_non_recoverable);
750              match reset_daq(&mut bus) {//, &mt_address) {
751                Err(err) => error!("Can not reset DAQ, error {err}"),
752                Ok(_)    => ()
753              }
754              n_non_recoverable = 0;
755            } 
756            n_non_recoverable += 1;
757          }
758          _ => ()
759        }
760      },
761      Some(Ok(_ev)) => {
762        if _ev.event_id == last_event_id {
763          error!("We got a duplicate event from the MTB!");
764          continue;
765        }
766        if _ev.event_id > last_event_id + 1 {
767          if last_event_id != 0 {
768            error!("We skipped {} events!", _ev.event_id - last_event_id); 
769            heartbeat.n_ev_missed += (_ev.event_id - last_event_id) as u64;
770          }
771        }
772        last_event_id = _ev.event_id;
773        heartbeat.n_events += 1;
774        match mt_sender.send(_ev) {
775          Err(err) => {
776            error!("Can not send MasterTriggerEvent over channel! {err}");
777            heartbeat.n_ev_unsent += 1;
778          },
779          Ok(_) => ()
780        }
781      }
782    }
783    
784    if preload_cache > 0 {
785      preload_cache -= 1;
786      continue;
787    }
788    if hb_timer.elapsed() >= hb_interval {
789      match EVQ_NUM_EVENTS.get(&mut bus) {
790        Err(err) => {
791          error!("Unable to query {}! {err}", EVQ_NUM_EVENTS);
792        }
793        Ok(num_ev) => {
794          evq_num_events += num_ev as u64;
795          heartbeat.evq_num_events_last = num_ev as u64;
796          n_iter_loop    += 1;
797          heartbeat.evq_num_events_avg = (evq_num_events as u64)/(n_iter_loop as u64);
798        }
799      }
800      
801      heartbeat.total_elapsed += hb_timer.elapsed().as_secs() as u64;
802      match TRIGGER_RATE.get(&mut bus) {
803        Ok(trate) => {
804          heartbeat.trate = trate as u64;
805        }
806        Err(err) => {
807          error!("Unable to query {}! {err}", TRIGGER_RATE);
808        }
809      }
810      match LOST_TRIGGER_RATE.get(&mut bus) {
811        Ok(lost_trate) => {
812          heartbeat.lost_trate = lost_trate as u64;
813        }
814        Err(err) => {
815          error!("Unable to query {}! {err}", LOST_TRIGGER_RATE);
816        }
817      }
818      match TRACK_TRIG_PRESCALE.get(&mut bus) {
819        Ok(ps) => {
820          heartbeat.prescale_track = (ps as f32) / (u32::MAX as f32) ;
821        }
822        Err(err) => {
823          error!("Unable to query {}! {err}", LOST_TRIGGER_RATE);
824        }
825      }
826      match GAPS_TRIG_PRESCALE.get(&mut bus) {
827        Ok(ps) => {
828          heartbeat.prescale_gaps = (ps as f32) / (u32::MAX as f32) ;
829        }
830        Err(err) => {
831          error!("Unable to query {}! {err}", LOST_TRIGGER_RATE);
832        }
833      }
834      heartbeat.version = ProtocolVersion::V1; 
835      if verbose {
836        println!("{}", heartbeat);
837      }
838
839      let pack = heartbeat.pack();
840      match moni_sender.send(pack) {
841        Err(err) => {
842          error!("Can not send MTB Heartbeat over channel! {err}");
843        },
844        Ok(_) => ()
845      }
846      hb_timer = Instant::now();
847    } 
848  }
849}
liftof_lib/master_trigger.rs

liftof_lib/
master_trigger.rs
