Python Iteratee

Python mini Hack-a-thon

参加しました。

今日はPythonでIterateeを書きました。

IterateeはIOストリームを抽象的に扱うもの・・・らしい。
composableでおもしろいなー
とちょっとした興味でさわってるのでそこまで詳しい訳じゃないです。

とりあえずソースコード。

	class Maybe(object):
	pass
	Nothing = Maybe()
	class Just(Maybe):
	def __init__(self, value):
	self.value = value
	def __eq__(self, other):
	return isinstance(other, Just) and other.value == self.value
	def __str__(self):
	return "Just({0})".format(self.value)

view raw maybe.py hosted with ❤ by GitHub

	from functools import partial
	from maybe import *
	class Input(object):
	def input(self, el, empty, eof):
	if self == Empty:
	return empty
	elif self == EOF:
	return eof
	else:
	return el(self)
	class El(Input):
	def __init__(self, e):
	self.e = e
	Empty = Input()
	EOF = Input()
	class IterV(object):
	def bind_(self, i):
	return self.bind(lambda _: i)
	def file_lines(self, path):
	f = open(path)
	try:
	return self._handle(f)
	finally:
	f.close()
	class Done(IterV):
	def __init__(self, a, i):
	self.a = a
	self.i = i
	def enum(self, i):
	return self
	def run(self):
	return self._maybe()
	def _maybe(self):
	return self.a
	def bind(self, f):
	return f(self.a)._bind(self.i)
	def _bind(self, i):
	return Done(self.a, i)
	def _handle(self, f):
	return self
	class Cont(IterV):
	def __init__(self, k):
	self.k = k
	def enum(self, i):
	if len(i) == 0:
	return self
	else:
	x = i[0]
	xs = i[1:]
	return self.k(El(x)).enum(xs)
	def run(self):
	return self.k(EOF)._maybe()
	def _maybe(self):
	raise Exception()
	def bind(self, f):
	return Cont(lambda i: self.k(i).bind(f))
	def _bind(self, i):
	return self.k(i)
	def _handle(self, f):
	line = f.readline()
	if line:
	return self.k(El(line))._handle(f)
	else:
	return self
	def head():
	def step(i):
	return i.input(el = lambda el: Done(Just(el.e), Empty), empty = Cont(step), eof = Done(Nothing, EOF))
	return Cont(step)
	def peek():
	def step(i):
	return i.input(el = lambda el: Done(Just(el.e), el), empty = Cont(step), eof = Done(Nothing, EOF))
	return Cont(step)
	def drop(n):
	if n == 0:
	return Done((), Empty)
	else:
	def step(i):
	return i.input(el = lambda _: drop(n - 1), empty = Cont(step), eof = Done((), EOF))
	return Cont(step)
	def length():
	def step(acc, i):
	return i.input(el = lambda _: Cont(partial(step, acc + 1)), empty = Cont(partial(step, acc)), eof = Done(acc, EOF))
	return Cont(partial(step, 0))

view raw iteratee.py hosted with ❤ by GitHub

	from unittest import TestCase, main
	from iteratee import *
	class IterateeTest(TestCase):
	def test_input(self):
	self.assertNotEqual(Empty, EOF)
	def test_head(self):
	self.assertEqual(head().enum([1, 2, 3]).run(), Just(1))
	self.assertEqual(head().enum([]).run(), Nothing)
	def test_peek(self):
	self.assertEqual(peek().enum([1, 2, 3]).run(), Just(1))
	self.assertEqual(peek().enum([]).run(), Nothing)
	def test_drop(self):
	self.assertEqual(drop(1).enum([1, 2, 3]).run(), ())
	def test_length(self):
	self.assertEqual(length().enum([1, 2, 3]).run(), 3)
	self.assertEqual(length().enum([]).run(), 0)
	def test_bind_(self):
	self.assertEqual(drop(0).bind_(peek()).enum([1, 2, 3]).run(), Just(1))
	self.assertEqual(drop(1).bind_(head()).enum([1, 2, 3]).run(), Just(2))
	self.assertEqual(peek().bind_(peek()).enum([1, 2, 3]).run(), Just(1))
	self.assertEqual(head().bind_(peek()).enum([1, 2, 3]).run(), Just(2))
	def test_file_lines(self):
	self.assertEqual(head().file_lines('test').run(), Just('hello\n'))
	self.assertEqual(drop(2).bind_(head()).file_lines('test').run(), Just('python\n'))
	self.assertEqual(drop(1).bind_(length()).file_lines('test').run(), 2)
	if __name__ == '__main__':
	main()

view raw test_iteratee.py hosted with ❤ by GitHub

	hello
	world
	python

view raw test hosted with ❤ by GitHub

これで何が嬉しいかというと

• 処理が合成できる
• IOストリームだけでなくリストなども処理できる

他にもハイパフォーマンスで型安全という特徴があるらしいのですが、この実装だと遅いし、Pythonだと型安全とか意味をなさないのであまり旨味がないかなあと思いました。