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 <ed****@majakari.net>
"""
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
r = RRArray(10) # creates a round-robin array of length 10
len(r) 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
r = RRArray(3)
r.insert(42)
r.most_recent() == 42 and r.used_slots() == 1 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
