RRArray for Python?

[Edvard Majakari]

I was wondering if there is already some 'official' module for round robin
arrays. Of course, it is relatively easy to implement your own if you need it,
but IMHO it feels like there should be one, because they are so
general-purpose.

Here's my implementation, if someone needs one. If there are enough people
supporting it's release, I could add it to some static web address.

"""
Round-robin array module

This module contains class for a round-robin array and associated round-robin
array database to contain long-time spanning sample values (cf. rrdtool).

Arrays can be arbitrary length, and default slot values can be defined by the
user.

Edvard Majakari <[email protected]>
"""

class RRArrayError(Exception): pass

class RRArray:
"""Simple round-robin class for arbitrary items"""

def __init__(self, slots, default=None):
"""Instantiate RRArray object

@param default: value to fill empty slots with
10
"""

self.default = default
self._slots = slots

self._array = [default]*slots

self._next = 0 # points to next free index in array
self._array_full = False

def clear(self):
"""Erase array by setting all values to default fill value"""

self._array = [self.default]*self._slots
self._next = 0
self._array_full = False

def contents(self):
"""Return contents of RRArray object as a list

so that most recent item is the last position.

>>> r = RRArray(3, 0)
>>> r.contents() [0, 0, 0]
>>> r.insert(2)
>>> r.contents()

[0, 0, 2]
"""

retval = self.latest(self._slots)
retval.reverse()

return retval

def insert(self, item):
"""Insert an item to object
True
"""

self._array[self._next % self._slots] = item

if self._next == self._slots - 1:
self._array_full = True
self._next = 0 # wrap-around

else:
self._next += 1

def latest(self, count):
"""Return count most recent items

Note that it is possible to receive default values which were never
inserted, eg.

r = RRArray(5)
r.insert(2), r.latest(3) # would return [2, None, None]

>>> r = RRArray(3)
>>> r.insert(42)
>>> r.insert(5)
>>> r.insert(7)
>>> r.insert(11)
>>> r.latest(2) [11, 7]
>>> r.contents()

[5, 7, 11]
"""

if count > self._slots:
err = "attempted to get %d items from rra of size %d"
raise RRArrayError(err % (count, self._slots))

latest_idx = self._latest_index()

head = self._array[0:latest_idx+1]
head.reverse()

if count >= len(head):
tail = self._array[latest_idx+1:self._slots]
tail.reverse()
head.extend(tail)

return head[0:count]

def most_recent(self):
"""Return most recent item inserted.

An IndexError is raised if no items have been inserted yet.
"""

if self._next == 0 and not self._array_full:
raise IndexError("no items inserted yet")

return self._array[(self._next - 1) % self._slots]

def used_slots(self):
"""Return number of used slots."""

if self._next < self._slots and not self._array_full:
return self._next
else:
return self._slots

### private and special methods

def __len__(self):
"""Return number of slots in the object."""

return self._slots

def _latest_index(self):
return (self._next - 1) % self._slots

def _test():
import doctest, rrarray
doctest.testmod(rrarray)

if __name__ == '__main__':
_test()

--
# Edvard Majakari Software Engineer
# PGP PUBLIC KEY available Soli Deo Gloria!

"Debugging is twice as hard as writing the code in the firstplace. Therefore,
if you write the code as cleverly as possible, you are, by definition,
not smart enough to debug it." -- Brian W. Kernighan