ifbtorr.cc

Go to the documentation of this file.

//
// \brief
//    This program is used as a replacement to the XML-RPC for data logging.
//    It is used to fetch the data from the intensity frontier database and
//    archive them into Minos mbeam.root files that can be unpacked later
//    using usual tools we normally use to read (analyze) the data.
//
//    The script evokes the intensity frontier data handling client put into
//    place by Marc Mengel from the computing division.
//
//    Originally written by: Joseph M. Kiveni <mkiveni@fnal.gov>
//    Modified by: Robert Hatcher <rhatcher@fnal.gov>
//
#include <fstream>
#include <sstream>
#include <iostream>
#include <iomanip>
#include <stdarg.h>
#include <stdlib.h>
#include <stdio.h>
#include <string.h>
#include <errno.h>
#include <math.h>

#include <unistd.h>
#include <getopt.h>
#include <time.h>
#include <sys/time.h>
typedef struct timespec timespec_t;
typedef struct tm       tm_t;
#include <ctime>

// ifbeam related (new beam database interface)
#include "WebAPI.h"
#include "ifbeam.h"

#include "BeamData/python/bdp/BeamData.h"
#include "BeamData/python/bdp/BeamData.cc"
#include "OnlineUtil/rototalk.h"

#include <algorithm>  // for sort
#include <map>

const char* exename = "ifbtorr";

//
// \class ifbConfig
// \brief a class for managing the configuration of ifbtorr
//
class ifbConfig {
public:
  ifbConfig();
  ~ifbConfig() { ; }
  void ParseArgv(int argc, char** argv);
  void Usage(std::string msg, int exitval);
  void PrintConfig();
public:
  int    Verbose;          // how noisy should this program be
  bool   PrintAndQuit;     // print out config and quit
  bool   CountAndQuit;     // count # of spills w/ no rotorooter connection
  bool   AllowRotoErrors;  // continue after rotorooter connection/file errors
  double TimeStamp;        // starting times (seconds since EPOCH)
  double WindowSize;       // window in seconds
  std::string BundleName;  // ifbeam DB bundle name
  std::string RotoNode;    // node where rootrooter will run
  std::string URLifbeam;   // URL for ibeam db connection
  double CacheWindow;      // BeamFolder time_width (-1 = use BF default 1200)
  double Epsilon;          // BeamFolder epsilon
  double DeltaCacheTime;   // new BeamFolder cache starts @ EndTime + Delta
  int    TClkTriggerEvent; // tclk_event to store w/ record
  int    TClkTriggerDelay; // tclk_delay to store w/ record
  int    DtTolerance;      // only complain about Dt larger than this
  std::string PrintOpts;   // print options
};

//
// \class ifbProcBase
// \brief an base class for processing ifbeam times
//        simply counts # of spills, and keeps track of devices
//
class ifbProcBase {
public:
  ifbProcBase(ifbConfig cfgIn);
  virtual ~ifbProcBase() { ; }

  // over-ride these
  virtual int Init();
  virtual int SpillInit(double when);
  virtual int SpillDevice(double when, std::string device);
  virtual int SpillFinish(double when);
  virtual int Finish();

  // these should be universal
  virtual int NewCache(double cacheStart);
  virtual int ScanWindow(double t1, double t2);
  virtual int ProcessTime(double when);
  virtual int Run();

  size_t SpillCount() { return nspills; }

protected:
  BeamFolder*                  bf;       // interface to ifbeam DB
  ifbConfig                    cfg;      // configuration info
  size_t                       nspills;  // number of spills processed

  // the following aren't for the whole window, but only the current cache
  std::vector<double>          cacheTimes;   // times list for the cache
  std::vector<std::string>     cacheDevices; // device list for the cache
};

//
// \class ifbProcRoto
// \brief process ifbeam times, sending info to rotorooter
//
class ifbProcRoto : public ifbProcBase {
public:
  ifbProcRoto(ifbConfig cfgIn) : ifbProcBase(cfgIn), maxDt(-99999999), minDt(999999) { ; }
  virtual ~ifbProcRoto() { ; }

  virtual int Init();
  virtual int SpillInit(double when);
  virtual int SpillDevice(double when, std::string device);
  virtual int SpillFinish(double when);
  virtual int Finish();

  std::map<std::string,size_t>& DeviceCount() { return devcnt; }

private:
  void CompareSpillDeviceTimes(const std::string& devname, 
                               int spillSec, int spillMSec,
                               int devSec, int devMSec);

  BeamData bd;  // Brett's class for packing data, sending to rotorooter

  std::map<std::string,size_t> devcnt;   // # times device was in a spill

  int maxDt;  // max time (ms) between spill and device data
  int minDt;  // min time (ms) between spill and device data
};

//
// forward declare helper functions
//
double                   ParseTimeString(const char*);
std::string              UTCTimeString(double);
std::string              LocalTimeString(double);
std::vector<std::string> UniqueNames(std::vector<std::string>, int verbose);

inline void DoubleToSecMSec(double t, int& sec, int& msec) 
{
  // add 0.5 before truncation on msec so that we round properly
  sec = (int)t;  msec = int((t - int(sec))*1.0e3+0.5);
}

int main(int argc, char** argv)
{
  exename = argv[0];

  //
  // Get configuration options
  //
  ifbConfig cfg;
  cfg.ParseArgv(argc,argv);
  if ( cfg.PrintAndQuit || cfg.Verbose > 0  ) cfg.PrintConfig();
  if ( cfg.PrintAndQuit ) exit(0);

  ifbProcBase* ifbp = 0;
  if ( cfg.CountAndQuit ) ifbp = new ifbProcBase(cfg);
  else                    ifbp = new ifbProcRoto(cfg);

  ifbp->Run();

} // end of main() program


ifbConfig::ifbConfig()
  : Verbose(0)               // quiet
  , PrintAndQuit(false)      // quit after parsing argv and print config
  , CountAndQuit(false)      // count # of spills w/o rotorooter connection
  , AllowRotoErrors(false)   // continue after rotorooter connection/file errs
  , TimeStamp(0)             // user must supply starting time
  , WindowSize(8*60*60)      // default window is 8 hours
  , BundleName("NuMI_all")   // default bundle
  , RotoNode("minos-nearline.fnal.gov") // default where rotorooter is running
  // URL for ifbeam db connection, default "" allows code to choose own default
  // might be something like "http://dbweb3.fnal.gov:8088/ifbeam" to override
  , URLifbeam("")
  , CacheWindow(-1)          // use BeamFolder default by default
  , Epsilon(0.05)            // BeamFolder matching epsilon?
  , DeltaCacheTime(0.0001)   // start new cache this long after previous
  , TClkTriggerEvent(169)
  , TClkTriggerDelay(500)
  , DtTolerance(500)         // allowed time between spill and device times
                             // without complaint
  , PrintOpts("")
{
  // example timestamp
  //   TimeStamp = 1373660869;  // Fri Jul 12 15:27:49 CDT 2013
  //                            // 2013-07-12 20:27:49 UTC
  //                            // 2013-07-12 15:27:49 local
}
void ifbConfig::PrintConfig() {

  long long tsec = (int)TimeStamp;
  long long tns  = (int)((TimeStamp - tsec)*1000000000);

  std::cout << exename << ": current configuration:" << std::endl
            << "  Verbose     " << Verbose 
            << "   " << ( (PrintAndQuit) ? "PrintAndQuit" : "" )
            << "   " << ( (CountAndQuit) ? "CountAndQuit" : "" )
            << "   " << ( (AllowRotoErrors) ? "AllowRotoErrors" : "" )
            << std::endl
            << "  TimeStamp   " << TimeStamp
            << " (" << tsec << "."
            << std::setfill('0') << std::setw(9) << tns
            << std::setfill(' ') << ")" << std::endl
            << "     " << UTCTimeString(TimeStamp) << std::endl
            << "     " << LocalTimeString(TimeStamp) << std::endl
            << "  WindowSize  " << WindowSize << " (seconds)" << std::endl
            << "  BundleName  \"" << BundleName << "\"" << std::endl
            << "  RotoNode    \"" << RotoNode << "\"" << std::endl
            << "  URLifbeam   \"" << URLifbeam << "\"" << std::endl
            << "  BeamFolder CacheWindow " << CacheWindow 
            << "  Epsilon " << Epsilon
            << "  DeltaCacheTime " << DeltaCacheTime << std::endl
            << "  TClkTrigger Event " << TClkTriggerEvent
            << "  Delay " << TClkTriggerDelay << std::endl
            << "  DtTolerance " << DtTolerance << std::endl
            << "  PrintOpts \"" << PrintOpts << "\"" << std::endl
            << std::endl;
}
void ifbConfig::ParseArgv(int argc, char** argv) {

  /*
   * Define the options
   *
   */

  static struct option long_opts[] = {
    { "help",         no_argument,       0, 'h' },
    { "timestamp",    required_argument, 0, 't' },
    { "timestring",   required_argument, 0, 'T' },
    { "window",       required_argument, 0, 'w' },
    { "bundle",       required_argument, 0, 'b' },
    { "rotonode",     required_argument, 0, 'r' },
    { "url",          required_argument, 0, 'u' },
    { "print-config", no_argument,       0, 'C' },
    { "count-spills", no_argument,       0, 'c' },
    { "verbose",      no_argument,       0, 'v' },
    { "bypass-roto",  no_argument,       0, 'A' },
    { "cache-window", required_argument, 0, 'z' },
    { "cache-delta",  required_argument, 0, 'd' },
    { "epsilon",      required_argument, 0, 'e' },
    { "tclk-event",   required_argument, 0, 'E' },
    { "tclk-delay",   required_argument, 0, 'D' },
    { "dt-tolerance", required_argument, 0, 'q' },
    { "print-opts",   required_argument, 0, 'P' },
    { 0, 0, 0, 0 } // signal end of list
  };
  const char short_opts[] = "ht:T:w:W:b:r:u:CcvAz:d:e:E:D:q:P:";

  /*
   * Parse the options
   *
   */

  // getopt_long stored the option index here
#ifndef MACOSX
  optind = 0; // getopt.h: Reset getopt to start of arguments
#else
  optind = 1; // skip 0th argument (executable) for MACOSX
#endif
  int opt_index = 0;
  int c;
  while ((c = getopt_long(argc,argv,short_opts,long_opts,&opt_index)) != -1 ) {
    if (Verbose>1) {
      std::cout << " c from getopt_long " << c << " '" << (char)c << "'"
                << " optarg \"" << (optarg?optarg:"") << "\""
                << std::endl;
    }
    switch ( c ) {
    case 0: {
      // if option set a flag, do nothing else
      if ( long_opts[opt_index].flag != 0 ) {
        if (Verbose>1) {
          std::cout << " c==0  opt_index " << opt_index
                    << " flag " << long_opts[opt_index].flag << std::endl;
        }
        break;
      }
      std::string opt_name = long_opts[opt_index].name;
      if (Verbose>1) {
        std::cout << "c==0  opt_index " << opt_index
                  << " opt_name=\"" << opt_name
                  << "\" optarg=\"" << optarg << "\"" << std::endl;
      }
      break;
    }
    case 'h': {
      Usage("",0);
      break;
    }
    case 'C': {
      PrintAndQuit = true;
      break;
    }
    case 'c': {
      CountAndQuit = true;
      break;
    }
    case 'v': {
      ++Verbose;
      break;
    }
    case 'A': {
      AllowRotoErrors = true;
      break;
    }
    case 't': {
      TimeStamp = atof(optarg);
      break;
    }
    case 'T': {
      TimeStamp = ParseTimeString(optarg);
      break;
    }
    case 'w': {
      float wval;
      char units[128] = { 0 };
      //int nitems = 
        sscanf(optarg,"%f%s",&wval,units);
      switch ( units[0] ) {
      case 's':
        WindowSize = wval;
        break;
      case 'm':
        WindowSize = wval*60;
        break;
      case 'h':
        WindowSize = wval*60*60;
        break;
      case '\0': 
        // cout << "RWH no units, " << nitems << " nitems" << endl;
        WindowSize = wval*60*60;
        break;
      default: 
        std::ostringstream oss;
        oss << "ERROR, --window set with unknown units"
            << " \"" << units << "\"";
        Usage(oss.str(),1);
        break;
      } // done unit handling
      break;
    }
    case 'b': {
      BundleName = optarg;
      break;
    }
    case 'r': {
      RotoNode = optarg;
      break;
    }
    case 'u': {
      URLifbeam = optarg;
      break;
    }
    case 'd': {
      DeltaCacheTime = atof(optarg);
      break;
    }
    case 'z': {
      CacheWindow = atof(optarg);
      break;
    }
    case 'e': {
      Epsilon = atof(optarg);
      break;
    }
    case 'E': {
      TClkTriggerEvent = atoi(optarg);
      break;
    }
    case 'D': {
      TClkTriggerDelay = atoi(optarg);
      break;
    }
    case 'q': {
      DtTolerance = atoi(optarg);
      break;
    }
    case 'P': {
      PrintOpts = optarg;
      break;
    }
    default: {
      std::ostringstream oss;
      oss << "ERROR, unhandled/unrecognized option '" << (char)optopt << "' ("
          << optopt << ")";
      Usage(oss.str(),1);
    }

    } // end of switch(c)
  } // end of getopt_long

  if ( CountAndQuit ) {
    if ( PrintOpts != "" ) PrintOpts += ",";
    PrintOpts += "spillcnt";
  }

  // handle non-flag (positional) arguments
  // do this in a manner to support old positional arguments
  //   expect: [ timestamp [ bundle [ window ]]]
  // later args are optional
  if ( optind < argc ) {
    if (Verbose>1) {
      std::cout << "non-option (positional) ARGV-elements: " << std::endl;
    }
    int posindx = -1;
    while ( optind < argc ) {
      ++posindx;
      std::string arg = argv[optind++];

      if (Verbose>1) {
        std::string posname[] = { "TimeStamp", "WindowSize",
                                  "BundleName", "UNKNOWN" };
        std::string name = (posindx<3) ? posname[posindx] : "UNKNOWN";
        std::cout << "Set " << setw(10) << name
                  << " from positional arg[" << posindx
                  << "] \"" << arg << "\"" << std::endl;
      }

      switch ( posindx ) {
      case 0: {
        // convert time as float (in case user gave fractional time)
        TimeStamp = atof(arg.c_str());
        break;
      }
      case 1: {
        // original code assumed this would be in hours
        WindowSize = atof(arg.c_str())*60*60;
        break;
      }
      case 2: {
        BundleName = arg;
        break;
      }
      default: {
        Usage("ERROR, unexpected extra positional argument",1);
        break;
      }
      } // end of switch( posindx )
    } // end of while
  } // end of "if ( optind < argc )"

}
void ifbConfig::Usage(std::string msg, int exitval) {

  std::cout 
    << std::endl
    << "Usage:  " << exename << " [OPTIONS] [ timestamp [ window(hours) [ bundle ]]]\n"
    << "  Read data from IFBeam DB, reformat and send to MINOS beam rotorooter.\n"
    << "Mandatory arguments to long options are mandatory for short options.\n"
    << "   -h, --help              display this help and exit\n"
    << "   -t, --timestamp=TIME    starting time in seconds since epoch\n"
    << "   -T, --timestring='YYYY-MM-DD hh:mm:ss'\n"
    << "   -w, --window=VAL        window time (default units hours)\n"
    << "                           use VALs or VALm for sec, min; e.g. 3600s\n"
    << "   -b, --bundle=NAME       bundle name\n"
    << "   -r, --rotonode=MACHINE  node where rotorooter is running\n"
    << "   -u, --url=IFB_URL       URL for IF Beam DB connection\n"
    << "   -C, --print-config      print configuration and exit\n"
    << "   -v, --verbose           increase verbosity\n"
    << "   -c, --count-spills      simply count spills w/ no rotorooter\n"
    << "   -z, --cache-window      BeamFolder cache window size (-1=default[1200])\n"
    << "   -d, --cache-delta       start new BF cache Delta after previous\n"
    << "   -e, --epsilon           BeamFolder epsilon\n"
    << "   -E, --tclk-event=EVENT  'event' (A9?) associated w/ block [169]\n"
    << "   -D, --tclk-delay=DELAY  'delay' associated w/ block [500]\n"
    << "   -q, --dt-tolerance=MS   complain about spill/device time diff\n"
    << "   -P, --print-opts=POPTS  what extra to print\n"
    << "       --bypass-roto       continue after rotorooter connection/file errors\n"


    << "Use of positional arguments is a deprecated feature, use flags instead.\n"
    << "PrintOpts POPTS is a comma separated string of things to print\n"
    << "     times    = all timestamps\n"
    << "     devices  = final list of devices w/ spill counts\n"
    << "     missing  = device only if it was missing from some spills\n"
    << "     dt       = excursion extremes between spill and device time (ms)\n"
    << "     spillcnt = # of spills\n"
    << std::endl;
  PrintConfig();
  if ( msg != "" ) std::cout << exename << ": " << msg << std::endl;
  if ( exitval >= 0 ) exit(exitval);
}


std::vector<std::string> UniqueNames(std::vector<std::string> names, 
                                     int verbose)
{
  //
  // Weed out duplicate names (also print out duplicates)
  //
  std::map<std::string,int> namecnt;
  for ( size_t iname = 0; iname < names.size(); ++iname ) {
    namecnt[names[iname]]++;
  }
  if ( names.size() != namecnt.size() ) {
    if ( verbose > 3 ) {
      std::cout << exename << ": bfp.GetDeviceList() returned " 
                << names.size() << " devices, " 
                << namecnt.size() << " unique"
                << std::endl;
    }
    // clear old list
    names.clear();
    // build new list from sorted map
    std::map<std::string,int>::const_iterator ncitr = namecnt.begin();
    for ( ; ncitr != namecnt.end(); ++ncitr ) {
      names.push_back(ncitr->first);
      int cnt = ncitr->second;
      if ( cnt > 1 && verbose > 4 ) {
        std::cout << "  " << std::left << std::setw(12) 
                  << ncitr->first << std::right
                  << " was in device list " << cnt 
                  << " times." << std::endl;
      }
    }
  }
  return names;
}
ifbProcBase::ifbProcBase(ifbConfig cfgIn) 
  : bf(0), cfg(cfgIn), nspills(0) 
{
  //
  // Create the BeamFolder w/ default [1200 sec] window
  // this will be less than the total window we want to scan (in general)
  // but we'll just push the BeamFolder cache window along
  //
  if ( cfg.CacheWindow > 1.0 ) {
    bf = new BeamFolder(cfg.BundleName, cfg.URLifbeam, cfg.CacheWindow); 
  } else {
    bf = new BeamFolder(cfg.BundleName, cfg.URLifbeam); 
  }
  bf->set_epsilon(cfg.Epsilon);  // matching criteria (?)
}

int ifbProcBase::Init() 
{ 
  return 0;
}
int ifbProcBase::SpillInit(double)
{
  return 0;
}
int ifbProcBase::SpillDevice(double,std::string)
{
  return 0;
}
int ifbProcBase::SpillFinish(double)
{
  return 0;
}
int ifbProcBase::Finish()
{
  return 0;
}
int ifbProcBase::NewCache(double cacheStart)
{
  cacheTimes.clear();
  cacheDevices.clear();

  try {
    bf->FillCache(cacheStart);
    cacheTimes = bf->GetTimeList();
    std::sort(cacheTimes.begin(),cacheTimes.end()); // ensure time ordered
    cacheDevices = UniqueNames( bf->GetDeviceList(), cfg.Verbose );
  } catch(WebAPIException we) {
    if ( cfg.Verbose > 0 ) {
      std::cout << exename << ": no spills in BeamFolder cache " 
                << std::fixed << std::setprecision(3) << cacheStart;
      if ( cfg.CacheWindow > 1 ) {
        std::cout << " (+ " << cfg.CacheWindow << ")";
      } else {
        std::cout << " (+ default BeamFolder window " << 1200 << "?)";
      }
      std::cout << std::endl;
    }
  }
  return 0;
}
int ifbProcBase::ScanWindow(double t1, double t2)
{
  bool printTime = ( cfg.PrintOpts.find("times") != std::string::npos );

  double tCache = t1;
  while ( tCache < t2 ) {
    NewCache( tCache );  // take a new BeamFolder cached time segment
    for (size_t it=0; it < cacheTimes.size(); ++it) {
      double t = cacheTimes[it];
      if ( t < t2 ) {
        if ( printTime ) {
          std::cout << "   [" << std::setw(6) << nspills << "] "
                    << std::fixed << std::setprecision(3) << t 
                    << " = " << UTCTimeString(t) << std::endl;
        }
        ++nspills;
        ProcessTime(t);
      }
    }
    // slide BeamFolder cache on to next time segment
    tCache = bf->GetCacheEndTime() + cfg.DeltaCacheTime;
  }
  return 0;
}
int ifbProcBase::ProcessTime(double t)
{
  SpillInit(t);
  for (size_t idevice=0; idevice < cacheDevices.size(); ++idevice) {
    std::string devname = cacheDevices[idevice];
    SpillDevice(t,devname);
  }
  SpillFinish(t);
  return 0;
}
int ifbProcBase::Run()
{
  int istat_init = Init();
  if ( istat_init != 0 ) {
    cfg.PrintConfig();
    std::cout << exename << ": init config returned " << istat_init
              << std::endl;
    exit(istat_init);
  }

  double t1 = cfg.TimeStamp;
  double t2 = t1 + cfg.WindowSize;
  int istat_scan = ScanWindow( t1, t2 );
  if ( istat_scan != 0 ) {
    cfg.PrintConfig();
    std::cout << exename << ": scan returned " << istat_scan
              << std::endl;
    exit(istat_scan);
  }

  int istat_finish = Finish();
  if ( istat_finish != 0 ) {
    cfg.PrintConfig();
    std::cout << exename << ": Finish() returned " << istat_init
              << std::endl;
    exit(istat_init);
  }
 
  if ( cfg.PrintOpts.find("spillcnt") != std::string::npos ) {
    std::cout << exename << ": saw " << nspills << " spills" << std::endl;
  }
  return 0;
}

int ifbProcRoto::Init() 
{
  //
  // Now, we need to start talking with the rotorooter by opening
  // the connection, send the data and close the connection.
  //
  int conn_status = bd.open_connection(cfg.RotoNode.c_str());
  if ( conn_status != 0 ) {
    std::cerr << exename << ": failed to make rotorooter connection (" 
              << conn_status << ")" << std::endl << std::flush;
    if ( ! cfg.AllowRotoErrors ) return conn_status;
  }

  //
  // Open a file in the rotorooter
  //
  // Do not use time of first spill in the window to name file
  // but instead use the start of the requested time window

  //int file_status = bd.open_file(sec, nsec);         /// time of first spill
  int file_status = bd.open_file((int)cfg.TimeStamp,0);

  if ( file_status != 0 ) {
    std::cerr << exename << ": failed to open rotorooter file (" 
              << file_status << ")" << std::endl << std::flush;
    if ( ! cfg.AllowRotoErrors ) return file_status;
  }

  return 0;
}
int ifbProcRoto::SpillInit(double tstamp)
{
    int secs = (int)tstamp;
    int secs_2ns = (tstamp - (int)tstamp)*1e9;
    //int secs_2ms = (tstamp - (int)tstamp)*1e3;

    // 3rd arg is 'tclk_event'   [169]
    // 4th arg is 'tclk_delay'   [500]
    bd.start_block(secs, secs_2ns, nspills, //spillcnt,
                    cfg.TClkTriggerEvent, cfg.TClkTriggerDelay);

  return 0;
}
int ifbProcRoto::SpillDevice(double tstamp, std::string devName)
{
  int spillSec, spillMSec;
  DoubleToSecMSec(tstamp,spillSec,spillMSec);

  if ( devName.find("[]") != std::string::npos ) {
    //
    // Vector device (has "[]" in the name)
    //
    std::vector<double> vDeviceValue;
    try {
      // GetNamed[Vector|Data] method uses 'double when' for the
      // 1st arg to pick closest time.
      double devTime;
      vDeviceValue = bf->GetNamedVector(tstamp,devName,&devTime);
      devcnt[devName]++;  // keep track of how often we succeeded
      
      int devSec, devMSec;
      DoubleToSecMSec(devTime,devSec,devMSec);
      CompareSpillDeviceTimes(devName,spillSec,spillMSec,devSec,devMSec);

      // 
      // Start the record entry for this device
      //
      bd.set_device_header(devName.c_str(), devSec, devMSec);

      // Add device's values to the record
      for ( size_t ih=0; ih < vDeviceValue.size(); ++ih ) {
        bd.add_device_value(devName.c_str(),vDeviceValue[ih]);
      }
      
    } catch(WebAPIException we) {
      // Device couldn't be read. we skip it and continue
      if ( cfg.Verbose > 3 ) {
        std::cout << "... skipping device " << devName
                  << " (" << (int)tstamp << " - not able to read)"
                  << std::endl;
      }
    }
    
  } else {
    //
    // Single valued device
    //
    double DeviceValue;
    try {
      // GetNamed[Vector|Data] method uses 'double when' for the
      // 1st arg to pick closest time.  Add "@" to name to get 
      // the associated time

      std::string devNameAndTime = devName + "@";
      double devTime;
      bf->GetNamedData(tstamp,devNameAndTime,&DeviceValue,&devTime);
      devcnt[devName]++;  // keep track of how often we succeeded

      int devSec, devMSec;
      DoubleToSecMSec(devTime,devSec,devMSec);
      CompareSpillDeviceTimes(devName,spillSec,spillMSec,devSec,devMSec);
      // 
      // Start the record entry for this device
      //
      bd.set_device_header(devName.c_str(), devSec, devMSec);

      // Add device's value to the record
      bd.add_device_value(devName.c_str(),DeviceValue);
      
    } catch(WebAPIException we) {
      // Device couldn't be read. we skip it and continue
      if ( cfg.Verbose > 3 ) {
        std::cout << "... skipping device " << devName 
                  << " (" << (int)tstamp << " - not able to read)"
                  << std::endl;
      }
    }
  }
  return 0;
}
int ifbProcRoto::SpillFinish(double)
{
  //
  // Send the record to the rotorooter
  //
  bd.send_data();
  return 0;
}
int ifbProcRoto::Finish()
{
  //
  // Close the file
  //
  bd.close_file();

  //
  // Close the connection to the rotorooter
  //
  bd.close_connection();

  //
  // optional print information
  //
  bool printDt = ( cfg.PrintOpts.find("times") != std::string::npos );
  if ( printDt ) {
    std::cout << exename << ": excursions spill vs. device time ( "
              << minDt << " " << maxDt << " )" << std::endl;
  }
  bool printDevices = ( cfg.PrintOpts.find("devices") != std::string::npos );
  bool printMissing = ( cfg.PrintOpts.find("missing") != std::string::npos );
  if ( printDevices || printMissing ) {
    std::cout << exename << ": " 
              << devcnt.size() << " devices " << nspills 
              << " spills";
    if ( ! printDevices ) cout << ", print only those missing from "
                               << "1 or more spills";
    std::cout << std::endl;
    int nMissingDevices = 0;
    std::map<std::string,size_t>::const_iterator itr = devcnt.begin();
    for ( ; itr != devcnt.end(); ++itr ) {
      size_t devSpills = itr->second;
      size_t missedSpills = ( nspills - devSpills);
      bool printDev = printDevices || ( missedSpills != 0 );
      if ( printDev ) {
        std::cout << "   " 
                  << std::left << std::setw(12) << itr->first << std::right
                  << " " << std::setw(6) << devSpills;
        if ( missedSpills > 0 ) {
          std::cout << ", missing from " << missedSpills;
          ++nMissingDevices;
        }
        std::cout << std::endl;
      }
    }
    std::cout << exename << ": " << nMissingDevices
              << " of " << devcnt.size() 
              << " devices missed at least one spill" << std::endl;
      
  }

  return 0;
}

void ifbProcRoto::CompareSpillDeviceTimes(const std::string& devname, 
                                          int spillSec, int spillMSec,
                                          int devSec, int devMSec)
{
  int diff = ( spillSec - devSec )*1000 + ( spillMSec - devMSec );
  if ( abs(diff) > cfg.DtTolerance ) {
    std::cout << "CompareSpillDeviceTimes " 
              << std::setw(12) << devname 
              << " sec " << std::setw(10) << spillSec 
              << "|" << std::setw(10) <<devSec
              << " msec " << std::setw(3) << spillMSec 
              << "|" << std::setw(3) << devMSec << " (delta " 
              << (diff) << ") " << std::endl;
  }
  if ( diff < minDt ) minDt = diff;
  if ( diff > maxDt ) maxDt = diff;

}


//_____________________________________________________________________________
bool IsLeapYear(int year)
{
   // Is the given year a leap year.


   // The calendar year is 365 days long, unless the year is exactly divisible
   // by 4, in which case an extra day is added to February to make the year
   // 366 days long. If the year is the last year of a century, eg. 1700, 1800,
   // 1900, 2000, then it is only a leap year if it is exactly divisible by
   // 400. Therefore, 1900 wasn't a leap year but 2000 was. The reason for
   // these rules is to bring the average length of the calendar year into
   // line with the length of the Earth's orbit around the Sun, so that the
   // seasons always occur during the same months each year.

   if (year%4 != 0) {
      return false;
   }
   else {
      if (year%400 == 0) {
         return true;
      }
      else { 
         if (year%100 == 0) {
            return false;
         }
         else {
            return true;
         }
      }
   }

}

//_____________________________________________________________________________
time_t MktimeFromUTC(tm_t *tmstruct)
{
   // Equivalent of standard routine "mktime" but
   // using the assumption that tm struct is filled with UTC, not local, time.

   // This version *ISN'T* configured to handle every possible
   // weirdness of out-of-range values in the case of normalizing
   // the tm struct.

   // This version *DOESN'T* correctly handle values that can't be
   // fit into a time_t (i.e. beyond year 2038-01-18 19:14:07, or
   // before the start of Epoch).

   const int days[]     = {31, 28, 31, 30, 31, 30, 31, 31, 30, 31, 30, 31};
   const int daysLeap[] = {31, 29, 31, 30, 31, 30, 31, 31, 30, 31, 30, 31};

   int year = tmstruct->tm_year + 1900;
   bool isleap = IsLeapYear(year);

   const int *daysInMonth = days;
   if (isleap) daysInMonth = daysLeap;

   // fill in tmstruct->tm_yday

   int &ref_tm_mon = tmstruct->tm_mon;
   int &ref_tm_mday = tmstruct->tm_mday;
   // count days in months past
   tmstruct->tm_yday = 0;
   for (int imonth = 0; imonth < ref_tm_mon; imonth++) {
      tmstruct->tm_yday += daysInMonth[imonth];
   }
   tmstruct->tm_yday += ref_tm_mday - 1;  // day [1-31] but yday [0-365]

   // adjust if day in this month is more than the month has
   while (ref_tm_mday > daysInMonth[ref_tm_mon]) {
      ref_tm_mday -= daysInMonth[ref_tm_mon];
      ref_tm_mon++;
   }

   // *should* calculate tm_wday (0-6) here ...

   // UTC is never DST
   tmstruct->tm_isdst = 0;

   // Calculate seconds since the Epoch based on formula in
   // POSIX  IEEEE Std 1003.1b-1993 pg 22

   int utc_sec = tmstruct->tm_sec +
                 tmstruct->tm_min*60 +
                 tmstruct->tm_hour*3600 +
                 tmstruct->tm_yday*86400 +
                 (tmstruct->tm_year-70)*31536000 +
                 ((tmstruct->tm_year-69)/4)*86400;

   return utc_sec;
}

//_____________________________________________________________________________
void DumpTMStruct(const tm_t &tmstruct) 
{
   // Print out the "tm" structure:
   // tmstruct.tm_year = year;    // years since 1900
   // tmstruct.tm_mon  = month-1; // months since Jan [0,11]
   // tmstruct.tm_mday = day;     // day of the month [1,31]
   // tmstruct.tm_hour = hour;    // hours since midnight [0,23]
   // tmstruct.tm_min  = min;     // minutes after the hour [0,59]
   // tmstruct.tm_sec  = sec;     // seconds after the minute [0,59]
   // tmstruct.tm_wday            // day of week [0,6]
   // tmstruct.tm_yday            // days in year [0,365]
   // tmstruct.tm_isdst           // DST [-1/0/1]  (unknown,false,true)

   printf(" tm { year %4d, mon   %2d, day   %2d,\n",
          tmstruct.tm_year,
          tmstruct.tm_mon,
          tmstruct.tm_mday);
   printf("      hour   %2d, min   %2d, sec   %2d,\n",
          tmstruct.tm_hour,
          tmstruct.tm_min,
          tmstruct.tm_sec);
   printf("      wday   %2d, yday %3d, isdst %2d",
          tmstruct.tm_wday,
          tmstruct.tm_yday,
          tmstruct.tm_isdst);
#ifdef linux
//#ifdef __GNUC__
// special GCC extras
   printf(",\n      tm_gmtoff %7ld,  tm_zone \"%s\"",
#ifdef __USE_BSD
          tmstruct.tm_gmtoff,tmstruct.tm_zone);
#else
          tmstruct.__tm_gmtoff,tmstruct.__tm_zone);
#endif
#endif
   printf("}\n");
}
//_____________________________________________________________________________

double ParseTimeString(const char* s)
{
   // assume string of the form 'YYYY-MM-DD hh:mm:ss[.nnnnn]'
   bool isUTC = true;

   int year=0, month=0, day=0;
   int hour=0, min=0, sec=0;
   float fltsec=0, frac_sec;
   char buff[128] = {0};

   int nitems = sscanf(s,"%d-%d-%d %d:%d:%f%s",
                       &year,&month,&day,&hour,&min,&fltsec,buff);
   //std::cout << "rest of time string \"" << buff << "\"" << std::endl;

   if ( nitems != 6 ) {
     std::cout << "ParseTimeString only got " << nitems << " items" 
               << std::endl;
   }
   // possibility of fractional seconds
   sec = (int)fltsec;
   frac_sec = ( fltsec - sec );

   // check trailing text
   // if anything but "", "Z", "UTC" or "GMT" assume time given was
   // in local time ... don't try to parse it as a TZ ...
   std::string tzstr(buff);
   // trim lead/trailing blanks
   if( tzstr.find_first_not_of(" ()[]{}\n") != 0)
     tzstr.erase( 0,  tzstr.find_first_not_of(" ()[]{}\n")  );
   if( tzstr.find_last_not_of(" ()[]{}\n") != tzstr.length() )
     tzstr.erase( tzstr.find_last_not_of(" ()[]{}\n")+1, tzstr.length() );

   isUTC = ( tzstr == "" || tzstr == "Z" || tzstr == "UTC" || tzstr == "GMT" );

   // month & day both use normal 1..12 and 1..31 counting
   // hours, min, sec run from 0 to 23, 59, 59 respectively;
   // secOffset provides method for adjusting for alternative timezones
   //
   // "year"  |    0    1 ... 37 | 38...69   |   70 .. 100  101 ..  137
   // true    | 2000 2001   2037 | undefined | 1970   2000 2001 .. 2037
   //
   // "year"  | 138...1969 | 1970 .. 2037 | ...
   // true    | undefined  | 1970 .. 2037 | undefined 
   //

   // deal with special formats of year
   if (year <= 37)                year += 2000;
   if (year >= 70 && year <= 137) year += 1900;
   // tm.tm_year is years since 1900
   if (year >= 1900)              year -= 1900;

   struct tm tmstruct;
   tmstruct.tm_year  = year;    // years since 1900
   tmstruct.tm_mon   = month-1; // months since Jan [0,11]
   tmstruct.tm_mday  = day;     // day of the month [1,31]
   tmstruct.tm_hour  = hour;    // hours since midnight [0,23]
   tmstruct.tm_min   = min;     // minutes after the hour [0,59]
   tmstruct.tm_sec   = sec;     // seconds after the minute [0,59]     
   tmstruct.tm_isdst = -1;      // let "mktime" determine DST setting

   //const time_t bad_time_t = (time_t) -1;
   // convert tm struct to time_t, if values are given in UTC then
   // no standard routine exists and we'll have to use our homegrown routine,
   // if values are given in local time then use "mktime"
   // which also normalizes the tm struct as a byproduct
   time_t utc_sec = (isUTC) ? MktimeFromUTC(&tmstruct) : mktime(&tmstruct);

   double atime = utc_sec + frac_sec;
   return atime;
}

std::string UTCTimeString(double ts)
{
  char buffergmt[80];
  time_t gmttime = (time_t)ts;
  struct tm * timeinfogmt = gmtime( &gmttime );
  strftime (buffergmt,80,"%F %T (UTC)",timeinfogmt);
  return std::string( buffergmt );
}

std::string LocalTimeString(double ts)
{
  char bufferloc[80];
  time_t loctime = (time_t)ts;
  struct tm * timeinfoloc = localtime( &loctime );
  strftime (bufferloc,80,"%F %T (local)",timeinfoloc);
  return std::string( bufferloc );
}

---