featured – Acodemics

Time Bound Python Queue

An extension of the standard Python queue that only stores elements for a given number of seconds before being removed. Useful if you don’t know the volume of data being added to the queue but need to limit it in some way.

There is no decrease in performance when removing items.

Without item removal

Queue size: 10000

real 0m10.771s
user 0m0.217s
sys 0m0.028s

With item removal

Queue size: 4659

real 0m10.765s
user 0m0.203s
sys 0m0.028s

#!/usr/bin/python

"""A Time bounded queue, can use any Queue class as a base as long as it follows the standard Python Queue format (http://docs.python.org/2/library/queue.html)."""

from collections import deque
from time import time

from Queue import Queue

class TimeBoundQueue(Queue):

    def __init__(self, max_time_seconds):
        '''Elements older than max_time_seconds will be removed from the queue.'''
        Queue.__init__(self)
        self.timeout = max_time_seconds
        self.timeout_count = 100

    def _put(self, item):
        '''Add element to queue. Creates an internal tuple of (timestamp, element), also triggers removal
            of timed out elements every 100 adds.'''
        self.timeout_count -= 1
        if self.timeout_count <= 0:
            self.__timeout()
            self.timeout_count = 100
        elem = (time(), item)
        self.queue.append(elem)

    def _get(self):
        return self.queue.popleft()[1]

    def __timeout(self):
        '''Remove oldest elements up to time()-max_time_seconds.'''
        valid_timestamp = time() - self.timeout
        completed = False
        while not completed:
            try:
                timestamp, item = self.queue[0]
                if timestamp < valid_timestamp:
                    # Quicker to pop than to remove a value at random position.
                    self.queue.popleft()
                else:
                    completed = True
            except IndexError as e:
                # No elements left
                return

if __name__ == "__main__":

    from time import sleep

    t_queue = TimeBoundQueue(5)

    for i in range(0, 10000):
        t_queue.put(i)
        sleep(0.001)

    print "Queue size: {0}".format(t_queue.qsize())

#!/usr/bin/python

"""A Time bounded queue, can use any Queue class as a base as long as it follows the standard Python Queue format (http://docs.python.org/2/library/queue.html)."""

from collections import deque

from time import time

from Queue import Queue

class TimeBoundQueue(Queue):

def __init__(self, max_time_seconds):

'''Elements older than max_time_seconds will be removed from the queue.'''

Queue.__init__(self)

self.timeout = max_time_seconds

self.timeout_count = 100

def _put(self, item):

'''Add element to queue. Creates an internal tuple of (timestamp, element), also triggers removal

of timed out elements every 100 adds.'''

self.timeout_count -= 1

if self.timeout_count <= 0:

self.__timeout()

self.timeout_count = 100

elem = (time(), item)

self.queue.append(elem)

def _get(self):

return self.queue.popleft()[1]

def __timeout(self):

'''Remove oldest elements up to time()-max_time_seconds.'''

valid_timestamp = time() - self.timeout

completed = False

while not completed:

try:

timestamp, item = self.queue[0]

if timestamp < valid_timestamp:

# Quicker to pop than to remove a value at random position.

self.queue.popleft()

else:

completed = True

except IndexError as e:

# No elements left

return

if __name__ == "__main__":

from time import sleep

t_queue = TimeBoundQueue(5)

for i in range(0, 10000):

t_queue.put(i)

sleep(0.001)

print "Queue size: {0}".format(t_queue.qsize())

Unit Testing and Code Coverage Results in Jenkins CI

August 16, 2013 / Jack Concanon / 0 Comments

Unit testing is great, code coverage is great and continuous integration servers are great; but what isn’t great is combining all three into something useful!

If you are coding in Java then code coverage is fantastically simple but what if you use C++ and GoogleTest and want the same level of graphification? Luckily it is also simple if you know how – so you’re in the right place.

The scenario here is that you have a C++ project being built on a Jenkins/Hudson continuous integration setup and you also use Google Test for your unit tests. The software required for this is GCC 4.7 which should contain a tool called gcov, this will create the actual coverage files for your executable but will require some additional linker and library options to be added to your build scripts or make files. The final piece of software required is called gcovr, this is a Python script that is designed to read the output files from gcov linked executables and convert it into an XML format that can be read by the Cobertura plugin in Jenkins/Hudson.

It is a good idea to add a separate make target for coverage as it requires the executables to be built with no optimisation and a lot of debug symbols, so not very useful for production environments.

Add the following to your CPPFLAGS –

-fprofile-arcs -ftest-coverage -O0

1	-fprofile-arcs -ftest-coverage -O0

And this to your LDFLAGS

-lgcov

-lgcov

So for example you could have a section in your makefile that looks like this (assuming the test target builds your unit test.) –

coverage: CPPFLAGS += -fprofile-arcs -ftest-coverage -O0
coverage: LDFLAGS += -lgcov
coverage: test

coverage: CPPFLAGS += -fprofile-arcs -ftest-coverage -O0

coverage: LDFLAGS += -lgcov

coverage: test

After you have run the coverage build target you then have to run the test executable, this generates .gcov files that describe which lines of code are covered and by what functions.

Then ensure that you can access the gcovr script and run it with the following arguments –

./test
chmod a+x ./gcovr
./gcovr --gcov-exclude=".*gtest.*" -x -d &gt; coverage.xml

./test

chmod a+x ./gcovr

./gcovr --gcov-exclude=".*gtest.*" -x -d > coverage.xml

The –gcov_exlcude will remove any unnecessary parsing of files from the GoogleTest framework. The output is placed into the coverage.xml file which can then be accessed from Jenkins by adding a new Post-Build Action from the job configuration page.

Vinaigrette Sort

May 15, 2013 / Jack Concanon / 0 Comments

When programming in C for embedded systems where memory, disk usage and CPU usage are all very strict there are several ‘rules’ in place such as not use dynamically allocated arrays; this ensures that you can retain more control over memory management.

In a recent experience I had a fixed size array which was always populated below capacity, this array needed to be sorted which would normally leave my zero initialised elements at the start of my array and mess up my loop logic further on in the program.

I had an idea that I could use my sort comparison function to do two things – the first is the original sort purpose of ordering my elements, the second would be to move all unused elements to the end of the array.

Guess what, it’s very simple! I present the vinaigrette sort (so called because the oil and vinegar separate into layers in a vinaigrette).

/* Vinaigrette sort example.
   Author: Jack Concanon
   Date:   14/05/2013
   Website: www.acodemics.co.uk
*/

#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <time.h>

const int BUF_SIZE = 14;
const int NUM_SENSORS = 7;

typedef struct {
    unsigned int id;
    float value;
} Data;

void printBuffer(Data *buf, unsigned size);
int compare(const void *elem1, const void *elem2);

int main(int argc, char* argv[]) {

    Data d[BUF_SIZE];
    memset(d, 0, sizeof(Data)*BUF_SIZE); // Important to zero the array.
    srand(time(NULL));

    unsigned i;
    for(i = 0; i < NUM_SENSORS; ++i) {
        d[i].id = rand(); // random ID number.
        d[i].value = rand(); // random sample value.
    }

    printf("-------- Unsorted array --------\n");
    printBuffer(d, BUF_SIZE);

    // The Magic
    // Sort the d array by ID but at the same time
    //   bubble empty elements to the end.
    qsort(d, BUF_SIZE, sizeof(Data), compare);

    printf("\n\n-------- Sorted By ID --------\n");
    printBuffer(d, BUF_SIZE); 

    return 0;
}

int compare(const void *elem1, const void *elem2) {
    const Data *da = elem1;
    const Data *db = elem2;
    if(db->id == 0) { return -1; }  // Move empty elements to the end
    //if(da->id == 0) { return 1; } // Force empty structs to bubble to beginning.

    if(da->id > db->id) { return 1; }
    if(da->id < db->id) { return -1; }
    return 0;
}

void printBuffer(Data *buf, unsigned size) {
    // loop 'size' times over buf and print.
    unsigned i = 0;
    for(i = 0; i < size; ++i) {
        printf("%d %f\n", buf[i].id, buf[i].value);
    }
}

/* Vinaigrette sort example.

Author: Jack Concanon

Date: 14/05/2013

Website: www.acodemics.co.uk

#include <stdio.h>

#include <stdlib.h>

#include <string.h>

#include <time.h>

const int BUF_SIZE = 14;

const int NUM_SENSORS = 7;

typedef struct {

unsigned int id;

float value;

} Data;

void printBuffer(Data *buf, unsigned size);

int compare(const void *elem1, const void *elem2);

int main(int argc, char* argv[]) {

Data d[BUF_SIZE];

memset(d, 0, sizeof(Data)*BUF_SIZE); // Important to zero the array.

srand(time(NULL));

unsigned i;

for(i = 0; i < NUM_SENSORS; ++i) {

d[i].id = rand(); // random ID number.

d[i].value = rand(); // random sample value.

}

printf("-------- Unsorted array --------\n");

printBuffer(d, BUF_SIZE);

// The Magic

// Sort the d array by ID but at the same time

// bubble empty elements to the end.

qsort(d, BUF_SIZE, sizeof(Data), compare);

printf("\n\n-------- Sorted By ID --------\n");

printBuffer(d, BUF_SIZE);

return 0;

}

int compare(const void *elem1, const void *elem2) {

const Data *da = elem1;

const Data *db = elem2;

if(db->id == 0) { return -1; } // Move empty elements to the end

//if(da->id == 0) { return 1; } // Force empty structs to bubble to beginning.

if(da->id > db->id) { return 1; }

if(da->id < db->id) { return -1; }

return 0;

}

void printBuffer(Data *buf, unsigned size) {

// loop 'size' times over buf and print.

unsigned i = 0;

for(i = 0; i < size; ++i) {

printf("%d %f\n", buf[i].id, buf[i].value);

}

It is as simple as modifying the compare function required by the qsort function. Check if a field in your struct (or your data) is zero and therefore unused and then declare it as ‘greater’ than any comparison. I did find that an older GCC version (3.7) required the line as described in the above source code, GCC 4.7 compiled and ran as required with just the one comparison.

Note the memset line, if you leave the array uninitialised then the compare won’t have anything defined to search for that would signify an unused element.

Code Golf Progress Bar

April 3, 2012 / Jack Concanon / 0 Comments

Had a go at some code golf today, went for this challenge: http://codegolf.stackexchange.com/questions/5382/create-an-ascii-progress-bar

Here is my solution which is 125 characters small.

a,b=raw_input().split()
a=int(a)
b=float(b)
c=int(a*b)
d="|"*c+" "*(a-c)
e=len(d)/2
print"["+d[:e]+`int(b*100)`+"%"+d[e:]+"]"

a,b=raw_input().split()

a=int(a)

b=float(b)

c=int(a*b)

d="|"*c+" "*(a-c)

e=len(d)/2

print"["+d[:e]+`int(b*100)`+"%"+d[e:]+"]"

So the output for this would be as such:

jack@Icarus:~/Projects/code golf$ python progress_bar.py 
10 0.7
[|||||70%||   ]
jack@Icarus:~/Projects/code golf$ python progress_bar.py 
10 0.56
[|||||56%     ]
jack@Icarus:~/Projects/code golf$ python progress_bar.py 
10 0.8
[|||||80%|||  ]

jack@Icarus:~/Projects/code golf$ python progress_bar.py

10 0.7

[|||||70%|| ]

jack@Icarus:~/Projects/code golf$ python progress_bar.py

10 0.56

[|||||56% ]

jack@Icarus:~/Projects/code golf$ python progress_bar.py

10 0.8

[|||||80%||| ]

Categories

Meta

Tagfeatured

Time Bound Python Queue

Unit Testing and Code Coverage Results in Jenkins CI

Vinaigrette Sort

Code Golf Progress Bar

Recent Comments