fileinfo.py

Go to the documentation of this file.

#!/usr/bin/env python
import logging, time
from datetime import datetime, timedelta
from baseinfo import BaseInfo
log = logging.getLogger(__name__)

class DaqFiles(dict):
    def __init__(self, source , cfg={} ):
        """
        Queries ``SFO_TZ_FILE`` to find files with records that have not yet been propagated offline

        """
        kstf = source.kls_("SFO_TZ_FILE")
        def _kstf__repr__(self):
            return "%s %s %s %s" % ( self.__class__.__name__, self.RUNNR, self.OPENTIME, self.LFN )
        kstf.__repr__ = _kstf__repr__
 
        self.source = source
        self.cfg = cfg 
        self.kstf = kstf

        daq = self.source()            ## single session closured in 
        self.qdaq  = lambda *qwn:daq.query(*qwn).order_by(kstf.OPENTIME).filter(kstf.FILESTATE == 'CLOSED')
        self.qdaqr = lambda *qwn:daq.query(*qwn).order_by(kstf.OPENTIME.desc()).filter(kstf.FILESTATE == 'CLOSED')

    def range(self):
        """
        DAQ Query recording `first` and `last` closed entries, in OPENTIME order 
        """
        self.first = self._first()
        self.last  = self._last()
        log.info("first %s " % self.first )
        log.info("last  %s " % self.last  )

    def runs( self , startlocal ):
        """
        :param startlocal: local datetime after which are interested in DAQ files

        Return ``RUNNR`` corresponding to files closed after `startlocal` in ``OPENTIME`` order  
        """
        kstf = self.kstf
        maxFilesPerRun = self.cfg.get('maxFilesPerRun',1000)
        q = self.qdaq(kstf.RUNNR).filter(kstf.OPENTIME>startlocal).distinct()
        daqruns = [run for run, in q.limit(maxFilesPerRun).all()]   
        log.info("qfter %s yields %s daqruns " % ( startlocal, len(daqruns) )) 
        return daqruns

    def _first(self):
        return self.qdaq(self.kstf).first()
    def _last(self):
        return self.qdaqr(self.kstf).first()

    def newfiles( self , startlocal, offnames=[] ):
         """
         :param offnames: list of LFN of already collected files
         """
         kstf = self.kstf
         maxFilesPerRun = self.cfg.get('maxFilesPerRun',1000)
         daq = self.source()
         q = self.qdaq(kstf).filter(kstf.OPENTIME>startlocal)
         if len(offnames) > 0:    
             daqfiles = q.filter(~kstf.LFN.in_(offnames)).limit(maxFilesPerRun).all()  ## "NOT IN_" 
         else:
             daqfiles = q.limit(maxFilesPerRun).all()  
         return daqfiles

    

class OffFiles(dict):
    def __init__(self, target , cfg ):
        self.target = target 
        self.cfg = cfg
        self.kdrdfi = self.target.dbikls_("DaqRawDataFileInfo")

        def kdrdfi__repr__(self):
            return "%s %s %s %s %s %s" % ( self.__class__.__name__, self.runNo, self.fileName, self.TIMESTART, self.TIMEEND, self.SEQNO )
        self.kdrdfi.__repr__ = kdrdfi__repr__ 
        self.kdrdfi.localstart = property(lambda self:self.TIMESTART - timedelta(seconds=time.timezone))   ## time.timezone is  -28800

    def last(self):
        """
        Accesses highest SEQNO entry from offline DBI pair `DaqRawDataFileInfo` and updates self with values

        #    self.update(runNo=last.runNo,fileNo=last.fileNo,fileName=last.fileName,timeStart=last.TIMESTART) 
        """
        off   = self.target()
        kdrdfi = self.kdrdfi 

        last = off.query(kdrdfi).order_by(kdrdfi.SEQNO.desc()).first()
        self.last = last 
        return last 

    def offnames(self, runs=[]):
        """
        :param runs: list of runs

        offline filenames corresponding to those runs
        """
        off = self.target()
        kdrdfi = self.kdrdfi
        if len(runs) == 0:
            offnames = []
        else: 
            offnames = [fileName for fileName, in off.query(kdrdfi.fileName).filter(kdrdfi.runNo.in_(runs)).all()]      
        return offnames



class FileInfo(BaseInfo):
    def __init__(self, source="tmp_daqdb", target="tmp_offline_db" , cfg={} ):
        BaseInfo.__init__(self, source, target, cfg) 
        self.daqf = DaqFiles( self.source, cfg  )
        self.offf = OffFiles( self.target, cfg  )

    def __repr__(self):
        return BaseInfo.__repr__(self)

    def last(self):
        return self.offf.last()

    def after(self, startlocal , offset_seconds=None ):
        """
        Queries daq table SFO_TZ_FILE for closed files, that were opened after cutoff time `startlocal` 
        and that are not already in offline table DaqRawDataFileInfo

        Former ``N+1`` nested approach ``get_newFileList``:

        * iterated over DAQ file entries, doing offline queries at each iteration to check for preexisting file names 

        New flat ``3`` query approach:

        #. runs corresponding to new DAQ files
        #. offline files names corresponding to those runs
        #. DAQ files with names not already in offline 

        Could be done flat ``2``, but that would entail IN querying with all filenames, which is not sustainable.

        """
        if not offset_seconds:
            offset_seconds = self.cfg.get('file_seconds_offset', -60 )

        startlocal = startlocal + timedelta(seconds=offset_seconds)

        daqruns = self.daqf.runs( startlocal )
        log.info("qfter %s yields %s daqruns " % ( startlocal, len(daqruns) )) 

        daqfiles = []
        if len(daqruns) == 0:
            log.info("no new closed runs after %s " % startlocal )
        else:
            offnames = self.offf.offnames(daqruns) 
            daqfiles = self.daqf.newfiles( startlocal, offnames )
        return daqfiles





if __name__ == '__main__':

    logging.basicConfig(level=logging.INFO)

    from NonDbi import MetaDB
    daqm = MetaDB("tmp_daqdb")
    offm = MetaDB("tmp_offline_db")

    daqf = DaqFiles( daqm )
    daqf.range()

    finfo = FileInfo(source=daqm,target=offm, cfg={})
    print finfo

    last = finfo.last()
    print "last_collected %r " % last 
    print "localstart %r " % last.localstart 

    parse = lambda _:datetime.strptime(_,"%Y-%m-%d %H:%M:%S")

    #cut = last.localstart
    #cut = daqf.first.OPENTIME #+ timedelta(minutes=600)
    #cut = parse("2011-10-20 16:35:24")
    cut = parse("2011-10-18 10:03:47")

    log.info("cut %s " % cut )
    fis = finfo.after( cut , offset_seconds=-60  )

    for sli in slice(0,10), slice(-10,None):
        print sli
        for fi in fis[sli]:
            print fi