IPFactory Advent Calender 2019 14日目

IPFactory所属、ISC 1年のpycysです。
ナンプレをPythonで解くの後編です。

コード書いたよ

やっとできました。
大抵の問題は1秒以内で解けるはずです。
間に合わせるためにアルゴリズムをそのまま書いたので、読みやすくは書かれていません。
後でもうちょい綺麗に書きます。
追加したアルゴリズムもあるので、それの説明も含め後日また記事書きます

nampre.py

import time, copy as cp

# 3×3枠n確定判定&排除処理
def cubic_frame_judgment():
    flag = False
    for i in range(3):
        for j in range(3):
            indices = [() for n in range(9)]
            for sub_i in range(i*3, i*3+3):
                for sub_j in range(j*3, j*3+3):
                    for num in unconfirmed_numbers[sub_i][sub_j]:
                        indices[num-1] = (sub_i, sub_j) if not indices[num-1] else (9,9)
            for index in indices:
                if index != (9,9) and index not in subscripts:
                    flag = True
                    nampre[index[0]][index[1]] = indices.index(index)+1
                    column_excluder(index[0], index[1], indices.index(index)+1)
                    row_excluder(index[0], index[1], indices.index(index)+1)
                    cubic_frame_excluder(index[0], index[1], indices.index(index)+1)
                    subscripts.add((index[0], index[1]))
    return flag

# 縦列n確定判定&排除処理
def column_judgment():
    flag = False
    for j in range(9):
        indices = [() for n in range(9)]
        for i in range(9):
            for num in unconfirmed_numbers[i][j]:
                indices[num-1] = (i, j) if not indices[num-1] else (9,9)
        for index in indices:
            if index != (9,9) and index not in subscripts:
                flag = True
                nampre[index[0]][index[1]] = indices.index(index)+1
                column_excluder(index[0], index[1], indices.index(index)+1)
                row_excluder(index[0], index[1], indices.index(index)+1)
                cubic_frame_excluder(index[0], index[1], indices.index(index)+1)
                subscripts.add((index[0], index[1]))
    return flag

# 横列n確定判定&排除処理
def row_judgment():
    flag = False
    for i in range(9):
        indices = [() for n in range(9)]
        for j in range(9):
            for num in unconfirmed_numbers[i][j]:
                indices[num-1] = (i, j) if not indices[num-1] else (9,9)
        for index in indices:
            if index != (9,9) and index not in subscripts:
                flag = True
                nampre[index[0]][index[1]] = indices.index(index)+1
                column_excluder(index[0], index[1], indices.index(index)+1)
                row_excluder(index[0], index[1], indices.index(index)+1)
                cubic_frame_excluder(index[0], index[1], indices.index(index)+1)
                subscripts.add((index[0], index[1]))
    return flag

# 置ける数字が１つしかないマスを探す&排除処理
def only_one_judgment():
    flag = False
    for i in range(9):
        for j in range(9):
            if len(unconfirmed_numbers[i][j]) == 1 and (i,j) not in subscripts:
                flag = True
                num = unconfirmed_numbers[i][j][0]
                nampre[i][j] = num
                column_excluder(i, j, num)
                row_excluder(i, j, num)
                cubic_frame_excluder(i, j, num)
                subscripts.add((i,j))
    return flag

# 例外処理１
def cubic_tumor_excluder():
    flag = False
    # 3×3枠check
    for i in range(3):
        for j in range(3):
            overlapping_numbers = [[] for i in range(9)]
            for sub_i in range(i*3, i*3+3):
                for sub_j in range(j*3, j*3+3):
                    for num in unconfirmed_numbers[sub_i][sub_j]:
                        overlapping_numbers[num-1].append((sub_i, sub_j))
            for index_box in overlapping_numbers:
                if overlapping_numbers.count(index_box) == len(index_box) >= 2:
                    nums = [index+1 for index, indices in enumerate(overlapping_numbers) if indices == index_box]
                    for index in index_box:
                        if unconfirmed_numbers[index[0]][index[1]] != nums:
                            flag = True
                            unconfirmed_numbers[index[0]][index[1]] = cp.deepcopy(nums)
    # 横列check
    for i in range(9):
        overlapping_numbers = [[] for i in range(9)]
        for j in range(9):
            for num in unconfirmed_numbers[i][j]:
                overlapping_numbers[num-1].append((i, j))
        for index_box in overlapping_numbers:
            if overlapping_numbers.count(index_box) == len(index_box) >= 2:
                nums = [index+1 for index, indices in enumerate(overlapping_numbers) if indices == index_box]
                for index in index_box:
                    if unconfirmed_numbers[index[0]][index[1]] != nums:
                        flag = True
                        unconfirmed_numbers[index[0]][index[1]] = cp.deepcopy(nums)
    # 縦列check
    for j in range(9):
        overlapping_numbers = [[] for i in range(9)]
        for i in range(9):
            for num in unconfirmed_numbers[i][j]:
                overlapping_numbers[num-1].append((i, j))
        for index_box in overlapping_numbers:
            if overlapping_numbers.count(index_box) == len(index_box) >= 2:
                nums = [index+1 for index, indices in enumerate(overlapping_numbers) if indices == index_box]
                for index in index_box:
                    if unconfirmed_numbers[index[0]][index[1]] != nums:
                        flag = True
                        unconfirmed_numbers[index[0]][index[1]] = cp.deepcopy(nums)
    return flag

# 例外1の逆バージョン
def remainder_excluder():
    flag = False
    # 3×3枠
    for i in range(3):
        for j in range(3):
            cubic_frame_nums = []
            for sub_i in range(i*3, i*3+3):
                for sub_j in range(j*3, j*3+3):
                    cubic_frame_nums.append(cp.deepcopy(unconfirmed_numbers[sub_i][sub_j]))
            for nums in cubic_frame_nums:
                if len(nums) == cubic_frame_nums.count(nums) > 1:
                    for sub_i in range(i*3, i*3+3):
                        for sub_j in range(j*3, j*3+3):
                            if unconfirmed_numbers[sub_i][sub_j] != nums:
                                for num in nums:
                                    if num in unconfirmed_numbers[sub_i][sub_j]:
                                        unconfirmed_numbers[sub_i][sub_j].remove(num)
                                        flag = True
    # 横
    for i in range(9):
        row_line_nums = []
        for j in range(9):
            row_line_nums.append(cp.deepcopy(unconfirmed_numbers[i][j]))
        for nums in row_line_nums:
            if len(nums) == row_line_nums.count(nums) > 1:
                for j in range(9):
                    if unconfirmed_numbers[i][j] != nums:
                        for num in nums:
                            if num in unconfirmed_numbers[i][j]:
                                unconfirmed_numbers[i][j].remove(num)
                                flag = True
    # 縦
    for j in range(9):
        column_line_nums = []
        for i in range(9):
            column_line_nums.append(cp.deepcopy(unconfirmed_numbers[i][j]))
        for nums in column_line_nums:
            if len(nums) == column_line_nums.count(nums) > 1:
                for i in range(9):
                    if unconfirmed_numbers[i][j] != nums:
                        for num in nums:
                            if num in unconfirmed_numbers[i][j]:
                                unconfirmed_numbers[i][j].remove(num)
                                flag = True
    return flag

# 例外処理２
def line_confirm():
    flag = False
    for i in range(3):
        for j in range(3):
            # 横処理
            row_lines = []
            for sub_i in range(i*3, i*3+3):
                row_line = []
                for sub_j in range(j*3, j*3+3):
                    for num in unconfirmed_numbers[sub_i][sub_j]:
                        row_line.append(num)
                row_lines.append(list(set(row_line)))
            exclusive_union = row_lines[0] + row_lines[1] + row_lines[2]
            exclusive_union = [num for num in exclusive_union if not exclusive_union.count(num) >= 2]
            if exclusive_union:
                for number in exclusive_union:
                    for row in row_lines:
                        if number in row:
                            row_i = i*3+row_lines.index(row)
                            for sub_j in range(0, j*3):
                                if number in unconfirmed_numbers[row_i][sub_j] and len(unconfirmed_numbers[row_i][sub_j]) > 1:
                                    flag = True
                                    unconfirmed_numbers[row_i][sub_j].remove(number)
                            for sub_j in range(j*3+3, 9):
                                if number in unconfirmed_numbers[row_i][sub_j] and len(unconfirmed_numbers[row_i][sub_j]) > 1:
                                    flag = True
                                    unconfirmed_numbers[row_i][sub_j].remove(number)
            # 縦処理
            column_lines = []
            for sub_j in range(j*3, j*3+3):
                column_line = []
                for sub_i in range(i*3, i*3+3):
                    for num in unconfirmed_numbers[sub_i][sub_j]:
                        column_line.append(num)
                column_lines.append(list(set(column_line)))
            exclusive_union = column_lines[0] + column_lines[1] + column_lines[2]
            exclusive_union = [num for num in exclusive_union if not exclusive_union.count(num) >= 2]
            if exclusive_union:
                for number in exclusive_union:
                    for column in column_lines:
                        if number in column:
                            column_j = j*3+column_lines.index(column)
                            for sub_i in range(0, i*3):
                                if number in unconfirmed_numbers[sub_i][column_j] and len(unconfirmed_numbers[sub_i][column_j]) > 1:
                                    flag = True
                                    unconfirmed_numbers[sub_i][column_j].remove(number)
                            for sub_i in range(i*3+3, 9):
                                if number in unconfirmed_numbers[sub_i][column_j] and len(unconfirmed_numbers[sub_i][column_j]) > 1:
                                    flag = True
                                    unconfirmed_numbers[sub_i][column_j].remove(number)
    return flag

# 3次元配列同縦列から同じ数字を弾く
def column_excluder(i, j, n):
    flag = False
    for sub_i in range(9):
        if n in unconfirmed_numbers[sub_i][j]:
            unconfirmed_numbers[sub_i][j].remove(n)
            flag = True
    unconfirmed_numbers[i][j] = [n]

# 3次元配列同横列から同じ数字を弾く
def row_excluder(i, j, n):
    flag = False
    for sub_j in range(9):
        if n in unconfirmed_numbers[i][sub_j]:
            unconfirmed_numbers[i][sub_j].remove(n)
            flag = True
    unconfirmed_numbers[i][j] = [n]

# 3次元配列同枠内から同じ数字を弾く
def cubic_frame_excluder(i, j, n):
    flag = False
    for sub_i in range(i//3*3, i//3*3+3):
        for sub_j in range(j//3*3, j//3*3+3):
            if n in unconfirmed_numbers[sub_i][sub_j]:
                flag = True
                unconfirmed_numbers[sub_i][sub_j].remove(n)
    unconfirmed_numbers[i][j] = [n]

# 出力用
def nampre_print():
    print("____________________________________")
    for index, nums in enumerate(nampre):
        if index == 8:
            print("|", end="")
            [print(" "+str(n)+" |",end="") for i, n in enumerate(nums)]
            print("\n ￣￣￣￣￣￣￣￣￣￣￣￣￣￣￣￣￣￣")
        elif (index+1)%3 == 0:
            print("|", end="")
            [print(" "+str(n)+" |",end="") for i, n in enumerate(nums)]
            print("\n|===|===|===|===|===|===|===|===|===|")
        else:
            print("|", end="")
            [print(" "+str(n)+" |",end="") for i, n in enumerate(nums)]
            print("\n|---|---|---|---|---|---|---|---|---|")


# 時間を測る
start = time.time()

# 入力＆2次元配列に整形
nums = [int(i) for i in input().split()]
nampre = [[] for i in range(9)]
for index, i in enumerate(nums):
    nampre[index//9].append(i)

# 未確定数字参照用の3次元配列生成
unconfirmed_numbers = [[[n for n in range(1,10)] for j in range(9)] for i in range(9)]

# 処理継続フラグ
flag = False

# 数字確定済添字
subscripts = set()

# 3次元配列削る
for i in range(9):
    for j in range(9):
        if nampre[i][j] != 0 and (i,j) not in subscripts:
            flag = True
            num = nampre[i][j]
            column_excluder(i, j, num)
            row_excluder(i, j, num)
            cubic_frame_excluder(i, j, num)
            subscripts.add((i,j))

nampre_print()
while flag:
    flag = cubic_frame_judgment() or row_judgment() or column_judgment() or only_one_judgment() or cubic_tumor_excluder() or remainder_excluder() or line_confirm()
nampre_print()
print(str(time.time()-start)+"[sec]")

概要

• MySQL に文字セット utf8mb4 に設定したテーブルを作成して、Python 3 からデータ操作(参照・追加・更新・削除)を試す
• 環境: GMOデジロックのレンタルサーバー「コアサーバー」 + MySQL 5.7 + Python 3.6 + mysqlclient (MySQLdb)

MySQL のバージョンを確認

$ mysqld --version
mysqld  Ver 5.7.27 for Linux on x86_64 (MySQL Community Server (GPL))

MySQL の文字セットと照合順序を確認

起動している MySQL デーモンの文字セットと照合順序の設定を MySQL monitor にて確認。

mysql> SHOW VARIABLES LIKE 'character_set%';
+--------------------------+----------------------------+
| Variable_name            | Value                      |
+--------------------------+----------------------------+
| character_set_client     | utf8                       |
| character_set_connection | utf8                       |
| character_set_database   | utf8                       |
| character_set_filesystem | binary                     |
| character_set_results    | utf8                       |
| character_set_server     | utf8                       |
| character_set_system     | utf8                       |
| character_sets_dir       | /usr/share/mysql/charsets/ |
+--------------------------+----------------------------+
8 rows in set (0.00 sec)

mysql> SHOW VARIABLES LIKE 'collation%';
+----------------------+-----------------+
| Variable_name        | Value           |
+----------------------+-----------------+
| collation_connection | utf8_general_ci |
| collation_database   | utf8_general_ci |
| collation_server     | utf8_general_ci |
+----------------------+-----------------+
3 rows in set (0.00 sec)

参考:

• MySQL :: MySQL 5.6 リファレンスマニュアル :: 5.1.4 サーバーシステム変数
• MySQL :: MySQL 5.6 リファレンスマニュアル :: 10.1.4 接続文字セットおよび照合順序
• MySQL :: MySQL 5.6 リファレンスマニュアル :: 10.1.14.1 Unicode 文字セット

データベースを作成

MySQL monitor にてデータベースを作成。

mysql> create database test_db;
Query OK, 1 row affected (0.00 sec)

mysql> show databases;
+--------------------+
| Database           |
+--------------------+
| information_schema |
| test_db            |
+--------------------+
2 rows in set (0.01 sec)

mysql> use test_db;
Database changed

参考:

• MySQL :: MySQL 5.6 リファレンスマニュアル :: 13.1.10 CREATE DATABASE 構文

テーブルを作成

テーブル作成用の SQL を書いてファイルに保存。

create-table.sql

CREATE TABLE IF NOT EXISTS test_table (
  id INTEGER NOT NULL AUTO_INCREMENT,
  name VARCHAR(256) NOT NULL,
  PRIMARY KEY (id)
)
ENGINE=InnoDB
CHARACTER SET utf8mb4
COLLATE utf8mb4_bin;

mysql コマンドでテーブルを作成。

$ mysql -h localhost -u alice -p test_db < create-table.sql

MySQL monitor にてテーブルを確認。

mysql> show tables from test_db;
+-------------------+
| Tables_in_test_db |
+-------------------+
| test_table        |
+-------------------+
1 row in set (0.00 sec)

mysql> show create table test_db.test_table;
+------------+-----------------------------------------------
| Table      | Create Table                                  
+------------+-----------------------------------------------
| test_table | CREATE TABLE `test_table` (
  `id` int(11) NOT NULL AUTO_INCREMENT,
  `name` varchar(256) COLLATE utf8mb4_bin NOT NULL,
  PRIMARY KEY (`id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin |
+------------+-----------------------------------------------
1 row in set (0.00 sec)

参考:

• MySQL :: MySQL 5.6 リファレンスマニュアル :: 13.1.17 CREATE TABLE 構文
• MySQL :: MySQL 5.6 リファレンスマニュアル :: 3.5 バッチモードでの MySQL の使用

Python のバージョンを確認

$ python3 --version
Python 3.6.8

Python のパッケージ mysqlclient をインストール

mysqlclient は Python 2 で使えるパッケージ MySQL-python (MySQLdb1) から派生して Python 3 に対応したパッケージ。

今回は mysqlclient のバージョン 1.3.4 を使う。

$ python3 -m pip install mysqlclient==1.3.4

今回の動作確認環境であるGMOデジロックのレンタルサーバー「コアサーバー」では該当バージョンがすでにインストール済みだった。

$ python3 -m pip list | grep mysqlclient
mysqlclient            1.3.4  

参考:

• Python モジュールのインストール — Python 3.6.10rc1 ドキュメント
• GitHub - PyMySQL/mysqlclient-python: MySQL database connector for Python (with Python 3 support)
• GitHub - farcepest/MySQLdb1: MySQL database connector for Python (legacy version)

Python + mysqlclient でデータ操作

データを追加・取得・更新・削除する Python のソースコードを用意。

crud.py

# mysqlclient を使う
import MySQLdb

# connect db
connection = MySQLdb.connect(host="localhost",
                             user="alice",
                             passwd="your-password",
                             db="test_db",
                             charset="utf8mb4")
cursor = connection.cursor(MySQLdb.cursors.DictCursor)

# insert
insert_sql = "INSERT INTO test_table (name) VALUES (%s)"
cursor.execute(insert_sql, ("寿司ビール??????文字化けしないで",))
print(f"insert count={cursor.rowcount}")
connection.commit()

# select
select_sql = "SELECT id, name FROM test_table"
cursor.execute(select_sql)
print(f"select count={cursor.rowcount}")
for row in cursor:
  print(f"{row['id']}: {row['name']}")
  target_id = row["id"]

# update
update_sql = "UPDATE test_table set name=%s WHERE id=%s"
cursor.execute(update_sql, ("Alice", target_id))
print(f"update count={cursor.rowcount}")
connection.commit()

# delete
delete_sql = "DELETE FROM test_table WHERE id=%s"
cursor.execute(delete_sql, (target_id,))
print(f"delete count={cursor.rowcount}")
connection.commit()

connection.close()

実行結果。
寿司ビール絵文字などが文字化けせずに追加・取得・表示できている。

$ python3 crud.py 
insert count=1
select count=1
1: 寿司ビール??????文字化けしないで
update count=1
delete count=1

参考:

• Welcome to MySQLdb’s documentation! — MySQLdb 1.2.4b4 documentation
• 【Python3】MySQL 操作をひと通りマスター！導入方法とCRUDサンプルコード集 | ITエンジニアラボ
• Python3でMySQLを使う – 基本操作からエラー処理までサンプルコード付 | Crane & to.

テーブルとデータベースを削除

MySQL monitor から使い終わったテーブルとデータベースを削除。

mysql> drop table test_db.test_table;
Query OK, 0 rows affected (0.00 sec)

mysql> drop database test_db;
Query OK, 0 rows affected (0.00 sec)

　BlenderにはPythonが内蔵されていて、色々なスクリプトを行うことができます。ただし、導入されているモジュールはnumpy等に限られており、好きなモジュールを追加するには少し作業が必要なので解説します。

Windowsの場合

　デフォルトの設定の場合、Blenderは以下の場所に保存されているはずです。

C:\Program Files\Blender Foundation\Blender\2.80

　この下のPythonが内蔵されています。（自分で個別にインストールしたPythonとはまた別です）

C:\Program Files\Blender Foundation\Blender\2.80\Python\python.exe

　また、以下の場所にpip.exe（インストールするためのモジュール）があります。

C:\Program Files\Blender Foundation\Blender\2.80\python\Scripts

　！バージョンは2.81だったり2.80だったりするので、そこは各自合わせてください

　このpipにはパスが通っていない（このpipを参照するようになっていない）ので、コマンドプロンプトで例えば、

pip install pandas

　と打っても通りません。（別のPythonがインストールされている場合、そちらの方にインストールされます）

　パスが通っていないなら場所を直接参照すればよいだけの話で、

C:\Program Files\Blender Foundation\Blender\2.80\python\Scripts\pip install pandas

　とやればOK……とはならず、実はもう一手間必要で、これだとWindowsの管理者権限の関係でうまくいきません。

C:\Program Files\Blender Foundation\Blender\2.80\python\Scripts\pip install pandas --user

　これでうまくいきます。

　注意事項として、ワーキングディレクトリは

>>> import os
>>> os.getcwd()
'C:\\Program Files\\Blender Foundation\\Blender'

　になるので、なにかファイルを生成してそれを参照する、という操作を行う場合は管理者権限がないとエラーになります。管理者権限でBlenderを起動するか、ユーザーディレクトリ以下にファイルを生成・参照するようにしましょう。

Macの場合

　Macの場合、Blenderは以下の場所に保存されているはずです。

/Applications/Blender.app/Contents/Resources/2.81

　！バージョンは2.81だったり2.80だったりするので、そこは各自合わせてください

　ただ、なぜかpipがインストールされていません。pipをインストールする必要があります。

　ターミナルを開いて、

curl https://bootstrap.pypa.io/get-pip.py -o get-pip.py
/Applications/blender.app/Contents/Resources/2.81/python/bin/python3.7m get-pip.py

　！pythonのバージョンは3.7だったり3.5だったりするので、そこは各自合わせてください

　で、pipが生成されるはずです。

　後は同じように、

/Applications/blender.app/Contents/Resources/2.81/python/bin/pip install pandas

　でインストール完了。

_tkinterの罠

　ちなみに、Windowsでpyplotをインポートしようとすると以下のようなエラーが出ます。
　

　_tkinterがないと言われます。直前にインポートしているtkinter/__init__.pyの中身を見ると、

「もしこれがfailする場合、あなたのPythonはTk用に調整されていないかもしれません」

　個別にインストールする普通の（？）Pythonだとこれが出ないので、内蔵Python特有の問題かもしれません。

　一応、力技としてC:\Program Files\Blender Foundation\Blender\2.80\の下に別にインストールしたpythonをフォルダごとコピーするというものがあります（元々存在していたpythonフォルダはリネームして避難）。これだと環境をそのまま持っていけます。ただ、pyplot.show()でグラフを描画することはできませんでした。この方法は、内蔵Pythonではどうしてもモジュールをpipできない場合などは有効かもしれません。

　MacOSの場合、このエラーは出ませんが、やはりpyplot.show()で描画することはできませんでした。

　Blender上でグラフを描画するのは難しそうです（その必要性はあまりなさそうですが）。

要約

SAMの内部にあるpipのversionとローカルのpipのversionが合わないと動かない時があるので、ローカルのpipのversionを修正する必要がある。

実行環境

項目名	バージョン
OS	MacOS High Sierra 10.13.6
VSCode	1.40.2
SAM	0.37.0
Python	3.7.3
pip	19.3.1(記事内で19.2.3に変更)

問題

VSCodeの「Run Locally」からAWS Toolkitを使用してローカル環境で試験したい時に動かない場合があります。例えば以下のようなエラーです。

Local invoke of SAM Application has ended.
Preparing to run app.lambda_handler locally...
Building SAM Application...
An error occurred trying to run SAM Application locally: Error with child process: Building resource 'awsToolkitSamLocalResource'
,Running PythonPipBuilder:ResolveDependencies
,Error: PythonPipBuilder:ResolveDependencies - Traceback (most recent call last):
  File "<string>", line 1, in <module>
TypeError: 'module' object is not callable

原因(おそらく)

SAMの方で起きていたIssueがVSCodeのAWS Toolkitではまだ解決していないみたいです。
GitHub : sam build fails for python3.7 functions with pip==19.3 installed
試しに同じ設定で、ターミナルから以下のコマンドで実行した場合はSAMとpipのverisonは同じでも問題ありませんでした。

$ sam local invoke HelloWorldFunction --event events/event.json

対処

一時的な対処とはなりますが、pipのversionを以下に修正するとVSCodeからでも正常に実行することが可能になります

$ pip install pip==19.2.3

pipのversion変更後にVSCodeから「Run Locally」したログ

Fetching lambci/lambda:python3.7 Docker container image......
Mounting /tmp/aws-toolkit-vscode/vsctkytL1AY/output/awsToolkitSamLocalResource as /var/task:ro,delegated inside runtime container

...(中略)...

{"statusCode":200,"body":"{\"message\": \"hello worldXXX\"}"}

plt.showで複数の画像を表示できる。

グレースケール画像を表示するときは
plt.imshow(thresh, 'gray', vmin = 0, vmax = 255)
http://hikuichi.hatenablog.com/entry/2015/12/26/225623

はじめまして、来年はHoudiniの記事を書きたいなと思ってます、rateionn(読み方は低音)です。

この記事はMaya Advent Calendar 2019の15日目の記事になります。

先日のzebraed氏のpyside:MEventMessageを使ったcallbackはご覧になられましたでしょうか？

OpenMayaでscriptJobと同じ機能を軽くシンプルに扱えるとの事で、自分も今後活用していこうと思います。

今回はscriptJobを使って書きますが...

---

はじめに

ノード名が重複しているからと言ってMayaに致命的なエラーが発生することはありません。
しかし、melスクリプトを作成する際に正しい作法で処理を書かないと指定したオブジェクトが見つけられずにスクリプトが停止してしまう可能性があります。

• Mayaエラー：行0：複数のオブジェクトが名前と一致します

• Melの「More than one object matches name」の回避

• ノード名の重複があっても正しく動作するツールであれば問題ないし、最終的にデータを綺麗に出来るならこの記事は読まなくても大丈夫だと思います。

何か起こる前に自動でリネーム出来てたら楽だよね

自動リネーム処理

1. リネームする

1. uuidからノードを取得する。
   • 【Maya】オブジェクト名によらないオブジェクト選択 _ UUIDを使う
2. ノード名に|(バーティカルライン)が含まれていた場合、ノード名が重複している。
3. 正規表現でベースとなるノード名を取得する。
   • 赤くマーキングされている範囲をベース名として取得する。
     
4. リネームする
   • cmds.rename
     • 新しい名前の最後に「#」を 1 つ付けると、名前の変更コマンドは、新しい名前を固有にする番号で最後の「#」を置き換えます。

出来たコードはこんな感じ

def rerename(node_uuid):
    for node in cmds.ls(node_uuid):
        if "|" in node:
            basename = re.search("^(.+)?\|(.+?)(\d+)?$", node).group(2)
            rename_name = cmds.rename(node, "{}#".format(basename))

nonSameNameAlliance.py#L5-L9

2. 手動リネーム時の処理

• NameChangedが実行されたときにリネーム処理を実行します。
  • 手動でリネームをする場合はオブジェクトを1つしか選択していないと思うのでlsコマンドで選択しているオブジェクトを取得してUUIDに変換してリネーム処理を実行します。

def renameEvent():
    for node_uuid in cmds.ls(cmds.ls(sl=True), uuid=True):
        rerename(node_uuid)

nonSameNameAlliance.py#L11-L13

3. 複製時の処理

• 手動で複製を行ったときは、複製されたオブジェクトに選択が移動するのでSelectionChangedが実行されたときにリネーム処理を実行します。
  • dag=Trueで階層を取得して、tr=Trueでトランスフォームノードのみ取得できるようにし、UUIDに変換してリネーム処理を実行します。

def duplicateEvent():
    for node_uuid in cmds.ls(cmds.ls(sl=True, dag=True, tr=True), uuid=True):
        rerename(node_uuid)

nonSameNameAlliance.py#L15-L17

スクリプトジョブに登録

def main():
    try:
        if jobIds is not None:
            for jobId in jobIds:
                cmds.scriptJob(kill=jobId, force=True)
    except:pass

    jobIds = []
    jobIds.append(cmds.scriptJob(event=["NameChanged", renameEvent], protected=True))
    jobIds.append(cmds.scriptJob(event=["SelectionChanged", duplicateEvent], protected=True))

• スクリプトジョブに処理を登録すると、条件に合ったタイミングで登録された処理を実行してくれます。
• 今回はrenameEventとSelectionChangedそれぞれに処理を登録しています。

nonSameNameAlliance.py#L19-L28

Maya起動時に自動でスクリプトジョブを実行する

userSetup.py

import nonSameNameAlliance;nonSameNameAlliance.main()

nonSameNameAlliance.pyをMayaのScriptsフォルダ内に配置してuserSetup.pyに起動用の処理を追記します。
これでMaya起動時にscriptJobに処理が登録されて自動リネームされるようになります。

ダウンロード

今回のソースコードはこちら
https://github.com/teionn/nonSameNameAlliance

注意点

1. シーン内で操作をした時に自動的にノードがリネームされてしまうので、基本的にモデラー向けの内容かと思います。

2. トランスフォームノードのみに実行されるようになっています。

   • 中間シェイプノードなどの全てのノードで効果を得たい場合はtr=Trueを消してください。

3. このスクリプトは大文字/小文字の判定は行いません。
   Mayaは大文字/小文字違いのオブジェクトを同じ階層であっても保持することが出来てしまいます。
   

この状態のままUnityにアセットを持っていくと、オブジェクトobject_aとobject_Aがあった場合どちらかのオブジェクトが参照出来なくなってしまうらしいです。

---

明日の記事はrateionnさんによるちゃんと Remove: Unknown Plugins してますか?というお話です。

あれ… 書くの自分やん… まだ記事できてないんですけど…

間に合ったらスクリプトを無償公開する予定なのでじゃんじゃんDLしてただけると嬉しいです。

MDC Advent Calendar 2019 の15日目です。

Kafkaとは何か

LinkedInが作ったOSSの分散メッセージングシステム（メッセージングキュー）。
高スループット、高スケーラビリティ。
Java(Scala)で書かれている。

Producer、Broker、Consumerの3つのコンポーネントで構成される。
ProducerからストリーミングされたデータをConsumerへ中継する。対障害のためにデータの永続化もする。送達保証も実現。
　　・ Producer: メッセージを配信する
　　・ Broker: ProducerからConsumerへメッセージの受け渡しをする
　　・ Consumer: メッセージを受け取る

※仔細な仕組みやアーキテクチャ等については以下が参考になります。
　・Apache Kafkaの概要とアーキテクチャ
　・Apache Kafkaに入門した

何に使うのか

ユースケースとしては以下のようなものが挙げられます。

1. システムがサイロ化するのを防ぐためにデータハブとしてアーキテクチャに組み込む（マイクロサービスとかで）
2. Fluentdなどと連携してログ収集に使う
3. Webサイトのユーザのページ移動とかを収集してWebアクティビティ分析に使う
4. IoTデバイスのセンサーの値を集約し、可視化や分析、他のデバイスの制御などに使う
5. ビッグデータ、機械学習、etc

具体的なところだと以下。
・LINEの大規模データパイプライン
・Yahooリアルタイム検索
・大手ヘルスケアIT企業 Cerner社のKafka活用事例

動かしてみる

簡単なサンプルを作って試してみます。
今回はKafka本体はKafka-dockerを使って環境構築をして、ProducerとConsumerのクライアント側はkafka-pythonを使いました。

Kafka-dockerのインストール・起動

公式にあるとおり、kafka-dockerをダウンロードして、docker-compose.ymlのKAFKA_ADVERTISED_HOST_NAMEにdocker hostのIPアドレスを書いたあと、

docker-compose up -d

すればOKです。

参考： kafka in dockerのチュートリアル

Producerを実装する

ほんとうはTwitter Striming APIみたいなデータをとってきたり、IoTのセンサーの値を取得したかったですが、今回は用意がないので一旦適当に数値を取れるものを、ということでマウスのx座標を取得して1秒おきにKafkaに送付するスクリプトを書きました。

KafkaProducerの引数bootstrap_serverに渡す値は、docker-compose.ymlにも書いたdocker hostのIPアドレスと、kafkaのコンテナに割り当てられたPort番号を指定します。
割り当てられてるPort番号はdocker psで確認できます。

上の場合は32783がそれです。
以下が作成したProducer側のソースコードです。
procuer.send(でtestというTopicに現在のマウスのx座標を投げています。

from kafka import KafkaProducer
import pyautogui
import time

def main():
    producer = KafkaProducer(bootstrap_servers='{Docker HostのIPアドレス}:{Port}')
    while True:
        result = producer.send('test', str(pyautogui.position().x).encode()).get(timeout=60)
        print(result)
        time.sleep(1)

if __name__ == '__main__':
    main()

Consumerを実装する

次にConsumer側の実装です。同様にKafkaのIPアドレスとPortを指定します。
for message in consumer:でKafkaからデータを逐次pullしてきてくれます。

from kafka import KafkaConsumer

def main():
    consumer = KafkaConsumer(
            'test', 
            bootstrap_servers=['{Docker HostのIPアドレス}:{Port}'])

    for message in consumer:
        print("x = " + message.value.decode())

if __name__ == '__main__':
    main()

実行

下の左側でProducer側、右側でConsumerを実行しています。
Producer側からKafkaに送った値(マウスのx座標)をConsumer側でKafkaから取得して表示できているのが確認できます。

今後やりたいこと

・ラズパイ、Arduinoとか使ってセンサーの値を組み込む。
・取得したデータをグラフにしたり、解析したりする。

はじめに

この記事はAWS初心者 Advent Calendar 2019
の14日目の記事です。
もし誤りがあれば指摘してもらえると幸いです。

要約

Boto3の中ではサービスによってリソースAPI(boto3.reosurce('サービス名')で呼び出すもの)があります。リソースAPIの方がクライアントAPIより抽象化されていて不要な情報を書かずに実装できるので、それぞれのAPIで同じことができる場合はリソースAPIを優先して使用した方が良いです。

記事全体の構成

まずBoto3、クライアントAPI、リソースAPIとは何だったのかを再確認します。
その後に両方のAPIを使用した場合にSQS、S3、DynamoDBの使用パターンを１ケースずつとりあげて比較してどう違うのかを具体的に見てみます。
最後に現在リソースAPIが提供されているサービスの一覧と所感を書いてます。

Boto3とは

Pythonに提供されているAWSのSDKで、コードからAWSの各種サービス（EC2やDynamoDB、S3など）に接続する場合に使用します。
APIのリファレンスは以下の資料に書かれています。
Boto 3 Documentation

APIには主に低レベルなAPIであるクライアントAPIと高レベルなAPIであるリソースAPIがあります。具体的には以下のような記述で呼び出されます。

# Client API
boto3.client('sqs')

# Resource API
boto3.resource('sqs')

次にそれぞれのAPIにどんな特性があるか確認します。

クライアントAPI（低レベルAPI）

AWSのサービスで提供しているHTTP APIと１対１に対応するメソッドです。HTTP API
と完全にマッピングしているので、APIで可能な操作はすべてできるようになっています。ただ汎用的な設定になっているので、APIに設定するパラメータをメソッドに直接する必要があります。
以下はSQSのsend_messageを実行する例です。APIで使用するQueueUrlなどを指定しています。

import boto3

# SQS クライアントAPI版
sqs = boto3.client('sqs')

response = sqs.send_message(QueueUrl='https://sqs.us-east-2.amazonaws.com/123456789012/MyQueue', MessageBody='...')

参考元：Boto3のClient APIドキュメント(send_message)
https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/sqs.html#SQS.Client.send_message
※「QueueUrl='...'」の部分はURLの記述を追加

リソースAPI（高レベルAPI）

クライアントAPIと比べて高レベルで抽象化したAPIです。APIとの直接マッピングではなく、抽象化したクラスを間に挟むことでより開発者にとって扱いやすくなっています。
ここでは例として、クライアントAPIと同じようにリソースAPIでsend_messageを行う例を記載します。

queueName='target-queue-name'

# SQS リソースAPI版
sqs = boto3.resource('sqs')

response = sqs.get_queue_by_name
                QueueName = queueName
            ).send_message(
                MessageBody = '...'
            )

参考元：「Boto3(Python)で"Service Resource"を使ってみた(Lambda)」
https://cloudpack.media/16114

クライアントAPIと比較するとコード内にSQSのURLを記載せず、queue名の問い合わせからメッセージの送信が書けます。いちいちキューのURLを意識する必要がなくなるので開発者はキューの名前さえ記述すればよくなります。

クライアントAPIとリソースAPIの比較（S3）

次にS3のクライアントAPIとリソースAPIを比較します。
例としてバケットからファイル名を取得するコードを記述します。

# 共通定数(バケットとS3内ファイルのプレフィックス)
BUCKET_NAME= "xxx-bucket"
S3_PREFIX = "image-file-done/"

# S3 クライアントAPI版
s3_client = boto3.client('s3')
# 戻り値の型：<class 'dict'>
s3_objects = s3_client.list_objects_v2(Bucket=BUCKET_NAME, Prefix=S3_PREFIX)

for filename in s3_objects['Contents']:
    print('client:'+filename['Key'])

# S3 リソースAPI版
s3_resource = boto3.resource('s3')
# 戻り値の型：<class 'boto3.resources.collection.s3.Bucket.objectsCollection'>
s3_objects = s3_resource.Bucket(BUCKET_NAME).objects.filter(Prefix=S3_PREFIX)

for filename in s3_objects:
    print('resource:'+filename.key)

呼び出しまではほとんど変わりませんが、呼び出し後の型が異なります。
クライアントAPIで呼び出した場合は、辞書型で返ってくるので実装のたびに返却される辞書型の形式を意識したデータの取り出しが必要になります。リソースAPIであればboto3用のオブジェクトになっているので、他のサービスでリソースAPIを使用する場合でも似たような記述でコードが書けます。

クライアントAPIとリソースAPIの比較（DynamoDB）

最後にDynamoDBのクライアントAPIとリソースAPIを比較します。
こちらは特にBoto3のドキュメントがよく分からなくて、他の記述を参考にして間違ってコピーしてしまうことが多そうです。（昔の自分のことです…）

# 共通定数(テーブル名とハッシュキー名、ソートキー名)
TABLE_NAME='XXXXX_IFO'
HASH_KEY_NAME='XXXXX_CODE'
SORT_KEY_NAME='DATE_TIME'

# DynamoDB クライアント版
dynamodb_client = boto3.client('dynamodb')
# 戻り値の型: <class 'dict'>
response = dynamodb_client.get_item(
    TableName=TABLE_NAME,
    Key={
        HASH_KEY_NAME:{
            'S': '54620100'  
        },
        SORT_KEY_NAME:{
            'S': '2019050621'
        }
    }
)

# DynamoDB リソース API
dynamodb_resource = boto3.resource('dynamodb')
table = dynamodb_resource.Table(TABLE_NAME)
# 戻り値の型: <class 'dict'>
response = table.get_item(
    Key={
        HASH_KEY_NAME: '54620100',
        SORT_KEY_NAME: '2019050621'
    }
)

久々にクライアントAPIで書きましたがパラメータをCLIと同じ形式で書くので少ししんどいです。戻り値の型自体はどちらも辞書型ですが、テーブル内部の操作をTableオブジェクトからでき、Key項目の記述は型情報を逐一書かなくて済むのでかなり楽になってます。

リソースAPIが提供されているサービス(2019/12/14時点)

リソースAPIは提供されているサービスが限られています。具体的には以下のサービスが対応しています。
- CloudFormation
- Cloud Watch
- DynamoDB
- EC2
- Glacier
- IAM
- OpsWorks
- S3
- SNS
- SQS

参考元：「Boto 3 Documentation」
https://boto3.amazonaws.com/v1/documentation/api/latest/index.html

上記サービスを利用する場合は、クライアントAPIの利用よりも先にまずリソースAPIの検討を行う必要がありそうです。Boto3のCodeExamples(サンプルドキュメント)でDynamoDBはresourceになっていますが、他のEC2やCloudWatchはclientなのでCodeExamplesだけでなくAvailable Servicesを見る必要がありそうです。

所感

今までクライアントAPIやリソースAPIなどをあまり意識せずに使用していたため反省もかねて色々調べた結果をまとめました。SQSの例でもわかるようにコード部分に不要な情報を書く必要がなくなるので、今後自分もまずはリソースAPIの提供があるサービスか確認し、提供されているならうまく使用できないか考えてからクライアントAPIを使いたいと思います。
QiitaやネットにはクライアントAPIで書かれた実装が多々あるようなので、今後自分が記事を書くことでリソースAPIの情報を増やしたいと思います。

はじめに

単純に上記エラーで悩んでいる人向けの記事です。かなりレアケースです。

問題点

import hashlib
すると死ぬ時があります。

結果

Python 2.7.16 (default, Mar  4 2019, 09:01:38) 
[GCC 4.2.1 Compatible Apple LLVM 10.0.0 (clang-1000.11.45.5)] on darwin
Type "help", "copyright", "credits" or "license" for more information.
>>> import hashlib
ERROR:root:code for hash md5 was not found.
Traceback (most recent call last):
  File "/usr/local/Cellar/python@2/2.7.16/Frameworks/Python.framework/Versions/2.7/lib/python2.7/hashlib.py", line 147, in <module>
    globals()[__func_name] = __get_hash(__func_name)
  File "/usr/local/Cellar/python@2/2.7.16/Frameworks/Python.framework/Versions/2.7/lib/python2.7/hashlib.py", line 97, in __get_builtin_constructor
    raise ValueError('unsupported hash type ' + name)
ValueError: unsupported hash type md5
ERROR:root:code for hash sha1 was not found.
Traceback (most recent call last):
  File "/usr/local/Cellar/python@2/2.7.16/Frameworks/Python.framework/Versions/2.7/lib/python2.7/hashlib.py", line 147, in <module>
    globals()[__func_name] = __get_hash(__func_name)
  File "/usr/local/Cellar/python@2/2.7.16/Frameworks/Python.framework/Versions/2.7/lib/python2.7/hashlib.py", line 97, in __get_builtin_constructor
    raise ValueError('unsupported hash type ' + name)
ValueError: unsupported hash type sha1
ERROR:root:code for hash sha224 was not found.
Traceback (most recent call last):
  File "/usr/local/Cellar/python@2/2.7.16/Frameworks/Python.framework/Versions/2.7/lib/python2.7/hashlib.py", line 147, in <module>
    globals()[__func_name] = __get_hash(__func_name)
  File "/usr/local/Cellar/python@2/2.7.16/Frameworks/Python.framework/Versions/2.7/lib/python2.7/hashlib.py", line 97, in __get_builtin_constructor
    raise ValueError('unsupported hash type ' + name)
ValueError: unsupported hash type sha224
ERROR:root:code for hash sha256 was not found.
Traceback (most recent call last):
  File "/usr/local/Cellar/python@2/2.7.16/Frameworks/Python.framework/Versions/2.7/lib/python2.7/hashlib.py", line 147, in <module>
    globals()[__func_name] = __get_hash(__func_name)
  File "/usr/local/Cellar/python@2/2.7.16/Frameworks/Python.framework/Versions/2.7/lib/python2.7/hashlib.py", line 97, in __get_builtin_constructor
    raise ValueError('unsupported hash type ' + name)
ValueError: unsupported hash type sha256
ERROR:root:code for hash sha384 was not found.
Traceback (most recent call last):
  File "/usr/local/Cellar/python@2/2.7.16/Frameworks/Python.framework/Versions/2.7/lib/python2.7/hashlib.py", line 147, in <module>
    globals()[__func_name] = __get_hash(__func_name)
  File "/usr/local/Cellar/python@2/2.7.16/Frameworks/Python.framework/Versions/2.7/lib/python2.7/hashlib.py", line 97, in __get_builtin_constructor
    raise ValueError('unsupported hash type ' + name)
ValueError: unsupported hash type sha384
ERROR:root:code for hash sha512 was not found.
Traceback (most recent call last):
  File "/usr/local/Cellar/python@2/2.7.16/Frameworks/Python.framework/Versions/2.7/lib/python2.7/hashlib.py", line 147, in <module>
    globals()[__func_name] = __get_hash(__func_name)
  File "/usr/local/Cellar/python@2/2.7.16/Frameworks/Python.framework/Versions/2.7/lib/python2.7/hashlib.py", line 97, in __get_builtin_constructor
    raise ValueError('unsupported hash type ' + name)
ValueError: unsupported hash type sha512

これの原因はPyenvとかhomebrewとか色々あります。
まずMacOSのPythonはpythonを読みに行くんじゃなく、python2を読みに行く点に注意です。
Pyenvで指定しているもんだからwhich pythonとかやっても本物は出てきません。
python2でエイリアス設定されているのか、基本python2が呼ばれます。
which python2ですれば正体がわかります。
そもそもなんでPython3があるのにpython2がpyenvにねーんだ！って話なんですけどね。

まぁこれらは間接的な原因でしかなくて、直接的なぶっ壊れ原因は全くわかりません。

解決法

brewでアップグレードしちゃう

これに限ります。

brew upgrade python@2

これで環境がとりあえずはなんとかなります。

まだまだ解決中なので、続報があり次第お伝えします。

概要

この記事はフューチャー Advent Calendar 2019 14日目の記事です。
昨日の記事は@RuyPKGさんによる新人でも、楽がしたい！ ~議事録の準備~となってます。

先日社内勉強会にて、@shibukawaさんがゲームエンジンの紹介をしていたのがこの記事を書いたきっかけになります。
Pythonで実装したボードゲームのUIが欲しく、HTMLによる表現だと動きを表現するのが難しかったのでPythonのゲームエンジンであるpygameを触ってみることにしました。

pygameとは

Pythonで動かすことの出来る、クロスプラットフォームのゲームエンジンです。
ゲームのグラフィック部分をPythonで書くことができるならば、Pythonで書かれたゲームロジックをそのまま埋め込むことが出来るのでとても便利です。

このpygameですが、2009年に一旦開発が停止してしばらく動いていなかったのでpygameをご存知の方は乗り換えした人も多いのではないでしょうか。
しかし、pygame2リリースに向けた開発が2019年になって活発になり、来年中にpygame2の本バージョンがリリースされそうです。
2.0.0.dev6からはpython3.8もサポートされていて先が楽しみです。
今後注目を浴びてくる気がします。

環境

• pygame: 2.0.0.dev6
• python: 3.8.0

環境構築

pygame公式のGetting Startedを参考に環境構築を進めます。

pip経由で、pygameをインストールします。
この際、バージョンを指定しないと最新安定版である1.9.6が入ってしまうので注意しましょう。(1.9.6はpython3.6までしか対応していません)

pip install pygame==2.0.0.dev6

サンプルを動かしてみる

多数のサンプルゲームがpygameには含まれています。
試しに一つ実行してみましょう。

python -m pygame.examples.aliens

インベーダーゲームのようなものが実行されました！
サンプルの一覧とそれぞれどんなゲームかはgitのexampleフォルダのREADMEにあるので色々実行してみると面白いです。

1からサンプルを作る

サンプルを解読しながら、自分でコードを描き上げていきます。
今回は、背景に文字を書くだけのサンプルを目指します。

処理の流れ

静的な画面とは違い、ゲームエンジンを用いて描画する場合は高速で画面を更新する必要があります。
そのため、以下のような流れで処理を行います。

1. 画像読み込み等の初期化
2. 画面をクリアし、背景画像を描画
3. 画面に表示するオブジェクトの位置や値を更新して描画
4. 画面の更新の反映
5. 2に戻る

ゲームエンジンでは、画面に表示するオブジェクトはスプライトを使って描画されているので、スプライトクラスを継承したクラスを使って実装時は描画します。
スプライトが何か気になった方は、調べてみてください。
描画を高速化するためのハードウェア実装の話等、歴史的経緯があって面白いですが知らなくても特に問題はないと思います。

import os
import pygame as pg

# game constants
SCREENRECT = pg.Rect(0, 0, 640, 480)
SCORE = 0

main_dir = os.path.split(os.path.abspath(__file__))[0]


def load_image(file):
    """ loads an image, prepares it for play
    """
    file = os.path.join(main_dir, "data", file)
    try:
        surface = pg.image.load(file)
    except pg.error:
        raise SystemExit('Could not load image "%s" %s' % (file, pg.get_error()))
    return surface.convert()

class Score(pg.sprite.Sprite):
    """ to keep track of the score.
    """

    def __init__(self):
        pg.sprite.Sprite.__init__(self)
        self.font = pg.font.Font(None, 40)
        self.font.set_italic(1)
        self.color = pg.Color("white")
        self.lastscore = -1
        self.update()
        self.rect = self.image.get_rect().move(10, 450)

    def update(self):
        """ We only update the score in update() when it has changed.
        """
        if SCORE != self.lastscore:
            self.lastscore = SCORE
            msg = "Score: %d" % SCORE
            self.image = self.font.render(msg, 0, self.color)

def main(winstyle=0):
    pg.init()
    # Set the display mode
    winstyle = 0  # |FULLSCREEN
    bestdepth = pg.display.mode_ok(SCREENRECT.size, winstyle, 32)
    screen = pg.display.set_mode(SCREENRECT.size, winstyle, bestdepth)

    # create the background, tile the bgd image
    bgdtile = load_image("background.jpg")
    background = pg.Surface(SCREENRECT.size)
    background.blit(bgdtile, (0, 0))
    screen.blit(bgdtile, (0, 0))
    pg.display.flip()

    # Initialize Game Groups
    all = pg.sprite.RenderUpdates()

    # Create Some Starting Values
    clock = pg.time.Clock()
    global SCORE
    if pg.font:
        all.add(Score())

    # Run our main loop whilst the player is alive.
    while True:
        all.clear(screen, background)
        SCORE += 123456789
        all.update()

        # draw the scene
        dirty = all.draw(screen)
        pg.display.update(dirty)

        # cap the framerate at 40fps. Also called 40HZ or 40 times per second.
        clock.tick(40)


if __name__ == "__main__":
    main()

data以下に背景画像(background.jpg)を準備しておく必要があります。
ちょっと長いですが、上記サンプルを実行すると、以下のように背景の描画と文字の表示ができます。

以下ゲームエンジン独特の部分を説明します。

Spriteのグループ化

all = pg.sprite.RenderUpdates() # Spriteグループの作成
all.add(Score()) # Spriteグループへの追加
all.update() # Spriteグループの一括更新
all.draw(screen) # Spriteグループの描画

pygameにはスプライトをグループ化できる機能が備わっていて、画面の全ての要素や、特定のグループ（例えば敵キャラクターのみ等）の要素を一斉に更新したり描画するのが簡単にできるようなっています。
Spriteのグループには順序付きグループや単体スプライト用のグループや様々な種類があり、用途に応じて使い分けると良いでしょう。
公式ドキュメントで一覧は確認できます。

画面のリフレッシュレートの指定

clock = pg.time.Clock() # Clockの生成
clock.tick(40) # Clockを用いて1/40秒経過するまで処理を待つ

オブジェクトを動かす際には、処理の重さに関わらず同じ速さで動かすために処理が早く終わりすぎた場合は待つ必要があります。
これを実現するため、pygameではClockオブジェクトを使用します。
Clockオブジェクトを作ってclock.tick()を呼ぶだけで、前回の呼び出しからフレームレートに応じた時間sleepしてくれるのでとても便利です。

最後に

今回紹介した描画の機能以外にも、pygameには

• 入力の受けつけ
• 音の再生
• Android, iOS対応

等様々な機能があります。
リアルタイムに更新されるUIをPythonで書けるのは非常に便利なので、pygame2の正式リリースが待ちきれないですね！

LINEWORKS Advent Calendar 14日目です。

今回は、LINEWORKS Advent Calendar 7日目 で紹介したリマインダBOT の実装について紹介します。

[再掲]BOTの画面と全体構成

リマインダBOTは3つのLambdaで構成されており、Python3.7 で実装してます。

①. LINEWORKSから送信されるメッセージの処理およびSQSへの通知
②. テーブル内に保存されたイベントをポーリングおよびSQSへの通知
③. SQSから受信したメッセージをLINEWORKSサーバに通知

今回は、①に焦点を当てて紹介します。

状態遷移表とメッセージリスト

ユーザとのBOTのやり取りを状態遷移表で表現しました。リマインダBOTは、以下の4つのイベントを扱います。

• ユーザ参加
  • ユーザがBOTを追加時に発生
• テキスト入力
  • ユーザがBOTに対して任意のテキスト入力
• イベント入力ボタン押下
  • BOT内のメニューに表示される「イベント登録」を押下
• イベント出力ボタン押下
  • BOT内のメニューに表示される「イベント参照」を押下

それぞれのユーザイベントに対して4つの状態を管理します。
BOTは、ユーザイベントとBOTの状態に対応するメッセージをユーザに返答します。
メッセージの内容は、メッセージリストとして定義しておきます。

Lambdaの実装

では、本題のLambdaの実装です。

まずは、Lambda関数の全体の処理です。
リクエストボディの検証とメッセージのメインの処理を担う自作のon_event関数を呼び出します。
リクエストボディの検証は、ヘッダーのx-works-signatureの値に基づいて処理します。

"""
index.py
"""

import os
import json
from base64 import b64encode, b64decode
import hashlib
import hmac

import reminderbot

API_ID = os.environ.get("API_ID")


def validate(payload, signature):
    """
    x-works-signatureの検証
    """

    key = API_ID.encode("utf-8")
    payload = payload.encode("utf-8")

    encoded_body = hmac.new(key, payload, hashlib.sha256).digest()
    encoded_base64_body = b64encode(encoded_body).decode()

    return encoded_base64_body == signature


def handler(event, context):
    """
    main関数
    """

    # リクエストボディの検証
    if not validate(event["body"], event["headers"].get("x-works-signature")):
        return {
            "statusCode": 400,
            "body": "Bad Request",
            "headers": {
                "Content-Type": "application/json"
            }
        }

    body = json.loads(event["body"])

    # メッセージのメイン処理
    reminderbot.on_event(body)

    return {
        "statusCode": 200,
        "body": "OK",
        "headers": {"Content-Type": "application/json"}
    }

続いて、on_event関数についてです。
今回事前に定めた、4つの状態、4つのユーザイベント、メッセージリストを定数で定義しておきます。

"""
reminderbot.py
"""

import os

import json
import datetime
import dateutil.parser
from dateutil.relativedelta import relativedelta

import boto3
from boto3.dynamodb.conditions import Key, Attr

# 状態遷移表に基づき4つの状態を定義
STATUS_NO_USER = "no_user"
STATUS_WATING_FOR_BUTTON_PUSH = "status_waiting_for_button_push"
STATUS_WATING_FOR_NAME_INPUT = "status_waiting_for_name_input"
STATUS_WATING_FOR_TIME_INPUT = "status_waiting_for_time_input"

# メッセージリストに基づき定義
MESSAGE_LIST = [
    "こんにちは、リマインドボットだよ。メニューボタンを押してね。",
    "イベント名を入力してね",
    "メニューボタンを押してね。",
    "イベントの内容はこちら！",
    "イベント時間を入力してね。",
    "登録完了！",
    "エラーだよ。もう一度入力してね。",
]

# ユーザのイベントをpostbackイベントとして定義
# BOTのメニュー登録時は、以下のpostbackイベントの値と同じにすること
POSTBACK_START = "start"
POSTBACK_MESSAGE = "message"
POSTBACK_PUSH_PUT_EVENT_BUTTON = "push_put_event_button"
POSTBACK_PUSH_GET_EVENT_BUTTON = "push_get_event_button"

# ステータスを管理するテーブル
dynamodb = boto3.resource("dynamodb")
table = dynamodb.Table("lineworks-sample-table")


def on_event(event):
    """
    botの全体のイベントの処理
    """

    account_id = event["source"]["accountId"]
    content = event["content"]

    postback =  content.get("postback") or "message"

    # ユーザの今の状態を確認
    response = table.get_item(
        Key={
            "Hash": "status_" + account_id,
            "Range": "-"
        }
    )

    status = STATUS_NO_USER
    message = None

    if response.get("Item") is not None:
        status = response.get("Item")["Status"]

    # 各ユーザイベント(postback)毎の分岐処理
    try:

        if postback == POSTBACK_START:
            message = on_join(account_id, status)

        elif postback == POSTBACK_MESSAGE:
            text = content["text"]
            message = on_message(account_id, status, text)

        elif postback == POSTBACK_PUSH_PUT_EVENT_BUTTON:
            message = on_pushed_put_event_button(account_id, status)

        elif postback == POSTBACK_PUSH_GET_EVENT_BUTTON:
            message = on_pushed_get_event_button(account_id, status)

    except Exception as e:
        print(e)
        message = MESSAGE_LIST[6]

    # SQSにメッセージ内容を通知
    sqs = boto3.resource("sqs")
    queue = sqs.get_queue_by_name(QueueName="lineworks-message-queue")

    queue.send_message(
        MessageBody=json.dumps(
            {
                "content": {
                    "type": "text",
                    "text": message,
                },
                "account_id": account_id,
            }
        ),
    )

    return True

最後に、各イベントごとの処理の実装です。
それぞれのイベントの中で、各状態ごとの分岐処理を状態遷移表に基づいて実装しています。
重複する処理はまとめています。

def on_join(account_id, status):
    """
    bot追加時のイベントの処理
    """

    # ステータスに応じた分岐処理
    if status == STATUS_NO_USER:

        table.put_item(
            Item={
                "Hash": "status_" + account_id,
                "Range": "-",
                "Status": STATUS_WATING_FOR_BUTTON_PUSH,
            }
        )
        return MESSAGE_LIST[0]

    else:

        table.delete_item(
            Key={
                "Hash": "status_" + account_id,
                "Range": "-"
            }
        )

        table.put_item(
            Item={
                "Hash": "status_" + account_id,
                "Range": "-",
                "Status": STATUS_WATING_FOR_BUTTON_PUSH,
            }
        )

        return MESSAGE_LIST[0]

def on_message(account_id, status, text):
    """
    テキスト入力時のイベントの処理
    """

    if status == STATUS_WATING_FOR_BUTTON_PUSH:

        table.put_item(
            Item={
                "Hash": "status_" + account_id,
                "Range": "-",
                "Status": STATUS_WATING_FOR_BUTTON_PUSH,
            }
        )
        return MESSAGE_LIST[2]

    elif status == STATUS_WATING_FOR_NAME_INPUT:

        table.update_item(
            Key={
                "Hash": "status_" + account_id,
                "Range": "-",
            },
            UpdateExpression="set #st = :s, Title = :t",
            ExpressionAttributeNames = {
                "#st": "Status" # Statusは予約語なので#stに置き換える
            },
            ExpressionAttributeValues={
                ":s": STATUS_WATING_FOR_TIME_INPUT,
                ":t": text,
            },
        )
        return MESSAGE_LIST[4]

    elif status == STATUS_WATING_FOR_TIME_INPUT:

        # dateutil.parserで日付は変換
        time_dt = dateutil.parser.parse(text)
        time = time_dt.strftime("%Y/%m/%d %H:%M:%S")

        response = table.get_item(
            Key={
                "Hash": "status_" + account_id,
                "Range": "-",
            }
        )

        table.put_item(
            Item={
                "Hash": "event_" + account_id,
                "Range": time,
                "Title": response["Item"]["Title"],
                # utc -> 日本時間変換のため、9時間の差分をとる
                # utc -> 当初の予定 + 1h後に削除するように設定
                "ExpireTime": int((time_dt - relativedelta(hours=9) + relativedelta(hours=1)).timestamp()),
                "SentFlag": False
            }
        ),

        table.put_item(
            Item={
                "Hash": "status_" + account_id,
                "Range": "-",
                "Status": STATUS_WATING_FOR_BUTTON_PUSH,
            }
        )

        return MESSAGE_LIST[5]

def on_pushed_put_event_button(account_id, status):
    """
    「イベント登録」ボタン押下時のイベントの処理
    """

    if status == STATUS_WATING_FOR_BUTTON_PUSH:

        table.put_item(
            Item={
                "Hash": "status_" + account_id,
                "Range": "-",
                "Status": STATUS_WATING_FOR_NAME_INPUT,
            }
        )
        return MESSAGE_LIST[1]

    elif status == STATUS_WATING_FOR_NAME_INPUT:

        return MESSAGE_LIST[1]

    elif status == STATUS_WATING_FOR_TIME_INPUT:

        table.put_item(
            Item={
                "Hash": "status_" + account_id,
                "Range": "-",
                "Status": STATUS_WATING_FOR_NAME_INPUT,
            }
        )
        return MESSAGE_LIST[1]

def on_pushed_get_event_button(account_id, status):
    """
    「イベント参照」ボタン押下時のイベントの処理
    """

    current_jst_time = (datetime.datetime.utcnow() + relativedelta(hours=9)).strftime("%Y/%m/%d %H:%M:%S")

    # event取得処理 
    response = table.query(
        KeyConditionExpression=Key("Hash").eq("event_" + account_id) & Key("Range").gt(current_jst_time)
    )

    items = response["Items"] or []

    message = MESSAGE_LIST[3]

    if len(items) == 0:
        message += "\n-----"
        message += "\nなし"
        message += "\n-----"

    for item in items:

        message += "\n-----"
        message += "\n タイトル: {title}".format(title=item["Title"]) 
        message += "\n 日時: {time}".format(time=item["Range"]) 
        message += "\n-----"

    return message

まとめ

状態遷移表を作成することで、各イベント時にどのような処理を実装すべきか、
どのメッセージを返すべきか、が明確になるので迷いなく実装することができました。

今回は、シンプルなアプリだったので状態やイベントの数も少ないですが、
より複雑な処理をBOTにさせようとすると状態遷移表がより役に立ってくるかと思います。

はじめに

TouchDesigner(以下TD) ではpythonスクリプトを実行することができますが、
HTTP通信のような処理に時間のかかるものだと、その処理が終わるまでちょっと固まってしまいます。
このとき、TD全体のタイムラインも止まってしまい描画が更新されない状態になってしまうので、
それを防ぐために非同期で通信する方針をとりました。

ただ非同期で値を取ってきたいだけであれば、TDに加えてnodeなどなどを使えば良かったりもするのですが、
環境の都合によりTDだけで完結させたい状況だったので、以下の方法を検討しました。

• Web DAT を使う
• threading モジュールを使う（マルチスレッド）
• asyncio モジュールを使う（シングルスレッド）
• multiprocessモジュールを使う（マルチプロセス）

それぞれについて説明していきます。

※ ちなみにpythonで非同期通信する方法に関しては、以下の記事がわかりやすかったです。
https://qiita.com/icoxfog417/items/07cbf5110ca82629aca0

サンプルコード

今回のサンプルを以下にアップしました。

(github)

Web DAT を使う

これがいちばん簡単です。
URLを入力して Fetch ボタンをpulseしてあげれば、
TDのタイムラインは止まらずに裏で（たぶんcurlとか？）通信をしてくれます。
参考: https://docs.derivative.ca/Web_DAT

GETの場合はURLに直接入れてしまえばOKで、
POSTの場合は、上側のインレット Input 0 に送りたいデータのtableを繋げた状態で Submit ボタンをpulseすればOKです。
（ちなみに下側のインレットにはカスタムHTTPヘッダを入れることができます）

(gif?)

そして返ってきたものの形式に合わせて、XML DAT や jsonモジュールなどを使って必要な値を引っ張ってくれば完成です。

参考： TouchDesignerでJSONをDAT Tableにパースする

threading モジュールを使う

threading モジュール は、マルチスレッドで並列処理をしてくれるモジュールです。
TD内部のpython3.5に標準で入っています。

threading.Thread() メソッドの引数に、別スレッドで実行したい関数名およびその関数の引数（タプル）を入れることで、別スレッドで実行してくれます。

Matthew Ragan大先生の記事を大いに参考にしました。

欠点と対策

こちらもとても便利なのですが、欠点がいくつかあります。

• 別スレッドでの処理がいつ終了したかを検知できない
• 別スレッドの処理中にTDのオペレータを参照できない

別スレッドでの処理がいつ終了したかを検知できない

メインスレッドの方から停止の命令は送れるのですが、別スレッドでの処理が終了したタイミングを検知することができません。
なので、「別スレッドでURL叩いて返ってきたらこの処理を実行する」的なことが簡単にはできません。。

別スレッドの処理中にTDのオペレータを参照できない

これもよく考えれば当然ではあるのですが、別のスレッドで処理を回しているので、
TDのメインスレッドに存在しているオペレータにアクセスすることができません。
なので、「別スレッドでURL叩いて返ってきた結果をそのまま Table DAT にパースする」的なことはできません。

（ちなみに、それをするとこんなダイアログが出ます）

対策

なので、かなりの力技ですが、値が返ってきたかどうかを Timer CHOP で逐次監視することにしました。

おわりに

• TouchDesignerだけで非同期通信する方法についてまとめました。
• 特に最後の力技の部分、もっとスマートに解決できる気がしてならないので、もしもっと良い方法を思いついた方いらっしゃいましたら是非コメントいただけると嬉しいです…！

おまけ

• 今回、TouchPlayerを使って複数端末でそれぞれ動かしていたのですが、TouchPlayerだとTextportが見れないので、ログを外部ファイルに吐きだしてエラーが出たらそれを確認するという方法をとっていました。
  • （TouchDesignerをReadOnlyで開くという選択肢もあります。）
• その際、ログを吐き出すモジュールとしてloguruがとても便利だったのでオススメです。

参考にしたリンク集

estie Advent Calendar 2019　14日目の記事になります。
是非他の記事もご覧くださいね

はじめに

こんにちは、estie.incでエンジニアやってます、marushoです。
estieでは「テクノロジーの力で、世界を自由に、楽しく。」を合言葉に、不動産分野の「めんどくさい」を解消するため

• オフィス探しサービス estie
• 不動産データの可視化サービス estiepro

を運営しており、日々更新される不動産データを分析し、新たな価値の創出にチャレンジしています。

データ分析や分析結果の反映をスピーディに行うためには、セキュリティ構成を担保しつつ気軽にDBへアクセスする必要があります。
弊社ではpandas<->DBのやりとりが頻繁に行われるのですが、一度csvファイルに変換したり踏み台サーバに入ったりするのは何かと時間を消費してしまいます。

ということで、
今回は踏み台経由でDBのデータを、pandasで基本的なCRUD操作を直接やってみます。

環境

DBはprivateなサブネットにいて踏み台サーバを経由しないとアクセスできない、というよくある環境を前提とします。
今回はAWS上のEC2/RDS(MySQL5.7)で動作させています。

ちなみにローカル環境は

• MacOS Mojave
• Python 3.6.8
• Pandas 0.24.1

必要なパッケージをインストール

PythonでDB情報を扱うので、定番ORMのSQLAlchemyを使います。
また、MySQLのドライバと踏み台にSSHを張るためのSSHtunnelもインストールします

$ pip install SQLAlchemy PyMySQL sshtunnel

SSH config

普段sshに接続するために、.ssh/configにHostを登録しておくことが多いと思います。
今回もsshtunnelでconfigに書かれたHost情報を利用するので、以下のように踏み台の接続情報を書き込んでおきます。

~/.ssh/config

Host rds_bastion
    Hostname [踏み台IP]
    Port 22
    User [UserName]
    IdentityFile ~/.ssh/[KeyName]

RDSに接続

まずはmoduleのimportと、DBの接続に必要な情報を書いておきます

import pandas as pd
import sqlalchemy as sa
from sshtunnel import SSHTunnelForwarder

DB_USER = 'test_user' # DBのユーザー名
DB_PASS =  'db_passward' # DBのパスワード
ENDPOINT = 'hogehoge.fugafuga.ap-northeast-1.rds.amazonaws.com' # RDSエンドポイント
PORT = 3306 # ポート
DB_NAME = 'test_db' # DB名
CHARSET = 'utf8' # 文字コード

次にSSHポートフォワードを使って、踏み台越しのDBに接続します。

server = SSHTunnelForwarder(ssh_address_or_host = 'rds_bastion',
                            ssh_config_file = '~/.ssh/config',
                            remote_bind_address=(ENDPOINT,PORT))
server.start()

接続を終了するときはcloseしましょう

server.close()

sshを接続した状態で、SQLAlqhemyのエンジンを取得します。

#SQLAlchemyの接続URLを生成
URL = f"mysql+pymysql://{DB_USER}:{DB_PASS}@127.0.0.1:{server.local_bind_port}/{DB_NAME}?charset={CHARSET}"

#engineの取得
engine = sa.create_engine(URL)

このengineを使ってPandasでのデータ操作をやっていきます

Pandasでごにょごにょする

さて、本題です。
pandasでcreate,read,update,delete操作ができるか試してみましょう。

サンプルとして、DB名test_dbにmembersテーブルを作成しておきます

MySQL [test_db]> SELECT * FROM members;
+----+------------------+-----+
| id | name             | age |
+----+------------------+-----+
|  1 | 雪村 あおい       | 15  |
|  2 | 倉上 ひなた       | 15  |
|  3 | 斎藤 楓          | 16  |
|  4 | 青羽 ここな       | 13  |
+----+------------------+-----+

Read:読み込み

まずはpandas.read_sqlを使ってmembersテーブルをDataFrameとして読み込んでみましょう

テーブル全てのデータを読み込む場合は、テーブル名を指定します

df = pd.read_sql('members', engine)

	id	name	age
0	1	雪村 あおい	15
1	2	倉上 ひなた	15
2	3	斎藤 楓	16
3	4	青羽 ここな	13

綺麗に読み込めてますね

indexカラムの指定や、取得したいカラム名をリスト指定することもできます。

df= pd.read_sql('members', engine, index_col='id', columns=['name'])

id	name
1	雪村 あおい
2	倉上 ひなた
3	斎藤 楓
4	青羽 ここな

もちろんSQLクエリでレコード指定することも可能です。

df= pd.read_sql('SELECT * FROM members WHERE id = 2', engine)

	id	name	age
1	2	倉上 ひなた	15

Create:テーブル作成

to_sqlを使ってDataFrameのデータから新しいテーブルを作成できます。
(DataFarameの)indexの有無や、どれをindexとして取り込むかの指定もできます。

df = pd.read_sql('SELECT * FROM members WHERE age < 14', engine)
df.to_sql('jc_members', engine, index=False, index_label='id')

MySQL [test_db]> select * from jc_members;
+------+------------------+------+
| id   | name             | age  |
+------+------------------+------+
|    4 | 青羽 ここな      | 13   |
+------+------------------+------+

Update:レコードの挿入/更新

こちらもto_sqlで実行できますが、
if_existオプションで挙動が異なるので注意が必要です。

if_exist=appendとすると、新しいレコードとして追加し、同じレコードがあった場合はエラーになります。

insert_df = pd.DataFrame({'id':['5'],'name' : ['黒崎 ほのか'],'age':['14']})
insert_df.to_sql('members', engine, index=False, index_label='id', if_exists='append')

id	name	age
1	雪村 あおい	15
2	倉上 ひなた	15
3	斎藤 楓	16
4	青羽 ここな	13
5	黒崎 ほのか	14

INSERTとおなじ挙動ですね。ちゃんと追加されています。

しかしif_exist=replaceとすると、指定テーブルのデータをすぺてdeleteして、DataFrameを追加します。

insert_df = pd.DataFrame({'id':['5'],'name' : ['黒崎 ほのか'],'age':['14']})
insert_df.to_sql('members', engine, index=False, index_label='id', if_exists='replace')

id	name	age
5	黒崎 ほのか	14

UPDATEでもUPSERTでもなく、はたまたREPLACEとも異なる挙動なので注意が必要です！

特定レコードだけ更新する、などの操作はまだto_sqlに実装されいないようです。
今回は割愛しますが、SQLAlchemyのupsertを使う方法や、to_sqlのmethodオプションでSQLの挙動を変更するやり方があるようなので、試してみようと思います。

Delete:レコード/テーブルの削除

read_sqlでdrop/delete操作をするとreturnが無くエラーになるのですが、
実はDB側には削除操作が実行されてしまいます。

pd.read_sql('DROP TABLE members', engine)

MySQL [test_db]> SELECT * FROM members;
ERROR 1146 (42S02): Table 'test_db.members' doesn't exist

これは本来の用途ではないので、delete操作を行うときは素直にsqlalchemyでのクエリ実行をお勧めします

engine.execute('DROP TABLE members')

おわりに

離れたDBの情報を手軽にDataFrameにできるのは魅力ですね。
更新系のメソッドはかゆいところに手が届いてない感じなので、今後のpandasの発展を注視したいと思います。

---

estieではWebエンジニアを募集しています！
Wantedly
お気軽にオフィスに遊びに来てくださいね！

実施日：2019年12月14日
テーマ：Pythonのインストール

実施手順

• OS環境：MacOS_10.13.6（17G65）
• Pythonのダウンロード：https://www.python.org/downloads/
• コードのダウンロード：https://www.oreilly.co.jp/books/9784873118741/
  • 「第1刷正誤表」タブ中の内容を本に反映する
  • 「関連ファイル」タブ中の「サンプルコード」のリンクをクリックして、ソースコードや関連ファイルをダウンロードする
• インストール手順は省略します。
• 動作確認：アプリケーションの中に「Python 3.8.0」の中の「IDLE.app」を起動してシェル環境の画面が見える。
  

TypeScriptを一年程楽しく書いていたのですが、１ヶ月程前から業務でPython3を触ることになりました。
Pythonは簡潔にかけて楽しいものの、他の人から引き継いだ箇所もあり、開発していて型がないのが辛くなってきました。

一番つらいのはコードリーディングしていて、この関数は何を返すのか？　とか、この変数は何が入ってるのか、とかひと目見て分からないこと。……ドキュメントとしての型が無いことです。
無いなら導入しようということで、型アノテーションとmypyを導入することにしました。

いきなりガチガチに型を導入しても逆に辛くなるので、型のない状態から無理せず型を導入していく方法を取り、結構うまく導入できたので、今回はその時の手順や得られた知見、Tipsを紹介しようと思います。

環境

今回はpipenvの環境でテストしていますが、mypyに付いてはpip等でも特に変わらないと思います。

python3.7.5
pipenv, version 2018.11.26

pipenvの導入はこちらを参考にしてください
https://pipenv-ja.readthedocs.io/ja/translate-ja/index.html

インストール

プロジェクトディレクトリ下でmypyをインストールします。

cd ./myproject
pipenv install mypy -d

静的型検査の実行

グローバルのpipにmypyがインストールされていてsrcディレクトリ以下のコードを静的型検査をしたい場合はmypy ./srcで型検査ができますが、pipenvにしかmypyが入っていない場合はエラーになります。

$ mypy ./src
Command 'mypy' not found, but can be installed with:

sudo apt install mypy

pipenvの仮想環境に入れば問題なく実行できます。

$ pipenv shell
(myproject) $ mypy ./src
Success: no issues found in 2 source files

毎回仮想環境に入るのは面倒なので、Pipfileにスクリプトを登録しておきましょう。

[scripts]
type-check = "mypy ./src"

参考　https://pipenv-ja.readthedocs.io/ja/translate-ja/advanced.html#custom-script-shortcuts

スクリプトで実行したコマンドはpipenvの環境下で実行されるのでpipenv shellを行わなくてもmypyを実行できます。

$ pipenv run type-check
mypy.ini: No [mypy] section in config file
Success: no issues found in 1 source files

CIなどで型検査を実行するときはこのコマンドを使用します。

Python3の組み込み型

殆どの人が既知だと思いますが、一応基本的な型について復習しておきます
公式のドキュメントにあるうち、普通型検査で使用するのはせいぜい以下の８種類くらいに限られると思います。
https://docs.python.org/ja/3.7/library/stdtypes.html

迷ったら組み込み関数のtype()に入れて結果を見ればいいので覚える必要すら無いです。

型の種類	型名	例
真偽値型	bool	True
整数型	int	10
浮動小数点数型	float	1.2
テキストシーケンス型（文字列型）	str	'hoge'
リスト型	list	[1, 2, 3]
タプル型	tuple	('a', 'b')
辞書型（マッピング型）	dict	{ 'a': 'hoge', 'b': 'fuga'}
集合型	set	{ 'j', 'k', 'l'}

型のない関数に型を付けてみる

まず型のない関数を作ってみます。
./src以下にmy_module.pyを作成します。

my_module.py

def get_greeting(time):
  if 4 <= time < 10:
    return 'Good morning!'
  elif 10 <= time < 14:
    return 'Hello!'
  elif 14 <= time < 24:
    return 'Goog afternoon.'
  elif 0 <= time < 4:
    return 'zzz..'
  else:
    return ''


if __name__ == "__main__":
    print(get_greeting('morning'))

0から24までの時間を受け取って挨拶を返してくれる関数にしてみました。
これで型検査を実行してみると...

$ pipenv run type-check
Success: no issues found in 1 source file

何もエラーになりません！
なぜかというと、型アノテーションを行っていないので関数の返り値や引数の方は基本Any型（何でもありの型）になってしまうためです。
（既存のコードベースが存在する場合に、型導入時にエラーが出まくって心が折れたりしないのである意味これでいいと思います）
そこで次に、型アノテーションをつけてみます。

def get_greeting(time: int) -> str:
  if 4 <= time < 10:
    return 'Good morning!'
  elif 10 <= time < 14:
    return 'Hello!'
  elif 14 <= time < 20:
    return 'Goog afternoon.'
  elif 0 <= time < 4:
    return 'zzz..'
  else:
    return None

if __name__ == "__main__":
    print(get_greeting('morning'))

１行目に「整数型を受け取って文字列型を返す」ことを表す型アノテーションを追加しました。
この状態で型検査を再び実行してみます。

$ pipenv run type-check
src/my_module.py:14: error: Argument 1 to "get_greeting" has incompatible type "str"; expected "int"
Found 1 error in 1 file (checked 1 source file)

今度はちゃんとエラーが出ました。
エラーメッセージを読むと14行目で関数get_greetingを呼び出す際に文字列を渡してしまっています。このまま実行すると実行時エラーが発生してしまっていました。
整数型を渡すようにコード変更して、再度型検査を実行するとエラーが出なくなります。

    print(get_greeting(10))

型アノテーションを付けることで、コードが理解しやすくなり、さらに実行時エラーを未然に防ぐことができました。

設定ファイルmypy.iniを使う

そうはいっても型アノテーションを強制させたいときもあると思います。
その場合は設定ファイルを作成します。

mypy.ini

[mypy]
python_version = 3.7
disallow_untyped_calls = True
disallow_untyped_defs = True

スクリプトも設定ファイルを指定するように修正します。

[scripts]
type-check = "mypy ./src --config-file ./mypy.ini"

こうしておけば型アノテーションを付け忘れた場合にエラーを返してくれるようになります。

$ pipenv run type-check
src/my_module.py:1: error: Function is missing a type annotation
src/my_module.py:14: error: Call to untyped function "get_greeting" in typed context
Found 2 errors in 1 file (checked 1 source file)

参考　https://mypy.readthedocs.io/en/latest/config_file.html

ゆるく導入するために活用したいテクニック

Any許容で既存のコードベースに導入できるとはいえ、元のコードベースが大きいと導入の際に大量のエラーが発生するのは避けられません。
心が折れそうになる前にちょっとまってください。
次に紹介する２つを実行するだけでエラーの９割は消えるはずです。

参考 https://mypy.readthedocs.io/en/latest/existing_code.html#start-small

型のないモジュールのインポートを無視する

例えば以下のようなコードがあるとします。

import request

型検査を行うとエラーがなんと３行も返ってきます。

$ pipenv run type-check
src/my_module.py:1: error: Cannot find implementation or library stub for module named 'request'
src/my_module.py:1: note: See https://mypy.readthedocs.io/en/latest/running_mypy.html#missing-imports
Found 1 error in 1 file (checked 1 source file)

importしたモジュールの型定義ファイル(stub)が無いためです。

導入時に型定義ファイルを全部用意したりするのは大変なのでmypy.iniの設定で無視します。

mypy.ini

[mypy-request.*]
ignore_missing_imports = True

これでrequestからのインポートにstubが無いことを無視してくれます。

$ pipenv run type-check
Success: no issues found in 1 source file

これで平和が戻りました。

その行だけ無視する

あまり推奨はできないのですが、testなどであえて間違った型を代入させたいときにはよく使用する方法です。
無視したいコードの行末尾に#type: ignoreのコメントをつけます。

    print(get_greeting('hoge'))  #type: ignore

この行で発生するはずだった型エラーを抑制することができます。

おまけ：stubを自動生成する

「いやstubを使いたいんだ」という場合もあると思います。しかしサードパーティのモジュールの開発者がstubを用意していてくれている保証はありません。
自分で作るのは面倒です。
そういうときは自動生成しましょう。

mypyをいれると使えるようになる、stubgenコマンドでファイルやディレクトリを指定してstubを自動生成することができます。

$ stubgen foo.py bar.py

インポートしたモジュールであれば*.__path__でモジュールのpathを確認することができるのでそのpathを直接指定してstubを作ることもできます。

>>> import request
>>> request.__path__
['/home/username/.local/share/virtualenvs/myproject-xxxxxxxx/lib/python3.7/site-packages/request']
>>>

pathがわかったら、stubgenを実行します。

(myproject) $ stubgen /home/username/.local/share/virtualenvs/myproject-xxxxxxxx/lib/python3.7/site-packages/request
Processed 1 modules
Generated out/request/__init__.pyi

stubgenを実行するとプロジェクトrootにoutディレクトリが作成されるのでこのpathをmypyが見るようにmypy.iniに指定します。

mypy.ini

[mypy]
python_version = 3.7
mypy_path = ./out

型検査が通るようになりました。

$ pipenv run type-check
Success: no issues found in 1 source file

stubgenで生成される型は完全なものではありません。
大体Any型になってしまうので本格的に使用したければstubファイルを自分で修正する必要があります。

参考　https://github.com/python/mypy/blob/master/docs/source/stubgen.rst

よく使う発展的な型

組み込み型以外にもよく使う型が存在するので紹介しておきます。
mypyでは組み込み型以外はtypingモジュールやtyping_extensionsモジュールからそれらの型のクラスを呼び出して使用します。
typescriptとはちょっと使い勝手が違いますが、これらのモジュールにジェネリック型含めて殆どの型が網羅されているのでガチガチに型プログラミングしたいという人も満足できそうです。

参考　https://mypy.readthedocs.io/en/latest/

Optional

通常は整数を返し、間違った値を受け取った場合などにはNoneを返すなどの関数はよくあります。
その場合の返り値はintもしくはNoneですが、これを表現できるのがOptionalです。

from typing import Optional

def sample(time: int) -> Optional[int]:
  if 24 < time:
    return None
  else:
    return time

List, Dict

整数のリストや文字列のリストなどをListで表現できます。

from typing import List

# 整数のリスト
intList: List[int] = [1, 2, 3, 4] 

# 文字列のリスト
strList: List[str] = ['a', 'b', 'c']

同様にDictを使えば辞書型でも「keyが文字でvalueが整数」のような表現ができます。

from typing import Dict

# keyが文字でvalueが整数の辞書型
strIntDict: Dict[str, int] = {'a': 1, 'b': 2, 'c': 3, 'd': 4}

Union

複数の方を組み合わせたユニオン型を作成できます。

from typing import Union

strOrInt: Union[str, int] = 1  # OK
strOrInt = 'hoge'  # OK
strOrInt = None # error: Incompatible types in assignment (expression has type "None", variable has type "Union[str, int]")

Any

型を指定したくない場合はAny型ももちろん可能です

from typing import Any

string: str = 'hoge' 

any: Any = string
any = 10  # OK

notAny = string
notAny = 10  # error: Incompatible types in assignment (expression has type "int", variable has type "str")

Callable

Callableで関数の型を表現できます。

from typing import Callable

# 整数型の引数を一つ受け取って文字列型を返す関数型定義
func: Callable[[int], str]

def sample(num: int) -> str:
  return str(num)

func = sample

TypedDict

マップでkeyの値を指定してそのkeyは何の型のvalueを持っているかを指定したい場合があると思います。typescriptだとinterfaceで表現されるものです。

例えばmovieと言う辞書型の値があったとします。

movie = {'name': 'Blade Runner', 'year': 1982}

moveはnameとyearというkeyを持ちますが、上書きするときに間違ってnameに整数を入れてしまったり、yearに文字列を入れてしまったら困りますよね。
TypedDictを使うと簡単に型として表現できます。

from typing_extensions import TypedDict

Movie = TypedDict('Movie', {'name': str, 'year': int})

movie1: Movie = {'name': 'Blade Runner', 'year': 1982} # OK

movie2: Movie = {'name': 'Blade Runner', 'year': '1982'} # error: Incompatible types (expression has type "str", TypedDict item "year" has type "int")

クラスの形でも表現できます。個人的にはTSのinterfaceっぽくかけるのでこちらの方が好みです。

from typing_extensions import TypedDict

class Movie(TypedDict):
    name: str
    year: int

詳細は公式ドキュメントをご確認ください。
https://mypy.readthedocs.io/en/latest/more_types.html#typeddict

まとめ

いかがでしたでしょうか？
Pythonで型を始めるのは意外とハードルが低いんだなと感じていただけたら幸いです。

Pythonで型がなくて辛いという方は今すぐ導入しましょう！　思ったより大体揃っているので幸せになれます。

不満点としてはエディタの支援があんまりしっかりしてないということでしょうか。
VS-Codeの拡張機能でPyrightを使用しているのですが、もっと良いのがあれば乗り換えたいです。

それでは良い年末を！

この記事は kaggle その2 Advent Calendar 2019 - Qiita 14日目の記事です。

0. 導入

深層学習フレームワークはいずれも開発が非常に速く盛り上がっている分野だと思います。
TensorFlow や jax 等もある中、つい先日 PFN のニュースもあり、PyTorch もより盤石となりそうです。おそらくこれからも PyTorch ユーザーは増えると思われます(Chainer にもあった公式 Trainer が PyTorch 内に実装されるとこの記事の存在が危ぶまれるので、そこには触れないこととします)。

しかし PyTorch は自由度が高い一方、学習周りのコード(各 epoch のループ周りとか)は個々人に委ねられており、非常に個性豊かなコードとなりがちです。

これらのコードを自分で書くことは非常に学びが多く、PyTorch を始める場合には必ず通るべきだと私は思います。しかしあまりに個性が強すぎると、他の人との共有やコンペ間での使い回し等のシーンで辛いときがあります(ex. Winner Solutions でよく見かけるオレオレ Trainer)。

PyTorch の場合、学習周りのコードを簡略化するためのフレームワークは自身の中にはない(以前 Trainer があったが廃止された)のですが、Ecosystem | PyTorch の中では以下の PyTorch 用フレームワークが紹介されています。

• catalyst
• fastai
• ignite
• pytorch-lightning

多いですね。
全部試して自分に合ったものを見つけろというのは正論です。しかしそれらは楽ではないので本記事では各フレームワークの紹介と簡単な比較をしてみて、皆さんが触ってみる何かしらの目安になればと思います。

なお fastai については頭一つ抜けて抽象度が高い(コードが短くなりやすい)のですが、自身で細かい操作を加えるための学習コストが高く感じたため、本記事内の比較では予め省いております。

そのため本記事では Catalyst・Ignite・Lightning の3つに絞り、かつ Kaggle のコンペに参加することを想定して比較を行っていきます。

ちなみにこれらの 3つのフレームワークについては予めある程度動作することは確認しました。本記事を読んでもう少し踏み込みたくなった方はご参照いただければと思います。

• 本記事内のコード
  • https://github.com/yukkyo/Compare-PyTorch-Catalyst-Ignite-Lightning
• Catalyst: yukkyo/PyTorch-FilterResponseNormalizationLayer
• Ignite: PyTorch-Ignite で学習用コードをスマートにする - ふぁむたろうのブログ
• Lightning: yukkyo/Kaggle-Understanding-Clouds-69th-solution

先に述べますが、いずれもコンペに参加できるだけのポテンシャルはあります。

1. この記事の対象(とか対象外)

• Kaggle コンペに興味ある
• 画像系コンペに興味ある
  • 特に Classification・Segmentation・Detection あたりに興味ある
• PyTorch 触ったことある
  • 触ったことない人はこの記事読んでる場合じゃないです
  • 以下のページや本とかで始めましょう
    • Welcome to PyTorch Tutorials — PyTorch Tutorials 1.3.1 documentation
    • つくりながら学ぶ！PyTorchによる発展ディープラーニング | 小川 雄太郎 | 工学 | Kindleストア | Amazon

2. 各フレームワーク(Catalyst・Ignite・Lightning)比較

2019年12月13日時点の pip 上での最新版を使いました。
Python のバージョンは 3.7.5 です。また NVIDIA/apexもインストール済を想定しています。
このコードを動かす分には apex は不要です。

torch==1.3.1
torchvision==0.4.2
catalyst==19.12
pytorch-ignite==0.2.1
pytorch-lightning==0.5.3.2

2.1 Star 数遷移(2019年12月10日時点)

Catalyst・Ignite が順調に伸びている一方、Lightning は今年4月からすごい勢いで伸びてきました。
一方 Lightning はまだ世に出て一年も経っていないので開発中の機能も多く、まだ unstable(バージョン上げたときに後方互換性がない等)であることには注意です。

また最近の Kaggle Notebook 上では Catalyst をよく見かけるため、Catalyst が Ignite を追い抜かすこともありえそうです。

2.2 書き方

ここでは素の PyTorch 学習用コードに対し各フレームワークを適用したらどうなるのか確認します。

2.2.1 共通部分

今回は cifer10 dataset に対して Resnet18 で学習してみようと思います。
下記のコードのように、モデルや Dataloader の定義は予め関数にしておきます。

共通部分のコード(長いのでたたみました)

share_funcs.py

import torch
import torch.nn as nn
from torch import optim
from torch.utils.data import DataLoader
from torchvision import datasets, models, transforms

def get_criterion():
    """Loss をよしなに返してくれる関数"""
    return nn.CrossEntropyLoss()

def get_loaders(batch_size: int = 16, num_workers: int = 4):
    """各 Dataloader をよしなに返してくれえる関数"""
    transform = transforms.Compose([transforms.ToTensor()])

    # Dataset
    args_dataset = dict(root='./data', download=True, transform=transform)
    trainset = datasets.CIFAR10(train=True, **args_dataset)
    testset = datasets.CIFAR10(train=False, **args_dataset)

    # Data Loader
    args_loader = dict(batch_size=batch_size, num_workers=num_workers)

    train_loader = DataLoader(trainset, shuffle=True, **args_loader)
    val_loader = DataLoader(testset, shuffle=False, **args_loader)
    return train_loader, val_loader

def get_model(num_class: int = 10):
    """モデルをよしなに返してくれる関数"""
    model = models.resnet18(pretrained=True)
    num_features = model.fc.in_features
    model.fc = nn.Linear(num_features, num_class)
    return model

def get_optimizer(model: torch.nn.Module, init_lr: float = 1e-3, epoch: int = 10):
    optimizer = optim.SGD(model.parameters(), lr=init_lr, momentum=0.9)
    lr_scheduler = optim.lr_scheduler.MultiStepLR(
        optimizer,
        milestones=[int(epoch*0.8), int(epoch*0.9)],
        gamma=0.1
    )
    return optimizer, lr_scheduler

2.2.1 ベースコード(素の学習用コード)

あまり深く考えずに愚直に書くと下記のようになると思います。
.to(device) や loss.backward()、optimizer.step() は書かなきゃいけないので、どうしても長くなりがちです。
また with torch.no_grad() は torch.set_grad_enabled(bool) を使うことで Train と Eval 時の両方に対応させることは可能なのですが、Train と Eval 時は違う処理が多く(ex. optimizer.step() や metrics 等)、両方対応させるような関数を作るとかえって見通しが悪くなりがちです。

ベースコード(長いのでたたみました)

def train(model, data_loader, criterion, optimizer, device, grad_acc=1):
    model.train()

    # zero the parameter gradients
    optimizer.zero_grad()

    total_loss = 0.
    for i, (inputs, labels) in tqdm(enumerate(data_loader), total=len(data_loader)):
        inputs = inputs.to(device)
        labels = labels.to(device)

        outputs = model(inputs)

        loss = criterion(outputs, labels)
        loss.backward()

        # Gradient accumulation
        if (i % grad_acc) == 0:
            optimizer.step()
            optimizer.zero_grad()

        total_loss += loss.item()

    total_loss /= len(data_loader)
    metrics = {'train_loss': total_loss}
    return metrics


def eval(model, data_loader, criterion, device):
    model.eval()
    num_correct = 0.

    with torch.no_grad():
        total_loss = 0.
        for inputs, labels in tqdm(data_loader, total=len(data_loader)):
            inputs = inputs.to(device)
            labels = labels.to(device)

            outputs = model(inputs)
            _, preds = torch.max(outputs, 1)

            loss = criterion(outputs, labels)

            total_loss += loss.item()
            num_correct += torch.sum(preds == labels.data)

        total_loss /= len(data_loader)
        num_correct /= len(data_loader.dataset)
        metrics = {'valid_loss': total_loss, 'val_acc': num_correct}
    return metrics


def main():
    epochs = 10

    device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
    model = get_model()
    train_loader, val_loader = get_loaders()
    optimizer, lr_scheduler = get_optimizer(model=model)
    criterion = get_criterion()

    # Model を multi-gpu したり、FP16 対応したりする
    model = model.to(device)

    print('Train start !')
    for epoch in range(epochs):
        print(f'epoch {epoch} start !')
        metrics_train = train(model, train_loader, criterion, optimizer, device)
        metrics_eval = eval(model, val_loader, criterion, device)

        lr_scheduler.step()

        # Logger 周りの処理
        # print するためのごちゃごちゃした処理
        print(f'epoch: {epoch} ', metrics_train, metrics_eval)

        # tqdm 使ってたらさらにごちゃごちゃする処理をここに書く
        # Model を保存するための処理
        # Multi-GPU の場合さらに注意して書く

2.2.2 Catalyst

Catalyst の場合、ライブラリ内の SupervisedRunner に必要なものを渡せば終わりです。すごいスマートですね！
また Accuracy や Dice 等のメジャーな metrics であれば Catalyst 内にあるため、それらを使えば自分で書くことはほとんどありません(独自 metrics の導入も比較的楽そうでした)。
大抵デフォルトのままで困らなそうですが、自分で細かい処理を加えたい場合若干調べる必要がありそうです。

import catalyst
from catalyst.dl import SupervisedRunner
from catalyst.dl.callbacks import AccuracyCallback
from share_funcs import get_model, get_loaders, get_criterion, get_optimizer

def main():
    epochs = 5
    num_class = 10
    output_path = './output/catalyst'

    model = get_model()
    train_loader, val_loader = get_loaders()
    loaders = {"train": train_loader, "valid": val_loader}

    optimizer, lr_scheduler = get_optimizer(model=model)
    criterion = get_criterion()

    runner = SupervisedRunner(device=catalyst.utils.get_device())
    runner.train(
        model=model,
        criterion=criterion,
        optimizer=optimizer,
        scheduler=lr_scheduler,
        loaders=loaders,
        logdir=output_path,
        callbacks=[AccuracyCallback(num_classes=num_class, accuracy_args=[1])],
        num_epochs=epochs,
        main_metric="accuracy01",
        minimize_metric=False,
        fp16=None,
        verbose=True
    )

2.2.3 Ignite

Ignite は Catalyst や後述する Lightning とは少し毛色が違います。
下記のように @trainer.on(Events.EPOCH_COMPLETED) 等で自分が挟みたい処理を各タイミングに対して差し込んでいくようなイメージです。
また Ignite も公式で Accuracy 等は用意されているのでメジャーな評価指標であれば自分で定義せずに済みそうです。

一方使いこなすのに慣れが必要そうなのと、イベント挟み方の自由度が高い(trainder.append のような足し方もできる)ので、一歩間違えると全体の見通しが悪くなる可能性もあります。

import torch
from ignite.engine import Events, create_supervised_trainer, create_supervised_evaluator
from ignite.metrics import Accuracy, Loss, RunningAverage
from ignite.contrib.handlers import ProgressBar
from share_funcs import get_model, get_loaders, get_criterion, get_optimizer

def run(epochs, model, criterion, optimizer, scheduler,
        train_loader, val_loader, device):
    trainer = create_supervised_trainer(model, optimizer, criterion, device=device)
    evaluator = create_supervised_evaluator(
        model,
        metrics={'accuracy': Accuracy(), 'nll': Loss(criterion)},
        device=device
    )

    RunningAverage(output_transform=lambda x: x).attach(trainer, 'loss')

    pbar = ProgressBar(persist=True)
    pbar.attach(trainer, metric_names='all')

    @trainer.on(Events.EPOCH_COMPLETED)
    def log_training_results(engine):
        scheduler.step()
        evaluator.run(train_loader)
        metrics = evaluator.state.metrics
        avg_accuracy = metrics['accuracy']
        avg_nll = metrics['nll']
        pbar.log_message(
            "Training Results - Epoch: {}  Avg accuracy: {:.2f} Avg loss: {:.2f}"
            .format(engine.state.epoch, avg_accuracy, avg_nll)
        )

    @trainer.on(Events.EPOCH_COMPLETED)
    def log_validation_results(engine):
        evaluator.run(val_loader)
        metrics = evaluator.state.metrics
        avg_accuracy = metrics['accuracy']
        avg_nll = metrics['nll']
        pbar.log_message(
            "Validation Results - Epoch: {}  Avg accuracy: {:.2f} Avg loss: {:.2f}"
            .format(engine.state.epoch, avg_accuracy, avg_nll))

        pbar.n = pbar.last_print_n = 0

    trainer.run(train_loader, max_epochs=epochs)

def main():
    epochs = 10
    train_loader, val_loader = get_loaders()
    model = get_model()
    device = 'cuda' if torch.cuda.is_available() else 'cpu'
    optimizer, scheduler = get_optimizer(model)
    criterion = get_criterion()

    run(
        epochs=epochs,
        model=model,
        criterion=criterion,
        optimizer=optimizer,
        scheduler=scheduler,
        train_loader=train_loader,
        val_loader=val_loader,
        device=device
    )

2.2.4 Lightning

Lightning の場合、LightningModule を継承したクラス(Trainer クラス的なもの)を定義する必要があります。

各step(ex. training_step)の名前は決まっており、各 step を自分で埋めていきます。
また学習の実行自体は pytorch_lightning.Trainer クラスによって行われ、GPU や MixedPrecision、gradient accumulation 等の設定はこのクラスで設定します。
また metrics については Lightning 内には用意されていないため、自分で記載する必要があります。

import torch
import pytorch_lightning as pl
from pytorch_lightning import Trainer
from share_funcs import get_model, get_loaders, get_criterion, get_optimizer

class MyLightninModule(pl.LightningModule):
    def __init__(self, num_class):
        super(MyLightninModule, self).__init__()
        self.model = get_model(num_class=num_class)
        self.criterion = get_criterion()

    def forward(self, x):
        return self.model(x)

    def training_step(self, batch, batch_idx):
        # REQUIRED
        x, y = batch
        y_hat = self.forward(x)
        loss = self.criterion(y_hat, y)
        logs = {'train_loss': loss}
        return {'loss': loss, 'log': logs, 'progress_bar': logs}

    def validation_step(self, batch, batch_idx):
        # OPTIONAL
        x, y = batch
        y_hat = self.forward(x)
        preds = torch.argmax(y_hat, dim=1)
        return {'val_loss': self.criterion(y_hat, y), 'correct': (preds == y).float()}

    def validation_end(self, outputs):
        # OPTIONAL
        avg_loss = torch.stack([x['val_loss'] for x in outputs]).mean()
        acc = torch.cat([x['correct'] for x in outputs]).mean()
        logs = {'val_loss': avg_loss, 'val_acc': acc}
        return {'avg_val_loss': avg_loss, 'log': logs}

    def configure_optimizers(self):
        # REQUIRED
        optimizer, scheduler = get_optimizer(model=self.model)
        return [optimizer], [scheduler]

    @pl.data_loader
    def train_dataloader(self):
        # REQUIRED
        return get_loaders()[0]

    @pl.data_loader
    def val_dataloader(self):
        # OPTIONAL
        return get_loaders()[1]


def main():
    epochs = 5
    num_class = 10
    output_path = './output/lightning'

    model = MyLightninModule(num_class=num_class)

    # most basic trainer, uses good defaults
    trainer = Trainer(
        max_nb_epochs=epochs,
        default_save_path=output_path,
        gpus=[0],
        # use_amp=False,
    )
    trainer.fit(model)

2.3 各フレームワークで実行したときのコンソール画面とアウトプット

2.3.1 デフォルト

コンソール画面

$ python train_default.py
Files already downloaded and verified
Files already downloaded and verified
Train start !
epoch 0 start !
100%|_____| 196/196 [00:05<00:00, 33.44it/s]
100%|_____| 40/40 [00:00<00:00, 50.43it/s]
epoch: 0  {'train_loss': 1.3714478426441854} {'valid_loss': 0.992230711877346, 'val_acc': tensor(0, device='cuda:0')}

アウトプット

なし

2.3.1 Catalyst

コンソール画面

$ python train_catalyst.py
1/5 * Epoch (train): 100% 196/196 [00:06<00:00, 30.09it/s, accuracy01=61.250, loss=1.058]
1/5 * Epoch (valid): 100% 40/40 [00:00<00:00, 49.75it/s, accuracy01=56.250, loss=1.053]
[2019-12-14 08:47:33,819]
1/5 * Epoch 1 (train): _base/lr=0.0010 | _base/momentum=0.9000 | _timers/_fps=58330.0450 | _timers/batch_time=0.0071 | _timers/data_time=0.0045 | _timers/model_time=0.0026 | accuracy01=52.0863 | loss=1.3634
1/5 * Epoch 1 (valid): _base/lr=0.0010 | _base/momentum=0.9000 | _timers/_fps=77983.3850 | _timers/batch_time=0.0146 | _timers/data_time=0.0126 | _timers/model_time=0.0019 | accuracy01=65.6250 | loss=0.9848
2/5 * Epoch (train): 100% 196/196 [00:06<00:00, 30.28it/s, accuracy01=63.750, loss=0.951]

アウトプット

• Tensorboard 等はデフォルトで出力されます
• weight も保存されます
• デフォルトで code まで残してくれるのはちょっとうれしいですね

catalyst
├── checkpoints
│   └── train.1.exception_KeyboardInterrupt.pth
├── code
│   ├── share_funcs.py
│   ├── train_catalyst.py
│   ├── train_default.py
│   └── train_lightning.py
├── log.txt
└── train_log
    └── events.out.tfevents.1576306176.FujimotoMac.local.41575.0

2.3.2 Ignite

コンソール画面

Catalyst よりもややスッキリした画面です。

$ python train_ignite.py
Epoch [1/10]: [196/196] 100%|________________, loss=1.14 [00:05<00:00]
Training Results - Epoch: 1  Avg accuracy: 0.69 Avg loss: 0.88
Validation Results - Epoch: 1  Avg accuracy: 0.65 Avg loss: 0.98
Epoch [2/10]: [196/196] 100%|________________, loss=0.813 [00:05<00:00]
Training Results - Epoch: 2  Avg accuracy: 0.78 Avg loss: 0.65
Validation Results - Epoch: 2  Avg accuracy: 0.70 Avg loss: 0.83

アウトプット

• なし
• 自分で保存する部分を書くか、Ignite 内のクラスを使う必要がありそうです

2.3.3 Lightning

コンソール画面

Lightning ではデフォルトでは tqdm 内のバーに全て表示するようです。

$ python train_lightning.py
Epoch 1: 100%|_____________| 236/236 [00:07<00:00, 30.75batch/s, batch_nb=195, gpu=0, loss=1.101, train_loss=1.06, v_nb=5]
Epoch 4:  41%|_____________| 96/236 [00:03<00:04, 32.28batch/s, batch_nb=95, gpu=0, loss=0.535, train_loss=0.524, v_nb=5]

アウトプット

• Lightning はディレクトリが重複した場合に version_x のように次のディレクトリを作って保存します。(それがかえって邪魔な場合もあり自分で checkpoint を定義することもありますが)
• Lightning の場合、meta_tags.csv に LightningModule に渡したパラメータが自動で保存されます
• Tensorboard 用の log もデフォルトで作成されます
• weight も checkpoints 内に保存されます
  • デフォルトでは各 epoch 毎に _ckpt_epoch_X.ckpt が作成され、古い epoch の ckpt を削除しているようです

lightning
└── lightning_logs
    ├── version_0
    │   └── checkpoints
    │       └── _ckpt_epoch_4.ckpt
    │   ├── media
    │   ├── meta.experiment
    │   ├── meta_tags.csv
    │   ├── metrics.csv
    │   └── tf
    │       └── events.out.tfevents.1576305970
    ├── version_1
    │   └── checkpoints
    │       └── _ckpt_epoch_3.ckpt
    │   ├── ...

2.4 その他めぼしいところ

いずれも Early Stopping 等は対応しています。

2.4.1 Catalyst

• catalyst.utils.set_global_seed() 等の再現性周りの関数も用意されている
• Dataset をより簡略化して書けるような関数もサポートされている
  • create_dataset, create_dataframe, prepare_dataset_labeling, split_dataframe
  • catalyst.utils.pandas
• Multi GPU や FP16 もサポートされている
• 公式 Tutorial のクオリティが高い
• 公式に Docker ファイルも置いてあり、インフラ周りの構成管理も意識したフレームワークを目指してるっぽい(多分)

2.4.2 Ignite

• Tensorboard や Logger も Ignite 内にあり、呼び出して使える
• 自由度が最も高そう
  • イベントの挟み方は慣れが必要そうだけど
• 公式リポジトリの下に置いてある

2.4.3 Lightning

• Multi GPU や FP16 もサポートされている

2.5 おすすめするとしたら

いずれもポテンシャルはあるため強制ではないです。下記は個人の感想です。

• 画像系コンペ初めてで、何からやればよいかよくわからない
  • → PyTorch Catalyst
• 画像系コンペは慣れきってて、殺意(金メダルを取りにいく強い気持ち)を持ってコンペに参加したい
  • → PyTorch Lightning か PyTorch Catalyst
• Classification・Segmentation に限らず色んな画像系タスクを取り組みたい
  • → PyTorch Lightning か PyTorch Catalyst
    • Catalyst 内には強化学習用のサンプルコードもある
• オサレに書きたい
  • → PyTorch Ignite
• PyTorch 公式のお膝元で安心してフレームワークを使いたい
  • → PyTorch Ignite
  • Ignite は 公式 PyTorch のリポジトリに置いてある

3. Catalyst・Ignite・Lightning を自由に行き来するために

ここで使うフレームワークを絞ってしまってもよいのですが、そもそも各フレームワークを行き来しやすいように書いていれば困らないはずです。 ですので PyTorch のコードを書き散らす上で意識しておくと良さそうなことをここにまとめます。

• ループの中身はなるべく取り出しておく
  • 各 step(ex. train 内の 1バッチごとの処理)の処理は抜き出せるように意識しておくと良さそうです
  • 少なくとも三重ループまで書き始めたら、ループの中身を抜き出せないか意識すると良さそうです
• 関数の引数を増やしすぎない
  • Class にしてインスタンス変数を使ったり、Config を一つにまとめて使っても良いと思います
• optimizer や model を呼び出す関数を作る
  • 個人の所感です
• Config を一つにまとめる
  • コンペに参加するときはなるべく一箇所に設定をまとめた方が管理が楽
  • 例
    • Config クラスを作る
    • Addict 等で呼び出しやすい辞書みたいなものを作る
    • YAML ファイルで書いた設定を読み出す

4. さいごに

本記事では PyTorch Catalyst・Ignite・Lightning の比較を行いました。
いずれも定型文を無くしたいという部分は一致していますが、それぞれ個性が出る結果となりました。
どのフレームワークもポテンシャルはあるため、もし触ってみて自分に合っていると思ったらコンペに出て使い倒してみると良いかと思います。

良い Kaggle(with PyTorch) ライフを！