CUBE SUGAR CONTAINER

技術系のこと書きます。

Python: pandas-should というパッケージを作ってみた

pandas を使ってデータ分析などをしていると、自分が意図した通りのデータになっているか、たまに確認することになると思う。 確認する方法としてはグラフにプロットしてみたり、あるいは assert 文を使って shape などを確認することが考えられる。

今回紹介する pandas-should は後者の「assert 文を使った内容の確認」を、なるべく簡単に分かりやすく記述するために作ってみた。

github.com

使った環境は次の通り。 なお、パッケージ自体は Python 3.6 以降で動作する。

$ sw_vers
ProductName:    Mac OS X
ProductVersion: 10.14.5
BuildVersion:   18F132
$ python -V
Python 3.7.4

インストール

インストールは pip からできる。

$ pip install pandas-should

使い方

パッケージをインストールできたら、とりあえず Python のインタプリタを起動しておく。

$ python

あとは pandas_should をインポートするだけ。

>>> import pandas_should

インポートすると pandas の DataFrame と Series のインスタンスに should というアトリビュートがひそかに生えてくる。

>>> import pandas as pd
>>> df = pd.DataFrame([1, 2, 3], columns=['id'])
>>> df.should
<pandas_should.dataframe.ShouldDataFrameAccessor object at 0x1083d6160>
>>> s = pd.Series([1, 2, 3])
>>> s.should
<pandas_should.series.ShouldSeriesAccessor object at 0x1196a36a0>

この should 経由で色々とできて、例えば行数が一致することを確認したいなら have_length() を使う。

>>> df.should.have_length(3)
True

基本的にメソッドは真偽値を返すので、アサーションに使うならこうする。

>>> assert df.should.have_length(3)

ここからは使いそうな API を幾つか紹介していく。

DataFrame

まずは DataFrame から。

要素に Null (NaN or NaT) が含まれるか調べたい

普通に書くと、こんな感じになると思う。

>>> not df.isnull().any(axis=None)
True

pandas-should を使うと、こう書ける。

>>> df.should.have_not_null()
True

あるいは Null が含まれることを期待するのであれば、こう。

>>> df.should.have_null()
False

要素のレンジを調べたい

各要素が特定のレンジ (値の範囲) に収まっているか知りたいときは、こう書く。 値の範囲には、指定した最小値と最大値も含まれる。

>>> df.should.fall_within_range(1, 3)
True

下限だけ指定したいときは greater_than() を使う。

>>> df.should.greater_than(0)
True

greater_than() では指定した値は含まれないので、含みたいときは greater_than_or_equal() を使う。

>>> df.should.greater_than_or_equal(1)
True

長いのでエイリアスとして gt()gte() も使える。

>>> df.should.gt(1)
False
>>> df.should.gte(1)
True

上限についても同様。 こちらもエイリアスとして lt()lte() が使える。

>>> df.should.less_than(3)
False
>>> df.should.less_than_or_equal(3)
True

形状 (Shape) を調べたい

続いて DataFrame の形状を調べる方法について。

比較対象が必要なので新たに DataFrame を用意しておく。

>>> data1 = [
...     ('apple', 98, True),
...     ('banana', 128, True),
... ]
>>> df1 = pd.DataFrame(data1, columns=['name', 'price', 'fruit'])
>>> data2 = [
...     ('carrot', 198, False),
...     ('dates', 498, True),
... ]
>>> df2 = pd.DataFrame(data2, columns=['name', 'price', 'fruit'])

同じ行数や列数であることを確認したいときは have_same_length()have_same_width() を使う。

>>> df1.should.have_same_length(df2)
True
>>> df1.should.have_same_width(df2)
True

前述した通り、整数で指定したいときは have_width()have_length() が使える。

>>> df1.should.have_width(2)
True
>>> df1.should.have_length(2)
True

ちなみに have_same_*() は複数の DataFrame との比較もできる。

>>> data3 = [
...     ('eggplant', 128, False),
... ]
>>> df3 = pd.DataFrame(data3, columns=['name', 'price', 'fruit'])
>>> data4 = [
...     ('fig', 298, True),
... ]
>>> df4 = pd.DataFrame(data4, columns=['name', 'price', 'fruit'])

例えば二つの DataFrame の行数を加算したものと同じになるか調べたいときは以下のようにする。 具体的なユースケースとしては、結合前と結合後の DataFrame の行数が一致しているか調べるときとか。

>>> df1.should.have_same_length(df3, df4)
True

行数についても同様。

>>> df1.should.have_same_width(df3, df4)
True

行と列を別々に比較するのがめんどいときは be_shaped_like() で一気に比較できる。

>>> df1.should.be_shaped_like(df2)
True

このメソッドにはタプルとか整数も渡せる。

>>> df1.should.be_shaped_like(df2.shape)  # tuple
True
>>> df1.should.be_shaped_like(df2.shape[0], df2.shape[1])  # int, int
True

Series

続いては Series について。

要素に Null (NaN or NaT) が含まれるか調べたい

要素に Null が含まれるか調べたいときは DataFrame と同じやり方が使える。

>>> s.should.have_not_null()
True
>>> s.should.have_null()
False

要素のレンジを調べたい

Series に関しても DataFrame と同じように、要素のレンジ (値の範囲) を調べられる。 追加で説明することは特にないかな。

>>> s.should.fall_within_range(1, 3)
True
>>> s.should.gt(1)
False
>>> s.should.gte(1)
True
>>> s.should.lt(3)
False
>>> s.should.lte(3)
True

形状 (Shape) を調べたい

Series に関しては列数という概念がないけど、次のように行数に関しては DataFrame と同じやり方が使える。

>>> s2 = pd.Series([4, 5, 6])
>>> s.should.have_same_length(s2)
True
>>> s.should.have_length(3)
True

そんなかんじで。 こういう API があると便利で欲しいみたいなのがあれば教えてほしい。

Python: Kivy で Matplotlib のグラフをプロットする

Kivy は最近人気のある Python のクロスプラットフォームな GUI のフレームワーク。 今回はそんな Kivy で作った GUI 上に Matplotlib のグラフをプロットしてみる。

使った環境は次の通り。

$ sw_vers
ProductName:    Mac OS X
ProductVersion: 10.14.5
BuildVersion:   18F132
$ python -V                    
Python 3.7.3

下準備

まずは Kivy と Matplotlib をインストールしておく。

$ pip install kivy matplotlib numpy

続いて Kivy Garden を使って Matplotlib 用のプラグイン (garden.matplotlib) をインストールする。

$ garden install matplotlib

これで Kivy で Matplotlib を使う準備ができた。

Kivy で Matplotlib のグラフをプロットする

以下の Kivy で Matplotlib のグラフをプロットするサンプルコードを示す。 garden.matplotlib を使うと Figure#canvas のインスタンスをウィジェットとして追加できるようになる。

#!/usr/bin/env python
# -*- coding: utf-8 -*-

from kivy.app import App
from kivy.uix.boxlayout import BoxLayout
from kivy.uix.label import Label
import numpy as np
import matplotlib.pyplot as plt
import matplotlib
# Kivy 上で Matplotlib を使うために必要な準備
matplotlib.use('module://kivy.garden.matplotlib.backend_kivy')


class GraphApp(App):
    """Matplotlib のグラフを表示するアプリケーション"""

    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.title = 'Matplotlib graph on Kivy'

    def build(self):
        # メインの画面
        main_screen = BoxLayout()
        main_screen.orientation = 'vertical'

        # 上部にラベルを追加しておく
        label_text = 'The following is a graph of Matplotlib'
        label = Label(text=label_text)
        label.size_hint_y = 0.2
        main_screen.add_widget(label)

        # サイン波のデータを用意する
        x = np.linspace(-np.pi, np.pi, 100)
        y = np.sin(x)
        # 描画する領域を用意する
        fig, ax = plt.subplots()
        # プロットする
        ax.plot(x, y)
        # Figure#canvas をウィジェットとして追加する
        main_screen.add_widget(fig.canvas)

        return main_screen


def main():
    # アプリケーションを開始する
    app = GraphApp()
    app.run()


if __name__ == '__main__':
    main()

上記を実行する。

$ python kvplot.py

すると、次のような結果が得られる。

f:id:momijiame:20190710212812p:plain

ちゃんと描画できてるね。

FigureCanvasKivyAgg を使う場合

別のやり方として Figure オブジェクトを FigureCanvasKivyAgg でラップするやり方もある。

サンプルコードは次の通り。

#!/usr/bin/env python
# -*- coding: utf-8 -*-

from kivy.app import App
from kivy.uix.boxlayout import BoxLayout
from kivy.garden.matplotlib.backend_kivyagg import FigureCanvasKivyAgg
from kivy.uix.label import Label
import numpy as np
import matplotlib.pyplot as plt


class GraphApp(App):
    """Matplotlib のグラフを表示するアプリケーション"""

    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.title = 'Matplotlib graph on Kivy'

    def build(self):
        main_screen = BoxLayout()
        main_screen.orientation = 'vertical'

        label_text = 'The following is a graph of Matplotlib'
        label = Label(text=label_text)
        label.size_hint_y = 0.2
        main_screen.add_widget(label)

        x = np.linspace(-np.pi, np.pi, 100)
        y = np.sin(x)
        fig, ax = plt.subplots()
        ax.plot(x, y)

        # Figure をラップする
        widget = FigureCanvasKivyAgg(fig)
        # ウィジェットとして追加する
        main_screen.add_widget(widget)

        return main_screen


def main():
    app = GraphApp()
    app.run()


if __name__ == '__main__':
    main()

表示される内容は変わらない。

ウィジェットのクラスとして定義する

さらに、ウィジェットをクラスとして定義した上で、その中にグラフを埋め込む場合には、次のようにする。

#!/usr/bin/env python
# -*- coding: utf-8 -*-

from kivy.app import App
from kivy.uix.boxlayout import BoxLayout
from kivy.uix.label import Label
from kivy.garden.matplotlib.backend_kivyagg import FigureCanvasKivyAgg
import numpy as np
import matplotlib.pyplot as plt


class GraphView(BoxLayout):
    """Matplotlib のグラフを表示するためのウィジェット"""

    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)

        x = np.linspace(-np.pi, np.pi, 100)
        y = np.sin(x)
        fig, ax = plt.subplots()
        ax.plot(x, y)

        widget = FigureCanvasKivyAgg(fig)
        self.add_widget(widget)


class GraphApp(App):
    """Matplotlib のグラフを表示するアプリケーション"""

    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.title = 'Matplotlib graph on Kivy'

    def build(self):
        main_screen = BoxLayout()
        main_screen.orientation = 'vertical'

        label_text = 'The following is a graph of Matplotlib'
        label = Label(text=label_text)
        label.size_hint_y = 0.2
        main_screen.add_widget(label)

        # ウィジェットを生成して追加する
        graph = GraphView()
        main_screen.add_widget(graph)

        return main_screen


def main():
    # アプリケーションを開始する
    app = GraphApp()
    app.run()


if __name__ == '__main__':
    main()

こちらも表示される内容は変わらない。

KV 言語を使う

Kivy は KV 言語という DSL でレイアウトを制御できる。 もし、KV 言語も併用したいという場合であれば次のようにする。

#!/usr/bin/env python
# -*- coding: utf-8 -*-

from kivy.app import App
from kivy.uix.boxlayout import BoxLayout
from kivy.uix.label import Label
from kivy.lang import Builder
from kivy.garden.matplotlib.backend_kivyagg import FigureCanvasKivyAgg
import numpy as np
import matplotlib.pyplot as plt


kv_def = '''
<RootWidget>:
    orientation: 'vertical'

    Label:
        text: 'The following is a graph of Matplotlib'
        size_hint_y: 0.2

    GraphView:

<GraphView>:
'''
Builder.load_string(kv_def)


class GraphView(BoxLayout):
    """Matplotlib のグラフを表示するためのウィジェット"""

    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)

        x = np.linspace(-np.pi, np.pi, 100)
        y = np.sin(x)
        fig, ax = plt.subplots()
        ax.plot(x, y)

        widget = FigureCanvasKivyAgg(fig)
        self.add_widget(widget)


class RootWidget(BoxLayout):
    """子を追加していくためのウィジェットを用意しておく"""


class GraphApp(App):
    """Matplotlib のグラフを表示するアプリケーション"""

    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.title = 'Matplotlib graph on Kivy'

    def build(self):
        return RootWidget()


def main():
    # アプリケーションを開始する
    app = GraphApp()
    app.run()


if __name__ == '__main__':
    main()

こちらも表示される内容は変わらない。

グラフで表示される内容を動的に更新したい

グラフに表示される内容を動的に更新したい場合のサンプルも以下に示す。 基本的には普通に Matplotlib を使って動的なグラフを描くのと変わらない。

#!/usr/bin/env python
# -*- coding: utf-8 -*-

from kivy.app import App
from kivy.uix.boxlayout import BoxLayout
from kivy.clock import Clock
from kivy.lang import Builder
from kivy.garden.matplotlib.backend_kivyagg import FigureCanvasKivyAgg
import numpy as np
import matplotlib.pyplot as plt


kv_def = '''
<RootWidget>:
    orientation: 'vertical'

    Label:
        text: 'The following is a graph of Matplotlib'
        size_hint_y: 0.2

    GraphView:

<GraphView>:
'''
Builder.load_string(kv_def)


class GraphView(BoxLayout):
    """Matplotlib のグラフを表示するためのウィジェット"""

    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)

        # 初期化に用いるデータ
        x = np.linspace(-np.pi, np.pi, 100)
        y = np.sin(x)
        # 描画状態を保存するためのカウンタ
        self.counter = 0

        # Figure, Axis を保存しておく
        self.fig, self.ax = plt.subplots()
        # 最初に描画したときの Line も保存しておく
        self.line, = self.ax.plot(x, y)

        # ウィジェットとしてグラフを追加する
        widget = FigureCanvasKivyAgg(self.fig)
        self.add_widget(widget)

        # 1 秒ごとに表示を更新するタイマーを仕掛ける
        Clock.schedule_interval(self.update_view, 0.01)

    def update_view(self, *args, **kwargs):
        # データを更新する
        self.counter += np.pi / 100  # 10 分の pi ずらす
        # ずらした値を使ってデータを作る
        x = np.linspace(-np.pi + self.counter,
                        np.pi + self.counter,
                        100)
        y = np.sin(x)
        # Line にデータを設定する
        self.line.set_data(x, y)
        # グラフの見栄えを調整する
        self.ax.relim()
        self.ax.autoscale_view()
        # 再描画する
        self.fig.canvas.draw()
        self.fig.canvas.flush_events()


class RootWidget(BoxLayout):
    """子を追加していくためのウィジェットを用意しておく"""


class GraphApp(App):
    """Matplotlib のグラフを表示するアプリケーション"""

    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.title = 'Matplotlib graph on Kivy'

    def build(self):
        return RootWidget()


def main():
    # アプリケーションを開始する
    app = GraphApp()
    app.run()


if __name__ == '__main__':
    main()

実行すると、こんな表示が得られる。 うにょうにょ。

f:id:momijiame:20190714123106g:plain

いじょう。

gRPC の通信を Wireshark でキャプチャしてみる

今回は、最近よく使われている gRPC の通信を Wireshark でキャプチャしてみる。 ちなみに、現行の Wireshark だと gRPC をちゃんと解釈できるみたい。

使った環境は次の通り。

$ sw_vers
ProductName:    Mac OS X
ProductVersion: 10.14.5
BuildVersion:   18F132
$ python -V                    
Python 3.7.3
$ wireshark --version | head -n 1
Wireshark 3.0.2 (v3.0.2-0-g621ed351d5c9)

Python で gRPC サーバ・クライアントを書く

通信をキャプチャするためには、まず gRPC のサーバとクライアントを用意する必要がある。

まずは gRPC を使う上で必要なツールキットをインストールしておく。

$ pip install grpcio-tools

gRPC では様々なプログラミング言語を用いてサーバとクライアントを記述できる。 そのために、どの言語を使う場合にも共通のスキーマを定義した上で、それを各言語用にコンパイルする。 共通のスキーマを定義するには Protocol Buffers というフォーマットを用いる。

以下は Protocol Buffers のスキーマを定義するファイルとなっている。 この中では HelloWorld というサービス上で greet() という RPC が定義されている。 greet() でやり取りするのは HelloRequestHelloReply というメッセージ。

syntax = "proto3";

service HelloWorld {
  rpc greet (HelloRequest) returns (HelloReply) {}
}

message HelloRequest {
  string name = 1;
}

message HelloReply {
  string message = 1;
}

上記を Python 用にコンパイルする。

$ python -m grpc_tools.protoc \
  --proto_path=. \
  --grpc_python_out=. \
  --python_out=. \
  helloworld.proto

すると、次のように pb2 という名前を含むファイルが二つ生成される。

$ ls | grep pb2
helloworld_pb2.py
helloworld_pb2_grpc.py

これらは、先ほどのスキーマ定義から生成された Python のモジュールになっている。

$ file helloworld_pb2.py 
helloworld_pb2.py: Python script text executable, ASCII text
$ file helloworld_pb2_grpc.py 
helloworld_pb2_grpc.py: Python script text executable, ASCII text

生成されたモジュールを使って gRPC のサーバを書いてみよう。 Protocol Buffers の定義にはインターフェースしか記述されていない。 そのため、内部で何をやるか実装を書いてやる必要がある。

#!/usr/bin/env python
# -*- coding: utf-8 -*-

from concurrent import futures
import time

import grpc

import helloworld_pb2
import helloworld_pb2_grpc


class HelloWorldService(helloworld_pb2_grpc.HelloWorldServicer):
    """サービスを定義する"""

    def greet(self, request, context):
        """RPC の中身"""
        # 受け取った内容を使って返信するメッセージを組み立てる
        message = 'Hello, {name}'.format(name=request.name)
        reply = helloworld_pb2.HelloReply(message=message)
        return reply


def main():
    # gRPC のサーバを用意する
    executor = futures.ThreadPoolExecutor(max_workers=10)
    server = grpc.server(executor)
    service = HelloWorldService()
    helloworld_pb2_grpc.add_HelloWorldServicer_to_server(service, server)

    # サーバを 37564 ポートで動作させる
    server.add_insecure_port('localhost:37564')
    server.start()

    try:
        while True:
            time.sleep(1)
    finally:
        server.stop(0)


if __name__ == '__main__':
    main()

サーバを起動してみよう。

$ python server.py

別のターミナルを開いてポートの状態を確認してみよう。 TCP で localhost:37564 を Listen していれば上手くいっている。

$ lsof -i | grep 37564
python3.7 11805 amedama    5u  IPv6 0xbd776a50dc8d08b1      0t0  TCP localhost:37564 (LISTEN)
python3.7 11805 amedama    6u  IPv6 0xbd776a50dc8d4231      0t0  TCP localhost:37564 (LISTEN)

続いてクライアントを記述しよう。 サーバが稼働している localhost:37564 ポートに接続させる。

#!/usr/bin/env python
# -*- coding: utf-8 -*-

import grpc

import helloworld_pb2
import helloworld_pb2_grpc


def main():
    # 'localhost:37564' に接続する
    with grpc.insecure_channel('localhost:37564') as channel:
        # メッセージと共にリモートプロシジャを呼び出す
        stub = helloworld_pb2_grpc.HelloWorldStub(channel)
        reply = stub.greet(helloworld_pb2.HelloRequest(name='World'))
        # 返ってきた内容を表示する
        print('Reply:', reply.message)


if __name__ == '__main__':
    main()

上記を実行してみよう。 次のようにメッセージが表示されれば上手くいった。

$ python client.py 
Reply: Hello, World

通信を Wireshark でキャプチャする

さて、これだけだとあまり面白くないので、続いては上記の通信をキャプチャしてみよう。

パケットキャプチャをするために Wireshark をインストールする。

$ brew cask install wireshark

インストールできたら Wireshark を起動する。

$ wireshark

起動したら Loopback インターフェースのキャプチャを開始する。 また、ディスプレイフィルタのバーに tcp.port == 37564 を指定する。 これで余計な内容が表示されなくて済む。

f:id:momijiame:20190630152902p:plain

準備ができたら、先ほどの gRPC サーバを起動する。

$ python server.py

そして、gRPC クライアントを実行する。

$ python client.py

すると、次のように TCP の通信内容がキャプチャされる。

f:id:momijiame:20190630153107p:plain

適当にフレームを選択して「Follow > TCP Stream」すると一連の TCP の通信内容が確認できる。

f:id:momijiame:20190630153245p:plain

HTTP/2.0 という文字列から読み取れる通り、gRPC は通信部分のレイヤーに HTTP2 を採用している。

生の TCP だと読みにくいので、通信を HTTP/2 として解釈させてみよう。 先ほどと同じようにフレームを選択したら「Decode As...」を選択する。

f:id:momijiame:20190630153445p:plain

上記のような画面が開いたら右端のカラムを「HTTP2」にする。 これで、通信プロトコルが HTTP/2 として解釈される。

HTTP/2 において HTTP のリクエストとレスポンスは HeadersData のメッセージでやり取りされる。 以下のように、まずリクエストが出ている。 メッセージの内容に World が含まれることが分かる。 また、呼び出す対象のリモートプロシジャは HTTP のパス部分に格納されるようだ。

f:id:momijiame:20190630153533p:plain

同様に、上記に対するレスポンスが以下になる。 メッセージに Hello, World という内容が含まれることが分かる。

f:id:momijiame:20190630153821p:plain

なかなか分かりやすいね。

Python: py4j で Java の API を Python から使う

今回は py4j を使って Java の API を Python から利用してみる。

py4j のアーキテクチャはサーバ・クライアントモデルになっている。 つまり、まず Java の API を Python から叩けるように、Java でゲートウェイサーバとなるプログラムを書く。 そして、Python からはネットワーク経由でそのゲートウェイサーバにアクセスする。 これは、RPC (Remote Procedure Call) の考え方に近い。

使った環境は次の通り。

$ sw_vers
ProductName:    Mac OS X
ProductVersion: 10.14.5
BuildVersion:   18F132
$ python -V
Python 3.7.3
$ java -version 
openjdk version "12.0.1" 2019-04-16
OpenJDK Runtime Environment (build 12.0.1+12)
OpenJDK 64-Bit Server VM (build 12.0.1+12, mixed mode, sharing)

下準備

下準備として Java (JDK) と py4j をインストールしておく。

$ brew cask install java
$ pip install py4j

ゲートウェイサーバを記述する

最初に、Python から利用したい Java の API に対してゲートウェイサーバを書く。

以下のサンプルコードでは HelloWorld クラスの API についてゲートウェイサーバを用意している。 基本的にはクラスのインスタンスを GatewayServer クラスに渡してやるだけ。 提供されるのは割り算の機能を持った div() メソッドになる。

import py4j.GatewayServer;

public class HelloWorld {

  public int div(int a, int b) {
    // 割り算の機能を提供するメソッド
    return a / b;
  }

  public static void main(String[] args) {
    // GatewayServer 経由で機能を提供する
    HelloWorld application = new HelloWorld();
    GatewayServer gateway = new GatewayServer(application);
    gateway.start();
    System.out.println("Starting server...");
  }
}

続いて、上記を Java バイトコードにコンパイルする。 ただし、それには py4j の jar ファイルが必要になる。

jar ファイルは py4j のインストール先にある。 なので、まずは py4j のインストールされている場所を確認する。

$ python -c "import py4j; print(py4j.__path__)"
['/Users/amedama/.virtualenvs/py37/lib/python3.7/site-packages/py4j']

次のように share ディレクトリ以下に jar ファイルがあった。

$ ls ~/.virtualenvs/py37/share/py4j
py4j0.10.8.1.jar

この jar ファイルにクラスパスを通しながら、先ほどのプログラムをコンパイルする。

$ javac -cp ~/.virtualenvs/py37/share/py4j/py4j0.10.8.1.jar HelloWorld.java

次のように class ファイルが完成すれば上手くいっている。

$ file HelloWorld.class 
HelloWorld.class: compiled Java class data, version 56.0

コンパイルできたら、ゲートウェイサーバのプログラムを起動する。 この際にも、jar ファイルにクラスパスを通す必要がある。

$ java -cp ~/.virtualenvs/py37/share/py4j/py4j0.10.8.1.jar:. HelloWorld
Starting server...

これで、デフォルトでは TCP:25333 で py4j のサービスが起動する。

$ lsof -i | grep 25333
java      10364 amedama    6u  IPv6 0xbd776a50dc8d3c71      0t0  TCP localhost:25333 (LISTEN)

Python から利用する

これで準備ができたらので、Python から利用してみよう。

まずは Python のインタプリタを起動する。

$ python

py4j 経由で Java の API を呼び出すために JavaGateway クラスのインスタンスを用意する。

>>> from py4j.java_gateway import JavaGateway
>>> java_gateway = JavaGateway()
>>> java_app = java_gateway.entry_point

あとは Java の API を呼び出すだけ。

>>> java_app.div(20, 10)
2

ちゃんと割り算ができている。

試しに Java のプログラム上で例外を発生させてみよう。 すると、次のように py4j.protocol.Py4JJavaError 例外となる。 例外の中には、Java 上で発生した例外の情報が入っている。

>>> java_app.div(1, 0)
Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
  File "/Users/amedama/.virtualenvs/py37/lib/python3.7/site-packages/py4j/java_gateway.py", line 1286, in __call__
    answer, self.gateway_client, self.target_id, self.name)
  File "/Users/amedama/.virtualenvs/py37/lib/python3.7/site-packages/py4j/protocol.py", line 328, in get_return_value
    format(target_id, ".", name), value)
py4j.protocol.Py4JJavaError: An error occurred while calling t.div.
: java.lang.ArithmeticException: / by zero
    at HelloWorld.div(HelloWorld.java:7)
    at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
    at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
    at java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
    at java.base/java.lang.reflect.Method.invoke(Method.java:567)
    at py4j.reflection.MethodInvoker.invoke(MethodInvoker.java:244)
    at py4j.reflection.ReflectionEngine.invoke(ReflectionEngine.java:357)
    at py4j.Gateway.invoke(Gateway.java:282)
    at py4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:132)
    at py4j.commands.CallCommand.execute(CallCommand.java:79)
    at py4j.GatewayConnection.run(GatewayConnection.java:238)
    at java.base/java.lang.Thread.run(Thread.java:835)

いじょう。

Python: ユニットテストを書いてみよう

ソフトウェアエンジニアにとって、不具合に対抗する最も一般的な方法は自動化されたテストを書くこと。 テストでは、書いたプログラムが誤った振る舞いをしないか確認する。 一口に自動テストといっても、扱うレイヤーによって色々なものがある。 今回は、その中でも最もプリミティブなテストであるユニットテストについて扱う。 ユニットテストでは、関数やクラス、メソッドといった単位の振る舞いについてテストを書いていく。

Python には標準ライブラリとして unittest というパッケージが用意されている。 これは、文字通り Python でユニットテストを書くためのパッケージとなっている。 このエントリでは、最初に unittest パッケージを使ってユニットテストを書く方法について紹介する。 その上で、さらに効率的にテストを記述するためにサードパーティ製のライブラリである pytest を使っていく。

使った環境は次の通り。

$ sw_vers 
ProductName:    Mac OS X
ProductVersion: 10.14.5
BuildVersion:   18F132
$ python -V        
Python 3.7.3

はじめてのユニットテスト

まずは最も単純な例を使ってユニットテストの書き方を説明していく。

そもそもテストを書くからには、テストする対象が必要になる。 そこで、次のように greet() という関数を用意した。 この関数は、呼び出されると特定の文字列を返すようになっている。

# -*- coding: utf-8 -*-


def greet():
    """挨拶のメッセージを返す関数"""
    return 'Hello, World!'

上記の内容を helloworld.py という名前で保存する。

これで、上記を helloworld モジュールとしてインポートして使えるようになる。

$ python -c "import helloworld; print(helloworld.greet())"
Hello, World!

続いて、上記のモジュールに対応するテストを記述する。 まず、テストを書くには unittest.TestCase クラスを継承したクラスを定義する。 そのクラスの中にテストをメソッドとして記述していく。 テストのメソッドは、必ず名前の先頭が test から始まるようにする。 これは後述するテストランナーが名前を元にテストコードを探すため。 そして、テストではテスト対象から得られる値もしくは状態が期待する内容と一致するかを比較する。 比較する方法として unittest.TestCase クラスには assertEqual()assertTrue() といったメソッドが用意されている。

以下に unittest パッケージを使ったユニットテストのサンプルコードを示す。

#!/usr/bin/env python
# -*- coding: utf-8 -*-

import unittest

import helloworld  # テスト対象のモジュールをインポートする


class TestHelloWorld(unittest.TestCase):
    """helloworld モジュールのテストを記述するクラス"""

    def test_greet(self):
        """greet() 関数をテストするメソッド"""
        # テスト対象の関数を呼び出す
        message = helloworld.greet()
        # 関数の返り値が期待した内容と一致するか確認する
        self.assertEqual(message, 'Hello, World!')


if __name__ == '__main__':
    # スクリプトとして実行された場合の処理
    unittest.main(verbosity=2)

上記を test_helloworld.py という名前で保存しよう。 実はこの名前が重要で、後述するテストランナーは test から始まる名前を元にテストコードを探索する。

テストを実行する準備が整ったので、手始めに上記をスクリプトとして実行してみよう。 先ほどのテストコードは、スクリプトとして実行された場合にも unittest.main() 関数が呼ばれるようにしてある。

$ python test_helloworld.py
test_greet (__main__.TestHelloWorld) ... ok

----------------------------------------------------------------------
Ran 1 test in 0.000s

OK

上記から、先ほど記述した test_greet() 関数が実行されて、テストが正しくパスしたことが分かる。

テストはスクリプトとして実行する以外にも、特定のディレクトリ以下から自動的に探して実行する方法もある。 それには、次のように Python のインタプリタで -m unittest として unittest モジュールが実行されるようにする。 その上で discover というコマンドを実行するとカレントディレクトリ以下のテストコードを名前を頼りに自動で探して実行できる。

$ python -m unittest discover -v
test_greet (test_helloworld.TestHelloWorld) ... ok

----------------------------------------------------------------------
Ran 1 test in 0.000s

OK

なお、上記は Python の公式では「テストディスカバリ」という名前の機能として提供されている。 一般的には、テストを探して実行する機能は「テストランナー」と呼ばれる。

もう少し実用的な例を見てみる

先ほどはテストする対象が固定の文字列を返すだけだったので、あまりテストをする意味合いが感じられなかったかもしれない。 続いては、もう少しだけ実用的な例を見ていこう。

以下のサンプルコードでは、有名な FizzBuzz を実装している。 fizzbuzz() 関数では、渡された整数が 3 または 5 で割り切れるかを判定して返す値を切り替える。 3 と 5 の両方で割れるときは 'FizzBuzz' を、3 だけで割れるときは 'Fizz' を、5 だけで割れるときは 'Buzz' を返す。 なお、いずれでも割れないときは単に数字を文字列にして返すこととする。

# -*- coding: utf-8 -*-


def fizzbuzz(n):
    if n % 3 == 0 and n % 5 == 0:
        return 'FizzBuzz'

    if n % 3 == 0:
        return 'Fizz'

    if n % 5 == 0:
        return 'Buzz'

    return str(n)

上記を fizzbuzz.py という名前で保存する。

それでは、先ほどの FizzBuzz が正しく振る舞うかテストコードを書いて確かめてみることにしよう。 要領は先ほどと変わらない。 関数に入力される値と返り値に対して、期待される内容を比較していけば良い。

#!/usr/bin/env python
# -*- coding: utf-8 -*-

import unittest

import fizzbuzz


class TestFizzBuzz(unittest.TestCase):

    def test_fizzbuzz(self):
        # 代表的な入力と出力のパターンを列挙する
        expects = {
            1: '1',
            2: '2',
            3: 'Fizz',
            4: '4',
            5: 'Buzz',
            6: 'Fizz',
            7: '7',
            8: '8',
            9: 'Fizz',
            10: 'Buzz',
            11: '11',
            12: 'Fizz',
            13: '13',
            14: '14',
            15: 'FizzBuzz',
            16: '16',
        }
        for n, expect in expects.items():
            # 特定の入力に大して期待される値が返ってくるか確認する
            result = fizzbuzz.fizzbuzz(n)
            self.assertEqual(result, expect)


if __name__ == '__main__':
    unittest.main(verbosity=2)

上記を test_fizzbuzz.py という名前で保存しよう。

先ほどと同じようにテストディスカバリを実行してみよう。 新たに追加されたテストコードが正しくパスすれば上手くいっている。

$ python -m unittest discover -v
test_fizzbuzz (test_fizzbuzz.TestFizzBuzz) ... ok
test_greet (test_helloworld.TestHelloWorld) ... ok

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

OK

テストしにくい部分をモックと入れ替える

ユニットテストを書いていると、どうしてもテストしにくい部分が出てくる。 典型的な例としては、テスト対象が動作するのに何らかの依存関係があって、それがないと動かないような場合がある。 あるいは、厳密に処理するとあまりにも多くの時間がかかってしまうような場合も考えられる。 そういった場合には、依存している部分をモックと呼ばれる代用部品に入れ替えると良い。

テストしにくい例として以下のサンプルコードを用意した。 このコードの中では do_something() という関数が定義されている。 また、この関数は内部で _take_a_long_time_to_do() という時間のかかる処理を実行している。 なお、実際にやっている処理は最初の greet() 関数と同じで特定の文字列を返すだけとなっている。

# -*- coding: utf-8 -*-

import time


def do_something():
    _take_a_long_time_to_do()
    return 'Hello, World!'


def _take_a_long_time_to_do():
    time.sleep(10)

上記を foobar.py という名前で保存しておこう。

まずは愚直にテストコードを書いてみる

最初は、何も考えずに上記に対応するテストコードを書いてみよう。 やっていることは最初の例と何ら変わらない。

#!/usr/bin/env python
# -*- coding: utf-8 -*-

import unittest

import foobar


class TestFooBar(unittest.TestCase):

    def test_do_something(self):
        # do_something() を呼ぶ
        message = foobar.do_something()
        # 返り値を比較する
        self.assertEqual(message, 'Hello, World!')


if __name__ == '__main__':
    unittest.main(verbosity=2)

上記を test_foobar.py という名前で保存しておく。

準備ができたら上記のテストコードを実行してみよう。 このテストが完了するには 10 秒を要する。

$ python test_foobar.py
test_do_something (__main__.TestFooBar) ... ok

----------------------------------------------------------------------
Ran 1 test in 10.001s

OK

内部的に読んでいる関数をモックに置き換えてみる

続いては、テストコードの実行時間を短縮するために内部的に呼んでいる関数をモックに置き換えてみよう。 モックへの置き換えはいくつかのやり方があるものの、今回は @patch デコレータを使うことにする。

なお、標準ライブラリの unittest にモックの機能が入ったのは Python 3 系から。 なので、もし万が一にも 2 系を使っているときは、次のように別途インストールする必要がある。

$ pip install mock

また、インポートするときのパスも unittest.mock ではなく mock に変わる点に注意しよう。

以下のサンプルコードでは foobar モジュールの _take_a_long_time_to_do() 関数をモックに置き換えている。 モックへの置き換えは、テストコードに @patch() デコレータで置き換えたいオブジェクトのパスを指定する。 置き換えられたオブジェクトの振る舞いは、テストコードのメソッドに引数として渡されるモックオブジェクトでカスタマイズできる。 ただし、今回は置き換えるオブジェクトの動作にテスト対象の関数が特に依存していないので特にカスタマイズは必要ない。 本来であれば返り値やプロパティをいじることになる。

#!/usr/bin/env python
# -*- coding: utf-8 -*-

import unittest
from unittest.mock import patch

import foobar


class TestFooBar(unittest.TestCase):

    # foobar._take_a_long_time_to_do() をモックに置き換える
    @patch('foobar._take_a_long_time_to_do')
    def test_do_something(self, patched_object):
        # モックに置き換えられた状態で do_something() 関数をテストする
        message = foobar.do_something()
        self.assertEqual(message, 'Hello, World!')
        # モックが呼び出されたことを確認する
        self.assertTrue(patched_object.called)


if __name__ == '__main__':
    unittest.main(verbosity=2)

先ほどと同じようにテストコードを実行してみよう。 今度は 10 秒もかからずにテストが完了する。

$ python test_foobar.py 
test_do_something (__main__.TestFooBar) ... ok

----------------------------------------------------------------------
Ran 1 test in 0.001s

OK

本当に必要な部分をモックに置き換える

先ほどの例では、テスト対象が内部的に呼び出している関数をモックに置き換えることで実行時間を短縮できた。 しかし、実は先ほどのやり方には問題がある。 というのも、モックに置き換えたのがアンダースコアから始まる隠し関数だったため。 これの何が問題かというと、テストコードが実装に依存することを意味している。

テストコードが実装に依存することの最大の問題点は、メンテナンスコストが高くつくこと。 例えば、リファクタリングなどをしただけでもテストが正しくパスしなくなる恐れがある。 そのため、テストコードは外部に公開しているインターフェースに対して記述するのが基本となる。 もし、内部的にしか呼び出されていない関数にテストコードを書いていると感じたなら、それは要注意な状態といえる。 テストを書くのであれば、まずインターフェースは何処なのか、それはどう振る舞うべきなのかを考えた上で書くようにしよう。

例えば先ほどの例であれば、内部的に呼んでいる _take_a_long_time_to_do() 関数よりも time.sleep() 関数をモックに置き換えてしまった方が良いかもしれない。 この先 time.sleep() の挙動が変わるような事態は、ちょっとやそっとでは起こらないだろう。

#!/usr/bin/env python
# -*- coding: utf-8 -*-

import unittest
from unittest.mock import patch

import foobar


class TestFooBar(unittest.TestCase):

    # time.sleep() をモックに置き換えてみる
    @patch('time.sleep')
    def test_do_something(self, patched_object):
        message = foobar.do_something()
        self.assertEqual(message, 'Hello, World!')
        self.assertTrue(patched_object.called)


if __name__ == '__main__':
    unittest.main(verbosity=2)

実行結果は先ほどと変わらないけど、変更に対する耐性は先ほどとは段違いなはず。

$ python test_foobar.py 
test_do_something (__main__.TestFooBar) ... ok

----------------------------------------------------------------------
Ran 1 test in 0.001s

OK

標準ライブラリの unittest を使った例は、ここまでで一旦おわりにする。

pytest で、より効率的にテストを書く

ここからは、サードパーティ製のテストフレームワークである pytest について見ていこう。 実際のところ、巷のライブラリなどで標準ライブラリの unittest をそのまま使ってテストを書いている例は少ない。 多くの場合、サードパーティ製のテストフレームワークとして pytest や nose などを使う場合が多い。 その中でも、最近は pytest がデファクトになりつつある。

pytest はサードパーティ製のライブラリなので pip を使ってインストールする必要がある。

$ pip install pytest

実は pytest は unittest と上位互換性がある。 そのため、既存の unittest を使ったプロジェクトにも後から導入しやすい。 試しに pytest のテストランナーで、これまでに書いた unittest のテストコードを実行してみよう。

$ pytest -v         
======================================================================= test session starts ========================================================================
platform darwin -- Python 3.7.3, pytest-4.6.3, py-1.8.0, pluggy-0.12.0 -- /Users/amedama/.virtualenvs/py37/bin/python3.7
cachedir: .pytest_cache
rootdir: /Users/amedama/Documents/temporary/ut
collected 3 items                                                                                                                                                  

test_fizzbuzz.py::TestFizzBuzz::test_fizzbuzz PASSED                                                                                                         [ 33%]
test_foobar.py::TestFooBar::test_do_something PASSED                                                                                                         [ 66%]
test_helloworld.py::TestHelloWorld::test_greet PASSED                                                                                                        [100%]

===================================================================== 3 passed in 0.08 seconds =====================================================================

ちゃんとテストが実行できてパスしたことが分かる。

最初の例を pytest 流に書き直してみる

先ほどは unittest で書いたテストコードも pytest から実行できることを示した。 とはいえ、pytest には pytest 流のテストコードの書き方がある。 試しに、最初に書いたテストコードを pytest 流に書き直してみよう。

書き直したサンプルコードが次の通り。 最初の例よりも、だいぶこざっぱりしている。 例えばテストを書くのにクラスを定義する必要はなく、単なる関数で構わない。 また、値を比較するにも専用の関数やメソッドは必要なくて単なる assert 文を使っている。

#!/usr/bin/env python
# -*- coding: utf-8 -*-

import pytest

import helloworld


def test_greet():
    """テストコードは単なる関数で良い"""
    message = helloworld.greet()
    # 比較は assert 文を使うだけで良い
    assert message == 'Hello, World!'


if __name__ == '__main__':
    pytest.main(['-v', __file__])

上記を実行してみよう。 こざっぱりした内容でも、ちゃんとテストとして機能していることが分かる。

$ python test_helloworld.py 
======================================================================= test session starts ========================================================================
platform darwin -- Python 3.7.3, pytest-4.6.3, py-1.8.0, pluggy-0.12.0 -- /Users/amedama/.virtualenvs/py37/bin/python
cachedir: .pytest_cache
rootdir: /Users/amedama/Documents/temporary/ut
collected 1 item                                                                                                                                                   

test_helloworld.py::test_greet PASSED                                                                                                                        [100%]

===================================================================== 1 passed in 0.03 seconds =====================================================================

FizzBuzz のテストも書き直してみる

続いては FizzBuzz のテストも書き直してみよう。 こちらは、pytest の parametrize という機能を使うとキレイに書くことができる。

以下が parametrize を使った FizzBuzz のテストコードになる。 この機能ではデコレータを使うことでテストの外側に入力と期待される出力の組を定義できる。

#!/usr/bin/env python
# -*- coding: utf-8 -*-

import pytest

import fizzbuzz


# 入力と期待される出力を parametrize で定義する
@pytest.mark.parametrize('n, expect', [
    (1, '1'),
    (2, '2'),
    (3, 'Fizz'),
    (4, '4'),
    (5, 'Buzz'),
    (6, 'Fizz'),
    (7, '7'),
    (8, '8'),
    (9, 'Fizz'),
    (10, 'Buzz'),
    (11, '11'),
    (12, 'Fizz'),
    (13, '13'),
    (14, '14'),
    (15, 'FizzBuzz'),
    (16, '16'),
])
def test_fizzbuzz(n, expect):
    # テストコードがシンプルに保たれる
    assert fizzbuzz.fizzbuzz(n) == expect


if __name__ == '__main__':
    pytest.main(['-v', __file__])

上記を実行すると、各パラメータの組み合わせに応じてテストが走ることが確認できる。

$ python test_fizzbuzz.py 
======================================================================= test session starts ========================================================================
platform darwin -- Python 3.7.3, pytest-4.6.3, py-1.8.0, pluggy-0.12.0 -- /Users/amedama/.virtualenvs/py37/bin/python
cachedir: .pytest_cache
rootdir: /Users/amedama/Documents/temporary/ut
collected 16 items                                                                                                                                                 

test_fizzbuzz.py::test_fizzbuzz[1-1] PASSED                                                                                                                  [  6%]
test_fizzbuzz.py::test_fizzbuzz[2-2] PASSED                                                                                                                  [ 12%]
test_fizzbuzz.py::test_fizzbuzz[3-Fizz] PASSED                                                                                                               [ 18%]
test_fizzbuzz.py::test_fizzbuzz[4-4] PASSED                                                                                                                  [ 25%]
test_fizzbuzz.py::test_fizzbuzz[5-Buzz] PASSED                                                                                                               [ 31%]
test_fizzbuzz.py::test_fizzbuzz[6-Fizz] PASSED                                                                                                               [ 37%]
test_fizzbuzz.py::test_fizzbuzz[7-7] PASSED                                                                                                                  [ 43%]
test_fizzbuzz.py::test_fizzbuzz[8-8] PASSED                                                                                                                  [ 50%]
test_fizzbuzz.py::test_fizzbuzz[9-Fizz] PASSED                                                                                                               [ 56%]
test_fizzbuzz.py::test_fizzbuzz[10-Buzz] PASSED                                                                                                              [ 62%]
test_fizzbuzz.py::test_fizzbuzz[11-11] PASSED                                                                                                                [ 68%]
test_fizzbuzz.py::test_fizzbuzz[12-Fizz] PASSED                                                                                                              [ 75%]
test_fizzbuzz.py::test_fizzbuzz[13-13] PASSED                                                                                                                [ 81%]
test_fizzbuzz.py::test_fizzbuzz[14-14] PASSED                                                                                                                [ 87%]
test_fizzbuzz.py::test_fizzbuzz[15-FizzBuzz] PASSED                                                                                                          [ 93%]
test_fizzbuzz.py::test_fizzbuzz[16-16] PASSED                                                                                                                [100%]

==================================================================== 16 passed in 0.07 seconds =====================================================================

多彩なプラグインを使いこなす

pytest には色々な機能を持ったプラグインが存在することも魅力の一つといえる。

例えばテストを実行するのと一緒に flake8 を実行できる pytest-flake8 は使われることが多い。

$ pip install pytest-flake8

このプラグインを使うと、テストランナーに --flake8 オプションを渡すことで flake8 を実行できるようになる。

$ pytest -v --flake8
======================================================================= test session starts ========================================================================
platform darwin -- Python 3.7.3, pytest-4.6.3, py-1.8.0, pluggy-0.12.0 -- /Users/amedama/.virtualenvs/py37/bin/python3.7
cachedir: .pytest_cache
rootdir: /Users/amedama/Documents/temporary/ut
plugins: flake8-1.0.4
collected 24 items                                                                                                                                                 

fizzbuzz.py::FLAKE8 PASSED                                                                                                                                   [  4%]
foobar.py::FLAKE8 PASSED                                                                                                                                     [  8%]
helloworld.py::FLAKE8 PASSED                                                                                                                                 [ 12%]
test_fizzbuzz.py::FLAKE8 PASSED                                                                                                                              [ 16%]
test_fizzbuzz.py::test_fizzbuzz[1-1] PASSED                                                                                                                  [ 20%]
test_fizzbuzz.py::test_fizzbuzz[2-2] PASSED                                                                                                                  [ 25%]
test_fizzbuzz.py::test_fizzbuzz[3-Fizz] PASSED                                                                                                               [ 29%]
test_fizzbuzz.py::test_fizzbuzz[4-4] PASSED                                                                                                                  [ 33%]
test_fizzbuzz.py::test_fizzbuzz[5-Buzz] PASSED                                                                                                               [ 37%]
test_fizzbuzz.py::test_fizzbuzz[6-Fizz] PASSED                                                                                                               [ 41%]
test_fizzbuzz.py::test_fizzbuzz[7-7] PASSED                                                                                                                  [ 45%]
test_fizzbuzz.py::test_fizzbuzz[8-8] PASSED                                                                                                                  [ 50%]
test_fizzbuzz.py::test_fizzbuzz[9-Fizz] PASSED                                                                                                               [ 54%]
test_fizzbuzz.py::test_fizzbuzz[10-Buzz] PASSED                                                                                                              [ 58%]
test_fizzbuzz.py::test_fizzbuzz[11-11] PASSED                                                                                                                [ 62%]
test_fizzbuzz.py::test_fizzbuzz[12-Fizz] PASSED                                                                                                              [ 66%]
test_fizzbuzz.py::test_fizzbuzz[13-13] PASSED                                                                                                                [ 70%]
test_fizzbuzz.py::test_fizzbuzz[14-14] PASSED                                                                                                                [ 75%]
test_fizzbuzz.py::test_fizzbuzz[15-FizzBuzz] PASSED                                                                                                          [ 79%]
test_fizzbuzz.py::test_fizzbuzz[16-16] PASSED                                                                                                                [ 83%]
test_foobar.py::FLAKE8 PASSED                                                                                                                                [ 87%]
test_foobar.py::TestFooBar::test_do_something PASSED                                                                                                         [ 91%]
test_helloworld.py::FLAKE8 PASSED                                                                                                                            [ 95%]
test_helloworld.py::test_greet PASSED                                                                                                                        [100%]

==================================================================== 24 passed in 0.28 seconds =====================================================================

あるいはテストカバレッジを計測するための pytest-cov というプラグインも有名。 これは ptyest と coverage のインテグレーションを提供している。

$ pip install pytest-cov

このプラグインは --cov というオプションをつけることでテストカバレッジの計測ができるようになる。

$ pytest -v --cov=.
======================================================================= test session starts ========================================================================
platform darwin -- Python 3.7.3, pytest-4.6.3, py-1.8.0, pluggy-0.12.0 -- /Users/amedama/.virtualenvs/py37/bin/python3.7
cachedir: .pytest_cache
rootdir: /Users/amedama/Documents/temporary/ut
plugins: cov-2.7.1, flake8-1.0.4
collected 18 items                                                                                                                                                 

test_fizzbuzz.py::test_fizzbuzz[1-1] PASSED                                                                                                                  [  5%]
test_fizzbuzz.py::test_fizzbuzz[2-2] PASSED                                                                                                                  [ 11%]
test_fizzbuzz.py::test_fizzbuzz[3-Fizz] PASSED                                                                                                               [ 16%]
test_fizzbuzz.py::test_fizzbuzz[4-4] PASSED                                                                                                                  [ 22%]
test_fizzbuzz.py::test_fizzbuzz[5-Buzz] PASSED                                                                                                               [ 27%]
test_fizzbuzz.py::test_fizzbuzz[6-Fizz] PASSED                                                                                                               [ 33%]
test_fizzbuzz.py::test_fizzbuzz[7-7] PASSED                                                                                                                  [ 38%]
test_fizzbuzz.py::test_fizzbuzz[8-8] PASSED                                                                                                                  [ 44%]
test_fizzbuzz.py::test_fizzbuzz[9-Fizz] PASSED                                                                                                               [ 50%]
test_fizzbuzz.py::test_fizzbuzz[10-Buzz] PASSED                                                                                                              [ 55%]
test_fizzbuzz.py::test_fizzbuzz[11-11] PASSED                                                                                                                [ 61%]
test_fizzbuzz.py::test_fizzbuzz[12-Fizz] PASSED                                                                                                              [ 66%]
test_fizzbuzz.py::test_fizzbuzz[13-13] PASSED                                                                                                                [ 72%]
test_fizzbuzz.py::test_fizzbuzz[14-14] PASSED                                                                                                                [ 77%]
test_fizzbuzz.py::test_fizzbuzz[15-FizzBuzz] PASSED                                                                                                          [ 83%]
test_fizzbuzz.py::test_fizzbuzz[16-16] PASSED                                                                                                                [ 88%]
test_foobar.py::TestFooBar::test_do_something PASSED                                                                                                         [ 94%]
test_helloworld.py::test_greet PASSED                                                                                                                        [100%]

---------- coverage: platform darwin, python 3.7.3-final-0 -----------
Name                 Stmts   Miss  Cover
----------------------------------------
fizzbuzz.py              8      0   100%
foobar.py                6      0   100%
helloworld.py            2      0   100%
test_fizzbuzz.py         6      1    83%
test_foobar.py          10      1    90%
test_helloworld.py       7      1    86%
----------------------------------------
TOTAL                   39      3    92%


==================================================================== 18 passed in 0.13 seconds =====================================================================

一般的な pytest のディレクトリ構成

ところでここまでテスト対象とテストコードを一つのディレクトリに雑然と放り込んできた。 巷のライブラリなどを見ると pytest を使ったプロジェクトでは、次のように tests というディレクトリを専用に用意することが多いように思う。

$ mkdir tests
$ mv test_*.py tests
$ touch tests/__init__.py

テスト対象のモジュール・パッケージについては、tests と同じ階層の別ディレクトリに入れられる場合が多い。

$ mkdir example
$ mv *.py example
$ touch example/__init__.py

ようするに、こんな感じ。 これはつまり example と tests という Python のパッケージを用意していることになる。

$ tree
.
├── example
│   ├── __init__.py
│   ├── fizzbuzz.py
│   ├── foobar.py
│   └── helloworld.py
└── tests
    ├── __init__.py
    ├── test_fizzbuzz.py
    ├── test_foobar.py
    └── test_helloworld.py

2 directories, 8 files

ただ、上記のような変更を加えると、先ほど書いたテストコードは少しだけ修正が必要になる。 というのも helloworldfizzbuzz モジュールが example パッケージ配下に移動しているため。 そこで、次のようにインポート文を修正する。

#!/usr/bin/env python
# -*- coding: utf-8 -*-

import pytest

# example 配下にある helloworld モジュールをインポートする
from example import helloworld


def test_greet():
    message = helloworld.greet()
    assert message == 'Hello, World!'


if __name__ == '__main__':
    pytest.main(['-v', __file__])

インポート文を変更したらテストランナーを実行してみよう。 次のように、ちゃんとテストがパスすれば上手くいっている。

$ pytest -v        
======================================================================= test session starts ========================================================================
platform darwin -- Python 3.7.3, pytest-4.6.3, py-1.8.0, pluggy-0.12.0 -- /Users/amedama/.virtualenvs/py37/bin/python3.7
cachedir: .pytest_cache
rootdir: /Users/amedama/Documents/temporary/ut
plugins: cov-2.7.1, flake8-1.0.4
collected 18 items                                                                                                                                                 

tests/test_fizzbuzz.py::test_fizzbuzz[1-1] PASSED                                                                                                            [  5%]
tests/test_fizzbuzz.py::test_fizzbuzz[2-2] PASSED                                                                                                            [ 11%]
tests/test_fizzbuzz.py::test_fizzbuzz[3-Fizz] PASSED                                                                                                         [ 16%]
tests/test_fizzbuzz.py::test_fizzbuzz[4-4] PASSED                                                                                                            [ 22%]
tests/test_fizzbuzz.py::test_fizzbuzz[5-Buzz] PASSED                                                                                                         [ 27%]
tests/test_fizzbuzz.py::test_fizzbuzz[6-Fizz] PASSED                                                                                                         [ 33%]
tests/test_fizzbuzz.py::test_fizzbuzz[7-7] PASSED                                                                                                            [ 38%]
tests/test_fizzbuzz.py::test_fizzbuzz[8-8] PASSED                                                                                                            [ 44%]
tests/test_fizzbuzz.py::test_fizzbuzz[9-Fizz] PASSED                                                                                                         [ 50%]
tests/test_fizzbuzz.py::test_fizzbuzz[10-Buzz] PASSED                                                                                                        [ 55%]
tests/test_fizzbuzz.py::test_fizzbuzz[11-11] PASSED                                                                                                          [ 61%]
tests/test_fizzbuzz.py::test_fizzbuzz[12-Fizz] PASSED                                                                                                        [ 66%]
tests/test_fizzbuzz.py::test_fizzbuzz[13-13] PASSED                                                                                                          [ 72%]
tests/test_fizzbuzz.py::test_fizzbuzz[14-14] PASSED                                                                                                          [ 77%]
tests/test_fizzbuzz.py::test_fizzbuzz[15-FizzBuzz] PASSED                                                                                                    [ 83%]
tests/test_fizzbuzz.py::test_fizzbuzz[16-16] PASSED                                                                                                          [ 88%]
tests/test_foobar.py::TestFooBar::test_do_something PASSED                                                                                                   [ 94%]
tests/test_helloworld.py::test_greet PASSED                                                                                                                  [100%]

==================================================================== 18 passed in 0.13 seconds =====================================================================

そんなかんじで。

インターネットに疎通のないマシンに SSH Remote Port Forwarding + Squid で Web にアクセスさせる

インターネットに直接つながっていないマシンというのは意外とよくある。 とはいえ、そういったマシンでも当然のことながらセットアップ等の作業は必要になる。 その際、作業に必要なファイルは大抵の場合に SCP などで転送することになると思う。 とはいえ、もしマシンから直接 Web につながれば楽ができるはず。 今回は、そういった状況で SSH Remote Port Forwarding と Squid (Web プロキシのソフトウェア) を使って Web への疎通を提供する方法を試してみる。 なお、提供できるのはあくまで HTTP/HTTPS 等に限られ、ICMP や UDP といったプロトコルは通らない。

構成について

最初に、ざっくりとした構成図を以下に示す。 インターネットにつながらないマシンは SSH Server を想定する。 そして、SSH Client がインターネットにつながるマシンで、SSH Server に接続しに行く。 SSH Client のマシンでは Squid を 8080 ポートで起動している。 今回の主眼は、SSH Client のマシンの 8080 ポートを Remote Port Forwarding で SSH Server にも見えるようにするところ。 SSH Server は自身の 8080 ポートを Web プロキシとして利用することで、Web への疎通が手に入る。

f:id:momijiame:20190616102958p:plain

使った環境について

続いては、今回の検証に使った環境について説明する。

インターネットにつながっているマシン (SSH Client) としては、以下の通り macOS のマシンを用意した。

$ sw_vers    
ProductName:    Mac OS X
ProductVersion: 10.14.5
BuildVersion:   18F132

このマシンは、次のようにインターネットに疎通がある。

$ ping -c 3 8.8.8.8
PING 8.8.8.8 (8.8.8.8): 56 data bytes
64 bytes from 8.8.8.8: icmp_seq=0 ttl=56 time=9.156 ms
64 bytes from 8.8.8.8: icmp_seq=1 ttl=56 time=30.316 ms
64 bytes from 8.8.8.8: icmp_seq=2 ttl=56 time=18.479 ms

--- 8.8.8.8 ping statistics ---
3 packets transmitted, 3 packets received, 0.0% packet loss
round-trip min/avg/max/stddev = 9.156/19.317/30.316/8.659 ms

そして、インターネットにつながっていないマシン (SSH Server) は、以下の通り Ubuntu 18.04 LTS のマシンを用意した。 このマシンは VirtualBox を使って先ほどの macOS 上で仮想マシンとして稼働させている。

$ cat /etc/lsb-release 
DISTRIB_ID=Ubuntu
DISTRIB_RELEASE=18.04
DISTRIB_CODENAME=bionic
DISTRIB_DESCRIPTION="Ubuntu 18.04.2 LTS"
$ uname -r
4.15.0-51-generic

こちらは以下のようにインターネットに疎通がない。 かろうじて 192.168.33.0/24 のネットワークを経由して前述した macOS のマシンと疎通がある。

$ ip addr show
1: lo: <LOOPBACK,UP,LOWER_UP> mtu 65536 qdisc noqueue state UNKNOWN group default qlen 1000
    link/loopback 00:00:00:00:00:00 brd 00:00:00:00:00:00
    inet 127.0.0.1/8 scope host lo
       valid_lft forever preferred_lft forever
    inet6 ::1/128 scope host 
       valid_lft forever preferred_lft forever
2: enp0s8: <BROADCAST,MULTICAST,UP,LOWER_UP> mtu 1500 qdisc fq_codel state UP group default qlen 1000
    link/ether 08:00:27:42:06:7b brd ff:ff:ff:ff:ff:ff
    inet 192.168.33.10/24 brd 192.168.33.255 scope global enp0s8
       valid_lft forever preferred_lft forever
    inet6 fe80::a00:27ff:fe42:67b/64 scope link 
       valid_lft forever preferred_lft forever
$ ip route show
192.168.33.0/24 dev enp0s8 proto kernel scope link src 192.168.33.10 
$ ping -c 3 8.8.8.8
connect: Network is unreachable

なお、このマシンは Vagrant + VirtualBox を使って仮想マシンとして用意した。 前述した macOS 上で稼働している。 ただし SSH については vagrant コマンドを使う代わりに、次のようにしてログインしている。

$ ssh -i .vagrant/machines/default/virtualbox/private_key \
  -o StrictHostKeyChecking=no -o UserKnownHostsFile=/dev/null \
  -l vagrant 192.168.33.10

Web プロキシをインストールする

ここからは、実際に SSH Server のマシンに Web への疎通を提供するまでに必要な作業について記載していく。 まずは SSH Client 側のマシンに Squid (Web プロキシ) をインストールする。

使っているのが macOS なので Homebrew を使うと楽にインストールできる。

$ brew install squid

コンフィグを編集する場合には、初期状態のものをコピーしてバックアップしておく。

$ cp $(brew --prefix)/etc/squid.conf{,.bak}

今回はコンフィグの編集例として、使うポートをよく利用される 8080 に変更してみる。

$ sed -i -e "
  s:^http_port.*$:http_port 8080:
" $(brew --prefix)/etc/squid.conf
$ grep http_port $(brew --prefix)/etc/squid.conf      
http_port 8080

あと、念のため次のようにしてどのような ACL が入っているのか、あらかじめ確認しておいた方が良いと思う。 オープンプロキシになってると危険なので。 とはいえ、おそらくデフォルトでプライベートアドレスからのアクセスしか認めないようになっているはず。

$ grep acl $(brew --prefix)/etc/squid.conf
$ grep deny $(brew --prefix)/etc/squid.conf

Squid のサービスを開始する。

$ brew services start squid

次のように 8080 ポートを Listen していれば良い。

$ lsof -i:8080 | grep -i squid
squid     58763 amedama   14u  IPv6 0x4e9127c629e616d5      0t0  TCP *:http-alt (LISTEN)

Remote Port Forwarding で Web プロキシを利用する

続いては SSH Server 側の作業に入る。

やることは単純で、次のように SSH するときに -R オプションで Remote Port Forwarding する。 以下では自身の 8080 ポートをリモートの localhost:8080 で見られるようにしている。

$ ssh -i .vagrant/machines/default/virtualbox/private_key \
      -R 8080:localhost:8080 \
      -o StrictHostKeyChecking=no -o UserKnownHostsFile=/dev/null \
      -l vagrant 192.168.33.10

ログインしたらポートを localhost:8080 で Listen していることを確認しておこう。

$ ss -tlnp | grep 8080
LISTEN   0         128               127.0.0.1:8080             0.0.0.0:*       
LISTEN   0         128                   [::1]:8080                [::]:*       

これで localhost:8080 経由で Web プロキシが使えるようになった。 あとは一般的なプロキシを使うのと同じ。 例えば環境変数に設定を入れておこう。

$ export http_proxy=http://localhost:8080
$ export https_proxy=$http_proxy

試しに wget を使って Ubuntu のイメージファイルのハッシュファイルを取得してみよう。

$ wget http://ftp.riken.jp/Linux/ubuntu-releases/bionic/SHA256SUMS

次のように、ちゃんと取得できれば上手くいっている。

$ cat SHA256SUMS 
22580b9f3b186cc66818e60f44c46f795d708a1ad86b9225c458413b638459c4 *ubuntu-18.04.2-desktop-amd64.iso
ea6ccb5b57813908c006f42f7ac8eaa4fc603883a2d07876cf9ed74610ba2f53 *ubuntu-18.04.2-live-server-amd64.iso

いじょう。

Squid Proxy Server 3.1: Beginner's Guide (English Edition)

Squid Proxy Server 3.1: Beginner's Guide (English Edition)

Squid: The Definitive Guide: The Definitive Guide (Definitive Guides) (English Edition)

Squid: The Definitive Guide: The Definitive Guide (Definitive Guides) (English Edition)

Python: 定期実行のアルゴリズムについて

今回は割と小ネタで、特定の処理を定期実行するようなプログラムを書く場合について考えてみる。 ただし、前提としてあくまで定期実行は Python の中で処理して cron 的なものには頼らないものとする。

使った環境は次の通り。

$ sw_vers
ProductName:    Mac OS X
ProductVersion: 10.14.5
BuildVersion:   18F132
$ python -V       
Python 3.7.3

ダメなパターン: 定期実行の時間だけ単純に sleep する

最初に考えられるのは、定期実行したい間隔で time.sleep() のような関数を使ってインターバルを入れるというもの。 ただし、このパターンでは肝心の定期実行したい処理にかかる時間が考慮できていない。

以下のサンプルコードでは 3 秒ごとに定期実行しているつもりでいる。 しかし、肝心の定期実行したい処理には 2 秒かかっている。

#!/usr/bin/env python
# -*- coding: utf-8 -*-

from datetime import datetime
import time


def interval_task():
    """定期的に実行したい何か時間のかかる処理"""
    now = datetime.now()
    print(now.strftime('%H:%M:%S.%f'))
    # 実行に 2 秒くらいかかる
    time.sleep(2)


def schedule(interval_sec, callable_task, args=None, kwargs=None):
    """何らかの処理を定期的に実行する関数"""
    args = args or []
    kwargs = kwargs or {}
    while True:
        callable_task(*args, **kwargs)  # ここで時間を食う
        time.sleep(interval_sec)  # さらにスリープしてしまう


def main():
    # 3 秒ごとに実行している...つもり
    schedule(interval_sec=3, callable_task=interval_task)


if __name__ == '__main__':
    main()

上記を実行してみる。 表示を見て分かる通り 3 + 2 = 5 秒の間隔で時刻が表示されてしまっている。

$ python sched1.py 
11:24:16.698956
11:24:21.709522
11:24:26.710433
11:24:31.718227
...

ダメなパターン: 処理にかかる時間を開始・終了の前後で毎回計測して計算する

続いて考えられるのが、定期実行したい処理の前後で開始・終了時刻を計測してスリープする時間を補正するというもの。 ただし、この場合は定期実行したい処理以外にかかる処理時間が考慮できていない。

以下のサンプルコードでは定期実行したい処理の前後で時刻を計測してスリープする時間を補正している。

#!/usr/bin/env python
# -*- coding: utf-8 -*-

from datetime import datetime
import time


def interval_task():
    """定期的に実行したい何か時間のかかる処理"""
    now = datetime.now()
    print(now.strftime('%H:%M:%S.%f'))
    # 実行に 2 秒くらいかかる
    time.sleep(2)


def schedule(interval_sec, callable_task, args=None, kwargs=None):
    """何らかの処理を定期的に実行する関数"""
    args = args or []
    kwargs = kwargs or {}
    while True:
        # 処理開始時間を取得する
        start_timing = datetime.now()

        callable_task(*args, **kwargs)

        # 処理完了時間を取得する
        end_timing = datetime.now()
        # 実行間隔との差分を取る
        time_delta_sec = (end_timing - start_timing).total_seconds()
        # スリープすべき時間を計算する
        sleep_sec = interval_sec - time_delta_sec

        time.sleep(max(sleep_sec, 0))


def main():
    # 3 秒ごとに実行される...と良いなあ
    schedule(interval_sec=3, callable_task=interval_task)


if __name__ == '__main__':
    main()

上記を実行してみると、だいたい 3 秒ごとに時刻が表示されるため一見すると上手くいっているように見える。 しかし、よく見ると 1 ~ 10 ミリ秒の単位は単調に時刻が増加していることが分かる。 これは、定期実行以外の処理にかかる時間が考慮できていないため、実際には間隔がわずかに長くなってしまっている。

$ python sched2.py 
11:25:41.820993
11:25:44.826223
11:25:47.831453
11:25:50.836683
11:25:53.840000
11:25:56.844355
11:25:59.849602
...

また、間隔が少し長くなる以外にも、もう一つの問題がある。 定期実行したい処理が実行間隔よりもかかってしまうと、スリープする時間がマイナスになってしまう。 そのため、スリープを全く入れなくても実際より実行間隔が長くなってしまう。 いわゆるバッチの突き抜けみたいな状態。

以下のサンプルコードでは実行間隔として 3 秒を意図しているにもかかわらず、実際には定期実行の処理には 4 秒かかる。

#!/usr/bin/env python
# -*- coding: utf-8 -*-

from datetime import datetime
import time


def interval_task():
    """定期的に実行したい何か時間のかかる処理"""
    now = datetime.now()
    print(now.strftime('%H:%M:%S.%f'))
    # 実行に 4 秒くらいかかる
    time.sleep(4)


def schedule(interval_sec, callable_task, args=None, kwargs=None):
    """何らかの処理を定期的に実行する関数"""
    args = args or []
    kwargs = kwargs or {}
    while True:
        start_timing = datetime.now()

        # ここが意図した実行間隔よりも長くかかる (バッチの突き抜け)
        callable_task(*args, **kwargs)

        end_timing = datetime.now()
        time_delta_sec = (end_timing - start_timing).total_seconds()
        sleep_sec = interval_sec - time_delta_sec

        # sleep_sec が負の値になる
        time.sleep(max(sleep_sec, 0))


def main():
    schedule(interval_sec=3, callable_task=interval_task)


if __name__ == '__main__':
    main()

上記を実行すると、次のように本来意図した 3 秒ではなく 4 秒間隔で実行される。

$ python sched3.py 
11:26:52.933087
11:26:56.938491
11:27:00.943913
11:27:04.948757
11:27:08.949384

解決策: 特定の基準時刻を元にスリープする時間を補正しつつ別のスレッドで実行する

先ほどの問題点を解決するために二つの施策が必要となる。 まずひとつ目は一つの基準時刻を設けて、それを元にスリープする時刻を補正する。 これで、実行間隔がやや長くなってしまう問題が解決できる。 もうひとつは定期実行の処理を別のスレッドで実行することで、バッチの突き抜けを防止できる。

以下のサンプルコードでは特定の基準時刻を元にスリープする時間を補正している。 具体的には、処理の最初で取得した時刻から剰余演算でスリープすべき時間を計算する。 その上で定期実行の処理は別のスレッドを起動している。

#!/usr/bin/env python
# -*- coding: utf-8 -*-

from datetime import datetime
import time
import threading


def interval_task():
    """定期的に実行したい何か時間のかかる処理"""
    now = datetime.now()
    print(now.strftime('%H:%M:%S.%f'))
    # 実行に 4 秒くらいかかる
    time.sleep(4)


def schedule(interval_sec, callable_task,
             args=None, kwargs=None):
    """何らかの処理を定期的に実行する関数"""
    args = args or []
    kwargs = kwargs or {}
    # 基準時刻を作る
    base_timing = datetime.now()
    while True:
        # 処理を別スレッドで実行する
        t = threading.Thread(target=callable_task,
                             args=args, kwargs=kwargs)
        t.start()

        # 基準時刻と現在時刻の剰余を元に、次の実行までの時間を計算する
        current_timing = datetime.now()
        elapsed_sec = (current_timing - base_timing).total_seconds()
        sleep_sec = interval_sec - (elapsed_sec % interval_sec)

        time.sleep(max(sleep_sec, 0))


def main():
    schedule(interval_sec=3, callable_task=interval_task)


if __name__ == '__main__':
    main()

上記の実行結果は次の通り。 定期実行の処理に 4 秒かかったとしても、正しく 3 秒間隔で処理が実行できていることが分かる。 また、ミリ秒単位についても単調増加していない。

$ python sched4.py 
11:27:38.941382
11:27:41.946648
11:27:44.942504
11:27:47.946613
11:27:50.942494
...

オプション: スレッドプールを使う

先ほどの処理では単純に定期実行の度にスレッドを起動していた。 しかし、スレッドの生成には時間的・空間的な計算量がかかる。 もし、オーバーヘッドを小さくしたり、メモリを過剰に使われたくないときはスレッドプールを利用することが検討できるはず。

以下のサンプルコードではワーカーが 10 のスレッドプールを使って先ほどと同じ処理を実行している。 これで、10 を越えるスレッドが同時に生成されることがなくなる。 また、一度作られたスレッドは再利用されるため時間的な計算量でもわずかながら有利になるはず。

#!/usr/bin/env python
# -*- coding: utf-8 -*-

from datetime import datetime
import time
from concurrent.futures import ThreadPoolExecutor


def interval_task():
    """定期的に実行したい何か時間のかかる処理"""
    now = datetime.now()
    print(now.strftime('%H:%M:%S.%f'))
    # 実行に 4 秒くらいかかる
    time.sleep(4)


def schedule(interval_sec, callable_task,
             args=None, kwargs=None,
             workers_n=10):
    """何らかの処理を定期的に実行する関数"""
    args = args or []
    kwargs = kwargs or {}
    base_timing = datetime.now()

    # 必要以上にスレッドが生成されないようにスレッドプールを使う
    with ThreadPoolExecutor(max_workers=workers_n) as executor:
        while True:
            future = executor.submit(callable_task,
                                     *args, **kwargs)

            current_timing = datetime.now()
            elapsed_sec = (current_timing - base_timing).total_seconds()
            sleep_sec = interval_sec - (elapsed_sec % interval_sec)

            time.sleep(max(sleep_sec, 0))


def main():
    schedule(interval_sec=3, callable_task=interval_task)


if __name__ == '__main__':
    main()

上記の実行結果は次の通り。

$ python sched5.py 
11:30:30.000741
11:30:33.005632
11:30:36.002308
11:30:39.003651
11:30:42.002279
...

ただし、スレッドプールにはプールの上限に達したときにバッチの突き抜けが起こるという問題がある。 メモリが枯渇して OOM Killer に殺されるか、殺されないけど突き抜けるかは状況によって選ぶのが良いと思う。 とはいえメモリが枯渇するほどスケジュール実行のスレッドが生成される状況って、暴走しているような場合くらいな気もする?

以下のサンプルコードではワーカーの数を 1 に制限することで、意図的に突き抜けを起こるようにしている。

#!/usr/bin/env python
# -*- coding: utf-8 -*-

from datetime import datetime
import time
from concurrent.futures import ThreadPoolExecutor


def interval_task():
    """定期的に実行したい何か時間のかかる処理"""
    now = datetime.now()
    print(now.strftime('%H:%M:%S.%f'))
    # 実行に 4 秒くらいかかる
    time.sleep(4)


def schedule(interval_sec, callable_task,
             args=None, kwargs=None,
             workers_n=10):
    """何らかの処理を定期的に実行する関数"""
    args = args or []
    kwargs = kwargs or {}
    base_timing = datetime.now()

    with ThreadPoolExecutor(max_workers=workers_n) as executor:
        while True:
            future = executor.submit(callable_task,
                                     *args, **kwargs)

            current_timing = datetime.now()
            elapsed_sec = (current_timing - base_timing).total_seconds()
            sleep_sec = interval_sec - (elapsed_sec % interval_sec)

            time.sleep(max(sleep_sec, 0))


def main():
    # 並列度を 1 にして時間のかかる処理を 1 秒ごとに実行した場合
    # スレッドが空くまで待たされる
    schedule(interval_sec=1, callable_task=interval_task,
             workers_n=1)


if __name__ == '__main__':
    main()

上記を実行してみよう。 たしかに突き抜けて処理に 4 秒かかっていることが分かる。

$ python sched7.py 
11:32:48.013018
11:32:52.014658
11:32:56.018407
11:33:00.024064
11:33:04.029571
...

なお、巷にはスケジュール実行するためのライブラリも色々とあって、それらを使うことで色々と楽ができる。 ただし、意図通りに動作させるためには上記のような考慮点についてあらかじめ検討しておく必要がある。 また、リアルタイム OS でない限り今回用いたようなコードで正しく定期実行されるという保証は実のところないはず。

いじょう。