Skip to content

Instantly share code, notes, and snippets.

@macrat
Last active Oct 19, 2019
Embed
What would you like to do?
本気のFizzBuzz -> [移動した](https://github.com/macrat/fizzbuzz-framework)
""" FizzBuzzのフレームワーク
本気でFizzBuzzを作ったら、パイプライン処理のライブラリが出来た。
"""
from typing import Any, List
class EndOfList(Exception):
""" 処理を終了したいときに送出するエラー """
pass
class Operator:
""" 値を生成したり、加工したり、出力したりするもの
オペレータは1個の入力値を受け取り、0個以上の出力値を返す。
もしくは、これ以上処理しないことを示す EndOfList を送出する。
複数のオペレータを OperatorChain を使って繋ぐことで、プログラムの処理を表現する。
"""
def execute(self, value: Any) -> List[Any]:
""" 何らかの値を受け取り、何らかの値のリストを返す関数
:param value: 上流のオペレータが作った入力値。無視しても良い。
:return: 下流のオペレータに渡される出力値のリスト。
"""
raise NotImplementedError()
class OperatorChain(Operator):
""" 複数のオペレータを繋いで、一個のオペレータにしたもの
>>> class TestOperator(Operator): # テスト用の値を倍にするオペレータ
... def execute(self, x):
... return [x * 2]
一つのオペレータだけを与えた場合は、与えられたオペレータと同じ挙動になる。
>>> TestOperator().execute(2)
[4]
>>> OperatorChain(TestOperator()).execute(2)
[4]
二つ以上のオペレータを渡すと、与えられた順に実行した結果を返す。
>>> OperatorChain(TestOperator(), TestOperator()).execute(2)
[8]
一つもオペレータが与えられなかった場合、入力をそのまま出力するオペレータになる。
>>> OperatorChain().execute(2)
[2]
"""
def __init__(self, *operators: Operator) -> None:
"""
:param operators: 繋ぎたいオペレータ。引数に与えた順番で実行される。
"""
self.operators = operators
def _unfoldExecute(self, operator: Operator, values: List[Any]) -> List[Any]:
""" 複数の値の分だけオペレータを実行し、結果を結合して返す
:param operator: 実行したいオペレータ。
:param values: オペレータに渡したい引数のリスト。
:return: 全部の値についてオペレータを実行した結果を結合したリスト。
>>> class TestOperator(Operator): # テスト用の値を2回繰り返すオペレータ
... def execute(self, x):
... return [x, x]
[1, 2, 3] を渡すと、それぞれの値が2回ずつ繰り返した物が返ってくる。
>>> OperatorChain()._unfoldExecute(TestOperator(), [1, 2, 3])
[1, 1, 2, 2, 3, 3]
"""
result = []
for v in values:
r = operator.execute(v)
try:
result.extend(r)
except TypeError:
pass # オペレータがリストを返さないのは許容してあげよう
return result
def execute(self, value: Any) -> List[Any]:
""" 結合されたオペレータを実行する
:param value: 入力値。
:return: 出力値のリスト。
"""
val = [value]
for o in self.operators:
val = self._unfoldExecute(o, val)
return val
def execute(*operators: Operator) -> None:
""" オペレータを繋いで、 EndOfList が送出されるまで繰り返し実行する
:param operators: 実行したいオペレータのリスト
>>> execute(
... RangeGenerator(0, 5), # 0から5までの数字を作って、
... ConsolePrinter(), # コンソールに表示する。
... )
0
1
2
3
4
5
最初のオペレータの入力は常に None になる。
>>> class TestPrinter(Operator): # テスト用の5回だけ入力を表示するオペレータ
... def __init__(self):
... self.count = 0
... def execute(self, x):
... self.count += 1
... if self.count > 5:
... raise EndOfList
... print(x)
>>> execute(TestPrinter())
None
None
None
None
None
"""
o = OperatorChain(*operators)
while True:
try:
o.execute(None)
except EndOfList:
break
class SequenceGenerator(Operator):
""" 無限の数列を作るジェネレータ """
def __init__(self, from_: float = 0, step: float = 1) -> None:
"""
:param from_: 最初に出力される数。省略すると0になる。
:param step: 一回の呼び出しで変化する数。省略すると1ずつ増える。
"""
self.current = from_ - step
self.step = step
def execute(self, value: Any) -> List[float]:
""" オペレータを実行する
:param value: 入力値だが、常に無視される。
:return: 昇順の数値。
"""
self.current += self.step
return [self.current]
class LimitFilter(Operator):
""" 指定の回数で処理を停止するフィルタ
>>> execute(
... SequenceGenerator(), # 無限に数列を生成して、
... LimitFilter(5), # 先頭から5つだけ値を取り出し、
... ConsolePrinter(), # コンソールに表示する。
... )
0
1
2
3
4
"""
def __init__(self, num: int) -> None:
"""
:param num: 停止するまでに処理する数。
"""
self.ttl = num
def execute(self, value: Any) -> List[Any]:
""" 実行回数が上限を越えていないか確認して、入力をそのまま出力として出す
:raise EndOfList: 実行回数が上限を越えると送出し、実行を停止させる。
"""
if self.ttl <= 0:
raise EndOfList()
self.ttl -= 1
return [value]
def RangeGenerator(from_: float, to: float, step: float = 1) -> OperatorChain:
""" 指定の数値の間をカウントする数列を生成するジェネレータ
中身は SequenceGenerator と LimitFilter を組み合わせただけのもの。
:param from_: カウントを開始する値。
:param to: カウントを終了する値。
:param step: 一回の実行で増える数。
>>> execute(
... RangeGenerator(1, 8, 2), # 1から8までの2刻みの数字を作って、
... ConsolePrinter(), # コンソールに表示する。
... )
1
3
5
7
"""
return OperatorChain(
SequenceGenerator(from_, step),
LimitFilter(int((max(from_, to) - min(from_, to)) // step + 1)),
)
class ModReplaceFilter(Operator):
""" 指定の値の倍数が入力されたときに値を置換するフィルタ
ほぼFizzBuzz用。
"""
def __init__(self, num: int, replace: Any) -> None:
"""
:param num: 何の倍数のときに置換するか。
:param replace: どんな値に置換するか。
"""
self.num = num
self.replace = replace
def execute(self, value: Any) -> List[Any]:
""" 入力を受け取り、指定の値の倍数が来たら置換する
:param value: 入力値。整数でない場合は割りようがないのでそのまま出力する。
:return: 置換された値、もしくは入力値そのものが入ったリスト。
"""
if isinstance(value, int) and value % self.num == 0:
return [self.replace]
return [value]
class ConsolePrinter(Operator):
""" 入力値をコンソールに出力するプリンタ """
def execute(self, value: Any) -> List[Any]:
""" 入力を受け取り、コンソールに出力する
入力値をそのまま出力するので、チェーンさせることも出来る。
:param value: 入力値。コンソールに出力される。
:return: 入力値をそのまま含む要素数が1つのリスト。
"""
print(value)
return [value]
def FizzBuzzFilter() -> OperatorChain:
""" 入力値に対してFizzBuzz変換掛けるフィルタ
中身は単純に ModReplaceFilter を組み合わせただけ。
"""
return OperatorChain(
ModReplaceFilter(15, "fizzbuzz"),
ModReplaceFilter(3, "fizz"),
ModReplaceFilter(5, "buzz"),
)
class FactorialFilter(Operator):
""" 階乗的な計算をするフィルタ
前回の出力を記憶しておいて、新しい入力と掛けたものを次の出力とする。
"""
def __init__(self) -> None:
self.current: float = 1
def execute(self, value: float) -> List[float]:
""" 階乗的な計算をした結果を出力する
:param value: 入力値。少なくともかけ算が出来る値である必要がある。
:return: 階乗的な計算の結果。
"""
self.current = value * self.current
return [self.current]
if __name__ == "__main__":
execute(OperatorChain(
RangeGenerator(1, 30),
FizzBuzzFilter(),
ConsolePrinter(),
))
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment