Usability improvements: Reader's now support the iterator protocol; CallbackReader supports a post-read converter

PoolReader: made the pool and task ref functions part of the internal interface, added pool() and task() functions which return the dereferenced weakref directly

Author: Byron

__init__.py

@@ -28,3 +28,9 @@ def thread_interrupt_handler(signum, frame):
 
 _init_atexit()
 _init_signals()
+
+
+# initial imports
+from task import *
+from pool import *
+from channel import *

channel.py

@@ -142,7 +142,24 @@ class Reader(object):
 	#{ Interface
 	def __init__(self, device):
 		"""Initialize the instance with the device to read from"""
+	
+	#{ Iterator protocol
+	
+	def __iter__(self):
+		return self
+
+	def next(self):
+		"""Implements the iterator protocol, iterating individual items"""
+		items = self.read(1)
+		if items:
+			return items[0]
+		raise StopIteration
+		
+	#} END iterator protocol
 
+		
+	#{ Interface
+	
 	def read(self, count=0, block=True, timeout=None):
 		"""read a list of items read from the device. The list, as a sequence
 		of items, is similar to the string of characters returned when reading from 
@@ -160,6 +177,8 @@ def read(self, count=0, block=True, timeout=None):
 			returned."""
 		raise NotImplementedError()
 
+	#} END interface
+		
 
 class ChannelReader(Reader):
 	"""Allows reading from a channel. The reader is thread-safe if the channel is as well"""
@@ -253,6 +272,7 @@ class CallbackReaderMixin(object):
 	def __init__(self, *args):
 		super(CallbackReaderMixin, self).__init__(*args)
 		self._pre_cb = None
+		self._post_cb = None
 
 	def set_pre_cb(self, fun = lambda count: None):
 		"""Install a callback to call with the item count to be read before any 
@@ -264,11 +284,26 @@ def set_pre_cb(self, fun = lambda count: None):
 		prev = self._pre_cb
 		self._pre_cb = fun
 		return prev
+		
+	def set_post_cb(self, fun = lambda items: items):
+		"""Install a callback to call after items have been read, but before 
+		they are returned to the caller. The callback may adjust the items and/or
+		the list
+		If no function is provided, the callback is uninstalled
+		:return: the previously installed function"""
+		prev = self._post_cb
+		self._post_cb = fun
+		return prev
 
 	def read(self, count=0, block=True, timeout=None):
 		if self._pre_cb:
 			self._pre_cb(count)
-		return super(CallbackReaderMixin, self).read(count, block, timeout)
+		items = super(CallbackReaderMixin, self).read(count, block, timeout)
+		
+		if self._post_cb:
+			items = self._post_cb(items)
+		return items
+		
 
 
 class CallbackChannelReader(CallbackReaderMixin, ChannelReader):

pool.py

@@ -71,11 +71,8 @@ def __del__(self):
 	#{ Internal
 	def _read(self, count=0, block=True, timeout=None):
 		return CallbackChannelReader.read(self, count, block, timeout)
-	
-	#} END internal
-
-	#{ Interface
-	
+		
+		
 	def pool_ref(self):
 		""":return: reference to the pool we belong to"""
 		return self._pool_ref
@@ -84,6 +81,27 @@ def task_ref(self):
 		""":return: reference to the task producing our items"""
 		return self._task_ref
 
+	#} END internal
+
+	#{ Interface
+	
+	def task(self):
+		""":return: task we read from
+		:raise ValueError: If the instance is not attached to at task"""
+		task = self._task_ref()
+		if task is None:
+			raise ValueError("PoolReader is not associated with at task anymore")
+		return task
+		
+	def pool(self):
+		""":return: pool our task belongs to
+		:raise ValueError: if the instance does not belong to a pool"""
+		pool = self._pool_ref()
+		if pool is None:
+			raise ValueError("PoolReader is not associated with a pool anymore")
+		return pool
+		
+	
 	#} END interface 
 
 	def read(self, count=0, block=True, timeout=None):

task.py

@@ -52,7 +52,7 @@ def __init__(self, id, fun, apply_single=True, min_count=None, max_chunksize=0,
 		self._wlock = threading.Lock()
 		self.fun = fun
 		self.min_count = None
-		self.max_chunksize = 0				# note set
+		self.max_chunksize = 0				# not set
 		self.apply_single = apply_single
 
 	def is_done(self):

test/test_channel.py

@@ -48,28 +48,34 @@ def test_base(self):
 		# test callback channels
 		wc, rc = mkchannel(wtype = CallbackChannelWriter, rtype = CallbackChannelReader)
 
-		cb = [0, 0]		# set slots to one if called
+		cb = [0, 0, 0]		# set slots to one if called
 		def pre_write(item):
 			cb[0] = 1
 			return item + 1
 		def pre_read(count):
 			cb[1] = 1
+		def post_read(items):
+			assert isinstance(items, list)
+			cb[2] = 1
+			return [ i+1 for i in items]
+			
 
 		# set, verify it returns previous one
 		assert wc.set_pre_cb(pre_write) is None
 		assert rc.set_pre_cb(pre_read) is None
+		assert rc.set_post_cb(post_read) is None
 		assert wc.set_pre_cb(pre_write) is pre_write
 		assert rc.set_pre_cb(pre_read) is pre_read
+		assert rc.set_post_cb(post_read) is post_read
 
 		# writer transforms input
 		val = 5
 		wc.write(val)
 		assert cb[0] == 1 and cb[1] == 0
 
 		rval = rc.read(1)[0]		# read one item, must not block
-		assert cb[0] == 1 and cb[1] == 1
-		assert rval == val + 1
-		
+		assert cb[0] == 1 and cb[1] == 1 and cb[2] == 1
+		assert rval == val + 1 + 1
 
 
 		# ITERATOR READER
@@ -83,5 +89,10 @@ def pre_read(count):
 		# doesn't work if item is not an iterator
 		self.failUnlessRaises(ValueError, IteratorReader, list())
 
+		
+		# test general read-iteration - its supported by all readers
+		reader = IteratorReader(iter(range(10)))
+		assert len(list(reader)) == 10
+		
 		# NOTE: its thread-safety is tested by the pool 
 

test/test_pool.py

@@ -413,6 +413,12 @@ def test_base(self):
 		urc2 = p.add_task(t2)
 		assert p.num_tasks() == 2
 
+		# test pool reader
+		assert urc1.pool_ref()() is p
+		assert urc1.task_ref()() is t1
+		assert urc1.pool() == p
+		assert urc1.task() == t1
+		
 		## SINGLE TASK #################
 		self._assert_single_task(p, False)
 		assert p.num_tasks() == 2