zynq_readout_cpp/client.cc

#include <iostream>
#include <sys/socket.h>
#include <stdlib.h>
#include <netinet/in.h>
#include <string.h>
#include <arpa/inet.h>
#include <unistd.h>
#include <errno.h>
#include <functional>
#include <fstream>
#include <chrono>
#include <thread>
#include <ctime>
#include <iomanip>
#include <csignal>
#include "channel.h"
#include "spdlog/spdlog.h"

const char * HOST = "192.168.30.89";
const uint32_t PORT = 9999;
const size_t MAX_EVENT_SIZE = 33484; // in bytes
const size_t MIN_EVENT_SIZE = 204; // in bytes
const size_t MAX_CHANNEL_SIZE = 4184;
const size_t MIN_CHANNEL_SIZE = 24;
const uint8_t EVENT_MARKER = 0xA5;
const uint8_t CHANNEL_MARKER = 0x5A;

int sock = 0;
bool running = true;
bool bad_data = true;

int open_socket();
void signalHandler( int signum );
void setup_logger();
size_t process_channel(char const * ptr, size_t len);
size_t process_block(char const * ptr, size_t len);
size_t process_block(char const * ptr, size_t len){
   return len;
}
size_t process_channel(char const * ptr, size_t len){
   if ((len < MIN_CHANNEL_SIZE) || (len > MAX_CHANNEL_SIZE)) {
      spdlog::debug("channel length must be between {} and {}.",
            MIN_CHANNEL_SIZE, MAX_CHANNEL_SIZE);
      return len;
   }

   size_t bytes_left = len;
   bool valid_channel = false;
   uint32_t * wordData = (uint32_t *)ptr;

   while ((bytes_left > 0) && !valid_channel) {
      if ((ptr[0] & 0xFF) != CHANNEL_MARKER){
         spdlog::debug("bad_data, expecting {0:x}, received {0:x}",
               static_cast<uint32_t>(CHANNEL_MARKER), static_cast<uint32_t>(ptr[0]));
         bytes_left -= 4;
      } else {
         valid_channel = true;

         size_t channel_size = ((wordData[0] & 0xFFFFFF00) >> 8) * sizeof(uint32_t);
         if ((channel_size > MAX_CHANNEL_SIZE) || (channel_size < MIN_CHANNEL_SIZE)) {
            spdlog::debug("Invalid channel size: {}", channel_size);
         }
         uint32_t version = wordData[0] && 0xFF;
         spdlog::debug("channel size {}, version {}", channel_size, version);

         int32_t integral[3];
         for (int i = 0; i < 3; i++) 
            integral[i] = (wordData[i + 1] & 0xFFFFFF00)>>8;
         spdlog::debug("integrals: {}, {}, {}", integral[0], integral[1], integral[2]);

         uint32_t crossing_count = wordData[4] & 0x00FFFFFF;
         uint32_t block_count = (wordData[5] &0xFF000000) >> 24;
         uint32_t channel = (wordData[5] & 0x00FF0000) >> 16;
         uint32_t spill_id = (wordData[5] & 0x0000FFFF);
         spdlog::debug("crossing_count {}, block_count {}, channel {}, spill_id {}",
               crossing_count, block_count, channel, spill_id);

         for (uint32_t i = 0; i < block_count; i++) {
            process_block(ptr, channel_size); // not correct yet!
         }

         // the number of bytes consumed by this function, regardless of block
         // processing
         bytes_left -= channel_size;

         Channel c(ptr, channel_size);
         c.Print();
      }
   }

   // return number of bytes cosumed of the channel
   spdlog::debug("processed {} bytes out of {}", (len - bytes_left), len);
   return len - bytes_left;
}


int main(int argc, char * argv[])
{
   setup_logger();
   signal(SIGINT, signalHandler);
   int ret = open_socket();
   if (ret != 0)
      return -1;

   char char_buffer[MAX_EVENT_SIZE];
   uint32_t * wordData = (uint32_t*) char_buffer;
   ssize_t nread;

   fd_set select_fds;                // fd's used by select
   struct timeval timeout;           // Time value for time out
   //Timeout set for 0 sec + 200 millisec
   timeout.tv_sec = 0;         
   timeout.tv_usec = 200*1000;
   std::chrono::duration<int, std::milli> polling_period(2);

   while (running) {
      // Setup the descriptor set for select()
      FD_ZERO(&select_fds);            
      FD_SET(sock, &select_fds);           
      if ( select(sock + 1, &select_fds, NULL, NULL, &timeout) == 0 ){
         std::this_thread::sleep_for(polling_period);
      }
      else{
         // read in first 4 bytes
         nread = recv(sock, char_buffer, 4, 0);
         spdlog::info("read in {:d} bytes: 0x{:X}", nread, wordData[0]);
         // if not A5 marker, go to next iteration (read in 4 more bytes)
         if ((char_buffer[0] & 0xFF) != 0xA5){
            bad_data = true;
            continue;
         }
         else {
            bad_data = false;
            size_t frameSize = 4 * (wordData[0] >> 8);
            nread = recv(sock, char_buffer, frameSize - 4, 0);
            // nread + 4 must be between MIN_EVENT_SIZE and MAX_EVENT_SIZE, and equal to frameSize.
            if (((size_t)nread != frameSize - 4) || (frameSize < MIN_EVENT_SIZE)|| (frameSize > MAX_EVENT_SIZE)){
               bad_data = true;
               continue;
            }
            else {
               // ok, this seems to be good data
               spdlog::info("next {:d} bytes (expecting {:d}): {:X} {:X} {:X} {:X}",
                     nread, frameSize - 4, wordData[0], wordData[1], wordData[2], wordData[3]);
               spdlog::info("{:X} {:X} {:X} {:X}", wordData[4], wordData[5], wordData[6], wordData[7]);
               uint8_t board_id = (wordData[0] >> 24) & 0xFF;
               uint8_t version = wordData[0] & 0xFF;
               uint16_t spill_id = (wordData[1] & 0x00FFFF00) >> 8;
               uint8_t channel_count = (wordData[1]) & 0x000000FF;
               spdlog::info("board {}, version {}, spill {}, channel_count {}",
                     static_cast<uint32_t>(board_id),
                     static_cast<uint32_t>(version),
                     static_cast<uint32_t>(spill_id),
                     static_cast<uint32_t>(channel_count));

               if ((board_id > 4) || (board_id < 1) || (version != 1) || (channel_count > 8) || (channel_count < 1)) {
                  bad_data = true;
                  continue;
               }
               else {
                  size_t bytes_left = (frameSize - 3 * sizeof(uint32_t));
                  size_t bytes_processed = 0;
                  while ((bytes_left - bytes_processed >= MIN_CHANNEL_SIZE) && (bytes_processed < bytes_left)) {
                     bytes_processed += process_channel(char_buffer + 2 * sizeof(uint32_t) + bytes_processed,
                           bytes_left - bytes_processed);
                  }
                  spdlog::info("Done, {} bytes processed.", bytes_processed);
               }

            }
         }
      }
   }

   close(sock);
   return 0;
}

void signalHandler( int signum ) {
   spdlog::info("Interrupt signal ({}) received, exiting...", signum);
   running = false;
}

int open_socket(){
   struct sockaddr_in serv_addr;
   if ((sock = socket(AF_INET, SOCK_STREAM, 0)) < 0) {
      std::cerr << "Cannot create socket" << std::endl;
      return -1;
   }

   memset(&serv_addr, '0', sizeof(serv_addr));
   serv_addr.sin_family = AF_INET;
   serv_addr.sin_port = htons(PORT);

   if(inet_pton(AF_INET, HOST, &serv_addr.sin_addr)<=0) {
      std::cerr << "Invalid address or Address not supported" << std::endl;
      return -1;
   }

   if (connect(sock, (struct sockaddr *)&serv_addr, sizeof(serv_addr)) < 0) {
      std::cerr << "Cannot connect to " << HOST << ":" << PORT << std::endl;
      return -1;
   }

   return 0;
}

void setup_logger(){
}