Python3 Programming

●ジェネレータ (2)

• 関数 (サブルーチン, sub-routine) は call してから return するまで途中で処理を中断することはできない
• ところが、コルーチン (co-routine) は途中で処理を中断し、そこから実行を再開することができる
• コルーチンに親子関係があるものを「セミコルーチン (semi-coroutine)」という
• コルーチン A からコルーチン B を呼び出した場合、A が親で B が子になる
• Python でコルーチンを使う場合、Python 3.5 で導入された async / await を用いるのが一般的
• それ以前はジェネレータで代用していた (今でもできる)

• ジェネレータのコルーチン的な使い方
• ジェネレータ関数を定義する
• ジェネレータ関数を呼び出すとジェネレータオブジェクト g が返される
• 関数 next(g) またはメソッド g.send(None) でジェネレータ関数 (子コルーチン) を実行する
• next(g) または g.send(None) を呼び出した処理が親コルーチンになる
• 子コルーチンで評価した yield の引数が、親コルーチンで評価した next() または send() の返り値になる
• そして、send() の引数が子コルーチンで評価した yield の返り値になる
• つまり、yield と send() を使って親子間でデータの受け渡しができる
• next() の場合は None が渡される
• 最初に子コルーチンを実行する時、send() の引数を受け取ることができない
• 初回は next() を使う、または send() の引数を None にすること
• このほかに、ジェネレータを終了するメソッド close() や例外を送出するメソッド throw() がある
• 例外処理も使用できる
• 詳細は Python のドキュメント 6.2. アトム、原子的要素 (atom) などを参照
• コルーチンの詳しい説明は拙作のページをお読みください
• Lua 入門: コルーチン (1) (2)
• Scheme 入門: コルーチン (1) (2)

• 簡単な実行例

>>> def foo():
...     a = yield "foo"
...     print("foo: ", a)
...     b = yield "bar"
...     print("foo: ", b)
...     c = yield "baz"
...     print("foo: ", c)
...
>>> g = foo()
>>> print(g.send(None))
foo
>>> print(g.send("foo1"))
foo:  foo1
bar
>>> print(g.send("foo2"))
foo:  foo2
baz
>>> print(g.send("foo3"))
foo:  foo3
Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
StopIteration

>>> for x in foo(): print(x)
...
foo
foo:  None
bar
foo:  None
baz
foo:  None

リスト : 複数のコルーチンを呼び出す

def make_coroutine(code):
    while True:
        print(code, end="")
        yield None

def test1(n):
    a = make_coroutine("h")
    b = make_coroutine("e")
    c = make_coroutine("y")
    d = make_coroutine("!")
    e = make_coroutine(" ")
    for _ in range(n):
        for g in [a, b, c, d, e]: next(g)
    for g in [a, b, c, d, e]: g.close()

>>> test1(5)
hey! hey! hey! hey! hey! >>>
>>> test1(10)
hey! hey! hey! hey! hey! hey! hey! hey! hey! hey! >>>

リスト : 子コルーチンから子コルーチンを呼び出す

def make_coroutine2(code, g):
    while True:
        print(code, end="")
        if g: next(g)
        yield None

def test2(n):
    e = make_coroutine2(" ", None)
    d = make_coroutine2("!", e)
    c = make_coroutine2("y", d)
    b = make_coroutine2("e", c)
    a = make_coroutine2("h", b)
    for _ in range(n): next(a)
    for g in [a, b, c, d, e]: g.close()

>>> test2(5)
hey! hey! hey! hey! hey! >>>
>>> test2(15)
hey! hey! hey! hey! hey! hey! hey! hey! hey! hey! hey! hey! hey! hey! hey!

リスト : エラトステネスの篩

# n から始まる整数列
def integers(n):
    while True:
        yield n
        n += 1

# フィルター
def stream_filter(pred, s):
    while True:
        x = next(s)
        if pred(x): yield x

# n 個の素数を求める
def sieve(n):
    # 変数 x の値をクロージャ内に保持する
    def make_pred(x):
        return lambda y: y % x != 0
    #
    nums = integers(2)
    ps = []
    for _ in range(n):
        x = next(nums)
        ps.append(x)
        nums = stream_filter(make_pred(x), nums)
    return ps

# 再帰にすると lambda だけでも動作する
def sieve1(n, nums, ps):
    if n == 0:
        return ps
    else:
        x = next(nums)
        ps.append(x)
        return sieve1(n - 1, stream_filter(lambda y: y % x != 0, nums), ps)

• sieve(), sieve1() ともに n を大きな値 (たとえば 500) にすると RecursionError

>>> sieve(25)
[2, 3, 5, 7, 11, 13, 17, 19, 23, 29, 31, 37, 41, 43, 47, 53, 59, 61, 67, 71, 73, 79,
 83, 89, 97]
>>> sieve(100)
[2, 3, 5, 7, 11, 13, 17, 19, 23, 29, 31, 37, 41, 43, 47, 53, 59, 61, 67, 71, 73, 79,
 83, 89, 97, 101, 103, 107, 109, 113, 127, 131, 137, 139, 149, 151, 157, 163, 167,
 173, 179, 181, 191, 193, 197, 199, 211, 223, 227, 229, 233, 239, 241, 251, 257, 263,
 269, 271, 277, 281, 283, 293, 307, 311, 313, 317, 331, 337, 347, 349, 353, 359, 367,
 373, 379, 383, 389, 397, 401, 409, 419, 421, 431, 433, 439, 443, 449, 457, 461, 463,
 467, 479, 487, 491, 499, 503, 509, 521, 523, 541]

>>> sieve1(25, integers(2), [])
[2, 3, 5, 7, 11, 13, 17, 19, 23, 29, 31, 37, 41, 43, 47, 53, 59, 61, 67, 71, 73, 79,
 83, 89, 97]
>>> sieve1(100, integers(2), [])
[2, 3, 5, 7, 11, 13, 17, 19, 23, 29, 31, 37, 41, 43, 47, 53, 59, 61, 67, 71, 73, 79,
 83, 89, 97, 101, 103, 107, 109, 113, 127, 131, 137, 139, 149, 151, 157, 163, 167, 
 173, 179, 181, 191, 193, 197, 199, 211, 223, 227, 229, 233, 239, 241, 251, 257, 263,
 269, 271, 277, 281, 283, 293, 307, 311, 313, 317, 331, 337, 347, 349, 353, 359, 367,
 373, 379, 383, 389, 397, 401, 409, 419, 421, 431, 433, 439, 443, 449, 457, 461, 463,
 467, 479, 487, 491, 499, 503, 509, 521, 523, 541]

リスト :  簡単なマルチプロセス

# 中断中のプロセスを格納するキュー
proc_queue = []

# キューに追加
def enqueue(x):
    proc_queue.append(x)

# キューからデータを取り出す
def dequeue():
    x = proc_queue[0]
    del proc_queue[0]
    return x

# キューは空か?
def isempty():
    return len(proc_queue) == 0

# プロセスの生成
def fork(fn):
    g = fn()
    g.send(None)
    proc_queue.append(g)

# メインプロセス
def main_process(*args):
    # プロセスの登録
    for fn in args: fork(fn)
    # 実行
    while not isempty():
        p = dequeue()
        if p.send(False):
            enqueue(p)
        else:
            p.close()

# 子プロセス
def mtest0(name, n):
    while n > 0:
        print(name, n)
        yield True
        n -= 1
    yield False

>>> main_process(lambda : mtest0("foo", 5), lambda : mtest0("bar", 7))
foo 5
bar 7
foo 4
bar 6
foo 3
bar 5
foo 2
bar 4
foo 1
bar 3
bar 2
bar 1
>>> main_process(lambda : mtest0("foo", 5), lambda : mtest0("bar", 7),
 lambda : mtest0("oops", 6))
foo 5
bar 7
oops 6
foo 4
bar 6
oops 5
foo 3
bar 5
oops 4
foo 2
bar 4
oops 3
foo 1
bar 3
oops 2
bar 2
oops 1
bar 1

●asyncio

• 標準モジュール asyncio は Python で「並行処理」を行うためのモジュール
• async / await 構文を使う
• asyncio --- 非同期 I/O, (本家, https://docs.python.org/ja/3/library/asyncio.html)
• 並行処理については、拙作のページをお読みください
• Lua 入門: コルーチン (2)
• Scheme 入門: コルーチン (2)
• Go 言語入門: 並行プログラミング
• Erlang 入門: プロセス (前編)

• 基本的な使い方
• 並行処理を行うにはモジュール asyncio を import する
• async を使ってコルーチン関数を定義する

async def name(args, ...): ...

• コルーチン関数を実行するとコルーチンオブジェクト coro が返される
• 関数 run() でメインとなるコルーチンを実行する

asyncio.run(coro)

• run() はイベントループを生成し、そこにメインコルーチンを登録して実行する
• 正確にはコルーチンを Task オブジェクトに変換し、それをイベントループに登録・実行する
• run() はメインコルーチンの返り値をそのまま返す
• コルーチンの中では await を使って子コルーチンを呼び出すことができる

await coro

• 親コルーチンは停止して、子コルーチンに制御が移る
• 子コルーチンの返り値が await の返り値になる
• 簡単な実行例

リスト : コルーチンの簡単な例

import asyncio
import time

async def hello():
  print('Hello ...')
  await asyncio.sleep(1)
  print('... World!')
  return True

>>> asyncio.run(hello())
Hello ...
... World!
True

• 関数 sleep() は現在のコルーチンを引数で指定した時間 (秒) だけ停止する
• Google Colab の場合は await hello() とする

asyncio.sleep(delay, result = None)

• このとき、他のコルーチンの実行が許可される
• result が指定された場合、result が sleep() の返り値になる
• 複数のコルーチンを実行するとき、await coro を順番に並べるだけでは平行に動作しない

リスト : 複数のコルーチン (1)

async def hello2():
  print('Hello2 ...')
  await asyncio.sleep(2)
  print('... World2')
  return 2

async def hello3():
  print('Hello3 ...')
  await asyncio.sleep(3)
  print('... World3')
  return 3

async def hello4():
  print('Hello4 ...')
  await asyncio.sleep(4)
  print('... World4')
  return 4

async def test00():
  s = time.time()
  a = await hello4()
  b = await hello3()
  c = await hello2()
  print(a, b, c)
  print(time.time() - s)

>>> asyncio.run(test00())
Hello4 ...
... World4
Hello3 ...
... World3
Hello2 ...
... World2
4 3 2
9.057184934616089

• Google Colab の場合は await test00() とする
• await hello4() は hello4() を実行し、それが終了するまで待機する
• そのあと、await hello3(), await hello2() が実行される
• つまり、hello4(), hello3(), hello2() は逐次実行されることになる
• 並行に処理したい場合はコルーチンを Task オブジェクトに変換して、イベントループに登録する

async.create_task(coro) => Task object

• 関数 create_task(coro) は引数の coro を Task オブジェクトに変換し、それをイベントループに登録・実行する
• await で Task オブジェクトの終了を待機して、返り値を取得するのは今までと同じ

リスト : 複数のコルーチン (2)

async def test1():
  t1 = asyncio.create_task(hello4())
  t2 = asyncio.create_task(hello3())
  t3 = asyncio.create_task(hello2())
  s = time.time()
  a = await t1
  b = await t2
  c = await t3
  print(a, b, c)
  print(time.time() - s)

>>> asyncio.run(test1())
Hello4 ...
Hello3 ...
Hello2 ...
... World2
... World3
... World4
4 3 2
4.002448081970215

• Google Colab の場合は await test1() とする
• 関数 gather() を使うと、複数のコルーチンを起動して、終了するまで待つことができる

asyncio.gather(coro, ...) => [result, ...]

• gather() はコルーチンの結果を格納したリストを返す
• ただし、Task オブジェクトのメソッドを利用することはできない

リスト : 複数のコルーチン (3)

async def test2():
  s = time.time()
  result = await asyncio.gather(hello4(), hello3(), hello2())
  print(result)
  print(time.time() - s)

>>> asyncio.run(test2())
Hello4 ...
Hello3 ...
Hello2 ...
... World2
... World3
... World4
[4, 3, 2]
4.019838809967041

• Google Colab の場合は await test2() とする

リスト : 複数のコルーチン (4)

async def test_sub(name, n):
  while n > 0:
    print(name, n)
    await asyncio.sleep(0)
    n -= 1
  return False

async def test3():
  result = await asyncio.gather(test_sub("foo", 5), test_sub("bar", 3), test_sub("baz", 7))
  print(result)

>>> asyncio.run(test3())
foo 5
bar 3
baz 7
foo 4
bar 2
baz 6
foo 3
bar 1
baz 5
foo 2
baz 4
foo 1
baz 3
baz 2
baz 1
[False, False, False]

• Google Colab の場合は await test3() とする
• sleep(0) は実行権を他のコルーチンに切り替えるだけ
• なお、Python ver 3.11 からは asyncio.TaskGroup() を使用することが推奨されている
• M.Hiroi は Python 3.10 を使っているので動作は未確認

• キュー
• asynic.Queue は並行処理用の待ち行列
• 基本動作はキューと同じだが、キューにデータを書き込むとき、満杯であれば待ち合わせを行う
• キューからデータを取り出すとき、空であれば待ち合わせを行う
• これによって、コルーチン間での同期処理が可能になる
• キューの大きさが少ない場合でも、データを書き込むコルーチンと取り出すコルーチンが並行に動作することで、キューの大きさ以上のデータを受け渡すことができる

リスト : キューの使用例

async def send_color(color, n, queue):
  while n > 0:
    await queue.put(color)
    await asyncio.sleep(0)
    n -= 1

async def receive_color(n, queue):
  while n > 0:
    item = await queue.get()
    print(item, end=" ")
    n -= 1

async def test4():
  q = asyncio.Queue(4)
  await asyncio.gather(send_color("red", 9, q),
                       send_color("blue", 7, q),
                       send_color("yellow", 8, q),
                       receive_color(24, q))

>>> asyncio.run(test4())
red blue yellow red blue yellow red blue yellow red blue yellow red blue yellow 
red blue yellow red blue yellow red yellow red 

• Google Colab の場合は await test4() とする

• 簡単な例題「哲学者の食事」
• 詳しい説明は拙作のページをお読みください
• Lua 入門: コルーチン (2)
• Scheme 入門: コルーチン (2)
• Go 言語入門: 並行プログラミング (2)
• Erlang 入門: プロセス (後編)

リスト : 哲学者の食事

# フォークの初期化
def init_forks():
  global forks
  forks = [True for _ in range(5)]

# フォークの番号を求める
def fork_index(person, side):
  return person if side == 'right' else (person + 1) % 5

# フォークがあるか
def isFork(person, side):
  return forks[fork_index(person, side)]

# フォークを取る
async def get_fork(person, side):
  while True:
    if isFork(person, side):
      forks[fork_index(person, side)] = False
      break
    await asyncio.sleep(1)
  await asyncio.sleep(1)
  return True

# フォークを置く
async def put_fork(person, side):
  forks[fork_index(person, side)] = True
  await asyncio.sleep(1)
  return True

# 哲学者の動作 (デッドロック)
async def person0(n):
  for _ in range(2):
    print(f'Philosopher {n} is thinking')
    await get_fork(n, 'right')
    await get_fork(n, 'left')
    print(f'Philosopher {n} is eating')
    await asyncio.sleep(1)
    await put_fork(n, 'right')
    await put_fork(n, 'left')
  print(f'Philosopher {n} is sleeping')

async def test50():
  init_forks()
  ps = [person0(n) for n in range(5)]
  await asyncio.gather(*ps)

>>> asyncio.run(test50())
Philosopher 0 is thinking
Philosopher 1 is thinking
Philosopher 2 is thinking
Philosopher 3 is thinking
Philosopher 4 is thinking
^C

• Google Colab の場合は await test50() とする
• Google Colab でプログラムの実行を中止する場合は ■ボタンを押す

リスト : デッドロックの解消

async def person2(n):
  for _ in range(2):
    print(f'Philosopher {n} is thinking')
    if n % 2 == 0:
      await get_fork(n, 'right')
      await get_fork(n, 'left')
    else:
      await get_fork(n, 'left')
      await get_fork(n, 'right')
    print(f'Philosopher {n} is eating')
    await asyncio.sleep(1)
    await put_fork(n, 'right')
    await put_fork(n, 'left')
  print(f'Philosopher {n} is sleeping')

async def test52():
  init_forks()
  ps = [person2(n) for n in range(5)]
  await asyncio.gather(*ps)

>>> asyncio.run(test52())
Philosopher 0 is thinking
Philosopher 1 is thinking
Philosopher 2 is thinking
Philosopher 3 is thinking
Philosopher 4 is thinking
Philosopher 0 is eating
Philosopher 3 is eating
Philosopher 0 is thinking
Philosopher 1 is eating
Philosopher 3 is thinking
Philosopher 0 is eating
Philosopher 1 is thinking
Philosopher 2 is eating
Philosopher 4 is eating
Philosopher 0 is sleeping
Philosopher 2 is thinking
Philosopher 1 is eating
Philosopher 4 is thinking
Philosopher 3 is eating
Philosopher 1 is sleeping
Philosopher 2 is eating
Philosopher 3 is sleeping
Philosopher 4 is eating
Philosopher 2 is sleeping
Philosopher 4 is sleeping

• Google Colab の場合は await test52() とする

●非同期ジェネレータ

• async def で定義したコルーチン関数の中で yield を使うことができる
• これを「非同期ジェネレータ (Asynchronous Generators)」という
• yield の使い方はジェネレータと同じ
• コルーチン関数を実行すると非同期ジェネレータオブジェクト (agen) が返される
• 値を取得する場合、for ではなく async for を、メソッド send() ではなく asend() を使う

async for item in agen:
    ...

await agen.send(value) => item

• next() に相当する関数はないが、async for で使用されるメソッド __anext__() を呼び出すことは可能
• asend() の基本的な使い方は send() と同じ
• ただし、async for や asend() はコルーチン関数の中でしか呼び出すことはできない

リスト : 非同期ジェネレータの簡単な使用例

import asyncio

# 非同期ジェネレータ
async def async_gen():
    a = yield "foo"
    print(a)
    await asyncio.sleep(0.5)
    b = yield "bar"
    print(b)
    await asyncio.sleep(0.5)
    c = yield "baz"
    print(c)

async def test0():
    xs = []
    async for x in async_gen():
        xs.append(x)
    return xs

async def test1():
    return [x async for x in async_gen()]

async def test2():
    g = async_gen()
    try:
        a = await g.asend(None)
        print(a)
        b = await g.asend(1)
        print(b)
        c = await g.asend(2)
        print(c)
        d = await g.asend(3)
        print(d)
    except StopAsyncIteration:
        pass
    return [a, b, c]

async def test3():
    g = async_gen()
    try:
        a = await g.__anext__()
        print(a)
        b = await g.__anext__()
        print(b)
        c = await g.__anext__()
        print(c)
        d = await g.__anext__()
        print(d)
    except StopAsyncIteration:
        pass
    return [a, b, c]

>>> asyncio.run(test0())
None
None
None
['foo', 'bar', 'baz']

>>> asyncio.run(test1())
None
None
None
['foo', 'bar', 'baz']

>>> asyncio.run(test2())
foo
1
bar
2
baz
3
['foo', 'bar', 'baz']

>>> asyncio.run(test3())
foo
None
bar
None
baz
None
['foo', 'bar', 'baz']

• Google Colab の場合は await testX() とする

●仮想環境とパッケージ

• pip は Python のパッケージ管理ツール
• pip を使うとパッケージ (外部ライブラリ) のインストールや管理が簡単にできる
• 最近は OS の Python 環境を保護するため、pip でパッケージをインストールすることはできないようだ
• Ubuntu 24.04 では、この制限 (PEP 668) が厳格に適用される
• pip でインストールしたライブラリが Ubuntu のシステム管理ツール (apt) と競合して OS が不安定になるのを防ぐための仕様
• コマンド "pip install" を実行すると以下のエラーが表示される

error: externally-managed-environment

• OS の Python 環境にパッケージをインストールするには OS のシステム管理ツール (apt) を使う
• このため、Python の仮想環境を作成し、そこにパッケージをインストールすることが推奨されている
• 仮想環境を作成する標準ツールに venv がある
• 通常、pip と venv は Python の配布パッケージに含まれているが、インストールされていない場合もある
• Ubuntu 24.04 の Python には含まれていない
• その場合は以下のコマンドでインストールする

sudo apt install python3-venv python3-pip

• Python の仮想環境を使うと、プロジェクトごとに使用するパッケージやそのバージョンを指定することができる

• 仮想環境の作成
• プロジェクト用ディレクトリの中で以下のコマンドを実行する

python3 -m venv venv-dir

• プロジェクト用ディレクトリの中にディレクトリ venv-dir が生成され、そこに仮想環境に必要なファイルなどがセットされる
• venv-dir の名前は .venv とすることが多いようだ
• 仮想環境の起動
• 以下のコマンドを実行する

$ source .venv/bin/activate 

• プロンプトの先頭に (.venv) が表示される
• これ以降、コマンド python で python3 が起動し、pip で pip3 が起動する
• パッケージのインストールと管理
• 仮想環境を起動した状態で pip を使う
• インストール: pip install パッケージ名
• 一覧表示: pip list
• 設定ファイルの書き出し: pip freeze > requirements.txt
• 他人が同じ環境を再現する際に、このファイルを使う
• 設定ファイルから一括インストール: pip install -r requirements.txt
• 仮想環境を終了するには以下のコマンドを入力する

$ deactivate 

• 元のシステム環境に戻る
• 仮想環境を削除っするには、仮想環境のディレクトリ (たとえば .venv）を削除するだけで OK