python-app-rocketbot/rscollections.py

"""
Subclasses of list, set, and dict with maximum sizes.
"""

class BoundList(list):
	"""
	Subclass of `list` that enforces a maximum size.

	If the number of elements following a mutating operation exceeds
	`self.max_element_count`, then each element will be tested for its "age,"
	and the "oldest" elements will be removed until the total size is back
	within the limit. "Age" is determined via a provided lambda function. It is
	a value with arbitrary numeric value that is used comparitively to find
	the element with the smallest value. If no `age_test` is provided, the
	default is to remove elem 0.

	The age lambda takes two arguments: the element index and the element value.
	It must return an int or float.

	`self.age_test` and `self.max_element_count` can be modified at runtime,
	however elements will only be discarded following the next mutating operation.
	"""
	def __init__(self, max_element_count = 10, age_test = (lambda index, elem: index), *args, **kwargs):
		super().__init__(*args, **kwargs)
		self.age_test = age_test
		self.max_element_count = max_element_count

	def __cull_old_elements(self):
		while len(self) > self.max_element_count:
			oldest_age = None
			oldest_index = -1
			for i in range(len(self)):
				elem = self[i]
				age = self.age_test(i, elem)
				if oldest_age is None or age < oldest_age:
					oldest_age = age
					oldest_index = i
			self.pop(oldest_index)

	# Passthrough methods

	def __add__(self, *args):
		ret_val = super().__add__(*args)
		self.__cull_old_elements()
		return ret_val
	def __iadd__(self, *args):
		ret_val = super().__iadd__(*args)
		self.__cull_old_elements()
		return ret_val
	def append(self, *args):
		ret_val = super().append(*args)
		self.__cull_old_elements()
		return ret_val
	def copy(self):
		return BoundList(self.max_element_count, self.age_test, super())
	def extend(self, *args):
		ret_val = super().extend(*args)
		self.__cull_old_elements()
		return ret_val
	def insert(self, *args):
		ret_val = super().insert(*args)
		self.__cull_old_elements()
		return ret_val

class BoundSet(set):
	"""
	Subclass of `set` that enforces a maximum size.

	Same behavior as `BoundList`.

	`age_test` lambda function takes one argument, the element value.
	"""
	def __init__(self, max_element_count = 10, age_test = (lambda elem: float(elem)), *args, **kwargs):
		super().__init__(*args, **kwargs)
		self.age_test = age_test
		self.max_element_count = max_element_count

	def __cull_old_elements(self):
		while len(self) > self.max_element_count:
			oldest_age = None
			oldest_elem = None
			for elem in self:
				age = self.age_test(elem)
				if oldest_age is None or age < oldest_age:
					oldest_age = age
					oldest_elem = elem
			self.remove(oldest_elem)

	def add(self, *args):
		ret_val = super().add(*args)
		self.__cull_old_elements()
		return ret_val
	def copy(self):
		return BoundSet(self.max_element_count, self.age_test, super())
	def update(self, *args):
		ret_val = super().update(*args)
		self.__cull_old_elements()
		return ret_val

class BoundDict(dict):
	"""
	Subclass of `dict` that enforces a maximum size.

	Same behavior as `BoundList`.

	`age_test` lambda takes two arguments: the key and the value of a pair.
	"""
	def __init__(self, max_element_count = 10, age_test = (lambda key, value: float(key)), *args, **kwargs):
		super().__init__(*args, **kwargs)
		self.age_test = age_test
		self.max_element_count = max_element_count

	def __cull_old_elements(self):
		while len(self) > self.max_element_count:
			oldest_age = None
			oldest_key = None
			for key, value in self.items():
				age = self.age_test(key, value)
				if oldest_age is None or age < oldest_age:
					oldest_age = age
					oldest_key = key
			del self[oldest_key]
	def __ior__(self, *args):
		ret_val = super().__ior__(*args)
		self.__cull_old_elements()
		return ret_val
	def __setitem__(self, *args):
		super().__setitem__(*args)
		self.__cull_old_elements()
	def copy(self):
		return BoundDict(self.max_element_count, self.age_test, super())
	@classmethod
	def fromkeys(cls, iter, value=None):
		raise RuntimeError('fromkeys not available')
	def update(self, *args):
		ret_val = super().update(*args)
		self.__cull_old_elements()
		return ret_val