Last active
October 19, 2019 12:56
-
-
Save macrat/dd77d11f9b09aeec681344b35c1ec558 to your computer and use it in GitHub Desktop.
本気のFizzBuzz -> [移動した](https://github.com/macrat/fizzbuzz-framework)
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
""" 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