source: trunk/FACT++/src/datalogger.cc@ 11218

//****************************************************************
/** @class DataLogger
  
  @brief Logs all message and infos between the services
  
  This is the main logging class facility. 
  It derives from StateMachineDim and DimInfoHandler. the first parent is here to enforce 
  a state machine behaviour, while the second one is meant to make the dataLogger receive
  dim services to which it subscribed from.
  The possible states and transitions of the machine are:
  \dot
  digraph datalogger {
          node [shape=record, fontname=Helvetica, fontsize=10];
      e [label="Error" color="red"];
   r [label="Ready"]
   d [label="NightlyOpen"]
   w [label="WaitingRun"]
      l [label="Logging"]
   b [label="BadNightlyconfig" color="red"]
   c [label="BadRunConfig" color="red"]
  
  e -> r
  r -> e
  r -> d
  r -> b
  d -> w
  d -> r
  w -> r
  l -> r
  l -> w
  b -> d
  w -> c
  w -> l
  b -> r
  c -> r
  c -> l
  }
  \enddot
 */
 //****************************************************************
#include "Dim.h"
#include "Event.h"
#include "Time.h"
#include "StateMachineDim.h"
#include "WindowLog.h"
#include "Configuration.h"
#include "ServiceList.h"
#include "Converter.h"
#include "MessageImp.h"
#include "LocalControl.h"
#include "DimDescriptionService.h"

#include "Description.h"

//#include "DimServiceInfoList.h"
#include "DimNetwork.h"
//for getting stat of opened files
#include <unistd.h>
//for getting disk free space
#include <sys/statvfs.h>
//for getting files sizes
#include <sys/stat.h>

#include <fstream>
#include <mutex>

#include <boost/bind.hpp>
#if BOOST_VERSION < 104400
#if (__GNUC__ > 4 || (__GNUC__ == 4 && __GNUC_MINOR__ > 4))
#undef BOOST_HAS_RVALUE_REFS
#endif
#endif
#include <boost/thread.hpp>

#ifdef HAVE_FITS
#include "Fits.h"
#endif

//Dim structures
///Distributes the writing statistics
struct DataLoggerStats {
    long sizeWritten;
    long freeSpace;
    long writingRate;
};
///distributes the number of opened subscriptions and fits files
struct NumSubAndFitsType {
    int numSubscriptions;
    int numOpenFits;
};
///distributes which files were opened.
struct OpenFileToDim {
    int code;
    char fileName[FILENAME_MAX];
};

///Run number record. Used to keep track of which run numbers are still active
struct RunNumberType {
#ifdef RUN_LOGS
    ///the run number log file
    shared_ptr<ofstream> logFile;
#endif
    ///the run number report file
    shared_ptr<ofstream> reportFile;
#ifdef HAVE_FITS
    ///the run number group fits file
    shared_ptr<CCfits::FITS> runFitsFile;
#endif
#ifdef RUN_LOGS
    ///the log filename
    string logName;
#endif
    ///the report filename
    string reportName;
    ///the actual run number
    uint32_t runNumber;
    ///the time at which the run number was received
    Time time;
    ///list of opened fits used to create the fits grouping when the run ends
    map<string, vector<string> > openedFits;
    ///default constructor
    RunNumberType()
    {
#ifdef RUN_LOGS
        logFile = shared_ptr<ofstream>(new ofstream());
#endif
        reportFile = shared_ptr<ofstream>(new ofstream());
#ifdef HAVE_FITS
        runFitsFile = shared_ptr<CCfits::FITS>();
#endif
        runNumber = 0;
    }
    ///default destructor
    ~RunNumberType()
    {

    }

    void addServiceToOpenedFits(const string& fileName, const string& serviceName)
    {
         //most likely I should add this service name.
         //the only case for which I should not add it is if a service disapeared, hence the file was closed
         //and reopened again. Unlikely to happen, but well it may

         if (find(openedFits[fileName].begin(), openedFits[fileName].end(),
                  serviceName)==openedFits[fileName].end())
            openedFits[fileName].push_back(serviceName);
    }
};
///Dim subscription type. Stores all the relevant info to handle a Dim subscription
struct SubscriptionType
{
#ifdef HAVE_FITS
    ///Nightly FITS output file
    Fits    nightlyFile;
    ///run-specific FITS output file
    Fits    runFile;
#endif
    ///the actual dimInfo pointer
    shared_ptr<DimStampedInfo> dimInfo;
    ///the server
    string server;
    ///the service
    string service;
    ///the converter for outputting the data according to the format
    shared_ptr<Converter> fConv;
    ///the current run number used by this subscription
    uint32_t runNumber;
    ///time of the latest received event
    Time lastReceivedEvent;
    ///whether or not the fits buffer was allocated already
    bool fitsBufferAllocated;

    ///Dim info constructor
    SubscriptionType(DimStampedInfo* info=NULL)
    {
        dimInfo = shared_ptr<DimStampedInfo>(info);
        fConv = shared_ptr<Converter>();
        runNumber = 0;
        lastReceivedEvent = Time::None;
        fitsBufferAllocated = false;
    }

    ///default destructor
    ~SubscriptionType()
    {
    }
};

class DataLogger : public StateMachineDim, DimServiceInfoList
{
public:
    /// The list of existing states specific to the DataLogger
    enum
    {
        kSM_NightlyOpen = 20, ///< Nightly file openned and writing
        kSM_WaitingRun = 30, ///< waiting for the run number to open the run file
        kSM_Logging = 40, ///< both files openned and writing
        kSM_BadNightlyConfig = 0x101, ///< the folder specified for Nightly logging does not exist or has bad permissions
        kSM_BadRunConfig = 0x102, ///<  the folder specified for the run logging does not exist or has wrong permissions or no run number
        kSM_WriteError = 0x103, ///< Denotes that an error occured while writing a file (text or fits).
    } localstates_t;
    
    DataLogger(ostream &out);
    ~DataLogger(); 

    bool SetConfiguration(Configuration& conf);

private:
    /************************************************
     * MEMBER VARIABLES
     ************************************************/
    /// ofstream for the NightlyLogfile
    ofstream fNightlyLogFile;
    /// ofstream for the Nightly report file
    ofstream fNightlyReportFile;
    /// base path of the Nightlyfile
    string fNightlyFilePath;
    ///base path of the run file
    string fRunFilePath;
    ///run numbers
    list<RunNumberType> fRunNumber;
    ///old run numbers time-out delay (in seconds)
    long fRunNumberTimeout;
    ///previous run number. to check if changed while logging
    int fPreviousRunNumber;
    ///Current Service Quality
    int fQuality;
    ///Modified Julian Date
    double fMjD;
    ///for obtaining the name of the existing services
//    ServiceList fServiceList;
    typedef map<const string, map<string, SubscriptionType> > SubscriptionsListType;
    ///All the services to which we have subscribed to, sorted by server name.
    SubscriptionsListType fServiceSubscriptions;
    ///full name of the nightly log file
    string fFullNightlyLogFileName;
    ///full name of the nightly report file
    string fFullNightlyReportFileName;
    ///variable to track when the statistic were last calculated
    Time fPreviousStatsUpdateTime;
    Time fPreviousOldRunNumberCheck;
    ///boolean to know whether we should close and reopen daily files or not
    bool fDailyFileDayChangedAlready;

private:
    /***************************************************
     * DIM INFO HANDLER
     ***************************************************/
    //overloading of DIM's infoHandler function
    void infoHandler(); 

    /***************************************************
     * TRANSITION FUNCTIONS
     ***************************************************/
    ///Reporting method for the services info received
    void ReportPlease(DimInfo* I, SubscriptionType& sub);  

    int ConfigureFileName(string &target, const string &type, const EventImp &evt);
    ///Configuration of the nightly file path
    int ConfigureNightlyFileName(const Event& evt); 
    ///Configuration fo the file name
    int ConfigureRunFileName(const Event& evt); 
    ///DEPREC - configuration of the run number
    int ConfigureRunNumber(const Event& evt); 
    ///print the current state of the dataLogger
    int PrintStatePlease(const Event& evt);
    ///checks whether or not the current info being treated is a run number
    void CheckForRunNumber(DimInfo* I);
    /// start transition
    int StartPlease(); 
    ///from waiting to logging transition
    int StartRunPlease(); 
    /// from logging to waiting transition
    int StopRunPlease(); 
    ///stop and reset transition
    int GoToReadyPlease(); 
    ///from NightlyOpen to waiting transition
    int NightlyToWaitRunPlease(); 
    /// from writing to error
    std::string SetCurrentState(int state, const char *txt="", const std::string &cmd="");
#ifdef HAVE_FITS
    ///Open fits files
    void OpenFITSFilesPlease(SubscriptionType& sub, RunNumberType* cRunNumber);
    ///Write data to FITS files
    void WriteToFITS(SubscriptionType& sub);
    ///Allocate the buffers required for fits
    void AllocateFITSBuffers(SubscriptionType& sub);
#endif//has_fits

    /***************************************
     * DIM SERVICES PROVIDED BY THE DATA LOGGER
     ***************************************/
    ///monitoring notification loop
    void ServicesMonitoring();
    inline void NotifyOpenedFile(const string &name, int type, DimDescribedService* service);
    ///variables for computing statistics
    DataLoggerStats fStatVar;
    ///mutex to make sure that the Stats are not accessed while updating
//    mutex fStatsMutex;
    ///services notification thread
//    boost::thread fMonitoringThread;
    ///end of the monitoring
//    bool fContinueMonitoring;
    ///stores the size of each file that is or was open
    map<string, long> fFileSizesMap;
    ///total size of the opened files BEFORE they were opened by the logger
    long fBaseSizeNightly;
    long fPreviousSize;
    long fBaseSizeRun;
    ///Service for opened files
    DimDescribedService* fOpenedNightlyFiles;
    DimDescribedService* fOpenedRunFiles;
    DimDescribedService* fNumSubAndFits;
    DimDescribedService* fStatsMonitoring;
    NumSubAndFitsType fNumSubAndFitsData;
    ///Small function for calculating the total size written so far
    bool calculateTotalSizeWritten(DataLoggerStats& statVar, bool isPrinting);

    /***************************************************
     * DATA LOGGER's CONFIGURATION STUFF
     ***************************************************/
    ///black/white listing
    set<string> fBlackList;
    set<string> fWhiteList;
    ///list of services to be grouped
    set<string> fGrouping;
    ///configuration flags
    bool fDebugIsOn;
    float fStatsPeriodDuration;
    bool fOpenedFilesIsOn;
    bool fNumSubAndFitsIsOn;
    //functions for controlling the services behavior
    int SetDebugOnOff(const Event& evt);
    int SetStatsPeriod(const Event& evt);
    int SetOpenedFilesOnOff(const Event& evt);
    int SetNumSubsAndFitsOnOff(const Event& evt);
    int SetRunTimeoutDelay(const Event& evt);

    ///boolean to prevent DIM update while desctructing the dataLogger
    bool fDestructing;    
    /***************************************************
     * UTILITIES
     ***************************************************/
    ///vectors to keep track of opened Fits files, for grouping purposes.
    map<string, vector<string> > fOpenedNightlyFits;
    ///creates a group fits file based on a list of files to be grouped
    void CreateFitsGrouping(map<string, vector<string> >& filesToGroup, int runNumber);

    bool OpenStream(shared_ptr<ofstream> stream, const string &filename);
    ///Open the relevant text files related to a particular run
    int OpenRunFile(RunNumberType& run);
    ///add a new run number
    void AddNewRunNumber(int64_t newRun, Time time);
    ///removes the oldest run number, and close the relevant files.
    void RemoveOldestRunNumber();
    ///retrieves the size of a file
    off_t GetFileSize(const string&);
    ///Get the digits of year, month and day for filenames and paths
    void GetYearMonthDayForFiles(unsigned short& year, unsigned short& month, unsigned short& day);
    ///Appends the relevant year month day to a given path
    void AppendYearMonthDaytoPath(string& path);
    ///Form the files path
    string CompileFileNameWithPath(const string &path, const string &service, const string & extension, const Time &time=Time());
    ///Form the file names only
    string CompileFileName(const string& service, const string& extension, const Time& time=Time());
    ///Form the files path
    string CompileFileNameWithPath(const string &path, const uint32_t run, const string &service, const string & extension, const Time &time=Time());
    ///Form the file names only
    string CompileFileName(const uint32_t run, const string& service, const string& extension);//, const Time& time=Time());
    ///Check whether service is in black and/or white list
    bool ShouldSubscribe(const string& server, const string& service);
    ///Subscribe to a given server and service
    DimStampedInfo* SubscribeToPlease(const string& server, const string& service);
    ///Open a text file and checks for ofstream status
    bool OpenTextFilePlease(ofstream& stream, const string& name);
    ///Check if a dir is . and returns the actual string corresponding to .
//    string CheckIfDirIsDot(const string& dir);
    ///Remembers the size of newly opened files. for statistic purposes
    bool RememberFileOrigSizePlease(string& fileName, bool nightly);
    ///Checks if the input osftream is in error state, and if so close it.
    bool CheckForOfstreamError(ofstream& out);
    ///Goes to Write error state and also closes all opened files
    void GoToWriteErrorState();
    ///Checks if a given path exist
    bool DoesPathExist(string path);
    ///Check if the statistics service should be updated, and if so, do it
    void UpdateStatisticsService();
    ///Check if old run numbers can be trimmed, and if so, do it
    void TrimOldRunNumbers();
    ///Create a given directory
    bool CreateDirectory(string path);
    /***************************************************
    * INHERITED FROM DIMSERVICEINFOLIST
    ***************************************************/
    ///Add a new service subscription
    void AddService(const string&, const string&, const string&, bool);
    ///Remove a given service subscription
    void RemoveService(const string&, const string&, bool);
    ///Remove all the services associated with a given server
    void RemoveAllServices(const string&);
}; //DataLogger

// --------------------------------------------------------------------------
//
//! Check if a given path exists
//! @param path the path to be checked
//! @return whether or not the creation has been successfull
//
bool DataLogger::CreateDirectory(string path)
{
    //remove last '/', if present
    if (path[path.size()-1] == '/')
        path = path.substr(0, path.size()-1);

    //create boost path
    const boost::filesystem::path fullPath =  boost::filesystem::system_complete(boost::filesystem::path(path));

    //if path does not exist, check if upper levels exist already
    if (boost::filesystem::exists(fullPath))
    {
        //if path already exist, make sure it does not designate a file (filenames cannot be checked if they do not exist)
        if (boost::filesystem::is_directory(fullPath))
            return true;

        Error("Path to be created contains a file name: '" + path + "'");
        return false;
    }

    if (path.size() <= 1)
    {//we're hitting "/", which SHOULD have existed...
        Error("Something unexpected happened while creating a path");
    }
    CreateDirectory(path.substr(0, path.find_last_of('/')));

    //path does not exist, and upper level have been created already by recusrion.
    const mode_t rightsMask = S_IRWXU | S_IXGRP | S_IRGRP | S_IXOTH | S_IROTH; //everybody read, owner writes

    const int returnValue = mkdir(path.c_str(), rightsMask);

    if (returnValue != 0)
    {
        ostringstream str;
        str << "Could not create directory " << path << " mkdir error code: " << errno;
        Error(str.str());
        return false;
    }

    return true;
}
// --------------------------------------------------------------------------
//
//! Check if a given path exists
//! @param path the path to be checked
//! @return whether or not the given path exists
//
bool DataLogger::DoesPathExist(string path)
{
    const boost::filesystem::path fullPath = boost::filesystem::system_complete(boost::filesystem::path(path));

    if (!boost::filesystem::exists(fullPath))
       return false;

    if (!boost::filesystem::is_directory(fullPath))
    {
        Error("Path given for checking '" + path + "' designate a file name. Please provide a path name only");
        return false;
    }

    if (access(path.c_str(), R_OK|W_OK|X_OK) != 0)
    {
        Error("Missing read, write or execute permissions on directory '" + path + "'");
        return false;
    }

    return true;
}
// --------------------------------------------------------------------------
//
//! Add a new service subscription
//! @param server the server for which the subscription should be created
//! @param service the service for which the subscription should be created
//! @param isCmd whether this is a Dim Command or not. Commands are not logged
//
void DataLogger::AddService(const string& server, const string& service, const string&, bool isCmd)
{
         //dataLogger does not subscribe to commands
    if (isCmd)
        return;

    //check the given subscription against black and white lists
    if (!ShouldSubscribe(server, service))
        return;

    map<string, SubscriptionType> &list = fServiceSubscriptions[server];

    if (list.find(service) != list.end())
    {
        Error("Service " + server + "/" + service + " is already in the dataLogger's list. ignoring its update.");
        return;
    }

    list[service].dimInfo = shared_ptr<DimStampedInfo>(SubscribeToPlease(server, service));
    list[service].server  = server;
    list[service].service = service;
    fNumSubAndFitsData.numSubscriptions++;
    if (fDebugIsOn)
        Debug("Added subscription to " + server + "/" + service);
}
// --------------------------------------------------------------------------
//
//! Remove a given service subscription
//! @param server the server for which the subscription should be removed
//! @param service the service that should be removed
//! @param isCmd whether or not this is a command
//
void DataLogger::RemoveService(const string& server, const string& service, bool isCmd)
{
    if (isCmd)
        return;

    if (fServiceSubscriptions[server].erase(service) != 1)
    {
        //check the given subscription against black and white lists
        if (!ShouldSubscribe(server, service))
            return;

        ostringstream str;
        str << "Subscription " << server << "/" << service << " could not be removed as it is not present";
        Error(str.str());
        return;
    }
    fNumSubAndFitsData.numSubscriptions--;
    if (fDebugIsOn)
    {
        Debug("Removed subscription to " + server + "/" + service);
    }
}
// --------------------------------------------------------------------------
//
//! Remove all the services associated with a given server
//! @param server the server for which all the services should be removed
//
void DataLogger::RemoveAllServices(const string& server)
{
    fNumSubAndFitsData.numSubscriptions -= fServiceSubscriptions[server].size();
    fServiceSubscriptions[server].clear();
    if (fDebugIsOn)
    {
        Debug("Removed all subscriptions to " + server + "/");
    }
}
// --------------------------------------------------------------------------
//
//! Checks if the given ofstream is in error state and if so, close it
//! @param out the ofstream that should be checked
//
bool DataLogger::CheckForOfstreamError(ofstream& out)
{
    if (out.good())
        return true;

    Error("An error occured while writing to a text file. Closing it");
//    if (out.is_open())
//        out.close();
    GoToWriteErrorState();

    return false;
//    SetCurrentState(kSM_WriteError);
}
// --------------------------------------------------------------------------
//
//! Checks the size on disk of a given size, and remembers it in the relevant member variable
//! @param fileName the file for which the size on disk should be retrieved
//! @param nightly whether this is a run or nightly file, so that its size is added to the correct member variable
//
bool DataLogger::RememberFileOrigSizePlease(string& fileName, bool nightly)
{
    //get the size of the file we're about to open
    if (fFileSizesMap.find(fileName) != fFileSizesMap.end())
        return false;

    if (nightly)
        fBaseSizeNightly += GetFileSize(fileName);
    else
        fBaseSizeRun += GetFileSize(fileName);
    fFileSizesMap[fileName] = 0;
    return true;
}

// --------------------------------------------------------------------------
//
//! Open a text file and checks for error code
//! @param stream the ofstream for which the file should be opened
//! @name the file name
//
bool DataLogger::OpenTextFilePlease(ofstream& stream, const string& name)
{
    Info("Opening: "+name);

    errno = 0;
    stream.open(name.c_str(), ios_base::out | ios_base::app);
    if (!stream)
    {
        ostringstream str;
        str << "Trying to open file " << name << ": " << strerror(errno) << " (errno=" << errno << ")";
        Error(str);
        return false;
    }

    return true;
}

// --------------------------------------------------------------------------
//
//! Create a new dim subscription to a given server and service
//! @param server the server name
//! @param service the service name
//
DimStampedInfo* DataLogger::SubscribeToPlease(const string& server, const string& service)
{
    if (fDebugIsOn)
    {
        ostringstream str;
        str << "Subscribing to service " << server << "/" << service;
        Debug(str);
    }
    return new DimStampedInfo((server + "/" + service).c_str(), (void*)NULL, 0, this);
}
// --------------------------------------------------------------------------
//
//! Check whether a service should be subscribed to, based on the black/white list entries
//! @param server the server name associated with the service being checked
//! @param service the service name associated with the service being checked
//
bool DataLogger::ShouldSubscribe(const string& server, const string& service)
{
    if (fWhiteList.size()>0 &&
        (fWhiteList.find(server + "/") == fWhiteList.end()) &&
        (fWhiteList.find(server + "/" + service) == fWhiteList.end()) &&
        (fWhiteList.find("/" + service) == fWhiteList.end()))
        return false;

    if ((fBlackList.find(server + "/") != fBlackList.end()) ||
         (fBlackList.find(server + "/" + service) != fBlackList.end()) ||
         (fBlackList.find("/" + service) != fBlackList.end()))
        return false;

    return true;
}
// --------------------------------------------------------------------------
//
//! Compiles a file name
//! @param path the base path where to put the file
//! @param time the time at which the file is created
//! @param service the service name, if any
//! @param extension the extension to add, if any
//
//string DataLogger::CompileFileName(const string &path, const string &service, const string & extension, const Time &time)
string DataLogger::CompileFileName(const string& service, const string& extension, const Time& time)
{
    ostringstream str;
    //calculate time suitable for naming path.
    const Time ftime(time-boost::posix_time::time_duration(12,0,0));

    //output base of file name
    str << Time::fmt("%Y_%m_%d") << ftime;

    //output service name
    if (!service.empty())
        str  << "_" << service;

    //output appropriate extension
    if (!extension.empty())
        str << "." << extension;

    return str.str();
}

string DataLogger::CompileFileNameWithPath(const string& path, const string& service, const string& extension, const Time& time)
{
    ostringstream str;
    //calculate time suitable for naming files.
    const Time ftime(time-boost::posix_time::time_duration(12,0,0));

    //output it
    str << path << Time::fmt("/%Y/%m/%d") << ftime;

    //check if target directory exist
    if (!DoesPathExist(str.str()))
        CreateDirectory(str.str());

    str << '/' << CompileFileName(service, extension, time);

    return str.str();


}
// --------------------------------------------------------------------------
//
//! Compiles a file name
//! @param path the base path where to put the file
//! @param time the time at which the file is created
//! @param run the run number
//! @param service the service name, if any
//! @param extension the extension to add, if any
//
//string DataLogger::CompileFileName(const string &path, uint32_t run, const string &service, const string & extension, const Time &time)
string DataLogger::CompileFileName(const uint32_t run, const string& service, const string& extension)//, const Time& time)
{
       ostringstream str;
       //output base of file name
       str << setfill('0') << setw(8) << run;

       //output service name
       if (!service.empty())
           str  << "_" << service;

       //output appropriate extension
       if (!extension.empty())
           str << "." << extension;
       return str.str();
}

string DataLogger::CompileFileNameWithPath(const string& path, const uint32_t run, const string& service, const string& extension, const Time& time)
{
    ostringstream str;
    //calculate suitable time for naming files and output it
    str << path << Time::fmt("/%Y/%m/%d") << (time-boost::posix_time::time_duration(12,0,0));

    //check if target directory exist
    if (!DoesPathExist(str.str()))
        CreateDirectory(str.str());

    str << '/' << CompileFileName(run, service, extension);//, time);

    return str.str();

}
// --------------------------------------------------------------------------
//
//!retrieves the size on disk of a file
//! @param fileName the full file name for which the size on disk should be retrieved
//! @return the size of the file on disk, in bytes. 0 if the file does not exist or if an error occured
//
off_t DataLogger::GetFileSize(const string& fileName)
{
    errno = 0;
    struct stat st;
    if (!stat(fileName.c_str(), &st))
        return st.st_size;

    if (errno != 0 && errno != 2)//ignoring error #2: no such file or directory is not an error for new files
    {
        ostringstream str;
        str << "Unable to stat " << fileName << ". Reason: " << strerror(errno) << " [" << errno << "]";
        Error(str);
    }

    return 0;
}
// --------------------------------------------------------------------------
//
//! Removes the oldest run number and closes the fits files that should be closed
//! Also creates the fits grouping file
//
void DataLogger::RemoveOldestRunNumber()
{
    if (fDebugIsOn)
    {
        ostringstream str;
        str << "Removing run number " << fRunNumber.front().runNumber;
        Debug(str);
    }
    CreateFitsGrouping(fRunNumber.front().openedFits, fRunNumber.front().runNumber);

    //crawl through the subscriptions to see if there are still corresponding fits files opened.
    for (SubscriptionsListType::iterator x=fServiceSubscriptions.begin();
         x!=fServiceSubscriptions.end(); x++)
        for (map<string, SubscriptionType>::iterator y=x->second.begin();
             y!=x->second.end(); y++)
            if (y->second.runFile.fRunNumber == fRunNumber.front().runNumber && y->second.runFile.IsOpen())
            {
                y->second.runFile.Close();

                Info("Closed: "+y->second.runFile.fFileName);
            }
    //if a grouping file is on, decrease the number of opened fits manually
    if (fRunNumber.front().runFitsFile)
        (fNumSubAndFitsData.numOpenFits)--;
    //remove the entry
    fRunNumber.pop_front();
}

// --------------------------------------------------------------------------
//
//! Calculate the total number of written bytes since the logger was started
//! @param statVar the data structure that should be updated
//! @param shouldWarn whether or not error messages should be outputted
//! @param isPrinting whether this function was called from the PRINT command or not. If so, displays relevant information
//
bool DataLogger::calculateTotalSizeWritten(DataLoggerStats& statVar, bool isPrinting)
{
//mutex
//    if (!isPrinting)
//        fStatsMutex.lock();
#ifdef HAVE_FITS
    if (isPrinting)
    {
        ostringstream str;
        str << "There are " << fNumSubAndFitsData.numOpenFits << " FITS files open:";
        Message(str);
    }

    ///TODO the grouping file is dealt with several times. This should not be a problem but well, better to fix it I guess.
    for (SubscriptionsListType::const_iterator x=fServiceSubscriptions.begin();
         x!=fServiceSubscriptions.end(); x++)
    {
        for (map<string, SubscriptionType>::const_iterator y=x->second.begin();
             y!=x->second.end(); y++)
        {
            if (y->second.runFile.IsOpen())
            {
                fFileSizesMap[y->second.runFile.fFileName] = y->second.runFile.GetWrittenSize();
                if (isPrinting)
                    Message("-> "+y->second.runFile.fFileName);
            }
            if (y->second.nightlyFile.IsOpen())
            {
                fFileSizesMap[y->second.nightlyFile.fFileName] = y->second.nightlyFile.GetWrittenSize();
                if (isPrinting)
                    Message("-> "+y->second.nightlyFile.fFileName);
            }
        }
    }
#else
    if (isPrinting)
        Message("FITS output disabled at compilation");
#endif
    //gather log and report files sizes on disk
    if (fNightlyLogFile.is_open())
        fFileSizesMap[fFullNightlyLogFileName] = GetFileSize(fFullNightlyLogFileName);
    if (fNightlyReportFile.is_open())
        fFileSizesMap[fFullNightlyReportFileName] = GetFileSize(fFullNightlyReportFileName);
    for (list<RunNumberType>::iterator it = fRunNumber.begin(); it != fRunNumber.end(); it++)
    {
        if (it->reportFile->is_open())
            fFileSizesMap[it->reportName] = GetFileSize(it->reportName);
#ifdef RUN_LOGS
        if (it->logFile->is_open())
            fFileSizesMap[it->logName] = GetFileSize(it->logName);
#endif
    }

    bool shouldWarn = false;
    struct statvfs vfs;
    if (!statvfs(fNightlyFilePath.c_str(), &vfs))
        statVar.freeSpace = vfs.f_bsize*vfs.f_bavail;
    else
    {
        ostringstream str;
        str << "Unable to retrieve stats for " << fNightlyFilePath << ". Reason: " << strerror(errno) << " [" << errno << "]";
        if (!shouldWarn)
            Error(str);
        statVar.freeSpace = -1;
    }
    //sum up all the file sizes. past and present
    statVar.sizeWritten = 0;
    for (map<string, long>::const_iterator it=fFileSizesMap.begin(); it != fFileSizesMap.end();  it++)
        statVar.sizeWritten += it->second;
    statVar.sizeWritten -= fBaseSizeNightly;
    statVar.sizeWritten -= fBaseSizeRun;

//mutex
//    if (!isPrinting)
//        fStatsMutex.unlock();

    return shouldWarn;
}

// --------------------------------------------------------------------------
//
//! Monitor the number of opened files and total size written, and distributes this data through a Dim service
//
//
/*
void DataLogger::ServicesMonitoring()
{
       struct statvfs vfs;
        if (!statvfs(fNightlyFilePath.c_str(), &vfs))
            fStatVar.freeSpace = vfs.f_bsize*vfs.f_bavail;
        else
            fStatVar.freeSpace = -1;

        DimDescribedService srvc ("DATA_LOGGER/STATS", "X:3", fStatVar, "Add description here");
        fPreviousSize = 0;
        //loop-wait for broadcast
        while (fContinueMonitoring)
        {
            if (fStatsPeriodDuration == 0.0f)
            {
                sleep(0.1f);
                continue;
            }

            sleep(fStatsPeriodDuration);


            fStatsMutex.lock();

            if (fStatVar.writingRate != 0) //if data has been written
            {
                srvc.updateService();

                if(fDebugIsOn)
                {
                    ostringstream str;
                    str << "Size written: " << fStatVar.sizeWritten/1000 << " kB; writing rate: ";
                    str << fStatVar.writingRate/1000 << " kB/s; free space: ";
                    str << fStatVar.freeSpace/(1000*1000) << " MB";
                    Debug(str);
                }
            }
            fStatsMutex.unlock();
        }
}
*/
// --------------------------------------------------------------------------
//
//! Default constructor. The name of the machine is given DATA_LOGGER
//! and the state is set to kSM_Ready at the end of the function.
//
//!Setup the allows states, configs and transitions for the data logger
//
DataLogger::DataLogger(ostream &out) : StateMachineDim(out, "DATA_LOGGER")
{
    //initialize member data
    fNightlyFilePath = ".";
    fRunFilePath = ".";

    //Give a name to this machine's specific states
    AddStateName(kSM_NightlyOpen,      "NightlyFileOpen",  "The summary files for the night are open.");
    AddStateName(kSM_WaitingRun,       "WaitForRun",       "The summary files for the night are open and we wait for a run to be started.");
    AddStateName(kSM_Logging,          "Logging",          "The summary files for the night and the files for a single run are open.");
    AddStateName(kSM_BadNightlyConfig, "ErrNightlyFolder", "The folder for the nighly summary files is invalid.");
    AddStateName(kSM_BadRunConfig,     "ErrRunFolder",     "The folder for the run files is invalid.");
    AddStateName(kSM_WriteError,       "ErrWrite",          "An error occured while writing to a file.");

    // Add the possible transitions for this machine
    AddEvent(kSM_NightlyOpen, "START", kSM_Ready, kSM_BadNightlyConfig)
        (boost::bind(&DataLogger::StartPlease, this))
        ("Start the nightly logging. Nightly file location must be specified already");

    AddEvent(kSM_Ready, "STOP", kSM_NightlyOpen, kSM_WaitingRun, kSM_Logging, kSM_WriteError)
        (boost::bind(&DataLogger::GoToReadyPlease, this))
        ("Stop all data logging, close all files.");

    AddEvent(kSM_Logging, "START_RUN", kSM_WaitingRun, kSM_BadRunConfig)
        (boost::bind(&DataLogger::StartRunPlease, this))
        ("Start the run logging. Run file location must be specified already.");

    AddEvent(kSM_WaitingRun, "STOP_RUN", kSM_Logging)
        (boost::bind(&DataLogger::StopRunPlease, this))
        ("Wait for a run to be started, open run-files as soon as a run number arrives.");

    AddEvent(kSM_Ready, "RESET", kSM_Error, kSM_BadNightlyConfig, kSM_BadRunConfig, kSM_WriteError)
        (boost::bind(&DataLogger::GoToReadyPlease, this))
        ("Transition to exit error states. Closes the any open file.");

    AddEvent(kSM_WaitingRun, "WAIT_FOR_RUN_NUMBER", kSM_NightlyOpen)
        (boost::bind(&DataLogger::NightlyToWaitRunPlease, this))
        ("Go to waiting for run number state. In this state with any received run-number a new file is opened.");

    // Add the possible configurations for this machine
    AddEvent("SET_NIGHTLY_FOLDER", "C", kSM_Ready, kSM_BadNightlyConfig)
        (boost::bind(&DataLogger::ConfigureNightlyFileName, this, _1))
        ("Configure the base folder for the nightly files."
         "|Path[string]:Absolute or relative path name where the nightly files should be stored.");

    AddEvent("SET_RUN_FOLDER", "C", kSM_Ready, kSM_BadNightlyConfig, kSM_NightlyOpen, kSM_WaitingRun, kSM_BadRunConfig)
        (boost::bind(&DataLogger::ConfigureRunFileName, this, _1))
        ("Configure the base folder for the run files."
         "|Path[string]:Absolute or relative path name where the run files should be stored.");

    AddEvent("SET_RUN_NUMBER", "X", kSM_Ready, kSM_NightlyOpen, kSM_WaitingRun, kSM_BadRunConfig, kSM_Logging)
        (boost::bind(&DataLogger::ConfigureRunNumber, this, _1))
        ("Configure the run number. Cannot be done in logging state");

     // Provide a print command
     AddEvent("PRINT")
            (boost::bind(&DataLogger::PrintStatePlease, this, _1))
            ("Print information about the internal status of the data logger.");

     OpenFileToDim fToDim;
     fToDim.code = 0;
     fToDim.fileName[0] = '\0';

     fOpenedNightlyFiles = new DimDescribedService(GetName() + "/FILENAME_NIGHTLY", "I:1;C", fToDim,
                               "Path and base name which is used to compile the filenames for the nightly files."
                               "|Type[int]:type of open files (1=log, 2=rep, 4=fits)"
                               "|Name[string]:path and base file name");

     fOpenedRunFiles = new DimDescribedService(GetName() + "/FILENAME_RUN", "I:1;C", fToDim,
                               "Path and base name which is used to compile the filenames for the run files."
                               "|Type[int]:type of open files (1=log, 2=rep, 4=fits)"
                               "|Name[string]:path and base file name");

     fNumSubAndFitsData.numSubscriptions = 0;
     fNumSubAndFitsData.numOpenFits = 0;
     fNumSubAndFits = new DimDescribedService(GetName() + "/NUM_SUBS", "I:2", fNumSubAndFitsData,
                               "Shows number of services to which the data logger is currently subscribed and the total number of open files."
                               "|Subscriptions[int]:number of dim services to which the data logger is currently subscribed."
                               "|NumOpenFiles[int]:number of files currently open by the data logger");

     //services parameters
     fDebugIsOn = false;
     fStatsPeriodDuration = 1.0f;
     fOpenedFilesIsOn = true;
     fNumSubAndFitsIsOn = true;

     // provide services control commands
     AddEvent("SET_DEUG_MODE", "B:1", kSM_NightlyOpen, kSM_Logging, kSM_WaitingRun, kSM_Ready)
         (boost::bind(&DataLogger::SetDebugOnOff, this, _1))
         ("Switch debug mode on or off. Debug mode prints ifnormation about every service written to a file."
          "|Enable[bool]:Enable of disable debug mode (yes/no).");

     AddEvent("SET_STATISTICS_UPDATE_INTERVAL", "F", kSM_NightlyOpen, kSM_Logging, kSM_WaitingRun, kSM_Ready)
         (boost::bind(&DataLogger::SetStatsPeriod, this, _1))
         ("Interval in which the data-logger statistics service (STATS) is updated."
          "|Interval[s]:Floating point value in seconds.");

     AddEvent("ENABLE_FILENAME_SERVICES", "B:1", kSM_NightlyOpen, kSM_Logging, kSM_WaitingRun, kSM_Ready)
         (boost::bind(&DataLogger::SetOpenedFilesOnOff ,this, _1))
         ("Switch service which distributes information about the open files on or off."
          "|Enable[bool]:Enable of disable filename services (yes/no).");

     AddEvent("ENABLE_NUMSUBS_SERVICE", "B:1", kSM_NightlyOpen, kSM_Logging, kSM_WaitingRun, kSM_Ready)
         (boost::bind(&DataLogger::SetNumSubsAndFitsOnOff, this, _1))
         ("Switch the service which distributes information about the number of subscriptions and open files on or off."
          "|Enable[bool]:Enable of disable NUM_SUBS service (yes/no).");

     AddEvent("SET_RUN_TIMEOUT", "L:1", kSM_Ready, kSM_NightlyOpen, kSM_Logging, kSM_WaitingRun)
         (boost::bind(&DataLogger::SetRunTimeoutDelay, this, _1))
         ("Set the timeout delay for old run numbers."
          "|timeout[min]:Time out in minutes after which files for expired runs are closed.");

     fDestructing = false;

     //start the monitoring service
     fStatVar.sizeWritten = 0;
     fStatVar.freeSpace = 0;
     fStatVar.writingRate = 0;
     fPreviousStatsUpdateTime = Time().Mjd();
     fPreviousOldRunNumberCheck = Time().Mjd();
     fPreviousSize = 0;

     struct statvfs vfs;
     if (!statvfs(fNightlyFilePath.c_str(), &vfs))
         fStatVar.freeSpace = vfs.f_bsize*vfs.f_bavail;
     else
         fStatVar.freeSpace = -1;

     fStatsMonitoring = new DimDescribedService(GetName() + "/STATS", "X:3", fStatVar, "Add description here");

//mutex
//     fContinueMonitoring = true;
//     fMonitoringThread = boost::thread(boost::bind(&DataLogger::ServicesMonitoring, this));
     fBaseSizeNightly = 0;
     fBaseSizeRun = 0;
     fDailyFileDayChangedAlready = true;
     fRunNumberTimeout = 60; //default run-timeout set to 1 minute
     if(fDebugIsOn)
     {
         Debug("DataLogger Init Done.");
     }
}

// --------------------------------------------------------------------------
//
//! Destructor
//
DataLogger::~DataLogger()
{
    if (fDebugIsOn)
    {
        Debug("DataLogger destruction starts");    
    }
    fDestructing = true;
    //first let's go to the ready state
    GoToReadyPlease(); 
    //release the services subscriptions
    fServiceSubscriptions.clear();
    //exit the monitoring loop
//mutex
//    fContinueMonitoring = false;
//    fMonitoringThread.join();
    //clear any remaining run number (should remain only one)
     while (fRunNumber.size() > 0)
     {
         RemoveOldestRunNumber();
     }

    delete fOpenedNightlyFiles;
    delete fOpenedRunFiles;
    delete fNumSubAndFits;

    if (fDebugIsOn)
    {
        Debug("DataLogger desctruction ends");    
    }
}
// --------------------------------------------------------------------------
//
//! checks if the statistic service should be updated, and if so, do it
//
void DataLogger::UpdateStatisticsService()
{
    //update the fits files sizes
    const Time cTime = Time();

    if ((fStatsPeriodDuration == 0) || ((cTime - fPreviousStatsUpdateTime).total_seconds() < fStatsPeriodDuration))
        return;

    calculateTotalSizeWritten(fStatVar, false);
    fStatVar.writingRate = (fStatVar.sizeWritten - fPreviousSize)/((cTime - fPreviousStatsUpdateTime).total_seconds());
    fPreviousSize = fStatVar.sizeWritten;
    fPreviousStatsUpdateTime = cTime;
    //update the service. No need to check if data has been written, because some must have been, otherwise we would not have hit this piece of code
    fStatsMonitoring->updateService();

    if(fDebugIsOn)
    {
        ostringstream str;
        str << "Size written: " << fStatVar.sizeWritten/1000 << " kB; writing rate: ";
        str << fStatVar.writingRate/1000 << " kB/s; free space: ";
        str << fStatVar.freeSpace/(1000*1000) << " MB";
        Debug(str);
    }
}
// --------------------------------------------------------------------------
//
//! checks if old run numbers should be trimmed and if so, do it
//
void DataLogger::TrimOldRunNumbers()
{
    const Time cTime = Time();

    if ((cTime - fPreviousOldRunNumberCheck).total_seconds() < fRunNumberTimeout)
        return;

    while (fRunNumber.size() > 1 && (cTime - fRunNumber.back().time) > boost::posix_time::seconds(fRunNumberTimeout))
    {
         RemoveOldestRunNumber();
    }
    fPreviousOldRunNumberCheck = cTime;
}
// --------------------------------------------------------------------------
//
//! Inherited from DimInfo. Handles all the Infos to which we subscribed, and log them
//
void DataLogger::infoHandler()
{
    // Make sure getTimestamp is called _before_ getTimestampMillisecs
    if (fDestructing)
        return;

    DimInfo* I = getInfo();

    if (I==NULL)
        return;
    //check if the service pointer corresponds to something that we subscribed to
    //this is a fix for a bug that provides bad Infos when a server starts
    bool found = false;
    SubscriptionsListType::iterator x;
    map<string, SubscriptionType>::iterator y;
     for (x=fServiceSubscriptions.begin(); x != fServiceSubscriptions.end(); x++)
    {//find current service is subscriptions
        for (y=x->second.begin(); y!=x->second.end();y++)
            if ((y->second.dimInfo).get() == I)
            {
                found = true;    
                break;
            }
        if (found)
            break;
    }
    if (!found)
    {
        DimServiceInfoList::infoHandler();
        return;
    }
    if (I->getSize() <= 0 || I->getData()==NULL)
        return;

        // Make sure that getTimestampMillisecs is NEVER called before
        // getTimestamp is properly called
        // check that the message has been updated by something, i.e. must be different from its initial value
    if (I->getTimestamp() == 0)
        return;

    // FIXME: Here we have to check if we have received the
    //        service with the run-number.
    //        CheckForRunNumber(I); has been removed because we have to
    //        subscribe to this service anyway and hence we have the pointer
    //        (no need to check for the name)

    ReportPlease(I, y->second);

    //update the fits files sizes
    UpdateStatisticsService();

    //remove old run numbers
    TrimOldRunNumbers();
}

bool DataLogger::OpenStream(shared_ptr<ofstream> stream, const string &filename)
{
    Info("Opening: "+filename);

    if (stream->is_open())
    {
        ostringstream str;
        str << filename << " was already open when trying to open it.";
        Error(str);
        return false;
    }

    errno = 0;
    stream->open(filename.c_str(), ios_base::out | ios_base::app);
    if (errno != 0)
    {
        ostringstream str;
        str << "Unable to open " << filename << ": " << strerror(errno) << " (errno=" << errno << ")";
        Error(str);
        return false;
    }

    if (!stream->is_open())
    {
        ostringstream str;
        str << "File " << filename << " not open as it ought to be.";
        Error(str);
        return false;
    }

    return true;
}

// --------------------------------------------------------------------------
//
//! Open the text files associated with the given run number
//! @param run the run number to be dealt with
//
int DataLogger::OpenRunFile(RunNumberType& run)
{
#ifdef RUN_LOGS
    // open log file
    run.logName = CompileFileName(fRunFilePath, run.runNumber, "", "log");
    if (!OpenStream(run.logFile, run.logName))
        return -1;
#endif

    // open report file
    run.reportName = CompileFileNameWithPath(fRunFilePath, run.runNumber, "", "rep");
    if (!OpenStream(run.reportFile, run.reportName))
        return -1;

    //get the size of the newly opened file.
#ifdef RUN_LOGS
    RememberFileOrigSizePlease(run.logName, false);
#endif
    RememberFileOrigSizePlease(run.reportName, false);

    //TODO this notification scheme might be messed up now.... fix it !
    const string baseFileName = CompileFileNameWithPath(fRunFilePath, run.runNumber, "", "");
    NotifyOpenedFile(baseFileName, 3, fOpenedRunFiles);
    run.openedFits.clear();
    return 0;
}
// --------------------------------------------------------------------------
//
//! Add a new active run number
//! @param newRun the new run number
//! @param time the time at which the new run number was issued
//
void DataLogger::AddNewRunNumber(int64_t newRun, Time time)
{

    if (newRun > 0xffffffff)
    {
        Error("New run number too large, out of range. Ignoring.");
        return;
    }
    if (fDebugIsOn)
    {
        ostringstream str;
        str << "Adding new run number " << newRun << " that was issued on " << time;
        Debug(str);
    }
    //Add new run number to run number list
    fRunNumber.push_back(RunNumberType());
    fRunNumber.back().runNumber = uint32_t(newRun);
    fRunNumber.back().time = time;

    ostringstream str;
    str << "The new run number is: " << fRunNumber.back().runNumber;
    Message(str);

    if (GetCurrentState() != kSM_Logging)
        return;
    //open the log and report files
    if (OpenRunFile(fRunNumber.back()) != 0)
    {//an error occured. close current run files and go to error state
        for (list<RunNumberType>::iterator it=fRunNumber.begin(); it != fRunNumber.end(); it++)
        {
            if (it->reportFile->is_open())
                it->reportFile->close();
#ifdef RUN_LOGS
            if (it->logFile->is_open())
                it->logFile->close();
#endif
        }
        StopRunPlease();
        SetCurrentState(kSM_BadRunConfig);
    }
}
// --------------------------------------------------------------------------
//
//! Checks whether or not the current info is a run number.
//! If so, then remember it. A run number is required to open the run-log file
//! @param I
//!        the current DimInfo
//
void DataLogger::CheckForRunNumber(DimInfo* I)
{
    if (strstr(I->getName(), "SET_RUN_NUMBER") != NULL)
    {//assumes that the run number is an integer
        //check if some run number entries can be deleted leave one so that two remain after adding the new one
        AddNewRunNumber(I->getLonglong(), Time(I->getTimestamp(), I->getTimestampMillisecs()*1000));
    }
}

// --------------------------------------------------------------------------
//
//! write infos to log files.
//! @param I
//!     The current DimInfo 
//! @param sub
//!        The dataLogger's subscription corresponding to this DimInfo
//
void DataLogger::ReportPlease(DimInfo* I, SubscriptionType& sub)
{
    const string fmt(I->getFormat());

    const bool isItaReport = fmt!="C";

    if (!fNightlyReportFile.is_open())
        return;

    if (fDebugIsOn && string(I->getName())!="DATA_LOGGER/MESSAGE")
    {
        ostringstream str;
        str << "Logging " <<  I->getName() << " [" << I->getFormat() << "] (" << I->getSize() << ")";
        Debug(str);
    }

    //Check whether we should close and reopen daily text files or not
    //This should work in any case base of the following:
    // - fDailyFileDayChangedAlready is initialized to true. So if the dataLogger is started around noon, no file will be closed
    // - fDailyFileDayChangedAlready is set to false if (time != 12), so the file will be closed and reopened only if the logger runs since before noon (which is the required behavior)
    //This only applies to text files. Fits are closed and reopened based on the last and current service received time.
    //this was not applicable to text files, because as they gather several services, we have no guarantee that the received time will be greater than the previous one,
    //which could lead to several close/reopen instead of only one.
    if (Time().h() == 12 && !fDailyFileDayChangedAlready)
    {
        if (fDebugIsOn)
            Debug("Its Noon! Closing and reopening nightly text files");

        fNightlyLogFile.close();
        fNightlyReportFile.close();

        Info("Closed: "+fFullNightlyLogFileName);
        Info("Closed: "+fFullNightlyReportFileName);

        fFullNightlyLogFileName = CompileFileNameWithPath(fNightlyFilePath, "", "log");
        if (!OpenTextFilePlease(fNightlyLogFile, fFullNightlyLogFileName))
        {
            GoToReadyPlease();
            SetCurrentState(kSM_BadNightlyConfig);
            return;
        }

        fFullNightlyReportFileName = CompileFileNameWithPath(fNightlyFilePath, "", "rep");
        if (!OpenTextFilePlease(fNightlyReportFile, fFullNightlyReportFileName))
        {
            GoToReadyPlease();
            SetCurrentState(kSM_BadNightlyConfig);
            return;
        }

        fDailyFileDayChangedAlready = true;
    }
    if (Time().h() != 12 && fDailyFileDayChangedAlready)
        fDailyFileDayChangedAlready = false;

    //create the converter for that service
    if ((!sub.fConv.get()) && isItaReport)
    {
        sub.fConv = shared_ptr<Converter>(new Converter(Out(), I->getFormat()));
        if (!sub.fConv)
        {
            ostringstream str;
            str << "Couldn't properly parse the format... service " << sub.dimInfo->getName() << " ignored.";
            Error(str);
            return;    
        }
    }
    //construct the header
    ostringstream header;
    const Time cTime(I->getTimestamp(), I->getTimestampMillisecs()*1000);
    fQuality = I->getQuality();
    fMjD = cTime.Mjd();

    //figure out which run file should be used
    ofstream* targetRunFile = NULL;
    RunNumberType* cRunNumber = NULL;
    if (GetCurrentState() == kSM_Logging)
    {
        list<RunNumberType>::reverse_iterator rit;
        for (rit=fRunNumber.rbegin(); rit!=fRunNumber.rend(); rit++)
        {
            if (rit->time < cTime) //this is the run number that we want to use
            {
                //Find something better to convert iterator to pointer than the ugly line below....
                cRunNumber = &(*rit);
                sub.runNumber = rit->runNumber;
#ifdef RUN_LOGS
                targetRunFile = isItaReport ? (rit->reportFile).get() : (rit->logFile).get();
#else
                targetRunFile = isItaReport ? (rit->reportFile).get() : NULL;
#endif
                break;
            }
        }
        if (rit == fRunNumber.rend() && fRunNumber.size() != 0)
        {
            ostringstream str;
            str << "Could not find an appropriate run number for info coming at time: " << cTime;
            Error(str);
            Error("Active run numbers: ");
            for (rit=fRunNumber.rbegin(); rit != fRunNumber.rend(); rit++)
            {
                str.str("");
                str << rit->runNumber;
                Error(str);
            }

        }
    }

    if (isItaReport)
    {
        //write text header
        header << I->getName() << " " << fQuality << " ";
        header << cTime.Y() << " " << cTime.M() << " " << cTime.D() << " ";
        header << cTime.h() << " " << cTime.m() << " " << cTime.s() << " ";
        header << cTime.ms() << " " << I->getTimestamp() << " ";

        string text;
        try
        {
            text = sub.fConv->GetString(I->getData(), I->getSize());
        }
        catch (const runtime_error &e)
        {
            ostringstream str;
            str << "Parsing service " << sub.dimInfo->getName();
            str << " failed: " << e.what();
            Error(str);
            return;
        }

        if (text.empty())
        {
            ostringstream str;
            str << "Service " << sub.dimInfo->getName() << " sent an empty string";
            Info(str);
            return;
        }
        //replace bizarre characters by white space
        replace(text.begin(), text.end(), '\n', '\\');
        replace_if(text.begin(), text.end(), ptr_fun<int, int>(&iscntrl), ' ');
        
        //write entry to Nightly report
        if (fNightlyReportFile.is_open())
        {
            fNightlyReportFile << header.str() << text << endl;
            if (!CheckForOfstreamError(fNightlyReportFile))
                return;
        }
        //write entry to run-report
        if (targetRunFile && targetRunFile->is_open())
        {
            *targetRunFile << header.str() << text << endl;
            if (!CheckForOfstreamError(*targetRunFile))
                return;
        }
    }
    else
    {//write entry to both Nightly and run logs
        ostringstream msg;
        msg << I->getName() << ": " << I->getString();

        if (fNightlyLogFile.is_open())
        {
            MessageImp(fNightlyLogFile).Write(cTime, msg.str().c_str(), fQuality);
            if (!CheckForOfstreamError(fNightlyLogFile))
                return;
        }
        if (targetRunFile && targetRunFile->is_open())
        {
            MessageImp(*targetRunFile).Write(cTime, msg.str().c_str(), fQuality);
            if (!CheckForOfstreamError(*targetRunFile))
                return;
        }
    }

#ifdef HAVE_FITS
    if (isItaReport)
    {
        //check if the last received event was before noon and if current one is after noon.
        //if so, close the file so that it gets reopened.
        if (sub.nightlyFile.IsOpen())
            if ((sub.lastReceivedEvent != Time::None) && (sub.lastReceivedEvent.h() < 12) && (cTime.h() >= 12))
            {
                sub.nightlyFile.Close();
            }
        sub.lastReceivedEvent = cTime;
        if (!sub.nightlyFile.IsOpen() || !sub.runFile.IsOpen() || sub.runNumber != sub.runFile.fRunNumber)
            OpenFITSFilesPlease(sub, cRunNumber);
        WriteToFITS(sub);
    }    
#endif

}

// --------------------------------------------------------------------------
//
//! print the dataLogger's current state. invoked by the PRINT command
//! @param evt
//!        the current event. Not used by the method
//! @returns 
//!        the new state. Which, in that case, is the current state
//!
int DataLogger::PrintStatePlease(const Event& )
{
    Message("------------------------------------------");
    Message("------- DATA LOGGER CURRENT STATE --------");
    Message("------------------------------------------");

    //print the path configuration
    Message("Nightly Path: " + boost::filesystem::system_complete(boost::filesystem::path(fNightlyFilePath)).directory_string());
    Message("Run Path: " + boost::filesystem::system_complete(boost::filesystem::path(fRunFilePath)).directory_string());

    //print active run numbers
    ostringstream str;
    str << "Active Run Numbers:";
    for (list<RunNumberType>::const_iterator it=fRunNumber.begin(); it!=fRunNumber.end(); it++)
        str << " " << it->runNumber;
    if (fRunNumber.size()==0)
        str << " <none>";
    Message(str);
    //timeout value
    str.str("");
    str << "Timeout delay for old run numbers: " << fRunNumberTimeout << " seconds";
    Message(str);

    //print all the open files. 
    Message("------------ OPENED FILES ----------------");
    if (fNightlyLogFile.is_open())
        Message("Nightly log-file: OPEN");

    if (fNightlyReportFile.is_open())
        Message("Nightly report-file: OPEN");

    for (list<RunNumberType>::const_iterator it=fRunNumber.begin(); it!=fRunNumber.end(); it++)
    {
#ifdef RUN_LOGS
        if (it->logFile->is_open())
            Message("Run log-file: " + it->logName + " (OPEN)");
#endif
        if (it->reportFile->is_open())
            Message("Run report-file: " + it->reportName + " (OPEN)");
    }

    DataLoggerStats statVar;
    /*const bool statWarning =*/ calculateTotalSizeWritten(statVar, true);

    Message("----------------- STATS ------------------");
    str.str("");
    str << "Total Size written: " << statVar.sizeWritten << " bytes.";
        Message(str);
    str.str("");
    str << "Disk free space:    " << statVar.freeSpace   << " bytes.";
        Message(str);
    str.str("");
    str << "Statistics are updated every " << fStatsPeriodDuration << " seconds";
    if (fStatsPeriodDuration != 0)
        Message(str);
    else
        Message("Statistics updates are currently disabled");

    Message("------------ DIM SUBSCRIPTIONS -----------");
    str.str("");
    str << "There are " << fNumSubAndFitsData.numSubscriptions << " active DIM subscriptions.";
    Message(str);
    for (map<const string, map<string, SubscriptionType> >::const_iterator it=fServiceSubscriptions.begin(); it!= fServiceSubscriptions.end();it++)
    {
        Message("Server "+it->first);
        for (map<string, SubscriptionType>::const_iterator it2=it->second.begin(); it2!=it->second.end(); it2++)
            Message(" -> "+it2->first);
    }
    Message("--------------- BLOCK LIST ---------------");
    for (set<string>::const_iterator it=fBlackList.begin(); it != fBlackList.end(); it++)
        Message(" -> "+*it);
    if (fBlackList.size()==0)
        Message(" <empty>");

    Message("--------------- ALLOW LIST ---------------");
    for (set<string>::const_iterator it=fWhiteList.begin(); it != fWhiteList.end(); it++)
        Message(" -> "+*it);
    if (fWhiteList.size()==0)
        Message(" <empty>");

    Message("-------------- GROUPING LIST -------------");
    Message("The following servers and/or services will");
    Message("be grouped into a single fits file:");
    for (set<string>::const_iterator it=fGrouping.begin(); it != fGrouping.end(); it++)
        Message(" -> "+*it);
    if (fGrouping.size()==0)
        Message(" <no grouping>");

    Message("------------------------------------------");
    Message("-------- END OF DATA LOGGER STATE --------");
    Message("------------------------------------------");

    return GetCurrentState();
}

// --------------------------------------------------------------------------
//
//! turn debug mode on and off
//! @param evt
//!        the current event. contains the instruction string: On, Off, on, off, ON, OFF, 0 or 1
//! @returns 
//!        the new state. Which, in that case, is the current state
//!
int DataLogger::SetDebugOnOff(const Event& evt)
{
    const bool backupDebug = fDebugIsOn;

    fDebugIsOn = evt.GetBool();

    if (fDebugIsOn == backupDebug)
        Message("Debug mode was already in the requested state.");

    ostringstream str;
    str << "Debug mode is now " << fDebugIsOn;
    Message(str);

    return GetCurrentState();
}
// --------------------------------------------------------------------------
//
//! set the statistics update period duration. 0 disables the statistics
//! @param evt
//!        the current event. contains the new duration.
//! @returns 
//!        the new state. Which, in that case, is the current state
//!
int DataLogger::SetStatsPeriod(const Event& evt)
{
    const float backupDuration = fStatsPeriodDuration;

    fStatsPeriodDuration = evt.GetFloat();

    if (fStatsPeriodDuration < 0)
    {
        Error("Statistics period duration should be greater than zero. Discarding provided value.");
        fStatsPeriodDuration = backupDuration;
        return GetCurrentState();    
    }
    if (!finite(fStatsPeriodDuration))// != fStatsPeriodDuration)
    {
        Error("Provided duration does not appear to be a valid float. Discarding it.");
        fStatsPeriodDuration = backupDuration;
        return GetCurrentState();    
    }
    if (backupDuration == fStatsPeriodDuration)
        Warn("Statistics period not modified. Supplied value already in use.");

    if (fStatsPeriodDuration == 0.0f)
        Message("Statistics are now OFF");
    else
    {
        ostringstream str;
        str << "Statistics period is now " << fStatsPeriodDuration << " seconds";
        Message(str);
    }

    return GetCurrentState();
}
// --------------------------------------------------------------------------
//
//! set the opened files service on or off. 
//! @param evt
//!        the current event. contains the instruction string. similar to setdebugonoff
//! @returns 
//!        the new state. Which, in that case, is the current state
//!
int DataLogger::SetOpenedFilesOnOff(const Event& evt)
{
    const bool backupOpened = fOpenedFilesIsOn;

    fOpenedFilesIsOn = evt.GetBool();

    if (fOpenedFilesIsOn == backupOpened)
        Message("Opened files service mode was already in the requested state.");

    ostringstream str;
    str << "Opened files service mode is now " << fOpenedFilesIsOn;
    Message(str);

    return GetCurrentState();
}

// --------------------------------------------------------------------------
//
//! set the number of subscriptions and opened fits on and off
//! @param evt
//!        the current event. contains the instruction string. similar to setdebugonoff
//! @returns 
//!        the new state. Which, in that case, is the current state
//!
int DataLogger::SetNumSubsAndFitsOnOff(const Event& evt)
{
    const bool backupSubs = fNumSubAndFitsIsOn;

    fNumSubAndFitsIsOn = evt.GetBool();

    if (fNumSubAndFitsIsOn == backupSubs)
        Message("Number of subscriptions service mode was already in the requested state");

    ostringstream str;
    str << "Number of subscriptions service mode is now " << fNumSubAndFitsIsOn;
    Message(str);

    return GetCurrentState();
}
// --------------------------------------------------------------------------
//
//! set the timeout delay for old run numbers
//! @param evt
//!        the current event. contains the timeout delay long value
//! @returns
//!        the new state. Which, in that case, is the current state
//!
int DataLogger::SetRunTimeoutDelay(const Event& evt)
{
    const long backupTimeout = fRunNumberTimeout;
    fRunNumberTimeout = evt.GetXtra();

    if (fRunNumberTimeout == 0)
    {
        fRunNumberTimeout = backupTimeout;
        Error("Timeout delays for old run numbers must be greater than 0. Ignored.");
        return GetCurrentState();
    }

    if (fRunNumberTimeout == backupTimeout)
        Message("New timeout for old run numbers is same value as previous one.");

    ostringstream str;
    str  << "Timeout delay for old run numbers is now " << fRunNumberTimeout << " seconds";
    Message(str);

    return GetCurrentState();
}

int DataLogger::ConfigureFileName(string &target, const string &type, const EventImp &evt)
{
    if (!evt.GetText())
    {
        Error("Empty "+type+" folder given. Please specify a valid path.");
        return GetCurrentState();
    }

    const string givenPath = evt.GetText();
    if (!DoesPathExist(givenPath))
    {
        Error("Provided "+type+" path '"+givenPath+"' is not a valid folder. Ignored");
        return GetCurrentState();
    }

    Message("New "+type+" folder: "+givenPath);

    target = givenPath;

    return GetCurrentState();
}

// --------------------------------------------------------------------------
//
//!    Sets the path to use for the Nightly log file.
//! @param evt
//!     the event transporting the path
//! @returns
//!        currently only the current state.
//
int DataLogger::ConfigureNightlyFileName(const Event& evt)
{
    return ConfigureFileName(fNightlyFilePath, "nightly", evt);
}

// --------------------------------------------------------------------------
//
//! Sets the path to use for the run log file.
//! @param evt
//!        the event transporting the path
//! @returns
//!     currently only the current state
int DataLogger::ConfigureRunFileName(const Event& evt)
{
    return ConfigureFileName(fRunFilePath, "run", evt);
}

// --------------------------------------------------------------------------
//
//! Sets the run number.
//! @param evt
//!        the event transporting the run number
//! @returns
//!     currently only the current state
int DataLogger::ConfigureRunNumber(const Event& evt)
{
    AddNewRunNumber(evt.GetXtra(), evt.GetTime());
    return GetCurrentState();
}
// --------------------------------------------------------------------------
//
//! Notifies the DIM service that a particular file was opened
//! @ param name the base name of the opened file, i.e. without path nor extension. 
//!     WARNING: use string instead of string& because I pass values that do not convert to string&.
//!        this is not a problem though because file are not opened so often.
//! @ param type the type of the opened file. 0 = none open, 1 = log, 2 = text, 4 = fits
inline void DataLogger::NotifyOpenedFile(const string &name, int type, DimDescribedService* service)
{
    if (!fOpenedFilesIsOn)
        return;

    if (fDebugIsOn)
    {
        ostringstream str;
        str << "Updating " << service->getName() << " file '" << name << "' (type=" << type << ")";
        Debug(str);

        str.str("");
        str << "Num subscriptions: " << fNumSubAndFitsData.numSubscriptions << " Num open FITS files: " << fNumSubAndFitsData.numOpenFits;
        Debug(str);
    }

    if (name.size()+1 > FILENAME_MAX)
    {
        Error("Provided file name '" + name + "' is longer than allowed file name length.");
        return;
    }

    OpenFileToDim fToDim;
    fToDim.code = type;
    memcpy(fToDim.fileName, name.c_str(), name.size()+1);

    service->setData(reinterpret_cast<void*>(&fToDim), name.size()+1+sizeof(int));
    service->setQuality(0);
    service->updateService();
}
// --------------------------------------------------------------------------
//
//! Implements the Start transition.
//! Concatenates the given path for the Nightly file and the filename itself (based on the day), 
//! and tries to open it.
//! @returns 
//!        kSM_NightlyOpen if success, kSM_BadNightlyConfig if failure
int DataLogger::StartPlease()
{
    if (fDebugIsOn)
    {
        Debug("Starting...");    
    }
    fFullNightlyLogFileName = CompileFileNameWithPath(fNightlyFilePath, "", "log");
    if (!OpenTextFilePlease(fNightlyLogFile, fFullNightlyLogFileName))
        return kSM_BadNightlyConfig;


    fFullNightlyReportFileName = CompileFileNameWithPath(fNightlyFilePath, "", "rep");
    if (!OpenTextFilePlease(fNightlyReportFile, fFullNightlyReportFileName))
    {
        fNightlyLogFile.close();
        return kSM_BadNightlyConfig;
    }
    //get the size of the newly opened file.
    fBaseSizeNightly = GetFileSize(fFullNightlyLogFileName);
    fBaseSizeNightly += GetFileSize(fFullNightlyReportFileName);
    fFileSizesMap.clear();
    fBaseSizeRun = 0;
    fPreviousSize = 0;

    //notify that a new file has been opened.
    const string baseFileName = CompileFileNameWithPath(fNightlyFilePath, "", "");
    NotifyOpenedFile(baseFileName, 3, fOpenedNightlyFiles);

    fOpenedNightlyFits.clear();
    
    return kSM_NightlyOpen;     
}

#ifdef HAVE_FITS
// --------------------------------------------------------------------------
//
//! open if required a the FITS files corresponding to a given subscription
//! @param sub
//!     the current DimInfo subscription being examined
void DataLogger::OpenFITSFilesPlease(SubscriptionType& sub, RunNumberType* cRunNumber)
{
    string serviceName(sub.dimInfo->getName());

    //if run number has changed, reopen a new fits file with the correct run number.
     if (sub.runFile.IsOpen() && sub.runFile.fRunNumber != sub.runNumber)
     {
         sub.runFile.Close();
         Info("Closed: "+sub.runFile.fFileName+" (new run number)");
     }

     //we must check if we should group this service subscription to a single fits file before we replace the / by _
    bool hasGrouping = false;
    if (!sub.runFile.IsOpen() && (GetCurrentState() == kSM_Logging))
    {//will we find this service in the grouping list ?
        for (set<string>::const_iterator it=fGrouping.begin(); it!=fGrouping.end(); it++)
        {
            if (serviceName.find(*it) != string::npos)
            {
                hasGrouping = true;
                break;
            }
        }
    }
    for (unsigned int i=0;i<serviceName.size(); i++)
    {
        if (serviceName[i] == '/')
        {
            serviceName[i] = '_';
            break;    
        }    
    }
    //we open the NightlyFile anyway, otherwise this function shouldn't have been called.
    if (!sub.nightlyFile.IsOpen())
    {
        string partialName = CompileFileNameWithPath(fNightlyFilePath, serviceName, "fits");

        const string fileNameOnly = partialName.substr(partialName.find_last_of('/')+1, partialName.size());
        if (!sub.fitsBufferAllocated)
            AllocateFITSBuffers(sub);
        //get the size of the file we're about to open
        if (RememberFileOrigSizePlease(partialName, true))//and remember that the file was opened (i.e. not an update)
            fOpenedNightlyFits[fileNameOnly].push_back(serviceName);

        ostringstream str;
        str << "Opening: " << partialName << " (Nfits=" << fNumSubAndFitsData.numOpenFits << ")";
        Info(str);

        if (!sub.nightlyFile.Open(partialName, serviceName, NULL, &fNumSubAndFitsData.numOpenFits, this, 0))
        {
            GoToWriteErrorState();
            //SetCurrentState(kSM_WriteError);
            return;
        }
        //notify the opening
        const string baseFileName = CompileFileNameWithPath(fNightlyFilePath, "", "");
        NotifyOpenedFile(baseFileName, 7, fOpenedNightlyFiles);
        if (fNumSubAndFitsIsOn)
            fNumSubAndFits->updateService();
    }
    //do the actual file open
    if (!sub.runFile.IsOpen() && (GetCurrentState() == kSM_Logging) && sub.runNumber != 0)
    {//buffer for the run file have already been allocated when doing the Nightly file
        string fileNameOnly;
        string partialName;
        if (hasGrouping)
        {
            partialName = CompileFileNameWithPath(fRunFilePath, sub.runNumber, "", "fits");
            fileNameOnly = partialName.substr(partialName.find_last_of('/')+1, partialName.size());
        }
        else
        {
            partialName = CompileFileNameWithPath(fRunFilePath, sub.runNumber, serviceName, "fits");
            fileNameOnly = partialName.substr(partialName.find_last_of('/')+1, partialName.size());
        }
        //get the size of the file we're about to open
        if (RememberFileOrigSizePlease(partialName, false))//and remember that the file was opened (i.e. not an update)
            cRunNumber->openedFits[fileNameOnly].push_back(serviceName);
        else
            if (hasGrouping)
            {
             cRunNumber->addServiceToOpenedFits(fileNameOnly, serviceName);
            }

        if (hasGrouping && (!cRunNumber->runFitsFile.get()))
            try
            {
                cRunNumber->runFitsFile = shared_ptr<CCfits::FITS>(new CCfits::FITS(partialName, CCfits::RWmode::Write));
                (fNumSubAndFitsData.numOpenFits)++;
            }    
            catch (CCfits::FitsException e)
            {
                ostringstream str;
                str << "Open FITS file " << partialName << ": " << e.message();
                Error(str);
                cRunNumber->runFitsFile = shared_ptr<CCfits::FITS>();//NULL;
                GoToWriteErrorState();
                //SetCurrentState(kSM_WriteError);
                return;
            }

        const string baseFileName = CompileFileNameWithPath(fRunFilePath, sub.runNumber, "", "");
        NotifyOpenedFile(baseFileName, 7, fOpenedRunFiles);// + '_' + serviceName, 4);

        ostringstream str;
        str << "Opening: " << partialName << " (Nfits=" << fNumSubAndFitsData.numOpenFits << ")";
        Info(str);

        if (hasGrouping)
        {
            if (!sub.runFile.Open(partialName, serviceName, (cRunNumber->runFitsFile).get(), &fNumSubAndFitsData.numOpenFits, this, sub.runNumber))
            {
                GoToWriteErrorState();
                //SetCurrentState(kSM_WriteError);
                return;
            }
        }
        else
        {
            if (sub.runFile.Open(partialName, serviceName, NULL, &fNumSubAndFitsData.numOpenFits, this, sub.runNumber))
            {
                GoToWriteErrorState();
                //SetCurrentState(kSM_WriteError);
                return;
            }
        }
        if (fNumSubAndFitsIsOn)
            fNumSubAndFits->updateService();
    }
}    
// --------------------------------------------------------------------------
//
//! Allocates the required memory for a given pair of fits files (nightly and run)
//! @param sub the subscription of interest.
//
void DataLogger::AllocateFITSBuffers(SubscriptionType& sub)
{
    //Init the time columns of the file
    Description dateDesc(string("Time"), string("Modified Julian Date"), string("MjD"));
    sub.nightlyFile.AddStandardColumn(dateDesc, "1D", &fMjD, sizeof(double));
    sub.runFile.AddStandardColumn(dateDesc, "1D", &fMjD, sizeof(double));

    Description QoSDesc("Qos", "Quality of service", "None");
    sub.nightlyFile.AddStandardColumn(QoSDesc, "1J", &fQuality, sizeof(int));
    sub.runFile.AddStandardColumn(QoSDesc, "1J", &fQuality, sizeof(int));

    const Converter::FormatList flist = sub.fConv->GetList();
    // Compilation failed
    if (!sub.fConv->valid())
    {
        Error("Compilation of format string failed.");
        return;
    }

    //we've got a nice structure describing the format of this service's messages.
    //Let's create the appropriate FITS columns
    int size = 0;

    vector<string> dataFormatsLocal;
    for (unsigned int i=0;i<flist.size()-1;i++)
    {
         ostringstream dataQualifier;

         dataQualifier << flist[i].second.first;
         switch (flist[i].first.first->name()[0])
         {
         case 'c': dataQualifier << "B"; break; // FIXME: To be checked!
         case 's': dataQualifier << "I"; break;
         case 'i': dataQualifier << "J"; break;
         case 'l': dataQualifier << "J"; break;
         case 'f': dataQualifier << "E"; break;
         case 'd': dataQualifier << "D"; break;
         case 'x': dataQualifier << "K"; break;
         case 'S': //we skip the variable length strings
             continue;

         default:
             Fatal("THIS SHOULD NEVER BE REACHED.");
         };

         size += flist[i].first.second * flist[i].second.first;
         dataFormatsLocal.push_back(dataQualifier.str());
     }
     sub.nightlyFile.InitDataColumns(GetDescription(sub.server, sub.service), dataFormatsLocal, sub.dimInfo->getData(), size);
     sub.runFile.InitDataColumns(GetDescription(sub.server, sub.service), dataFormatsLocal, sub.dimInfo->getData(), size);
     sub.fitsBufferAllocated = true;
}
// --------------------------------------------------------------------------
//
//! write a dimInfo data to its corresponding FITS files
//
void DataLogger::WriteToFITS(SubscriptionType& sub)
{
        //nightly File status (open or not) already checked
        if (sub.nightlyFile.IsOpen())
        {
            if (!sub.nightlyFile.Write(sub.fConv.get()))
            {
                GoToWriteErrorState();
                return;
              //SetCurrentState(kSM_WriteError);
            }
         }

        if (sub.runFile.IsOpen())
        {
            if (!sub.runFile.Write(sub.fConv.get()))
            {
                GoToWriteErrorState();
                return;
                //SetCurrentState(kSM_WriteError);
            }
        }
}
#endif //if has_fits

void DataLogger::GoToWriteErrorState()
{
    GoToReadyPlease();
    SetCurrentState(kSM_WriteError);
}

std::string DataLogger::SetCurrentState(int state, const char *txt, const std::string &cmd)
{
//    if (state == kSM_WriteError && GetCurrentState() == kSM_WriteError)
//        return "";
    return StateMachineImp::SetCurrentState(state, txt, cmd);
}
// --------------------------------------------------------------------------
//
//! Implements the StartRun transition.
//! Concatenates the given path for the run file and the filename itself (based on the run number),
//! and tries to open it.
//! @returns
//!        kSM_Logging if success, kSM_BadRunConfig if failure.
int DataLogger::StartRunPlease()
{
    if (fDebugIsOn)
    {
        Debug("Starting Run Logging...");    
    }
    //open all the relevant run-files. i.e. all the files associated with run numbers.
    for (list<RunNumberType>::iterator it=fRunNumber.begin(); it != fRunNumber.end(); it++)
        if (OpenRunFile(*it) != 0)
        {
            StopRunPlease();
            return kSM_BadRunConfig;
        }

    return kSM_Logging;
}

#ifdef HAVE_FITS
// --------------------------------------------------------------------------
//
//! Create a fits group file with all the run-fits that were written (either daily or run)
//! @param filesToGroup a map of filenames mapping to table names to be grouped (i.e. a
//!        single file can contain several tables to group
//! @param runNumber the run number that should be used for grouping. 0 means nightly group
//
void DataLogger::CreateFitsGrouping(map<string, vector<string> > & filesToGroup, int runNumber)
{
    if (fDebugIsOn)
    {
        ostringstream str;
        str << "Creating fits group for ";
        if (runNumber != 0)
            str << "run files";
        else
            str << "nightly files";
        Debug(str);
    }
    //create the FITS group corresponding to the ending run.
    CCfits::FITS* groupFile;
    unsigned int numFilesToGroup = 0;
    unsigned int maxCharLength = 0;
    for (map<string, vector<string> >::const_iterator it=filesToGroup.begin(); it != filesToGroup.end(); it++)
    {
        //add the number of tables in this file to the total number to group
        numFilesToGroup += it->second.size();
        //check the length of all the strings to be written, to determine the max string length to write
        if (it->first.size() > maxCharLength)
            maxCharLength = it->first.size();
        for (vector<string>::const_iterator jt=it->second.begin(); jt != it->second.end(); jt++)
            if (jt->size() > maxCharLength)
                maxCharLength = jt->size();
    }

    if (fDebugIsOn)
    {
        ostringstream str;
        str << "There are " << numFilesToGroup << " tables to group";
        Debug(str);
    }
    if (numFilesToGroup <= 1)
    {
        filesToGroup.clear();
        return;
    }
    string groupName;
    if (runNumber != 0)
        groupName = CompileFileNameWithPath(fRunFilePath, runNumber, "", "fits");
    else
        groupName = CompileFileNameWithPath(fNightlyFilePath, "", "fits");

    Info("Creating FITS group in: "+groupName);

    CCfits::Table* groupTable;
//    const int maxCharLength = FILENAME_MAX;
    try
    {
        groupFile = new CCfits::FITS(groupName, CCfits::RWmode::Write);
        //setup the column names
        ostringstream pathTypeName;
        pathTypeName << maxCharLength << "A";
        vector<string> names;
        vector<string> dataTypes;
        names.push_back("MEMBER_XTENSION");
        dataTypes.push_back("8A");
        names.push_back("MEMBER_URI_TYPE");
        dataTypes.push_back("3A");
        names.push_back("MEMBER_LOCATION");
        dataTypes.push_back(pathTypeName.str());
        names.push_back("MEMBER_NAME");
        dataTypes.push_back(pathTypeName.str());

        groupTable = groupFile->addTable("GROUPING", numFilesToGroup, names, dataTypes);
//TODO handle the case when the logger was stopped and restarted during the same day, i.e. the grouping file must be updated
     }
     catch (CCfits::FitsException e)
     {
         ostringstream str;
         str << "Creating FITS table GROUPING in " << groupName << ": " << e.message();
         Error(str);
         return;
     }

    //CCfits seems to be buggy somehow: can't use the column's function "write": it create a compilation error: maybe strings were not thought about.
    //use cfitsio routines instead
    groupTable->makeThisCurrent();
    //create appropriate buffer.
    const unsigned int n = 8 + 3 + 2*maxCharLength + 1; //+1 for trailling character

    vector<unsigned char> realBuffer;
    realBuffer.resize(n);
    unsigned char* fitsBuffer = &realBuffer[0];
    memset(fitsBuffer, 0, n);

    char* startOfExtension = reinterpret_cast<char*>(fitsBuffer);
    char* startOfURI       = reinterpret_cast<char*>(&fitsBuffer[8]);
    char* startOfLocation  = reinterpret_cast<char*>(&fitsBuffer[8 + 3]);
    char* startOfName      = reinterpret_cast<char*>(&fitsBuffer[8+3+maxCharLength]);

    strcpy(startOfExtension, "BINTABLE");
    strcpy(startOfURI,       "URL");

    int i=1;
    for (map<string, vector<string> >::const_iterator it=filesToGroup.begin(); it!=filesToGroup.end(); it++)
        for (vector<string>::const_iterator jt=it->second.begin(); jt != it->second.end(); jt++, i++)
        {
            strcpy(startOfLocation, it->first.c_str());
            strcpy(startOfName, jt->c_str());

            if (fDebugIsOn)
            {
                ostringstream str;
                str << "Grouping " << it->first << " " << *jt;
                Debug(str);
            }

            int status = 0;
            fits_write_tblbytes(groupFile->fitsPointer(), i, 1, 8+3+2*maxCharLength, fitsBuffer, &status);
            if (status)
            {
                char text[30];//max length of cfitsio error strings (from doc)
                fits_get_errstatus(status, text);
                ostringstream str;
                str << "Writing FITS row " << i << " in " << groupName << ": " << text << " (file_write_tblbytes, rc=" << status << ")";
                Error(str);
                GoToWriteErrorState();
                return;
            }
        }

    filesToGroup.clear();
    delete groupFile;
}
#endif //HAVE_FITS

// --------------------------------------------------------------------------
//
//! Implements the StopRun transition.
//! Attempts to close the run file.
//! @returns
//!        kSM_WaitingRun if success, kSM_FatalError otherwise
int DataLogger::StopRunPlease()
{

    if (fDebugIsOn)
    {
        Debug("Stopping Run Logging...");    
    }
    for (list<RunNumberType>::const_iterator it=fRunNumber.begin(); it != fRunNumber.end(); it++)
    {
#ifdef RUN_LOGS
        if (!it->logFile->is_open() || !it->reportFile->is_open())
#else
        if (!it->reportFile->is_open())
#endif
            return kSM_FatalError;
#ifdef RUN_LOGS
        it->logFile->close();
#endif
        it->reportFile->close();
    }

#ifdef HAVE_FITS
    for (SubscriptionsListType::iterator i = fServiceSubscriptions.begin(); i != fServiceSubscriptions.end(); i++)
        for (map<string, SubscriptionType>::iterator j = i->second.begin(); j != i->second.end(); j++)
        {
            if (j->second.runFile.IsOpen())
                j->second.runFile.Close();
        }
#endif
    NotifyOpenedFile("", 0, fOpenedRunFiles);
    if (fNumSubAndFitsIsOn)
        fNumSubAndFits->updateService();

    while (fRunNumber.size() > 0)
    {
        RemoveOldestRunNumber();
    }

    return kSM_WaitingRun;
}
// --------------------------------------------------------------------------
//
//! Implements the Stop and Reset transitions.
//! Attempts to close any openned file.
//! @returns
//!     kSM_Ready
int DataLogger::GoToReadyPlease()
{
   if (fDebugIsOn)
   {
        Debug("Going to the Ready state...");
   }
   if (GetCurrentState() == kSM_Logging)
       StopRunPlease();

    if (fNightlyLogFile.is_open())
        fNightlyLogFile.close();
    if (fNightlyReportFile.is_open())
        fNightlyReportFile.close();
        
#ifdef HAVE_FITS
    for (SubscriptionsListType::iterator i = fServiceSubscriptions.begin(); i != fServiceSubscriptions.end(); i++)
        for (map<string, SubscriptionType>::iterator j = i->second.begin(); j != i->second.end(); j++)
        {
            if (j->second.nightlyFile.IsOpen())
                j->second.nightlyFile.Close();;
        }
#endif
    if (GetCurrentState() == kSM_Logging || 
        GetCurrentState() == kSM_WaitingRun || 
        GetCurrentState() == kSM_NightlyOpen)
    { 
        NotifyOpenedFile("", 0, fOpenedNightlyFiles);
        if (fNumSubAndFitsIsOn)
            fNumSubAndFits->updateService();
    }
#ifdef HAVE_FITS
    CreateFitsGrouping(fOpenedNightlyFits, 0);
#endif
    return kSM_Ready;
}

// --------------------------------------------------------------------------
//
//! Implements the transition towards kSM_WaitingRun
//! Does nothing really.
//!    @returns
//!        kSM_WaitingRun
int DataLogger::NightlyToWaitRunPlease()
{
    if (fDebugIsOn)
    {
        Debug("Going to Wait Run Number state...");    
    }
    return kSM_WaitingRun;    
}
// --------------------------------------------------------------------------
//
//! Setup Logger's configuration from a Configuration object
//! @param conf the configuration object that should be used
//!
bool DataLogger::SetConfiguration(Configuration& conf)
{
    fDebugIsOn = conf.Get<bool>("debug");

    //Set the block or allow list
    fBlackList.clear();
    fWhiteList.clear();

    //Adding entries that should ALWAYS be ignored
    //fBlackList.insert("DATA_LOGGER/");
    fBlackList.insert("/SERVICE_LIST");
    fBlackList.insert("DIS_DNS/");

    //set the black list
    if (conf.Has("block"))
    {
        const vector<string> vec = conf.Get<vector<string>>("block");

        fBlackList.insert(vec.begin(), vec.end());
    }

    //set the white list
    if (conf.Has("allow"))
    {
        const vector<string> vec = conf.Get<vector<string>>("allow");
        fWhiteList.insert(vec.begin(), vec.end());
    }

    //Set the grouping
    if (conf.Has("group"))
    {
        const vector<string> vec = conf.Get<vector<string>>("group");
        fGrouping.insert(vec.begin(), vec.end());
    }

    //set the old run numbers timeout delay
    if (conf.Has("runtimeout"))
    {
        const long timeout = conf.Get<long>("runtimeout");
        if (timeout <= 0)
        {
            Error("Time out delay for old run numbers should be greater than 0 minute");
            return false;
        }
        fRunNumberTimeout = timeout;
    }
    return true;
}

// --------------------------------------------------------------------------
int RunDim(Configuration &conf)
{
    WindowLog wout;

    ReadlineColor::PrintBootMsg(wout, conf.GetName(), false);

    //log.SetWindow(stdscr);
    if (conf.Has("log"))
        if (!wout.OpenLogFile(conf.Get<string>("log")))
            wout << kRed << "ERROR - Couldn't open log-file " << conf.Get<string>("log") << ": " << strerror(errno) << endl;

    // Start io_service.Run to use the StateMachineImp::Run() loop
    // Start io_service.run to only use the commandHandler command detaching
    DataLogger logger(wout);
    if (!logger.SetConfiguration(conf))
        return -1;

    logger.Run(true);

    return 0;
}
// --------------------------------------------------------------------------
void RunThread(DataLogger* logger)
{
    // This is necessary so that the StateMachine Thread can signal the 
    // Readline thread to exit
    logger->Run(true);
    Readline::Stop();    
}
// --------------------------------------------------------------------------
template<class T>
int RunShell(Configuration &conf)
{
    static T shell(conf.GetName().c_str(), conf.Get<int>("console")!=1);

    WindowLog &win  = shell.GetStreamIn();
    WindowLog &wout = shell.GetStreamOut();

    if (conf.Has("log"))
        if (!wout.OpenLogFile(conf.Get<string>("log")))
            win << kRed << "ERROR - Couldn't open log-file " << conf.Get<string>("log") << ": " << strerror(errno) << endl;

    DataLogger logger(wout);

    if (!logger.SetConfiguration(conf))
        return -1;
    
    shell.SetReceiver(logger);

    boost::thread t(boost::bind(RunThread, &logger));

    if (conf.Has("cmd"))
    {
        const vector<string> v = conf.Get<vector<string>>("cmd");
        for (vector<string>::const_iterator it=v.begin(); it!=v.end(); it++)
            shell.ProcessLine(*it);
    }

    if (conf.Has("exec"))
    {
        const vector<string> v = conf.Get<vector<string>>("exec");
        for (vector<string>::const_iterator it=v.begin(); it!=v.end(); it++)
            shell.Execute(*it);
    }

    if (conf.Get<bool>("quit"))
        shell.Stop();

    
    shell.Run(); // Run the shell
    
    logger.Stop();
    
    //Wait until the StateMachine has finished its thread
    //before returning and destroyinng the dim objects which might 
    //still be in use.
    t.join();

    return 0;
}

/*
 Extract usage clause(s) [if any] for SYNOPSIS.
 Translators: "Usage" and "or" here are patterns (regular expressions) which
 are used to match the usage synopsis in program output.  An example from cp
 (GNU coreutils) which contains both strings:
  Usage: cp [OPTION]... [-T] SOURCE DEST
    or:  cp [OPTION]... SOURCE... DIRECTORY
    or:  cp [OPTION]... -t DIRECTORY SOURCE...
 */
void PrintUsage()
{
    cout << "\n"
        "The data logger connects to all available Dim services and "
        "writes them to ascii and fits files.\n"
        "\n"
        "The default is that the program is started without user interaction. "
        "All actions are supposed to arrive as DimCommands. Using the -c "
        "option, a local shell can be initialized. With h or help a short "
        "help message about the usage can be brought to the screen.\n"
        "\n"
        "Usage: datalogger [-c type] [OPTIONS]\n"
        "  or:  datalogger [OPTIONS]\n";
    cout << endl;

}
// --------------------------------------------------------------------------
void PrintHelp()
{
    /* Additional help text which is printed after the configuration
     options goes here */
    cout <<
        "\n"
        "If the allow list has any element, only the servers and/or services "
        "specified in the list will be used for subscription. The black list "
        "will disable service subscription and has higher priority than the "
        "allow list. If the allow list is not present by default all services "
        "will be subscribed."
        "\n"
        "For example, block=DIS_DNS/ will skip all the services offered by "
        "the DIS_DNS server, while block=/SERVICE_LIST will skip all the "
        "SERVICE_LIST services offered by any server and DIS_DNS/SERVICE_LIST "
        "will skip DIS_DNS/SERVICE_LIST.\n"
        << endl;
}

// --------------------------------------------------------------------------
void SetupConfiguration(Configuration &conf)
{
    const string n = conf.GetName()+".log";

    po::options_description configp("Program options");
    configp.add_options()
        ("dns",       var<string>("localhost"),  "Dim nameserver host name (Overwites DIM_DNS_NODE environment variable)")
        ("log,l",     var<string>(n), "Write log-file")
        ("console,c", var<int>(),     "Use console (0=shell, 1=simple buffered, X=simple unbuffered)")
        ("cmd",       vars<string>(), "Execute one or more commands at startup")
        ("exec,e",    vars<string>(), "Execute one or more scrips at startup")
        ("quit,q",    po_switch(),    "Quit after startup");
        ;

    po::options_description configs("DataLogger options");
    configs.add_options()
        ("block,b", vars<string>(), "Black-list of services")
        ("allow,a", vars<string>(), "White-list of services")
        ("debug",   po_bool(),      "Debug mode. Print clear text of received service reports to log-stream")
        ("group,g", vars<string>(), "Grouping of services into a single run-Fits")
        ("runtimeout", var<long>(), "Time out delay for old run numbers")
        ;

    conf.AddEnv("dns", "DIM_DNS_NODE");

    conf.AddOptions(configp);
    conf.AddOptions(configs);
}
// --------------------------------------------------------------------------
int main(int argc, const char* argv[])
{
    Configuration conf(argv[0]);
    conf.SetPrintUsage(PrintUsage);
    SetupConfiguration(conf);

    po::variables_map vm;
    try
    {
        vm = conf.Parse(argc, argv);
    }
#if BOOST_VERSION > 104000
    catch (po::multiple_occurrences &e)
    {
        cerr << "Program options invalid due to: " << e.what() << " of '" << e.get_option_name() << "'." << endl;
        return -1;
    }
#endif
    catch (exception& e)
    {
        cerr << "Program options invalid due to: " << e.what() << endl;
        return -1;
    }

    if (conf.HasVersion() || conf.HasPrint())
        return -1;

    if (conf.HasHelp())
    {
        PrintHelp();
        return -1;
    }

    Dim::Setup(conf.Get<string>("dns"));

//    try
    {
        // No console access at all
        if (!conf.Has("console"))
            return RunDim(conf);

        // Console access w/ and w/o Dim
        if (conf.Get<int>("console")==0)
            return RunShell<LocalShell>(conf);
        else
            return RunShell<LocalConsole>(conf);
    }
/*    catch (exception& e)
    {
        cerr << "Exception: " << e.what() << endl;
        return -1;
    }*/

    return 0;
}