gistfile1.py

import random
import ctypes
import operator
import cProfile
import pstats
import os
import argparse
from time import time
import numpy
from numpy import frombuffer, complex64, empty, mean, ctypeslib

SAMPLE_SIZE = 1*10**6
RUNS = 1

isamples = None
fsamples = None
data = None

def build_data():
    global isamples, fsamples
    isamples = ctypes.c_byte * SAMPLE_SIZE
    fsamples = ctypes.c_float * SAMPLE_SIZE

    return [isamples(*[random.randrange(0, 255) for i in xrange(SAMPLE_SIZE)]) for j in xrange(RUNS)]



def original(b):
    # Reflect current implementation in rtlsdr master
    iq = empty(len(b)//2, 'complex')
    iq.real, iq.imag = b[::2], b[1::2]
    iq /= (255/2)
    iq -= (1 + 1j)
    return iq


def masse1234(b):
    b_np = ctypeslib.as_array(b)
    iq = empty(len(b)//2, complex64)
    iq.real, iq.imag = b_np[::2], b_np[1::2]
    iq /= (255/2)
    iq -= (1 + 1j)
    return iq


def sliceassign(b):
    f = fsamples()
    f[:] = b
    c = frombuffer(f, complex64)
    c /= (255/2)
    c -= (1 + 1j)
    return c


def starassign(b):
    f = fsamples(*b)
    c = frombuffer(f, complex64)
    c /= (255/2)
    c -= (1 + 1j)
    return c

def pijyoi(b):
    bytes_np = ctypeslib.as_array(b)
    iq = bytes_np.astype(numpy.float32).view(complex64)
    iq /= (255/2)
    iq -= (1 + 1j)
    return iq

def profile_run(func):
    pr = cProfile.Profile()
    results = []
    for b in data:
        pr.enable()
        r = func(b)
        pr.disable()
        results.append(b)
    sort_by = 'cumulative'
    ps = pstats.Stats(pr)
    ps.strip_dirs()
    ps.sort_stats(sort_by)
    print(func.__name__)
    print('-'*80)
    ps.print_stats()
    print('-'*80)
    return ps, results

def run():
    global data
    if data is None:
        data = build_data()
    results = []
    stats = {}
    for test in original, masse1234, sliceassign, starassign, pijyoi:
        ps, r = profile_run(test)
        results.append(r)
        stats[test.__name__] = ps
    return stats, results

if __name__ == '__main__':
    p = argparse.ArgumentParser()
    p.add_argument('-d', dest='data_dir', help='Path for dumping profile results')
    p.add_argument('--sample-size', dest='sample_size')
    p.add_argument('--runs', dest='runs')
    args = p.parse_args()
    if args.sample_size:
        SAMPLE_SIZE = int(args.sample_size)
    if args.runs:
        RUNS = int(args.runs)
    print "starting test: sample_size={}, runs={}".format(SAMPLE_SIZE, RUNS)
    stats, results = run()
    if args.data_dir:
        for testname, ps in stats.items():
            fn = os.path.join(args.data_dir, '{}.pstats'.format(testname))
            print('dumping stats: {}'.format(fn))
            ps.dump_stats(fn)