簡単な流れ 1.pandas.read_csvでcsvファイルを一括読み込み 2.itertuplesメソッドによって得たイテレータをfor文で回し，一行ずつ解析 3.要素の更新にはatまたはiatを利用し個別に操作(replace等を使えばまとめての更新が可能) 4.to_csvで書き込み サンプル.py 関数として実装すると下記のようになります． sample.py def csvEdit(path_csv): csv = pd.read_csv(path_csv) for row in csv.itertuples(): csv.at[row[0], 'hoge'] = huga csv.to_csv(path_csv) 少しだけitertuplesについて説明します．rowに格納されるタプルですが，これには先頭にindex名，そしてその行のデータが格納されています．rowはただのコピーなので代入等の要素操作は元データであるcsvに行ってください． また，csvファイルの形によってはiterrowsメソッドの使用の検討をおすすめします．インデックスを明記している場合や特別に扱う必要がある場合ですね． タスク管理用の実装 自分が制作しているタスク管理用プログラム内での実装は以下のようになりました． 習慣として登録されており(isRoutine)，deadlineを過ぎたものは自動更新がかかるようになってます． TaskNotify.py def loadTasks(path_csv): csv_tasks = pd.read_csv(path_csv, encoding='shift-jis') list_tasks =[] for row in csv_tasks.itertuples(): date = datetime.datetime.strptime(row.deadline, '%Y年%m月%d日').date() if row.isRoutine and (date < today): date += datetime.timedelta(days=row.isRoutine) csv_tasks.at[row[0], 'deadline'] = date.strftime('%Y年%m月%d日') list_tasks.append(Task(row.name, date)) csv_tasks.to_csv(path_csv, encoding='shift-jis', index=False) return list_tasks 次回　第-7回では... 一行ごとのデータ数を増やします．deadlineに細かく時間まで設定できるようにしたり，習慣の場合は曜日指定ができるようにする予定です．

はじめに これは学生時代に勉強していたテンソルネットワークを再度まとめたものになっております。ネット上にはテンソルネットワークの解説はありますが、ソースコードが公開されているのはあまりないので、今回はテンソル繰り込み群(Tensor Renormalizations Group, 以下 TRG)を例に解説とソースコードをあげさせていただきます。 本稿が、何かに役立てば幸いです。なお、間違いなどがございましたらご指摘していただけたら幸いです。できる限り対応させていただきたいと思います。 なお、本稿及びプログラムは金沢大学の武田真滋様の資料1を参考に作成いたしました。大変わかりやすい資料ですので、一読をおすすめいたします。 また、参考資料の数式が非常に詳しく書かれているため、数式の結果だけを記述しております。なお、一部文字などを改変しております。 概要 テンソルネットワークとは、量子多体系の基底状態を求める数値計算の手法として開発されました。このテンソルネットワークはモンテカルロ法での符号問題が存在しない、巨大な体積の計算が可能という様々な利点があります。また、物理学の諸問題だけではなく機械学習の分野でも応用されるなど、個人的にホットトピックだと思っております。本稿では先に述べたとおり、二次元イジングをテンソルネットワーク表記で表し、TRG を用いて厳密解を求めてみたいと思います。 実行環境 今回の実行環境は次のようになります。 Ubuntu 18.04LTS Ryzen 7 3700X RAM 16GB Python(miniconda) ver 3.8.8 numpy ver 1.19.5 Pytorch 1.8.0 テンソルネットワークとは テンソルネットワークとは、量子多体系の波動関数、統計力学の分配関数などを多数のテンソルの縮約として表現する手法です。 まず、テンソルというのは、次のようなものを言います。 A_{ij}\\B_{ijk}\\C_{ijkl} Aがランク2のテンソル、B がランク3のテンソル、C がランク4のテンソルといい、それぞれの添字は足と呼ばれます。量子力学を学んだ方ならとっつきやすいかもしれません。さて、ランク4のテンソルが4つあった場合、それらをテンソルネットワークで表すと次のようになります。 上記図の数式をみると ijkl で和を取ることになります。アインシュタインの縮約記法と似ています。 TRGでは、上記図の正方形のネットワークが複数個結合するネットワークとなります。 二次元イジングのテンソルくりこみ群 繰り込み群の流れは 二次元イジングの分配関数をテンソルネットワーク表示にする。 テンソルを特異値分解し、新しい添字を用いて2つのテンソルに分ける。 古い添字の和を実行して新しいテンソルを作る。 2と3を繰り返し、途中でDをDcutにし低ランク行列による近似を行う。 最後にテンソルのトレースを取る。 テンソルネットワーク表示 まず、二次元イジングの分配関数Zをテンソルネットワーク表示1にします。 Z = \sum_s \exp{(\sum_{<x,y>}\beta s_x s_y)} = 2^V(\cosh{\beta})^{2V}\sum_{\dots ijkl \dots} \dots T_{ijkl}T_{mnop} \dots\\ \ast T_{ijkl} = (\sqrt{\tanh{\beta}})^{i_{xy}+i_{xz}+i_{xw}+i_{xv}}\delta(mod(i_{xy}+i_{xz}+i_{xw}+i_{xv},2)) このとき Tijkl を2次元の行列として表すと(４x４行列になります。添字は最初は０か１のみ数値を取るようになっております。) \left( \begin{array}{ccc} T_{0000} & \dots & T_{0011}\\ \vdots & \ddots & \vdots\\ T_{1100} & \dots &T_{1111} \end{array} \right) = \left( \begin{array}{cccc} 1&0&0&\tanh{\beta}\\ 0&\tanh{\beta}&\tanh{\beta}&0\\ 0&\tanh{\beta}&\tanh{\beta}&0\\ \tanh{\beta}&0&0&(\tanh{\beta})^2 \end{array} \right) となります。 この初期の行列を作成するソースコードは class initial_tensor(): def Delta_func(self,x): y= np.array([0.,1.],dtype='float64') if x ==0: return y[1] else: return y[0] def Tensor(self,K): K = torch.tensor(K,dtype=torch.float64,device=torch.device("cpu")) T = torch.zeros(2,2,2,2,dtype=torch.float64) for i in range(2): for j in range(2): for k in range(2): for l in range(2): T[i,j,k,l] += (torch.sqrt(torch.tanh(1/K))**(i+j+k+l) * self.Delta_func(np.mod(i+j+k+l,2))) T = T.view(2,2,2,2) return T になります。後のテンソルの扱いを楽にするため、4次元のテンソルとして扱っています。今後添字の数=次元として扱うようにしています。 　テンソルの分解 テンソルの分解は特異値分解 (SVD) を実行します。テンソルの分解方法は次の二通りあります。 これを式で表すと T_{ijkl} = \sum^{D^2}_{n=1}u_{ijn}\sqrt{s_n}\sqrt{s_n}v^{\dagger}_{nkl} = \sum^{D^2}_{n=1}(S_1)_{ijn}(S_3)_{kln}\\ T_{ijkl} = \sum^{D^2}_{n=1}u_{iln}\sqrt{s_n}\sqrt{s_n}v^{\dagger}_{njk} = \sum^{D^2}_{n=1}(S_2)_{iln}(S_4)_{jkn} とそれぞれ対応しております。 このときの、記号の対応は n = i+D(j-1)\\ k,j = 1,2,\dots,D\\ n = 1,2,\dots,D^2 となります。 Ma = T.view(D**2,D**2) ua,sa,va = torch.svd(Ma) Mb = T.permute(1,2,3,0).contiguous().view(D**2,D**2) ub,sb,vb = torch.svd(Mb) .viewというのはテンソルを任意の大きさに変更するものです。このときはランク４のテンソルをランク２のテンソルに変更します。このとき、要素数は一緒になります。 T.permute(1,2,3,0).contiguous()というのは、行列の入れ替えに対応します。今回はテンソルの各次元の順番を入れ替えました。たとえばij行とkl列に対応するのをjk行とli列に変更するのに一致します。行列の分解方法はしっかりと違うので注意してください。私はこれで悩みました。 　新しいテンソルを作成 イメージは この式は T_{abcd} = \sum^{ALL}_{i,j,k,l}(S_1)_{ila}(S_2)_{jib}(S_3)_{kjc}(S_4)_{lkd} で表せます。 ここでは def S_n(self,x,y,D,D_cut): u = (x[:,:D_cut]).view(D,D,D_cut) s = torch.sqrt(y[:D_cut]) return torch.einsum('kjm,m->kjm',u,s) 低ランク近似とトレース Dが大きくなっておくと、その分テンソルが大きくなってしまいます。そうなると、計算に時間が非常にかかってしまいます。そのため、途中で低ランク近似を行います。 T_{ijkl} = \sum^{D^2}_{n=1}(S_1)_{ijn}(S_3)_{kln} \approx \sum^{Dcut}_{n=1}(S_1)_{ijn}(S_3)_{kln}\\ T_{ijkl} = \sum^{D^2}_{n=1}(S_1)_{iln}(S_3)_{jkn} \approx \sum^{Dcut}_{n=1}(S_1)_{ijn}(S_3)_{kln} テンソルの分解と新しいテンソルの作成を繰り返し、途中で低ランク近似を行い続けると最終的に 残るテンソルは一つになります。それのテンソルのトレースが分配関数の近似になります。 プログラム全体 以上より、分配関数が求まったため、比熱を計算するとそのピークとなる温度が厳密解になります。 ソースコードを一括で表示すると import matplotlib.pyplot as plt import numpy as np import torch from torch import einsum from torch.linalg import svd import time class initial_tensor(): def Delta_func(self,x): y= np.array([0.,1.],dtype='float64') if x ==0: return y[1] else: return y[0] def Tensor(self,beta): beta = torch.tensor(beta,dtype=torch.float64,device=torch.device("cpu")) T = torch.zeros(2,2,2,2,dtype=torch.float64) for i in range(2): for j in range(2): for k in range(2): for l in range(2): T[i,j,k,l] += (torch.sqrt(torch.tanh(beta))**(i+j+k+l) * self.Delta_func(np.mod(i+j+k+l,2))) T = T.view(2,2,2,2) return T class TRG(): def S_n(self,x,y,D,D_cut): a = (x[:,:D_cut]).view(D,D,D_cut) b = torch.sqrt(y[:D_cut]) return torch.einsum('kjm,m->kjm',a,b) def Renorm(self,T,Dcut): D = T.shape[0] D_cut = min(D**2, Dcut) Ma = T.view(D**2,D**2) ua,sa,va = torch.svd(Ma) Mb = T.permute(1,2,3,0).contiguous().view(D**2,D**2) ub,sb,vb = torch.svd(Mb) # einsum 'ila,jib,kjc,lkd->abcd' S1,S2,S3,S4 return torch.einsum('ila,jib,kjc,lkd->abcd', (self.S_n(ua,sa,D,D_cut),self.S_n(ub,sb,D,D_cut), self.S_n(va,sa,D,D_cut), self.S_n(vb,sb,D,D_cut))) def Loop_func(self,T,loop,Dcut): norm = 0. for i in range(loop): T_new = T/T.abs().max() norm+= torch.log(T.abs().max())/(2**(i)) T = self.Renorm(T_new,Dcut) D = T.shape[0] logZ = torch.log(torch.trace(T.view(D*D,D*D))) return(logZ/2**loop) +norm Dcut = int(input("Dcut:")) loop = int(input("Number of renormalizations:")) s_beta = 0.4 e_beta = 0.5 step = 51 temp = [] logZ_list = [] for beta in np.linspace(s_beta,e_beta,step): T = initial_tensor().Tensor(beta) start = time.time() logZ = TRG().Loop_func(T,loop,Dcut) + np.log(2) + 2*np.log(np.cosh(beta)) run_time = time.time()-start print('beta : {:.4f},trace(logZ/V):{:.5f},run time:{:.2f}'.format(beta,logZ,run_time)) temp.append(beta) logZ_list.append(logZ) #temp.logZ_list def Energy(x,y): size = len(x)-1 E_list = [] E_Beta_list = [] for i in range(size): d_beta = x[i+1]-x[i] d_log = y[i+1] - y[i] E_list.append(-1*d_log/d_beta) E_Beta_list.append((x[i+1]+x[i])/2) return E_Beta_list,E_list def S_heat(x,y): size = len(x)-1 Cv_list= [] Cv_beta_list = [] for i in range(size): d_beta = x[i+1]-x[i] d_E = y[i+1] - y[i] beta = (x[i+1]+x[i])/2 Cv_list.append(-1*beta*beta*d_E/d_beta) Cv_beta_list.append(beta) return Cv_beta_list,Cv_list def Sort_heat(x,y): y_r = sorted(y) y_r.sort(reverse = True) a = y_r[0] b = y.index(a) c = x[b] a = a.to('cpu').detach().numpy().copy() return c,a E_Beta_list,E_list = Energy(temp,logZ_list) Cv_beta_list,Cv_list = S_heat(E_Beta_list,E_list) Max_beta ,Max_Heat = Sort_heat(Cv_beta_list,Cv_list) fig = plt.figure() plt.rcParams['ytick.direction'] = 'in' plt.rcParams['xtick.direction'] = 'in' x = Max_beta*1.5 y = Max_Heat/10 fig.add_subplot(xlim=(s_beta,e_beta),ylabel="Heat",xlabel="Beta") fig.text(x,y,'Max_heat:{:.4f}\nbeta : {:.2f}'.format(Max_Heat ,Max_beta ),size=12) plt.plot(Cv_beta_list,Cv_list) fig.savefig("Heat_{}_{}.png".format(Dcut,loop)) 上記プログラムでの def Loop_func(self,T,loop,Dcut): norm = 0. for i in range(loop): T_new = T/T.abs().max() norm+= torch.log(T.abs().max())/(2**(i)) T = self.Renorm(T_new,Dcut) D = T.shape[0] logZ = torch.log(torch.trace(T.view(D*D,D*D))) return(logZ/2**loop) +norm では、オーバーフローを防ぐためにくりこみをするテンソルTを規格化しています。それぞれのループではその時の体積で割ることにしています。 結果 上記プログラムの実行結果がこちらになります。 Dcut=32,くりこみ回数(loop)=20のときの結果です。およその実行時間は約７分ほどとなっております。(温度のメッシュは５１となっております。１度辺り７秒ほどかかっている計算になります。） 比熱は 次は β 表示ではないですが、くりこみ回数による違いは次のようになります。Dcut=32となっております。 比熱のピークがT = 2.27となっており厳密解=2.269...1に近い値となっております。 上の図と下図の赤い線を比較するとスケールの違いからかBetaでの計算よりケルビンでの計算のほうがスムーズに見えます。メッシュのスケールが違うからでしょう。 まとめ 今回は、主としてプログラムや数式とプログラムの対応を中心に記述しました。 なかなか、プログラムで書き起こすのは苦労しましたが、最終的な結果は満足する出来だと思います。 テンソルネットワークは機械学習、ディープラーニングなどにも応用は利くので、今後何かしらの応用したプログラムを組んでいきたいと思います。 参考文献 著者:武田真滋様　タイトル:テンソルくりこみ群による素粒子物理学の諸問題へのアプローチ 筑波大学のシンポジウムでの資料のようです。 著者:武田真滋様　タイトル:テンソルくりこみ群による素粒子物理学の諸問題へのアプローチ ↩

Python Win32プログラミング Win32 API自体、今では使う機会は少ないですが - Win32 API(C言語)で作成されたソフトをPythonに移植したり - exeしかないソフトに対してPythonでちょっとしたハックをしたり するのに役立つかもしれません。 参考にした記事 Python: ctypesパターン集 前提 Win32 APIの初歩的な知識 例：ウインドウを前面に出す ウインドウタイトルが完全にわかっている場合 使用するAPI - FindWindowExW - SetForegroundWindow 例えば、エクスプローラで"テスト"という名前のフォルダを開いているとき from ctypes import * FindWindowEx = windll.user32.FindWindowExW SetForegroundWindow = windll.user32.SetForegroundWindow hwnd = FindWindowEx(None, None, None, "テスト") if hwnd != None: SetForegroundWindow(hwnd) でエクスプローラが前面に出ます。 C言語で言うと、 #include <windows.h> int main(void) { HWND hwnd = FindWindowExW(NULL, NULL, NULL, L"テスト"); if (hwnd != NULL) { SetForegroundWindow(hwnd); } } このように、文字列はそのまま、NULLはNoneに置き換えます。 Win32 APIには～A,～Wと二種類存在する場合がありますが(この場合はFindWindowExA,FindWindowExW)、ASCII文字列しか使わない場合は～Aでも構いません。 FindWindowExWの仕様は、 https://docs.microsoft.com/ja-jp/windows/win32/api/winuser/nf-winuser-findwindowexw これを見るとエラー(ウインドウがみつからなかった)の場合はNULLを返すことがわかるので、 if hwnd != None:でエラーチェックしています。 部分文字列や正規表現で検索したい場合 使用するAPI - EnumWindows - GetWindowText - GetWindowTextLength - SetForegroundWindow FindWindowExは文字列が完全に一致しないとエラーになるので、この場合はEnumWindowsを使います。 EnumWindowsで現在開いているウインドウの一覧を取得し、コールバックにてウインドウタイトルと比較(GetWindowText)、目的のウインドウがみつかったら所望の処理をする、という流れになります。 from ctypes import * EnumWindows = windll.user32.EnumWindows GetWindowText = windll.user32.GetWindowTextW GetWindowTextLength = windll.user32.GetWindowTextLengthW SetForegroundWindow = windll.user32.SetForegroundWindow callback = WINFUNCTYPE(c_bool, POINTER(c_int), POINTER(c_int)) ghwnd = None def EnumWindowsProc(hwnd, lParam): global ghwnd length = GetWindowTextLength(hwnd) buff = create_unicode_buffer(length + 1) GetWindowText(hwnd, buff, length + 1) # print(buff.value) if buff.value.startswith("テス"): ghwnd = hwnd return False return True EnumWindows(callback(EnumWindowsProc), 0) if ghwnd != None: SetForegroundWindow(ghwnd) 一気に複雑になるので萎えますが、ポイントはウインドウタイトルを受け取るためのバッファをcreate_unicode_bufferで確保しているところです。文字列そのものは.valueに格納されています。 C言語では以下のようになるところですが、Pythonではかえって面倒くさくなってしまうのが難点です。 char buff[20]; GetWindowTextA(hwnd, buff, 20); printf("%s\n", buff);

センシティブなデータを暗号化してソースコードに埋め込む。 手順 暗号化文字列を作成する方法 import boto3 import base64 def main(): key_id = 'マスターキーのARN' Plaintext = '暗号化対象の文字列' response = kms.encrypt( KeyId = key_id, Plaintext = Plaintext )['CiphertextBlob'] Ciphertext = base64.b64encode(response).decode('utf-8') print('Ciphertext='+Ciphertext) if __name__ == '__main__': main() 上記処理で得られたCiphertextはKMSのマスターキーによって暗号化されている。 プログラム中で使用する場合は以下のようになる key_id = 'マスターキーのARN' Ciphertext = '上記処理で得られた暗号化文字列' response = kms_client.decrypt( CiphertextBlob=Ciphertext, KeyId=key_id ) plaintext = response['Plaintext'].decode('utf-8') KMSで暗号化しているため、仮にソースコードが流出したとしても、KMSのDecrypt API実行権限 がなければ機密情報が漏れる心配がなく安心ですね。 おわりに 暗号化文字列の複合化にKMSの権限が必要なため、プログラムを実行するサービスに対して 忘れずにKMS実行権限を付与しましょう。

はじめに 本記事は爆速（1日）でHerokuに幼稚園児並みのアプリをデプロイしていくまでの軌跡である。 Herokuへのデプロイの仕方が主で、Flaskについては幼稚園児並みにしか触れないので注意されたい。 環境 動作OS：Windows10 home ターミナル：Powershell 言語：Python 3 実施手順 venvでの環境構築 Herokuへのデプロイ パイプラインの実装 venvでの環境構築 任意のディレクトリに任意名のフォルダを作成する。 kindergartenというのは幼稚園という意味である。 幼稚園児でも理解できる記事にしていきたいと思う。 $ mkdir flask_kindergarten $ cd flask_kindergarten pythonでvenvを実行しアクティベートしていく。 WindowsやMac、Linuxでコマンドに差分があるので注意。 ターミナルがPowershellかCOMかでも差分があるので注意。 今回はWindows-Powershellでの実行だが環境差分がある人は、当該環境でvenvアクティベートの仕方を調べると良いだろう。 $ python3 -m venv venv $ Set-ExecutionPolicy RemoteSigned -Scope CurrentUser -Force $ venv/Scripts/Activate.ps1 Flaskのインストール。 $ python3 -m pip install Flask==1.1.2 herokuへデプロイする用のrequirements.txtを作成。venvは仮想環境ごとにpipのスコープを持つので仮想環境下でインストールしたものだけがテキストファイルに出力される。 $ python3 -m pip freeze > requirements.txt 次に、お待ちかねFlaskのメインのコードを書いていく。 MVCモデルで言うところのC=Controllerにあたり、アプリの背骨の部分である。 現在のディレクトリにapp.pyを作成。 $ type nul > app.py 今のディレクトリ構成はこんな感じ。 flask_kindergerten/ │ ├── venv/ │ ├── app.py └── requirements.txt では、app.pyを記述して一度Flaskを立ち上げていこう。 app.py from flask import Flask app = Flask(__name__) @app.route("/") def index(): return "お前、どこ園だ、バブゥ！？" $ flask run ローカルで立ち上がったWebサーバを覗きに行く。デフォルトでは127.0.0.1:5000/であった。 ブラウザに遷移して「お前、どこ園だ、バブゥ！？」と表示されていたら勝ちである。 これは不良中学生の「お前、どこ中だ、オラァ！？」のオマージュである。 コードに変更があった場合にいちいちWebサーバを落として再起動しなくても済むように以下のコマンドを打っておく。 例によって、環境差分があるので注意。 $ $env:FLASK_ENV = 'development' 次に、herokuにデプロイするためにgitの設定をしていく。 'git init'でレポジトリの初期化を行ったあとに'.gitignore'でコミットしないファイルを指定する。.gitignoreはvenvや__pycache__のほかにSECRET KEYを書いたconfigファイルとかを指定する場合があると思う。 また、'>' この大なりの記号は出力リダイレクトといって出力結果をリダイレクト先に書き込むことができる。 '>>'で追記となる。 $ git init $ echo venv > .gitignore $ echo __pycache__ >> .gitignore $ git add .gitignore app.py requirements.txt $ git commit -m "my first commit" Herokuへのデプロイ まず、Herokuに未登録の人はこちらから登録しよう。 次に、CLIを使うためにこちらからインストールしよう。 ここまでできたらHerokuにログインしよう。 $ heroku login 次に、Herokuにデプロイする準備をする。 まずはHerokuの設定ファイル'Procfile'を作成していく。 このProcfileだが、間違えると痛手になる。 具体的には以下２つには気をつけて欲しい。私はこれに2時間くらい足踏みし、結果として爆速味が失われてしまった。 最初のPは必ず大文字にすること UTF-8でエンコードしていること 以下のようにProcfileを作成する。 Procfile web: gunicorn app:app 次に、PythonのWSGIであるgunicornをインストールする。 WSGIとはWeb Server Gateway Interfaceの略であり、アプリとサーバの差分をなくして標準化するようなインターフェースだそうだ。 $ python3 -m pip install gunicorn==20.0.4 $ python3 -m pip freeze > requirements.txt gunicornをインストールしたので、requirements.txtを更新した。 では、追加分をgitに反映する。 $ git add Procfile requirements.txt $ git commit -m "Add Heroku deployment files" Herokuにリポジトリを作成して、git pushしていく。 いつもは'origin'でpushしてるところは'heroku'に変更になるから注意だ。 あと、Herokuはアンダーバーが使用不能なのでリポジトリ名をハイフンで代用する。 $ heroku create flask-kindergarten $ git push heroku master がんばってクラウド上のサーバでデプロイしてまっせ、みたいなメッセージ群が流れ終わったら完了だ。 では、実際に確認してみよう。 $ heroku open Herokuのドメインで先の幼稚園児アプリが確認できたらOKだ！ パイプラインの実装 Herokuへのデプロイまでなら前項で終了なのだが、それだとあまりに幼稚園児すぎるので、せめて小学生に顔向けできるように最後にパイプラインを構築する。 パイプラインというのは複数の環境をワークフローで結んだグループのことである。 複数の環境というのは、主に開発環境・ステージング環境・本番環境のことである。 ここまでで使用した環境はローカルの開発環境とHerokuの本番環境だけだが、これだと実運用では困るだろう。 例えば、往々にして開発環境と本番環境には環境差がある。 開発環境ではうまく作動した機能が本番環境でもうまく作動するかわからない。なんなら、機能追加してしまったせいで本番環境がうまく動作しなくなるなんてこともあるだろう。 そんな時に役立つのがステージング環境である。 ステージング環境は本番環境と同様の環境をつくり、ここで新機能などのテストを行うのだ。 各環境のイメージは以下。 これらを結ぶものがワークフローであり、ワークフローによってグループ化されたものがパイプラインだ。 それでは、早速ステージング環境をHerokuに構築していこう。 $ heroku create flask-kindergarten-staging --remote staging $ git push staging master https://flask-kindergarten-staging.herokuapp.com/ みたいな名前で本番環境と同じものが確認できればOKだ！ では、次にパイプラインを作成していく。 flask-pipelineというパイプラインに本番環境としてflask-kindergartenのアプリを登録する。 $ heroku pipelines:create --app flask-kindergarten ` --stage production ` flask-pipeline 次に、gitを反映させる。 gitでの名前はprodとする。 $ heroku git:remote --app flask-kindergarten --remote prod そして、ステージング環境としてflask-kindergarten-stagingのアプリをパイプラインに加えていく。 $ heroku pipelines:add flask-pipeline \ --app flask-kindergarten-staging \ --stage staging では、先ほどのapp.pyの中身を少し変更し、ステージング環境だけに反映してみよう。 app.py from flask import Flask # make app instance app = Flask(__name__) # routing @app.route('/') def index(): return 'あなた様はどこ園かしら？' なんとなく女子にしてみた。 それでは、ステージング環境にファイルをcommitしていく。 $ git add app.py $ git commit -m "changed" $ git push staging master ブラウザでステージング環境と本番環境のURLを叩いて、別々の表示になっていればOKだ。 最後に、ステージング環境の変更を本番環境へパイプラインを用いて反映させる。 $ heroku pipelines:promote --remote staging 本番環境にも変更が反映されていれば完了だ！ おわりに 今回は爆速でFlaskアプリをHerokuにデプロイしたわけだが、Flaskアプリ部分があまりにも幼並完（幼稚園児並みの完成度）であった。 今度はちゃんとHTMLなりを実装し、まともなFlaskアプリを構築してみようと思う。 ではまた！ 参考 Deploying a Python Flask Example Application Using Heroku

※備忘録的に記述します。 RailsでもPythonでもそうですが、エラーが発生したときにやりがちなことが 「エラーの文言をそのままGoogle検索」 だと思うんですが、はっきり言ってこれは間違っている。 まず最初にやることはエラーの内容を理解して、どこが問題なのかを理解すること。 JavaScriptが読み込まない Nomethodと出る どんなエラーでも同じですが、 ①どのようなエラーが発生しているのか ②何が原因なのか ということをきちんと把握することが大事。 プログラミング初心者だと、ついエラーから逃げがちだけども、一番大事なことはここなのかと。 いろんな言語を経験しているわけではないので、全てではないですが、どの言語においても共通していることだと思います。 まずは真剣にエラーと向き合う。 どこでエラーが発生しているのか、原因となっている箇所は何なのか、どのようなエラー内容なのか、ということを正確に把握することが大事。 ということをすごく学びました。

MacでPythonのバージョンを2系(2.7.10)から3系(3.9.5)にアップデートしていたのですが、ターミナルを終了して再度開くとバージョンが戻る現象が発生しました。 解消してみると、くだらないオチだったのですが、経験があまりない方々は割と詰まってしまいそうな気がしましたので、メモとして記載(共有)しようと思います。 想定読者 MacでPythonのアップデートを行われている方を想定致します。 また、アップデート時にバージョンが変わらないという方の参考にもなるかと思います。 ちなみに私はアップデートを行う際、以下の記事を参考にさせて頂きました。 MacのPythonを2系から3系にアップデートする MacのPython2系からPython3系に3分で切り替える方法 補足: シェルの環境設定(.bash_profileへの書き込み)は2の記事を参考にしました。 再現するまで 参考にした記事の通り、 Homebrewでpyenvをインストール pyenvの設定のため、.bash_profileを修正 pyenvを使用して最新(対応時は3.9.5)のPythonをインストール pyenv global 3.9.5 コマンドをターミナルで実行してPythonのバージョン切り替え を行いました。 pyenv versions を実行すると、 $ pyenv versions system * 3.9.5 (set by $HOME/.pyenv/version) pyenvのPythonのバージョンの切り替えは問題なし。 python --version でPythonのバージョンも確認すると、 $ python --version Python 3.9.5 こちらもOK。 しかし、ターミナルを終了して再度起動し、 python --version すると、 $ python --version Python 2.7.10 元のバージョンに戻りました。(pyenv versionsの方は更新が正常に反映されていました) 原因 私のPCのシェルがbashではなくzshを使っていたことが原因でした。(echo $SHELL を実行すると「/bin/zsh」と表示されました。) どうやら、シェルの環境設定(「再現するまで」の2の手順)の修正対象ファイルを誤っていたようです。 対策 参考にした記事で.bash_profileに追記されていた下記の設定を.zshrcに追記しました。 export PATH="$HOME/.pyenv/shims:$PATH" これで、ターミナルを再起動してもpythonのバージョンは元に戻らなくなりました。 シェルの環境設定の際は、まず、 echo $SHELL をターミナルで実行して、自分のPCのシェルが何を使っているか確認しないといけませんね... 最後に 「何かをする時は、まず己自身を知りなさい。」 そんな声が聞こえた気がしました。 良い教訓でした、反省。。。

はじめに 以前、社内のRedashをEC2からECSに移行してv5からv8にした話という記事を投稿しました。redashの古いバージョンを最新まで上げつつ、EC2からECS Fargateに載せ替えたという内容です。 当時の記事ではバージョンアップ方法や環境移行方法に焦点を当てていた為、今回はECS Fargateでredashを利用する方法、メリットについてまとめたいと思います。また、当社ではこの環境でPythonを用いており、記事の後半ではPython利用時の方法、ライブラリの扱いについても記載します。 1. 当社の環境 上図は当社のredash環境を簡単に表した図です。redashはFargateでホスティングしており、redash_serivceの中でredash_worker、redash_serverコンテナが稼働しています。redashのダッシュボードやクエリ等のメタデータを管理するPostgreSQLはRDS、キューイングを管理するRedisはElastiCacheと、それぞれマネージドサービスに逃しています。図では割愛していますが、別で設定したALBのターゲットグループにコンテナを指定して利用しています。 1.1 この環境のメリット 移行前はredash本体、Redis、PostgreSQLを全て同一のEC2インスタンスでホスティングしていました。加えてredashで参照する分析用DBのホスティングや同DBへのマスキング処理も同じインスタンスで行っていました。お察しの通り、たびたびリソースが逼迫してredashが機能しなくなっていました。 Fargate環境にしてからはそういった事は起きていません。redash_serviceを登録する際の必要タスク数を1としており、仮にコンテナが落ちても必要なタスク数に応じて起動し直してくれます。その際メタデータやキューイングデータはそれぞれマネージドサービスに逃しているので運用上問題はありません。いつredashが落ちるか分からない、落ちたら手動でどうにかしなければならない、といった悩みの種を駆逐できる恩恵は非常に大きなものだと言えます。 2. 環境の詳細 2.1 利用サービスの一覧 Secrets Manager ALB RDS ElastiCache ECS ECSで用いるredashコンテナは公式のイメージを利用し、バージョンは現行(2021年6月現在)の最新(redash/redash:8.0.0.b32245)です。 2.2 設定や注意点 2.2.1 Secrets Manager RDSやElastCacheの認証情報をSecrets Managerに保存しておき、ECSのタスク定義時に呼び出すことが可能です。変数入力欄に認証情報を直接記述しないで済みます。詳細は公式ドキュメントをご覧下さい。 2.2.2 ALB 利用する場合は事前に設定する必要がありますが、Route53を用いて任意のレコード名とredashを紐付けできたり、あれば何かと便利です。基本的には環境に応じて作成すれば問題ありませんが、手順4のルーティング設定ではターゲットの種類をIPで設定しましょう。ヘルスチェックは/pingで通ります。 当然ですが、VPC・サブネット・セキュリティグループ等の周辺も事前に設計、設定しておく必要があります。 2.2.3 RDS redashのメタデータを保存するDBとしてPostgreSQLインスタンスを作成する必要があります。環境にも依存するかとは思いますが、そこまで大きなインスタンスタイプにする必要はなさそうな印象です。スモールスタートして、リソースが足りなさそうであれば少しずつインスタンスタイプを変更していく運用で良いかと思います。IDやPWはECSタスク定義時に利用するので、控えておくかSecrets Managerに保存しましょう。 参考までに、当社のredashは以下の様な利用状況ですが、PostgreSQLインスタンスはdb.t3.mediumタイプで安定稼働しています。 2.2.4 ElastiCache Redisでクラスターを作成します。スモールスタートする場合は、まずは1台構成からでも良いかもしれません。デフォルトだと大きめのノードタイプが選択されているので注意が必要です。 2.2.5 ECS クラスター作成 利用するクラスターを作成します。Fargateを利用する場合はネットワーキングのみで問題ありません。 タスク定義 当社ではECSでredashを利用する場合のタスク定義を2種類用意しています。初回構築時のみ利用する、Redis及びPostgreSQLの利用準備を行うcreate_dbタスクとredash本体を稼働させるredashタスクです。 create_dbタスク create_dbタスクはredashのCLIで/app/manage.py database create_tablesコマンドを発行するタスクです。コマンドの内容はここに記載されており、この辺りに記述されているテーブルを作成するものと思われます。このコマンドはredashのdocker-composeなどでも呼び出されているものです。 タスク定義の詳細は次の通りです。 タスクメモリを0.5GiBに設定したところコンテナが落ちてしまったので3GiBとしています。 コンテナ定義は次の通りです。 環境の欄にcreate_dbを記入します。 環境変数の欄にREDASH_DATABASE_URL(PostgreSQLの認証情報)、REDASH_REDIS_URL(Redisの認証情報)を記入します。記法は以下の通りですが、Secrets Managerを利用する場合はそちらの値を記入します。記入を終えたら完成です。 postgresql://user:password@hostname:port/redash redis://hostname:port/0 タスク定義が完成したら、クラスターの新しいタスクの実行から実行しましょう。このタスクは初回構築時にのみ必要となるので、サービスとして実行する必要はありません。 redashタスク こちらはECSクラスターのサービスとして起動することになる、redash本体のタスクです。当社ではserverとworkerの2つのコンテナを立ち上げています。定義は次の通りです。 serverコンテナ コマンド欄にserverと記入。各種変数は環境に合わせて読み替えてください。 workerコンテナ コマンド欄にscheduler,workerと記入。各種変数は環境に合わせて読み替えてください。 こちらのタスクはサービスとして登録したいので、サービスの作成を押下して設定します。 サービスの設定時、パブリックIP割り当ての項目があります。ALB経由の設定を行う場合は必要はありませんが、redashが起動しているのか確認する等のデバッグ用であれば割り当てても良いかもしれません。 3. redash on ECS環境でPythonを使う redashでPythonを利用する為にはredash本体での設定とECSタスク定義での設定がそれぞれ必要です。ライブラリの導入も同じ手順で行います。 注意点 2021年6月現在、redash v8で利用できるのは既にサポートが終了したPython2系です。v9になるとPython3が利用できるようなので正式リリースが非常に楽しみです。 3.1 ECS側の設定 実施する設定は、redashでデータソースとしてPython自体を利用可能にするものと、Pythonで利用するライブラリをコンテナにインポートするものの2種類があります。 3.1.1 Pythonデータソースを利用可能にする タスク定義にて、server、workerそれぞれのコンテナの環境変数欄にREDASH_ADDITIONAL_QUERY_RUNNERSを追加し、redash.query_runner.pythonを記入します。 作成後はリビジョンを新しいものにしてタスクを再起動すればredashでPythonが利用可能になります。 3.1.2 Pythonライブラリを導入する 分析を行う中で利用したいライブラリが出てくる場合もあるかと思いますが、その場合もタスク定義にて導入することになります。 タスク定義のworkerコンテナのコマンド欄にて、worker、schedulerに続いてインストールコマンドを記入します。 記法はbash -c 'pip install [ライブラリ名]==[バージョン] --user'です。複数導入する場合はカンマを挟んで続けて記入します。上記はstatsmodelsのv0.9.0とscipyのv0.14を導入する例です。作成後はリビジョンを新しいものにしてタスクを再起動すれば記述したライブラリがコンテナにインストールされます。 3.2 redash側の設定 データソース設定からPythonを作成し、利用予定のライブラリをModules to import prior to running the scriptに記入します。AdditionalModulesPathsの欄には、インポート先である/home/redash/.local/lib/python2.7/site-packagesを記入します。 まとめ redash on ECS Fargate環境を構築する方法についてまとめました。この環境であればそこまでリソースに対して気を配る必要がなくなるため、運用上の負荷が軽減されます。後半ではpythonを利用する方法についてまとめました。redash上でPythonを利用できるようになると分析の幅が広がっていく為、非常に有用だと思います。反面、段々と増えていくダッシュボードやクエリの管理が煩雑になってしまう点については新たな課題となってしまいました。これに関しては良い解決策を発見次第、記事にまとめたいと考えています。

画像データ（フォルダ内の一連のjpegファイル）をモノクロ化して取得日時を書き込み、別のフォルダに出力します。 ファイルのプロパティを見ると、「変更日時」が取得日時に相当していたので、こちらのコードからちょっと変えます。 Windows 10、Python 3.8.5で実行しました。 英数字だけを書き込んでいますが、OpenCVのデフォルトのフォントではなくArialを使いたかったので、日本語の書き込み方を参考にしました。 Python import cv2 as cv import numpy as np import PIL from PIL import ImageFont, ImageDraw, Image import os import glob import datetime # 出力先フォルダを指定 output_path = './output/' # ファイルの変更日時を取得 def get_exif(image_path): t = os.path.getmtime(image_path) d = datetime.datetime.fromtimestamp(t) captured_datetime = d.strftime("%Y/%m/%d %H:%M:%S") return captured_datetime # 各ファイルに変更日時を書き込む def write_text(input_path): captured_datetime = get_exif(input_path) image_name = os.path.split(input_path)[1] img = cv.imread(input_path) # モノクロ化 img_gray = cv.cvtColor(img, cv.COLOR_BGR2GRAY) # 文字入れ fontpath ='C:/Windows/Fonts/arialbd.ttf' #Arial Bold font = ImageFont.truetype(fontpath, 64) # フォントサイズ64 # 表示位置：画像に合わせて調整 position = (30, 1820) img_pil = Image.fromarray(img_gray) draw = ImageDraw.Draw(img_pil) draw.text(position, captured_datetime, font = font , fill = 255) # fill = 255で白 img_gray = np.array(img_pil) # PIL を配列に変換 cv.imwrite(os.path.join(output_path, image_name), img_gray) # ファイル名は同じ # 元フォルダから一括変換 # 一括変換する前に1枚テストしてフォントサイズや位置を確認しましょう for f in glob.glob('./input/*.jpg'): write_text(f) 出力結果イメージ 文字部分だけ切り取っています。 参考 Python, OpenCV, NumPyでカラー画像を白黒（グレースケール）に変換 画像にテキストを書き込む方法 Pythonでファイルの作成・更新日時を取得する（os.path.getmtimeなど）：作成日時の取得はOSごとに変わるので注意 TypeError: Can't convert 'datetime.timedelta' object to str implicitly strftime() と strptime() の振る舞い OpenCV 3.4 + Python3 文字を描画する Python + OpenCV でWindowsの日本語フォントを描画する pythonでフォルダ内のファイル名のみ取得する方法

初版: 2021年6月7日 著者: 橋本恭佑、柿田将幸, 株式会社 日立製作所 はじめに 本投稿では、学習済みの機械学習モデルをAWS/Azureが提供するコンテナイメージを用いてサービングする方法を紹介します。 Amazon SageMakerとAzure Machine Learning (Azure ML)で実際にサービングを試して分かった比較結果を解説します。 今回の検証では、AWS/Azureが提供するコンテナイメージをコンテナ基盤にデプロイするときに、必要なライブラリの追加またはバージョン変更を行い、作成済みの前処理・モデル・後処理を追加する方法(下記表1の赤枠部分)を紹介します。 表1: オンプレミス環境で学習させた機械学習モデルをAWSおよびMicrosoft Azure上でサービングする際のサービス一覧(2021年2月現在) 投稿一覧 AWSとMicrosoft Azureにおける機械学習モデルのサービング技術の概要 Amazon SageMakerとAzure MLにおける機械学習モデルのサービング技術比較: AWS/Azureが提供するコンテナイメージを利用する場合・・・本投稿 Amazon SageMakerとAzure MLにおける機械学習モデルのサービング技術比較: 自作コンテナイメージを利用する場合・・・近日公開 AWS FargateとAzure Container Instancesにおける機械学習モデルのサービング技術比較・・・近日公開 AWSとAzureにおけるパターン1の実現技術の違い パターン1を実現する場合の手順を図1に、パターン1を実現する場合に利用する技術をAWSとAzureで比較した結果を表2に示します。 Amazon SageMakerとAzure MLの双方で、機械学習モデルのサービングに必要なサービスが提供されておりますが、AWSには手順2を実現する手段がないことや、モデルのアップロード先が双方のパブリッククラウドで異なるなどの差があります。本稿ではこの手順に沿ってパターン1が実現可能かを実機検証した結果を紹介します。 図1: パブリッククラウドにおいてパターン1を実現する場合の手順 表2: AWSとAzureでパターン1を実現する場合に利用する技術の比較 項番 手順 AWSで手順を実現する技術 Microsoft Azureで手順を実現する技術 1 コンテナイメージを用意する 機械学習モデルの用途に合ったコンテナイメージをAmazon ECR(Elastic Container Registry)から検索する 機械学習モデルの用途に合ったコンテナイメージをAzure Container Registryから検索する 2 不足するライブラリがあれば追加する 対応技術なし(Amazon SageMakerのノートブックインスタンス上でライブラリを追加できない) Azure MLのノートブックインスタンス上でライブラリを追加する 3 モデルと前処理・モデル・後処理を呼び出す推論コードをアップロードする Amazon S3にモデルをアップロードし、Amazon SageMakerのノートブックインスタンス上に推論コードをアップロードする Azure MLのノートブックインスタンス上にモデルと推論コードをアップロードする 4 コンテナイメージ・モデル・推論コード・コンテナ基盤を指定しコンテナを立ち上げる Amazon SageMakerのノートブックインスタンス上でコンテナイメージ・モデル・推論コード・コンテナ基盤を指定する Azure MLのノートブックインスタンス上でコンテナイメージ・モデル・推論コード・コンテナ基盤を指定する 本検証の事前準備 本投稿では、以前の投稿の検証シナリオで作成した下記ファイルを利用しますので、事前に用意してください。 学習済みの手書き文字認識モデル: model.joblib 前処理・モデル利用・後処理を記載したスクリプト: api_server.py Amazon SageMakerでサービングする場合 学習済みの手書き文字認識モデルを、AWSが提供するコンテナイメージでサービングできるか検証しました。 Amazon SageMakerのノートブックインスタンスを起動して、作成した機械学習モデルをノートブックインスタンスへアップロードします。 それから、下記コードをノートブックインスタンス上のJupyter Notebookから実行すると、ノートブックインスタンスからS3ストレージへ機械学習モデルをアップロードできます。下記コードを利用せずに機械学習モデルを直接S3ストレージへアップロードしても問題ありません。 from sagemaker import get_execution_role from sagemaker.sklearn.model import SKLearnModel import sagemaker sagemaker_session = sagemaker.Session() role = get_execution_role() # ノートブックインスタンスを起動したときにアタッチされるS3ストレージのimportディレクトリへ、 # 機械学習モデルをアップロードする uploaded_model = sagemaker_session.upload_data(path='model.tar.gz', key_prefix='import') print(uploaded_model) # 実行後、機械学習モデルがアップロードされたディレクトリ名を確認できる。例えば下記など。 # 's3://sagemaker-ap-northeast-1-aws_account_id/import/model.tar.gz' 次に、以前の投稿に示した推論用のソースコード(api_server.py)を、SageMakerで実行可能な形式に書き換えます。下記に書き換え後のソースコード(entry.py)を示します。 entry.py import os import joblib from sklearn.neural_network import MLPClassifier import numpy as np import sys from six import BytesIO from PIL import Image def model_fn(model_dir): clf = joblib.load(os.path.join(model_dir, "model.joblib")) return clf def _npy_loads(data): """ Deserializes npy-formatted bytes into a numpy array """ stream = BytesIO(data) return np.load(stream) def input_fn(request_body, request_content_type): if request_content_type == 'application/x-npy': return _npy_loads(request_body) elif request_content_type == "image/jpeg": image_number = np.array(Image.open(request_body).convert('L')) array = [image_number.reshape(RESHAPED)] return array else: raise ValueError('Content type must be application/x-npy') def predict_fn(input_data, model): prediction = model.predict(input_data) return prediction[0] ここでポイントとなるのは以下の2点です。 1. 機械学習モデルのロード、入力データの変換、推論をmodel_fn, input_fn, predict_fnと各々で別の関数に分けて書きます。前処理はinput_fn、後処理はpredict_fnへ記入することで、それぞれ前処理と後処理を推論コンテナへ実装できます。元のapi_server.pyから書き換える必要があります。 2. input_fn関数の引数に、入力データ(request_body)だけでなく、入力データで受け付けるデータの形式(request_content_type)をとります。そして、入力データで受け付けるデータの形式(たとえばapplication/x-npyやimage/jpeg)ごとに前処理を書きます。こうすることで、サービングした後の機械学習モデルが複数の形式の入力データを受け付けることが可能となります。 次に、AWSが提供するコンテナイメージの中から、今回サービングに利用するOSSの入ったコンテナイメージをデプロイします。AWSではApache MXNet、tensorflow、scikit-learnなど複数の機械学習向けOSSについて、推論環境に利用できるコンテナイメージを提供しています。今回はscikit-learnのサービング用コンテナを利用しました。前処置・後処理・モデルを内包するscikit-learnの入った推論用コンテナをデプロイするには、ノートブックインスタンス上のJupyter Notebookから下記のコードを実行し、上記の推論用コード(entry.py)・モデルのファイル・scikit-learnのバージョンを指定します。 from sagemaker.sklearn.model import SKLearnModel # SKLearnModelのコンテナイメージを呼び出して、推論用コードを内包させる sklearn_model = SKLearnModel(model_data='s3://sagemaker-ap-northeast-1-aws_account_id/import/model.tar.gz', role=role, entry_point="./entry.py", framework_version="0.20.0") # 推論用コードを内包させたコンテナイメージをデプロイする predictor = sklearn_model.deploy(instance_type="ml.c4.xlarge", initial_instance_count=1) これでサービング処理の記述は完了です。推論コンテナのデプロイが成功するかを確認します。 # CloudWatchから確認できるログ Traceback (most recent call last): File "/miniconda3/lib/python3.7/site-packages/sagemaker_containers/_modules.py", line 258, in import_module module = importlib.import_module(name) File "/miniconda3/lib/python3.7/importlib/__init__.py", line 127, in import_module return _bootstrap._gcd_import(name[level:], package, level) File "<frozen importlib._bootstrap>", line 1006, in _gcd_import File "<frozen importlib._bootstrap>", line 983, in _find_and_load File "<frozen importlib._bootstrap>", line 967, in _find_and_load_unlocked File "<frozen importlib._bootstrap>", line 677, in _load_unlocked File "<frozen importlib._bootstrap_external>", line 728, in exec_module File "<frozen importlib._bootstrap>", line 219, in _call_with_frames_removed File "/miniconda3/lib/python3.7/site-packages/fail_entry.py", line 7, in <module> from PIL import Image ModuleNotFoundError: No module named 'PIL' 上記の実行結果より、AWSが提供するコンテナイメージには画像の前処理に用いるPillowのライブラリが含まれず、処理が完了していないことがわかります。エラーを解消するために、Pillowと同等の機能を持つ画像処理ライブラリ(例えばscikit-image)を利用可能か試しましたが、2021年2月現在、AWSの提供するscikit-learn用イメージコンテナには、scikit-imageなどの画像処理ライブラリが含まれていませんでした。 また、コンテナイメージをコンテナ基盤にデプロイするときに、pip, condaといったパッケージ管理ツールやAWSが提供するAPIを用いて必要なライブラリ(今回の場合は例えばpillowなど)を追加できるかを調査しましたが、2021年2?現在、そのような機能は提供されていませんでした。 以上の検証から、AWSにおいて「パブリッククラウドが提供するコンテナイメージをコンテナ基盤にデプロイするときに、ライブラリの追加またはバージョン変更して、前処理・モデル・後処理を追加する」ことはできないことがわかりました。 Azure MLでサービングする場合 学習済みの手書き文字認識モデルを、Azureが提供するコンテナイメージでサービングできるか検証しました。 Azure Machine Learningのノートブックインスタンスを起動し、ノートブックインスタンスへ作成した機械学習モデルをアップロードします。アップロードはOSSのJupyterと同様に、ブラウザからアップロード可能でした。 次に、前回の投稿に示した推論用のソースコード(api_server.py)を、Azure Machine Learningで実行可能な形式に書き換えます。下記に書き換え後のソースコード(score.py)を示します。 score.py import os import joblib import numpy as np import json from PIL import Image from io import BytesIO import base64 # 推論開始時に呼び出される関数 def init(): global model # モデルのPATHを指定 model_path = os.path.join(os.getenv('AZUREML_MODEL_DIR'), 'model3.joblib') # モデルをロード model = joblib.load(model_path) # 推論リクエストがあった場合の処理を定義 def run(raw_data): try: # JSONリクエストのtext propertyを抽出 data = json.loads(raw_data)['data'] img = base64.b64decode(data) # base64に変換された画像データを元のバイナリデータに変換 img = BytesIO(img) # _io.BytesIO pillowで扱えるように変換 RESHAPED = 784 image_number = np.array(Image.open(img).convert('L')) array = [image_number.reshape(RESHAPED)] prediction = model.predict(array) # このpredictionの型はarray型 return str(prediction[0]) # array型がJSONでシリアライズ不可のためstr型に変換 except Exception as e: error = str(e) print('array: {0}'.format(array)) return json.dumps({"error": error}) ここでポイントとなるのは以下の2点です。 1. 機械学習モデルのロードと、入力データの変換・推論をinit, runの2つの関数に分けて書きます。Amazon SageMakerにおける書き換え時と比べて、前処理や後処理を厳密に分ける必要がないといえます。 2. Azure MLの仕様上、run関数の入力データ(raw_data)および出力データの型はJSONのみです。したがって、今回の様に画像を扱う場合は、JSON化された入力データに含まれる画像のデータを読み込む前処理が必要です。同様に、推論結果もJSONでシリアライズ可能な形式で出力する必要があります。 次に、推論用のスクリプト(score.py)と、推論に利用するライブラリ群を紐づけた、前処理・後処理・ライブラリに関連するコンフィグ(inference_config)のオブジェクトを生成します。 Azure Machine Learningのノートブックインスタンス上のJupyter Notebookで下記のコードを実行します。 from azureml.core import Workspace from azureml.core.model import Model from azureml.core import Workspace, Environment ws = Workspace.from_config() from azureml.core.model import InferenceConfig from azureml.core.environment import Environment from azureml.core.conda_dependencies import CondaDependencies myenv = Environment(name="oss_center") conda_dep = CondaDependencies() # 推論に利用するライブラリ群を定義 conda_dep.add_conda_package("numpy") conda_dep.add_conda_package("scikit-learn") conda_dep.add_conda_package("pillow") conda_dep.add_conda_package("joblib") myenv.python.conda_dependencies=conda_dep # 推論用のスクリプトと推論に利用するOSS群を紐づけた、前処理・後処理・ライブラリに関連するコンフィグ(inference_config)のオブジェクトを生成 inference_config = InferenceConfig(entry_script="score.py", environment=myenv) 最後に、推論のためのコンテナのスペックのコンフィグ(deployment_config)のオブジェクトを生成し、上記で生成した、前処理・後処理・ライブラリに関連するコンフィグ(inference_config)と紐づけて、推論のためのコンテナをデプロイします。 Azure Machine Learningのノートブックインスタンス上のJupyter notebookから下記のコードを実行すると、推論のためのコンテナをAzure Container Instancesで起動できます。 from azureml.core.webservice import AciWebservice, Webservice # 推論のためのコンテナのスペックのコンフィグのオブジェクトを生成 deployment_config = AciWebservice.deploy_configuration(cpu_cores = 1, memory_gb = 1) model = Model(ws, name='hash-mnist') # 前処理・後処理・ライブラリに関連するコンフィグ(inference_config)とdeploymet_configを紐づけて、推論のためのコンテナをデプロイ service = Model.deploy(ws, 'hash-mnist-first', [model], inference_config, deployment_config) service.wait_for_deployment(True) print(service.state) print("scoring URI: " + service.scoring_uri) # 下記の様な出力が得られることを確認する # Running............................................................................ # Succeeded # ACI service creation operation finished, operation "Succeeded" # Healthy # scoring URI: http://7032b584-13d9-4bcb-a096-9fe0d9dc4941.japaneast.azurecontainer.io/score これでサービング完了です。推論が成功するかを確認します。SageMakerの場合と違って、画像送信時に画像をJSONで送信可能な型に変換する必要があることに注意してください。 import requests from PIL import Image import json import base64 from io import BytesIO img = Image.open("x_test_50.jpeg") # PillowImageをbytesに変換してさらにbase64に変換 buffered = BytesIO() img.save(buffered, format="JPEG") img_byte = buffered.getvalue() # bytes img_base64 = base64.b64encode(img_byte) # base64でエンコードされたbytes ※strではない # まだbytesなのでjson.dumpsするためにstrに変換(jsonの要素はbytes型に対応していないため) img_str = img_base64.decode('utf-8') # str files = { "data":img_str } response = requests.post( 'http://7032b584-13d9-4bcb-a096-9fe0d9dc4941.japaneast.azurecontainer.io/score', json.dumps(files), headers={'Content-Type': 'application/json'}) pprint.pprint(response.json()) # ‘6’と返答が返ってくることを確認する これで動作検証が完了し、サービングが成功したことを確認できました。以上の検証結果から、Azure Machine Learningでは「パブリッククラウドが提供するコンテナイメージをコンテナ基盤にデプロイするときに、ライブラリの追加またはバージョン変更して、前処理・モデル・後処理を追加する」が可能であることを確認できました。 検証結果の考察 パブリッククラウドの提供する汎用のコンテナイメージに推論に必要なライブラリが含まれない場合も、後からライブラリの追加が可能であることから、「パブリッククラウドが提供するコンテナイメージをコンテナ基盤にデプロイするときに、ライブラリの追加またはバージョン変更して、前処理・モデル・後処理を追加する」パターンの実現手段はAzure Machine Learningに限られることがわかりました。この手法は、初回デプロイ時に追加で開発する項目が最も少なく、モデル訓練後からランタイム作成までのリードタイムを短くできるため、顧客にすぐに推論システムを提供する必要あるユースケースに最も適しています。 おわりに 次回の投稿において、パブリッククラウドが提供するコンテナイメージを利用せず、オンプレミス環境で必要なライブラリを含んだコンテナイメージを作成して、コンテナ基盤へデプロイするときに、前処理・モデル・後処理を追加する手法について、Amazon SageMakerとAzure Machine Learningで実証した結果を解説します。

インストール php pi@raspberry:~ $ sudo apt install php php php-cli php-curl pi@raspberry:~ $ sudo a2enmod userdir pi@raspberry:~ $ sudo systemctl restart apache2 MariaDB pi@raspberry:~ $ sudo apt-get install mariadb-server-10.0 python pi@raspberry:~ $ pip install mysql-connector-python pi@raspberry:~ $ pip install mysqlclient ソースコード pi@raspberry:~ $ cd /var/www/html pi@raspberry:/var/www/html $ git clone https://github.com/kawanishi291/proen pi@raspberry:/var/www/html $ cd ~ pi@raspberry:~ $ git clone https://github.com/kawanishi291/proen2021 初回準備 pi@raspberry:~ $ cd ./proen2021 pi@raspberry:~/proen2021 $ chmod 777 setup.sh pi@raspberry:~/proen2021 $ ./setup.sh 実行 pi@raspberry:~/proen2021 $ ./RS232C_raspberry_pi

インストール php pi@raspberry:~ $ sudo apt install php php php-cli php-curl pi@raspberry:~ $ sudo a2enmod userdir pi@raspberry:~ $ sudo systemctl restart apache2 MariaDB pi@raspberry:~ $ sudo apt-get install mariadb-server-10.0 python pi@raspberry:~ $ pip install mysql-connector-python pi@raspberry:~ $ pip install mysqlclient ソースコード pi@raspberry:~ $ cd /var/www/html pi@raspberry:/var/www/html $ git clone https://github.com/kawanishi291/proen pi@raspberry:/var/www/html $ cd ~ pi@raspberry:~ $ git clone https://github.com/kawanishi291/proen2021 初回準備 pi@raspberry:~ $ cd ./proen2021 pi@raspberry:~/proen2021 $ chmod 777 setup.sh pi@raspberry:~/proen2021 $ ./setup.sh Enter password: raspberry 初期セットアップ実行開始 DBユーザ作成完了 DBテーブル作成完了 テキストファイル作成完了 ファイル権限変更完了 Cコンパイル完了 この画面表示ならおけ。 TOPページの確認 実行と結果 em board側 rs232c.mtpjをCS+で実行 RaspberryPi側 pi@raspberry:~/proen2021 $ ./RS232C_raspberry_pi fb = 3 s 0 0 2 , 0 0 1 , 0 0 0 , 0 3 9 True 現在の日付時刻のデータが増えていれば成功 webサイトの確認

S3にファイルがアップされたらLambdaでメール通知する設定 この手の設定記事も何番煎じか分かりませんが、今回はついにAWS Lambdaに手を出していこうと思います。PythonとかRubyとか全然できないので敬遠してたんですけど、諸事情あり勉強がてらの記事投稿です。 今回のゴール S3の特定バケットに特定の拡張子ファイル（今回はテキストファイル）をアップロードしたら、Lambdaでpythonコードが実行されからAmazon SNS経由でメール通知が届く。 以下イメージです。 設定項目 ①テキストファイルを格納するS3バケットとサブフォルダを作成 ②Lambda関数を作成 ③LambdaにIAMロールを設定 ④Lambda関数のトリガーを設定 ⑤SNSのトピック、サブスクリプションを作成、設定 ⑥Lambda関数にpythonコードを記述 実際にやってみた まず①ですがLambdaを設定するリージョンと同じリージョンを選択して、S3バケットを作成します。基本的にデフォルトの設定のままでOKです。 続いて②のLambda関数を作成ですが、AWSが用意したテンプレートを使うこともできますが今回は「一から作成」で進めていきます。 「関数名」は任意の名前を、「ランタイム」はPython3.7を選択します。 実行ロールは「基本的なLambdaアクセス権限で新しいロールを作成」を選択します。 これでいったんLambda関数の作成は完了です。 次に③のLambdaにIAMロールを設定を行います。 Lambdaは作成された時点の状態では以下の図のようにCloudWatch Logs以外のアクセス権がついていない状態となります。 今回のゴールが「Amazon SNS でメール通知する」のため、LambdaからSNSを呼びさせるようにアクセス権を設定する必要があります。 アクセス権の設定方法については色々なやり方があると思いますが今回は単純に「新しく作成されたロール」にSNSのアクセス権をアタッチするやり方で進めます。 「リソースの概要」の上部に「実行ロール」の表示があり、LambdaにアタッチされたIAMロールが表示されています。このロールにSNSの権限を以下のようにアタッチします。※別にFullAccessでなくてもOKです これでIAMロールの設定はOKです。 次に④Lambda関数のトリガーを設定をしていきます。トリガーとは、Lambdaが実行されるきっかけとなるアクションのことです。今回の場合、「トリガー」はS3バケットへのデータアップロードになります。 トリガーの設定は下図の赤丸の「トリガーを追加」をクリックして、以下のように選択していきます。 トリガーはS3 バケットは①で作成したバケットを指定 イベントタイプはとりあえず今回は「全てのオブジェクト作成イベント」を選択 プレフィックスはサブフォルダを指定　　　※「サブフォルダ/」 という形で最後の”/”も記載する サフィックスは今回はテキストファイルなので、「.txt」を選択 これでトリガーの設定は完了です。S3の指定バケットのテキストファイルがアップロードされたらLamdba関数が実行されるようになります。 続いて⑤Amazon SNSのトピック、サブスクリプションを作成、設定をしていきます。Lambdaが呼び出すSNSを設定し、通知ができるように有効化しておきます。 下図の赤い丸の「トピック」をクリックし、まずトピックを作成します。 「名前」、「表示オプション」は任意のものでOKです。 トピックが作成されるとARNが発行されますので、ARNをメモしておいてください。Pythonコードに後ほど埋め込んで使います。 次にサブスクリプションを設定していきます。下図の赤い丸の「サブスクリプションの作成」をクリックします。 プロトコルは「Eメール」を選択し、エンドポイントに「通知したいメールアドレス」を設定します。 サブスクリプションを作成するとエンドポイントで設定したメールアドレスに確認依頼のようなメールが届きます。そのメールの「Confirm subscription」をクリックすると確認OKになり、以下ような画面が表示されます。 これでSNSの準備が整いました。 最後にいよいよ⑥のLambda関数にコード記載していきます。lambda_function.py となっているファイルをダブルクリックして、既存コードを削除して以下のコードを記述してください。コードを書く場所やlambda_function.py の場所は下図参照。 import boto3 def lambda_handler(event, context): client = boto3.client('sns') TOPIC_ARN = 'Amazon SNSのトピックのARNを記載' msg = 'Pythonからのテストメール\nS3バケットにファイルがアップロードされました' subject = 'S3バケットを確認してください!!!!' response = client.publish( TopicArn = TOPIC_ARN, Message = msg, Subject = subject ) return response エラー表示がでなければ、「Deploy」をクリックして、コードを保存してください。 もし、Lambda関数に適切なアクセス権が設定されていない場合はテスト実行すると以下のようなResponse結果になります。IAMロールにアタッチしているポリシーを見直しましょう。 あとは該当のS3バケットのサブフォルダにテキストファイルをアップロードしてみましょう。 こんな感じでメール通知が出ると思います。 注意点 トリガーを追加する画面の右側に「送信先を追加」というものがあり、ここにAmazon SNSを設定するのかな？...と思って設定すると通知は来るのですが、タイムスタンプやリクエストIDなどの情報が羅列されたものが通知されるだけでした。 トリガーの設定で、プレフィックスにサブフォルダを記載しますが、サブフォルダ名の最後に"/"をつけ忘れると上手く動作しなくなります。 またサブフォルダを作成せず、バケット直下を指定してしてもなぜか上手く動作しませんので、適当にバケット下にサブフォルダを作成してください。 参考にしたサイト https://qiita.com/tsumita7 https://docs.aws.amazon.com/ja_jp/lambda/latest/dg/with-s3-example.html https://blog.smtps.jp/entry/2019/10/04/171230 https://recipe.kc-cloud.jp/archives/10035 https://dev.classmethod.jp/articles/lambda-my-first-step/ https://nopipi.hatenablog.com/entry/2019/01/11/224201 終わりに Lambda が使えるようになると一気にAWSで構築できる内容が広がります。何より楽しくなってきます（笑） pythonをそこそこ書けるようになったら、さぞ楽しいことでしょう...精進しますm(_ _)m

はじめに 前回の記事でスクレイピングでレースのデータを取得しました。 そのデータを使ってGUIのアプリケーションを作ります。 オブジェクト指向を使ったtkinterの具体的な実装法のメモになればいいかなと。 様々なサイトを参考にさせていただきました。 手順 スクレイピングでレースデータを取得 データを使ってGUIアプリケーションを作る（検索＆TODOリスト登録）【今回やる】 使用開発環境 Python 3.8.3 pandas 1.0.5 tkinter 8.6.10 エディタ：VScode 要件 適当に要件を決めます。 レースを名前で検索できる 検索したレースを選択してTODOリストに追加できる TODOリストのチェックボックスをチェックするとそのレースを消すことができる レースを期間で昇順ソートできる レースのデータ（期間、場所など）も表示させる ソースコード 順番に実装していきます。 インポート、定数の宣言 この辺りの定数は他の実装をしつつ定義したものです。 ほとんどがtkinterのGUI設計用の定数です。 import pandas as pd from tkinter import * PATH='race.csv' SEP=20 CHK_X=10 CHK_Y=10 LIST_X=10 LIST_Y=80 SER_ENT_WIDTH=20 SER_ENT_X=10 SER_ENT_Y=10 MAIN_TITLE='ウマ娘 レース計画ソフト' MAIN_WINDOW_SIZE='600x500' SER_BTN_TEXT='検索' SER_BTN_X=10 SER_BTN_Y=50 UPD_BTN_TEXT = '更新' UPD_BTN_X=80 UPD_BTN_Y=50 ADD_BTN_TEXT='レースを追加' ALL_BTN_TEXT='全選択' ALL_BTN_X=130 DONE_RACE = 0 MONTH_ZEN = 0.1 MONTH_KOU = 0.2 Raceクラスの定義 レースに必要なプロパティと、レースデータを表示させるメソッドを定義する。 要件の、「レースのデータ（期間、場所など）も表示させる」はこれで実装完了です。 class Race: def __init__(self, year, month, name, race_class, place, length, direction, tk): self.year = year self.month = month self.name = name self.race_class = race_class self.place = place self.length = length self.direction = direction self.bln = BooleanVar() self.bln.set(False) self.chk = Checkbutton(tk, variable=self.bln, text=self.generate_view_race()) def generate_view_race(self): txt = f'{self.year} {self.month} {self.race_class} {self.name} ' + \ f'{self.length} {self.place} {self.direction}' return txt アプリのクラスを定義 tkinterのウェジットをインスタンスのプロパティとして実装します。 class Race_plan_app(): def __init__(self): # メイン画面を定義 self.tk = Tk() self.tk.title(MAIN_TITLE) self.tk.geometry(MAIN_WINDOW_SIZE) # 検索窓を定義 self.search_entry = Entry(width=SER_ENT_WIDTH) self.search_entry.place(x=SER_ENT_X, y=SER_ENT_Y) # 検索ボタンを定義 self.search_btn = Button(self.tk, text=SER_BTN_TEXT, command=self.search_race) self.search_btn.place(x=SER_BTN_X, y=SER_BTN_Y) # 更新ボタンを定義 self.update_btn = Button(self.tk, text=UPD_BTN_TEXT, command=self.update) self.update_btn.place(x=UPD_BTN_X, y=UPD_BTN_Y) # レースのリストを入れるプロパティを定義 self.race_list = [] self.read_race_data() # レースのデータを読み込む def read_race_data(self): self.df_race = pd.read_csv(PATH) def run(self): self.tk.mainloop() 基本的なウェジットは実装完了です。 あとは、runメソッドを呼び出すことでmainloopさせるように実装しました（やり方違ってたらすみません！）。 race_app = Race_plan_app() race_app.run() ただ、ボタンで起動するメソッドがまだ定義されてないのでこれだけ実行してもエラーを吐かれます。 　検索機能の実装 class Race_plan_app(): # ~~　省略 ~~ # レース検索機能 def search_race(self): # 入力されたレース名をもとにデータフレームから検索する self.ent_txt = self.search_entry.get() self.search_result = self.df_race[self.df_race['レース名'].str.contains(self.ent_txt)] self.search_result = list(self.search_result.itertuples()) # 新しいウィンドウを表示 new_window = Toplevel() new_window.title('レース選択') new_window.geometry('500x1000') # 検索したレースデータをチェックボックス付きで表示（インスタンスも作成） self.search_races = [] for i in range(len(self.search_result)): self.search_races.append(Race(self.search_result[i][2], self.search_result[i][3], self.search_result[i][4], self.search_result[i][5], self.search_result[i][6], self.search_result[i][7], self.search_result[i][8], new_window)) self.search_races[-1].chk.place(x=CHK_X, y=CHK_Y + SEP*i) # TODOリストへのレース追加を行う add_btn = Button(new_window, text=ADD_BTN_TEXT, command=self.add_race) add_btn.place(x=CHK_X, y=CHK_Y + SEP*len(self.search_result)) # 検索結果を全部選択する all_btn = Button(new_window, text=ALL_BTN_TEXT, command=self.chose_all) all_btn.place(x=ALL_BTN_X, y=CHK_Y + SEP*len(self.search_result)) new_window.mainloop() これで「レースを名前で検索できる」の実装完了です。 キモとしては、新しいウィンドウを表示させるにはToplevelメソッドを使うところです。 そのウィンドウの変数を使って新しいウィンドウに表示させるウィジットを定義していきます。 最後にmainloopさせているので、ウィンドウが閉じられたら新しいウィンドウの処理も停止します。 検索だけでなく、レースの結果をTODOリストに追加する機能も実装する必要があります。 各ボタンに設定しているメソッドを実装していきます。 　TODOリストへの追加 class Race_plan_app(): # ~~　省略 ~~ # TODOリストに追加する def add_race(self): for i in range(len(self.search_races)): if self.search_races[i].bln.get(): race = self.search_races[i] self.race_list.append(Race(race.year, race.month, race.name, race.race_class, race.place, race.length, race.direction, self.tk)) self.update() # レースの検索結果を全部選択する def chose_all(self): for i in self.search_races: i.bln.set(True) これで「検索したレースを選択してTODOリストに追加できる」機能の実装完了です。 これらは新しいウィンドウのボタンウィジットに登録しています。 ボタンに登録するメソッドもインスタンスないのメソッドを使用できるのでわかりやすいですね。 検索結果を全選択する機能も便利かなと思ってついでにつけました。 更新機能と、ソート機能 ちょっと長くなりますが、以下のように実装。 更新機能は、ボタン以外にもレースを追加する際に更新を行う。 更新を行うと、レースのソート、完了済みのレースを削除という処理が行われる。 ソートは…汚いです。動けば正義という気持ちで実装しました… class Race_plan_app(): # ~~　省略 ~~ # 更新する def update(self): # TODOリストのレースをチェックボックス付きで表示 s = 0 for i in range(len(self.race_list)): if self.race_list[i].bln.get(): self.race_list[i].chk.destroy() self.race_list[i] = DONE_RACE else: self.race_list[i].chk.place(x=LIST_X, y=LIST_Y + SEP*s) s+=1 # 完了済みのレースは削除 while DONE_RACE in self.race_list: self.race_list.remove(DONE_RACE) # レースをソートする self.sort_race() # レースを昇順にする def sort_race(self): # 年、月毎にデータを分ける、月も前後半で昇順になるようにする race_yms = [] for race in self.race_list: race_yms.append([int(race.year[0]), int(race.month[:-3])]) if race.month[-2:] == '前半': race_yms[-1][1] += MONTH_ZEN elif race.month[-2:] == '後半': race_yms[-1][1] += MONTH_KOU years = [i[0] for i in race_yms] # 年をソートする for i in range(len(self.race_list)): for j in range(len(self.race_list)): if race_yms[i][0] < race_yms[j][0]: tmp = race_yms[i] race_yms[i] = race_yms[j] race_yms[j] = tmp tmp = self.race_list[i] self.race_list[i] = self.race_list[j] self.race_list[j] = tmp # 1年目を月毎にソート if 1 in years: ri = self.count_rindex([i[0] for i in race_yms], 1) # 1年目の最後尾のインデックス番号取得 for i in range(len(self.race_list[:ri+1])): for j in range(len(self.race_list[:ri+1])): if race_yms[i][1] < race_yms[j][1]: tmp = race_yms[i] race_yms[i] = race_yms[j] race_yms[j] = tmp tmp = self.race_list[i] self.race_list[i] = self.race_list[j] self.race_list[j] = tmp # 2年目を月毎にソート if 2 in years: ind = [i[0] for i in race_yms].index(2) ri = self.count_rindex([i[0] for i in race_yms], 2) for i in range(ind, len(self.race_list[:ri+1])): for j in range(ind, len(self.race_list[:ri+1])): if race_yms[i][1] < race_yms[j][1]: tmp = race_yms[i] race_yms[i] = race_yms[j] race_yms[j] = tmp tmp = self.race_list[i] self.race_list[i] = self.race_list[j] self.race_list[j] = tmp # 3年目を月毎にソート if 3 in years: ind = [i[0] for i in race_yms].index(3) for i in range(ind, len(self.race_list)): for j in range(ind, len(self.race_list)): if race_yms[i][1] < race_yms[j][1]: tmp = race_yms[i] race_yms[i] = race_yms[j] race_yms[j] = tmp tmp = self.race_list[i] self.race_list[i] = self.race_list[j] self.race_list[j] = tmp # 年ごとの最終要素のインデックス番号を取得するメソッド def count_rindex(self, lis, n): return len(lis) - (lis[::-1].index(n)+1) これで、「TODOリストのチェックボックスをチェックするとそのレースを消すことができる」、「レースを期間で昇順ソートできる」という機能の実装完了です。 使用例 有馬記念で検索してみるとこんな感じ。2年目と3年目の違いで、2年目を追加してみます。 色々追加して、2年目のレースが終わったと仮定すると、 このようになります。 今後の展望 追加するなら、セーブ機能ですね。 実装予定ではあったんですが、自分用だけなら別にいらんなと思って実装してません。 まとめ 簡潔にはなりますが、ウマ娘のレース管理アプリ作成でした。 自分が色々とPythonで書いていく中で考えた書き方なので癖があるかもしれませんが、参考になれば幸いです。 参考文献 なんとか記憶を掘り出して参考サイトとか思い出しました。漏れがあれば申し訳ないです。 Pythonによるプログラミング 【Python】GUI 画面（ウィンドウ）を作る（tkinter） 【Python】ボタンのクリックイベントを取得する Entryの値を取得してみる · tkinter - Nな人（N na hito） 【Python】チェックボックス（チェックボタン）を作成する（Checkbutton） Tkinter のボタンをクリックして新しいウィンドウを作成する方法 Python Tkinter 入門メモ Git Hub https://github.com/sugimochi97/uma_musume/tree/master/race_plan

概要 "i += 1"と"i = i + 1"の処理速度を比較します。 すでに[小ネタ] Pythonでは、 i+=1 よりも i = i+1の方が微妙に早い、というお話に詳しくまとまっているので、ここでは箱ひげ図で可視化してみる。 方法 "i += 1"と" i = i + 1"をそれぞれ10,000,000回計算してそれを1セットとして、100回分計測して箱ひげ図で比較します。 箱ひげ図は、プログラマーのための統計学】箱ひげ図を参考にしました。 処理速度の比較 import time import matplotlib.pyplot as plt number_of_set = 100 list_1 = []# "i = i + 1" list_2 = []# "i += 1" n = int(1e+7) # "i = i + 1"と"i += 1"の計測 for _ in range(number_of_set): # "i = i + 1" i = 0 start_1 = time.time() for _ in range(n): i = i + 1 list_1.append(time.time() - start_1) # "i += 1" i = 0 start_2 = time.time() for _ in range(n): i += 1 list_2.append(time.time() - start_2) # 箱ひげ図 time = (list_1, list_2)# タプル化 fig, ax = plt.subplots() bp = ax.boxplot(time) ax.set_xticklabels(['i = i + 1', 'i += i']) plt.title('100 times with 10,000,000 loops as one set') plt.xlabel('type') plt.ylabel('time [sec]') plt.grid() # 描画 plt.show() 結論 箱ひげ図から定性的になりますが、"i = i + 1"の方が速い結果となりました。 ただ、"i += 1"の方がすっきりしているので、臨機応変に使い分けたいところです。 参考資料 [小ネタ] Pythonでは、 i+=1 よりも i = i+1の方が微妙に早い、というお話

線形計画法 (Linear Programming, LP)を解くためのPythonライブラリとしてPuLPがあるので、これの利用例として、簡単な線形計画法と、連立方程式とナップサック問題を解いた例を以下記します。 インストール !pip install pulp Collecting pulp [?25l Downloading https://files.pythonhosted.org/packages/14/c4/0eec14a0123209c261de6ff154ef3be5cad3fd557c084f468356662e0585/PuLP-2.4-py3-none-any.whl (40.6MB) [K |████████████████████████████████| 40.6MB 104kB/s [?25hCollecting amply>=0.1.2 Downloading https://files.pythonhosted.org/packages/f3/c5/dfa09dd2595a2ab2ab4e6fa7bebef9565812722e1980d04b0edce5032066/amply-0.1.4-py3-none-any.whl Requirement already satisfied: docutils>=0.3 in /usr/local/lib/python3.7/dist-packages (from amply>=0.1.2->pulp) (0.17.1) Requirement already satisfied: pyparsing in /usr/local/lib/python3.7/dist-packages (from amply>=0.1.2->pulp) (2.4.7) Installing collected packages: amply, pulp Successfully installed amply-0.1.4 pulp-2.4 基本的な使い方 目的の関数を最小 (あるいは最大) にする変数の値を求める「数理計画法 (Mathematical Programing)」のうち、制約条件と目的関数が線形方程式で表される問題を「線形計画法 (Linear Programming, LP)」と呼びます。 PuLPではまず、次のようにして線形計画法のためのインスタンスを作成します。 import pulp prob = pulp.LpProblem(name='A_simple_Linear_Programming', sense=pulp.LpMaximize) この問題で取り扱う変数を次のように定義します。ここでは、最低値がゼロという制約を加えています。 x = pulp.LpVariable('x', lowBound = 0) y = pulp.LpVariable('y', lowBound = 0) 最大化または最小化したい目的関数を設定します。最大なのか最小なのかは、インスタンス作成時に sense=pulp.LpMaximize のようにして定義済みです。 prob += x + y 制約条件を設定します。 prob += x + 2 * y <= 6 prob += 2 * x + y <= 6 これまで設定した制約条件をチェックしましょう。 prob A_simple_Linear_Programming: MAXIMIZE 1*x + 1*y + 0 SUBJECT TO _C1: x + 2 y <= 6 _C2: 2 x + y <= 6 VARIABLES x Continuous y Continuous 次の文を実行すればソルバーが頑張って解を探してくれます。 status = prob.solve() 無事、解が見つかったかどうかは次のようにして分かります。 pulp.LpStatus[status] 'Optimal' 最適解の各変数の値は次のようにして確認できます。 print("Optimal x=", x.value()) print("Optimal y=", y.value()) print("Optimal z=", prob.objective.value()) Optimal x= 2.0 Optimal y= 2.0 Optimal z= 4.0 連立方程式 PuLP は連立方程式にも用いることができます。例としてつぎのような連立方程式を解いてみましょう。 3 x_1 + 2 x_2 + 7 x_3 + x_4 = 8 \\ x_1 + 5 x_2 + x_3 - x_4 = 5 \\ 4 x_1 + x_2 + 3 x_3 - 2 x_4 = 7 \\ x_1 + 6 x_2 + 4 x_3 + 3 x_4 = 13 変数の定義 import pulp x_1 = pulp.LpVariable('x_1') x_2 = pulp.LpVariable('x_2') x_3 = pulp.LpVariable('x_3') x_4 = pulp.LpVariable('x_4') 制約条件の設定 prob = pulp.LpProblem(name='Simultaneous_linear_equations') prob += 3 * x_1 + 2 * x_2 + 7 * x_3 + x_4 == 8 prob += x_1 + 5 * x_2 + x_3 - x_4 == 5 prob += 4 * x_1 + x_2 + 3 * x_3 - 2 * x_4 == 7 prob += x_1 + 6 * x_2 + 4 * x_3 + 3 * x_4 == 13 制約条件の確認 prob Simultaneous_linear_equations: MINIMIZE None SUBJECT TO _C1: 3 x_1 + 2 x_2 + 7 x_3 + x_4 = 8 _C2: x_1 + 5 x_2 + x_3 - x_4 = 5 _C3: 4 x_1 + x_2 + 3 x_3 - 2 x_4 = 7 _C4: x_1 + 6 x_2 + 4 x_3 + 3 x_4 = 13 VARIABLES x_1 free Continuous x_2 free Continuous x_3 free Continuous x_4 free Continuous ソルバー実行 status = prob.solve() pulp.LpStatus[status] 'Optimal' 解の確認。 x_1.value(), x_2.value(), x_3.value(), x_4.value() (3.5, 1.0, -1.0, 2.5) ナップサック問題 アイテムの集合 $I$ = {$1, 2, ..., N$} 中の各アイテム $i \in I$ の価値を $v_i$、重みを $w_i$とし、それを入れるナップサックの最大容量を $W_{capacity}$ としたとき、下記の最適化問題をナップサック問題と言います。 max \sum_{i=1}^{N} v_i x_i \\ s.t. \sum_{i=1}^{N} w_i x_i \le W_{capacity} \\ \qquad \qquad x_i \in \mathbb{N} \qquad (\forall i \in I) $\in$ は英語で「in」、 $\forall$ は「for all」の意味 $s.t.$ は「such that」の略で「下記を満たす条件の元で上記を求める」くらいの意味 $\mathbb{N}$ は自然数全体の集合 $x_i \in \mathbb{N}$ を $x_i \in$ {$0$, $1$} に置き換えたものを特に「0-1ナップサック問題」と言います。 ナップサック問題は「動的計画法」で解ける問題として典型的なものですが、ここでは「線形計画法」としてPuLPで解きます。 データの用意 n = 5 W_capacity = 175 W = [95, 56, 65, 54, 32] V = [79, 9, 57, 55, 27] 制約条件の設定 import pulp prob = pulp.LpProblem('kp01', sense = pulp.LpMaximize) xs = [pulp.LpVariable('{}'.format(x), cat='Binary', lowBound = 0) for x in range(n)] prob += pulp.lpDot(V, xs) prob += pulp.lpDot(W, xs) <= W_capacity prob kp01: MAXIMIZE 79*0 + 9*1 + 57*2 + 55*3 + 27*4 + 0 SUBJECT TO _C1: 95 0 + 56 1 + 65 2 + 54 3 + 32 4 <= 175 VARIABLES 0 <= 0 <= 1 Integer 0 <= 1 <= 1 Integer 0 <= 2 <= 1 Integer 0 <= 3 <= 1 Integer 0 <= 4 <= 1 Integer ソルバーの実行 status = prob.solve() pulp.LpStatus[status] 'Optimal' 答え print([x.value() for x in xs]) print(prob.objective.value()) [0.0, 0.0, 1.0, 1.0, 1.0] 139.0 個数制限なしナップサック問題 $x_i \in \mathbb{N}$ をゼロ以上の自然数に拡張したものを、個数制限なしのナップサック問題と言います。動的計画法では、0-1ナップサック問題と個数制限なしのナップサック問題とでアルゴリズムを変更する必要が出てきますが、PuLPでは、次のように少しだけ変更すればOKです。 データの用意 n = 3 W_capacity = 7 W = [3, 4, 2] V = [4, 5, 3] 制約条件の設定 import pulp prob = pulp.LpProblem('kp_unlimited', sense = pulp.LpMaximize) xs = [pulp.LpVariable('{}'.format(x), cat='Integer', lowBound = 0) for x in range(n)] prob += pulp.lpDot(V, xs) prob += pulp.lpDot(W, xs) <= W_capacity prob kp_unlimited: MAXIMIZE 4*0 + 5*1 + 3*2 + 0 SUBJECT TO _C1: 3 0 + 4 1 + 2 2 <= 7 VARIABLES 0 <= 0 Integer 0 <= 1 Integer 0 <= 2 Integer ソルバーの実行 status = prob.solve() pulp.LpStatus[status] 'Optimal' 答え print([x.value() for x in xs]) print(prob.objective.value()) [1.0, 0.0, 2.0] 10.0

ウマ娘をやってて、ミッションなどで特定のレースに出走する必要があったりする場合、そのレースがいつ開催されるのか、そしてそのミッションをクリアしたかどうかをアプリで管理できたら良いと思いませんか？ まぁ、大手攻略サイトと、TODOリストアプリ使えばいいんですが、Pythonのコーディング練習とか含めてやってみようと思います。 スクレイピングや、tkinterの使い方のメモなどになればいいかなと。 手順 スクレイピングでレースデータを取得【今回やる】 データを使ってGUIアプリケーションを作る（検索＆TODOリスト登録） 使用開発環境 Python 3.8.3 beautifulsoup4 4.9.1 pandas 1.0.5 エディタ：VScode スクレイピングでレースデータを取得 レースデータを一つ一つ入力していくのは時間がかかるのでGame With様から取得させていただきます。 サイトはこちら。 ただ、スクレイピングをする上で自分が気をつけているのは、「インターネット上にある誰かのデータを借りている」「データ取得のリクエストにはサーバーに負担をかけている」ということを意識することです。 なので簡単に実行できないようにURLや内部のクラス名などは明記しません。 from bs4 import BeautifulSoup import requests import pandas as pd URL='' RACE_CLASS='' # リクエスト＆読み込み load_url = URL html = requests.get(load_url) soup = BeautifulSoup(html.content, 'html.parser') lis_td = [] # td要素を入れるためのリスト lis_a = [] # a要素を入れるためのリスト # レース毎に設定されてるクラスから一つずつ取り出す for element in soup.find_all(class_=RACE_CLASS): # 要素内のデータを整形して、項目毎のレースデータを作成 lis = [i.text for i in element.find_all('td')] lis_td.append(lis[0].split(' ') + lis[1].split(' / ')) lis_a.append([i.text for i in element.find_all('a')]) # 空リストを削除する while [] in lis_a: i = lis_a.index([]) lis_a.pop(i) lis_td.pop(i) # リスト内の空要素を削除 lis_a = [i[1] for i in lis_a] # データフレーム形式でデータを整形 df = pd.DataFrame(lis_td, columns=RACE_COLUMNS) se = pd.Series(lis_a) df['レース名'] = se # 出力 df.to_csv(OUTPUT_CSV) ソースコードの細かい処理などはコード内のコメントを見てください。 参考文献 【ウマ娘】レース一覧 - ゲームウィズ(GameWith) PythonでHTMLを解析してデータ収集してみる？ スクレイピングが最初からわかる『Python 2年生』 米国データサイエンティストがやさしく教えるデータサイエンスのためのPython講座（Pandasの説明がわかりやすいコースです） Git Hub https://github.com/sugimochi97/uma_musume/tree/master/race_plan

NumPyの数列から、隣り合う２項の平均値の数列を計算する最も速い方法 数列 $[a_1, a_2, a_3, ..., a_i]$ から、数列 $[\frac{a_1+a_2}{2}, \frac{a_2+a_3}{2}, \frac{a_3+a_4}{2}, ..., \frac{a_{i-1}+a_i}{2}]$ を得たいとする。 In [1]: import numpy as np In [2]: a = np.random.default_rng(0).integers(0, 10, 10) ...: a Out[2]: array([8, 6, 5, 2, 3, 0, 0, 0, 1, 8], dtype=int64) In [3]: func_hoge(a) Out[3]: array([7. , 5.5, 3.5, 2.5, 1.5, 0. , 0. , 0.5, 4.5]) どのようにすればよいだろうか。 動機 Qiitaの『Pythonでデータの挙動を見やすくする可視化ツールを作成してみた（ヒストグラム・確率分布編）』という記事を見て、以下のコードが気になった。 # フィッティングの残差平方和を計算 (参考https://rmizutaa.hatenablog.com/entry/2020/02/24/191312) hist_y, hist_x = np.histogram(x, bins=20, density=True) # ヒストグラム化して標準化 hist_x = (hist_x + np.roll(hist_x, -1))[:-1] / 2.0 # ヒストグラムのxの値をビンの左端→中央に移動 この部分は、コメント行にある通り、『あてはまりのよい確率分布を探したい - rmizutaの日記』に掲載されている以下のコードの引用のようである。 y, x = np.histogram(data.iloc[:,1], bins=20, density=True) #xの値はヒストグラムの左端の値なので中心点に修正 x = (x + np.roll(x, -1))[:-1] / 2.0 一般的NumPyユーザーとしての直感では、np.roll()というのは非常に遅い、つまり（大した処理じゃないのに）やたらと時間のかかる鈍い関数である。しかも、特に、ここで行っている処理は、配列における隣り合った２項の平均をとるというものであるから、np.roll()を使う必然性を感じないので、このコードにはひどく違和感を覚えた。 方法 この処理は、以下のようにスライシングして行うのが、スタンダードだと思う。 # xの値はヒストグラムの左端の値なので中心点に修正 x = (x[1:] + x[:-1]) / 2 NumPyのヘビーユーザーならnp.correlate()/np.convolve()を用いる方法も考えられると思う。あまり知られていないが、一般的NumPyユーザーとしての直感では、この関数はとても速い有能な関数である。 x = np.correlate(x, [.5, .5], 'valid') 今回、先に引用したコードの場合、xはビンのエッジ配列だから、等差数列である。冒頭で述べた「数列 $[a_1, a_2, a_3, ..., a_i]$ から、数列 $[\frac{a_1+a_2}{2}, \frac{a_2+a_3}{2}, \frac{a_3+a_4}{2}, ..., \frac{a_{n-1}+a_n}{2}]$ を得たい」という目的からは外れてしまうが、今回のような条件のときにのみ適用されるよりよい方法を考えると、公差の半分を配列に足せばいいのだと気づく。 x = x[:-1] + (x[1] - x[0]) / 2 似た要領で、np.linspace()を使って１から作るというやり方でもいける。 x = np.linspace(x[:2].mean(), x[-2:].mean(), x.size-1) いろいろな方法をあげたが、結局どれが速いのか。どうやらもとの配列xの長さによって変わるようである。 計測してみた 比較するコードは以下の通り。 import numpy as np def np_roll(x): return (x + np.roll(x, -1))[:-1] / 2 def slice_mean(x): return (x[1:] + x[:-1]) / 2 def np_correlate(x): return np.correlate(x, [.5, .5], 'valid') def diff_add(x): return x[:-1] + (x[1] - x[0]) / 2 def np_linspace(x): return np.linspace(x[:2].mean(), x[-2:].mean(), x.size-1) benchitを用いて計測する。 import benchit funcs = [np_roll, slice_mean, np_correlate, diff_add, np_linspace] inputs = [np.linspace(1, n/9, n) for n in 10 ** np.arange(1, 7)] # <- xの中身は適当 # すべての関数が同じ結果を返すことを確認 all(np.allclose(funcs[i](inputs[-1]), funcs[i+1](inputs[-1])) for i in range(len(funcs)-1)) # -> True t = benchit.timings(funcs, inputs) t.plot(figsize=(8, 8), logx=True) print(t) t Functions np_roll slice_mean np_correlate diff_add np_linspace Len 10 0.000021 0.000002 0.000004 0.000002 0.000058 100 0.000021 0.000002 0.000004 0.000002 0.000058 1000 0.000024 0.000004 0.000005 0.000003 0.000061 10000 0.000041 0.000014 0.000012 0.000007 0.000075 100000 0.000192 0.000121 0.000077 0.000050 0.000221 1000000 0.013711 0.008622 0.003737 0.003750 0.005741 配列xの長さにかかわらず、公差を求めて配列に足す方法が最も速いことがわかる。ただし、これはxが等差数列のときにのみ使える手である。 それを除外すると、xの長さが3000未満程度ならば、スタンダードと呼んだスライシングして計算する方法がやはり速い。xの長さが3000以上の場合はnp.correlate()を用いた方法が他を圧倒しており、なんと、xの長さが100万になると「xが等差数列のときにのみ使える裏技」をも凌駕している。対して、np.roll()を用いた方法はきわめて遅い。 おおむね予想通りの結果となったが、np.correlate()の有能さにはもっと注目を浴びせたい。そしてやはり、こんなときにnp.roll()は使うべきではない。

前提 某プログラミングスクールでRubyを学習しました。 Rubyの復習、Pythonの学習を兼ねてじゃんけんゲームを作って見ました。 Ruby.ver # ＜じゃんけんゲーム＞ # 間違った番号を入れた場合のループ def validate(player_hand, hands) if player_hand < 0 || player_hand > 2 loop{ puts "0~2の番号を選択してください" puts "0.#{hands[0]} 1.#{hands[1]} 2.#{hands[2]}" player_hand = gets.chomp.to_i if player_hand >= 0 && player_hand <= 2 break end } end end # 入力フォーム hands = ["グー","チョキ","パー"] puts "__________________________" puts "名前を入力してください" player_name = gets.chomp puts "0.#{hands[0]} 1.#{hands[1]} 2.#{hands[2]}" puts "0~2の番号を選択してください" player_hand = gets.chomp.to_i validate(player_hand, hands) puts "#{player_name}さんの選択された手は#{hands[player_hand]}です。" # コンピュータとの勝敗判定 program_hand = rand(3) if player_hand == program_hand puts "コンピュータの手は#{hands[program_hand]}あいこです" elsif (player_hand == 0 && program_hand == 1) || (player_hand == 1 && program_hand == 2) || (player_hand == 2 && program_hand == 0) puts "コンピュータの手は#{hands[program_hand]}で#{player_name}さんの勝ちです" else puts "コンピュータの手は#{hands[program_hand]}で#{player_name}さんの負けです" end Python.ver # ＜じゃんけんゲーム＞ import random # 間違った番号を入れた場合のループ def validate(player_hand, hands): if player_hand < 0 and player_hand > 2: while player_hand != 0 or player_hand != 1 or player_hand != 2: print ('0~2の番号を選択してください') player_hand = int(input(f'0.{hands[0]} 1.{hands[1]} 2.{hands[2]}')) if player_hand >= 0 and player_hand <= 2: break # 入力フォーム hands = ['グー','チョキ','パー'] print ('__________________________') print ('名前を入力してください') player_name = input('') print (f"0.{hands[0]} 1.{hands[1]} 2.{hands[2]}") print ('0~2の番号を選択してください') player_hand = int(input('')) validate(player_hand, hands) print (f'{player_name}さんの選択された手は{hands[player_hand]}です。') # コンピュータとの勝敗判定 program_hand = random.randint(0,2) if player_hand == program_hand: print (f'コンピュータの手は{hands[program_hand]}あいこです') elif (player_hand == 0 and program_hand == 1) or (player_hand == 1 and program_hand == 2) or (player_hand == 2 and program_hand == 0): print (f'コンピュータの手は{hands[program_hand]}で{player_name}さんの勝ちです') else : print (f'コンピュータの手は{hands[program_hand]}で{player_name}さんの負けです') じゃんけんゲームで学んだRubyとPythonの違い Rubyと違いPythonはインデントに注意する必要がある。（インデントを間違えると正しい挙動を示さない） メソッド（関数）やif文などpythonではendをつけず、：（コロン）をつける。 式展開を行う際、Rubyは#{}で記述を行うがPythonは(f'{}')で記述（一例） まとめ 同じオブジェクト指向の言語であったが、色々と記述方法が違うのでこれから引き続き学習を行って行きたいと思います。 また、Pythonの式展開は一例なので今後、まとめて行きたいと思います。

オプションの理論価格を算出するブラックショールズ方程式について調べてみましたので、メモとして残しておきたいと思います。 コールオプションは、あらかじめ定められた金融商品(原資産)をあらかじめ定められた期間（満期)にあらかじめ定められた価格（行使価格)で買う権利をあたえる金融商品です。プットオプションは同じ仕組みを用いて、売る権利をあたえます。原資産を株式や為替レートとしたものが活発に取引されています。売買する権利が満期のみに与えられているものをヨーロピアンオプション、満期前であればいつでも権利を行使できるものをアメリカンオプションといいます。 原資産である株式の価格が幾何ブラウン運動にしたがうとします。それは $\mathrm{d}S_t=\mu S_t \mathrm{d}t+\sigma S_t\mathrm{d}W_t$　(1) の確率微分方程式であらわされます。$S_t$は$t$時点の株価です。$\sigma$は価格変動の程度を$\mu$は株価のトレンドをあらわす定数です。$W_t$は標準ウィナー過程です。$\mu$も$\sigma$も株価がもつ特有の特性で、過去の価格データから推定されます。株価の動きはよくランダムウォークにたとえられます。もっとも単純な直線上を動くランダムウォークはゼロから始まり同じ確率で+1または-1というようにとびとびな値をとりながら動く離散的な確率過程です。幾何ブラウン運動は期待値$\mu t$、分散$\sigma^2t$で連続的に動く確率過程です。上述の幾何ブラウン運動の$\mu$に着目しましょう。これは確定的トレンドと解釈できます。ただしこれはリターンとして考えた場合です。株価でみると$\mu$の影響は各$t$時点での株価に左右されます。したがって、ドリフトと呼ばれるこの動きは確率過程の一部です。まずそれを確かめてみましょう。 import matplotlib.pyplot as plt from scipy.stats import norm from scipy.stats import lognorm from scipy import stats import numpy as np def genS(mu0,sigma0,nDays,nSim): mu=mu0/250 sigma=sigma0/np.sqrt(250) S = [[0 for i in range(nDays)] for j in range(nSim)] M = [[0 for i in range(nDays)] for j in range(nSim)] for j in range(nSim): S[j][0]=1 M[j][0]=1 for i in range(1,nDays): S[j][i]=S[j][i-1]*(1+mu+sigma*norm.rvs(size=1)[0]) M[j][i]=S[j][i-1]*mu+M[j][i-1] return np.array(S),np.array(M) S,M=genS(0.1,0.4,250,10000) リターンの年率(mu)を１０％とし、ボラティリティーと呼ばれる年率換算の標準偏差は40%とします。これは典型的な個別株の動きです。上述のシミュレーションはこの株式の一日の終値を250回生成しそれを１年の株価の動きとします。それを10000回繰り返しています。さらに、株価の動きとはべつにドリフト項の動きも保存しておきます。 つぎの図は生成された株価の動きをプロットしたものです。 for i in S: plt.plot(i) ランダムウォークというと期待値の周りを行ったり来たりするというイメージがあるかもしれませんが、動きは上下に大きく振れるときもあります。つぎにドリフトの部分を見てみましょう。 for i in M: plt.plot(i) こちらは線形の確定的なトレンドをもっているように見えますが、実はランダムな動きをしています。その証拠に10000回の試行のどれもが独自の動きをしています。これは株価の変動がmuの結果に影響を与えているからです。一般的に見られる $S_t=S_{t-1}+\mu\mathrm{d}t+\sigma W_t\mathrm{d}t$ で表現されるドリフト項付きの株価モデルの動きとは異なります。こちらは確定的トレンドの項をもちます。 株価の動きとドリフトの動きを注意深くみていきましょう。株価の動きは時間の経過とともに振れ幅が大きくなっています。これは初期値と$t$時点での株価の差の分散が時間の経過とともに大きくなっていることを意味します。どうように初期値と$t$時点での株価の差の期待値も時間の経過とともに大きくなります。これはリターンのモデルとは全く違う性質です。また、株価の動きにはドリフト項が作るトレンド以上に強いトレンドが出ているものがあります。これはランダム項の影響です。生成された乱数の多くが正(負)の値であったために作られたトレンドで、確率的トレンドといいます。この影響はボラティリティが大きくなるとさらに大きくなります。 このような株式を原資産にもつヨーロピアンのコールオプションの満期でのペイオフを$V(S,T)$とします。このペイオフの満期前の価値を得るためには、株価と時間の変化の関数として$V(S,T)$をとらえる必要があります。伊藤のレンマを用いると$V$の価値の時間変化を表現できます。それは $\mathrm{d}V=\left(\mu S\frac{\partial V}{\partial S}+\frac{\partial V}{\partial t}+\frac{1}{2}\sigma^2S^2 \frac{\partial^2 V}{\partial S^2}\right)\mathrm{d}t+\sigma S\frac{}{}\mathrm{d}W$ --- (2) の確率微分方程式になります。この式は確率的項もドリフト項も含んでいますので、$\mathrm{d}V$の動きはランダムです。 　株式市場では、同じ株式銘柄を取引したい売り手と買い手がそろって初めて取引が成立します。しかし、常に取引相手がいるとは限りません。そこで、仲介業者が、リスクに見合った報酬を得ることで仲立ちをしています。オプションの取引では通常の株式の取引以上の困難があります。それは同じ株式であってもさまざまなオプションがあるということです。取引したいオプションのバリエーションが多いということは、取引の出会いの機会は減ります。それは仲介業者にとっては大きなリスクです。そこで仲介業者は仲介をするのではなく、取引相手の注文を自身で受け、そのポジションのリスクをなくす取引戦略を考えます。これをヘッジ戦略といいます。そしてこのヘッジ戦略にかかる費用をオプションを供給することの対価として徴収しようと考えます。この戦略はつぎのように構築します。コールオプションを買いたい投資家が現れたら、そのオプションを取引相手に提供します。仲介業者(マーケットメイカ:MM)にとってはコールオプションを売ったことになります。しかし、このままではコールオプションのリスクを抱えることになります。満期に株価が行使価格を大きく上昇していたら、取引相手は買う権利を行使してきます。MMは行使価格でかれらに株式を売る必要がでてきます。しかし、MMはその株式をもっていないので、その時点で市場から時価で購入する必要があり、市場で高く買い、行使相手に安く売ることになるので大きな損失につながります。このようなリスクをなくすためにMMは$\partial V/ \partial S$の株価を買っておきます。このヘッジに必要なポジション(在庫)の価値は $\Pi=-V+\frac{\partial V}{\partial S}S$ になります。$-V$はオプションの売りポジションを示しています。しかし、時間が経過したり、株価が動いたりすると、$\partial V/ \partial S$も動いてしまいます。その変化は $\Delta \Pi=-\Delta V+\frac{\partial \Delta V}{\partial S}\Delta S$　---- (3) になります。(1)と(2)の式の微小変化を離散の変化に書き換えて、それを上述の式に代入すると $\Pi= \left(-\frac{\partial V}{\partial t}-\frac{1}{2}\sigma^2S^2 \frac{\partial^2 V}{\partial S^2}\right)\Delta t$ ---(4) となります。ここで不確定要素である$W$と$\mu$がなくなりました。リスクのないポジションになったので、そのリターンは無リスク金利になります。$\Delta t$の間のリターンは $\Pi r\Delta t=\Delta \Pi$ になります。(3),(4)式から $\left(-\frac{\partial V}{\partial t}-\frac{1}{2}\sigma^2S^2 \frac{\partial^2 V}{\partial S^2}\right)\Delta t=\left(-\Delta V+\frac{\partial \Delta V}{\partial S}\Delta S\right)r \Delta t$ を得ます。これを整理するとブラック-ショールズの偏微分方程式が得られます。 $\frac{\partial V}{\partial t}+\frac{1}{2}\sigma^2S^2 \frac{\partial^2 V}{\partial S^2}+r\frac{\partial \Delta V}{\partial S}\Delta S -r\Delta V =0$　　(5) これを満期のペイオフなどの幾つかの境界条件のもとで解くとブラック-ショールのオプション価格の評価式が得られます。(5)式は拡散方程式なので定形の解き方があります。それによってえられる解は $C(S_t,t)=S_tN(d_1)-Ke^{-r(T-t)}N(d_2)$ ここで $N(d)=\frac{1}{2\pi}\int_{-\infty}^{d}e^{-\frac{z^2}{2}}$ $d_1=\frac{\log\frac{S_t}{K}+(r+\sigma^2/2)(T-t)}{\sigma\sqrt{T-t}}$ $d_2=\frac{\log\frac{S_t}{K}+(r-\sigma^2/2)(T-t)}{\sigma\sqrt{T-t}}$ となります。結果をPythonで確かめてみましょう。まずヨーロピアンコールオプションの関数をつくります。 def ecall(s,k,t,r,v): d1=(np.log(s/k)+(r+v**2/2)*t)/v/np.sqrt(t) d2=(np.log(s/k)+(r-v**2/2)*t)/v/np.sqrt(t) return s*norm.cdf(d1)-k*np.exp(-r*t)*norm.cdf(d2) 満期のペイオフができるかどうかを満期までの期間tを極力ゼロに近づけ、価格Sを0から2まで動かして確かめます。その他の変数は行使価格k=1,無リスク金利r=0,ボラティリティv=0.4と固定します。 c=[];s=[] for i in range(1,200): s.append(i/100) c.append(ecall(s[i-1],1,0.00001,0,0.4)) plt.plot(s,c, label="payoff of maturity") plt.xlabel('price') plt.ylabel('option price') plt.legend() plt.show() 満期において行使価格を境にペイオフが４５度に折れ曲がるのが確認できます。これは満期時のペイオフそのものです。満期の時点での株価を$s_T$であらわすとmax($s_T-k,0$)と表現されることがあります。株価が行使価格よりも安ければ、オプションの買い手は行使価格で買う権利を放棄して市場で直接株式を購入することを示しています。この満期時のペイオフを本質的価値と呼びます。 つぎに満期から１年前のオプション価格をチェックしてみましょう。t=1とします。 c=[] for i in range(1,200): c.append(ecall(i/100,1,1,0,0.4)) plt.plot(c, label="payoff of 1 year to maturity") plt.xlabel('price') plt.ylabel('option price') plt.legend() plt.show() 一年前に株価がゼロのときはオプションの価値もゼロ、株価が2のときにはオプションの価値は1になることが確認できます。これは株価が行使価格が極端に離れているときのオプションの境界条件です。これはMMがヘッジをしたときに株価がゼロで、行使価格が1のときには行使される可能性がゼロであるので、ヘッジをする必要がないことを意味しています。つぎに、行使価格が1で株価が2であるときには満期に行使されるのが確実なために、100%の株式を保有していればヘッジになることを意味しています。どちらもMMにとってはヘッジの費用はゼロなので、オプションの価格は株価と行使価格の差の１になります。 つぎに$d_1$、$d_2$に着目したいのですが、その前に(2)をもちいて満期時点で株価と行使価格(初期値)の差がどのような分布になるかをを求めてみます。それは$N(z)$となります。ここで $z=\frac{\log\frac{S_t}{K}-(\mu-\sigma^2/2)(T-t)}{\sigma\sqrt{T-t}}$ です。これはつぎのように解釈できます。幾何ブラウン運動にしたがう株価と行使価格の差の期待値は、この２つの価格の比の対数が正規分布にしたがうことから計算されます。その期待値は$(\mu-\sigma^2/2)(T-t)$で、標準偏差は$\sigma\sqrt{T-t}$です。この価格差の分布を対数を取らずに描くと右にひずんだ分布になります。これを確かめてみましょう。これは簡単に確かめられます。それは幾何ブラウン運動のシミュレーションの結果$S$を利用します。まず$S$の頻度図を描きそれを対数正規分布にあてはめます。当てはまりが良ければ、この差の対数が正規分布にしたがっていることになります。 fig, ax = plt.subplots(1, 1) ax.hist(S[:,249],density=True) s,l,sc = stats.lognorm.fit(S[:,249]) print('shape,loc,scale',params) x = np.linspace(lognorm.ppf(0.01, s), lognorm.ppf(0.99, s), 100) ax.plot(x, lognorm.pdf(x, s,l,sc), 'r-', lw=5, alpha=0.6, label='lognorm pdf') plt.show() # shape,loc,scale (0.41460495758207605, 0.021039875375615552, 0.999017608428401) shape,loc,scaleはscipyの分布関数に使われるパラメータです。これについての説明はscipyのマニュアルをご覧ください。グラフのあてはまりを目で確認します。よく当てはまっているといえるのではないでしょうか？ つぎにSの具体的な統計量を求めてみましょう。 np.mean(S[:,249]-1),np.mean(np.log(S[:,249]/1)),np.std(np.diff(np.log(S[:,249]/1))) # (0.09633981256696895, 0.011974958624954268, 0.5645790288879695) ボラティリティが少し大きいですが、株価のリターンは9.6%、対数リターンは1.2%です。つぎにscipyではデータ$S$を対数分布であてはめて、その理論的統計量を計算してくれます。それをやってみましょう。 lognorm.stats(s, moments='mvsk') #(array(1.07936312), array(0.19225792), array(1.28573556), array(3.0766001)) "moments='mvsk'"は平均、分散、歪度、尖度を計算することを指定しています。平均は1.079となっていますが、これは行使価格を差し引かずに計算した結果なので、実際は7.9%です。分散は0.19なのでかなり良い結果です。ここまでの結果で$z$の確率が正規分布にしたがいそうなことは理解できました。その結果、満期時点の株価と行使価格の差の期待値は $S_tN(z)-Ke^{-r(T-t)}N(z)$ になります。これをオプションの理論価格の考察に使うためにはリスク中立な世界で考えなければならないので、$\mu$を$r$に変える必要があります。それを$z^*$とします。そうするとE(S-K)は $S_tN(z^*)-Ke^{-r(T-t)}N(z^*)$ となります。この式とブラック―ショールズの理論価格式の違いはコールオプションがE(max{S-K,0})を求めていることからきます。つまり、違いはE(max{S-K,0})のペイオフを複製するための複製費用(ヘッジ費用)であると考えられます。 オプション複製戦略 実際にオプションの複製に要する費用とペイオフの期待値がオプションプレミアムになるということをブラック―ショールズの方程式を解く過程で理解することは大変な困難をようすることから、複製戦略を構築して、実際の複製戦略の収益がオプションプレミアムと同等になることを見たほうは近道なので、実際の複製戦略を構築してみましょう。 def option_replication(S,k,t,r,v): payoff=[] close=[] pl=[] prem=[] interest=[] for s in S: y=pd.DataFrame(s).copy() y.columns=['Close'] y['pl']=0 #一日の損益を入れます。 y['pos']=0 y['pos_x_c']=0 y['int_income']=0 #init---------------------------------- n=0 buy=0 #買った時の価格です。フラグの役割もしています。 buyF=0 #買いシグナルのフラグ。 sellF=0 #売りシグナルのフラグ size=0 #トレードできる資産の最大の単位数を示します。 delta=norm.cdf((np.log(1/k)+(r+v**2/2)*t)/v/np.sqrt(t)) y.iloc[0,2]=delta y.iloc[0,3]=delta*1 delta0=delta for i in range(1,len(y)): tt=t-i/250 c=y.Close.iloc[i] #その日の価格の取得 c0=y.Close.iloc[i-1] #前日の価格の取得 delta=norm.cdf((np.log(c/k)+(r+v**2/2)*tt)/v/np.sqrt(tt)) y.iloc[i,2]=delta# デルタ y.iloc[i,3]=y.iloc[i-1,3]+(delta-delta0)*c# 在庫の管理 y.iloc[i,1]=delta*c-y.iloc[i,3]# 損益の計算 y.iloc[i,4]=(1-delta)*r/250#金利収入の計算 delta0=delta payoff.append(max(c-k,0)) close.append(c) pl.append(y.iloc[-1:,1]) prem.append(max(c-k,0)-y.iloc[-1:,1]) interest.append(sum(y.iloc[:,4])) return np.array(payoff),np.array(close),np.array(pl),np.array(prem),np.array(interest) 実際に実行してみます。 payoff,close,pl,prem,interest=option_replication(S,1,1,0.1,0.4) payoffはそれぞれの１年のシミュレーション結果の満期時点のペイオフ、closeは終値、plは損益、premはオプションプレミアム、interestは金利収入を保存しています。これらをcloseをｘ軸として散布図を描いて、特徴を捕えます。 plt.scatter(close,payoff,label='payoff') plt.scatter(close,pl,label='trading pl') plt.scatter(close,prem,label='premium') plt.scatter(close,interest,label='interest income') plt.legend() plt.show() シミュレーションは自己資金をもってコールのオプションの複製を試みる戦略なので、株式を購入していない資金に対しては無リスク金利が付きます。この金利の影響は大きく、株価が上昇していて株式を多くもたなくてはならないときには金利収入が減り、ヘッジ戦略の収益は減ってしまっています。損益、金利収入の平均をみていると np.mean(prem)+np.mean(interest),ecall(1,1,1,0.1,0.4) # (0.1936994107690406, 0.203184693100587) ブラックショールズモデルから計算したコールの理論価格は20.3%ですが、複製戦略の収益は19.3%と1%程ずれていますが、オプションのプレミアムがヘッジの期待費用とペイオフの期待収益から構成されているのが分かると思います。 参考 Black–Scholes equation Black–Scholes model Itô's lemma

二分探索法は、検索したい値が配列の中にあるか判定するだけでなく、「所与の条件を満たす最小値を求めよ」という最適化問題にも応用でき、さらには「N個すべての値をx以下にできるか判定せよ」という判定問題にも応用が効く、使いこなせるとカッコいいアルゴリズムの一つです。 配列から目的の値を探索する二分探索法 BinarySearch1.py # 二分探索法 # 計算量：O(logN) def binary_search(key, a): left = 0 right = len(a) while right >= left: mid = left + (right-left)//2 if a[mid] == key: return mid elif a[mid] > key: right = mid - 1 elif a[mid] < key: left = mid + 1 return -1 N = int(input()) a = list(map(int, input().split())) binary_search(N, a) # 入力 # 10 # 3 5 8 10 14 17 21 39 # 出力 # 3 条件を満たす最小値を求める二分探索法 BinarySearch2.py # 最小値を返す二分探索法 # 配列aの中でa[i]≧keyという条件を満たす最小の添字iを返す def binary_search2(key, a): left = 0 right = len(a) while right >= left: mid = left + (right-left)//2 if a[mid] == key: return mid elif a[mid] > key: right = mid - 1 elif a[mid] < key: left = mid + 1 return right+1 # ペア和問題 # N個の整数a0,...,aN-1と、　N個の整数b0,...,bN-1からから１個ずつ整数を選び、 # その和が整数K以上となる範囲での最小値を求める # 入力受取 N, K = map(int, input().split()) a = list(map(int, input().split())) b = list(map(int, input().split())) # bに無限大を表す値を追加 b.append(99999) # bを昇順にソート b.sort() # 暫定最小値を格納 min_ = 99999 # aを固定して解く for i in range(N): iter_ = binary_search2(K-a[i], b) if min_ > a[i] + b[iter_]: min_ = a[i] + b[iter_] print(min_) 最適化問題を判定問題に帰着して解く二分探索法 BinarySearch3.py # AtCoder Beginner Contest 023 D - 射撃王 # N個の風船が初期値h[i]の高さにあり、秒速s[i]で上昇するとし、1秒に1つの速さで風船を # 割っていき、風船が取り得る最大の高さの最小値を求める # 入力受取 N = int(input()) h = list(map(int, input().split())) s = list(map(int, input().split())) # 二分探索法で解く left, right = 0, 99999 while right - left > 1: mid = (left + right) // 2 # mid以内で風船を割り切れる（風船が取り得る最大の高さの最小値）かの判定 ok = True # 各風船を割るまでの制限時間t[i]を用意 t = [0 for i in range(N)] # 初期値の高さと比較し、各風船の制限時間を格納 for i in range(N): if mid < h[i]: ok = False else: t[i] = (mid - h[i]) // s[i] # 制限時間に余裕がない順にソートし、順に判定 t.sort() for i in range(N): if t[i] < i: ok = False # 判定結果に基づき、midを更新 if ok: right = mid else: left = mid print(right)

二分探索法は、検索したい値が配列の中にあるか判定するだけでなく、「所与の条件を満たす最小値を求めよ」という最適化問題にも応用でき、さらには「N個すべての値をx以下にできるか判定せよ」という判定問題にも応用が効く、使いこなせるとカッコいいアルゴリズムの一つです。 配列から目的の値を探索する二分探索法 BinarySearch1.py # 二分探索法 # 計算量：O(logN) def binary_search(key, a): left = 0 right = len(a) while right >= left: mid = left + (right-left)//2 if a[mid] == key: return mid elif a[mid] > key: right = mid - 1 elif a[mid] < key: left = mid + 1 return -1 N = int(input()) a = list(map(int, input().split())) binary_search(N, a) # 入力 # 10 # 3 5 8 10 14 17 21 39 # 出力 # 3 条件を満たす最小値を求める二分探索法 BinarySearch2.py # 最小値を返す二分探索法 # 配列aの中でa[i]≧keyという条件を満たす最小の添字iを返す def binary_search2(key, a): left = 0 right = len(a) while right >= left: mid = left + (right-left)//2 if a[mid] == key: return mid elif a[mid] > key: right = mid - 1 elif a[mid] < key: left = mid + 1 return right+1 # ペア和問題 # N個の整数a0,...,aN-1と、　N個の整数b0,...,bN-1からから１個ずつ整数を選び、 # その和が整数K以上となる範囲での最小値を求める # 入力受取 N, K = map(int, input().split()) a = list(map(int, input().split())) b = list(map(int, input().split())) # bに無限大を表す値を追加 b.append(99999) # bを昇順にソート b.sort() # 暫定最小値を格納 min_ = 99999 # aを固定して解く for i in range(N): iter_ = binary_search2(K-a[i], b) if min_ > a[i] + b[iter_]: min_ = a[i] + b[iter_] print(min_) 最適化問題を判定問題に帰着して解く二分探索法 BinarySearch3.py # AtCoder Beginner Contest 023 D - 射撃王 # N個の風船が初期値h[i]の高さにあり、秒速s[i]で上昇するとし、1秒に1つの速さで風船を # 割っていき、風船が取り得る最大の高さの最小値を求める # 入力受取 N = int(input()) h = list(map(int, input().split())) s = list(map(int, input().split())) # 二分探索法で解く left, right = 0, 99999 while right - left > 1: mid = (left + right) // 2 # mid以内で風船を割り切れる（風船が取り得る最大の高さの最小値）かの判定 ok = True # 各風船を割るまでの制限時間t[i]を用意 t = [0 for i in range(N)] # 初期値の高さと比較し、各風船の制限時間を格納 for i in range(N): if mid < h[i]: ok = False else: t[i] = (mid - h[i]) // s[i] # 制限時間に余裕がない順にソートし、順に判定 t.sort() for i in range(N): if t[i] < i: ok = False # 判定結果に基づき、midを更新 if ok: right = mid else: left = mid print(right)

出発地 Python実践データ分析100本ノック の「ノック61：輸送最適化問題を解いてみよう」で用いられているライブラリ pulp ortoolpy のインストールで迷ったので記録しておきます。 環境によって対処方法が異なると思いますが、私の場合はこうでした。 最初の状態 OS: windows 10.0 python: 3.7.3 numpy: 1.16.2 実施日 2021/06/07 利用ツール Anaconda Prompt 実施コマンド conda update conda conda update numpy pip install pulp pip install ortoolpy 到着地 python: 3.7.3 numpy: 1.20.2 pulp: 2.4 ortoolpy: 0.2.38 conda list # packages in environment at C:\ProgramData\Anaconda3: # # Name Version Build Channel （中略） numpy 1.20.2 py37ha4e8547_0 （中略） ortoolpy 0.2.38 pypi_0 pypi （中略） pulp 2.4 pypi_0 pypi （後略）

出発地 Python実践データ分析100本ノック の「ノック61：輸送最適化問題を解いてみよう」で用いられているライブラリ pulp ortoolpy のインストールで迷ったので記録しておきます。 環境によって対処方法が異なると思いますが、私の場合はこうでした。 最初の状態 OS: windows 10.0 python: 3.7.3 numpy: 1.16.2 実施日 2021/06/07 利用ツール Anaconda Prompt 実施コマンド conda update conda conda update numpy pip install pulp pip install ortoolpy 到着地 python: 3.7.3 numpy: 1.20.2 pulp: 2.4 ortoolpy: 0.2.38 conda list # packages in environment at C:\ProgramData\Anaconda3: # # Name Version Build Channel （中略） numpy 1.20.2 py37ha4e8547_0 （中略） ortoolpy 0.2.38 pypi_0 pypi （中略） pulp 2.4 pypi_0 pypi （後略）

はじめに 本記事はSeleniumで私が個人的に使うためのスニペットのまとめです。 自分が使うためのスニペット集なので内容は偏っています。また一部Seleniumとは直接関係ないものも含まれています。 Seleniumの一通りの使い方を知りたい方は、以下のような素敵な記事がありますので、そちらをご参照ください。 Python + Selenium で Chrome の自動操作を一通り Selenium webdriverよく使う操作メソッドまとめ 環境 主に以下の環境向けの情報です。他の環境では一部動作しないものがあると思います。 ・macOS Catalina ・Python 3.8.3 ・selenium 3.141.0 ・Google Chrome 91.0.4472.19 目次（スニペット+α） タップで展開 とりあえず動かす 終了時に閉じる WebDriverを自動更新する 要素の取得・操作 普通のスクリーンショット取得 ページ全体のスクリーンショット(ヘッドレスモード) Headlessモードで起動 JavaScript実行 要素をクリックする直前・直後に何か処理をする Basic認証を突破する Chrome DevTools Protocolコマンドの実行 【Chrome】ページ全体のスクリーンショットを撮る 【Headless Chrome】WebページをPDFに保存する 【Chrome】ページをMHTML形式で1つのファイルに保存する 【Chrome】カスタムヘッダの付与 ネットワーク情報を取得する LINEに通知を送る 【Google Colab】Seleniumを使うための準備 【Google Colab】ノートブック上に画像を表示する 【Google Colab】Googleドライブをマウントする 【Google Colab】Googleスプレッドシートの中身を読み取る テスト用サイト 開発者ツールでXPathで要素を取得する 最新のChromeDriverのバージョンが知りたい ドキュメント等 とりあえず動かす $ pip install selenium $ pip install webdriver-manager selenium-sample.py import time from selenium import webdriver from webdriver_manager.chrome import ChromeDriverManager # Webdriver ManagerでChromeDriverを取得 driver = webdriver.Chrome(ChromeDriverManager().install()) # Qiitaトップページにアクセス driver.get('https://qiita.com/') time.sleep(5) driver.close() driver.quit() 終了時に閉じる Seleniumを触っていくうちに、いつの間にかWebDriverのプロセスが増殖しがちです。 それをできるだけ防ぐためにプログラム終了時にDriverを終了させておきます。 import atexit # プログラム終了時に実行させる関数 def tear_down(): driver.close() driver.quit() # 終了ハンドラの設定 atexit.register(tear_down) Seleniumのドキュメントのサンプルコードを見ると、with構文で書かれていたので、こちらの方がスマートかもしれません。 with webdriver.Chrome(ChromeDriverManager().install()) as driver: # Qiitaトップページにアクセス driver.get('https://qiita.com/') time.sleep(5) WebDriverを自動更新する 既に上で使っていますが、Webdriver Managerを使うと、自分でブラウザのバージョンに合ったdriverを取ってきたりする手間が省けます。 Webブラウザが更新されて、driverの対応バージョンと合わずエラーになるということも防げます。 $ pip install webdriver-manager from webdriver_manager.chrome import ChromeDriverManager # Webdriver ManagerでChromeDriverを取得 driver = webdriver.Chrome(ChromeDriverManager().install()) 参考：Python+SeleniumWebDriverではwebdriver_managerを使うといちいちdriverのexeを置き換えなくて済む 要素の取得・操作 PythonのSelenium APIドキュメントやSelenium API(逆引き)、Python + Selenium で Chrome の自動操作を一通りを参照するのがわかりやすいです。 あえて書くこともないので、ここでは、テスト自動化練習サイトの会員登録ページで使ったコードを貼り付けておきます。 from selenium.webdriver.common.keys import Keys from selenium.webdriver.support.select import Select # フォームに会員情報を登録する関数 def register_member(email, password, username, rank, address, tel, gender, birthday, notification): # 性別用 gender_index = {"男性": "1", "女性": "2", "その他": "9"} if gender in gender_index: gender_value = gender_index[gender] else: gender_value = "0" # 会員種別用 rank_value = 'normal' if rank == '一般会員' else 'premium' # 各項目の入力 driver.find_element_by_name("email").send_keys(email) driver.find_element_by_name("password").send_keys(password) driver.find_element_by_name("password-confirmation").send_keys(password) driver.find_element_by_name("username").send_keys(username) driver.find_element_by_xpath('//input[@name="rank" and @value="' + rank_value + '"]').click() driver.find_element_by_name("address").send_keys(address) driver.find_element_by_name("tel").send_keys(tel) # 性別の入力 select = Select(driver.find_element_by_id("gender")) select.select_by_value(gender_value) # 生年月日の入力 driver.find_element_by_name("birthday").clear() driver.find_element_by_name("birthday").send_keys(birthday) # お知らせを受け取る場合にチェックボックスにチェック if notification == "受け取る": driver.find_element_by_name("notification").click() driver.find_element_by_xpath('//button[text()="登録"]').click(); # 登録完了ページのスクリーンショットを取得 driver.save_screenshot('complete.png') driver.find_element_by_xpath('//button[text()="ログアウト"]').click(); # テスト用の会員登録フォームに移動 driver.get('https://hotel.testplanisphere.dev/ja/signup.html') # 会員登録 register_member("murtwiry@example.com", "fpfuo2x9", "鈴木　陽葵", "一般会員", "北海道札幌市中央区北３条西６丁目", "09022223333", "女性", "11/22/1995", "受け取る") スクリーンショットを撮る Chrome DevTools Protocolコマンドを使ったChromeでのページ全体のスクリーンショット取得は、本記事内の【Chrome】ページ全体のスクリーンショットを撮るもご参照ください。 普通のスクリーンショット取得 普通にブラウザに表示されている範囲のスクリーンショットを取得する場合は、以下のコードでできます。 import os driver.get('https://qiita.com/') # スクリーンショットを取得し保存 filename = os.path.join(os.path.dirname(os.path.abspath(__file__)), "screenshot.png") driver.save_screenshot(filename) ページ全体のスクリーンショット(ヘッドレスモード) JavaScriptでページ全体のサイズを取得し、ウィンドウをページ全体のサイズに設定してから、スクリーンショットを撮影します。 def save_full_screenshot(driver, img_path): width = driver.execute_script("return document.body.scrollWidth;") height = driver.execute_script("return document.body.scrollHeight;") driver.set_window_size(width, height) driver.save_screenshot(img_path) save_full_screenshot(driver, filename) Headlessモードで起動 ChromeでHeadlessモードを利用する場合は、--headlessのオプションを使います。 環境や対象ページによって、その他にオプションが必要になることがあります。 from selenium.webdriver.chrome.options import Options # Headless Chromeのためのオプション options = webdriver.ChromeOptions() options.add_argument('--headless') options.add_argument('--no-sandbox') options.add_argument('--disable-gpu') # Webdriver ManagerでChromeDriverを取得 driver = webdriver.Chrome( executable_path=ChromeDriverManager().install(), options=options ) 参考：Getting Started with Headless Chrome JavaScript実行 driver.execute_scriptメソッドを使います。 # Qiitaトップページにアクセス driver.get('https://qiita.com/') # 投稿記事のタイトル一覧を取得して表示 titles = driver.execute_script(''' var nodeList = document.querySelectorAll('article > h2 > a'); titles = Array.from(nodeList).map(a => a.textContent); return titles; ''') for title in titles: print(title) 参考：JavaScriptを書ける人であれば誰でもスクレイピングはできるよっていう話 要素をクリックする直前・直後に何か処理をする 要素のクリックイベントの前後に処理を行うことができます。 その他にもいくつかイベントの前後に処理を行うことができるものが存在します。詳しくは、abstract_event_listenerのAPIドキュメントをご参照ください。 from selenium.webdriver.support.events import EventFiringWebDriver, AbstractEventListener # イベント補足クラスを定義 class CustomListener(AbstractEventListener): # クリック前に要素までスクロール def before_click(self, element, driver): driver.execute_script('arguments[0].scrollIntoView({behavior: "smooth", block: "center"});', element) # クリック後に3秒待つ def after_click(self, element, driver): time.sleep(3) base_driver = webdriver.Chrome(ChromeDriverManager().install()) driver = EventFiringWebDriver(base_driver, CustomListener()) 参考：【Python】before_click/after_click・・・要素がクリックされる直前/直後の処理を実施する Basic認証を突破する driver.get("http://username:password@example.com") 参考：SeleniumでBasic認証有のサイトにログインする ユーザ名が例えば、mail@example.comのようなメールアドレスの場合は、@を%40に置き換えて、以下のようにします。 driver.get("http://mail%40example.com:password@example.com") @や#等、URLの中で特別な意味を持つ文字は、パーセントエンコーディングしてあげる必要があります。 参考：Percent-encoding (パーセントエンコーディング) | MDN Web Docs Chrome DevTools Protocol関連 Chrome系のブラウザに限定されるかもしれませんが、ブラウザの開発者ツールにあるような機能を実行できるChrome DevTools Protocolに関するスニペットです。 Chrome DevTools Protocolコマンドの実行 driver.execute_cdp_cmd("Chrome DevTools Protocolのコマンド", "コマンドのパラメータ") 参考：execute_cdp_cmd 【Chrome】ページ全体のスクリーンショットを撮る 詳細記事：【Chrome】Seleniumでページ全体のスクリーンショットを撮る import base64 def take_full_screenshot(driver, file_path): width = driver.execute_script("return document.body.scrollWidth;") height = driver.execute_script("return document.body.scrollHeight;") viewport = { "x": 0, "y": 0, "width": width, "height": height, "scale": 1 } # Chrome Devtools Protocolコマンドを実行し、取得できるBase64形式の画像データをデコードしてファイルに保存 image_base64 = driver.execute_cdp_cmd("Page.captureScreenshot", {"clip": viewport, "captureBeyondViewport": True}) image = base64.b64decode(image_base64["data"]) with open(file_path, 'bw') as f: f.write(image) driver.get('https://qiita.com/') take_full_screenshot(driver, 'screenshot.png') 【Headless Chrome】WebページをPDFに保存する 詳細記事：SeleniumとHeadless ChromeでページをPDFに保存する import base64 def save_to_pdf(driver, file_path): parameters = { "printBackground": True, # 背景画像を印刷 "paperWidth": 8.27, # A4用紙の横 210mmをインチで指定 "paperHeight": 11.69, # A4用紙の縦 297mmをインチで指定 } # Chrome Devtools Protocolコマンドを実行し、取得できるBase64形式のPDFデータをデコードしてファイルに保存 pdf_base64 = driver.execute_cdp_cmd("Page.printToPDF", parameters) pdf = base64.b64decode(pdf_base64["data"]) with open(file_path, 'bw') as f: f.write(pdf) driver.get('https://qiita.com/') save_to_pdf(driver, 'qiita.pdf') 【Chrome】ページをMHTML形式で1つのファイルに保存する WebページをMHTML形式(Webページを1つのファイルに保存するための形式)で保存することができます。 cdp-mhtml.py from selenium import webdriver from webdriver_manager.chrome import ChromeDriverManager import time with webdriver.Chrome(ChromeDriverManager().install()) as driver: driver.get('https://qiita.com/') # ページをMHTML形式で保存 mhl = driver.execute_cdp_cmd("Page.captureSnapshot", {}) time.sleep(3) with open('qiita.mhtml', 'w') as f: f.write(mhl["data"]) Page.captureSnapshotメソッドを利用 {"code":-32000,"message":"Failed to generate MHTML"}というエラーが発生することがありました。原因はわかりませんが、time.sleep(3)をとりあえず入れて回避しています。 【Chrome】カスタムヘッダの付与 custom_headers = { "X-Custom-Header1": "value1", "X-Custom-Header2": "value2", } driver.execute_cdp_cmd("Network.enable", {}) driver.execute_cdp_cmd("Network.setExtraHTTPHeaders", {"headers": custom_headers}) ネットワーク情報を取得する パフォーマンス情報内に含まれるネットワーク情報を取得できます。 from selenium.webdriver.common.desired_capabilities import DesiredCapabilities # ネットワークに関するログを取得するための設定 capabilities = DesiredCapabilities.CHROME capabilities['goog:loggingPrefs'] = { 'performance': 'ALL' } driver = webdriver.Chrome( executable_path=ChromeDriverManager().install(), desired_capabilities=capabilities ) driver.get('https://qiita.com/') # ネットワーク情報のログをファイルに書き出す with open('network.log', 'w') as f: for entry_json in driver.get_log('performance'): print(entry_json, file=f) 参考：【Python】Selenium側からネットワーク情報を取得する LINEに通知を送る 直接、Seleniumと関係ありませんが、Seleniumを使ったWebサイトの死活監視等の際の通知用に。 詳細記事：SeleniumでWeb監視をしてLINEに通知する import os import requests # LINE Notifyのアクセストークン ACCESS_TOKEN = "YOUR ACCESS TOKEN" LINE_API = "https://notify-api.line.me/api/notify" # 結果をLINEへ送信 def notify_to_line(message, image_file_path=None): headers = {'Authorization': 'Bearer ' + ACCESS_TOKEN} payload = {'message': message} files = {} if ((image_file_path is not None) and os.path.exists(image_file_path)): files = {'imageFile': open(image_file_path, 'rb')} response = requests.post(LINE_API, headers=headers, params=payload, files=files) # テキストのみ送信 notify_to_line('テストメッセージ') # テキスト + 画像を送信 notify_to_line('スクリーンショット付きメッセージ', 'qiita.png') Google Colaboratory関連 Google Colaboratory（Google Colab）を使うと、Webブラウザ上でSeleniumを使う環境を整えることができます。 また、Googleドライブにファイルを保存したり、Googleスプレッドシートと連携することもできます。 関連記事：【Selenium】Googleスプレッドシートを読み込んで会員登録フォームに入力する 【Google Colab】Seleniumを使うための準備 ChromeDriverと日本語フォントを導入して、WebDriverを使えるようにします。 # ChromeDriver、日本語フォントの導入 !apt-get update !apt-get install -y chromium-chromedriver fonts-ipafont fonts-ipaexfont !ln -s /usr/lib/chromium-browser/chromedriver /usr/bin/chromedriver !pip install selenium from selenium import webdriver # WebDriverを準備 options = webdriver.ChromeOptions() options.add_argument('--headless') options.add_argument('--no-sandbox') options.add_argument('--disable-dev-shm-usage') options.add_argument('--disable-gpu') options.add_argument('--window-size=1280,1400') driver = webdriver.Chrome(executable_path="/usr/bin/chromedriver", options=options) 参考：Googlecolab を使って iRuby で selenium と chromium で徘徊する為に 【Google Colab】ノートブック上に画像を表示する ヘッドレスChromeを使うため動作が見えないので、スクリーンショットを撮って見ていきます。 この時に、以下のようにすることで、ノートブック上に画像を表示することができます。 # ノートブック上に画像を表示するために導入 from IPython.display import Image, display_png # スクリーンショット取得 driver.get('https://qiita.com/') driver.save_screenshot('qiita.png') # ノートブック上に表示 display_png(Image('qiita.png')) 【Google Colab】Googleドライブをマウントする Google ColabからGoogleドライブを使うことができます。 実行すると、Googleドライブ利用の許可を求められるので、指示に従ってコードを入力します。 from google.colab import drive # Googleドライブをマウント drive.mount('/content/drive') # スクリーンショットをGoogleドライブの「Colab」という事前に作成したフォルダに保存 driver.save_screenshot('/content/drive/MyDrive/Colab/screenshot.png') 【Google Colab】Googleスプレッドシートの中身を読み取る Googleスプレッドシートにアクセスすることもできます。 from google.colab import auth from oauth2client.client import GoogleCredentials import gspread # 認証処理 auth.authenticate_user() gc = gspread.authorize(GoogleCredentials.get_application_default()) # 「url_list」という名前のスプレッドシートの先頭ワークシートをオープンして、値を読み込み worksheet = gc.open('url_list').get_worksheet(0) url_list = worksheet.get_all_values() # 中身を表示 for url in url_list: print(url) その他 スニペットとは違いますが、関連する情報です。 テスト用サイト Seleniumの検証でお世話になっているサイトです。 テスト用サイト - 日本Seleniumユーザーコミュニティ HOTEL PLANISPHERE - テスト自動化練習サイト 開発者ツールでXPathで要素を取得する Chromeの開発者ツールで、XPathに対応する要素を取得することができます。 $x('//input[@id="term"]')等を開発者ツールのコンソールに入力します。 確認してみると、XPathで一つに特定できていると思っていたのが、実は該当する要素が2つ以上あったというのは、良くあることです。 参考：ChromeでXPathを取る・検証する 最新のChromeDriverのバージョンが知りたい Linux等にChromeDriverをインストールする際のTipsです。 $ CHROMEDRIVER_VERSION=`curl -sS chromedriver.storage.googleapis.com/LATEST_RELEASE` $ echo ${CHROMEDRIVER_VERSION} 関連：【2019年版】Ubuntu18.04 にChromeとSeleniumをインストール ドキュメント等 https://pypi.org/project/selenium/ https://www.selenium.dev/documentation/en/ selenium package API Selenium API(逆引き) おわりに Seleniumから脱線している箇所も多々ありますが、自分のためのスニペット集をまとめました。もし気が向いたら更新をしたいと思います。

はじめに 本記事はSeleniumで私が個人的に使うためのスニペットのまとめです。 自分が使うためのスニペット集なので内容は偏っています。また一部Seleniumとは直接関係ないものも含まれています。 Seleniumの一通りの使い方を知りたい方は、以下のような素敵な記事がありますので、そちらをご参照ください。 Python + Selenium で Chrome の自動操作を一通り Selenium webdriverよく使う操作メソッドまとめ ○ 環境 主に以下の環境向けの情報です。他の環境では一部動作しないものがあると思います。 ・macOS Catalina ・Python 3.8.3 ・selenium 3.141.0 ・Google Chrome 91.0.4472.19 目次（スニペット+α） とりあえず動かす 終了時に閉じる WebDriverを自動更新する 要素の取得・操作 普通のスクリーンショット取得 ページ全体のスクリーンショット(ヘッドレスモード) Headlessモードで起動 JavaScript実行 要素をクリックする直前・直後に何か処理をする Basic認証を突破する Chrome DevTools Protocolコマンドの実行 【Chrome】ページ全体のスクリーンショットを撮る 【Headless Chrome】WebページをPDFに保存する 【Chrome】ページをMHTML形式で1つのファイルに保存する 【Chrome】カスタムヘッダの付与 ネットワーク情報を取得する LINEに通知を送る 【Google Colab】Seleniumを使うための準備 【Google Colab】ノートブック上に画像を表示する 【Google Colab】Googleドライブをマウントする 【Google Colab】Googleスプレッドシートの中身を読み取る テスト用サイト 開発者ツールでXPathで要素を取得する 最新のChromeDriverのバージョンが知りたい ドキュメント等 スニペット集 ここからが、スニペット集です。 とりあえず動かす $ pip install selenium $ pip install webdriver_manager selenium-sample.py import time from selenium import webdriver from webdriver_manager.chrome import ChromeDriverManager # Webdriver ManagerでChromeDriverを取得 driver = webdriver.Chrome(ChromeDriverManager().install()) # Qiitaトップページにアクセス driver.get('https://qiita.com/') time.sleep(5) driver.close() driver.quit() 終了時に閉じる Seleniumを触っていくうちに、いつの間にかWebDriverのプロセスが増殖しがちです。 それをできるだけ防ぐためにプログラム終了時にDriverを終了させておきます。 import atexit # プログラム終了時に実行させる関数 def tear_down(): driver.close() driver.quit() # 終了ハンドラの設定 atexit.register(tear_down) Seleniumのドキュメントのサンプルコードを見ると、with構文で書かれていたので、こちらの方がスマートかもしれません。 with webdriver.Chrome(ChromeDriverManager().install()) as driver: # Qiitaトップページにアクセス driver.get('https://qiita.com/') time.sleep(5) WebDriverを自動更新する 既に上で使っていますが、Webdriver Managerを使うと、自分でブラウザのバージョンに合ったdriverを取ってきたりする手間が省けます。 Webブラウザが更新されて、driverの対応バージョンと合わずエラーになるということも防げます。 $ pip install webdriver_manager from webdriver_manager.chrome import ChromeDriverManager # Webdriver ManagerでChromeDriverを取得 driver = webdriver.Chrome(ChromeDriverManager().install()) 参考：Python+SeleniumWebDriverではwebdriver_managerを使うといちいちdriverのexeを置き換えなくて済む 要素の取得・操作 PythonのSelenium APIドキュメントやSelenium API(逆引き)、Python + Selenium で Chrome の自動操作を一通りを参照するのがわかりやすいです。 あえて書くこともないので、ここでは、テスト自動化練習サイトの会員登録ページで使ったコードを貼り付けておきます。 from selenium.webdriver.common.keys import Keys from selenium.webdriver.support.select import Select # フォームに会員情報を登録する関数 def register_member(email, password, username, rank, address, tel, gender, birthday, notification): # 性別用 gender_index = {"男性": "1", "女性": "2", "その他": "9"} if gender in gender_index: gender_value = gender_index[gender] else: gender_value = "0" # 会員種別用 rank_value = 'normal' if rank == '一般会員' else 'premium' # 各項目の入力 driver.find_element_by_name("email").send_keys(email) driver.find_element_by_name("password").send_keys(password) driver.find_element_by_name("password-confirmation").send_keys(password) driver.find_element_by_name("username").send_keys(username) driver.find_element_by_xpath('//input[@name="rank" and @value="' + rank_value + '"]').click() driver.find_element_by_name("address").send_keys(address) driver.find_element_by_name("tel").send_keys(tel) # 性別の入力 select = Select(driver.find_element_by_id("gender")) select.select_by_value(gender_value) # 生年月日の入力 driver.find_element_by_name("birthday").clear() driver.find_element_by_name("birthday").send_keys(birthday) # お知らせを受け取る場合にチェックボックスにチェック if notification == "受け取る": driver.find_element_by_name("notification").click() driver.find_element_by_xpath('//button[text()="登録"]').click(); # 登録完了ページのスクリーンショットを取得 driver.save_screenshot('complete.png') driver.find_element_by_xpath('//button[text()="ログアウト"]').click(); # テスト用の会員登録フォームに移動 driver.get('https://hotel.testplanisphere.dev/ja/signup.html') # 会員登録 register_member("murtwiry@example.com", "fpfuo2x9", "鈴木　陽葵", "一般会員", "北海道札幌市中央区北３条西６丁目", "09022223333", "女性", "11/22/1995", "受け取る") スクリーンショットを撮る Chrome DevTools Protocolコマンドを使ったChromeでのページ全体のスクリーンショット取得は、本記事内の【Chrome】ページ全体のスクリーンショットを撮るもご参照ください。 普通のスクリーンショット取得 普通にブラウザに表示されている範囲のスクリーンショットを取得する場合は、以下のコードでできます。 import os driver.get('https://qiita.com/') # スクリーンショットを取得し保存 filename = os.path.join(os.path.dirname(os.path.abspath(__file__)), "screenshot.png") driver.save_screenshot(filename) ページ全体のスクリーンショット(ヘッドレスモード) JavaScriptでページ全体のサイズを取得し、ウィンドウをページ全体のサイズに設定してから、スクリーンショットを撮影します。 def save_full_screenshot(driver, img_path): width = driver.execute_script("return document.body.scrollWidth;") height = driver.execute_script("return document.body.scrollHeight;") driver.set_window_size(width, height) driver.save_screenshot(img_path) save_full_screenshot(driver, filename) Headlessモードで起動 ChromeでHeadlessモードを利用する場合は、--headlessのオプションを使います。 環境や対象ページによって、その他にオプションが必要になることがあります。 from selenium.webdriver.chrome.options import Options # Headless Chromeのためのオプション options = webdriver.ChromeOptions() options.add_argument('--headless') options.add_argument('--no-sandbox') options.add_argument('--disable-gpu') # Webdriver ManagerでChromeDriverを取得 driver = webdriver.Chrome( executable_path=ChromeDriverManager().install(), options=options ) 参考：Getting Started with Headless Chrome JavaScript実行 driver.execute_scriptメソッドを使います。 # Qiitaトップページにアクセス driver.get('https://qiita.com/') # 投稿記事のタイトル一覧を取得して表示 titles = driver.execute_script(''' var nodeList = document.querySelectorAll('article > h2 > a'); titles = Array.from(nodeList).map(a => a.textContent); return titles; ''') for title in titles: print(title) 参考：JavaScriptを書ける人であれば誰でもスクレイピングはできるよっていう話 要素をクリックする直前・直後に何か処理をする 要素のクリックイベントの前後に処理を行うことができます。 その他にもいくつかイベントの前後に処理を行うことができるものが存在します。詳しくは、abstract_event_listenerのAPIドキュメントをご参照ください。 from selenium.webdriver.support.events import EventFiringWebDriver, AbstractEventListener # イベント補足クラスを定義 class CustomListener(AbstractEventListener): # クリック前に要素までスクロール def before_click(self, element, driver): driver.execute_script('arguments[0].scrollIntoView({behavior: "smooth", block: "center"});', element) # クリック後に3秒待つ def after_click(self, element, driver): time.sleep(3) base_driver = webdriver.Chrome(ChromeDriverManager().install()) driver = EventFiringWebDriver(base_driver, CustomListener()) 参考：【Python】before_click/after_click・・・要素がクリックされる直前/直後の処理を実施する Basic認証を突破する driver.get("http://username:password@example.com") 参考：SeleniumでBasic認証有のサイトにログインする ユーザ名が例えば、mail@example.comのようなメールアドレスの場合は、@を%40に置き換えて、以下のようにします。 driver.get("http://mail%40example.com:password@example.com") @や#等、URLの中で特別な意味を持つ文字は、パーセントエンコーディングしてあげる必要があります。 参考：Percent-encoding (パーセントエンコーディング) | MDN Web Docs Chrome DevTools Protocol関連 Chrome系のブラウザに限定されるかもしれませんが、ブラウザの開発者ツールにあるような機能を実行できるChrome DevTools Protocolに関するスニペットです。 Chrome DevTools Protocolコマンドの実行 driver.execute_cdp_cmd("Chrome DevTools Protocolのコマンド", "コマンドのパラメータ") 参考：execute_cdp_cmd 【Chrome】ページ全体のスクリーンショットを撮る 詳細記事：【Chrome】Seleniumでページ全体のスクリーンショットを撮る import base64 def take_full_screenshot(driver, file_path): width = driver.execute_script("return document.body.scrollWidth;") height = driver.execute_script("return document.body.scrollHeight;") viewport = { "x": 0, "y": 0, "width": width, "height": height, "scale": 1 } # Chrome Devtools Protocolコマンドを実行し、取得できるBase64形式の画像データをデコードしてファイルに保存 image_base64 = driver.execute_cdp_cmd("Page.captureScreenshot", {"clip": viewport, "captureBeyondViewport": True}) image = base64.b64decode(image_base64["data"]) with open(file_path, 'bw') as f: f.write(image) driver.get('https://qiita.com/') take_full_screenshot(driver, 'screenshot.png') 【Headless Chrome】WebページをPDFに保存する 詳細記事：SeleniumとHeadless ChromeでページをPDFに保存する import base64 def save_to_pdf(driver, file_path): parameters = { "printBackground": True, # 背景画像を印刷 "paperWidth": 8.27, # A4用紙の横 210mmをインチで指定 "paperHeight": 11.69, # A4用紙の縦 297mmをインチで指定 } # Chrome Devtools Protocolコマンドを実行し、取得できるBase64形式のPDFデータをデコードしてファイルに保存 pdf_base64 = driver.execute_cdp_cmd("Page.printToPDF", parameters) pdf = base64.b64decode(pdf_base64["data"]) with open(file_path, 'bw') as f: f.write(pdf) driver.get('https://qiita.com/') save_to_pdf(driver, 'qiita.pdf') 【Chrome】ページをMHTML形式で1つのファイルに保存する WebページをMHTML形式(Webページを1つのファイルに保存するための形式)で保存することができます。 cdp-mhtml.py from selenium import webdriver from webdriver_manager.chrome import ChromeDriverManager import time with webdriver.Chrome(ChromeDriverManager().install()) as driver: driver.get('https://qiita.com/') # ページをMHTML形式で保存 mhl = driver.execute_cdp_cmd("Page.captureSnapshot", {}) time.sleep(3) with open('qiita.mhtml', 'w') as f: f.write(mhl["data"]) Page.captureSnapshotメソッドを利用 {"code":-32000,"message":"Failed to generate MHTML"}というエラーが発生することがありました。原因はわかりませんが、time.sleep(3)をとりあえず入れて回避しています。 【Chrome】カスタムヘッダの付与 custom_headers = { "X-Custom-Header1": "value1", "X-Custom-Header2": "value2", } driver.execute_cdp_cmd("Network.enable", {}) driver.execute_cdp_cmd("Network.setExtraHTTPHeaders", {"headers": custom_headers}) ネットワーク情報を取得する パフォーマンス情報内に含まれるネットワーク情報を取得できます。 from selenium.webdriver.common.desired_capabilities import DesiredCapabilities # ネットワークに関するログを取得するための設定 capabilities = DesiredCapabilities.CHROME capabilities['goog:loggingPrefs'] = { 'performance': 'ALL' } driver = webdriver.Chrome( executable_path=ChromeDriverManager().install(), desired_capabilities=capabilities ) driver.get('https://qiita.com/') # ネットワーク情報のログをファイルに書き出す with open('network.log', 'w') as f: for entry_json in driver.get_log('performance'): print(entry_json, file=f) 参考：【Python】Selenium側からネットワーク情報を取得する LINEに通知を送る 直接、Seleniumと関係ありませんが、Seleniumを使ったWebサイトの死活監視等の際の通知用に。 詳細記事：SeleniumでWeb監視をしてLINEに通知する import os import requests # LINE Notifyのアクセストークン ACCESS_TOKEN = "YOUR ACCESS TOKEN" LINE_API = "https://notify-api.line.me/api/notify" # 結果をLINEへ送信 def notify_to_line(message, image_file_path=None): headers = {'Authorization': 'Bearer ' + ACCESS_TOKEN} payload = {'message': message} files = {} if ((image_file_path is not None) and os.path.exists(image_file_path)): files = {'imageFile': open(image_file_path, 'rb')} response = requests.post(LINE_API, headers=headers, params=payload, files=files) # テキストのみ送信 notify_to_line('テストメッセージ') # テキスト + 画像を送信 notify_to_line('スクリーンショット付きメッセージ', 'qiita.png') Google Colaboratory関連 Google Colaboratory（Google Colab）を使うと、Webブラウザ上でSeleniumを使う環境を整えることができます。 また、Googleドライブにファイルを保存したり、Googleスプレッドシートと連携することもできます。 関連記事：【Selenium】Googleスプレッドシートを読み込んで会員登録フォームに入力する 【Google Colab】Seleniumを使うための準備 ChromeDriverと日本語フォントを導入して、WebDriverを使えるようにします。 # ChromeDriver、日本語フォントの導入 !apt-get update !apt-get install -y chromium-chromedriver fonts-ipafont fonts-ipaexfont !ln -s /usr/lib/chromium-browser/chromedriver /usr/bin/chromedriver !pip install selenium from selenium import webdriver # WebDriverを準備 options = webdriver.ChromeOptions() options.add_argument('--headless') options.add_argument('--no-sandbox') options.add_argument('--disable-dev-shm-usage') options.add_argument('--disable-gpu') options.add_argument('--window-size=1280,1400') driver = webdriver.Chrome(executable_path="/usr/bin/chromedriver", options=options) 参考：Googlecolab を使って iRuby で selenium と chromium で徘徊する為に 【Google Colab】ノートブック上に画像を表示する ヘッドレスChromeを使うため動作が見えないので、スクリーンショットを撮って見ていきます。 この時に、以下のようにすることで、ノートブック上に画像を表示することができます。 # ノートブック上に画像を表示するために導入 from IPython.display import Image, display_png # スクリーンショット取得 driver.get('https://qiita.com/') driver.save_screenshot('qiita.png') # ノートブック上に表示 display_png(Image('qiita.png')) 【Google Colab】Googleドライブをマウントする Google ColabからGoogleドライブを使うことができます。 実行すると、Googleドライブ利用の許可を求められるので、指示に従ってコードを入力します。 from google.colab import drive # Googleドライブをマウント drive.mount('/content/drive') # スクリーンショットをGoogleドライブの「Colab」という事前に作成したフォルダに保存 driver.save_screenshot('/content/drive/MyDrive/Colab/screenshot.png') 【Google Colab】Googleスプレッドシートの中身を読み取る Googleスプレッドシートにアクセスすることもできます。 from google.colab import auth from oauth2client.client import GoogleCredentials import gspread # 認証処理 auth.authenticate_user() gc = gspread.authorize(GoogleCredentials.get_application_default()) # 「url_list」という名前のスプレッドシートの先頭ワークシートをオープンして、値を読み込み worksheet = gc.open('url_list').get_worksheet(0) url_list = worksheet.get_all_values() # 中身を表示 for url in url_list: print(url) その他 スニペットとは違いますが、関連する情報です。 テスト用サイト Seleniumの検証でお世話になっているサイトです。 テスト用サイト - 日本Seleniumユーザーコミュニティ HOTEL PLANISPHERE - テスト自動化練習サイト 開発者ツールでXPathで要素を取得する Chromeの開発者ツールで、XPathに対応する要素を取得することができます。 $x('//input[@id="term"]')等を開発者ツールのコンソールに入力します。 確認してみると、XPathで一つに特定できていると思っていたのが、実は該当する要素が2つ以上あったというのは、良くあることです。 参考：ChromeでXPathを取る・検証する 最新のChromeDriverのバージョンが知りたい Linux等にChromeDriverをインストールする際のTipsです。 $ CHROMEDRIVER_VERSION=`curl -sS chromedriver.storage.googleapis.com/LATEST_RELEASE` $ echo ${CHROMEDRIVER_VERSION} 関連：【2019年版】Ubuntu18.04 にChromeとSeleniumをインストール ドキュメント等 https://pypi.org/project/selenium/ https://www.selenium.dev/documentation/en/ selenium package API Selenium API(逆引き) おわりに Seleniumから脱線している箇所も多々ありますが、自分のためのスニペット集をまとめました。20選というタイトルにしましたが、20になっていないのは、気にしないことにします。 もし気が向いたら更新をしたいと思います。

PyMC3のインストールでドツボったので、メモしておきます。 要は公式のGitHubをきちんと読みましょうということです。 環境 windows10 64bit anaconda3 2021/5版 PyMC3 3.11.2 はまったやり方 conda インストールの方法をググって、以下のサイトを見つけました。 conda-forge / packages / pymc3 ここにある通り、conda install -c conda-forge pymc3を実行してインストールすると、visual studio 2017が所定のパスにないと怒られてしまいました。 仕方なく、過去のインストール成功事例の記事を参考にやってみましたが、PyMC3のバージョン3.6など古いバージョンしか入れられなかったり、arvizのバージョン違いが怒られたりしました。 正しいやり方 PyMC3のGitHubから、以下のwindowsインストール手順が書いてあるサイトがありました。 PyMC3 Installation on Windows まず、以下のコマンドを実行して、mypm3envという仮想環境に必要なライブラリをインストールします。 # starting out with a fresh environment conda create -c msys2 -c conda-forge -n mypm3env python=3.8 mkl-service libpython m2w64-toolchain scipy matplotlib pandas conda activate mypm3env このあとで、PyMC3をインストールします。 pip install pymc3 こうやってPyMC3をインストールした後で、チュートリアルを動かすと、きちんと動きました！ 安易に記事を探すより、オリジナルの手順に沿ってやりましょうという話でした。

はじめに 　研究活動の一環でクラスタリング処理について調べる必要があったのでまとめます。この辺の記事はたくさん解説があるので、適宜ググってもらえるとわかると思います。今回は、勉強のためだと思って、アウトプットしたいと思います！ 　なお、参考にした記事は以下です。公式は英語のドキュメントです。今回の私の記事を読んでいただいて、公式ドキュメントの理解がスムーズに出来るようになっていただければ幸いです。また、@y-kさんの記事は、凝集型クラスタリングについての、とても分かりやすい解説が載っています。とても参考にさせていただきました！ ・参考１：scipy.cluster.hierarchy.linkage ・参考２：凝集型(階層型)クラスタリングを理解する 誰向けの記事？ 　この記事は、 ・凝集型クラスタリングについて知ってる ・pythonで凝集型クラスタリングをやってみたい ・1次元 or 2次元の座標データを凝集型クラスタリングで処理したい ・scipyのlikageメソッドを理解したい（入力・出力） 人向けです。 　機械学習未経験者が初心者向けに書く記事なので、ご指摘などございましたら幸いです。なお、主にlinkageメソッドの使い方を中心に書いていきます。 scipyのインストール 　一応手順として書いておきます。以下のコマンドなどを打ち込んで、scipyをインポートできる様にしましょう。 cmd_install_scipy pip install scipy linkageについて理解する 　pythonのscipyから使えるメソッドの一つである、linkageは凝集型クラスタリングのメソッドです。メソッドの使い方、指定できる融合法、結合されていくデータの格納、出力されるデータについて解説していきます。 メソッドの使い方 　メソッドの使い方は、method_linkageのようになります。入力であるX（座標データのリスト）を受け入れ、出力であるZ（結合手順を示したリスト）を出します。その際、"method_name"を引数に渡すことで、融合法を指定することができます。表１に各変数についてまとめておきます。 method_linkage Z = linkage(X, 'method_name') 表１　メソッドの変数についての説明 変数 役割 X 入力する座標データ群。リスト型 "method_name" 凝集型クラスタリングの融合法を指定する文字列。表２参考。 Z 出力されるデータ群。どのような結合手順をたどるかが記載されている。 指定できる融合法 　指定できる融合法は7つです。linkageの引数に、表２の文字列を指定すると、その手法で融合します。 　各手法についての解説は、公式ドキュメントか、@g-kさんの凝集型(階層型)クラスタリングを理解するを読んでいただけると理解できると思います。 表２　指定できる融合法　 指定する文字列 備考 "single" 単連結法(single linkage method) "complete" 完全連結法(complete linkage method) "average" 群平均法(group average method) "weighed" 重み付き平均法(weighed average method) "centroid" 重心法(centroid method) "median" メディアン法(median method) "ward" ウォード法(ward method) 結合されていくデータの格納 linkageでは、結合されていくデータを、以下の図のように管理しています。この図は処理対象の座標が5組あります。 　まず与えられる入力データXは座標のリスト型なので、一番左のように添え字が割り振られています。これら5つの座標のうち、（指定した融合法に従って）一番近い座標どうしを結合します。そのデータは、添え字が6として、データが格納されます。その時、結合した群の中に何個データあるかをカウントします。この結合処理を続けていって、最終的に結合した群の要素数が5になったら、終了です。 　とりあえず、インデックスの動きについて理解してくれればいいです。 　 出力されるデータ 　出力されるデータは、result_of_linkageのような2次元のリストです。各要素は以下の表のようになっています。結合対象は、さっきの図でいう座標（青）or結合（緑）のどっちかです。distanceは、指定した手法で割り出した距離で、number_of_sub_clusterはさっきの図でいう要素数です。 　result_of_linkageは、上から順に結合していきます。そして、必ず最後はnumber_of_sub_clusterが、元の座標数に等しくなります。 要素 備考 index1 結合対象１の添え字 index2 結合対象２の添え字 distance 結合対象１と結合対象２の距離 number_of_sub_cluster 結合対象１と結合対象２の要素数を合計した数 result_of_linkage Z = [ [index1, index2, distance, number_of_sub_cluster], [index1, index2, distance, number_of_sub_cluster], 　　　　　　　　　　　　　・ 　　　　　　　　　　　　　・ 　　　　　　　　　　　　　・ ] サンプルコード 　さあ！いつの間にか記事が長くなりました！サンプルコードを書いてみましょう。ここでは、座標が1次元の場合と、2次元の場合で書いてきます（確か公式には1次元と2次元が対応しているって書いてありました）。 入力の座標が1次元の場合 　入力の座標が1次元の場合、以下のようなサンプルコードになります。今回は単連結法を使いました。クラスタされる前の点は5つあり、Xには各点の座標（1次元）が入っています。 linkage_1dimension.py from scipy.cluster.hierarchy import linkage # 入力データ（1次元） X = [ [1], [2], [4], [5], [100] ] # 単連結法によるクラスタリング Z = linkage(X, method='single') print(Z) 　このコードを実行すると、以下のような表示がされます。これは、出力結果であるZを表示しています。 resut_of_1dimension [[ 0. 1. 1. 2.] [ 2. 3. 1. 2.] [ 5. 6. 2. 4.] [ 4. 7. 95. 5.]] 　これを解説すると、 手順１：index0とindex1を結合する（距離は１、結合した時の要素数は２）→index5に格納 手順２：index2とindex3を結合する（距離は１、結合した時の要素数は２）→index6に格納 手順３：index5とindex6を結合する（距離は２、結合した時の要素数は４）→index7に格納 手順４：index4とindex7を結合する（距離は95、結合した時の要素数は５）→要素数が5なので終了 となります。手順１で座標x=1とx=2（距離1）、手順２でx=4とx=5（距離1）が結合されるのは分かりやすいですね。それ以降はindex5以降に格納され、処理されています。最後は要素数が5になり、クラスタが1つになり、すべてがまとまりました。 入力座標が2次元の場合 　入力の座標が2次元の場合、以下のようなサンプルコードになります。1次元の時と同様に、単連結法を使いました。クラスタされる前の点は5つあり、Xには各点の座標（2次元）が入っています。 linkage_2dimension.py from scipy.cluster.hierarchy import linkage # 入力データ（2次元） X = [ [1, 1], [1, 2], [1, 4], [2, 4], [100, 100] ] # 単連結法によるクラスタリング Z = linkage(X, method='single') # 結果。クラスタリングの途中結果を示したもの。 print(Z) 　このコードを実行すると、以下のような表示がされます。これは、出力結果であるZを表示しています。 resut_of_1dimension [[ 0. 1. 1. 2. ] [ 2. 3. 1. 2. ] [ 5. 6. 2. 4. ] [ 4. 7. 137.18600512 5. ]] 　1次元の時と同様ですが、これを解説すると、 手順１：index0とindex1を結合する（距離は１、結合した時の要素数は２）→index5に格納 手順２：index2とindex3を結合する（距離は１、結合した時の要素数は２）→index6に格納 手順３：index5とindex6を結合する（距離は２、結合した時の要素数は４）→index7に格納 手順４：index4とindex7を結合する（距離は137、結合した時の要素数は５）→要素数が5なので終了 となります。 おわりに 　今回、凝集型クラスタリングの関数についてまとめました。linkageはpythonでサクッとできるのでいいですね♪。しかし、機械学習ど素人の私にとって、なんのこっちゃ分からなかったので、まとめました。特に入力と出力が分かりずらかったですね。。。 　とりあえず、自分なりのアウトプットができたので満足です。 それじゃ！

はじめに 　研究活動の一環でクラスタリング処理について調べる必要があったのでまとめます。この辺の記事はたくさん解説があるので、適宜ググってもらえるとわかると思います。今回は、勉強のためだと思って、アウトプットしたいと思います！ 　なお、参考にした記事は以下です。公式は英語のドキュメントです。今回の私の記事を読んでいただいて、公式ドキュメントの理解がスムーズに出来るようになっていただければ幸いです。また、@y-kさんの記事は、凝集型クラスタリングについての、とても分かりやすい解説が載っています。とても参考にさせていただきました！ ・参考１：scipy.cluster.hierarchy.linkage ・参考２：凝集型(階層型)クラスタリングを理解する 誰向けの記事？ 　この記事は、 ・凝集型クラスタリングについて知ってる ・pythonで凝集型クラスタリングをやってみたい ・1次元 or 2次元の座標データを凝集型クラスタリングで処理したい ・scipyのlikageメソッドを理解したい（入力・出力） 人向けです。 　機械学習未経験者が初心者向けに書く記事なので、ご指摘などございましたら幸いです。なお、主にlinkageメソッドの使い方を中心に書いていきます。 scipyのインストール 　一応手順として書いておきます。以下のコマンドなどを打ち込んで、scipyをインポートできる様にしましょう。 cmd_install_scipy pip install scipy linkageについて理解する 　pythonのscipyから使えるメソッドの一つである、linkageは凝集型クラスタリングのメソッドです。メソッドの使い方、指定できる融合法、結合されていくデータの格納、出力されるデータについて解説していきます。 メソッドの使い方 　メソッドの使い方は、method_linkageのようになります。入力であるX（座標データのリスト）を受け入れ、出力であるZ（結合手順を示したリスト）を出します。その際、"method_name"を引数に渡すことで、融合法を指定することができます。表１に各変数についてまとめておきます。 method_linkage Z = linkage(X, 'method_name') 表１　メソッドの変数についての説明 変数 役割 X 入力する座標データ群。リスト型 "method_name" 凝集型クラスタリングの融合法を指定する文字列。表２参考。 Z 出力されるデータ群。どのような結合手順をたどるかが記載されている。 指定できる融合法 　指定できる融合法は7つです。linkageの引数に、表２の文字列を指定すると、その手法で融合します。 　各手法についての解説は、公式ドキュメントか、@g-kさんの凝集型(階層型)クラスタリングを理解するを読んでいただけると理解できると思います。 表２　指定できる融合法　 指定する文字列 備考 "single" 単連結法(single linkage method) "complete" 完全連結法(complete linkage method) "average" 群平均法(group average method) "weighed" 重み付き平均法(weighed average method) "centroid" 重心法(centroid method) "median" メディアン法(median method) "ward" ウォード法(ward method) 結合されていくデータの格納 linkageでは、結合されていくデータを、以下の図のように管理しています。この図は処理対象の座標が5組あります。 　まず与えられる入力データXは座標のリスト型なので、一番左のように添え字が割り振られています。これら5つの座標のうち、（指定した融合法に従って）一番近い座標どうしを結合します。そのデータは、添え字が6として、データが格納されます。その時、結合した群の中に何個データあるかをカウントします。この結合処理を続けていって、最終的に結合した群の要素数が5になったら、終了です。 　とりあえず、インデックスの動きについて理解してくれればいいです。 　 出力されるデータ 　出力されるデータは、result_of_linkageのような2次元のリストです。各要素は以下の表のようになっています。結合対象は、さっきの図でいう座標（青）or結合（緑）のどっちかです。distanceは、指定した手法で割り出した距離で、number_of_sub_clusterはさっきの図でいう要素数です。 　result_of_linkageは、上から順に結合していきます。そして、必ず最後はnumber_of_sub_clusterが、元の座標数に等しくなります。 要素 備考 index1 結合対象１の添え字 index2 結合対象２の添え字 distance 結合対象１と結合対象２の距離 number_of_sub_cluster 結合対象１と結合対象２の要素数を合計した数 result_of_linkage Z = [ [index1, index2, distance, number_of_sub_cluster], [index1, index2, distance, number_of_sub_cluster], 　　　　　　　　　　　　　・ 　　　　　　　　　　　　　・ 　　　　　　　　　　　　　・ ] サンプルコード 　さあ！いつの間にか記事が長くなりました！サンプルコードを書いてみましょう。ここでは、座標が1次元の場合と、2次元の場合で書いてきます（確か公式には1次元と2次元が対応しているって書いてありました）。 入力の座標が1次元の場合 　入力の座標が1次元の場合、以下のようなサンプルコードになります。今回は単連結法を使いました。クラスタされる前の点は5つあり、Xには各点の座標（1次元）が入っています。 linkage_1dimension.py from scipy.cluster.hierarchy import linkage # 入力データ（1次元） X = [ [1], [2], [4], [5], [100] ] # 単連結法によるクラスタリング Z = linkage(X, method='single') print(Z) 　このコードを実行すると、以下のような表示がされます。これは、出力結果であるZを表示しています。 resut_of_1dimension [[ 0. 1. 1. 2.] [ 2. 3. 1. 2.] [ 5. 6. 2. 4.] [ 4. 7. 95. 5.]] 　これを解説すると、 手順１：index0とindex1を結合する（距離は１、結合した時の要素数は２）→index5に格納 手順２：index2とindex3を結合する（距離は１、結合した時の要素数は２）→index6に格納 手順３：index5とindex6を結合する（距離は２、結合した時の要素数は４）→index7に格納 手順４：index4とindex7を結合する（距離は95、結合した時の要素数は５）→要素数が5なので終了 となります。手順１で座標x=1とx=2（距離1）、手順２でx=4とx=5（距離1）が結合されるのは分かりやすいですね。それ以降はindex5以降に格納され、処理されています。最後は要素数が5になり、クラスタが1つになり、すべてがまとまりました。 入力座標が2次元の場合 　入力の座標が2次元の場合、以下のようなサンプルコードになります。1次元の時と同様に、単連結法を使いました。クラスタされる前の点は5つあり、Xには各点の座標（2次元）が入っています。 linkage_2dimension.py from scipy.cluster.hierarchy import linkage # 入力データ（2次元） X = [ [1, 1], [1, 2], [1, 4], [2, 4], [100, 100] ] # 単連結法によるクラスタリング Z = linkage(X, method='single') # 結果。クラスタリングの途中結果を示したもの。 print(Z) 　このコードを実行すると、以下のような表示がされます。これは、出力結果であるZを表示しています。 resut_of_1dimension [[ 0. 1. 1. 2. ] [ 2. 3. 1. 2. ] [ 5. 6. 2. 4. ] [ 4. 7. 137.18600512 5. ]] 　1次元の時と同様ですが、これを解説すると、 手順１：index0とindex1を結合する（距離は１、結合した時の要素数は２）→index5に格納 手順２：index2とindex3を結合する（距離は１、結合した時の要素数は２）→index6に格納 手順３：index5とindex6を結合する（距離は２、結合した時の要素数は４）→index7に格納 手順４：index4とindex7を結合する（距離は137、結合した時の要素数は５）→要素数が5なので終了 となります。 おわりに 　今回、凝集型クラスタリングの関数についてまとめました。linkageはpythonでサクッとできるのでいいですね♪。しかし、機械学習ど素人の私にとって、なんのこっちゃ分からなかったので、まとめました。特に入力と出力が分かりずらかったですね。。。 　とりあえず、自分なりのアウトプットができたので満足です。 それじゃ！

概要 初めてpythonファイルを階層化したとき import周りでハマった為、自分用に 環境 python 3.8.3 フォルダ構成 AAA/  ├AA.py  ├BBB/    ├BB.py    └CC.py ソース AA.py from BBB.BB import B class A(): def __init__(self): pass BB.py from CC import C class B(): def __init__(self): pass CC.py class C(): def __init__(self): pass 行動したこと BB.pyとCC.pyを実装して、動作確認した後 階層化して外から呼び出すため、フォルダ:BBBとAA.pyを作成した AA.pyを実行したら予想外のエラーが出た 発生したエラー AA.pyのerror from CC import C ModuleNotFoundError: No module named 'CC' 敗因 BB.py単体では実行できたから、import文は間違ってないはず なんで存在するのに呼び出されないのか 解決 BB.pyのimport文にディレクトリ名を追加する newBB.py from BBB.CC import C とても単純なことですが、実行できないことが意外だったため 自戒として心に刻みます 本当の敗因 フォルダ名,ファイル名,クラス名をすべて同じ名前にして作ったため 解決と同じ手段を取ったのに解決できず、発見が遅れた フォルダ名とファイル名が同じだと、importで自身を参照し、同じエラーが発生する 本当に戒めるべきなのは、ファイルの命名規則ですね