Sender.h

Go to the documentation of this file.

// Copyright (C) 2022 European Spallation Source, ERIC. See LICENSE file
//===----------------------------------------------------------------------===//
//===----------------------------------------------------------------------===//
#pragma once
 
#include "cluon-complete.hpp"
 
#include <atomic>
#include <cstdint>
#include <string>
#include <utility>
#include <optional>
#include <random>
 
#include "Structs.h"
#include "Readout.h"
#include "SenderConfigs.h"
#include "enums.h"
#include "hdf_interface.h"
#include "version.hpp"
#include "efu_time.h"
 
class RL_API Sender {
public:
  Sender(
      std::string IpAddress, const int UDPPort, const int TCPPort, const DetectorType detector_type, const ReadoutType readout_type, const pulse_grid grid
  ): detector_type(detector_type),
     readout_type(readout_type),
     ipaddr(std::move(IpAddress)),
     port(UDPPort),
     tcp_port(TCPPort),
     grid(grid),
     period(grid.period()),
     sender{ipaddr, static_cast<uint16_t>(UDPPort)}
  {
    // the latest pulse already past, as for a source that has been running all along
    set_pulse(grid.at_or_before(efu_time::now_nanoseconds()));
    newPacket();
  }
 
  Sender(
      std::string IpAddress, const int UDPPort, const int TCPPort, const DetectorType detector_type, const ReadoutType readout_type, const efu_time p = efu_time(1)
  ): Sender(std::move(IpAddress), UDPPort, TCPPort, detector_type, readout_type, pulse_grid::from_period(p)) {}
 
  explicit Sender(const SenderConfig & config, const pulse_grid grid)
      : Sender(config.ip_address, config.udp_port, config.tcp_port, config.detector_type, config.readout_type, grid) {}
 
  explicit Sender(const SenderConfig & config, const efu_time p = efu_time(1))
      : Sender(config, pulse_grid::from_period(p)) {}
 
  ~Sender() {
    // ensure any buffered readouts are sent before the object is destroyed
    if (DataSize > static_cast<int>(sizeof(struct PacketHeaderV0))) {
      send();
    }
    report_long_tof();
  }
 
  void addReadout(uint8_t Ring, uint8_t FEN, double tof, double weight, const void * data);
  void addReadout(uint8_t Ring, uint8_t FEN, efu_time t, const void * data);
  // Specializations for handled data types
  void addReadout(uint8_t Ring, uint8_t FEN, efu_time t, const CAEN_readout_t * data);
  void addReadout(uint8_t Ring, uint8_t FEN, efu_time t, const CDT_readout_t * data);
  void addReadout(uint8_t Ring, uint8_t FEN, efu_time t, const VMM3_readout_t * data);
  void addReadout(uint8_t Ring, uint8_t FEN, efu_time t, const BM0_readout_t * data);
  void addReadout(uint8_t Ring, uint8_t FEN, efu_time t, const BM2_readout_t * data);
  void addReadout(uint8_t Ring, uint8_t FEN, efu_time t, const BMI_readout_t * data);
 
 
  int send();
 
  void setPulseTime(uint32_t PHI, uint32_t PLO, uint32_t PPHI, uint32_t PPLO);
 
  void begin_pulse();
  void begin_pulse_at(uint64_t at);
  [[nodiscard]] uint64_t next_pulse() const;
  void announce_pulse();
  void flush();
  void hold_pulse(const bool hold) { hold_pulse_ = hold; }
 
  void fold_tof(const bool fold) { fold_tof_ = fold; }
  [[nodiscard]] uint64_t long_tof_count() const { return long_tof_; }
 
  // Query the current pulse and previous pulse times
  [[nodiscard]] std::pair<uint32_t, uint32_t> lastPulseTime() const;
  [[nodiscard]] uint64_t lastPulseNanoseconds() const { return pulse_ns; }
  [[nodiscard]] std::pair<uint32_t, uint32_t> prevPulseTime() const;
  [[nodiscard]] std::pair<uint32_t, uint32_t> lastEventTime() const;
 
  void newPacket();
 
  int command_shutdown() const;
 
  int verbose(const Verbosity v){
    switch (v) {
      case Verbosity::details: verbosity=3; break;
      case Verbosity::info: verbosity=2; break;
      case Verbosity::warnings: verbosity=1; break;
      case Verbosity::errors: verbosity=0; break;
      case Verbosity::silent: verbosity=-1; break;
      default: verbosity=0;
    }
    return verbosity;
  }
  int verbose(const int v){verbosity = v; return verbosity;}
private:
 
  void check_size_and_send();
 
  void maybe_advance_pulse();
 
  void set_pulse(uint64_t ns);
 
  // Packet header
  uint32_t phi{0}; // pulse and prev pulse high and low
  uint32_t plo{0};
  uint32_t pphi{0};
  uint32_t pplo{0};
 
  uint32_t lasthi{0};
  uint32_t lastlo{0};
 
  int SeqNum{0};
  int OutputQueue{0};
  DetectorType detector_type;
  ReadoutType readout_type;
 
  // TX Buffer
  // PacketHeaderV0 *hp{};
  char buffer[9000]{};
  const int MaxDataSize{8950};
  int DataSize{0};
  // IP and port number
  std::string ipaddr;
  int port{9000};
  int tcp_port{8888};
  int verbosity{0};
  bool fold_tof_{false};
  bool hold_pulse_{false};
  std::atomic<uint64_t> long_tof_{0};
  void report_long_tof() const;
 
  pulse_grid grid;
  efu_time period, time;
  uint64_t pulse_ns{0};
  cluon::UDPSender sender;
 
  std::mutex time_mutex, send_mutex;
};