pythonで処理待ちのぐるぐる(Spinner)を作る。

はじめに

pythonで関数の実行中にあの棒がぐるぐるするやつを表示したい。
調べてみるといろいろありそうだったが、関数の戻り値を取得しているケースがなかったので方法として残しておく。
また、時間のかかる関数の実装とぐるぐるの実装を分離できるようデコレーター化している。

ぐるぐるを表示しながら実行する関数を"目的の関数"と呼ぶことにする。

環境

PC：windows10
言語 ：python 3.9.1 64bit

実装

「追記:アノテーションに対応させる」 の項で改善したものを書いたので、コピペする場合はそちらを使ってください。

目的の関数を引数に取るSpinnerクラスをつくり、インスタンスに目的の関数に渡す引数を与え実行することでぐるぐるしながら実行する。

import itertools
import time
import sys
from concurrent.futures import ThreadPoolExecutor
from typing import Any

# 関数を引数に取るクラス
class Spinner:
  def __init__(self,func) -> None:
    self.done = False # funcが実行終了するとTrueになる
    self.func = func  # 目的の関数

  # ぐるぐるをコンソールに表示するメソッド
  def spinner(self):
    # 無限に繰り返すイテレータ
    chars = itertools.cycle(r'/-\|')

    while not self.done:
      # ぐるぐるを表示
      sys.stdout.write( '\b' + next(chars))
      sys.stdout.flush()
      # 待機
      time.sleep(0.2)

    # ぐるぐるを削除
    sys.stdout.write('\b') 
    sys.stdout.flush()

  # インスタンスを実行できるようにする特殊メソッド
  def __call__(self, *args: Any, **kwds: Any) -> Any:
    return self.run(*args, **kwds)

  # ぐるぐると目的の関数を並列処理するメソッド
  def run(self, *args: Any, **kwds: Any) -> Any:
    # ThreadPoolExecutor(2): 最大二つの処理を並行処理してくれるスレッドプール
    with ThreadPoolExecutor(2) as executor:
      # ぐるぐるメソッドをスレッドプールに登録
      executor.submit(self.spinner)
      # 目的の関数をスレッドプールに登録
      # 戻り値が欲しいのでFutureインスタンスを変数に格納
      future = executor.submit(self.func, *args, **kwds)

      # 実行結果を得る
      result = future.result()
      self.done = True
    return result

使い方

1.デコレータとして使う

自分で作った関数に常にぐるぐるを表示したい場合はこっちを使う。
関数定義時にデコレータとして使うことで関数呼び出し時に必ずぐるぐるする。
10秒待つだけの処理を例に挙げる。

import time

# デコレータとしてSpinnerをつける
@Spinner
# 目的の関数
def func(arg):
  time.sleep(10) # 時間のかかる処理
  return arg

result = func('hello') # 処理中にコンソールにぐるぐるがつく

print(result)
# -> 'hello'

2.実行時に使う

自分で定義した関数ではない場合や、常にぐるぐるが表示される必要のない場合はこっちを使う。
関数呼び出し時にSpinnerで関数をラップすることで、スピナー付き関数として実行できる。
10秒待つだけの処理を例に挙げる。

import time

# 目的の関数
def func(arg):
  time.sleep(10) # 時間のかかる処理
  return arg

# Spinnerでラップした関数を実行
result = Spinner(func)('hello') # 処理中にコンソールにぐるぐるがつく

print(result)
# -> 'hello'

解説

Spinner

関数を引数に取るだけの単純なクラス
Spinner.doneフィールドは目的の関数の実行が終わったらTrueになる。

class Spinner:
  def __init__(self,func) -> None:
    self.done = False
    self.func = func

Spinner.spinner

ぐるぐるを表示するメソッド。
self.doneがTrueになったらぐるぐるを消去し実行終了

  # ぐるぐるをコンソールに表示するメソッド
  def spinner(self):
    # itertools.cycleで要素を無限に繰り返すイテレータができる
    # / - \ | / - …　と繰り返す
    chars = itertools.cycle(r'/-\|')

    while not self.done:
      # print だと新しい行になってしまうので sys.stdout.write + sys.stdout.flush を使う
      # \b はバックスペース
      # next(chars)でcharsイテレータの次の文字を取得
      sys.stdout.write( '\b' + next(chars))
      sys.stdout.flush()
      # 待機
      time.sleep(0.2)

    # ぐるぐるを空白文字で書き換える
    sys.stdout.write('\b' + ' ' ) 
    sys.stdout.flush()

Spinner.run

ぐるぐると目的の関数を並行処理する
目的の関数は実行結果が欲しいのでFutureインスタンスをfutureに格納しておく
self.doneをTrueにして実行終了

  # ぐるぐると目的の関数を並列処理するメソッド
  def run(self, *args: Any, **kwds: Any) -> Any:
    # ThreadPoolExecutor(2): 最大二つの処理を並行処理してくれるスレッドプール
    with ThreadPoolExecutor(2) as executor:
      # ぐるぐるメソッドをスレッドプールに登録
      executor.submit(self.spinner)
      # 目的の関数をスレッドプールに登録
      # 戻り値が欲しいのでFutureインスタンスを変数に格納
      future = executor.submit(self.func, *args, **kwds)

      # 実行結果を得る
      result = future.result()
      self.done = True
    return result

Spinner.__call__

この特殊メソッドはインスタンス自体に引数を渡して実行したときの挙動を定義する

result = Spinner(func)('hello')

これを見るとSpinner(func)で生成したSpinnerインスタンスに引数'hello'を渡して実行していることがわかる。この時自動的にSpinner(func).__call__('hello')が呼ばれている。

また、デコレータとして機能させる条件は一つの関数を引数に取り実行可能なオブジェクトを返すことなので、__call__メソッドを定義することでSpinnerクラスがデコレータとして使用可能になる。

追記:アノテーションに対応させる

ラップした状態で引数のアノテーションに対応させたい。
クラスの状態でアノテーションに対応させる方法がわからなかったので関数デコレータとして再定義した。
使い方は上記の使い方とほぼ同じ。Spinnerをspinnerにすれば同じように動く。

import itertools
import time
import sys
from concurrent.futures import ThreadPoolExecutor
from typing import TypeVar

T = TypeVar('T')
def spinner(func:T):
  running = True

  # runningの間ぐるぐるを表示する
  def spin():
    chars = itertools.cycle(r'/-\|')
    while running:
      sys.stdout.write( '\b' + next(chars))
      sys.stdout.flush()
      time.sleep(0.2)
    sys.stdout.write('\b')
    sys.stdout.flush()

  parllrel:T
  # 並列処理
  def parallel(*arg,**kwarg):
    nonlocal running
    with ThreadPoolExecutor(2) as executor:
      executor.submit(spin)
      future = executor.submit(func, *arg, **kwarg)
      result = future.result()
      running = False
    return result

  return parallel

終わりに

結局ぐるぐるするやつの正式名称ってあるのでしょうか...?

初投稿なので読みづらいとは思いますが、お役に立てれば幸いです。
pythonの理解やら、Qiitaの記述方法やらに問題がありましたら是非教えてください！！

参考文献