DybIODaqSvc.cc

Go to the documentation of this file.

/*
 *  DybIODaqSvc.cc
 *  DybIOSvc
 *
 *  Created by Simon Patton on 3/7/11.
 *  Copyright 2011 DayaBay Collaboration. All rights reserved.
 *
 */
#include "DybIODaqSvc.h"

#include "DaqReadoutFormat/ByteBuffer.h"
#include "DaqFormatModules/DaqReadoutSvc.h"
#include "DaqXml/DaqXmlStream.h"
#include "Event/DaqBlock.h"
#include "Event/DaqBlockHeader.h"
#include "FileReadoutFormat/DaqFromStrStream.h"
#include "FileReadoutFormat/FileBuffer.h"
#include "FileReadoutFormat/FileTraits.h"
#include "GaudiKernel/MsgStream.h"
#include "GaudiKernel/IConversionSvc.h"
#include <string>

using DayaBay::DaqBlock;
using DayaBay::DaqBlockHeader;
using DybDaq::ByteBuffer;
using DybDaq::DaqFromStrStream;
using DybDaq::DaqXmlStream;
using DybDaq::FileBuffer;
using DybDaq::FileEndRecord;
using DybDaq::FileTraits;
using std::string;

static const string kTwoSpaces("  ");

DybIODaqSvc::DybIODaqSvc(const std::string& name,
                           ISvcLocator* svc) :
    Service(name,
            svc),
    m_rioSvc(0),
    m_input(0),
    m_convSvc(0),
    m_daqSvc(0),
    m_daqBlockCnv(0),
    m_nextInsertion(0),
    m_entry(0),
    m_pendingHeader(0) {
    declareProperty("DaqBlockLocation",
                    m_daqBlockLocation=DayaBay::DaqBlockHeaderLocation::Default,
                    "Location in file where DAQ block will be saved.");
}

StatusCode DybIODaqSvc::queryInterface(const InterfaceID& id,
                                        void** interface) {
    if (IDybIODaqSvc::interfaceID().versionMatch(id)) {
        // Good enough. Return this object
        *interface = dynamic_cast<IDybIODaqSvc*>(this);
        addRef();
        return StatusCode::SUCCESS;
    }
    // Try base class
    return Service::queryInterface(id,
                                   interface);
}

StatusCode DybIODaqSvc::finalize() {
    if (0 != m_pendingHeader) {
        delete m_pendingHeader;
    }
    if (0 != m_daqSvc) {
      m_daqSvc->release();
    }
    if (0 != m_convSvc) {
      m_convSvc->release();
    }
    if (0 != m_rioSvc) {
      m_rioSvc->release();
    }
    return Service::finalize();
}

StatusCode DybIODaqSvc::setEventEntry(const unsigned int entry) {
    findDaqReadoutSvc();
    if (0 != m_daqSvc) {
        m_daqSvc->setFileRecord((const FileEndRecord*)0);
    }
    while (m_nextInsertion == entry) {
        if (0 != m_pendingHeader) {
            parseDaqBlock(*(m_pendingHeader->daqBlock()));
            delete m_pendingHeader;
        }
        m_pendingHeader = readDaqBlockHeader();
        if (0 == m_pendingHeader) {
            return StatusCode::FAILURE;
        }
        m_nextInsertion += m_pendingHeader->deltaEvents();
    }
    return StatusCode::SUCCESS;
}

StatusCode DybIODaqSvc::parseDaqBlock(const DaqBlock& daqBlock) {
    string content((const char*)(&((*(daqBlock.bytes()))[0])),
                   daqBlock.size() * ByteBuffer::BYTES_IN_INT);
    DaqFromStrStream reader;
    reader.openWithString(content);
    const FileBuffer* record = reader.nextRecord();
    while (0 != record) {
        if(record->isMarked(FileTraits::kDataSeparatorRecord)) {
            return StatusCode::FAILURE;
        } else {
            if (0 != m_daqSvc) {
                m_daqSvc->setFileRecord(record);
            }
            std::stringstream message;
            message << '\n';
            DaqXmlStream xmlStream(message,
                                   kTwoSpaces,
                                   true);
            xmlStream.dumpElement(*record);
            MsgStream log(msgSvc(),
                          "DaqFormatContext");
            log << MSG::INFO << message.str() << endreq;        
        }
        record = reader.nextRecord();
    }
    return StatusCode::SUCCESS;
}

StatusCode DybIODaqSvc::findDaqReadoutSvc() {
    if (0 != m_daqSvc) {
      return StatusCode::SUCCESS;
    }
    MsgStream log(msgSvc(),
                  "DybIODaqSvc");
    IService* isvc = 0; 
    StatusCode status = service("DaqReadoutSvc",
                                "DaqReadoutSvc",
                                isvc);
    if (status.isFailure() ) {
        log << MSG::ERROR
            << "Failed to get IDaqReadoutSvc as IService"
            << endreq;
        return status;
    }
    IDaqReadoutSvc* daqSvc = 0;
    status = isvc->queryInterface(IDaqReadoutSvc::interfaceID(),
                                  (void**)&daqSvc);
    if (status.isFailure()) {
        log << MSG::ERROR
            << "Service does not implement IDaqReadoutSvc interface"
            << endreq;
        return status;
    }
    m_daqSvc = dynamic_cast<DaqReadoutSvc*>(daqSvc);
    return StatusCode::SUCCESS;
}

StatusCode DybIODaqSvc::findInputStream() {
    if (0 != m_input) {
      return StatusCode::SUCCESS;
    }
    MsgStream log(msgSvc(),
                  "DybIODaqSvc");
    IService* isvc = 0; 
    StatusCode sc = service("RootIOCnvSvc",
                            isvc,
                            true);
    if(sc.isFailure()){
        log << MSG::ERROR
            << "Failed to get RootIOCnvSvc as IService"
            << endreq;
        return sc;
    }
    sc = isvc->queryInterface(IRootIOSvc::interfaceID(),
                              (void**)&m_rioSvc);
    if (sc.isFailure()) {
        log << MSG::ERROR
            << "Conversion service RootIOCnvSvc"
            << " does not implement IRootIOCnvSvc"
            << endreq;
        return sc;
    }

    IRootIOSvc::InputStreamMap& ism = m_rioSvc->inputStreams();
    const IRootIOSvc::InputStreamMap::iterator finished = ism.end();
    for (IRootIOSvc::InputStreamMap::iterator it = ism.begin();
         finished != it;
         ++it) {
        log << MSG::DEBUG
            << "checking input stream: "
            << it->first << endreq;
        if( it->first == m_daqBlockLocation ){
            m_input = it->second;
            return StatusCode::SUCCESS;
        }
    }
    return StatusCode::FAILURE;
}

StatusCode DybIODaqSvc::findConverter() {
    if (0 != m_daqBlockCnv) {
      return StatusCode::SUCCESS;
    }
    MsgStream log(msgSvc(),
                  "DybIODaqSvc");
    IService* isvc = 0; 
    StatusCode sc = service("RootIOCnvSvc",
                            isvc,
                            true);
    if(sc.isFailure()){
        log << MSG::ERROR
            << "Failed to get RootIOCnvSvc as IService"
            << endreq;
        return sc;
    }
    sc = isvc->queryInterface(IConversionSvc::interfaceID(),
                              (void**)&m_convSvc);
    if (sc.isFailure()) {
        log << MSG::ERROR
            << "Conversion service RootIOCnvSvc"
            << " does not implement IConversionSvc"
            << endreq;
        return sc;
    }

    DaqBlockHeader daqBlockHeader;
    IConverter* converter = m_convSvc->converter(daqBlockHeader.classID());
    if (0 == converter) {
        log << MSG::ERROR
            << "Failed to get DaqBlockHeader converter"
            << endreq;
        return StatusCode::FAILURE;
    }
    m_daqBlockCnv = dynamic_cast< RootIOTypedCnv<PerDaqBlockHeader, DaqBlockHeader>* >(converter);
    if (0 == m_daqBlockCnv) {
        log << MSG::ERROR
            << "Failed to cast DaqBlockHeader converter"
            << endreq;
        return StatusCode::FAILURE;
    }
    return StatusCode::SUCCESS;
}

DaqBlockHeader* DybIODaqSvc::readDaqBlockHeader() {
    MsgStream log(msgSvc(),
                  "DybIODaqSvc");

    StatusCode sc = findInputStream();
    if (sc.isFailure()) {
        return 0;
    }
    sc = findConverter();
    if (sc.isFailure()) {
        return 0;
    }

    m_input->setEntry(m_entry);
    log << MSG::DEBUG
        << "Return status of read: "
        << m_input->read()
        << "\t" 
        << m_input->obj()
        << endreq;
    PerDaqBlockHeader* perDaqBlockHeader = (PerDaqBlockHeader*)(m_input->obj());
    if(0 == perDaqBlockHeader){
        log << MSG::ERROR
            << "Failed to get persistent DaqBlockHeader"
            << endreq;
        return 0;
    }

    DaqBlockHeader* result = new DaqBlockHeader;
    sc = m_daqBlockCnv->PerToTran(*perDaqBlockHeader,
                                  *result);
    if (sc.isFailure()) {
        log << MSG::ERROR
            << "Failed to translate persistent DaqBlockHeader" 
            << endreq;
        return 0;
    } else {
        ++m_entry;
    }
    return result;
}