ジェネレータを使った反復処理のカスタマイズ

はじめに

この実験では、Python のジェネレータを使用して反復処理をカスタマイズする方法を学びます。また、カスタムクラスでイテレータ機能を実装し、ストリーミングデータソース用のジェネレータを作成します。

実験中に structure.py ファイルを変更し、follow.py という名前の新しいファイルを作成します。

Python ジェネレータの理解

ジェネレータは Python の強力な機能です。イテレータを作成する簡単でエレガントな方法を提供します。Python でデータシーケンスを扱う場合、イテレータは非常に便利で、一連の値を 1 つずつループ処理することができます。通常の関数は、一般的に単一の値を返してから実行を停止します。しかし、ジェネレータは異なります。時間をかけて一連の値を生成することができ、つまり、段階的に複数の値を生成することができます。

ジェネレータとは何か？

ジェネレータ関数は通常の関数に似ていますが、値を返す方法が大きく異なります。通常の関数が return 文を使って単一の結果を返すのに対し、ジェネレータ関数は yield 文を使います。yield 文は特殊な文で、実行されるたびに関数の状態が一時停止し、yield キーワードの後に続く値が呼び出し元に返されます。ジェネレータ関数が再度呼び出されると、中断したところから実行を再開します。

まずは簡単なジェネレータ関数を作成してみましょう。Python の組み込み関数 range() は小数ステップをサポートしていません。そこで、小数ステップで数値の範囲を生成できるジェネレータ関数を作成します。

1. まず、WebIDE で新しい Python ターミナルを開く必要があります。これを行うには、「Terminal」メニューをクリックし、「New Terminal」を選択します。
2. ターミナルが開いたら、以下のコードをターミナルに入力します。このコードはジェネレータ関数を定義し、それをテストします。

def frange(start, stop, step):
    current = start
    while current < stop:
        yield current
        current += step

## Test the generator with a for loop
for x in frange(0, 2, 0.25):
    print(x, end=' ')

このコードでは、frange 関数がジェネレータ関数です。current 変数を start 値で初期化します。そして、current が stop 値より小さい限り、current 値を生成し、current を step 値だけ増やします。for ループは frange ジェネレータ関数が生成する値を反復処理し、それらを出力します。

以下の出力が表示されるはずです。

0 0.25 0.5 0.75 1.0 1.25 1.5 1.75

ジェネレータの使い切り性

ジェネレータの重要な特性は、使い切り可能であるということです。つまり、ジェネレータが生成するすべての値を反復処理した後は、同じ値のシーケンスを再度生成するために使用することはできません。以下のコードでこれを実証してみましょう。

## Create a generator object
f = frange(0, 2, 0.25)

## First iteration works fine
print("First iteration:")
for x in f:
    print(x, end=' ')
print("\n")

## Second iteration produces nothing
print("Second iteration:")
for x in f:
    print(x, end=' ')
print("\n")

このコードでは、まず frange 関数を使ってジェネレータオブジェクト f を作成します。最初の for ループはジェネレータが生成するすべての値を反復処理し、それらを出力します。最初の反復処理の後、ジェネレータは使い切られています。つまり、生成できるすべての値をすでに生成しています。そのため、2 番目の for ループで再度反復処理を試みると、新しい値は生成されません。

出力結果：

First iteration:
0 0.25 0.5 0.75 1.0 1.25 1.5 1.75

Second iteration:

2 回目の反復処理では出力がないことに注意してください。これは、ジェネレータがすでに使い切られているためです。

クラスを使った再利用可能なジェネレータの作成

同じ値のシーケンスを複数回反復処理する必要がある場合は、ジェネレータをクラスでラップすることができます。こうすることで、新しい反復処理を開始するたびに、新しいジェネレータが作成されます。

class FRange:
    def __init__(self, start, stop, step):
        self.start = start
        self.stop = stop
        self.step = step

    def __iter__(self):
        n = self.start
        while n < self.stop:
            yield n
            n += self.step

## Create an instance
f = FRange(0, 2, 0.25)

## We can iterate multiple times
print("First iteration:")
for x in f:
    print(x, end=' ')
print("\n")

print("Second iteration:")
for x in f:
    print(x, end=' ')
print("\n")

このコードでは、FRange クラスを定義しています。__init__ メソッドは start、stop、step の値を初期化します。__iter__ メソッドは Python クラスの特殊メソッドで、イテレータを作成するために使用されます。__iter__ メソッドの中には、先ほど定義した frange 関数と同様に値を生成するジェネレータがあります。

FRange クラスのインスタンス f を作成し、複数回反復処理すると、各反復処理で __iter__ メソッドが呼び出され、新しいジェネレータが作成されます。そのため、同じ値のシーケンスを複数回取得することができます。

出力結果：

First iteration:
0 0.25 0.5 0.75 1.0 1.25 1.5 1.75

Second iteration:
0 0.25 0.5 0.75 1.0 1.25 1.5 1.75

今回は、__iter__() メソッドが呼び出されるたびに新しいジェネレータが作成されるため、複数回反復処理することができます。

カスタムクラスに反復処理機能を追加する

ここでは、ジェネレータの基本を理解したという前提で、カスタムクラスに反復処理機能を追加する方法を学びます。Python でクラスを反復可能（iterable）にするには、__iter__() 特殊メソッドを実装する必要があります。反復可能なクラスを使うと、リストやタプルのように、その要素をループで処理することができます。これは非常に強力な機能で、カスタムクラスをより柔軟で使いやすくします。

__iter__() メソッドの理解

__iter__() メソッドは、クラスを反復可能にするための重要な部分です。このメソッドはイテレータオブジェクトを返す必要があります。イテレータは、反復処理（ループ）できるオブジェクトです。これを実現する簡単で効果的な方法は、__iter__() をジェネレータ関数として定義することです。ジェネレータ関数は yield キーワードを使って、値を 1 つずつ生成します。yield 文に遭遇するたびに、関数は一時停止し、値を返します。次にイテレータが呼び出されると、関数は中断したところから再開します。

Structure クラスの修正

この実験のセットアップでは、基本となる Structure クラスを用意しています。Stock などの他のクラスは、この Structure クラスを継承することができます。継承は、既存のクラスのプロパティとメソッドを引き継ぐ新しいクラスを作成する方法です。Structure クラスに __iter__() メソッドを追加することで、そのすべてのサブクラスを反復可能にすることができます。つまり、Structure を継承する任意のクラスは、自動的にループ処理できるようになります。

1. WebIDE で structure.py ファイルを開きます。

cd ~/project

このコマンドは、カレントワーキングディレクトリを project ディレクトリに変更します。structure.py ファイルはこのディレクトリにあります。ファイルにアクセスして変更するには、正しいディレクトリにいる必要があります。

2. Structure クラスの現在の実装を見てみましょう。

class Structure(metaclass=StructureMeta):
    _fields = []
    def __init__(self, *args):
        if len(args) != len(self._fields):
            raise TypeError(f'Expected {len(self._fields)} arguments')
        for name, val in zip(self._fields, args):
            setattr(self, '_'+name, val)

Structure クラスには、属性名を格納する _fields リストがあります。__init__() メソッドはクラスのコンストラクタです。渡された引数の数がフィールドの数と等しいかをチェックして、オブジェクトの属性を初期化します。もし等しくなければ、TypeError を発生させます。そうでなければ、setattr() 関数を使って属性を設定します。

3. 属性値を順番に生成する __iter__() メソッドを追加します。

def __iter__(self):
    for name in self._fields:
        yield getattr(self, name)

この __iter__() メソッドはジェネレータ関数です。_fields リストをループし、getattr() 関数を使って各属性の値を取得します。そして、yield キーワードが値を 1 つずつ返します。

完成した structure.py ファイルは次のようになります。

class StructureMeta(type):
    def __new__(cls, name, bases, clsdict):
        fields = clsdict.get('_fields', [])
        for name in fields:
            clsdict[name] = property(lambda self, name=name: getattr(self, '_'+name))
        return super().__new__(cls, name, bases, clsdict)

class Structure(metaclass=StructureMeta):
    _fields = []
    def __init__(self, *args):
        if len(args) != len(self._fields):
            raise TypeError(f'Expected {len(self._fields)} arguments')
        for name, val in zip(self._fields, args):
            setattr(self, '_'+name, val)

    def __iter__(self):
        for name in self._fields:
            yield getattr(self, name)

この更新された Structure クラスには __iter__() メソッドがあり、これによりクラス自体とそのサブクラスが反復可能になります。

4. ファイルを保存します。
   structure.py ファイルに変更を加えた後は、変更を適用するために保存する必要があります。

5. では、Stock インスタンスを作成して反復処理をテストしてみましょう。

python3 -c "from stock import Stock; s = Stock('GOOG', 100, 490.1); print('Iterating over Stock:'); [print(val) for val in s]"

このコマンドは、Structure クラスを継承する Stock クラスのインスタンスを作成します。そして、リスト内包表記を使ってインスタンスを反復処理し、各値を出力します。

以下のような出力が表示されるはずです。

Iterating over Stock:
GOOG
100
490.1

これで、Structure を継承する任意のクラスは自動的に反復可能になり、反復処理では _fields リストで定義された順序で属性値が生成されます。つまり、Structure のサブクラスの属性をループで処理する際に、反復処理用の追加コードを書く必要がなくなります。

反復処理機能を持つクラスの強化

ここでは、Structure クラスとそのサブクラスが反復処理をサポートするようになったことを前提に進めます。反復処理は Python の強力な概念で、コレクション内のアイテムを 1 つずつループで処理できます。クラスが反復処理をサポートすると、より柔軟になり、多くの Python の組み込み機能と連携できるようになります。この反復処理のサポートが Python でどのように強力な機能を可能にするかを見ていきましょう。

反復処理を利用したシーケンス変換

Python には list() や tuple() などの組み込み関数があります。これらの関数は、任意の反復可能オブジェクトを入力として受け取ることができるため、非常に便利です。反復可能オブジェクトとは、リスト、タプル、または今回の Structure クラスのインスタンスのように、ループで処理できるものです。Structure クラスが反復処理をサポートするようになったので、そのインスタンスを簡単にリストやタプルに変換できます。

1. Stock インスタンスでこれらの操作を試してみましょう。Stock クラスは Structure のサブクラスです。ターミナルで以下のコマンドを実行します。

python3 -c "from stock import Stock; s = Stock('GOOG', 100, 490.1); print('As list:', list(s)); print('As tuple:', tuple(s))"

このコマンドはまず Stock クラスをインポートし、インスタンスを作成し、それを list() と tuple() 関数を使ってそれぞれリストとタプルに変換します。出力結果には、インスタンスがリストとタプルとして表されたものが表示されます。

As list: ['GOOG', 100, 490.1]
As tuple: ('GOOG', 100, 490.1)

アンパッキング

Python にはアンパッキングという非常に便利な機能があります。アンパッキングを使うと、反復可能オブジェクトの要素を一度に個別の変数に割り当てることができます。Stock インスタンスは反復可能なので、このアンパッキング機能を使うことができます。

python3 -c "from stock import Stock; s = Stock('GOOG', 100, 490.1); name, shares, price = s; print(f'Name: {name}, Shares: {shares}, Price: {price}')"

このコードでは、Stock インスタンスを作成し、その要素を name、shares、price の 3 つの変数にアンパッキングします。そして、これらの変数を出力します。出力結果には、これらの変数の値が表示されます。

Name: GOOG, Shares: 100, Price: 490.1

比較機能の追加

クラスが反復処理をサポートすると、比較操作を実装するのが容易になります。比較操作は、2 つのオブジェクトが等しいかどうかをチェックするために使用されます。Structure クラスに __eq__() メソッドを追加して、インスタンスを比較できるようにしましょう。

1. 再度 structure.py ファイルを開きます。__eq__() メソッドは、2 つのオブジェクトを == 演算子で比較するときに呼び出される Python の特殊メソッドです。structure.py ファイルの Structure クラスに以下のコードを追加します。

def __eq__(self, other):
    return isinstance(other, type(self)) and tuple(self) == tuple(other)

このメソッドはまず、isinstance() 関数を使って other オブジェクトが self と同じクラスのインスタンスであるかをチェックします。そして、self と other を両方ともタプルに変換し、これらのタプルが等しいかどうかをチェックします。

完成した structure.py ファイルは次のようになります。

class StructureMeta(type):
    def __new__(cls, name, bases, clsdict):
        fields = clsdict.get('_fields', [])
        for name in fields:
            clsdict[name] = property(lambda self, name=name: getattr(self, '_'+name))
        return super().__new__(cls, name, bases, clsdict)

class Structure(metaclass=StructureMeta):
    _fields = []
    def __init__(self, *args):
        if len(args) != len(self._fields):
            raise TypeError(f'Expected {len(self._fields)} arguments')
        for name, val in zip(self._fields, args):
            setattr(self, '_'+name, val)

    def __iter__(self):
        for name in self._fields:
            yield getattr(self, name)

    def __eq__(self, other):
        return isinstance(other, type(self)) and tuple(self) == tuple(other)

2. __eq__() メソッドを追加した後、structure.py ファイルを保存します。

3. 比較機能をテストしてみましょう。ターミナルで以下のコマンドを実行します。

python3 -c "from stock import Stock; a = Stock('GOOG', 100, 490.1); b = Stock('GOOG', 100, 490.1); c = Stock('AAPL', 200, 123.4); print(f'a == b: {a == b}'); print(f'a == c: {a == c}')"

このコードは 3 つの Stock インスタンス a、b、c を作成します。そして、== 演算子を使って a と b、a と c を比較します。出力結果には、これらの比較結果が表示されます。

a == b: True
a == c: False

4. すべてが正しく動作していることを確認するために、ユニットテストを実行する必要があります。ユニットテストは、プログラムのさまざまな部分が期待通りに動作しているかをチェックする一連のコードです。ターミナルで以下のコマンドを実行します。

python3 teststock.py

すべてが正しく動作していれば、テストが合格したことを示す出力が表示されます。

..
----------------------------------------------------------------------
Ran 2 tests in 0.001s

OK

__iter__() と __eq__() という 2 つの簡単なメソッドを追加するだけで、Structure クラスに Python らしい機能を追加し、使いやすくすることができました。

ストリーミングデータ用のジェネレータの作成

プログラミングにおいて、ジェネレータは強力なツールです。特に、ストリーミングデータソースの監視などの実世界の問題を扱う際に有用です。このセクションでは、これまで学んだジェネレータの知識を実際のシナリオに適用する方法を学びます。ログファイルを監視し、新しい行が追加されるたびにそれを返すジェネレータを作成します。

データソースのセットアップ

ジェネレータを作成する前に、データソースをセットアップする必要があります。この場合、株式市場データを生成するシミュレーションプログラムを使用します。

まず、WebIDE で新しいターミナルを開きます。ここでシミュレーションを開始するコマンドを実行します。

ターミナルを開いた後、株式シミュレーションプログラムを実行します。以下のコマンドを入力します。

cd ~/project
python3 stocksim.py

最初のコマンド cd ~/project は、カレントディレクトリをホームディレクトリ内の project ディレクトリに変更します。2 番目のコマンド python3 stocksim.py は、株式シミュレーションプログラムを実行します。このプログラムは株式市場データを生成し、それをカレントディレクトリ内の stocklog.csv という名前のファイルに書き込みます。監視コードを作成している間、このプログラムをバックグラウンドで実行しておきます。

シンプルなファイル監視プログラムの作成

データソースがセットアップされたので、stocklog.csv ファイルを監視するプログラムを作成しましょう。このプログラムは、価格が下落した株式を表示します。

1. まず、WebIDE で follow.py という新しいファイルを作成します。これを行うには、ターミナルで以下のコマンドを使用して project ディレクトリに移動します。

cd ~/project

2. 次に、以下のコードを follow.py ファイルに追加します。このコードは stocklog.csv ファイルを開き、ファイルポインタをファイルの末尾に移動し、新しい行を継続的にチェックします。新しい行が見つかり、それが価格下落を表す場合、株式名、価格、価格変動を表示します。

## follow.py
import os
import time

f = open('stocklog.csv')
f.seek(0, os.SEEK_END)   ## Move file pointer 0 bytes from end of file

while True:
    line = f.readline()
    if line == '':
        time.sleep(0.1)   ## Sleep briefly and retry
        continue
    fields = line.split(',')
    name = fields[0].strip('"')
    price = float(fields[1])
    change = float(fields[4])
    if change < 0:
        print('%10s %10.2f %10.2f' % (name, price, change))

3. コードを追加した後、ファイルを保存します。次に、ターミナルで以下のコマンドを使用してプログラムを実行します。

python3 follow.py

価格が下落した株式が表示されるはずです。出力は次のようになるかもしれません。

      AAPL     148.24      -1.76
      GOOG    2498.45      -1.55

プログラムを停止するには、ターミナルで Ctrl+C を押します。

ジェネレータ関数への変換

前のコードは動作しますが、ジェネレータ関数に変換することで、より再利用可能でモジュール性の高いコードにすることができます。ジェネレータ関数は、一時停止と再開が可能で、値を 1 つずつ生成する特殊な関数です。

1. 再度 follow.py ファイルを開き、ジェネレータ関数を使用するように修正します。以下は更新後のコードです。

## follow.py
import os
import time

def follow(filename):
    """
    Generator function that yields new lines in a file as they are added.
    Similar to the 'tail -f' Unix command.
    """
    f = open(filename)
    f.seek(0, os.SEEK_END)   ## Move to the end of the file

    while True:
        line = f.readline()
        if line == '':
            time.sleep(0.1)   ## Sleep briefly and retry
            continue
        yield line

## Example usage - monitor stocks with negative price changes
if __name__ == '__main__':
    for line in follow('stocklog.csv'):
        fields = line.split(',')
        name = fields[0].strip('"')
        price = float(fields[1])
        change = float(fields[4])
        if change < 0:
            print('%10s %10.2f %10.2f' % (name, price, change))

follow 関数は現在、ジェネレータ関数になっています。この関数はファイルを開き、末尾に移動し、新しい行を継続的にチェックします。新しい行が見つかると、その行を生成します。

2. ファイルを保存し、以下のコマンドを使用して再度実行します。

python3 follow.py

出力は前と同じになるはずです。ただし、現在はファイル監視のロジックが follow ジェネレータ関数にきれいにカプセル化されています。これは、ファイルを監視する必要がある他のプログラムでこの関数を再利用できることを意味します。

ジェネレータの強力さの理解

ファイル読み取りコードをジェネレータ関数に変換することで、コードをはるかに柔軟で再利用可能なものにしました。follow() 関数は、株式データだけでなく、ファイルを監視する必要がある任意のプログラムで使用できます。

たとえば、サーバーログ、アプリケーションログ、または時間の経過とともに更新される他の任意のファイルを監視するために使用できます。これは、ジェネレータがストリーミングデータソースをクリーンでモジュール性の高い方法で処理する素晴らしい方法であることを示しています。

まとめ

この実験では、Python でジェネレータを使って反復処理をカスタマイズする方法を学びました。yield 文を使って簡単なジェネレータを作成して値のシーケンスを生成し、__iter__() メソッドを実装することでカスタムクラスに反復処理のサポートを追加し、反復処理を利用してシーケンスの変換、アンパッキング、比較を行い、ストリーミングデータソースを監視する実用的なジェネレータを構築しました。

ジェネレータは Python の強力な機能で、最小限のコードでイテレータを作成できます。特に、大規模なデータセットの処理、ストリーミングデータの操作、データパイプラインの作成、カスタム反復パターンの実装に役立ちます。ジェネレータを使用することで、意図が明確で、メモリ効率の高いクリーンなコードを書くことができます。