この記事、何?

ちょっとしたスクリプトを作ろうという話の一つ。
今回はBase64。
たまに、末尾が「==」で終わる乱数チックな文字列を見るのでなんだろ思ったので気になってた。
たまに照合したりするのでEncode/Decodeをできるように。

どういう人向け?

まあ・・・自分あて。
アピールできるほどの記事ではないので・・・
ただ、たまに「昔やったんだけどなぁ・・・細かいパラメータが分からないわ～」ってなるのを避けるため、将来の自分あて。

参考にした記事

■Pythonでbase64エンコード・デコードする方法
http://yut.hatenablog.com/entry/20110825/1314228258

■Base64デコード・エンコード
https://qiita.com/mas9612/items/5d3bf90b04bf19a1bf20

できたコード

#18-01.py (エンコード)
import sys
import base64
args = sys.argv
arg = args[1]
print(arg)
print(base64.b64encode(arg.encode('utf-8')))

#18-02.py (デコード)
import sys
import base64
args = sys.argv
arg = args[1]
print(arg)
print(base64.b64decode(arg.encode('utf-8')))

(実行イメージ)
>python 18-01.py abcdefg
abcdefg
b'YWJjZGVmZw=='

>python 18-02.py YWJjZGVmZw==
YWJjZGVmZw==
b'abcdefg'

ということで、[abcdefg] から [YWJjZGVmZw==] が取得された。
で、戻して原文が取得できた。

あっ、クリップボードに欲しかったら | clip って書いてください。
あと、テキストファイルに欲しかったら > text.txt とかで。

感想

記事が薄いわぁ・・・
pythonいいな、という話以前に、他の方の書いたのを数行写経して終わり・・・
いや、今週はほら、いろいろイベントあって忙しかったし・・・

今度はもうちょっとましなものを投稿します。。。

C#で再起ロジック書いてみたかったのでやってみようかな、と思ってま~す。

はじめに

個人用のメモ書きです。Djangoの公式ドキュメント「はじめてのDjangoアプリ作成」をもとに作成しています。

環境

• PC ： MacBook Air (13-inch, Early 2014)
• OS ： macOS Mojave Ver. 10.14.2
• python ： Ver. 3.7
• Django ： Ver. 2.1.5

前提条件

Python、Djangoは既にインストール済であることを前提としています。

完成イメージ

ブラウザ上に「Hello, World!」を表示します。

プロジェクトを作成する

デモ用にプロジェクトを新規作成します。プロジェクト名は 「sample_project」とします。

コマンドライン

$ django-admin startproject sample_project

以下のファイルが生成されます。

sample_project/
    manage.py
    sample_project/
        __init__.py
        settings.py
        urls.py
        wsgi.py

外側のsample_project/ ルートディレクトリはDjangoに依存しないため、名前は好きなものに変更できます。既に存在するディレクトリにプロジェクトを作成する場合は、プロジェクト名に続いてディレクトリを入力して実行します。

コマンドライン

$ django-admin startproject sample_project (任意のディレクトリ)

なお、プロジェクト名に 「-（ハイフン）」は使用できません。また、プロジェクト名に限らず、予約語や組み込み関数名と衝突しないよう注意しましょう。

アプリケーションを作成する

プロジェクトを作成したら、次はアプリケーションを作成します。アプリケーション名を「app1」とします。

コマンドライン

$ python manage.py startapp app1

アプリケーションは実際に何らかの処理を行う個別のアプリーケーションを指します。プロジェクトはあるウェブサイト向けに設定とアプリケーションを集めたものです。例えば、販売管理システム（プロジェクト）に対し、売上管理アプリ、在庫管理アプリ、予算管理アプリ(アプリケーション）といった具合です。

ビューを作成する

views.pyを開き、以下のpythonコードを書きます。

app1/views.py

from django.http import HttpResponse


def index(request):
    return HttpResponse("Hello, world!")

URLconfを作成する

先ほど書いたビューを呼ぶために、URLを対応付けします。
まず、app1ディレクトリにurls.pyというファイルを作ります。アプリのディレクトリは以下のようになります。

app1/
    __init__.py
    admin.py
    migrations/
        __init__.py
    models.py
    tests.py
    urls.py
    views.py

作成した app1/urls.py に以下のコードを書きます。

app1/urls.py

from django.urls import path

from . import views


urlpatterns = [
    path('', views.index, name='index'),
]

次に、作成したアプリのURLconfをプロジェクト側のURLconfに反映させます。
今度は sample_project/urls.py に以下を追加します。

sample_project/urls.py

from django.contrib import admin
from django.urls import include, path


urlpatterns = [
    path('app1/', include('app1.urls')),
    path('admin/', admin.site.urls),
]

結果

コマンドラインで開発用サーバを動かします。

コマンドライン

$ python manage.py runserver

ブラウザで http://localhost:8000/app1/ にアクセスすると「Hello World!」が表示されます。

おわりに

公式により丁寧な説明がありますので、基本はそちらを使って勉強するのがよいと思います。ただ動かすだけであれば、プログラミング未経験者でも迷うポイントはあまりないと思います。
個人的引っかかりポイントは、プロジェクトとアプリケーションの区別（どのような単位でアプリケーションを作ればよいか？）とそれぞれの名前の付け方でした。

はじめに

勢いでIntelliJ All Products　Pack買いました！
衝動買い楽しい。
せっかくだしちゃんと勉強しながら色々と使いこなせるようにメモします。
この機能の紹介をぜひー！とかこれ忘れてないなどあればどんどんお教えいただけますと幸いです。追記していきます！！
あと、サムライズムさんとかIntelliJとかお金もらってかいているわけではないです。

色々あるけど・・・

いろんな機能があって色々と扱えるものも違ってきます。
ちなみに最強はIntelliJ IDEA Ultimate Editionです。
plugin で追加さえすればめちゃめちゃ捗ります。

僕がweb開発とアプリする人間なので、それ以外にどうするべきなのかはわからないです。
値段は年間ですが、３年目まで毎年安くなっていきます。

JavaScripter

webStormで要件を満たせます。
年間６３００円で、三年目だと3700円です。

webStormは、HTML/CSS、js, ts, coffe scriptを扱うことができます。なので、フロントエンドだろうが、バックエンド（nodeでの話）だろうがこれ一個で大丈夫です。
goに関してはプラグインでなんとかできるけど辛いみたい・・・。
そして、WebStormはかなり安いです。

ただし、こいつでDBを覗き見たりするのはpluginでできなくもないですが、Data Gripを買った方がいいかと。
二郎を食べたことない人が二郎の美味しさがわからないてきな感じでData Gripを触らない限りは他のフリーツールでも満足してやっていけます。
ただし、free trialでもData Grip触ろうものなら、もう戻ってこれなくなります。

pythoner

pyCharm professional Edition(PE)で要件を満たせます。
年間9600円で三年目だと5700円です**

pythonはCommunity Edition(CE)って無償のバージョンがありますが、これは基本的にはデータサイエンス向けでweb開発用に転用していくのは結構無理がでます。データサイエンスだけのためならCEでよいです！jupyterも動かせます！
そして、こいつはなんとHTML/CSS, js, ts, CoffeScriptも扱うことができます！
なのでこれを買えば、フロンエンドだろうが、バックエンド（python or node)だろうがこれ一個で大丈夫です。
ただし、値段はちょっと高いです。

phper

phpStormです。
年間9600円で三年目だと5700円です
webStormの上位互換ににあたります。
そして、こいつもなんとHTML/CSS, js, ts, CoffeScriptも扱うことができます！
なのでこれを買えば、フロンエンドだろうが、バックエンド（php or node)だろうがこれ一個で大丈夫です。
上位互換なので値段はやっぱちょっと高いです。

Rubyer

RubyMineです。
年間9600円で三年目だと5700円です**
僕はあんまりRubyに関して明るくないです・・・。
が、Rubyを扱うことができ、こいつもなんとHTML/CSS, js, ts, CoffeScriptも扱うことができます！
ということでこれもフロンエンドだろうが、バックエンド（ruby or node)だろうがこれ一個で大丈夫です。

go langer

go landです。
年間9600円で三年目だと5700円です

かっこいいですね。
これはHTML/CSS, js, tsを扱うことができます！

javer

IntelliJ IDEA ultimate Edition(UE)を使うことができます。
年間16100円で三年目だと9600円です

一応無償のCommunity Edition(UE)もありますが、で開店としてはHTML/CSS, js, markdownとかに対応していない点です。
java, kotolin, androidとかならCEで十分ですが、フロント周りをやろうとするならこれだとできないです。
pulanginをいれることで、多少python、Goなどに対して耐性をつけることもできます。
これだけ機能がたくさんあるためややお高い値段設定になっています。

swifter Objective-Cer

AppCodeがあります。
年間9600円で三年目だと5700円です

いままでXCode使ってきたので使用感はちょっと不明です・・・。というかReact Nativeエンジニアなので、そこまで必要なものでもない。
これは、swift, Objective-C/C++のほか、HTML/CSS、JavaScriptを扱えますが、ts、Coffieはできません。
なのでReact Nativeやる人からしてみてもAppCode買うかは微妙なところですね。

SQLer

Data Gripがあります。
年間9600円で三年目だと5700円です
これはめっちゃ良くてかなり重宝しています。

All Product Packは買いなの？

はっきし言ってAll Product Packは結構たかいです。
年間26900円で三年目だと16100円です

普通に開発しててもなんだかんだで２言語ぐらいできたら結構十分かと思います。
基本的に元を取ろうと思ったら３製品はちゃんと使わないともったいないです。
ですが、js系のものはだいたいどれもあつかえるので３製品使おうとするのは結構難しいです。

たとえば、僕の場合、

• フロントエンドやる
• アプリやるけどReact Nativeなのでswift、javaはそんなに触らない
• バックエンド node or python
• データサイエンスやる
• データベースみる

とかになってくると、pyCharmとDataGripあれば十分かなと言ったところです。
それと、基本的にアップデートの速さは固有の製品の方が早いみたいなので、前線をいく人は前線をいけるような製品チョイスを心がけた方がいいかもしれません。
なので中なかなかに絶妙な値段設定をしている感じがしますね・・・。

値段のまとめ

２6900円コース

All Product Pack

16100円コース

IntelliJ IDEA Ultimate Edition

9600円コース

phpStorm, RubyMine, PyCharm, DataGrip, Goland, AppCode

６３００円コース

webStorm

と言った感じです。なので、ここから自分に何が必要なのかを吟味して選んでいくといいかと思います！
あと、僕みたいに心配性だったり考えるのめんどくさ買ったりする人はAll Product Packを買うといいと思います。

intention Actions

option + enter
いろんなことができます。
例えば、勢いでコード書いたけどファイルつくってなかった時とか、ここからさささっと作ることができます。

post completion

あと出しでコードを書く機能。
みてもらったほうが早い機能。そんなにいらないんじゃないって思うかもですが、なんだかんだ言って覚えれば便利な機能です。

一覧はpreference > editor > general > Postfix Completionからみれます。

独自定義もできます。

割と便利な

console.log({
  a,
  b,
});

を作りました。

live template

色々な機能がある。アロー関数をfarrowとして追加してみた。
こちらを参考にした。

タブを押すたびに引数、処理と言ったことができる。

こんな感じで追加する。ワーニングっぽいのでてたらどの言語でできているかわからないっていっているので、どの言語で使えるかを設定してあげましょう。

($PARAMS$) => ({
    $BODY$
})$END$;

こんな感じで定義すると、param -> body -> endにtabを押すたびに遷移する。
一応、テンプレート的なのは、範囲を選択 -> tools -> save as livetemplateっていうのを押すと、選択範囲のコードが入った状態で始められる親切設計。

expand selection

いい感じに選択範囲をする機能です。
option + ↑で回数に応じて範囲が広がります。逆に戻したい場合はoption + ↓です。

変数化とインライン化

名前的に変数化したかったり、戻したりを一瞬でできます。

cmd + option + v 変数化します。
cmd + option + n で戻します。

parameter hint name

若干物議をかもすこの機能。

こんな形でパラーメーター名を表示できます。

設定はここで切り替えられます。

ショートカット

効果	コマンド
Undo	cmd + z
Redo	cmd + shift + z
検索	cmd + f or cmd + shift + f (プロジェクト全体)
置換	cmd + r or cmd + shift + r (プロジェクト全体)
複製	cmd + d
プロジェクトにwindowを開く閉じる	cmd + 1
ファイル作成	cmd + n
コードの折りみと展開	cmd + .
preferenceを開く	cmd + ,
help or Action	cmd + shift + a
expand selection	option + ↑ (回数で変わる)
予測補完を出す(macだとデフォで行けないので後述。ここに書いたのは自分用メモ)	cmd + 4
smart補完を出す(macだとデフォで行けないので後述。ここに書いたのは自分用メモ)	cmd + 5
intention Actions	option + enter
変数化	cmd + option + v
インライン化	cmd + option + n
パラメーターヒント	cmd + p
定義元ジャンプ(とにかく素晴らしい)	cmd + b

簡単な使い方

1. ブラウザで⇒アクセス　Gooogle colabo のページ
2. （パラメータ変更して）実行ボタンを押す

---

内部の概略

Quantlibからよく使われるブラックショールズ式を簡単に呼び出せるラッパークラスを作成。

インストールする場合

pip install simple_option

主なクラス

Portfolio
Option
Payoff

(後ほど追記）

簡単な使い方

1. ブラウザで⇒アクセス　Gooogle colabo のページ
2. （パラメータ変更して）実行ボタンを押す

---

パラメータ変更例1：

#　３月限ショートストラングル
p = Portfolio(
"""
    03/C21000[-1]
    03/P20000[-1]
""")

#マーケット情報設定書式
#setting(原資産価格, IV(％）, 日付【yyyymmdd】)   
setting(20250, 25, 20190204)    

内部の概略

Quantlibからよく使われるブラックショールズ式を簡単に呼び出せるラッパークラスを作成。

インストールする場合

pip install simple_option

主なクラス

Portfolio
Option
Payoff

クラス利用例：

Example1
---------
from simpleOption import *
#Simple Example
o = Option('02/P20500')
op_price = o.v(20625, 20.8, 20190124)
print(f"{o}@{op_price:.2f}  (nk=20625,IV=20.8%)  jan24 ")
OUTPUT 1
---------
02/P20500@285.49  (nk=20625,IV=20.8%)  jan24
Example2
---------
#underlying change: 20625 >>20500
op_price2 = o.v(20500)
print(f"{o}@{op_price2:.2f} (nk=20500,IV=20.8%)  jan24")
OUTPUT 2
---------
02/P20500@285.49  (nk=20625,IV=20.8%)  jan24
Example3
---------
#underlying & IV change: 20625>>20000 &IV=25%
op_price3 = o.v(20000, 25)
print(f"{o}@{op_price3:.2f} (nk=20000,IV=25%)  jan24")
OUTPUT 3
---------
02/P20500@703.62 (nk=20000,IV=25%)  jan24
Example4
---------
#use keyword
op = Option('02/P20500')
op_price4 = op(
    underlying=20250,
    iv=25,
    evaluationDate=20190122
)
"""

transitionsはPythonで状態遷移を実現するためのパッケージですが、今回は状態遷移時に実施されるアクションを実現するために「コールバック」について紹介したいと思います。
この記事では状態遷移を実現するソフトウェア的な機構をステートマシンと呼んでいます。

この記事の対象者と今回の内容

Pythonで状態遷移を実装したり動作確認をしたい方に、Pythonの状態遷移パッケージ「transitions」の使い方を説明していきたいと思います。状態遷移そのものは組込みとか制御などでよく使われるものですが、それをPythonで実現したい場合にこのパッケージが有用かと思います。

状態遷移を実現するだけであれば状態と遷移を定義しステートマシンを作れますが、実際はトリガーイベントが発生した際や遷移時に何か処理をすることも多いです。こういったトリガーイベント起因による処理をアクションなどといい、ソフトウェア的にはコールバック(callback)として実装されることが多いかと思います。transitionsパッケージでもこのコールバックを実装できますので詳細について説明していきたいと思います。
この状態遷移におけるアクションは必須ではありませんが、状態遷移を使って制御機構を実装する場合に非常に重要ですのでtransitionsで状態遷移を実現する場合は当記事の内容を参考に頂けたらと思います。

その他、transitionsの概要やインストール方法、当記事で作成している状態遷移図といったグラフ表示機能の導入(GraphMachine)や設定については準備編の記事を参照頂けたらと思います。
※今回の記事でも公式チュートリアルにならい遷移をtransitionsと呼んでますが、パッケージ名のtransitionsとややこしいので、当パッケージそのものを示す場合はtransitionsパッケージと明示することにします。

transitionsにおけるコールバック

transitionsパッケージにおけるコールバックについて、簡単なサンプルコードを例に説明します。
まずはコールバックをもつステートマシンの定義になります。

コールバック定義のサンプル

from transitions import Machine

states = ['A', 'B']   #状態の定義
transitions = [
    {'trigger':'fromAtoB', 'source':'A', 'dest':'B', 'after':'action_after'},
    {'trigger':'fromBtoA', 'source':'B', 'dest':'A'},
]

class Model(object):
    # afterのコールバック（自メソッド名を表示するだけ）
    def action_after(self):
        print('do "{}"'.format(sys._getframe().f_code.co_name))

model = Model()

machine = Machine(model=model, states=states, transitions=transitions, initial=states[0], 
                  auto_transitions=False, ordered_transitions=False)

上記コードはAとBの2状態で、AからBの遷移にafterコールバックを定義したものとなります。

上記で定義したステートマシンに対し、トリガーイベントを起こし、コールバックを呼んでみます。

コールバック動作のサンプル

>>> model.state         # 初期状態の確認
'A'
>>> model.fromAtoB()    # トリガーイベントを起こす
do "action_after"
True
>>> model.state         # 遷移後状態の確認
'B'

遷移を図示すると以下の通りになります。
今回はこの青矢印の遷移が起こった時にコールバックが呼ばれ実行されたことになります。
トリガーイベント（fromAtoB）を起こし、その際にtransitionリスト内の定義である'after'に指定された'action_after'メソッドが呼ばれている事を確認できると思います。

このようにtransitionsパッケージにおけるコールバックは、(一部を除いて)基本的にトリガーイベントを起点として発生する事になります。
またコールバックに指定するものは、Machineの引数modelに指定するオブジェクト内のメソッド（例ではmodelオブジェクトのクラスメソッド）になります。

コールバックの種類

冒頭の例では非常に簡単なコールバックですが、transitionsパッケージで定義できるコールバックは大きく分けて3つになります。

• 遷移に紐付くコールバック（trantisions辞書で設定）
• 各状態に紐付くコールバック（Stateクラス/state辞書で設定）
• 全状態に紐付くコールバック（Machineクラス/machineオブジェクトで設定）

上記以外にtimeout等特殊なコールバックもあります。今回は順次それぞれについて詳細説明します。

遷移に紐付くコールバック

こちらは遷移編1と内容が被る部分がありますが再度内容を記しておきます。
遷移に紐付くコールバックは、その名の通り遷移それぞれに対し個別にコールバックを設定できます。従って「ある遷移にのみコールバックを設定したい場合」にこちらの定義方法を利用することになります。

遷移準備/遷移前/遷移後コールバックについて

まずは遷移に関する基本的なコールバックになります。
冒頭の例もこの遷移に紐付くコールバックであり、以下の通りtransitions辞書などで設定できます。

transitions = [
    {'trigger':'fromAtoB', 'source':'A', 'dest':'B', 'after':'action_after'},
]

上記はfromAtoBトリガーイベントが起こった際の状態Aから状態Bへの遷移定義ですが、afterキーがコールバック定義の一つで、afterキーに指定しているaction_afterがコールバックされるクラス内メソッドになります。このようにコールバックとして呼び出されるクラス内メソッド名を文字列で記載します。
なお遷移に紐付くコールバックはトリガーイベントが起こった際、遷移元（source）と遷移後（dest）どちらで行うかでいくつか種類が別れています。

項目	定義名	実施される状態	説明
遷移準備コールバック	prepare	遷移元 (souce)	triggerが発生した際に実行されるコールバック
遷移前コールバック	before	遷移元 (souce)	状態が遷移する前に呼び出されるコールバック
遷移後コールバック	after	遷移先 (dest )	状態が遷移した後に呼び出されるコールバック

冒頭の例ではafterのみしか設定していませんが、複数種類のコールバックを一度に定義することも可能です。
もちろんこれらprepareやbeforeコールバックを実行するクラス内メソッドをModelクラス内に定義しなければ実行エラーになりますので注意が必要です。

# prepare, before, afterを一つの遷移に設定した場合
transitions = [
    {'trigger':'fromAtoB', 'source':'A', 'dest':'B', 
     'prepare':'action_prepare', 'before':'action_before', 'after':'action_after'}  
]

上記を図示すると以下の通りになります。

この図は、状態AにおいてfromAtoBトリガーイベントを起こした直後になります。
状態Aにおいてトリガーイベントを起こしprepare, beforeが実施され、その後状態Bに移ってafterのコールバックが実施されることになります。

ガード判定について

UMLなどでもあるガードも実装することができ、transitionsパッケージにおいてガードは遷移に紐付くコールバックになります。（当記事ではガードを便宜上ガード判定と呼んでいます）
ガード判定は遷移を伴うトリガーイベントが起こったとしても、特定の条件を満たしていないと遷移をさせないというものになります。
この判定をコールバックとして実現しているのがtrantisionsパッケージのガード判定コールバックになります。

trantisionsパッケージにおけるガード判定の定義は以下二種があげられます。

項目	定義名	実施される状態	説明
ガード判定(True)	conditions	遷移元 (souce)	指定されたコールバックがTrue時に遷移を許可
ガード判定(False)	unless	遷移元 (souce)	指定されたコールバックがFalse時に遷移を許可

遷移編の焼き直しになりますが、ガード判定の簡単な例について示します。

遷移成立時(conditions)の定義例

import sys
from transitions import Machine

states = ['A', 'B']   # 状態の定義
transitions = [
    {'trigger':'fromAtoB', 'source':'A', 'dest':'B', 'conditions':'action_conditions'}
]

class Model(object):
    # ガード判定用コールバック(conditionsやunlessはbool型の戻り値が必要)
    def action_conditions(self):
        print('do "{}" on state ({})'.format(sys._getframe().f_code.co_name, self.state))
        return False            # conditionsではTrueを返すと遷移を許可する(unlessではFalse時に遷移を許可)

model = Model()
machine = Machine(model=model, states=states, transitions=transitions, initial=states[0], 
                  auto_transitions=False, ordered_transitions=False)

上記コード定義後、以下のコードにより動作させてみると、conditionsに設定したクラス内メソッドがコールバックされ実施されている事が分かります。
内部的な一連の動きとしては、トリガーイベント発生 → condisionsのコールバック実施 → コールバックよりFalseが返る → 遷移阻止という流れになります。

遷移成立時(conditions)の挙動

>>> model.state
'A'
>>> model.fromAtoB()    # トリガーイベントを起こしてみる
do "action_conditions" on state (A)
False
>>> model.state         # 状態はAのままで遷移できていない
'A'

上記conditionsの定義において、action_conditionsの戻り値をTrueにすると、状態AにおいてfromAtoBトリガーイベントが起きた際、通常どおり状態Bに遷移する事になります。

unlessについてはconditionsと論理が逆になるだけ（Falseを返した際に遷移を許可）で、基本的な動作は変わりません。
また、conditionsとunlessを同時設定することも可能で、その場合はconditionsの判定が行われた後unlessが実施され、conditionsで遷移不許可となった場合はunlessは実施されない事に注意ください。

なお、prepareとbeforeの役割や違い、実行順については遷移編1でも述べてますが、以下コールバックの優先順位にて全コールバックの優先順位と絡めて述べたいと思います。

各状態に紐付くコールバック

こちらは状態編と内容が被る部分がありますが、もう少し詳しく説明したいと思います。

enterとexitコールバックについて

当コールバック設定は状態定義（stateリスト）で行うものになります。まずはサンプルコードを示します。

各状態に紐付くコールバックの定義例

from transitions import Machine, State

#状態の定義
states = [State(name='A', on_exit=['action_on_exit']),  # Stateクラスで定義可能
          {'name':'B', 'on_enter':'action_on_enter'}]   # 辞書でも定義可能

transitions = {'trigger':'fromAtoB', 'source':'A', 'dest':'B'}

class Model(object):
    # on_enterのコールバック
    def action_on_enter(self):
        print('do "{}" on state ({})'.format(sys._getframe().f_code.co_name, self.state))

    # on_exitのコールバック
    def action_on_exit(self):
        print('do "{}" on state ({})'.format(sys._getframe().f_code.co_name, self.state))

model = Model()
machine = Machine(model=model, states=states, transitions=transitions, initial='A', 
                  auto_transitions=False, ordered_transitions=False)

状態のみを定義する場合は、冒頭のサンプルコードの通りstate=['A', 'B']と文字列のリストのみで良いですが、各状態に「入った際」や「出た際」にコールバックを起こす際は、以下のように設定します。

states = [State(name='A', on_exit=['action_on_exit']),  # Stateクラス版（Machine.Stateのimportが必要）
          {'name':'B', 'on_enter':'action_on_enter'}]   # 辞書版

上記サンプルコードに対し実際に動作させてみます。

状態に紐付くコールバックの動作例

>>> model.state
'A'
>>> model.fromAtoB()
do "action_on_exit" on state (A)
do "action_on_enter" on state (B)
True
>>> model.state
'B'

上記の結果を図示すると以下の通りになります。

fromAtoBトリガーイベントが発生し遷移が行われ状態Aから抜けます。この際状態Aに定義されたon_exitのコールバックaction_on_exitが実施されてます。
その後、状態Bに入り、状態Bに定義されたon_enterのコールバックaction_on_enterが実施されているという流れになります。

このように、on_enterは定義した状態に入った時、on_exitは定義した状態から出た時に実施されることになります。
こちらを図示すると以下のようなイメージとなります。

時間切れコールバックについて

こちらはちょっと特殊なコールバックになりますが、状態に紐付くコールバックになります。
ある状態に入って、その状態にtimeoutで指定した時間滞在すると実施されるコールバックになります。定義例を以下に示します。

時間切れコールバックの定義例

from time import sleep
from transitions import Machine
from transitions.extensions.states import add_state_features, Timeout

@add_state_features(Timeout)
class MachineWithTimeout(Machine):
    pass

states = ['A',
          Timeout(name='B', timeout=10, on_timeout='action_timeout')]
#上記は{'name': 'B', 'timeout': 10, 'on_timeout': 'action_timeout'}でも良い

class Model:
    # on_timeoutのコールバック
    def action_timeout(self):
        print('do "{}" on state ({})'.format(sys._getframe().f_code.co_name, self.state))

model= Model()
machine = MachineWithTimeout(model=model, states=states, initial='A', 
                             auto_transitions=False, ordered_transitions=True)

注意が必要なのはtimeoutはtransitionsパッケージの拡張機能になるので「transitions.extensions.states」でTimeoutクラスとadd_state_fuaturesのimportおよび定義が必要になります。
上記は状態Bに遷移後、状態Bの状態に10秒間滞在するとon_timeoutコールバックが呼ばれるものになります。実際に動作させてみます。

時間切れコールバックの動作例

>>> model.state
'A'
>>> model.next_state()
True
>>> model.state
'B'
>>> sleep(30)
do "action_timeout" on state (B)

このように、sleep中にtimeoutコールバックが発生している事がわかります（next_state実施後、状態Bに移って10秒後に発生）
なお、timeout時間が設定されているのにもかかわらず、on_timeoutコールバックが指定されていない場合はtimeout例外が発生します。

各状態に関するコールバックのまとめ

各状態に関するコールバックをまとめると以下のとおりになります。
あくまでもコールバックが行われるのは「各状態に関するコールバック」が定義された"状態"になります。

項目	定義名	説明
enterコールバック	on_enter	定義された状態に入った際に実施されるコールバック
exitコールバック	on_exit	定義された状態から出る際に実施されるコールバック
時間切れコールバック	on_timeout	定義された状態にtimeout時間経過した際に実施されるコールバック

全状態に紐付くコールバック

最後に全状態に紐付くコールバックになります。
動作としては各状態に紐付くコールバックに近いものになりますが、ここで設定したコールバックは全状態が対象になります。
以下サンプルを見てみましょう。

全状態に紐付くコールバックの定義例

from transitions import Machine

states = ['A', 'B']   #状態の定義

class Model(object):
    # prepare_eventのコールバック
    def action_prepare_event(self):
        print('do "{}" on state ({})'.format(sys._getframe().f_code.co_name, self.state))

    # before_state_changeのコールバック
    def action_before_state_change(self):
        print('do "{}" on state ({})'.format(sys._getframe().f_code.co_name, self.state))

    # after_state_changeのコールバック
    def action_after_state_change(self):
        print('do "{}" on state ({})'.format(sys._getframe().f_code.co_name, self.state))

    # finalize_eventのコールバック
    def action_finalize_event(self):
        print('do "{}" on state ({})'.format(sys._getframe().f_code.co_name, self.state))

model = Model()
machine = Machine(model=model, states=states, initial=states[0], 
                  auto_transitions=False, ordered_transitions=True,
                  prepare_event='action_prepare_event',
                  before_state_change='action_before_state_change',
                  after_state_change='action_after_state_change',
                  finalize_event='action_finalize_event')

上記定義はステートマシンとしては単純な状態Aと状態Bによる2状態の順序遷移になります。
全状態に紐付くコールバックであるprepare_event、before_state_change、after_state_change、finalize_eventの4つ全てを定義しています。

こちらを実際に動かしてみます。
以下のサンプルでは状態A→状態Bへ遷移し、その後状態B→状態Aへと遷移する動作例になります。

全状態に紐付くコールバックの動作例

>>> model.state             # 初期状態の確認
'A'
>>> model.next_state()      # トリガーイベント：状態A→状態Bへ
do "action_prepare_event" on state (A)
do "action_before_state_change" on state (A)
do "action_after_state_change" on state (B)
do "action_finalize_event" on state (B)
True
>>> model.state             # 遷移後状態の確認
'B'
>>> model.next_state()      # トリガーイベント：状態B→状態Aへ
do "action_prepare_event" on state (B)
do "action_before_state_change" on state (B)
do "action_after_state_change" on state (A)
do "action_finalize_event" on state (A)
True
>>> model.state             # 遷移後状態の確認
'A'

動作例を見ると分かる通り、stateやtransitionsのコールバック設定がされていないのにも関わらず、状態遷移が発生する度に登録されたコールバックが呼ばれている事が確認できると思います。
このように、全状態に関するコールバックは設定したコールバックが各状態に入った際や出た際などで起こりますので、各状態における共通の初期化などに使えると思います。

なお、prepare_eventとfinalize_eventは少々特殊で、こちらは遷移してもしなくても実施されるコールバックになります。
したがって、ガード判定などで遷移が阻害されたとしても、prepare_eventはトリガーイベントが発生したらすぐに、finalize_eventは最後に実施されるコールバックになります。
ただし、他のコールバック内で例外などが発生してしまった場合は実施されませんので、その点はご注意ください。

「全状態に紐付くコールバック」をまとめると以下の通りになります。

項目	定義名	実施される状態	説明
準備コールバック	prepare_event	遷移元 (souce)	triggerが発生した際に実行されるコールバック
状態変化前コールバック	before_state_change	遷移元 (souce)	状態が遷移する前に呼び出されるコールバック
状態変化後コールバック	after_state_change	遷移先 (dest )	状態が遷移した後に呼び出されるコールバック
最終コールバック	finalize_event	※	最後に呼び出されるコールバック

※遷移出来た際は遷移先（dest）にて、ガード判定で遷移が中断された場合は遷移元（source）にて実施される。

コールバックの優先順位

さて、コールバックの種類として、「遷移に紐付くコールバック」「各状態に紐付くコールバック」「全状態に紐付くコールバック」の3種類を紹介しましたが、もちろんこれらコールバックは全て合わせて定義可能です。
一方で各種コールバックを複数合わせた際に気になるのが実施順だと思いますが、実施順には明確な定義があります。
以下、実施順序について記します。
なんらかのトリガーイベントが発生した時、下表の上から順に実施されることを示しています。
また下表において「遷移阻止時も実施」列はunlessやconditionsにより遷移が中断された際にコールバックが実施されるかどうかを示しています。

定義名	定義先	状態	遷移阻止時も実施
prepare_event	machine	source	YES
prepare	transitions	source	YES
conditions	transitions	source	YES ※3
unless	transitions	source	YES ※3, ※4
before_state_change	machine	source	NO
before	transitions	source	NO
on_exit	state	source ※1	NO
--- 状態変化 ---
on_enter	state	dest ※2	NO
after	transitions	dest	NO
after_state_change	machine	dest	NO
finalize_event	machine	source/dest	YES ※5

※1：遷移元（source)のstateで定義されたon_exitコールバックが呼ばれる
※2：遷移先（dest）のstateで定義されたon_enterコールバックが呼ばれる
※3：conditionsとunlessが同時設定されていた場合、両者の遷移条件が成立していないと遷移しません。
※4：conditionsとunlessが同時設定されていた場合、conditionsで遷移阻止された場合はunlessも実施されません。
※5：遷移中断時は遷移元で実施され、on_exitの後に実施される。

全コールバックを定義した際のサンプルコード

これらを確認するためのサンプルコードを以下に示します。
少々長いですが、以下はtimeoutを除く全てのコールバックを定義してあります。
以下サンプルは状態Aのみで、トリガーイベントeventにより自己遷移（状態A → 状態A）するステートマシンになります

全コールバックの定義例（timeout除く）

from transitions import State, Machine

states = [State(name='A', on_exit='action_on_exit', on_enter='action_on_enter')]
transitions = {'trigger':'event', 'source':'A', 'dest':'=', 
               'prepare'    : 'action_prepare',
               'conditions' : 'action_conditions',
               'unless'     : 'action_unless',
               'before'     : 'action_before',
               'after'      : 'action_after'}

class Model(object):
    # prepare_event(全遷移に紐付く)のコールバック
    def action_prepare_event(self):
        print('do "{}"'.format(sys._getframe().f_code.co_name))

    # prepare(遷移に紐付く)のコールバック
    def action_prepare(self):
        print('do "{}"'.format(sys._getframe().f_code.co_name))

    # conditions(遷移に紐付く)のコールバック
    def action_conditions(self):
        print('do "{}"'.format(sys._getframe().f_code.co_name))
        return True   #遷移を許可

    # unless(遷移に紐付く)のコールバック
    def action_unless(self):
        print('do "{}"'.format(sys._getframe().f_code.co_name))
        return False  #遷移を許可

    # before_state_change(全遷移に紐付く)のコールバック
    def action_before_state_change(self):
        print('do "{}"'.format(sys._getframe().f_code.co_name))

    # before(遷移に紐付く)のコールバック
    def action_before(self):
        print('do "{}"'.format(sys._getframe().f_code.co_name))

    # on_exit(状態に紐付く)のコールバック
    def action_on_exit(self):
        print('do "{}"'.format(sys._getframe().f_code.co_name))

    ### --- change state ---
    # on_enter(状態に紐付く)のコールバック
    def action_on_enter(self):
        print('do "{}"'.format(sys._getframe().f_code.co_name))

    # after(遷移に紐付く)のコールバック
    def action_after(self):
        print('do "{}"'.format(sys._getframe().f_code.co_name))

    # after_state_change(全遷移に紐付く)のコールバック
    def action_after_state_change(self):
        print('do "{}"'.format(sys._getframe().f_code.co_name))

    # finalize_event(全遷移に紐付く)のコールバック
    def action_finalize_event(self):
        print('do "{}"'.format(sys._getframe().f_code.co_name)) 

#ファイル出力する場合はMatter('test')等ファイル名指定する, Notebook上で表示する場合は引数に指定なし
model = Model()
machine = Machine(model=model, states=states, transitions=transitions, initial=states[0], 
                       auto_transitions=False, ordered_transitions=False)

# 全状態に紐付くコールバック定義の追加
machine.prepare_event = 'action_prepare_event'
machine.before_state_change = 'action_before_state_change'
machine.after_state_change = 'action_after_state_change'
machine.finalize_event = 'action_finalize_event'

こちらを動作させると以下の通りになります。

全コールバックの動作例（timeout除く）

>>> model.event()   #トリガーイベント：状態A→状態A
do "action_prepare_event"
do "action_prepare"
do "action_conditions"
do "action_unless"
do "action_before_state_change"
do "action_before"
do "action_on_exit"
do "action_on_enter"
do "action_after"
do "action_after_state_change"
do "action_finalize_event"
True

なお、もしもconditionsやunlessにより遷移が中断された場合は遷移や状態が変わった際のコールバックは実施されませんが、準備関連のコールバックや最終コールバックなどは実施されます。
以下にガード判定unlessでTrueを返し、遷移を阻止した場合の動作例を示します。

unlessで阻止（Trueを返した）際の動作例

>>> model.event()   #トリガーイベント：状態A→状態A
do "action_prepare_event"
do "action_prepare"
do "action_conditions"
do "action_unless"
do "action_finalize_event"
False

このようにunlessにより遷移が阻止されても準備系(prepareやprepare_event）や最終コールバック(finalize_event)が実施されている事に注意下さい。

各コールバックの使い分け

これまでで各種コールバックの実施順やタイミングなどは理解できたと思います。
色々とコールバックの設定があって大変ですが、

• 遷移それぞれにコールバックを設定したい場合は「遷移に紐付くコールバック」
• ある状態に入ったり出ていった際に実施したい場合は「状態に紐付くコールバック」
• 全状態に共通に処理を行わせたい場合は「全状態に紐付くコールバック」

を定義してやれば良いことになります。

基本的に制御関連では「ある状態において、あるイベントが起きた時、あるアクションが起きて別の状態に遷移する」という風な目線で設計することが多いと思いますので、大多数は「遷移に紐付くコールバック」を設計していけば良いと思いますが、設計していく中で、ある状態に入ったり出た際の初期化は「状態に紐付くコールバック」を、状態が変わるごとに初期化をしたい場合は「全状態に紐付くコールバック」をする場合が多いとは思います。
また「遷移に紐付くコールバック」を設計していて共通化できるものは、各状態や全状態に紐付くコールバックにした方がコールバックの見通しは良くなるので、全てのコールバックを「遷移に紐付くコールバック」として実装するのではなく、通常のソフトウェア設計と同じくコールバックが実施される範囲や共通化といった事を意識して定義することをオススメします。

prepare (prepare_event)やfinalize_eventの使いどころ

遷移阻止によらず実施されるコールバックとして、prepare（遷移に紐付く）、prepare_event（全状態に紐付く）、finalize_event（全状態に紐付く）がありますが、これらは使いどころが難しいかもしれません。
prepare（遷移に紐付くコールバック）は、特にガード判定の前処理として使う事をオススメします。
後のコールバック編2で紹介しますが、例えば「prepareコールバックでデータを受け取ったりして前処理を行い、前処理の結果、遷移しても良いと判定したら遷移許可を出す（conditionsコールバックの戻り値がTrueになるように処理する）」といった使い方も出来ると思います。特に遷移前のコールバックであるbeforeやbefore_state_changeは「遷移が確定しないと実施されない」ので、これらガード判定の前処理としては使えず注意が必要です。

全遷移に対しトリガーイベントが発生した際に必ず実施されるfinalize_eventは、個人的には遷移結果の表示やGraphMachineを用いた画像出力などをこちらのコールバックとしてやらせることが多いです。

まとめ

今回はtransitionsパッケージで設定出来るコールバックの種類や定義方法、および実施順序などについて紹介しました。
冒頭でも述べたとおり、コールバックはステートマシンに必須の機能ではありませんがコールバックなしには状態遷移で制御を行う事は難しくなるので、是非活用してもらえればと思います。その際、当記事が少しでもお役に立てたら幸いです。

今回も少々長くなってしまったので、一旦コールバック編をここで区切ります。次はコールバック編2としてコールバックメソッドへのデータの渡し方や、コールバックのキューといった内容について紹介する予定です。

これの続きです

深層強化学習 PyTorchによる実践プログラミング の6章の内容です。

前回の Deep Q-Netowork の発展版として、Double DQN と Dueling Network の実装が紹介されていました。

Double DQN の結果がこれ。

Dueling Network の結果がこれ。

この課題だともう学習後の上手さの違いなどはよくわからない。上記の本では Dueling Network を使うと少ない試行数でも学習が進むという話が書いてあったけどそのような傾向は見られなかった。

Double DQN

Main Q-Network: 次のステップで最大のQ値を持つようなaction、$a_m$を求めるネットワーク
Target Q-Network: $a_m$ の Q 値を評価するネットワーク
行動の決定と評価をそれぞれ別のネットワーク (ただし構造は同一) で行うことで学習を安定化させられるらしい。学習自体は Main Q-Network に対して行うが、たまに (下記の例では 2 エピソードに 1 回) Main Q-Network の weight を Target Q-Network にコピーしている。

Dueling Network

行動価値関数 $ Q(s, a) $ には、$s$だけで決まってしまう要素 $V(s)$ と行動次第で決まる要素 $A(s,a)$ があると考える。例えば cartpole では棒がもう倒れそうな状態だったらそこから右に押そうが左に押そうがあまり関係ない、など。そこで Q-Network の出力を $V(s)$ を出力する部分と各行動に対する $A(s,a)$ を出力する部分に分岐させ、$V(s) + A(s,a) = Q(s, a)$ として Q 値を求める。

DDQN.py

import numpy as np
import matplotlib.pyplot as plt
%matplotlib inline
import gym

from JSAnimation.IPython_display import display_animation
from matplotlib import animation 
from IPython.display import display

def display_frames_as_gif(frames):
    plt.figure(figsize=(frames[0].shape[1]/72.0, frames[0].shape[0]/72.0), dpi=72)
    patch = plt.imshow(frames[0])
    plt.axis('off')

    def animate(i):
        patch.set_data(frames[i])

    anim = animation.FuncAnimation(plt.gcf(), animate, frames=len(frames), interval=50)
    anim.save('movie_cart_ple_ddqn.mp4')
    display(display_animation(anim, default_mode='loop'))

from collections import namedtuple

Transition = namedtuple('Transition', ('state', 'action', 'next_state', 'reward'))

# 定数の設定
ENV = 'CartPole-v0'
GAMMA = 0.99
MAX_STEPS = 200
NUM_EPISODES = 500


# 経験を保存するメモリクラスを定義します。

class ReplayMemory:

    def __init__(self, CAPACITY):
        self.capacity = CAPACITY # メモリの最大長さ
        self.memory = []
        self.index = 0

    def push(self, state, action, state_next, reward):
        if len(self.memory) < self.capacity:
            self.memory.append(None) #メモリが満タンじゃないときには追加

        self.memory[self.index] = Transition(state, action, state_next, reward)
        self.index = (self.index + 1) % self.capacity

    def sample(self, batch_size):
        return random.sample(self.memory, batch_size)

    def __len__(self):
        return len(self.memory)

# ニューラルネットワークの定義します。
import torch.nn as nn
import torch.nn.functional as F

class Net(nn.Module):
    def __init__(self, n_in, n_mid, n_out):
        super(Net, self).__init__()
        self.fc1 = nn.Linear(n_in, n_mid)
        self.fc2 = nn.Linear(n_mid, n_mid)
        self.fc3 = nn.Linear(n_mid, n_out)

    def forward(self, x):
        h1 = F.relu(self.fc1(x))
        h2 = F.relu(self.fc2(h1))
        output = self.fc3(h2)
        return output

import random
import torch
from torch import nn
from torch import optim
import torch.nn.functional as F

BATCH_SIZE = 32
CAPACITY = 10000

# 行動決定を行うためのクラスです。
class Brain:
    def __init__(self, num_states, num_actions):
        self.num_actions = num_actions

        # メモリオブジェクトの生成
        self.memory = ReplayMemory(CAPACITY)

        # ニューラルネットワークの構築
        n_in, n_mid, n_out = num_states, 32, num_actions
        self.main_q_network = Net(n_in, n_mid, n_out)
        self.target_q_network= Net(n_in, n_mid, n_out)

        print(self.main_q_network)

        # オプティマイザの設定
        self.optimizer = optim.Adam(self.main_q_network.parameters(), lr=0.0001)

    def replay(self):
        '''Experience Replay'''

        # メモリサイズの確認
        # メモリサイズがミニバッチサイズより小さい間は何もしない。
        if len(self.memory) < BATCH_SIZE:
            return

        # ミニバッチの作成        
        self.make_minibatch()

        # 教師信号Q(s_t, a_t)を求める
        self.expected_state_action_values = self.get_expected_state_action_values()

        # ネットワークのパラメータ更新
        self.update_main_q_network()

    def decide_action(self, state, episode):
        '''state に応じて行動を決定する関数'''
        epsilon = 0.5 * (1 / (episode + 1))

        if epsilon <= np.random.uniform(0, 1):
            self.main_q_network.eval()
            with torch.no_grad():
                action = self.main_q_network(state).max(1)[1].view(1, 1)
        else:
            action = torch.LongTensor([[random.randrange(self.num_actions)]])

        return action

    def make_minibatch(self):
        # ミニバッチの作成
        transitions = self.memory.sample(BATCH_SIZE)

        batch = Transition(*zip(*transitions))
        self.batch = batch
        self.state_batch = torch.cat(batch.state)
        self.action_batch = torch.cat(batch.action)
        self.reward_batch = torch.cat(batch.reward)
        self.non_final_next_states = torch.cat([s for s in batch.next_state if s is not None])

    def get_expected_state_action_values(self):

        self.main_q_network.eval()
        self.target_q_network.eval()

        # state_batchをモデルに与え、推論結果からaction_batchで行った行動に対応するQ値を取ってくる
        # つまりあるstateにおける行ったactionの価値をとってきている。
        self.state_action_value = self.main_q_network(self.state_batch).gather(1, self.action_batch)

        # max{Q(s_t+1, a)}を求める
        non_final_mask = torch.ByteTensor(
        tuple(map(lambda s: s is not None, self.batch.next_state)))

        next_state_values = torch.zeros(BATCH_SIZE)

        a_m = torch.zeros(BATCH_SIZE).type(torch.LongTensor)

        # 次の状態での最大Q値のa_mをmain_q_networkから求める
        a_m[non_final_mask] = self.main_q_network(self.non_final_next_states).detach().max(1)[1]

        # 次の状態があるものだけをフィルターし、sizeをBATCH_SIZEからBATCH_SIZE*1へ
        a_m_non_final_next_states = a_m[non_final_mask].view(-1, 1)

        #行動a_mのQ値をtarget_q_networkで推定する。
        next_state_values[non_final_mask] = self.target_q_network(self.non_final_next_states).gather(
            1, a_m_non_final_next_states).detach().squeeze()

        expected_state_action_values = self.reward_batch + GAMMA * next_state_values

        return expected_state_action_values

    def update_main_q_network(self):
        # main_q_networkのネットワークパラメータの更新
        self.main_q_network.train()
        loss = F.smooth_l1_loss(self.state_action_value, self.expected_state_action_values.unsqueeze(1))
        self.optimizer.zero_grad()
        loss.backward()
        self.optimizer.step()

    def update_target_q_network(self):
        #target_q_network のパラメータをmain_q_networkと同じにする。
        self.target_q_network.load_state_dict(self.main_q_network.state_dict())

# エージェントを定義するクラスです。
class Agent:
    def __init__(self, num_states, num_actions):
        self.brain = Brain(num_states, num_actions)

    def update_q_function(self):
        self.brain.replay()

    def get_action(self, state, episode):
        action = self.brain.decide_action(state, episode)
        return action

    def memorize(self, state, action, state_next, reward):
        self.brain.memory.push(state, action, state_next, reward)

    def update_target_q_function(self):
        self.brain.update_target_q_network()

# 環境を定義するクラスです。
class Environment:
    def __init__(self):
        self.env = gym.make(ENV)
        self.num_states = self.env.observation_space.shape[0]
        self.num_actions = self.env.action_space.n

        self.agent = Agent(self.num_states, self.num_actions)

    def run(self):

        episode_10_list = np.zeros(10) #直近10エピソードで振り子が立ち続けたステップ数を記録

        complete_episodes = 0
        episode_final = False
        frames = []

        for episode in range(NUM_EPISODES):
            observation = self.env.reset()

            state = observation
            state = torch.from_numpy(state).type(torch.FloatTensor)
            state = torch.unsqueeze(state, 0)

            for step in range(MAX_STEPS):

                if episode_final is True:
                    frames.append(self.env.render(mode='rgb_array'))

                action = self.agent.get_action(state, episode)

                #行動actionの実行によってs_t+1とdoneフラグを取得
                observation_next, _, done, _ = self.env.step(action.item())

                if done:
                    state_next = None

                    episode_10_list = np.hstack((episode_10_list[1:], step + 1))

                    if step < 195:
                        reward = torch.FloatTensor([-1.0]) # 195ステップ未満で倒れたら報酬-1
                        complete_episodes = 0
                    else:
                        reward = torch.FloatTensor([1.0])
                        complete_episodes = complete_episodes + 1

                else:
                    reward = torch.FloatTensor([0.0])
                    state_next = observation_next
                    state_next = torch.from_numpy(state_next).type(torch.FloatTensor)
                    state_next = torch.unsqueeze(state_next, 0)

                self.agent.memorize(state, action, state_next, reward)
                self.agent.update_q_function()

                state = state_next

                if done:
                    print("%d episode: Finished after %d steps: 10試行の平均step数 = %.lf"%(
                    episode, step + 1, episode_10_list.mean()))

                    if episode % 2 == 0:
                        self.agent.update_target_q_function()

                    observation = self.env.reset()
                    break

            if episode_final is True:
                display_frames_as_gif(frames)
                break

            if complete_episodes >= 10:
                print("10回連続成功")
                episode_final = True


cartpole_env = Environment()
cartpole_env.run()

Dueling Network は Double DQN のネットワーク構造を以下のように変えるだけでよい。

DuelingNetwork.py

class Net(nn.Module):
    def __init__(self, n_in, n_mid, n_out):
        super(Net, self).__init__()
        self.fc1 = nn.Linear(n_in, n_mid)
        self.fc2 = nn.Linear(n_mid, n_mid)
        self.fc3_adv = nn.Linear(n_mid, n_out)
        self.fc3_v = nn.Linear(n_mid, 1)

    def forward(self, x):
        h1 = F.relu(self.fc1(x))
        h2 = F.relu(self.fc2(h1))
        adv = self.fc3_adv(h2)
        val = self.fc3_v(h2).expand(-1, adv.size(1))
        output = val + adv - adv.mean(1, keepdim=True).expand(-1, adv.size(1))

        return output

環境構築はさくっとできます
※Python 3.7.2時点

インストール

下記ページから
https://www.python.org/downloads/windows/

今回は「Windows x86-64 executable installer」でexeをダウンロード
※個人のPCによるので32bitの人は「Windows x86 executable installer」の方

ダウンロードしたらexe起動して「Install Now」をポチってインストールするだけ
この時に下の方にある「Add Python 3.7 to PATH」をチェックしてインストールすると後程やろうとしてる環境変数に追加してくれるみたいです（僕はやってないけど）

環境変数に追加

下記2つをを環境変数の「Path」に追加
~\Python\Python37\
~\Python\Python37\Scripts\

~は各自Pythonをインストールしたパス

コマンドで確認

>pip

Commands:
…
ってコマンドリストが出てきたらPythonを使用できる準備完了！

前回の記事

前回の記事→とにかくわかりやすく！Djangoでアプリ開発！ーその2ー

本記事の目的

python初心者の方が、本記事を見たあとに、一人でアプリ開発できることを目的にしております。
※インストールや開発環境については記載しません

環境

macOSX　Sierra
python3.7
django 2.1.5

前回まで

プロジェクトを立ち上げ(startproject)
→アプリの作成(startapp)
→view.pyを変更してレスポンスを書く
→urls.pyを修正する
→アプリの登録する
→index.html作る
→views.pyを直す
→htmlに変数入れる
→views.pyを直す
→複数ページ作るためにリンクつける
→views.pyを直す
→cssで装飾できるようにする

とここまででした。

ここからはアプリらしくフォームを作成していきます。

フォームを作る

目に見える画面に用意するものなので、templatesのindex.htmlを修正します。
前回同様にformというURLへの書き出しを行います。このformは後ほどurlpatternsに追加します。
またCSRF対策タグもつけておきました。これでフォーム送信時に、このトークンも一緒に受け渡されて、チェックが行われるので安心です。
※一旦不要なsecondは消します。

index.html

{% load static %}
<!doctype html>
<html lang="ja">
    <head>
        <meta charset="utf-8">
        <title>{{title}}</title>
        <link rel="stylesheet" type="text/css"
        href="{% static 'app1/css/style.css' %}"/>
    </head>
    <body>
        <h1>{{title}}</h1>　　　
        <p>{{msg}}</p>
        <form action= "{% url 'form' %}" method = "post"> <!--ここのフォーム使うとfromに飛びます-->    
            {% csrf_token %} <!--CSRF対策-->    
            <label for = "msg">ここに入力：</label> <!--入力項目ラベル-->    
            <input id = "msg" type="text" name ="msg"> <!--入力フォーム-->    
            <input type="submit" value="入力"> <!--ボタン-->    
        </form>
    </body>
</html>

htmlを変えたので,それを制御するviews.pyの変更も行います。
form関数を作ります。htmlのinputの部分に書いてある通り、ここで送信されたmsgと名付けられたtextを受け取るために、POSTメソッドを利用し送信された値を取り出します。
※一旦不要なsecondは消します。

views.py

from django.shortcuts import render
from django.http import HttpResponse

def aisatsu(request):
    params = {
        'title':'Hello World',
        'msg':'名前を入力してください',
    }
    return render(request,'app1/index.html', params)


def form(request):
    msg = request.POST['msg']
    params = {
        'title':'Hello World',
        'msg':'hello '+msg+'!',
    }
    return render(request,'app1/index.html', params)

新しい関数を定義したので、最後にこの関数を呼び出せるように、URLを準備します。urls.pyに追加を行います。　ここに追加することでhtmlのタグをつけるようになります。

urls.py

from django.urls import path
from . import views

urlpatterns = [
    path("",　views.aisatsu,　name=　"aisatsu"),
    path("form",views.form,　name="form"),
]

それではhttp://127.0.0.1:8000/app1/にアクセスして確認します。
※runserverしておいてください。キャッシュによって変更されない可能性があるので、その時はサーバーを止めてから、再度起動してください

以下のようにフォームができています。

入力して、送信！画面が以下のようになればOK！

今後のために

実はこの使い方は、せっかく入力された名前がクリアになってしまうので、DB登録とのチェック機能やできないなどあまり汎用性がありません。なのでdjangoに用意されているFormクラスを使います。※最初からそれを記載してもよかったのですが...念のため...

ということでまずはapp1フォルダ配下にforms.pyを作成し、以下のようにします。
書いてあるまんまですが、テキスト(char)と整数(int)の入力フィールドを用意して、labelで名前をつけました。大文字小文字に気をつけてください。

apps/forms.py

from django import forms

class aisatsuform(forms.Form):
    name = forms.CharField(label="name")
    area = forms.CharField(label="area")
    age = forms.IntegerField(label="age")

このフォームを利用できるような制御にするために、views.pyを修正していきます。
パラメタの中に、formを作ってインスタンスを入れています。
下の部分では、request.methodで、リクエストがPOSTかどうかチェックをして、そうだったらPOST用の処理を実施しします。GETならば、ページだけ返します。
※POSTとかGET:HTTPメソッド。フォームを使うならPOSTを利用します。他にもいくつかメソッドがありますが、メソッドを見れば、どんなことが行われているかわかるようになっています。
*波括弧の閉じる位置とreturnのインデントに注意

views.py

from django.shortcuts import render
from django.http import HttpResponse
from .forms import AisatsuForm

def aisatsu(request):
    params = {
        'title':'Hello World',
        'msg':'ちゃんと挨拶したいので情報の登録をしてください',
        'form': AisatsuForm(),
    }
    if (request.method=='POST'):
        params['msg'] = 'こんにちは!'+request.POST['name']+'さん!<br>'+request.POST['area']+'にお住まいで<br>年齢は'+request.POST['age']+'歳なんですね!<br>よろしくお願いします。'
        params['form']= AisatsuForm(request.POST)
    return render(request,'app1/index.html', params)

少し複雑ですね。冒頭でaisatsuformのインスタンスをformに代入し、もしPOSTリクエストだったら、もう一回POSTの内容でインスタンスを作って上書きを行なっています。同様にmsgも書き換えを行い、POSTされた情報を表示させるようにしています。ここでは<br>などのhtmlタグが埋め込まれているので、index.htmlで表示させる時に、そうなっていることを知らせなくてはなりません。下記で記述します。

それでは、画面表示直していきます。index.htmlを修正します。
先ほど作ったフォームは丸ごと置き換えます。
{{form}}のところをテーブル(as_table)にしたり、リスト(as_ul)にすることもできます。何も指定しないと横一列にフィールドが設置されます。※tableにする場合は

タグが必要です。

index.html

{% load static %}
<!doctype html>
<html lang="ja">
    <head>
        <meta charset="utf-8">
        <title>{{title}}</title>
        <link rel="stylesheet" type="text/css"
        href="{% static 'app1/css/style.css' %}"/>
    </head>
    <body>
        <h1>{{title}}</h1>　　　
        <p>{{msg|safe}}</p><!--ここでviews.pyのhtmlを有効化--> 
        <form action= "{% url 'aisatsu' %}" method = "post"> 
            {% csrf_token %} 
            {{form.as_ul}} <!--ここ-->    
            <input type="submit" value="入力"> 
        </form>
    </body>
</html>

ここまできたら、http://127.0.0.1:8000/app1/
にアクセスしてみます。すごい怪しい感じになりました。

情報を送信してみるとこんな感じ

この記事はここまで

次回は続きを投稿していきます。そろそろDBを使います。

環境　Windows Python3.7

これからやること

Beautiful Soup 4（以後bs4）を使ってスクレイピング。結果をJSONファイルで出力。
WEBから分析用のデータを収集する。
データ分析の勉強をしたかったが、そもそも良い感じのデータが無かったためスクレイピングで入手することに。

スクレイピング対象

• ニコニコチャート（http://www.nicochart.jp/)
• ニコニコ動画(https://www.nicovideo.jp/)

ニコニコ動画ゲームカテゴリランキングの上位100に入った動画と投稿者の情報を取得する。

Q. なぜニコニコ動画？
A. 私も投稿者だからです（宣伝）。https://www.nicovideo.jp/user/35233530

取得したい情報

• 動画情報
  • 動画ID
  • 投稿日時
  • 再生時間
  • 過去最高ランク
  • タグ
• 投稿者情報
  • 投稿者ID
  • ユーザー登録時のニコニコ動画のバージョン
  • フォロワー数
  • 投稿動画数

Beautiful Soupについて

事前準備

BeautifulSoupをインストール

$ pip install beautifulsoup4

基本的な使い方

import urllib.request
from bs4 import BeautifulSoup

url="対象のURL"
html = urllib.request.urlopen(url)
soup = BeautifulSoup(html, "html.parser")

#引数はどれか1つ以上でOK。最初に見つかったタグが返される。
hoge = soup.find("HTMLタグ名",class_="タグのクラス名",id="タグのID") 
#条件に一致する全てのタグが返される。
hoge = soup.find_all("HTMLタグ名",class_="タグのクラス名",id="タグのID") 
#あるタグの子要素に限定して検索することも可能。
hogehoge=hoge.find("HTMLタグ名",class_="タグのクラス名",id="タグのID")
#タグのテキストを取得
print(hoge.text)
#タグの属性値を取得
print(hoge["href"]) 

エラー対策

ネットワークエラーの対策。
503エラーが発生することがあるので。
時間を空けてリトライ。それでもダメなら諦め。

try:
    html = urllib.request.urlopen(_Url)
except urllib.error.HTTPError as e:
    print(e.code)
    sleep(60)
    html = urllib.request.urlopen(_Url)
soup = BeautifulSoup(html, "html.parser")

タグが見つからないエラーの対策。
HTMLが上手く読めてない可能性があるので時間を空けてリトライ。
それでもダメなら諦め。

try:
    tag_panel=soup.find(class_="クラス名1").find(class_="クラス名2")
except:
    try:
        print("error")
        sleep(60)
        #再読み込み
        html = urllib.request.urlopen(_Url)
        soup = BeautifulSoup(html, "html.parser")
        tag_panel=soup.find(class_="クラス名1").find(class_="クラス名2")
        result=tag_panel.text
    except:
        print("error")
        result = None

スクレイピングの注意点

• サイトの規約を読みましょう。
• 相手のサーバーに負担をかけるため、sleep等でリクエストの間隔を空けましょう。1秒以上は空けたほうが良さげ。

JSON出力

辞書型からJSONファイルで出力。エンコードはUTF-8で出力。Shift_Jisは文字化けしたり何かと厄介。
ensure_ascii=Falseも忘れずに。
ほんとはwith句でやると安全。

fw = open("ファイル名",'w',encoding='utf-8')
json.dump(_dictionary,fw,indent=4,ensure_ascii=False)
fw.close()

実際にやってみた

エラー対策は頻発する箇所のみ実施。

#%%
import urllib.request
import re
import datetime
import json
import sys
from bs4 import BeautifulSoup
from time import sleep 

#ニコニコチャートからスクレイピング
###ランキングページを開く
def setRankPage(_dateStr):
    url="http://www.nicochart.jp/ranking/game/"+_dateStr
    html = urllib.request.urlopen(url)
    print(url)
    soup = BeautifulSoup(html, "html.parser")
    return soup

##動画IDと動画URLと投稿者名、IDとURLの取得
def getRankInfo(_soup,_num):
    rank=_soup.find(id="rank"+str(_num))

    link=rank.find(class_="title").find("a")
    infoUrl="http://www.nicochart.jp/"+str(link["href"])
    videoId=re.findall('\d+',link["href"])[0]
    print(infoUrl,videoId)

    ##投稿者名とURLの取得
    nameInfo=rank.find("em",class_="user")
    if not nameInfo:
        return {"videoId":videoId,"videoUrl":infoUrl,"userName":None,"userUrl":None}
    nameInfo=nameInfo.find("a")
    nameUrl=str(nameInfo["href"])
    userId=re.findall('\d+',nameUrl)[0]
    nameUrl += "/video"
    name=rank.find("em",class_="name").find("a").text
    print(name,nameUrl)

    return {"videoId":videoId,"videoUrl":infoUrl,"userName":name,"userId":userId,"userUrl":nameUrl}

###動画URLからタグ、投稿日時、最高ランク、再生時間の取得
def getVideoInfo(_infoUrl):
    try:
        html = urllib.request.urlopen(_infoUrl)
    except urllib.error.HTTPError as e:
        print(e.code)
        sleep(60)
        html = urllib.request.urlopen(_infoUrl)
    soup = BeautifulSoup(html, "html.parser")

    ##タグの取得
    try:
        tag_panel=soup.find(class_="tab-panel").find(class_="cloud")
        tags=tag_panel.find_all("a",class_="word")
        tagWords=list(map(lambda tag: tag.text, tags))
    except:
        try:
            print("error")
            sleep(60)
            html = urllib.request.urlopen(_infoUrl)
            soup = BeautifulSoup(html, "html.parser")
            tag_panel=soup.find(class_="tab-panel").find(class_="cloud")
            tags=tag_panel.find_all("a",class_="word")
            tagWords=list(map(lambda tag: tag.text, tags))
        except:
            print("error")
            tagWords=None
    print(tagWords)

    ##投稿日時の取得
    date=soup.find("dd",class_="first-retrieve").find("span").text

    ##再生時間の取得
    length=soup.find("span",class_="length").text

    ##最高ランクの取得
    nicoRank=soup.find(class_="point-data").find_all("td",class_="fav")
    nicoRank=list(map(lambda rank:rank.find_all("a"),nicoRank))
    nicoRank=list(map(lambda rank:sys.maxsize if len(rank)==0 else int(rank[len(rank)-1].text),nicoRank))
    nicoRank=min(nicoRank)
    print(nicoRank)
    return {"tags":tagWords,"date":date,"length":length,"maxRank":nicoRank}

###投稿者情報から投稿者の基本情報を取得
#投稿者URLからフォロワー数、投稿動画数、ユーザーID、ニコ動Verを取得。
def getUserInfo(_nameUrl):
    try:
        html = urllib.request.urlopen(_nameUrl)
    except urllib.error.HTTPError as e:
        print(e.code)
        return {"followerNum":None,"videoNum":None,"nicoVer":None}
    soup = BeautifulSoup(html, "html.parser")

    ##基本情報の取得
    followerNum=soup.find(class_="stats channel_open_mb8").find_all(class_="num")[1].text
    followerNum=followerNum.replace(',',"")
    videoNum=soup.find(class_="content",id="video").find("h3").text
    videoNum=re.findall('\d+', videoNum)
    if videoNum:
        videoNum=videoNum[0]
    else:
        videoNum=None
    idInfo=soup.find(class_="accountNumber").find("span").text
    nicoVer=re.findall('(?<=\().*?(?=\))',idInfo)[0]

    print(followerNum,videoNum,nicoVer)
    return {"followerNum":followerNum,"videoNum":videoNum,"nicoVer":nicoVer}

###ランキングから動画URLと投稿者情報の取得
def getRankDataDay(_dateStr):  

    soup=setRankPage(_dateStr)
    rankData=[]
    for rankNum in range(1,100):
        print("//////",rankNum,"//////")
        rankData.append(getRankInfo(soup,rankNum))
    print(rankData)

    ###動画一覧、投稿者一覧の作成
    videoData={}
    userData={}
    user_videoData={}
    num=0
    for rank in rankData:
        num +=1
        videoId=rank["videoId"]
        videoData[videoId]={"videoUrl":rank["videoUrl"]}

        if not "userId" in rank:
            continue
        userId=rank["userId"]
        user_videoData[videoId]={"userId":rank["userId"]}
        if not userId in userData:
            userData[userId]={"userName":rank["userName"],"userUrl":rank["userUrl"]}


    ###動画一覧にタグ情報を追加
    for video in videoData.values():
        print("/////////",video["videoUrl"])
        videoInfo=getVideoInfo(video["videoUrl"])
        video.update(videoInfo)
        sleep(2)

    ###投稿者一覧に投稿者情報を追加
    for user in userData.values():
        userInfo=getUserInfo(user["userUrl"])
        user.update(userInfo)
        sleep(1)

    print(videoData)
    print(userData)
    return {"videoData":videoData,"userData":userData,"user_videoData":user_videoData}

###Json出力
def outputJson(_dateStr,_data):
    fw = open("VideoData/"+ _dateStr +".json",'w',encoding='utf-8')
    json.dump(_data,fw,indent=4,ensure_ascii=False)
    fw.close()
    return

###指定期間分のデータを取得。指定日の前日から取得開始。
date = datetime.date(2018,10,19)
count=200

for num in range(count):
    date -=datetime.timedelta(days=1)
    dateStr=date.strftime("%Y%m%d")
    rankDataDay=getRankDataDay(dateStr)
    outputJson(dateStr,rankDataDay["videoData"])
    outputJson(dateStr,rankDataDay["userData"])
    outputJson(dateStr,rankDataDay["user_videoData"])

結果

データ収集が完了。JSON形式で出力。
次回はこのデータをSQLデータベースにｼｭｳｳｳｳｳｳ!!!!超!!!ｴｷｻｲﾃｨﾝ!!!!!。

参考
https://www.crummy.com/software/BeautifulSoup/bs4/doc/#installing-beautiful-soup

1 動機

yahooさんのapiを使ってのキーワード抽出をしたいと思ったがサンプルコードがPHPしかなかった。
(https://developer.yahoo.co.jp/sample/jlp/sample3.html)
pythonでしたいよ〜

2 できること

文章を入れると、キーワード単語とその重要度が出力される。

3 サンプルリクエストURL

https://jlp.yahooapis.jp/KeyphraseService/V1/extractappid="yahoocliantID"&sentence=<対象のテキスト>
yahooのcliant ID取得:(https://developer.yahoo.co.jp/yconnect/v2/registration.html)

4 コード

keyword.py

import urllib.request
import sys
import xml.etree.ElementTree as ET
#sentence=解析したい文章
def key(sentence):
    url="https://jlp.yahooapis.jp/KeyphraseService/V1/extract?appid=<yahooのcliant ID>&sentence="+urllib.parse.quote(sentence)
    req = urllib.request.Request(url)
    with urllib.request.urlopen(req) as res:
        body = res.read()
        root = ET.fromstring(body)
    output=[]
    for child in root:
        sen=child[0].text+":"+child[1].text
        output.append(sen)
    return output

[参照]
・URLに日本語を含む時の対処(https://note.nkmk.me/python-urllib-parse-quote-unquote/)
・xmlを取得(https://qiita.com/Takaki_/items/4ba7f5e327296d403e65)

今回も引き続き、「setの集合演算」の続きについて書いていきます。
I'll write about "set operations in python" continuously on this page.

■ len関数により、setの要素数を調べる
 Research the number of elements with len function

・ リストやタプルと同じように、setの要素数をlen関数で知ることができます。

 As same as lists and tuples, you can get the number of elements with "len function".

>>> sandwich_set = {"egg", "bread", "mayonnaise", "salt", "pepper"}
>>> 
>>> len(sandwich_set)
5

■ 演算子でsetの要素を調べる
 Search elements of set with "in" or "notin" operators.

 setに特定の値が含まれているかどうかを、in または notin 演算子にて調べることができます。
 You can search for specific value in a set with "in" or "notin" operators.

>>> set = {"egg", "bread", "mayonnaise", "salt", "pepper"}
>>> 
>>> "egg" in set
True
>>> 
>>> "oiveoil" in set
False

■ setの比較 （Comparison of sets)

・set同士を比較する方法は、以下の3つに分けられます。

 There are three methods to compare sets below;

・ 「==」演算子により、setの完全一致の比較をする
 Compare elements for perfect matching

>>> a = {1, 2, 3}
>>> 
>>> b = {3, 1, 2}
>>> 
>>> c = {1, 2, 3, 4}
>>> 
>>> a == b
True
>>> 
>>> a == c
False
>>> 
>>> # aとbが一致していないかの比較
>>> 
>>> a != b
False
>>> 
>>> # aとcが一致していないかの比較
>>> 
>>> a != c
True

■isdisjoint()メソッドにより、共通要素の有無を比較する
 Know "presence or absence" for common elements with "isdisjoint()method"

>>> a = {"egg", "bread", "mayonnaise"}
>>> 
>>> b = {"pepper", "salt"}
>>> 
>>> c = {"egg", "oliveoil"}
>>> 
>>> #aとbに共通要素がない場合、True
>>> a.isdisjoint(b)
True
>>> 
>>> #aとcに共通要素がない場合、False
>>> a.isdisjoint(c)
False

■ issubset() / issuperset()メソッドにより、集合の包含関係を比較する
 Compare the inclusion relation for sets with issubset() / issuperset() methods

・2つのset（集合）を比較した際、どちらかが片方の要素を全て含んでいるかを調べることができます。
aの要素の全てがbに含まれている場合...

 ・「サブセット（部分集合）」： aがbに対して

 ・「スーパーセット（上位集合）」： bはaに対して

>>> a = {"egg", "bread"}
>>> 
>>> b = {"egg", "bread", "mayonnaise", "salt", "pepper"}
>>> 
>>> a.issubset(b)
True
>>> 
>>> #<= 演算子とissubset()メソッドは同じである
>>> 
>>> a <= b
True
>>> 
>>> #bはaのスーパーセットなので、Trueとなる
>>> 
>>> b.issuperset(a)
True
>>> 
>>> #issuperset()メソッドは、>= 演算子と同じである
>>> 
>>> b >= a
True

いかがでしたでしょうか

How was my post?

本ブログは、随時に更新していきますので、
定期的な購読をよろしくお願いします。

I'll update my blogs at all times.
So, please subscribe my blogs from now on.

本ブログについて、
何か要望等ありましたら、気軽にメッセージをください！

If you have some requests, please leave some messages! by You-Tarin

また、「Qiita」へ投稿した内容は、随時ブログへ移動して行きたいと思いますので、よろしくお願いします。

環境

• Raspberry Pi 3
• Raspbian Jessie
• Python 3.4 (Python 2系と共存)

起きたこと

ラズパイでFlask開発しようと思い，本家ページを見ながら導入しようとしたとき，以下のエラーが発生

$ python3 -m venv app
Error: Command '['/(パス)/app/bin/python3', '-Im', 'ensurepip', '--upgrade', '--default-pip']' returned non-zero exit status 1

このエラーを調べると，次のようにpipインストールを回避して，後からpip入れればいいやん，という対策が多くヒット．

$ python3 -m venv --without-pip app

ただこの後のpipインストールが仮想環境を構築するたび毎回するのは面倒．

対策

色々試していると，どうもensurepipのモジュールが入っていない様子．
sh
$ python3 -m ensurepip
/usr/bin/python3: No module named ensurepip


以下コマンドでvenvできるようになりました．

$ sudo apt install python3-venv

はじめに

マイニングに使われていたらしい「ビデオ出力のないビデオカード（RX470）」と「PCIExpress x1が10個以上あるマザーボード（CPU込）」を入手しました。「目的のためのお買い物」ではなく「目的は買ってから考える」のはいつものこと^^;、ということで。

今更マイニングもないし、機械学習とかかな？チャットボット作ってみたいな～、と調べてみると、大量の学習データが必要とか。学習データ、どうやって調達するのかしら？と先人の業を見るに、ネットにある会話コーパスを利用したり、twitterでreplyを集めたり、とからしく。

他に会話を集める方法はないのかしら？と考えていたら
「放送電波に垂れ流しになってる会話を利用すればいいんじゃね？」
となにかが降りてきました。
そういえば、データ放送に文字情報ってあるよね？

目的

というわけで。テレビの放送電波に乗ってる文字情報を、いろいろに喰わせられるテキストにしてみたいと思います。
実際になにをやるか、というと、地上波放送を録画したTSファイル、たとえば2019/01/07 23:50 から NHK総合 で放送された 「みんなで筋肉体操 腕立て伏せ(２)」 の録画ファイルから

皆さん 筋トレしてますか？
｢みんなで筋肉体操｣です｡
筋トレは 継続して行わなければ効果は上がりません｡
楽しんで 筋肉を追い込んでいきましょう｡
今日は 腕立て伏せです｡
分厚い胸板力強い上半身を作りましょう｡
１種目目は 60秒インターミッテント･プッシュアップです｡

……といったテキストファイルを得る、というのを目的とします。
楽しんで「筋肉は裏切らない」を学習させていきましょう（ぇー

環境

• OS: Ubuntu18.04
• 地上デジタルTVチューナー: PX-S1UD V2.0
• 録画ソフト: recdvb

今回は文字情報が目的になります。画質などは重要ではないですし、ワンセグで充分ですのでB-CASカードやカードリーダーは不要になります。つか、B-CASカードだのカードリーダーを用意するのがめんどい^^;。
ワンセグにしたのは画質が必要ない、と同時に、フルセグを録画したファイルとかサイズがバカでかいのでね……厳しいのですよ、HDDの容量的に。
フルセグチューナーをワンセグチューナーとして利用するのもなんですが、PX-S1UD が家で利用場所もなく転がっていましたので。秋葉原あたりで500円くらいで売ってるワンセグチューナーはLinuxで使うにはちょっとキツいしね、というのもあります。

ここではすでに PX-S1UD を使って recdvb で録画できる状態にある、ということで話を進めさせていただきます。

字幕抽出

字幕抽出、の前に。実験用に放送を実際に録画してファイルにしておきましょう。

recdvb --sid 1seg 27 120 test.m2ts

問題はこの録画したファイルに字幕テキストが入っているかどうか？です。VLCなどで再生できる環境があるのでしたら、字幕を表示することができますので見ておいてください。
NHKの番組であれば、日付変わってから朝7時くらいまでの放送以外はほぼ入っているようです。
NHKと契約していないなどで観れない場合、NHK以外の在京キー局の放送では、ドラマなどはほぼ入っていましたが、生放送では入ってない番組もあるようです。
私が確認した範囲では、日曜日の午前8時30分からのテレビ朝日の番組では字幕が入っていることを確認しています（んー？。

字幕抽出ソフト選択

字幕を抽出する方法ですが、今回以下の方法を試してみました。

1. Python3 で ariblib を利用して出力
2. Windows用のCaption2AssC.exe を wine を使って利用して出力
3. assdumperを利用して出力

まず 1.ですが、今回録画したワンセグファイルを入れた場合には、なにも出力されませんでした。ちなみに、家で録画PC（Windows10）で録画したフルセグ動画もなにも出力されず、でした（ちゃんと試していないのでおそらくですが、recdvb でフルセグ録画した場合は出力されると思います）。
次に 2.については、録画WinPCによるフルセグ動画からは字幕情報が取れましたが、今回録画したワンセグファイルからは出力できず。
最後に、3.は今回録画したワンセグファイルから出力できました。

なので、今回の用途では 3.を使います。
今回使用する 3. のassdumper（あるいは2.のCaption2AssC）では、Advanced SubStation Alpha（以下 ASS）という字幕の形式になります。これを後に通常のテキストにしていくことにします。

assdumperビルド

ソースを持ってきます

$ svn export https://github.com/eagletmt/eagletmt-recutils/trunk/assdumper

assdumper をそのままビルドしたのでは、話者を識別するために字幕でよく使われる文字色が出力されません。なので出力を追加します。
今回は「色が変わったことだけわかればいい」という考えなので、これをそのまま字幕として別のソフトなどに喰わせてもいいものかは知りませんので注意。

--- assdumper/assdumper.cc      2017-01-19 21:34:56.000000000 +0900
+++ assdumper_/assdumper.cc     2019-02-02 22:18:58.967834742 +0900
@@ -204,6 +204,8 @@ public:
   {
     const unsigned char *end = str + len;
     std::string ans;
+    unsigned char prev_color = 7;
+    std::string colors[] = {"000000","0000ff","00ff00","00ffff","ff0000","ff00ff","ffff00","ffffff"};
     for (const unsigned char *p = str; p < end; ++p) {
       if (0xa0 < *p && *p < 0xff) {
         char eucjp[3];
@@ -233,7 +235,11 @@ public:
         }
         ++p;
       } else if (0x80 <= *p && *p <= 0x87) {
-        // color code. ignore
+        if( prev_color == (*p - 0x80)){
+        }else{
+          ans += "{\\c&H" + colors[*p - 0x80] + "&}";
+        }
+        prev_color = *p - 0x80;
       } else if (*p == 0x0d) {
         // CR -> LF
         ans += "\\n";

パッチを当てたら、make して出来た assdumper をパスが通った場所にでも置いてください。

字幕抽出結果

2019/01/27 08:30 からテレビ朝日で放送された「HUGっと！プリキュア」が入った録画ファイルを入れるとこのようになります。

$ assdumper rec/20190127/rec24_20190127080818.m2ts
program_number = 1448, program_map_PID = 8136
1 pmt_pids
8136
388 caption pid, PCR_PID = 257
[Script Info]
ScriptType: v4.00+
Collisions: Normal
ScaledBorderAndShadow: yes
Timer: 100.0000

[Events]
　：
（中略）
　：
Dialogue: 0,00:14:18.76,00:14:22.46,Default,,,,,, {\c&H00ffff&}・うぅぅぅ…・\n{\c&Hffffff&}・（さあや）はい 息はいて〜・
Dialogue: 0,00:14:22.46,00:14:25.93,Default,,,,,, {\c&H00ffff&}はぁ〜！ うぅぅ…
Dialogue: 0,00:14:25.93,00:14:31.02,Default,,,,,, {\c&H00ffff&}あぁぁぁ〜…\n{\c&Hffffff&}（さあや）上手 上手
Dialogue: 0,00:14:31.02,00:14:37.04,Default,,,,,, ・（ほまれ）はな！・\n{\c&H00ffff&}うぅっ はぁ…  
Dialogue: 0,00:14:37.04,00:14:40.74,Default,,,,,, （ダイガン）おぉ！\n{\c&H00ffff&}ほまれ…
Dialogue: 0,00:14:40.74,00:14:43.05,Default,,,,,, 間に合ったね
Dialogue: 0,00:14:43.05,00:14:46.52,Default,,,,,, {\c&H00ffff&}来てくれたんだ
Dialogue: 0,00:14:46.52,00:14:50.92,Default,,,,,, はな！ フレフレ！
Dialogue: 0,00:14:50.92,00:14:53.23,Default,,,,,, {\c&H00ffff&}わぁ…
Dialogue: 0,00:14:53.23,00:14:54.85,Default,,,,,, がんばれ！
Dialogue: 0,00:14:54.85,00:14:56.93,Default,,,,,, {\c&H00ffff&}うん
Dialogue: 0,00:14:56.93,00:15:00.40,Default,,,,,, さぁ いくよ 赤ちゃん がんばってる！
Dialogue: 0,00:15:00.40,00:15:06.65,Default,,,,,, {\c&H00ffff&}うん！  
Dialogue: 0,00:15:06.65,00:15:09.89,Default,,,,,, {\c&H00ffff&}＜子どもの頃 なりたかったわたしに
Dialogue: 0,00:15:09.89,00:15:14.05,Default,,,,,, {\c&H00ffff&}わたしは なれたのかな…＞
Dialogue: 0,00:15:14.05,00:15:17.06,Default,,,,,, {\c&H00ffff&}うぅぅ〜 あぁ〜
Dialogue: 0,00:15:17.06,00:15:20.30,Default,,,,,, {\c&H00ffff&}うぅ〜…
Dialogue: 0,00:15:20.30,00:15:23.30,Default,,,,,, {\c&H00ffff&}＜未来は 楽しいことばかりじゃない＞
Dialogue: 0,00:15:23.30,00:15:26.77,Default,,,,,, {\c&H00ffff&}＜めげそうになることも いっぱいある＞
　：
（以下略）
　：

このように、ASSの形式で出力されます。
このままでは使い勝手が悪いので、使いやすいようテキスト整形をしていきます。

テキスト整形

assの字幕ファイルを整形していくわけですが、これが困ったことに「タグ〇〇が入ったら話者が変わる」「××が表れたら一文が終了」などという決まりはないようです。局によって、あるいは番組によってかなりまちまちなようです。あくまで映像・音声とともに流れる字幕ですので、字幕を読んだ人が他の情報と統合して補う、ということなのでしょう。

とはいえ、整形にはある程度ルールを決めていかないと仕方ないので、ルールを決めて整形をしていくことにします。

整形ルール

とりあえず、今回は以下のような整形ルールで整形をするようにします。

1. 字幕文字列が空の場合は無視する
2. 「音符 + "～"」の場合は音楽が流れているだけとみなし無視する
3. 字幕表示開始時間が、直前の字幕表示終了時刻から5秒以上経過していた場合は、別の文章・会話とする（ブランク行を入れる）
4. 文字色が直前と比較して変化した場合は、別の文章・人の発言とする
5. 「。（読点）」「！」「？」等が字幕行末に来た場合は、次の字幕は別の文章・人の発言とする
6. 字幕行末に「→」等、矢印の記号が来た場合は次の字幕に文章が続くとする
7. 丸括弧の中は、発言している人の名前、または状況説明とみなしてテキストには出力しない
8. カギ括弧（「」）、角括弧（［］）、山括弧（＜＞）などの中は発言や思ってることとしてテキストに出力する

これで "見ている範囲" では "今のところ" は "それなりに" 整形されているようです。
あくまで私が「今のところ」「見ている範囲」ですので頼りすぎるのは危険ですし、後述しますが、すでに駄目な場合もあります。あくまで「それなり」です（責任逃れの防壁準備）。
このルールについてはうまく整形できてない例を見つけ次第、適宜決めていくしかないような気がします。

皆さんにも「テレビを見ているときに迂闊に字幕をONにしてしまい、整形がうまくいきそうにない字幕を見つけて苦しむ」呪いがかかるといいと思います（ぁ？。

整形用スクリプト

整形のために作成したPythonスクリプトを示します。

ass2text.py

#!/usr/bin/python3
# -*- coding: utf-8 -*-

import sys
import re

noserif = ['♬～', '♬〜', '♪～']
cmark = ['→', '➡']
eos = ['｡', '。', '!', '！', '?', '？', '⁉', '‼']
ndbraket =  ['＜', '＞', '〈', '〉', '《', '》', '≪', '≫', '\[', '\]', '［', '］']
reb = '|'.join(ndbraket) + '|「|」[。｡]{0,1}';
ndbraket.extend(['「', '」', '」。', '」｡'])

def HMSms2as(t):
    ht = re.split('[:.]', t)
    r = ( int(ht[0]) * 60 * 60 * 100 ) + \
        ( int(ht[1]) * 60 * 100) + \
        ( int(ht[2]) * 100 ) + \
        ( int((ht[3]+"0")[:2]) % 100 )
    return r

def ass2array(fn):
    b = False
    r = []

    try:
        f = open(fn)
        l = f.readline()
        while l:
            s = l.replace("\n","")    
            if 0 < len(s):
                if '[' == s[0]:
                    if '[events]' == s[0:8].lower():
                        b = True
                        l = f.readline()
                        continue
                    else:
                        b = False
                else:
                    if b:
                        if 'dialogue:' == s[0:9].lower():
                            c = 'ffffff'
                            d = s[10:].split(',')
                            if 'default' == d[3].lower():
                                txt = [x for x in re.split('({|})', ','.join(d[9:]).strip()) if not ''==x]
                                st = HMSms2as(d[1])
                                et = HMSms2as(d[2])
                                tb = False
                                buf = []
                                for t in txt:
                                    if "{" == t:
                                        tb = True
                                        continue
                                    if "}" == t and tb:
                                        tb = False
                                        continue
                                    if tb:
                                        m = re.search('\\\d{0,1}c&H([0-9a-fA-F]+)', t)
                                        if m:
                                            c = m.group(1)
                                    else:
                                        ta = [x for x in re.split('\\\\[nN]', t.replace('\\h', ' ')) if not ''==x]
                                        for tf in ta:
                                            buf = [tf, c, st, et]
                                            r.append(buf)
            l = f.readline()
        f.close
    except Exception as e:
        print(e)
        pass
    return r


def tsplit(text):
    r = []
    ta = [x.strip() for x in re.split('(\(|\)|（|）|' + reb + ')', text) if not ''==x]
    tb = False
    for t in ta:
        if t in ['(', '（']:
            tb = True
            continue
        if t in [')', '）']:
            tb = False
            continue
        if t in ndbraket:
            continue

        if tb:
            pass
        else:
            if not t in noserif:
                r.append(t)
    return r

if __name__ == '__main__':
    if 2 != len(sys.argv):
        print('Usage: # python %s filename' % sys.argv[0])
        quit()

    aary = ass2array(sys.argv[1])

    j=0
    tary = []
    t = ""
    for i in range(len(aary)):
        if aary[i][0] in noserif:
            continue
        if 0 < i:
            if 500 > ( aary[i][2] - aary[i-1][3]):
                if aary[(i-1)][1] != aary[i][1]:
                    tary.extend(tsplit(t))
                    t = aary[i][0]
                else:
                    if t[-1:] in eos:
                        tary.extend(tsplit(t))
                        t = aary[i][0]
                    elif t[-1:] in cmark:
                        t = t[:-1] + aary[i][0]
                    else:
                        t = t + aary[i][0]
            else:
                tary.extend(tsplit(t))
                tary.append("")
                t = aary[i][0]
        else:
            t = aary[i][0]
    if "" != t:
        tary.extend(tsplit(t))
for t in tary:
    print (t)

整形結果

先に挙げました「HUGっと！プリキュア」のASSファイルをスクリプトを通して整形した結果、以下のようになりました。

・うぅぅぅ…・
・
はい 息はいて〜・
はぁ〜！ うぅぅ…あぁぁぁ〜…
上手 上手・
はな！・
うぅっ はぁ…
おぉ！
ほまれ…
間に合ったね
来てくれたんだ
はな！ フレフレ！
わぁ…
がんばれ！
うん
さぁ いくよ 赤ちゃん がんばってる！
うん！
子どもの頃 なりたかったわたしにわたしは なれたのかな…
うぅぅ〜 あぁ〜うぅ〜…
未来は 楽しいことばかりじゃない
めげそうになることも いっぱいある

複数人の発言が一行に入っていたのが複数行になっている（「はぁ〜！ うぅぅ…あぁぁぁ〜…」「上手 上手」とか）、1人のセリフが複数行にわたっていたのが1行になっている（「子どもの頃 なりたかったわたしにわたしは なれたのかな」とか）、というのがわかりますでしょうか。

このプリキュアの例ではあまり整形効果なさそうに見えますが（じゃぁなぜ例として挙げた^^;）、最初に挙げた筋肉体操では

Dialogue: 0,0:00:31.13,0:00:34.96,Default,,0000,0000,0000,,{\pos(264,438)\c&H00ffff&}筋トレは 継続して行わなければ\N
Dialogue: 0,0:00:31.13,0:00:34.96,Default,,0000,0000,0000,,{\pos(290,518)\c&H00ffff&}効果は上がりません｡\N
Dialogue: 0,0:00:34.96,0:00:38.26,Default,,0000,0000,0000,,{\pos(184,518)\c&H00ffff&}楽しんで 筋肉を追い込んでいきましょう｡\N
Dialogue: 0,0:00:40.77,0:00:43.17,Default,,0000,0000,0000,,{\pos(344,518)\c&H00ffff&}今日は 腕立て伏せです｡\N
Dialogue: 0,0:00:43.17,0:00:46.67,Default,,0000,0000,0000,,{\pos(370,438)\c&H00ffff&}分厚い胸板\N
Dialogue: 0,0:00:43.17,0:00:46.67,Default,,0000,0000,0000,,{\pos(397,518)\c&H00ffff&}力強い上半身を作りましょう｡\N

と一文が複数行（複数字幕）にわたることが多いため、整形の効果はわかっていただけるのではないでしょうか？

ただ、先にも書いたとおり「駄目な場合」もありました。先に挙げたプリキュアのASS、別の箇所にある次のような出力を見てみます。

Dialogue: 0,15:29:47.81,15:29:52.21,Default,,,,,, {\c&H00ffff&}実はわたし みんなを守るプリキュアなの
Dialogue: 0,15:29:52.21,15:29:54.75,Default,,,,,, {\c&H00ffff&}あっ この子は 空からふってきた
Dialogue: 0,15:29:54.75,15:29:58.45,Default,,,,,, {\c&H00ffff&}不思議な赤ちゃん はぐたん\n{\c&Hffffff&}はぎゅ！

整形結果はこうなります。

実はわたし みんなを守るプリキュアなのあっ この子は 空からふってきた不思議な赤ちゃん はぐたん
はぎゅ！

「なの」と「あっ」の間、同じ人物（同じ文字色）のセリフですが、読点や字幕の時間などでは文章の区切りがわからないため一行になってしまっています。
今回はこれについては妥協します。
こういった場合を機械的に別の文章とできる、なにかいい方法をご存知の方いましたら教えてくださいまし。

課題・問題

テレビ放送からテキストを取得する、という目的はおおよそ達成できたと思います。会話とかチャットボットの学習に使えるかはともかく（あれ？。

キャスターが淡々と伝えるようなニュース番組では、さすがに会話とはなりませんが、文章素材としてはかなり有用に使えるのではないかと思います。

トーク系のバラエティ番組では会話が取得できる、と思いきや、そうもいかないようです。以下は 2019/01/27 11:25 からテレビ東京で放送された「男子ごはん」のオープニングトーク部分です。

突然ですが 問題です。
わかりました。
はい 太一さん。
ブーッ！
ブーッ！
腹立つな。
ラストチャンス！
ピンポーン！
やった～！
大正解です。

「問題です」「ブーッ！」「ピンポーン！」などのやりとりの中で問題や解答をちゃんと言っているのですが、字幕テキストに入ってきていません。
画面の映像にはテロップ文字が出てきますから、映像にテロップで文字を入れているから字幕として入れる必要はない、という判断なのでしょう。そのようになってる番組は多く見られます。
このような場合は字幕だけを追っても会話が成立しないことになります。
バラエティ番組などでは、発言や対するツッコミなどがテロップ側に出る番組が多くなっていますので、こういったことが発生しやすいようです。

字幕を会話をデータとして取得したいならば、ドラマやアニメなどを中心に集めた方がいいかもしれません。……若干偏った会話になりそうな気はしますが（ぉ？。

もちろんバラエティ番組全てが会話を取得できないわけではなく、たとえば、司会者が「毎度おなじみ流浪の番組……」と言って始まる番組では、結構綺麗に会話が取得できます。

NHKがほぼ全ての番組で字幕テキストがあるのも印象的でした（2018年の紅白歌合戦の「勝手にシンドバッド」の曲中に「la la la…桑田君！」が入ってきてたくらい）。ただしETVについては、子供向け番組が多いせいでしょうか、漢字がほとんど含まれない番組も多く見られましたので、気に留めておくべきかもしれません。

テキストをどの用途によって使うか、という観点から番組を選択していく必要もあるでしょう。その際は放送波から抽出できる番組情報を使うことを考えていった方がいいかもしれません。

最後に

本当は、「切りかえしていこう」「なにが切りかえすのよ」「まぁまぁ無かった事にして」「後世まで語り継ぐよ」とか「負けて悔いなし」「さぁさぁ急ぎましょう」「キミの乗った馬車のように」とかの会話を取り出したかったのですが。現在関東で放送されているClassicには字幕情報はないようです。残念。

「ぼくはギャングスターになる」というアニメの「覚悟はいいか？オレはできてる」などのセリフも、と考えていたのですが。こちらも字幕情報がありませんでした。無念。

当初目的に対しての残念無念はありましたが、それなりに面白かったので、よしとしましょう。

「放送に含まれる字幕テキストを字幕以外に使う」というのが調べても、あまり出てこないように思えるのですが、「テキスト情報が常に得られるデバイス」と考えればなかなかに使いでがありそうに思えます。どうでしょう？
そうなるとバラエティ番組などで、映像側テロップに入ってるからと字幕には入ってこないような発言が取得できない、というのは非常に惜しいんですよね。どうにか字幕に入れてほしい、というのは、さすがにわがままですね。

余談

放送局ごとに別々のチャットボットに学習させていった場合、同じ質問をしたときに個性など出るのでしょうか？
……若干黒い考えも思い浮かびましたが、検証はおまかせします > ここまで読んでくださった面々

はじめに

pytestはPython用のテストツールです。

標準のunittestに比べ、テストエラー時の結果が分かりやすいのが特徴です。

例として、辞書オブジェクトを比較する以下のテストコードをunittestとpytestそれぞれで実行してみます。

# test_dict.py

import unittest

dict1 = {'name': 'Tom', 'age': 20}
dict2 = {'name': 'John', 'age': 23}


class TestUnitTest(unittest.TestCase):

    def test_one(self):
        assert dict1 == dict2

unittestでの実行例。

$ python -m unittest
F
======================================================================
FAIL: test_one (test_dict.TestUnitTest)
----------------------------------------------------------------------
Traceback (most recent call last):
  File "/Users/gaizaku/Develop/pytest_learning/test_dict.py", line 12, in test_one
    assert dict1 == dict2
AssertionError

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

FAILED (failures=1)

一方、pytestでの実行例。

$ pytest -vv test_dict.py
================================ test session starts ================================
platform darwin -- Python 3.7.1, pytest-4.1.1, py-1.7.0, pluggy-0.8.1 -- /Users/gaizaku/.pyenv/versions/3.7.1/envs/pytest_learning/bin/python
cachedir: .pytest_cache
rootdir: /Users/gaizaku/Develop/pytest_learning, inifile:
collected 1 item

test_dict.py::TestUnitTest::test_one FAILED                                   [100%]

===================================== FAILURES ======================================
_______________________________ TestUnitTest.test_one _______________________________

self = <test_dict.TestUnitTest testMethod=test_one>

    def test_one(self):
>       assert dict1 == dict2
E       AssertionError: assert {'age': 20, 'name': 'Tom'} == {'age': 23, 'name': 'John'}
E         Differing items:
E         {'name': 'Tom'} != {'name': 'John'}
E         {'age': 20} != {'age': 23}
E         Full diff:
E         - {'age': 20, 'name': 'Tom'}
E         ?          ^           ^ ^
E         + {'age': 23, 'name': 'John'}
E         ?          ^           ^ ^^

test_dict.py:12: AssertionError
============================= 1 failed in 0.08 seconds ==============================

見ての通り、pytestはキー・バリューの差分まで表示してくれるので、何故テストが失敗したのか非常に分かりやすいです。

また、pytestにはプラグイン機構等のいくつかの機能があります。

その中でも特に便利なフィクスチャ機能について、公式ドキュメントを参考に使い方をまとめました。

フィクスチャ（fixture）とは

テストの前処理(DBをセットアップする、モックを作成するなど)を行うためのpytestの機能です。

個人的には、フィクスチャがあるからpytestを使っている言っても過言ではありません。

前処理はテストコードの保守性を低くする大きな原因ですが、フィクスチャ機能を上手に使うことでテストコードをクリーンに保つことが可能になります。

フィクスチャの概要

仮想的なデータベースオブジェクトUsersクラスを使ってフィクスチャの概要を紹介します。

# users_db.py


class Users:
    def __init__(self):
        self.last_insert_id = 0
        self.rows = {}

    def insert(self, name, age):
        self.last_insert_id += 1
        self.rows[self.last_insert_id] = {
            'id': self.last_insert_id,
            'name': name,
            'age': age,
        }

    def get(self, id_):
        return self.rows[id_]

上記Usersクラスをテストするコードを記述します。

以下のテストコードではdbフィクスチャ(フィクスチャ関数とも言う)を定義しています。

dbフィクスチャはダミーレコードを追加したUsersデーターベースを作成するという前処理を行っています。

テストケースの中でそのデーターベースを使いたい場合は、テストケースの引数にdbと追加するだけでデーターベースを利用出来るようになります。

# test_fixture_sample.py

import pytest

from users_db import Users


# これがフィクスチャ
@pytest.fixture
def db():
    users = Users()
    users.insert('Bob', 10)
    users.insert('Alice', 12)
    return users


# dbフィクスチャを利用するテストケース
def test_one(db):
    assert db.get(1)['name'] == 'Bob'


# フィクスチャは複数のテストケースで共有できる
def test_two(db):
    assert db.get(2)['name'] == 'Alice'


# クラスベースのテストでも利用できる
class TestUers:
    def test_one(self, db):
        assert db.get(1)['name'] == 'Bob'

複数のフィクスチャを同時に使いたい場合は、単に引数に列挙すればOKです。

@pytest.fixture
def fixture1():
    pass


@pytest.fixture
def fiture2():
    pass


def test_foo(fixture1, fiture2):
    pass

unittestとの比較

比較として、unittestで同様のテストコードを記述してみます。

# test_unittest_setup.py

import unittest

from users_db import Users


class TestUnittest(unittest.TestCase):

    def setUp(self):
        self.db = Users()
        self.db.insert('Bob', 10)
        self.db.insert('Alice', 12)

    def test_one(self):
        assert self.db.get(1)['name'] == 'Bob'

    def test_two(self):
        assert self.db.get(2)['name'] == 'Alice'

$ python -m unittest test_unittest_setup.py
..
----------------------------------------------------------------------
Ran 2 tests in 0.000s

OK

conftest.pyによるフィクスチャの共有

先ほどのテストコードは規模も小さいので、unittestと比較した場合のフィクスチャのメリットが良く分からないかもしれません。

フィクスチャの真価はテスト間で共有出来る事にあります。

conftest.pyというファイルにフィクスチャを記述すると、そのフィクスチャは複数のテストコードから呼び出すことが出来るようになります。

例として、先ほどのdbフィクスチャをconftest.pyに移動します。

# conftest.py

import pytest

from users_db import Users


@pytest.fixture
def db():
    users = Users()
    users.insert('Bob', 10)
    users.insert('Alice', 12)
    return users

フィクスチャを移動した後でもテストは問題なくパスします。

$ pytest -q test_fixture_sample.py
...                                                                      [100%]
3 passed in 0.02 seconds

これで今後テストが増えたとしても、引数にdbと記述するだけでフィクスチャを利用出来ます。

unittestの場合は毎回setUpメソッドを記述しないといけないので、フィクスチャがテストコードをシンプルに保つのに便利な機能で有ることが分かります。

スコープ

フィクスチャにはスコープ(scope)という概念があります。

スコープを理解すると、フィクスチャ関数が実行される粒度を制御出来るようになります。

まず、以下にスコープの種類と実行粒度をまとめます。

スコープ名	実行粒度
function	テストケースごとに1回実行される（デフォルト）
class	テストクラス全体で1回実行される
module	テストファイル全体で1回実行される
session	テスト全体で1回だけ実行される

スコープの動作を理解するためにconftest.pyに4つのフィクスチャを追加します。

各々のフィクスチャは実行されると自身のスコープ名をprintするので、これらを使ってスコープ毎の実行粒度を検証します。

@pytest.fixture(scope='function')
def fixture_function():
    print('function')


@pytest.fixture(scope='class')
def fixture_class():
    print('class')


@pytest.fixture(scope='module')
def fixture_module():
    print('module')


@pytest.fixture(scope='session')
def fixture_session():
    print('session')

次に、3つのテストコードを作成します。

# test_scope_sample1.py


class TestScopeSample1:

    def test_one(self,
                 fixture_function,
                 fixture_class,
                 fixture_module,
                 fixture_session):
        pass

    def test_two(self,
                 fixture_function,
                 fixture_class,
                 fixture_module,
                 fixture_session):
        pass

# test_scope_sample2.py


class TestScopeSample2:

    def test_one(self,
                 fixture_function,
                 fixture_class,
                 fixture_module,
                 fixture_session):
        pass

    def test_two(self,
                 fixture_function,
                 fixture_class,
                 fixture_module,
                 fixture_session):
        pass

# test_scope_sample3.py

# 注：このテストコードはfixture_classは使わない


def test_one(fixture_function,
             fixture_module,
             fixture_session):
    pass


def test_two(fixture_function,
             fixture_module,
             fixture_session):
    pass

テストを実行します。

$ pytest -vs test_scope_sample*
============================= test session starts ==============================
platform darwin -- Python 3.7.1, pytest-4.1.1, py-1.7.0, pluggy-0.8.1 -- /Users/gaizaku/.pyenv/versions/3.7.1/envs/pytest_learning/bin/python
cachedir: .pytest_cache
rootdir: /Users/gaizaku/Develop/pytest_learning, inifile:
collected 6 items

test_scope_sample1.py::TestScopeSample1::test_one session
module
class
function
PASSED
test_scope_sample1.py::TestScopeSample1::test_two function
PASSED
test_scope_sample2.py::TestScopeSample2::test_one module
class
function
PASSED
test_scope_sample2.py::TestScopeSample2::test_two function
PASSED
test_scope_sample3.py::test_one module
function
PASSED
test_scope_sample3.py::test_two function
PASSED

=========================== 6 passed in 0.06 seconds ===========================

出力結果から、各フィクスチャの実行回数をまとめます。

functionスコープのフィクスチャが最も頻繁に実行され、逆にsessionスコープのフィクスチャはテスト全体を通して1回しか実行されていません。

つまり、下に行くほどコストが高い前処理に適している事が分かります。

スコープ名	実行回数
function	6回（テストケース数と等しい）
class	2回（クラス数と等しい）
module	3回（テストファイル数と等しい）
session	1回（テスト開始直後に1回だけ実行）

なお、function以外のスコープを使う場合、フィクスチャがpytestにキャッシュされる点に注意します。

下記のmoduled_dbフィクスチャはmoduleスコープなので、フィクスチャにより作成されたUsersデーターベースの中身はテスト全体で共有されており、test_threeは全てのレコードを参照することが出来ています。

# test_fixture_cache.py

import pytest

from users_db import Users


@pytest.fixture(scope='module')
def moduled_db():
    users = Users()
    return users


def test_one(moduled_db):
    moduled_db.insert('Bob', 10)


def test_two(moduled_db):
    moduled_db.insert('Alice', 12)


def test_three(moduled_db):
    assert moduled_db.get(1)['name'] == 'Bob'
    assert moduled_db.get(2)['name'] == 'Alice'

$ pytest -q test_fixture_cache.py
...                                                                 [100%]
3 passed in 0.02 seconds

　終了処理(teardown)

テストが終了した後に何らかの終了処理(DB接続のクローズや一時ファイルの削除等)をしたいシチュエーションは多いと思います。

フィクスチャを使う場合、2種類の方法で終了処理を記述出来ます。

yieldを使うパターン

フィクスチャからテストケースに対してデータを渡す際、通常はreturnを使用しますが、代わりにyieldを使用するとテストの終了後に任意の処理を実行出来ます。

例として、テスト終了時に全レコードを表示するという終了処理を記述してみます。

まず、データベースオブジェクトに全レコードを返させるallメソッドを追加します。

# users_db.py


class Users:

    def all(self):
        return self.rows.values()

次にフィクスチャとテストコードを作成します。

dbフィクスチャはyieldした段階でブロックされ、処理がテストケースへと移動します。

テストが終了すると処理はフィクスチャに戻り、yield以降の処理が実行されます。

# test_teardown_sample.py

import pytest

from users_db import Users


@pytest.fixture
def db():
    users = Users()
    yield users

    print()
    for v in users.all():
        print(v)


def test_one(db):
    db.insert('Bob', 10)


def test_two(db):
    db.insert('Tom', 15)
    db.insert('Alice', 12)

テストを実行すると期待通りに全レコードが出力されます。

$ pytest -vs test_teardown_sample.py
============================= test session starts ==============================
platform darwin -- Python 3.7.1, pytest-4.1.1, py-1.7.0, pluggy-0.8.1 -- /Users/gaizaku/.pyenv/versions/3.7.1/envs/pytest_learning/bin/python
cachedir: .pytest_cache
rootdir: /Users/gaizaku/Develop/pytest_learning, inifile:
collected 2 items

test_teardown_sample.py::test_one PASSED
{'id': 1, 'name': 'Bob', 'age': 10}

test_teardown_sample.py::test_two PASSED
{'id': 1, 'name': 'Tom', 'age': 15}
{'id': 2, 'name': 'Alice', 'age': 12}


=========================== 2 passed in 0.01 seconds ===========================

終了処理はテストの成否や例外の有無を問わず実行されるので、フィクスチャ内で例外キャッチ等のエラーハンドリングは必要ありません。（ただし、フィクスチャ関数内で発生するエラーは別です）

# test_teardown_sample.py


def test_three(db):
    db.insert('Tom', 15)
    raise
    db.insert('Alice', 12)

$ pytest -s test_teardown_sample.py::test_three
============================= test session starts ==============================
platform darwin -- Python 3.7.1, pytest-4.1.1, py-1.7.0, pluggy-0.8.1
rootdir: /Users/gaizaku/Develop/pytest_learning, inifile:
collected 1 item

test_teardown_sample.py F
{'id': 1, 'name': 'Tom', 'age': 15}


=================================== FAILURES ===================================
__________________________________ test_three __________________________________

db = <users_db.Users object at 0x104d8ecc0>

    def test_three(db):
        db.insert('Tom', 15)
>       raise
E       RuntimeError: No active exception to reraise

test_teardown_sample.py:29: RuntimeError
=========================== 1 failed in 0.10 seconds ===========================

また、functionスコープ以外のフィクスチャでは、終了処理はそのフィクスチャを利用する全テストが終わった後に1回だけ実行されます。

# test_teardown_sample.py


@pytest.fixture(scope='module')  # <-- スコープを変更
def db():
    ...

$ pytest -vs test_teardown_sample.py
============================= test session starts ==============================
platform darwin -- Python 3.7.1, pytest-4.1.1, py-1.7.0, pluggy-0.8.1 -- /Users/gaizaku/.pyenv/versions/3.7.1/envs/pytest_learning/bin/python
cachedir: .pytest_cache
rootdir: /Users/gaizaku/Develop/pytest_learning, inifile:
collected 3 items

test_teardown_sample.py::test_one PASSED
test_teardown_sample.py::test_two PASSED
test_teardown_sample.py::test_three FAILED
{'id': 1, 'name': 'Bob', 'age': 10}
{'id': 2, 'name': 'Tom', 'age': 15}
{'id': 3, 'name': 'Alice', 'age': 12}
{'id': 4, 'name': 'Tom', 'age': 15}


=================================== FAILURES ===================================
__________________________________ test_three __________________________________

db = <users_db.Users object at 0x106433cf8>

    def test_three(db):
        db.insert('Tom', 15)
>       raise
E       RuntimeError: No active exception to reraise

test_teardown_sample.py:29: RuntimeError
====================== 1 failed, 2 passed in 0.11 seconds ======================

addfinalizerを使うパターン

yield以外にはaddfinalizerを用いるパターンがあります。

yieldと比較すると、addfinalizerには3つの大きな違いがあります。

• 終了処理をコールバック関数として登録する
• コールバック関数は複数登録できる
• フィクスチャ内で例外が起きたとしても必ず実行される

以下にシンプルなメールクライアントを用いたaddfinalizerの例を示します。

clientsフィクスチャが受け取っているrequestsはpytestが渡してくれる特殊なフィクスチャで、addfinalizerメソッドを使って終了処理を登録する事が出来ます。

# test_addfinalizer_sample.py

import pytest


class MailClient:
    def close(self):
        print(f'Close smtp connection:', self)


@pytest.fixture
def clients(request):
    ret = []
    for i in range(3):
        client = MailClient()
        request.addfinalizer(client.close)
        ret.append(client)
    return ret


def test_one(clients):
    pass

テストを実行すると、メールクライアントのcloseメソッドが期待通りに3回呼ばれている事が分かります。

$ pytest -sv test_addfinalizer_sample.py
============================= test session starts ==============================
platform darwin -- Python 3.7.1, pytest-4.1.1, py-1.7.0, pluggy-0.8.1 -- /Users/gaizaku/.pyenv/versions/3.7.1/envs/pytest_learning/bin/python
cachedir: .pytest_cache
rootdir: /Users/gaizaku/Develop/pytest_learning, inifile:
collected 1 item

test_addfinalizer_sample.py::test_one PASSEDClose smtp connection: <test_addfinalizer_sample.MailClient object at 0x10ef6a940>
Close smtp connection: <test_addfinalizer_sample.MailClient object at 0x10ef6a908>
Close smtp connection: <test_addfinalizer_sample.MailClient object at 0x10ef6a8d0>


=========================== 1 passed in 0.01 seconds ===========================

ここで、clientsフィクスチャに意図的に例外を仕込んでみます。

# test_addfinalizer_sample.py

@pytest.fixture
def clients(request):
    ret = []
    for i in range(3):
        client = MailClient()
        request.addfinalizer(client.close)
        raise  # <-- 例外を追加
        ret.append(client)
    yield ret

テストを実行すると、メールクライアントの生成は1つ目で終わってしまっていますが、生成されたクライアントに対するcloseメソッドはちゃんと呼ばれている事が分かります。

$ pytest -sv test_addfinalizer_sample.py
============================= test session starts ==============================
platform darwin -- Python 3.7.1, pytest-4.1.1, py-1.7.0, pluggy-0.8.1 -- /Users/gaizaku/.pyenv/versions/3.7.1/envs/pytest_learning/bin/python
cachedir: .pytest_cache
rootdir: /Users/gaizaku/Develop/pytest_learning, inifile:
collected 1 item

test_addfinalizer_sample.py::test_one ERRORClose smtp connection: <test_addfinalizer_sample.MailClient object at 0x10561b828>


==================================== ERRORS ====================================
__________________________ ERROR at setup of test_one __________________________

request = <SubRequest 'clients' for <Function test_one>>

    @pytest.fixture
    def clients(request):
        ret = []
        for i in range(3):
            client = MailClient()
            request.addfinalizer(client.close)  # 終了処理を登録
>           raise
E           RuntimeError: No active exception to reraise

test_addfinalizer_sample.py:17: RuntimeError
=========================== 1 error in 0.08 seconds ============================

ファクトリ・フィクスチャ

これまで作成してきたフィクスチャは全て何らかの固定的なデータを返す物でした。

しかし、動的なデータが欲しいシチュエーションも多いと思います。

そういう時は、データを生成するための関数(ファクトリ)を返すフィクスチャを定義すると便利です。

そういったフィクスチャの事を公式ドキュメントではファクトリ・フィクスチャと呼んでいます。

以下は「任意の名前と長さを持つPersonオブジェクトのリスト」を返すファクトリ・フィクスチャの例です。

# test_factory_fixture.py

import pytest


class Person:
    def __init__(self, name):
        self.name = name

    def __repr__(self):
        return self.name


@pytest.fixture
def persons_factory():
    def factory(names):
        return [Person(v) for v in names]
    return factory


def test_one(persons_factory):
    persons1 = persons_factory(['Tom', 'Bob'])
    persons2 = persons_factory(['Alice', 'Nancy'])
    print(persons1, persons2)

$ pytest -qs test_factory_fixture.py
[Tom, Bob] [Alice, Nancy]
.
1 passed in 0.01 seconds

パラメタライズド・フィクスチャ

複数の入力値による網羅的なテストを行いたい場合はパラメタライズド・フィクスチャが便利です。

pytest.fixtureデコレータのparams引数にlist等のiterableを指定すると、含まれている各要素をパラメータとしたフィクスチャが自動で生成されます。

以下はStudentクラスをテストするコードです。

Studentクラスは生徒が小学生なのか中学生なのかを教えてくれるschool_typeプロパティを持っているので、そのプロパティの動作をテストしています。

# test_params_sample.py

import pytest


class Student:
    ELEMENTARY_SCHOOL = 1
    JUNIOR_HIGH_SCHOOL = 2

    def __init__(self, name, age):
        self.name = name
        self.age = age

    @property
    def school_type(self):
        if self.age >= 13:
            return self.JUNIOR_HIGH_SCHOOL
        else:
            return self.ELEMENTARY_SCHOOL


@pytest.fixture(params=[('Tom', 10), ('Bob', 15), ('Alice', 12)])
def student(request):
    return Student(request.param[0], request.param[1])


def test_one(student):
    if student.age <= 12:
        assert student.school_type == 1
    else:
        assert student.school_type == 2

studentフィクスチャは3つのStudentオブジェクトを返すので、それに合わせて３つのテストケースが実行されます。

$ pytest -sv test_params_sample.py
============================= test session starts ==============================
platform darwin -- Python 3.7.1, pytest-4.1.1, py-1.7.0, pluggy-0.8.1 -- /Users/gaizaku/.pyenv/versions/3.7.1/envs/pytest_learning/bin/python
cachedir: .pytest_cache
rootdir: /Users/gaizaku/Develop/pytest_learning, inifile:
collected 3 items

test_params_sample.py::test_one[student0] PASSED
test_params_sample.py::test_one[student1] PASSED
test_params_sample.py::test_one[student2] PASSED

=========================== 3 passed in 0.02 seconds ===========================

フィクスチャからフィクスチャを使う

フィクスチャには他のフィクスチャを利用できるという特性があります。

小さいメソッドが集まる事で複雑なクラスが出来上がるように、フィクスチャ同士を組み合わせる事で複雑な前処理を実現出来ます。

以下の例では、モック化されたメールサーバ接続を返すmock_connectionフィクスチャと、それを利用してモック化されたメールクライアントを返すclientフィクスチャを定義しています。

# test_fixture_in_fixture.py

import pytest


class MailClient:

    def __init__(self, connection):
        self.connection = connection

    def send(self, message):
        return self.connection.send(message)


@pytest.fixture
def mock_connection():
    class MockConnection:

        def __init__(self):
            self.sents = []

        def send(self, message):
            self.sents.append(message)
            return True

    return MockConnection()


@pytest.fixture
def client(mock_connection):
    return MailClient(mock_connection)


def test_send(client):
    msg = 'Hello, world!'
    client.send(msg)
    assert client.connection.sents[0] == msg

$ pytest test_fixture_in_fixture.py
=================================== test session starts ===================================
platform darwin -- Python 3.7.2, pytest-4.2.0, py-1.7.0, pluggy-0.8.1
rootdir: /Users/gaizaku/Develop/pytest_learning, inifile:
collected 1 item

test_fixture_in_fixture.py .                                                        [100%]

================================ 1 passed in 0.02 seconds =================================

テストクラス・モジュール全体にフィクスチャを自動摘要する

pytest.mark.usefixturesを使うと、テストクラスやテストモジュール(test_*.py)全体にフィクスチャを自動で摘要出来るようになります。

例としてまず、テストケース毎に初期化された一時ディレクトリを作成してくれるフィクスチャをconftest.pyに定義します。

# conftest.py

import tempfil


@pytest.fixture
def tmpdir():
    with tempfile.TemporaryDirectory() as dirname:
        os.chdir(dirname)
        yield

上記フィクスチャを使う場合、もちろんテストケースの引数として記述しても良いのですが、このフィクスチャは何のデータも返さないので引数として受け取り意味をあまり感じません。

また、テストケースが増える度にいちいちフィクスチャ名を記述するのも冗長です。

そういった時は以下の様にテストクラスをpytest.mark.usefixturesでデコレートしてあげると、各テストケースの実行前に自動でフィクスチャが実行されるようになります。

# test_usefixtures_class.py

import pytest


@pytest.mark.usefixtures('tmpdir')
class TestUseFixtures:
    def test_one(self):
        with open('hoge.csv', 'w') as f:
            f.write('Hello, world!')

    def test_two(self):
        with pytest.raises(FileNotFoundError):
            open('hoge.csv', 'r')

$ pytest -q test_usefixtures_class.py
..                                                                  [100%]
2 passed in 0.02 seconds

また、テストモジュール(.pyファイル)で利用する場合は以下の様にします。（pytestmarkという変数に代入しないと動作しないので注意）

# test_usefixtures_module.py

import pytest

pytestmark = pytest.mark.usefixtures('tmpdir')


def test_one():
    with open('hoge.csv', 'w') as f:
        f.write('Hello, world!')

複数フィクスチャを利用する場合は列挙すればOKです。

pytest.mark.usefixtures('fixture1', 'fixture2', 'fixtureN')

conftest.pyの階層化とオーバーライド

テストコードが階層構造を持っている場合、前述のconftest.pyも各階層に配置することが出来ます。

その場合、テストケースは自身と同階層にあるconftest.pyと自身より上位層にあるconftest.pyに定義してあるフィクスチャを利用出来ます。

また、子階層のフィクスチャは上位層のフィクスチャをオーバーライドする事も出来ます。

以下に階層構造のテストコードの例を示します。

$ tree hier_conftest/
hier_conftest/
├── conftest.py
├── sub
│   ├── conftest.py
│   └── test_sub.py
└── test_root.py

これは上位層のconftest.pyです。

# hier_conftest/conftest.py

import pytest


@pytest.fixture
def username():
    return 'username'

これは子階層のconftest.pyで、親のusernameフィクスチャを利用しつつ、その結果をオーバーライドしています。

# hier_conftest/sub/conftest.py

import pytest


@pytest.fixture
def username(username):
    return 'sub-' + username

親階層のテストコードです。

# hier_conftest/test_root.py


def test_username(username):
    assert username == 'username'

子階層のテストコードです。

# hier_conftest/sub/test_sub.py


def test_username(username):
    assert username == 'sub-username'

テストを実行すると期待通りにオーバーライドされている事が分かります。

$ pytest hier_conftest/
=========================== test session starts ===========================
platform darwin -- Python 3.7.2, pytest-4.2.0, py-1.7.0, pluggy-0.8.1
rootdir: /Users/gaizaku/Develop/pytest_learning, inifile:
collected 2 items

hier_conftest/test_root.py .                                        [ 50%]
hier_conftest/sub/test_sub.py .                                     [100%]

======================== 2 passed in 0.03 seconds =========================