pytables example

pyt.py

import random
import tables
import time

import os.path
from pprint import pprint

def timer(N=1):
    '''timing decorator'''
    def real_deca(func):
        def wrapper(*args, **kws):
            ts = time.time()
            for i in range(N):
                res = func(*args, **kws)
            e = time.time() - ts
            print('{} -> Did {} runs, avg {}, total {}'.format(
                func.__name__,
                N,
                e/N,
                e
                ))
            return res
        return wrapper
    return real_deca

class Data(tables.IsDescription):
    
    name = tables.StringCol(16)
    num  = tables.Int64Col()

def build(n = 10):
    '''create random data set of length = n'''
    ds = [] # data set
    fs = [] # filter set
    for i in range(n):
        ds.append({
            'name':''.join(chr(random.randrange(97,120)) for _ in range(15)),
            'num':random.randrange(1000,9999999)
            })
    for i in range(n//2):
        d = ds[i]
        ds.append({
            'name':d['name'],
            'num':random.randrange(1000,9999999),
            })        
    return ds

def build_filter(ds):
    '''build random filter set from ds'''
    random.shuffle(ds)
    fs = set()
    for i in range(len(ds)//4):
        fs.add(ds[i]['name'])

    random.shuffle(ds)
    return fs
    
def write_data(dat):
    '''write data set into pytable'''
    fn = os.path.normpath('c:/tmp/test.h5')

    h5file = tables.open_file(fn, mode = "w", title = "Testing")
    group = h5file.create_group("/", 'log', 'log information')
    table = h5file.create_table(group, 'dd', Data, "Readout example")
    d = table.row

    for i in dat:
        d['name'] = i['name']
        d['num'] =  i['num']
        d.append()

    table.cols.name.create_index()
    table.cols.num.create_index()
    h5file.close()

@timer()
def xc_mem(ds, fs):
    '''do compare with both data sets in memory'''
    res = []
    for nm in fs:
        candi = [d['num'] for d in ds if d['name'] == nm]
        res.append({'name':nm, 'num':max(candi)})
    return res

@timer()
def xc_hd5(fs):
    '''do compare with data set in pytables'''
    fn = os.path.normpath('c:/tmp/test.h5')
    h5file = tables.open_file(fn)
    tbl = h5file.root.log.dd

    res = []
    for nm in fs:
        q = 'name == b"{}"'.format(nm)
        candi = [d['num'] for d in tbl.read_where(q)]
        res.append({'name':nm, 'num':max(candi)})

    return res
        

def main():

    
    ds = build(10000)
    fs = build_filter(ds)
    
    print('data set {}, filter set {}'.format(len(ds),len(fs)))
    write_data(ds)
    hr = xc_hd5(fs)
    mr = xc_mem(ds, fs)
    print(mr == hr)
    
    
    
main()