asciicat.py

Go to the documentation of this file.

#!/usr/bin/env python
"""
Direct python access to the rows of DBI ascii catalog via csv parsing, implemented to
enable pre-insert checks prior to rloadcat of ascii catalogs into the DB

No need to use the more complex CSV 

#. as DBI Ascii catalogs are gauranteed valid .csv 
#. have some specifics regards the key names that need special handling

Ambiguity Avoidance
~~~~~~~~~~~~~~~~~~~~~

Need to write ascii cat with new ``INSERTDATE``s 
changed to UTC now of an update, to avoid the **window of ambiguity**  problem :dybsvn:`ticket:844`

#. first get new ``SEQNO``



"""
import os, re, logging
from datetime import datetime
from csv import DictReader
from StringIO import StringIO
from pprint import pformat
from copy import deepcopy

log = logging.getLogger(__name__)
pathx = lambda _:os.path.abspath(os.path.expanduser(os.path.expandvars(_)))



class AsciiRow(dict):
    """
    Used internally by AsciiCSV::
 
         t = AsciiCSV( "~/path/to/catdir") 
         r = AsciiRow.Create(d, t ) 
         print r.pk    ##  (1,10) 
         print r.fv
         print r

    """ 
    def Create(cls, d, t ):
        """
        :param d: dict with row content
        :param t: parent table AsciiCSV instance
        """
        r = cls(d)
        r.t = t
        return r
    Create = classmethod(Create)

    def clone(self):
        return AsciiRow.Create( dict(self).copy() , self.t )

    def get_pk(self):
        if self.t.intpk:
            fv = map(lambda k:int(self[k]),self.t.pkfields)
        else:
            fv = map(lambda k:self[k],self.t.pkfields)
        return fv[0] if len(fv) == 1 else tuple(fv)
    def set_pk( self, pk ):
        if type(pk) == int:pk = tuple([pk])     ## type regularize
        assert len(pk) == len(self.t.pkfields), "pk %r is wrong shape expecting %r " % ( pk, self.t.pkfields )
        for k,v in zip(self.t.pkfields,pk):
            self[k] = str(v)               ## so faked entries match kosher ones hailing from csv files   
    pk = property(get_pk, set_pk)
    fv = property(lambda self:map(lambda k:self[k],self.t.fields), doc="list of field values in order held by AsciiCSV instance")
    def __str__(self):
         return ",".join(map(lambda _:"\"%s\""%_[0] if _[1] else _[0], zip(self.fv,self.t.quote))) 


class AsciiCSV(dict):
    """
    AsciiCSV is a dict keyed by PK with values which are dicts keyed by fieldname
    which have string values. The PKs and field names are determined from the 
    header line of the csv file. For example::

        csv = AsciiCSV( None, "~/dybaux/catalog/tmp_offline_db/CalibPmtSpec/CalibPmtSpec.csv")()
              ## standalone use without a catalog

        csv[(50,167)] == {'PMTAFTERPULSE': '0',
                          'PMTDARKRATE': '0',
                          'PMTDESCRIB': 'DayaBayOWS-wall09-spot23-in',
                          'PMTEFFIC': '1',
                           ...
                          'ROW_COUNTER': '167',
                          'SEQNO': '50'}

        csv[(50,167)]['PMTEFFIC'] == '1'

        for pk,v in csv.items():
            print pk, v

    Typical PKs:
          
    #. (SEQNO,ROW_COUNTER) integer tuple for payload tables
    #. SEQNO integer for validity tables
    #. TABLENAME string for LOCALSEQNO tables 

    """
    def __init__(self, cat, path, **kwa ):
        dict.__init__(self)
        self.cat = cat
        self.path = path
        self.kwa = kwa
        self.tabbase = kwa.get('tdir',None)
        self.tabname = kwa.get('tnam',None)
        self.pk = kwa.get('tnam',None)
        self.seqnos_read = []
        self.pks_read = []
 
    # too memory wasteful apparently 
    #pks    = property(lambda self:list(set(map(lambda _:_.pk, self.values() ))), doc="for all tables" )
    #allseqno = property(lambda self:list(set(map(lambda _:int(_['SEQNO']), self.values() ))), doc="for payload/validity tables" )

    pks    = property(lambda self:self.pks_read, doc="for all tables" )
    allseqno = property(lambda self:self.seqnos_read, doc="for payload/validity tables" )
    maxseqno = property(lambda self:max(self.allseqno), doc="for payload/validity tables")
    minseqno = property(lambda self:min(self.allseqno), doc="for payload/validity tables")


    abspath = property(lambda self:self.cat.resolve(self.path))

    pretty = property(lambda self:pformat(dict(self)), doc="pretty print the AsciiCSV dict, only useful for small CSVs such as LOCALSEQNO" )    
    lastusedseqno = property(lambda self:dict(map(lambda k:(k,int(self[k]['LASTUSEDSEQNO'])) , self )), doc="for LOCALSEQNO tables" )   
    smry = property(lambda self:"\n".join( [self.pretty, pformat(self.lastusedseqno) ]), doc="for LOCALSEQNO tables" )

    orderpk = property(lambda self:sorted(self))
    lastpk = property(lambda self:self.orderpk[-1])
    last = property(lambda self:self[self.lastpk])

    def __repr__(self):
        return "%s [%-5d] %s  " % ( self.__class__.__name__, len(self), self.path  )

    def compare(self, other):
        """
        Equality operator for dict-of-dict works as would want : 
        recursive same keys with same values. As demonstrated::
 
            old = AsciiCSV( None, "~/dybaux/catalog/tmp_offline_db/CalibPmtSpec/CalibPmtSpec.csv.old")()
            new = AsciiCSV( None, "~/dybaux/catalog/tmp_offline_db/CalibPmtSpec/CalibPmtSpec.csv.new")()

            old == new   ## True

            keep = old[(50,167)]['PMTEFFIC'] 
            old[(50,167)]['PMTEFFIC'] = keep + "changed"
            old == new                                     ## False after changing a single value 

            old[(50,167)]['PMTEFFIC'] = keep
            old == new   ## True                           ## back to equality 

            list(old) == list(new)       ## False          ## key order cannot be relied on 
            sorted(old) == sorted(new)   ## True
            old.orderpk == new.orderpk   ## True


        The .old which matches dybaux r4968 is from before standardized order was established...
 
            [blyth@belle7 CalibPmtSpec]$ diff CalibPmtSpec.csv CalibPmtSpec.csv.old
            [blyth@belle7 CalibPmtSpec]$ diff CalibPmtSpec.csv CalibPmtSpec.csv.new | wc
            2470    4934  222803

        Standardize it by ``old.write()`` after which::

            [blyth@belle7 CalibPmtSpec]$ diff CalibPmtSpec.csv.old  CalibPmtSpec.csv.new
            [blyth@belle7 CalibPmtSpec]$ 

        """
        return self == other


    def parse_hdr(self, hdr ):
        """
        Make sense of hdr line like:: 

           SEQNO int(11),ROW_COUNTER int(11),SENSORID int(11),CHANNELID int(11),PRIMARY KEY (SEQNO,ROW_COUNTER)

        """
        fdesc,pkdesc = hdr.split(",PRIMARY KEY ")
        assert pkdesc[0] == '(' and pkdesc[-1] == ")"
        
        self.hdr = hdr 
        self.fields  =  map(lambda _:_.split()[0], fdesc.split(",") )
        self.dbtypes =  map(lambda _:_.split()[1], fdesc.split(",") )
        self.fdict = dict(zip(self.fields, self.dbtypes))
        self.pkfields = pkdesc[1:-1].split(",")
        self.pktypes =  map(lambda k:self.fdict[k], self.pkfields)
        self.intpk = len(filter( lambda k:k.startswith('int'), self.pktypes )) == len(self.pktypes)   ## all integer pk
        qtyps = ("char","varc","date",) 
        self.quote = map( lambda _:self.fdict[_][0:4] in qtyps, self.fields )    ## the type is quoted


    def __call__(self, content=None):
        return self.read(content)

    def resolve(self, base=None ):
        if self.cat:
            path = self.cat.resolve( self.path, base )
        else:
            path = pathx(self.path)
        return path

    def read(self, content=None):
        """
        Parse the csv storing as dict of dict keyed by pk 

        Typical DBI validity and payload tables are ordered by SEQNO/(SEQNO,ROW_COUNTER), but 
        the standard order is not followed by:

        #. LOCALSEQNO, not surprising as TABLENAME keyed and follows historic ordering 
        #. CalibPmtSpec, CalibPmtSpecVld  **SURPISE : EVIDENCE FOR NON-DBI HANDLING**

        """
        if content:
            pass
        else:
            path = self.resolve()
            log.debug("parse csv %s " % path ) 
            content = open( path , "r" ).read()

        feol = content.find("\n")
        self.parse_hdr( content[:feol] )  ## determine fields, pk etc...
        rdr = DictReader( StringIO(content[feol+1:]), fieldnames=self.fields )

        sqn=[] 
        pks=[]

        for i,d in enumerate(rdr):
            r = AsciiRow.Create(d, self )
            rpk = r.pk
            self[rpk] = r 
            pks.append(rpk)

            # collect seqno during filling to avoid memory expense of grabbing latter 
            seqno = int(r.get('SEQNO',0))
            if seqno:
                sqn.append(seqno)

            assert len(self) == i + 1, ("error non-unique pk %r for row %r " % ( rpk, r ) )  
            pass

        self.seqnos_read = list(set(sqn))
        self.pks_read = list(set(pks))
        return self       


    def merged(self, update , postpend="" ):
        """
        Return a new instance that merges `update` into `self`, changes in
        the `update` override things in self 
       
        NB `self` is not changed by this 

        :param update: instance of `AsciiCSV` 
        :return merged instance:
        :rtype dict of dict:

        """
        log.debug("merge csv %r " % update ) 
        mrgd = deepcopy(self)
        if mrgd.path:
            mrgd.path += postpend

        allpk = set(self).union(set(update))
        for pk in allpk:
            if pk in mrgd and pk in update:
                m = deepcopy(self[pk])
                m.update( update[pk] )      
                log.debug("merged of pk %r combines self %r with update %r yielding %r  " % ( pk, mrgd[pk],update[pk],m) )
                mrgd[pk] = m
            elif pk in mrgd:
                log.debug("merged of pk %r keeps entry %r " % ( pk, mrgd[pk] ) )
                pass
            elif pk in update:
                log.debug("merged of pk %r adds update entry %r " % ( pk, update[pk] ) )
                mrgd[pk] = deepcopy(update[pk])
        return mrgd

    def compare(self, other):
        """
        Return a dict with the differences  
        """
        pass

    def content(self):
        content = self.hdr + "\n"
        for pk in sorted(self):
            content += str(self[pk]) + "\n"
        return content

    def write(self, base=None):
        path = self.resolve( base=base )
        if base == None:
            log.warn("inplace overwriting %s " % path )        
        else:
            log.debug("writing to %s " % path )

        wdir = os.path.dirname( path )
        if not os.path.exists(wdir):
            os.makedirs(wdir)

        fp = open(path, "w")
        fp.write( self.content() )
        fp.close()




class AsciiCat(dict):
    """
    DBI Ascii catalog (a tablename keyed dict holding AsciiCSV representing each table), usage::

        import logging
        logging.basicConfig(level=logging.INFO)

        cat = AsciiCat("~/dybaux/catalog/tmp_offline_db")  

        for tn,csv in cat.items():
            print tn, csv

        for tn,lastseqno in cat.seqno.items():
            print cat[tn+'Vld'][lastseqno]         

        for tn in cat.orderpk:
            csv = cat[tn]
            print tn, csv.abspath, csv.tabname, csv.tabbase, csv.orderpk[0:100]


    """
    ptn = re.compile("(?P<tdir>[^/]*)/(?P<tnam>\S*)\.csv$")

    def parse_relpath(self, relpath):
        m = self.ptn.match(relpath)
        assert m, "failed to match %s " % relpath 
        d = m.groupdict()
        tab = d['tdir']
        d['tab'] = tab
        if d['tdir'] == d['tnam']:
            d['type'] = "pay"
        elif "%sVld" % d['tdir'] == d['tnam']:
            d['type'] = "vld"
        return dict(d, relpath=relpath) 

    showtables = property( lambda self:sorted(self.keys()))
    orderpk = property(lambda self:sorted(self))

    @classmethod
    def findcat(cls, dir): 
        catz = filter(lambda n:n[-4:] == ".cat" , os.listdir(dir) )
        assert len(catz) == 1, "must be only one .cat in dir %s " % dir
        return os.path.join( dir, catz[0] )

    def __init__(self, dir , skip_pay_check=False , allow_partial=False ):
        """
        :param dir:
        :param skip_pay_check: when True skip loading payload tables in memory 
        """
        dir = os.path.normpath(pathx(dir))
        assert os.path.isdir(dir) , dir
        catpath = self.findcat( dir ) 

        log.debug("AsciiCat catpath %s skip_pay_check %s allow_partial %s " % (catpath,skip_pay_check,allow_partial) )

        self.skip_pay_check = skip_pay_check   
        self.allow_partial = allow_partial   
        self.dir = dir
        self.v = {}
        self.p = {}
        pass
        self.read( catpath )

    def compare(self, other):
        """
        Checking cat comparison, setup old and new catalogs::

            rm -rf ~/dybaux/catalog/tmp_offline_db
            svn up -r 4968 ~/dybaux/catalog/tmp_offline_db      ## clean r4968 just prior to ordering standardization
            rm -rf ~/tmp_offline_db ; db.py offline_db rdumpcat ~/tmp_offline_db   ## same named catalog with the stanbdard rdumpcat ordering 

        .. code-block:: ipython

                In [15]: old = AsciiCat("~/dybaux/catalog/tmp_offline_db")
                INFO:__main__:read /home/blyth/dybaux/catalog/tmp_offline_db/tmp_offline_db.cat 
                INFO:__main__:done AsciiCat [13   ] /home/blyth/dybaux/catalog/tmp_offline_db {'CalibPmtSpec': 50, 'FeeCableMap': 3, 'HardwareID': 372, 'CalibFeeSpec': 113, 'CableMap': 460}  

                In [16]: new = AsciiCat("~/tmp_offline_db")
                INFO:__main__:read /home/blyth/tmp_offline_db/tmp_offline_db.cat 
                INFO:__main__:done AsciiCat [13   ] /home/blyth/tmp_offline_db {'CalibPmtSpec': 50, 'FeeCableMap': 3, 'HardwareID': 372, 'CalibFeeSpec': 113, 'CableMap': 460}  

                In [17]: old == new
                Out[17]: True

                In [18]: old['CableMap'][(1,1)]['CHANNELID']
                Out[18]: '16908545'

                In [19]: old['CableMap'][(1,1)]['CHANNELID'] += "changed"

                In [20]: old == new
                Out[20]: False

                In [21]: old['CableMap'][(1,1)]['CHANNELID'] = '16908545'

                In [22]: old == new
                Out[22]: True

                In [25]: old.diff(new)
                INFO:__main__:diffing with "diff -r --brief /home/blyth/dybaux/catalog/tmp_offline_db /home/blyth/tmp_offline_db | grep -v .svn "
                INFO:__main__:diff Files /home/blyth/dybaux/catalog/tmp_offline_db/CalibPmtSpec/CalibPmtSpec.csv and /home/blyth/tmp_offline_db/CalibPmtSpec/CalibPmtSpec.csv differ
                INFO:__main__:diff Files /home/blyth/dybaux/catalog/tmp_offline_db/CalibPmtSpec/CalibPmtSpecVld.csv and /home/blyth/tmp_offline_db/CalibPmtSpec/CalibPmtSpecVld.csv differ
                INFO:__main__:diff Files /home/blyth/dybaux/catalog/tmp_offline_db/LOCALSEQNO/LOCALSEQNO.csv and /home/blyth/tmp_offline_db/LOCALSEQNO/LOCALSEQNO.csv differ
                INFO:__main__:diff Files /home/blyth/dybaux/catalog/tmp_offline_db/tmp_offline_db.cat and /home/blyth/tmp_offline_db/tmp_offline_db.cat differ

                In [27]: old.write()
                WARNING:__main__:overwriting catalog into dir /home/blyth/dybaux/catalog/tmp_offline_db 

                In [28]: old.diff(new)
                INFO:__main__:diffing with "diff -r --brief /home/blyth/dybaux/catalog/tmp_offline_db /home/blyth/tmp_offline_db | grep -v .svn "

        """
        return self == other  

    def __repr__(self):
        return "%s [%-5d] %s %r " % ( self.__class__.__name__, len(self), self.dir, self.seqno )

    def read(self, catpath): 
        """
        Parses Vld and LOCALSEQNO tables by default, reading in the 
        csv content into memory.  Only reads payload tables when `skip_pay_check=False`
        Reading payload tables for larger tables is slow and memory expensive. 

        """
        log.info("read %s " % catpath ) 
        self.read_catalog( catpath )

        for tn in self.orderpk:
            read = False
            if tn[-3:] == 'Vld' or tn == 'LOCALSEQNO':
                read = True 
            else:
                read = not self.skip_pay_check
            if read:
                log.info("reading table %s " % tn ) 
                self[tn].read()
            else:
                log.debug("NOT reading table %s " % tn ) 
 
        self.read_seqno()
        self.check_seqno()
        log.info("done %r " % self ) 

    def read_catalog(self, catpath ):
        """
        Reads ascii catalog into memory.  

        This just reads the small .cat file and sets up lightweight 
        AsciiCSV objects without reading in csv content.
        """
        content = open( catpath , "r" ).read().strip()
        relpaths = content.split("\n")
        assert relpaths[0] == "name", "unexpected catalog header line %s " % relpaths[0]
        for d in map(lambda _:self.parse_relpath(_), relpaths[1:] ):
            typ, tab = d.get('type',None), d.get('tab', None)
            assert typ and tab 
            csv = AsciiCSV(self, d['relpath'], **d )
            if typ == 'pay': 
                self.p[tab] = csv   
            elif typ == 'vld':
                self.v[tab] = csv
            pass
            self[csv.pk] = csv   ## keyed by tablename
            pass



    def fake_write( self, tn , npay=1 ):
        """
        Fakes a single validity entry and ``npay`` payload entries, by copying the 
        last entries in validity and payload tables and with a 
        fabricated SEQNO or (SEQNO,ROW_COUNTER) pks.  The `tn` entry in the LOCALSEQNO
        table is also changed to mimic a real DBI write.

        :param tn: payload table name
        :param npay: number of payload rows to write

        Usage::

            cat = AsciiCat("~/dybaux/catalog/tmp_offline_db")
            cat.fake_write('HardwareID', 10 )
            cat.write()                       ## CAUTION inplace overwriting

        """
        p,v,m = self[tn], self[tn+'Vld'],self['LOCALSEQNO']
        assert len(p) > 1 and len(v) > 1 and len(m) > 1, ("cannot fake into empty tables ", len(p),len(v),len(m))

        a = v.last.clone()            ## pk ordered
        lastseqno = a.pk              ## get_pk

        l = p.last                    ## l.pk is tuple with SEQNO,ROW_COUNTER values
        assert l.pk[0] == lastseqno == int(m[tn]['LASTUSEDSEQNO']) , ( "lastseqno mismatch ", l, l.pk, lastseqno , m[tn] )

        fakeseqno = lastseqno + 1 
        a.pk = fakeseqno              ## set_pk     
        v[a.pk] = a                   ## fake validity entry 

        for n in range(npay):
           b = l.clone()
           b.pk = (fakeseqno, n+1 )   ## set_pk 
           p[b.pk] = b                ## fake payload entries

        m[tn]['LASTUSEDSEQNO'] = str(fakeseqno)  

        log.info("fake_write %s %s added LASTUSEDSEQNO %s " % ( tn,npay,fakeseqno )) 

        self.clear_cache()



    def read_seqno(self, tab="LOCALSEQNO"):
        """
        Reads the LASTUSEDSEQNO entries from the LOCALSEQNO table into seqno dict
        """ 
        self._seqno = {}
        for pk,row in self[tab].items():
            t = row['TABLENAME']
            n = int(row['LASTUSEDSEQNO'])
            if t == "*":continue
            self._seqno[t] = n       
        log.debug( "LASTUSEDSEQNO %r FROM %s  " % ( self._seqno, tab)  )
    def check_seqno(self):  
        """
        Checks correspondence between the max SEQNO in the Vld and Pay tables 
        with that recorded in the LOCALSEQNO table

        Note that without the very expensive db.py option: `--EXTRA_PAYLOAD_CHECK`
        which caused setting of `self.skip_pay_check` only the validity tables are checked. 

        The allseqno called inside the maxseqno property is very expensive, causing 
        MemoryError at IHEP
        """ 
        for t,n in self._seqno.items():
            v = self.get(t+'Vld',[])
            p = self.get(t,[])
            if len(v) > 0:
                vmax = v.maxseqno
                assert vmax == n , ( vmax , n )
            if len(p) > 0:
                pmax = p.maxseqno
                assert pmax == n , ( pmax , n )

    def get_seqno(self):
        """
        Tablename keyed dict of LASTUSEDSEQNO values obtained from LOCALSEQNO table
        """ 
        if hasattr(self,'_seqno'):
            return self._seqno
        self.read_seqno()
        self.check_seqno()
        return self._seqno
    seqno = property( get_seqno , doc=get_seqno.__doc__ )

        
    def clear_cache(self):
        """
        delete cached attributes forcing recalc on next access ...
        needed after faking for example  
        """
        if hasattr(self, '_fabseqno' ):
            del self._fabseqno
        if hasattr(self, '_seqno' ):
            del self._seqno
        if hasattr(self, '_allseqno' ):
            del self._allseqno 

    def read_allseqno(self):
        """
        Populates the table name keyed `self._allseqno` dict with results
        from the `.allseqno` property applied to individual `AsciiCSV` for each Vld table.  

        The `self.seqno` dict is populated by reading the LOCALSEQNO.csv so when the 
        tables represented in that do not correspond to those of the catalog. 

        """ 
        self._allseqno = {}
        for tn,lus in self.seqno.items():
            tnv = "%sVld"%tn
            csvv = self.get(tnv, None)
            if not csvv and self.allow_partial:continue  
            assert csvv, "read_allseqno failed to find AsciiCSV for %s : consider use of -P,--ALLOW_PARTIAL option to allow dealing in partial catalogs " % tnv 
            self._allseqno[tn] = csvv.allseqno
    def check_allseqno(self):
        if self.allow_partial:return
        assert sorted(self._allseqno.keys()) == sorted(self._seqno.keys()), "seqno keys mismatch "
        pass
    def get_allseqno(self):
        """
        Populates, checks and returns the `self._allseqno` dict which is keyed 
        on tablename and holds the collected allseqno from the indicdual AsciiCSV
        """ 
        if hasattr(self,'_allseqno'):
            return self._allseqno
        self.read_allseqno()
        self.check_allseqno()
        return self._allseqno
    allseqno = property( get_allseqno , doc=get_allseqno.__doc__ )

    def get_fabseqno(self):
        """
        Summarizes ``cat.allseqno``, by fabricating a dict keyed by table name containing 
        the number of Vld SEQNO (from length of values in ``cat.allseqno``) 

        This dict can be compared with ``cat.seqno``, which is obtained from 
        the LASTUSEDSEQNO entries in the ``LOCALSEQNO`` table::
        Assuming kosher DBI handling of tables this fabricated dict ``cat.fabseqno`` should 
        match ``cat.seqno``, meaning that SEQNO start from 1 and have no gaps.

       .. code-block:: ipython          

                In [1]: from DybPython import AsciiCat

                In [2]: cat = AsciiCat("~/dybaux/catalog/tmp_offline_db")

                In [3]: db.seqno   ## queries the LOCALSEQNO table in DB
                Out[3]: 
                {'CableMap': 213,
                 'CalibFeeSpec': 113,
                 'CalibPmtSpec': 29,
                 'FeeCableMap': 3,
                 'HardwareID': 172}

                In [4]: db.fabseqno    ## a summarization of db.allseqno
                Out[4]: 
                {'CableMap': 213,
                 'CalibFeeSpec': 111,
                 'CalibPmtSpec': 8,
                 'FeeCableMap': 3,
                 'HardwareID': 172}

                In [5]: db.miscreants()   ## assertions avoided by miscreant status
                Out[5]: ('CalibPmtSpec', 'CalibFeeSpec')


        """
        if hasattr(self, '_fabseqno' ):
            return self._fabseqno
        self._fabseqno =  dict(map(lambda(k,v):(k,len(v)),self.allseqno.items()))
        return self._fabseqno
    fabseqno = property( get_fabseqno, doc=get_fabseqno.__doc__ )


    def write(self, xdir=None ):
        """
        :param xdir: write directory 

        Writes catalog into the ``xdir`` directory named after directory basename
        """
        if xdir==None:
            xdir = self.dir
            log.warn("overwriting catalog into dir %s " % xdir )
        else:
            xdir = pathx(xdir)
            assert xdir != self.dir
        self.write_catalog( base=xdir )
        for tn in self.orderpk: 
            csv = self[tn]
            csv.write(base=xdir)

    def write_catalog(self, base ):
        """
        writes the .cat file into directory `base` the name of the `.cat` is 
        obtained from the basename of `base` 
        """
        if not os.path.exists(base):
            log.info("creating %s " % base ) 
            os.makedirs(base)
        outpath = os.path.join( base , "%s.cat" % os.path.basename(base) )
        log.debug("write_catalog to %s " % outpath )

        fp = open(outpath,"w")
        fp.write("\n".join(["name"] + map(lambda tn:self[tn].path, self.orderpk)) + "\n" ) 
        fp.close()           

    def resolve(self, relpath , base=None):
        """Resolve a relative path wrt the base""" 
        if base == None:
            base = self.dir 
        return os.path.join( base , relpath )


    def updates(self, target, tselect=[], fastforward=False ):
        """
        :param target: DBI store (AsciiCat or DB) but usually the target DB to be updated

        Return tablename (payload only) keyed dict listing new SEQNO 
        and prepares the ascii catalog for loading into DB, by fastforwarding INSERTDATE to UTC now. 

        """
        log.debug("updates tselect %r " % tselect )
        upls = self.seqno_updates(target, tselect=tselect)
        upas = self.allseqno_updates(target, tselect=tselect )
        assert upls.keys() == upas.keys(), ("updates keys mismatch ", upls.keys(), upas.keys() )

        # checking correspondence between LASTUSEDSEQNO and the last of the lists of allseqno
        for tn in upas:
            assert upls[tn] == upas[tn][-1] , ("seqno vs allseqno mismatch %s" % tn, upls[tn], upas[tn][-1] ) 
            #print tn,  upls[tn], upas[tn][-1] , upas[tn][0] 

        if fastforward: 
            now = datetime.utcnow()
            now = now.strftime("%Y-%m-%d %H:%M:%S")
            for tn in upas:
                n = 0 
                vld = self[tn+'Vld']
                for sq in upas[tn]:
                    n += 1
                    vld[sq]['INSERTDATE'] = now
                if n > 0: 
                    log.info("fastforward %s validity rows of %s to %s " % (n,tn,now) )
                    vld.write()
        else: 
            pass
        return upas


    def seqno_updates(self, target , tselect=[] ):
        """
        :param target: DBI store (AsciiCat or DB) but usually the target DB to be updated
        :param tselect: restrict comparisons to payload tables in tselect, if tselect list is populated 

        This operates by comparing LASTUSEDSEQNO entries in the LOCALSEQNO tables, via 
        the ``.seqno`` 

        Provides dict keyed by table names of LASTUSEDSEQNO values (payload names only) for 
        new or changed LASTUSEDSEQNO entries in the ascii catalog.  

        For example this AsciiCat might contain::

              {'Noodles': 15, 'CalibPmtSpec': 29, 'FeeCableMap': 3, 'HardwareID': 172, 'CalibFeeSpec': 113, 'CableMap': 213}

        Whereas the target DB contains::
           
              {'Noodles': 10, 'CalibPmtSpec': 29, 'FeeCableMap': 3, 'CalibFeeSpec': 113}  
 
        This would return::

              {'Noodles':15 ,'HardwareID': 172 ,'CableMap':213}
        

        Note the use of SEQNO 0 to mark not-present


        This way of checking for updates works fine without the payload tables in memory
        as it is just using the LASTUSEDSEQNO counts hailing from LOCALSEQNO.csv
        BUT make sure are rcmpcat comparing against a propa full DB not the decoupled subset of tables
        that tmp_offline_db often is.
        """
        upls = {}

        
        ttabs = filter(lambda t:len(tselect) == 0 or t in tselect, target.seqno )
        log.debug("target tables %r " % ttabs )         
        missing = list(set(ttabs).difference(set(self.seqno)))
        assert len(missing) == 0, "tables in target that are not in ascii catalog %r \n try restricting table selection with -t option " % missing
        
        log.debug("seqno_updates self.seqno %r " % self.seqno )         
        log.debug("seqno_updates target.seqno %r " % target.seqno )         

        alldbi = list(set(self.seqno).union(set(target.seqno)))
        for tn in alldbi:
            if len(tselect) > 0 and tn not in tselect:
                continue
            acls = self.seqno.get(tn,0)     ## ascii catalog LASTUSEDSEQNO for table
            tgls = target.seqno.get(tn,0)   ## LASTUSEDSEQNO in target db 
            #log.info("seqno_updates acls %s tgls %s " % (acls,tgls) )
            assert acls >= tgls, ("ERROR LASTUSEDSEQNO in target exceeds that in ascii cat %s " % tn, acls, tgls)
            if acls > tgls:
                upls[tn] = acls
            else:
                pass  
        log.info("seqno_updates : ascii catalog LASTUSEDSEQNO changes relative to target : %r " % upls )
        return upls

    def allseqno_updates(self, target, tselect=[] ):
        """
        :param target: DBI store (AsciiCat or DB) but usually the target DB to be updated
        
        This operates by comparing all SEQNO in the tables

        """ 
        upas = {}
        alldbi = list(set(self.allseqno).union(set(target.allseqno)))
        for tn in alldbi:
            if len(tselect) > 0 and tn not in tselect:
                continue
            acsq = set(self.allseqno.get(tn,[]))         
            tgsq = set(target.allseqno.get(tn,[]))      
            #log.info("allseqno_updates tn %s acsq %s tgsq %s " % (tn,acsq,tgsq))
            assert len(tgsq.difference(acsq)) == 0 , "ERROR SEQNO in db that are not present in ascii catalog table %s  " % tn
            upsq = sorted(list(acsq.difference(tgsq)))     ## SEQNO in ascii catalog but not target DB
            if len(upsq) > 0:
                upas[tn] = upsq 
            else:
                pass
        for tn,seq in upas.items():
            contiguous = seq[-1] - seq[0] + 1  == len(seq) 
            if not contiguous:
                msg = "non contiguous SEQNO in tn %s %s %s %s " % ( tn,  seq[0], seq[-1], len(seq)  )
                if tn in ('CalibPmtSpec',):    ## WHY NOT NORMALLY TICKLED ?
                    log.info(msg)
                else:
                    log.fatal(msg)       
                    assert contiguous, msg 

        #log.debug("allseqno_updates %r " % upas )
        return upas


    def diff(self, otherdir):
        if type(otherdir) == AsciiCat:
            otherdir = otherdir.dir  
        cmd = "diff -r --brief %s %s | grep -v .svn " % ( self.dir, otherdir )
        log.info("diffing with \"%s\"" % cmd )
        for line in os.popen(cmd).readlines():
            log.info("diff %s" % line.strip())


class DD(dict):
    """
    dict comparison
    """
    def __init__(self, a_, b_ , **kwa ):
        dict.__init__(self, **kwa ) 
        a, b = set(a_), set(b_)
        c = a.intersection(b)
        self['added']   = b - c
        self['removed'] = a - c
        self['changed'] = set(o for o in c if a_[o] != b_[o])
        self['unchanged'] = set(o for o in c if a_[o] == b_[o])
        ## specialized
        if kwa.get('increments'):
            self['allincrements'] = dict( (o,b_[o]-a_[o]) for o in c )
            self['increments'] = dict( (o,b_[o]-a_[o]) for o in c if a_[o] != b_[o]  )

    def __str__(self):
        return pformat(self) 



def validate_csv_update( tn , old_ , new_ ):
    """
    Standalone comparison of parsed .csv   

    :param tn: table name
    :param old_: path to starting csv 
    :param new_: path to changed csv 

    Hmm when operating decoupled the new LOCALSEQNO will 

    """
    if tn in ('LOCALSEQNO','GLOBALSEQNO',):
        old = AsciiCSV(None, old_)
        old.read()
        log.info( "oldcsv %r %r " % ( old, dict(old) ) )
        new = AsciiCSV(None, new_)
        new.read()
        log.info( "newcsv %r %r " % ( new, dict(new) ) )
        pass
        assert len(new) >= len(old) , "LOCALSEQNO cannot shrink "  
 


def test_diddle():
    cat = AsciiCat("~/dybaux/catalog/tmp_offline_db")
    diddle = False
    if diddle:
        now = datetime.utcnow()
        for pk,row in cat['CableMapVld'].items():
            row['INSERTDATE'] = now.strftime("%Y-%m-%d %H:%M:%S")
    #print cat.allseqno
    print cat.fabseqno
    #cat.write( "/tmp/demo" )  
    #cat.diff( "/tmp/demo" )


def test_merge():

    old_ = """TABLENAME char(64),LASTUSEDSEQNO int(11),PRIMARY KEY (TABLENAME)
"*",0
"CableMap",440
"CalibFeeSpec",113
"CalibPmtSpec",29
"FeeCableMap",3
"HardwareID",358
"""

    new_ = """TABLENAME char(64),LASTUSEDSEQNO int(11),PRIMARY KEY (TABLENAME)
"*",0
"CableMap",1440
"HardwareID",1358
"Lettuce",100
"Tomato",200
"CalibFeeZiggy",300
"""

    old = AsciiCSV( None, None )( old_ )
    print "old", old.lastusedseqno
    print "old content " + "*" * 100 
    print old.content()

    new = AsciiCSV( None, None )( new_ )
    print "new", new.lastusedseqno
    print "new content " + "*" * 100 
    print new.content()

    mrga = old.merged( new )
    print "mrga ", mrga.lastusedseqno
    print "mrga content " + "*" * 100 
    print mrga.content()

    mrgb = new.merged( old )    
    print "mrgb ", mrgb.lastusedseqno
    print "mrgb content " + "*" * 100 
    print mrgb.content()

    ## merge direction matters ... in the CableMap and HardwareID values 

    om = DD( old.lastusedseqno, mrga.lastusedseqno , name="om", increments=True )
    print om 



def test_fake_write():
    cat = AsciiCat("~/dybaux/catalog/tmp_offline_db")
    cat.fake_write('HardwareID', 10 )
    cat.write()


def test_compare_csv():
    old = AsciiCSV( None, "~/dybaux/catalog/tmp_offline_db/CalibPmtSpec/CalibPmtSpec.csv.old")()
    new = AsciiCSV( None, "~/dybaux/catalog/tmp_offline_db/CalibPmtSpec/CalibPmtSpec.csv.new")()
    assert old == new   ## True
    ## old.write()     standard-orderized 


def test_update_relative_to_target():
    from DybPython import DB
    target = DB("tmp_offline_db")
    cat = AsciiCat("~/dbicopy/tmp_offline_db", skip_pay_check=True, allow_partial=True)
    tselect = [tn]

    log.info("   cat.seqno=%r" % cat.seqno)
    log.info("target.seqno=%r" % target.seqno)
    log.info("   cat.allseqno=%r" % cat.allseqno)
    #log.info("target.allseqno=%r" % target.allseqno)

    updates = cat.updates( target , tselect=tselect , fastforward=False )   
    print "asciicat updates relative to target %s " % updates
    



catd = "~/dbicopy/tmp_offline_db"
tmpd = "/tmp/tdbicopy"

def test_copy_cat():
    """
    #. scrubs any pre-existing tmpd 
    #. copies asciicat into tmpd
    #. checks duplication
    """
    import shutil
    cat = AsciiCat(catd, skip_pay_check=True, allow_partial=True)
    if os.path.isdir(tmpd):
        log.info("removing %s " % tmpd )
        shutil.rmtree(tmpd)
    pass
    tn = 'PhysAd'     # conveniently small table
    cat[tn]()         # reads in payload CSV  
    cat.write(tmpd)   # info file system Ascii Catalog

    tmp = AsciiCat(tmpd, skip_pay_check=True, allow_partial=True)
    tmp[tn]()         # reads in payload CSV  
    assert cat == tmp 

def test_fake_write():
    """
    Requires test_copy_cat run before to create sacrificial AsciiCat to diddle with 

    #. fake write in memory 
    #. inplace write in filesystem tmpd
    """
    pass
    tmp = AsciiCat(tmpd, skip_pay_check=True, allow_partial=True)
    tn = 'PhysAd'
    tmp[tn]()         # reads in payload CSV  
    tmp.fake_write( tn , npay=1 )
    tmp.write()
    for tn,csv in tmp.items():
        log.info("csv %s %s " % ( tn, csv ))
        for pk,v in csv.items():
            log.info("pk %s v %s " % ( pk, v ))

def test_fake_updates():
    """
    #. for updates call the target slot (argument 1) `cat` is required to be at/behind base `tmp` slot (argument 0)
    """
    cat = AsciiCat(catd, skip_pay_check=True, allow_partial=True)
    tmp = AsciiCat(tmpd, skip_pay_check=True, allow_partial=True)
    tn = 'PhysAd'
    updates = tmp.updates( cat , tselect=[tn] , fastforward=False )   
    print updates

def test_dump_cat():
    cat = AsciiCat(basd, skip_pay_check=True, allow_partial=True)
    log.info("dumping cat %r " % cat )
    for tn,csv in cat.items():
        log.info("csv %s %s " % ( tn, csv ))
        for pk,v in csv.items():
            log.info("pk %s v %s " % ( pk, v ))


def test_partial_rloadcat():
    """
    TODO:

    #. make re-testable : by controlling DB status (use fixture control?), 
    #. currently does nothing after 1st usage as detects nothing to do  
    """
    tn = 'PhysAd'
    _tmpd = tmpd
    cmd = "db.py --noconfirm --ALLOW_PARTIAL -t %(tn)s tmp_offline_db rloadcat %(_tmpd)s " % locals()   
    print cmd


if __name__ == '__main__':

    logging.basicConfig(level=logging.INFO)
    #test_dump_cat()
    test_copy_cat()
    test_fake_write()           
    test_fake_updates()
    test_partial_rloadcat()