IoInputStreamItr.cxx

Go to the documentation of this file.

// $Id: IoInputStreamItr.cxx,v 1.17 2009/01/05 05:55:13 schubert Exp $
//
// Class for walking over the contents of and input data stream
// 
// messier@huhepl.harvard.edu
// schubert@hep.umn.edu (S. Kasahara) 
#include "IoModules/IoInputStreamItr.h"

#include <algorithm>
#include "TClass.h"
#include "TKey.h"
#include "TFile.h"
#include "TROOT.h"
#include "TSystem.h"
#include "TRegexp.h"
#include "MessageService/MsgService.h"
#include "Persistency/PerFile.h"
#include "Persistency/PerFileManager.h"
#include "Util/UtilString.h"

CVSID("$Id: IoInputStreamItr.cxx,v 1.17 2009/01/05 05:55:13 schubert Exp $");

//......................................................................

IoInputStreamItr::IoInputStreamItr(const char* streamlist) : fIsValid(true) {
//======================================================================
// Argument: Delimiter separated list of streams to open.
//======================================================================
  this -> Streams(streamlist);
}

//......................................................................

IoInputStreamItr::~IoInputStreamItr() {
//======================================================================
// Destructor
//======================================================================
  MSG("Io",Msg::kDebug) << 
    "Close streams " << this->GetSourceName() << "\n";
  fInputStreamManager.CloseStream(); 
}

//......................................................................

int IoInputStreamItr::LoadRecords(MomNavigator* m) {
//======================================================================
// Retrieve current record set from manager. Returns number of records
// loaded.
//======================================================================
  int nrecord = fInputStreamManager.Get(m);
  return nrecord;
}

//......................................................................

int IoInputStreamItr::Increment(int n, MomNavigator* m) {
//======================================================================
// Advance n record sets.  Returns number of record sets advanced.
//======================================================================

  return fInputStreamManager.Next(m,n);
}

//......................................................................

int IoInputStreamItr::Decrement(int n, MomNavigator* m) {
//======================================================================
// Rewind n record sets. Returns number of record sets rewound.
//======================================================================
  return fInputStreamManager.Previous(m,n);
}

//......................................................................

JobCResult IoInputStreamItr::GoTo(const VldContext& vld, MomNavigator* m) {
//======================================================================
// Go directly to a record with a specified validity context. Will
// stop at record set one past requested validity if exact match not 
// available.  Returns:
//      JobCResult::kBeginOfInputStream if at begin of streams
//      JobCResult::kEndOfInputStream if at end of streams
//      JobCResult::kAOK if all okay
//======================================================================
  const VldContext& newVld = fInputStreamManager.RecordsAt(m,vld);
  if ( newVld == Per::GetVldEnd() ) return JobCResult::kEndOfInputStream;
  if ( newVld == Per::GetVldBegin() ) return JobCResult::kBeginOfInputStream;
  return JobCResult::kAOK; // success
}

//......................................................................

int IoInputStreamItr::GoToEOF() {
//======================================================================
// Advance to the end of all files
//======================================================================

  fInputStreamManager.RecordsAt((MomNavigator*)0,Per::GetVldEnd());
  return 1;

}

//......................................................................

int IoInputStreamItr::DefineStream(const char* streamname,const char* treename)
{
//======================================================================
// Define stream with name streamname to serve tree of name treename.
//======================================================================

  MSG("Io",Msg::kDebug) <<
    "Define stream " << streamname << " to serve tree " << treename << endl;

  if ( fInputStreamManager.GetOpenedStream(streamname) ) {
    fInputStreamManager.CloseStream(streamname);
  }
  fInputStreamManager.OpenStream(streamname,treename);

  return 1;
}

//......................................................................

int IoInputStreamItr::Streams(const char* streams) {
//======================================================================
// Modify open streams to match those specified in streams.  streams
// is a delimiter separated list of streamnames. Returns number of
// streams open.
//======================================================================

  this -> SetSourceName(streams);
  MSG("Io",Msg::kDebug) << "Set streams " << streams << endl;

  // parse delimiter separate streamlist into a vector of streamnames
  std::vector<std::string> streamlist;
  UtilString::StringTok(streamlist,std::string(streams),":,;/ ");

  std::vector<std::string>::iterator strItr;
  // Add streams not previously opened
  for ( strItr = streamlist.begin(); strItr != streamlist.end(); strItr++ ) {
    std::string streamname = *strItr;
    if ( !fInputStreamManager.GetOpenedStream(streamname) ) {
      fInputStreamManager.OpenStream(streamname,streamname);
    }
  }
  
  // Close streams not in requested list
  const PerStreamManager::StreamMap& smap = fInputStreamManager.GetStreamMap();
  for ( PerStreamManager::StreamMapConstItr citr = smap.begin();
                                            citr!= smap.end(); ++citr ) {
    std::string streamname = citr->first;
    typedef std::vector<std::string>::const_iterator vs_citer;
    vs_citer vpos = std::find(streamlist.begin(),streamlist.end(),streamname);
    if ( vpos == streamlist.end() ) { 
      fInputStreamManager.CloseStream(streamname);
    }
  }

  // Open streams associated with those in requested list.  Since assocated
  // streams may in turn have associated streams, this is done as a while
  // loop until no more streams are added
  bool isNewStream = true;
  while ( isNewStream ) {
    isNewStream = false;
    for ( PerStreamManager::StreamMapConstItr citr = smap.begin();
                                              citr!= smap.end(); ++citr ) {
      std::string streamname = citr->first;
      std::string assocstreams 
                          = Per::GetAssociatedStreamList(streamname.c_str());
      if ( !assocstreams.empty() ) {
        // parse delimiter separated streamlist into a vector of streamnames
        std::vector<std::string> assocstreamlist;
        UtilString::StringTok(assocstreamlist,
                              std::string(assocstreams),":,;/ ");
        std::vector<std::string>::iterator astrItr;
        // Add associated streams not previously opened
        for ( astrItr = assocstreamlist.begin(); 
              astrItr!= assocstreamlist.end(); astrItr++ ) {
          std::string assocstreamname = *astrItr;
          if ( !fInputStreamManager.GetOpenedStream(assocstreamname) ) {
            fInputStreamManager.OpenStream(assocstreamname,assocstreamname);
            isNewStream = true;
          }
        }
      }
    }
  }


  return fInputStreamManager.GetNumStream();
}

//......................................................................

int IoInputStreamItr::Select(const char* stream, const char* selection,
                             bool isRequired) {
//======================================================================
// Set the selection cuts for named stream.  Returns 0 if stream not
// open, else 1.
//======================================================================
  int isOpen = 0;
  MSG("Io",Msg::kDebug) <<
    "Set selection string " << selection << " for stream " << stream << endl;

  if ( std::string(stream) == "*" ) {
    fInputStreamManager.SetSelection(stream,selection,isRequired);
    isOpen = 1;
  }
  else if ( fInputStreamManager.GetOpenedStream(stream) ) {
    fInputStreamManager.SetSelection(stream,selection,isRequired);
    isOpen = 1;
  }
  return isOpen;
}

//......................................................................

int IoInputStreamItr::SetPerOwnedDisabled(const char* stream, 
                                          bool perowneddisabled) {
//======================================================================
// Used to disable PerOwned for named stream.  Returns 0 if stream not
// open, else 1.
//======================================================================
  int isOpen = 0;
  MSG("Io",Msg::kDebug) << "SetPerOwnedDisabled " << perowneddisabled 
                        << " for stream " << stream << endl;

  if ( std::string(stream) == "*" ) {
    fInputStreamManager.SetPerOwnedDisabled(stream,perowneddisabled);
    isOpen = 1;
  }
  else if ( fInputStreamManager.GetOpenedStream(stream) ) {
    fInputStreamManager.SetPerOwnedDisabled(stream,perowneddisabled);
    isOpen = 1;
  }
  return isOpen;
}

//......................................................................

int IoInputStreamItr::SetSequenceMode(const char* stream, 
                                   Per::ESequenceMode sequenceMode) {
//======================================================================
// Set the sequence mode for named stream.  Returns 0 if stream not
// open, else 1.
//======================================================================
  int isOpen = 0;
  MSG("Io",Msg::kDebug) <<
    "Set sequence mode " << Per::AsString(sequenceMode) << " for stream " 
                         << stream << endl;

  if ( fInputStreamManager.GetOpenedStream(stream) ) {
    fInputStreamManager.SetSequenceMode(stream,sequenceMode);
    isOpen = 1;
  }
  return isOpen;
}

//......................................................................

int IoInputStreamItr::SetTestMode(const char* stream, 
                                  bool testMode) {
//======================================================================
// Set the test mode for named stream.  Returns 0 if stream not
// open, else 1.
//======================================================================
  int isOpen = 0;
  std::string testModeStr = "false";
  if ( testMode ) testModeStr = "true";

  MSG("Io",Msg::kDebug) <<
    "Set test mode " << testModeStr.c_str() << " for stream " 
                     << stream << endl;

  if ( std::string(stream) == "*" ) {
    fInputStreamManager.SetTestMode(stream,testMode);
    isOpen = 1;
  }
  else if ( fInputStreamManager.GetOpenedStream(stream) ) {
    fInputStreamManager.SetTestMode(stream,testMode);
    isOpen = 1;
  }
  return isOpen;

}

int IoInputStreamItr::SetWindow(const char* stream, double lower, double upper)
{
//======================================================================
//Set the window for named stream.  Returns 0 if stream not open, else
//1.
//======================================================================
  int isOpen = 0;
  MSG("Io",Msg::kDebug) <<
      "Set window [" << lower << ", " << upper << "] for stream " 
                        << stream << endl;

  if ( fInputStreamManager.GetOpenedStream(stream) ) {
    fInputStreamManager.SetWindow(stream,lower,upper);
    isOpen = 1;
  }
  return isOpen;
}

int IoInputStreamItr::SetMaxFileRepeat(const char* stream, int numRepeat)
{
//======================================================================
// Set the number of times to reuse a file before moving to the next one
// in the named stream.
// Returns 0 if stream not open, else 1.
//======================================================================
  int isOpen = 0;
  if (numRepeat < 0) numRepeat = 0;
  MSG("Io",Msg::kDebug)
    << "Set number of times to reuse each file from stream " << stream
    << " = " << numRepeat << endl;

  if ( fInputStreamManager.GetOpenedStream(stream) ) {
    fInputStreamManager.SetMaxFileRepeat(stream,numRepeat);
    isOpen = 1;
  }
  return isOpen;
}

int IoInputStreamItr::SetMeanMom(const char* stream, double mean)
{
//======================================================================
// Set the mean #evts for named stream.
// Returns 0 if stream not open, else 1.
//======================================================================
  int isOpen = 0;
  MSG("Io",Msg::kDebug)
    << "Set mean number of events to push to Mom from stream " << stream
    << " = " << mean << endl;

  if ( fInputStreamManager.GetOpenedStream(stream) ) {
    fInputStreamManager.SetMeanMom(stream,mean);
    isOpen = 1;
  }
  return isOpen;
}

int IoInputStreamItr::SetPushRandom(const char* stream, bool setRandom)
{
//======================================================================
// Set whether to push a random/constant #evts for named stream to Mom.
// Returns 0 if stream not open, else 1.
//======================================================================
  int isOpen = 0;
  MSG("Io",Msg::kDebug)
    << "Set push " << ((setRandom)? "random " : "constant ")
    << "number of events to push to Mom from stream " << stream
    << endl;

  if ( fInputStreamManager.GetOpenedStream(stream) ) {
    fInputStreamManager.SetPushRandom(stream,setRandom);
    isOpen = 1;
  }
  return isOpen;
}

void IoInputStreamItr::SetRandomSeed(int rSeed)
{
//======================================================================
// Set the random seet for the InputStreamManager
//======================================================================

  MSG("Io",Msg::kDebug)
    << "Set InputStreamManager random seed to " << rSeed << endl;

  fInputStreamManager.SetRandomSeed(rSeed);

}


void IoInputStreamItr::AddFile(const char* fullfilepathname, int at,
                               const char* streamlist ) { 
//======================================================================
// Purpose: Add file fullfilepathname to index position at in filelist
//          associated with the stream(s) in streamlist. If streamlist == "*",
//          files will be applied to all managed streams, otherwise streamlist
//          is a delimiter separated list of streams.
//
// Arguments: fullfilepathname  filename.
//            at        the index position in the file list at which to
//                      insert the new file. Set = -1 (default) => insert
//                      at end of file list.
//            streamlist Delimiter separated list of streams (e.g.
//                       "DaqSnarl,DaqMonitor") on which to apply action.
//                       If streamlist = "*" (default) => apply to all streams.
//
// Return: none.
//         
// Contact: S. Kasahara
//======================================================================

  // parse delimiter separate streamlist into a vector of streamnames
  std::vector<std::string> vstreams;
  UtilString::StringTok(vstreams,streamlist,":,;/ ");
  std::vector<std::string>::iterator vsItr;

  int nstream = 0;
  for ( vsItr = vstreams.begin(); vsItr != vstreams.end(); vsItr++ ) {
    std::string streamname = *vsItr;
    if ( fInputStreamManager.GetOpenedStream(streamname) ) { 
      nstream += fInputStreamManager.AddFile(fullfilepathname,at,streamname);
    }
    std::string assocstreams 
                     = Per::GetAssociatedStreamList(streamname.c_str());
    if ( !assocstreams.empty() ) {
      // parse delimiter separated streamlist into a vector of streamnames
      std::vector<std::string> assocstreamlist;
      UtilString::StringTok(assocstreamlist,
                            std::string(assocstreams),":,;/ ");
      std::vector<std::string>::iterator astrItr;
      // Add file to associated streams 
      for ( astrItr = assocstreamlist.begin(); 
            astrItr!= assocstreamlist.end(); astrItr++ ) {
        std::string assocstreamname = *astrItr;
        // Check to see if assocstreamname isn't already in input streamlist
        // This is lame at best and needs to be fixed.
        typedef std::vector<std::string>::const_iterator vs_citer;
        vs_citer aspos = std::find(vstreams.begin(),vstreams.end(),
                                   assocstreamname);
        if ( aspos == vstreams.end() ) { 
          if ( fInputStreamManager.GetOpenedStream(assocstreamname) ) {
            nstream 
            +=fInputStreamManager.AddFile(fullfilepathname,at,assocstreamname);
          }
        }
      }
    }
  }

  return;
}

const char* IoInputStreamItr::GetCurrentFile(const char* streamname) const { 
//======================================================================
// Purpose: Return current file of specified stream.
//
// Arguments: streamname If specified as "*", returns current file of
//                       first open stream.
//
// Return: current file name
//         
// Contact: S. Kasahara
//======================================================================

  return (fInputStreamManager.GetCurrentFile(streamname)).c_str();

}

JobCResult IoInputStreamItr::GoToFile( int n, const char* streamlist ) { 
//======================================================================
// Purpose: GoToFile at position n in filelist of specified streams.
//
// Arguments: n          index of file in filelist.
//            streamlist Delimiter separated list of streams (e.g.
//                       "DaqSnarl,DaqMonitor") on which to apply action.
//                       If streamlist = "*" (default) => apply to all streams.
// 
// Return: JobCResult::kAOK if advance successful and file opened okay on
//                          at least one requested stream, or
//         JobCResult::kWarning, if requested file number out of range, or
//                               file open error occured on all requested
//                               streams.
// 
// Contact: S. Kasahara
//======================================================================

  int nstream = 0;

  // parse delimiter separate streamlist into a vector of streamnames
  std::vector<std::string> vstreams;
  UtilString::StringTok(vstreams,streamlist,":,;/ ");
  std::vector<std::string>::iterator vsItr;

  for ( vsItr = vstreams.begin(); vsItr != vstreams.end(); vsItr++ ) {
    std::string streamname = *vsItr;
    nstream += fInputStreamManager.GoToFile(n,streamname);
  }

  if ( !nstream ) return JobCResult::kWarning;
  return JobCResult::kAOK;

}

JobCResult IoInputStreamItr::GoToFile(const char* filename, 
                                      const char* streamlist) { 
//======================================================================
// Purpose: GoToFile filename in filelist of specified streams.
//
// Arguments: filename   full file path name of file in streamlist.
//            streamlist Delimiter separated list of streams (e.g.
//                       "DaqSnarl,DaqMonitor") on which to apply action.
//                       If streamlist = "*" (default) => apply to all streams.
// 
// Return: JobCResult::kAOK if advance successful and file opened okay on
//                          at least one requested stream, or
//         JobCResult::kWarning, if requested file not found in file list or
//                               file open error occured on all requested
//                               streams.
// 
// Contact: S. Kasahara
//======================================================================

  int nstream = 0;

  // parse delimiter separate streamlist into a vector of streamnames
  std::vector<std::string> vstreams;
  UtilString::StringTok(vstreams,streamlist,":,;/ ");
  std::vector<std::string>::iterator vsItr;

  for ( vsItr = vstreams.begin(); vsItr != vstreams.end(); vsItr++ ) {
    std::string streamname = *vsItr;
    nstream += fInputStreamManager.GoToFile(filename,streamname);
  }

  if ( !nstream ) return JobCResult::kWarning;
  return JobCResult::kAOK;
}

JobCResult IoInputStreamItr::NextFile(int n, const char* streamlist) { 
//======================================================================
// Purpose: Advance forward n files in filelist of specified stream(s).
//
// Arguments: n   number of files by which to advance.(default = 1)
//            streamlist Delimiter separated list of streams (e.g.
//                       "DaqSnarl,DaqMonitor") on which to apply action.
//                       If streamlist = "*" (default) => apply to all streams.
// 
// Return: JobCResult::kAOK if advance successful and file opened okay for
//                          at least one stream, or
//         JobCResult::kWarning if advance successful, but file open
//                              failed.
//         JobCResult::kEndOfInputStream, if at end of file list for
//                                        all requested streams.
// 
// Contact: S. Kasahara
//======================================================================

  int nstream = 0;

  // parse delimiter separate streamlist into a vector of streamnames
  std::vector<std::string> vstreams;
  UtilString::StringTok(vstreams,streamlist,":,;/ ");
  std::vector<std::string>::iterator vsItr;

  if ( vstreams.empty() ) return JobCResult::kAOK; // no streams requested

  bool isWarning = false;
  for ( vsItr = vstreams.begin(); vsItr != vstreams.end(); vsItr++ ) {
    std::string streamname = *vsItr;
    int nOK = fInputStreamManager.NextFile(n,streamname);
    if ( !nOK ) { 
      // Failed to advance to file but not at end of files means open failure
      if ( !fInputStreamManager.IsEndOfFiles(streamname) ) isWarning = true;
    }
    else nstream += nOK;
  }

  if ( !nstream ) {
    if ( isWarning ) return JobCResult::kWarning;
    else return JobCResult::kEndOfInputStream;
  }

  return JobCResult::kAOK;

}

JobCResult IoInputStreamItr::PrevFile(int n, const char* streamlist) { 
//======================================================================
// Purpose: Rewind backward n files in filelist of specified stream(s).
//
// Arguments: n   number of files by which to rewind.(default = 1)
//            streamlist Delimiter separated list of streams (e.g.
//                       "DaqSnarl,DaqMonitor") on which to apply action.
//                       Set = "*" (default) => apply to all streams.
// 
// Return: JobCResult::kAOK if rewind successful and file opened okay for
//                          at least one stream, or
//         JobCResult::kWarning if rewind successful, but file open
//                              failed.
//         JobCResult::kBeginOfInputStream, if at end of file list for
//                                          all requested streams.
//
// Contact: S. Kasahara
//======================================================================

  int nstream = 0;

  // parse delimiter separate streamlist into a vector of streamnames
  std::vector<std::string> vstreams;
  UtilString::StringTok(vstreams,streamlist,":,;/ ");
  std::vector<std::string>::iterator vsItr;

  if ( vstreams.empty() ) return JobCResult::kAOK; // no streams requested

  bool isWarning = false;
  for ( vsItr = vstreams.begin(); vsItr != vstreams.end(); vsItr++ ) {
    std::string streamname = *vsItr;
    int nOK = fInputStreamManager.PrevFile(n,streamname);
    if ( !nOK ) { 
      // Failed to advance to file but not at begin of files means open failure
      if ( !fInputStreamManager.IsBeginOfFiles(streamname) ) isWarning = true;
    }
    else nstream += nOK;
  }

  if ( !nstream ) {
    if ( isWarning ) return JobCResult::kWarning;
    else return JobCResult::kBeginOfInputStream;
  }

  return JobCResult::kAOK;

}

void IoInputStreamItr::RemoveFile(const char* filename,const char* streamlist){
//======================================================================
// Purpose: Remove file filename from filelist associated with the
//          stream(s) in streamlist.  If filename = "*", all files are
//          removed.  
//
// Arguments: filename  full file path name of file to remove.  If "*",
//                      all files will be removed from file list. 
//            streamlist Delimiter separated list of streams (e.g.
//                       "DaqSnarl,DaqMonitor") on which to apply action.
//                       If streamlist = "*" (default) => apply to all streams.
//
// Return: none.
//
// Contact: S. Kasahara
//======================================================================

  int nstream = 0;

  // parse delimiter separate streamlist into a vector of streamnames
  std::vector<std::string> vstreams;
  UtilString::StringTok(vstreams,streamlist,":,;/ ");
  std::vector<std::string>::iterator vsItr;

  for ( vsItr = vstreams.begin(); vsItr != vstreams.end(); vsItr++ ) {
    std::string streamname = *vsItr;
    nstream += fInputStreamManager.RemoveFile(filename,streamname);
  }

  return;
}

std::ostream& IoInputStreamItr::ListFile(std::ostream& os, 
                                         const char* streamlist) const { 
//======================================================================
// Purpose: List files in file list of specified stream(s).
//
// Arguments: os  ostream to direct print.
//            streamlist Delimiter separated list of streams (e.g.
//                       "DaqSnarl,DaqMonitor") on which to apply action.
//                       If streamlist = "*" (default) => apply to all streams.
// 
// Return: ostream reference.
// 
// Contact: S. Kasahara
//======================================================================

  // parse delimiter separated streamlist into a vector of streamnames
  std::vector<std::string> vstreams;
  UtilString::StringTok(vstreams,streamlist,":,;/ ");
  std::vector<std::string>::iterator vsItr;

  for ( vsItr = vstreams.begin(); vsItr != vstreams.end(); vsItr++ ) {
    std::string streamname = *vsItr;
    fInputStreamManager.ListFile(os,streamname);
  }

  return os;
}

---