本気のFizzBuzz -> [移動した](https://github.com/macrat/fizzbuzz-framework)

fizzbuzz.py

""" 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(),
    ))