scan.py

Go to the documentation of this file.

import os, logging, stat, pickle, datetime
from pprint import pformat
from multiprocessing import Process
from DybPython import DB, envvar

log = logging.getLogger(__name__)

__all__ = ('Scan','perdict')

loc_mtime = lambda path:datetime.datetime.fromtimestamp(os.stat(path)[stat.ST_MTIME])
utc_mtime = lambda path:datetime.datetime.utcfromtimestamp(os.stat(path)[stat.ST_MTIME])


def dbifullscan( vs ):
    """
    Perform fullscan for all DBI tables in a single DB, saving the per-table scans 
    at location given by `vs`. This func is performed in subprocesses for full DBI stack control.

    :param dbconf: config section
    :param tab: table name or None to traverse all payload tables
    :param ctx: context dict passed to fullscan or None to traverse all contexts
    """ 
    from DybDbi import gDbi
    gDbi.level = 'WARNING'

    dbconf, tab, ctx, opts  = vs.coords_()

    with envvar(DBCONF=dbconf):
        db = DB()
        ptabs = [tab] if tab else db.showpaytables
        for tn in ptabs: 
            ta = db.tab(tn)
            insertdate = ta.vlast('INSERTDATE')
            if not insertdate:
                log.error("skipping empty table %s " % tn )
                continue 
            ctxs = [ctx] if ctx else ta.actual_ctxs()
            for _ctx in ctxs:
                v = vs.spawn( ctx=_ctx)
                path = v.path

                stamp = utc_mtime(path) if os.path.exists(path) else None
                if stamp == None or insertdate > stamp or vs.rescan:
                    log.info("cache miss : fullscan %s %s to %s %s " % (tn, _ctx, path, opts))
                    fs = ta.fullscan( _ctx , opts )
                    fss = Scan(fs)
                    fss.save( path )
                    if vs.txtscan:
                        fss.txtsave( path+'txt')
                else: 
                    log.info("cached hit : fullscan %s %s %s still valid " % (tn, _ctx, path))


class perdict(dict):
    def save(self, path):
        with open(path,"wb") as fp:
            pickle.dump( dict(self) , fp )
    def load(cls, path):
        with open(path,"rb") as fp:
            obj = pickle.load(fp)
        return cls(obj)
    load = classmethod(load)    
    def txtsave(self, path):
        with open(path,"w") as fp:
            fp.write( pformat(dict(self)) )
    def txtload(cls, path):
        with open(path,"w") as fp:
            obj = eval(fp.read())   ## sensitive to form of datetime import 
        return cls(obj)
    txtload = classmethod(txtload)    
 

class Scan(perdict):
    def compare( cls , vs  ):

        path = vs.path 
        log.info("Scan.compare vs %s creating %s  " % ( vs, path ))

        cf = vs.compare

        assert cf in ('vary','dbconf',)        
        assert len(vs[cf]) == 2 

        xv = vs.varycopy(0)
        x  = xv.make_scan() 
        
        yv = vs.varycopy(1)
        y = yv.make_scan() 

        sc = cls.compare_( x , y ) 
        log.info("Scan.compare saving comparison scan %s to %s " % ( sc['_stat'], path) ) 
        sc.save( path )
        return sc

    compare = classmethod( compare )

    def minmax( cls, d, pfx, val  ):
        mx, mi = pfx+'max',pfx+'min'
        if not d.has_key(mx) or val > d[mx]:d[mx] = val 
        if not d.has_key(mi) or val < d[mi]:d[mi] = val 
    minmax = classmethod( minmax )


    def compare_( cls , x , y , diffkeys=['dig','t','n',], i2vfn=lambda _:_['t'] ):
        """
        Compare two dict-of-list-of-dict, checking same shape for the
        non leaf parts. The leaf dict is compared kv by kv and differences
        result in the differing values being incorportated into a tuple of
        differers in the newly dict-of-list-of-dict



        :param diffkeys: list of keys to regard when detecting dict differences
                         default ignores collision and ncollision differences 

        Changes:

        #. moved from diffing on SEQNO with `rvs` to digest diffing with `dig` allowing comparison of non-SEQNO aligned 
 
        """
        assert type(x) == cls and type(y) == cls
        xk = sorted(x.keys())
        yk = sorted(y.keys())

        if xk != yk:
            log.info( "xk-yk: %s " % set(xk).difference(set(yk)) )
            log.info( "yk-xk: %s " % set(yk).difference(set(xk)) )
            for ix,xkk in enumerate(xk):
                print "x",ix,xkk
            for iy,ykk in enumerate(yk):
                print "y",iy,ykk

        assert xk == yk, (xk,yk)
        ndif = 0 
        cf = {}

        mm = dict( dif={} , all={} )
        for k in xk:             ## insertdate
            lxk = len(x[k])
            lyk = len(y[k])

            if lxk != lyk:
               print "x[k]", k
               for ix,dx in enumerate(x[k]):print ix,dx 
               print "y[k]", k
               for iy,dy in enumerate(y[k]):print iy,dy 

            assert lxk == lyk, ( lxk,lyk )
            cf[k] = []
            for i in range(lxk):
                dx = x[k][i]
                dy = y[k][i]   

                ivx = i2vfn(dx)   
                ivy = i2vfn(dy)   
                assert ivx == ivy , ("differing index to time conversion for x and y ", ivx,ivy )

                cls.minmax( mm['all'] , 'i' , ivx )
                cls.minmax( mm['all'] , 'k' , k )

                dxk = sorted(dx.keys())
                assert dxk == sorted(dy.keys()) 
                ddif = 0
                dcf = {}
                for kk in dxk:
                    if dx[kk] != dy[kk]:
                        dcf[kk] = (dx[kk],dy[kk])
                        if len(diffkeys) == 0 or kk in diffkeys:   
                            ddif += 1
                            cls.minmax( mm['dif'] , 'i' , ivx )
                            cls.minmax( mm['dif'] , 'k' , k )
                            log.debug("count difference in %s %s %s  " % ( kk, dcf[kk], diffkeys )) 
                        else:
                            log.debug("ignore difference in %s %s %s " % ( kk, dcf[kk], diffkeys )) 
                    else:
                        dcf[kk] = dx[kk]
                        log.debug("equality in %s %s %s " % ( kk, dx[kk], dy[kk] ))
                if ddif > 0: 
                    ndif += 1
                    log.warn( "dcf %s" % dcf )

                cf[k].append( dcf )
        sc = cls(cf)
        sc['_stat'] = dict(ndif=ndif, minmax=mm )
        log.warn("Scan.compare_ _stat %s " % sc['_stat'] )
        return sc
    compare_ = classmethod( compare_ )

    def get_or_create( cls, _vs ):
        """
        Handles both leaf and comparison VlutSpec 
        action is effected by options `_recompare` and `_rescan` and whether
        a prior scan exists

        """
        vs = _vs.copy( category="cache" )
        path = vs.path
        exists = os.path.exists( path )
        if vs.compare:
            if not exists or vs.recompare: 
                log.info("Scan.get_or_create making comparison scan %s " % path )
                scan = cls.compare( vs ) 
            else:
                log.info("Scan.get_or_create loading comparison scan %s " % path )
                scan = Scan.load( path )
        else:
            if not exists or vs.rescan:
                log.info("Scan.get_or_create making normal scan %s " % path )
                cls.make_scan( vs )
                scan = Scan.load( path )
            else:
                log.info("Scan.get_or_create loading normal scan from %s " % path )
                scan = Scan.load( path )
        return scan
    get_or_create = classmethod( get_or_create ) 

    def make_scan( cls, vs ):
        """
        Invokes `dbifullscan` for each `dbconf` in a separate process allowing 
        the full DBI stack to be booted in each process (working around DBI non-resetability).
        The scans are persisted at locations controlled by the `vs` instance.

        :param vs: VlutSpec instance specifiying dbconf/tn/ctx/opts
        """ 
        assert vs['category'] == 'cache', (vs, vs['category'] )
        if type(vs['dbconf']) == list:
            dbconfs = vs['dbconf']
        else:
            dbconfs = [vs['dbconf']]  
        
        for dbconf in dbconfs:
            v = vs.spawn( dbconf=dbconf )
            p = Process( target=dbifullscan , args=(v,) )
            p.start()
            p.join()    # blocks to serialize
            assert p.exitcode == 0, "dbiscan: subprocess retuned non-zero exitcode %d " % p.exitcode
        pass
    make_scan = classmethod( make_scan )

        




if __name__ == '__main__':
    logging.basicConfig(level=logging.INFO)
    ordering = ""            # "SEQNO desc" "SEQNO asc"
    from vlutspec import VlutSpec
    vs = VlutSpec( dbconf="tmp_offline_db".split(), tn='CableMap', ctx=None, opts=dict(_rescan=True,ordering=ordering) )
    scan = Scan.get_or_create( vs )