RootAFP.cpp

Go to the documentation of this file.

/***************************************************************************
 *   Copyright (C) 2004 by Olivier Stezowski                               *
 *   stezow(AT)ipnl.in2p3.fr                                               *
 *                                                                         *
 *   This program is free software; you can redistribute it and/or modify  *
 *   it under the terms of the GNU General Public License as published by  *
 *   the Free Software Foundation; either version 2 of the License, or     *
 *   (at your option) any later version.                                   *
 *                                                                         *
 *   This program is distributed in the hope that it will be useful,       *
 *   but WITHOUT ANY WARRANTY; without even the implied warranty of        *
 *   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the         *
 *   GNU General Public License for more details.                          *
 *                                                                         *
 *   You should have received a copy of the GNU General Public License     *
 *   along with this program; if not, write to the                         *
 *   Free Software Foundation, Inc.,                                       *
 *   59 Temple Place - Suite 330, Boston, MA  02111-1307, USA.             *
 ***************************************************************************/

#ifndef ADF_RootAFP
#include "RootAFP.h"
#endif

#include <iostream>
#include <iomanip>
#include <fstream>
#include <sstream>

using namespace ADF; 

RootAFP::RootAFP(): 
    NarvalProducer(),
    fSourceOfFrames(aMByte,ConfAgent::kRead),
    fPath("./"),
    fBaseForName("AFP_"),
    fCurrentFile(0x0),
    fCurrentFileNumber(0u),
    fMaxSize(kMaxInt_t),
    fConfMode(-1),
    fListOfFile(),
    TotalEntryFileSize(0),
    ReadEntryFileSize(0),
    LastReadEntryFileSize(0),
    fEndOfFrame("EndOfFrame")
{       
    Log.GetProcessName() = "RootAFP"; GetFrameIO().SetName("RootAFP");

    // to be sure unknown frames are written in the ouput
    GetFrameIO().SetModel(ConfAgent::kSafe);
    // to protect writings in the source of frames
    fSourceOfFrames.SetModeIO(ConfAgent::kRead);

    // the input datablock is owned by BasicAFP
    GetFrameIO().Attach(&fSourceOfFrames,0x0);
    fListOfFile.clear();
}

RootAFP::~RootAFP() 
{  
    // in principle not needed ... just in case reset it has not been called by narval
    UInt_t error = 0u; process_reset(&error) ;
} 

Bool_t RootAFP::NewFile()
{       
    Log.SetProcessMethod("NewFile");
    Bool_t ok = false;

    if ( fConfMode == -1 ) // not properly initialised
        return ok;

    // close the current file
    if ( fCurrentFile )
        ::fclose(fCurrentFile);
    fCurrentFile = 0x0;

    if ( fConfMode == 0 ) {
        // open the first file to be ready to be run
        std::string tmp = fPath;
        if ( tmp.size() > 0 && !(tmp.at(tmp.size()-1) == '/') )
            tmp += '/';

        // try to open the next file
        std::ostringstream filename;
        filename << tmp << fBaseForName
                 << std::setfill('0') << std::setw(4) << fCurrentFileNumber++ << std::setfill(' ')
                 << ".adf";

        fCurrentFile = ::fopen(filename.str().data(),"rb");
        if ( fCurrentFile != 0x0 ) {
            Log << "A new input file has just been open "
                << filename.str() << nline;
            fSourceOfFrames.SetFile
                    (fCurrentFile,fMaxSize);
            GetFrameIO().SetStatus
                    (BaseFrameIO::kIdle);
            ok = true;
        }
        else {
            Log << warning << "Cannot open file "
                << filename.str()
                << nline;
            GetFrameIO().SetStatus(BaseFrameIO::kFinished);
        }
    }
    else {
        // try to open the next file
        while ( fCurrentFileNumber < fListOfFile.size() ) {

            fCurrentFile = ::fopen(fListOfFile[fCurrentFileNumber].data(),"rb");
            if ( fCurrentFile == 0x0 ) {
                Log << warning << "File "
                    << fListOfFile[fCurrentFileNumber]
                       << " is not a valid file "
                       << nline ;

                fCurrentFileNumber++;
                continue;
            }
            Log << info << "A new input file has just been open "
                << fListOfFile[fCurrentFileNumber]
                   << nline;
            fSourceOfFrames.SetFile(fCurrentFile,fMaxSize);
            GetFrameIO().SetStatus(BaseFrameIO::kIdle);
            ok = true;
            fCurrentFileNumber++;

            break;
        }
        if ( !ok )
            GetFrameIO().SetStatus(BaseFrameIO::kFinished);
    }
    // send the log collected in this method
    Log << dolog;

    return ok;
}

void RootAFP::process_config (const Char_t *directory_path, UInt_t *error_code)
{       
    // first init narval and ADF stuff (if required)
    NarvalInterface::process_config(directory_path,error_code);

    if ( (*error_code) == 0u ) { // now init your stuff
    }
}

UInt_t RootAFP::ProcessBlock(ADF::FrameBlock &out) 
{
    Log.ClearMessage(); Log.SetProcessMethod("process_block");

    // attach the FrameBlock
    GetFrameIO().Attach(0x0,&out);

    // fill the output with successive frames read from the file until out is filled
    while ( GetFrameIO().Notify() ) {
    }
    // interruption: means end of file or out is full
    if ( GetFrameIO().GetCurrentBlockIN()->IsEoB() ) { // the file has been read, try to open the next one
        if ( NewFile() == false )
            GetFrameIO().SetStatus(BaseFrameIO::kFinished);
    }

    ReadEntryFileSize += out.GetSize();

    double Ratio = ((double)ReadEntryFileSize)/((double)TotalEntryFileSize)*100.;
    double LastRatio = ((double)LastReadEntryFileSize)/((double)TotalEntryFileSize)*100.;

    std::ostringstream toto;
    toto<<TotalEntryFileSize;
    int NDigits = toto.str().length();

    if(((int)Ratio) != ((int)LastRatio))
    {
        std::cout<<"\r"<<"Analysis progress : "<<std::setw(3)<<(int)(Ratio+0.5)<<" %"<<" ("<<std::setw(NDigits)<<ReadEntryFileSize<<"/"<<std::setw(NDigits)<<TotalEntryFileSize<<" bytes)"<<std::flush;
        if((int)(Ratio+0.5)==99 || (int)(Ratio+0.5)==100) std::cout<<std::endl;
    }

    LastReadEntryFileSize = ReadEntryFileSize;

    if ( GetFrameIO().GetStatus() == BaseFrameIO::kFinished ) {
        // set a trigger to send end of frames (single shot) in case the producer is in finish mode and the user ask for an new frame
        if ( fEndOfFrame.GetOutputFrame() == 0x0 ) {
            // a eof frame
            FactoryItem key_item("Agata","meta:eof",Version(1,0)), frame_item("Agata","meta:eof",Version(0,0));
            // allocate and record it using trigger
            SharedFP *eof = fEndOfFrame.SetOutputFrame( key_item, frame_item );
            if ( eof ) {
                // add a eof
                GetFrameIO().Register(&fEndOfFrame);
                fEndOfFrame.Fired(true);
                fEndOfFrame.GetOutputFrame()->Write();

                GetFrameIO().Record(&fEndOfFrame);
                //
                Log << dolog;
                // return with this frame

                return 0u;
            }
        }
        Log << dolog;
        return 1u;
    }

    // detach the output block
    GetFrameIO().Detach(0x0,&out);
    //
    Log << dolog;
    return 0u;
}       

void RootAFP::process_initialise (UInt_t *error_code)
{
    std::string tmp, option;

    *error_code = 0u; Log.ClearMessage(); Log.SetProcessMethod("process_initialise");

    // read an input file to get the base filename and the path
    std::string conffile
            = GetConfPath();
    conffile += "RootAFP.conf";

    fCurrentFileNumber = 0u;

    std::ifstream filein(conffile.data());
    if ( filein.is_open() == true ) {
        getline(filein,tmp);
        tmp += " ";
        while ( filein.good() && !filein.eof() ) { // read input stream line by line
            if ( tmp[0] == '#' ) {
                getline(filein,tmp);
                tmp += " ";
                continue;
            } // this line is a comment

            std::istringstream decode(tmp);

            if( tmp[0] == 'b' ){ // to emulate a BasicAFP

                std::string pathforfiles, basename;
                UInt_t starting_number;
                decode >> option >> pathforfiles >> basename >> starting_number;

                if ( decode.good() ) {

                    fPath = pathforfiles;
                    fBaseForName = basename;
                    fCurrentFileNumber = starting_number;

                    fConfMode = 0;

                    break;
                }
            }
            if( tmp[0] == 'l' ) { // the .conf file contains the list of file to be read

                std::string filename;
                decode >> option >> filename;
                if( decode.good() )
                    fListOfFile.push_back(filename);

                fConfMode = 1;
            }
            if( tmp[0] == 'r' ) { // the .conf contains a range of files to be read
                UInt_t starting_number, ending_number;
                std::string pathforfiles, basename;
                decode >> option >> pathforfiles >> basename >> starting_number >> ending_number;
                if( decode.good() && starting_number < ending_number ){
                    for( UInt_t number = starting_number; number <= ending_number; number++ ){
                        std::ostringstream osfilename;
                        osfilename  << pathforfiles << basename
                                    << std::setfill('0') << std::setw(4) << number << std::setfill(' ')
                                    << ".adf";
                        fListOfFile.push_back(osfilename.str());
                    }
                }
                fConfMode = 2;
            }
            getline(filein,tmp);
            tmp += " ";
        }
    }
    filein.close();

    switch (fConfMode) {
    case -1:
        *error_code = 1u;
        Log << error << "Not properly initialised" << nline ;
        break;
    case 0:
        Log << info << "Init in BasicAFP mode" << nline ;
        break;
    case 1:
        for(unsigned int i=0 ; i<fListOfFile.size() ; i++)
        {
            std::ifstream afile(fListOfFile[i].data(), std::ios_base::binary);
            afile.seekg( 0 , std::ios_base::end );
            unsigned long size = afile.tellg() ;
            afile.close();
            TotalEntryFileSize += size;
        }
        break;
    case 2:
        Log << info << "List of file to be read is " << fListOfFile.size() << nline ;
        break;
    default:
        break;
    }

    // open the first file
    if ( *error_code == 0u )
        if ( !NewFile() )
            *error_code = 2u;

    Log << dolog;
}


void RootAFP::process_reset (UInt_t *error_code)
{
    *error_code = 0u;
    Log.ClearMessage(); Log.SetProcessMethod("process_reset");

    if ( fCurrentFile )
        ::fclose(fCurrentFile);
    fCurrentFileNumber = 0u;
    fConfMode = -1;

    GetFrameIO().Print( Log() );

    Log << dolog;
}
/*
void BasicAFP::process_unload (UInt_terror_code)
{
  std::cout << "process_unload called with GetPID()" << GetPID() << std::endl;
  *error_code = 0;
} */