00

00.py

s = "stressed"
print(s[::-1])

実行結果

desserts

01

01.py

s = "パタトクカシーー"
print(s[::2])

実行結果

パトカー

02

02.py

ret = ''
s1 = 'パトカー'
s2 = 'タクシー'
s = ''.join([a + b for a, b in zip(s1, s2)])
print(s)

実行結果

パタトクカシーー

03

03.py

s = "Now I need a drink, alcoholic of course, after the heavy lectures involving quantum mechanics."
s = s.replace(',','')
s = s.replace('.', '')
l = s.split()
ret = [len(s) for s in l]
print(ret)

実行結果

[3, 1, 4, 1, 5, 9, 2, 6, 5, 3, 5, 8, 9, 7, 9]

04

04.py

s = "Hi He Lied Because Boron Could Not Oxidize Fluorine. New Nations Might Also Sign Peace Security Clause. Arthur King Can."
i = 1
n = (0, 4, 5, 6, 7, 8, 14, 15, 18) 
d = {i: word[0:1] if i in n else word[0:2] for i, word in enumerate(s.split())}
print(d)

実行結果

{0: 'H',
 1: 'He',
 2: 'Li',
 3: 'Be',
 4: 'B',
 5: 'C',
 6: 'N',
 7: 'O',
 8: 'F',
 9: 'Ne',
 10: 'Na',
 11: 'Mi',
 12: 'Al',
 13: 'Si',
 14: 'P',
 15: 'S',
 16: 'Cl',
 17: 'Ar',
 18: 'K',
 19: 'Ca'}

05

05.py

def ngram(s, n):
    return [s[i:i+n] for i, _ in enumerate(list(s))]

print(ngram("I am an NLPer", 2))

実行結果

['I ', ' a', 'am', 'm ', ' a', 'an', 'n ', ' N', 'NL', 'LP', 'Pe', 'er', 'r']

06

06.py

def ngram(s, n):
    return {s[i:i+n] for i, _ in enumerate(list(s))}

s1 = "paraparaparadise"
s2 = "paragraph"
X = ngram(s1, 2)
Y = ngram(s2, 2)
print("X | Y :", X | Y)
print("X & Y :", X & Y)
print("X - Y :", X - Y)
print("se" in X)
print("se" in Y)

実行結果

X | Y : {'ra', 'h', 'gr', 'pa', 'ph', 'ap', 'ar', 'ad', 'di', 'is', 'se', 'ag', 'e'}
X & Y : {'pa', 'ap', 'ra', 'ar'}
X - Y : {'ad', 'di', 'se', 'is', 'e'}
True
False

07

07.py

def template(x, y, z):
    return str(x) + "時の" + y + "は" + str(z)

print(template(x=12, y="気温", z=22.4))

実行結果

'12時の気温は22.4'

08

08.py

def cipher(s):
    return ''.join([chr(219-ord(s1)) if s1.islower() else s1 for s1 in s])

s = "Are you sure?"
s_encrypted = cipher(s)
print(s_encrypted)

実行結果

Aiv blf hfiv?

09

09.py

import random

def typoglycemia(s):
    return ' '.join([s1 if len(s1)<=4 else shuffle_string(s1) for s1 in s.split()])

def shuffle_string(s):
    return s[0] + ''.join(random.sample(s[1:-1], len(s[1:-1])))+ s[-1]

s = "I couldn't believe that I could actually understand what I was reading : the phenomenal power of the human mind ."

print(typoglycemia(s))

実行結果

I c'odulnt bievele that I colud aatcluly uasdnnertd what I was rdeaing : the poenhaenml poewr of the hamun mind .

前提条件

・Windows端末(著者はwin10)
・Pythonインストール済

行う事

AWS-CLIを使用してコマンドラインでRoute53にレコードをセットする

手順

pythonのバージョンの確認

pythonが入っているか確認(画像は3.8.1)

>python --version
Python 3.8.1

pipを使用して、AWS CLI をインストール

> pip3 install awscli

パスの確認

where /R c:\ aws

Pathの設定

Windowsボタンを押して検索ボックスで「環境変数」と入力して検索し、上で表示されたPathを設定する

【例】表示結果：「c:[userdirectory]\Python\Python37\Scripts\aws」の場合
　　　設定内容：　c:[userdirectory]\Python\Python37\Scripts
※pythonインストール時にPATHを設定してくれるオプションもありました。

AWS CLIのバージョンの確認

>aws --version
aws-cli/1.18.3 Python/3.8.1 Windows/10 botocore/1.15.3

IAMでユーザーを作成する

accessKeys.csvをダウンロードする。

認証情報の設定

> aws configure

以下を入力する。

> AWS Access Key ID [None]：csvファイルのAccess key IDを入力
> Secret access key:csvファイルのSecret access keyを入力
> Default region name [None]: ap-northeast-1
> Default output format [None]:json

実際にレコードセットしてみる

sample.jsonを作成する

{
            "Comment": "CREATE/DELETE/UPSERT a record ",
            "Changes": [{
            "Action": "CREATE",
                        "ResourceRecordSet": {
                                    "Name": "a.example.com",
                                    "Type": "A",
                                    "TTL": 300,
                                    "ResourceRecords": [{ "Value": "4.4.4.4"}]
}}]
}

コマンド change-resource-record-sets を使用し、ドメインのリソースレコードセットをホストゾーンに作成する。レコードを作成するための値は、​sample.json ファイルで指定する。

コマンド実行

aws route53 change-resource-record-sets --hosted-zone-id ZXXXXXXXXXX --change-batch file://sample.json

JSON ファイルにエラーがなければ、一意の ID とともにステータスとしてPENDING (保留中) が返される。

$ aws route53 change-resource-record-sets --hosted-zone-id ZXXXXXXXXXXX --change-batch file://sample.json
{
    "ChangeInfo": {
        "Status": "PENDING", 
        "Comment": "optional comment about the changes in this change batch request", 
        "SubmittedAt": "2018-07-10T19:39:37.757Z", 
        "Id": "/change/C3QYC83OA0KX5K"
    }
}

変更のステータスを確認するには、API コール get-change で change-resource-record-sets レスポンスの Id 値を使用します。

aws route53  get-change --id /change/C3QYC83OA0KX5K

・PENDING (保留中) は、このリクエストの変更が、まだサーバーには伝播済みでないことを示す。
・INSYNC (同期中) は、変更がサーバーに伝播済みであることを示す。

einsumを使うと多次元配列の掛け算が簡単に行えます。
記述方法には、癖がありますが、覚えてしまえば難しくありません。
einsumには、他にもいろいろな演算を行うことができますが、ここでは、多次元配列の掛け算について書きます。

2次元配列

アダマール積

各要素の積

\begin{pmatrix}
1 & 2 & 3 \\
4 & 5 & 6 \\
\end{pmatrix}
×
\begin{pmatrix}
1 & 2 & 3 \\
4 & 5 & 6 \\
\end{pmatrix}
=
\begin{pmatrix}
1 & 4 & 9 \\
16 & 25 & 36 \\
\end{pmatrix}

x = np.array([[1.,2.,3.],[4.,5.,6.]])
y = np.array([[1.,2.,3.],[4.,5.,6.]])

演算子

x*y

array([[ 1.,  4.,  9.],
       [16., 25., 36.]])

ループ

要素ごとに計算すると、以下のようになります。

# 要素ごとに計算
z = np.zeros((2,3))
for i in range(2):
    for j in range(3):
        z[i,j] += x[i,j] * y[i,j]
z

array([[ 1.,  4.,  9.],
       [16., 25., 36.]])

einsum

for文中の計算式の以下の部分の添え字をそのまま書けば良いです。
z[i,j] += x[i,j] * y[i,j]
上の例では、以下のように書きます。
xの添え字,yの添え字->zの添え字
（添え字に利用する文字な何でも構いません。）

# einsumで計算
np.einsum("ij,ij->ij", x, y)

同じ結果になりました。

array([[ 1.,  4.,  9.],
       [16., 25., 36.]])

内積

\begin{pmatrix}
1 & 2 & 3 \\
4 & 5 & 6 \\
\end{pmatrix}
\begin{pmatrix}
1 & 2 & 3 & 4 \\
5 & 6 & 7 & 8 \\
9 & 10 & 11 & 12 \\
\end{pmatrix}
=
\begin{pmatrix}
38 & 44 & 50 & 56 \\
83 & 98 & 113 & 128 \\
\end{pmatrix}

x = np.array([[1.,2.,3.],[4.,5.,6.]])
y = np.array([[1.,2.,3.,4.],[5.,6.,7.,8.],[9.,10.,11.,12.]])

演算

2次元の配列では、np.dotまたはnp.matmulで内積を計算できます。

dot

# ドット積で計算
np.dot(x, y)

array([[ 38.,  44.,  50.,  56.],
       [ 83.,  98., 113., 128.]])

matmul

# matmulで計算
np.matmul(x, y)

array([[ 38.,  44.,  50.,  56.],
       [ 83.,  98., 113., 128.]])

ループ

# 要素ごとに計算
z = np.zeros((2,4))
for i in range(3):
    for j in range(2):
        for k in range(4):
            z[j,k] += x[j,i] * y[i,k]
z

array([[ 38.,  44.,  50.,  56.],
       [ 83.,  98., 113., 128.]])

einsum

for文中の計算式の添え字をそのまま書けばよいのでした。
z[j,k] += x[j,i] * y[i,k]

# einsumで計算
np.einsum("ji,ik->jk", x, y)

array([[ 38.,  44.,  50.,  56.],
       [ 83.,  98., 113., 128.]])

3次元配列

2次元配列のバッチ処理を考えます。

バッチの次元が、１次元目の場合の内積

x = np.array([[[1.,2.,3.],[4.,5.,6.]],[[1.,2.,3.],[4.,5.,6.]]])
print("x.shape=", x.shape)
print("x=")
print(x)

x.shape= (2, 2, 3)
x=
[[[1. 2. 3.]
  [4. 5. 6.]]

 [[1. 2. 3.]
  [4. 5. 6.]]]

y = np.array([[[1.,2.,3.,4.],[5.,6.,7.,8.],[9.,10.,11.,12.]],[[1.,2.,3.,4.],[5.,6.,7.,8.],[9.,10.,11.,12.]]])
print("y.shape=", y.shape)
print("y=")
print(y)

y.shape= (2, 3, 4)
y=
[[[ 1.  2.  3.  4.]
  [ 5.  6.  7.  8.]
  [ 9. 10. 11. 12.]]

 [[ 1.  2.  3.  4.]
  [ 5.  6.  7.  8.]
  [ 9. 10. 11. 12.]]]

内積の結果は、以下となります。

array([[[ 38.,  44.,  50.,  56.],
        [ 83.,  98., 113., 128.]],

       [[ 38.,  44.,  50.,  56.],
        [ 83.,  98., 113., 128.]]])

演算

dot

# ドット積で計算
np.dot(x, y)

array([[[[ 38.,  44.,  50.,  56.],
         [ 38.,  44.,  50.,  56.]],

        [[ 83.,  98., 113., 128.],
         [ 83.,  98., 113., 128.]]],


       [[[ 38.,  44.,  50.,  56.],
         [ 38.,  44.,  50.,  56.]],

        [[ 83.,  98., 113., 128.],
         [ 83.,  98., 113., 128.]]]])

正しく計算できませんでした。

matmul

バッチが１次元目の場合、np.matmulで計算できます。

# matmulで計算
np.matmul(x, y)

array([[[ 38.,  44.,  50.,  56.],
        [ 83.,  98., 113., 128.]],

       [[ 38.,  44.,  50.,  56.],
        [ 83.,  98., 113., 128.]]])

ループ

# 要素ごとに計算
z = np.zeros((2,2,4))
for i in range(2):
    for j in range(3):
        for k in range(2):
            for l in range(4):
                z[i,k,l] += x[i,k,j] * y[i,j,l]
z

array([[[ 38.,  44.,  50.,  56.],
        [ 83.,  98., 113., 128.]],

       [[ 38.,  44.,  50.,  56.],
        [ 83.,  98., 113., 128.]]])

einsum

３次元でも同じです。for文中の計算式の添え字をそのまま書きます。
z[i,k,l] += x[i,k,j] * y[i,j,l]

# einsumで計算
np.einsum("ikj,ijl->ikl", x, y)

array([[[ 38.,  44.,  50.,  56.],
        [ 83.,  98., 113., 128.]],

       [[ 38.,  44.,  50.,  56.],
        [ 83.,  98., 113., 128.]]])

バッチの次元が、２次元目の場合の内積

RNNなどで、バッチの次元を２次元目としたい場合があります。

xt = x.transpose(1,0,2)
print("xt.shape=", xt.shape)
print("xt=")
xt

xt.shape= (2, 2, 3)
xt=
array([[[1., 2., 3.],
        [1., 2., 3.]],

       [[4., 5., 6.],
        [4., 5., 6.]]])

yt = y.transpose(1,0,2)
print("yt.shape=", yt.shape)
print("yt=")
yt

yt.shape= (3, 2, 4)
yt=
array([[[ 1.,  2.,  3.,  4.],
        [ 1.,  2.,  3.,  4.]],

       [[ 5.,  6.,  7.,  8.],
        [ 5.,  6.,  7.,  8.]],

       [[ 9., 10., 11., 12.],
        [ 9., 10., 11., 12.]]])

内積の結果は、以下となります。

array([[[ 38.,  44.,  50.,  56.],
        [ 38.,  44.,  50.,  56.]],

       [[ 83.,  98., 113., 128.],
        [ 83.,  98., 113., 128.]]])

演算

dot

# ドット積で計算
np.dot(xt, yt)

ValueError                                Traceback (most recent call last)
<ipython-input-24-a174c5fa02ae> in <module>
      1 # ドット積で計算
----> 2 np.dot(xt, yt)

<__array_function__ internals> in dot(*args, **kwargs)

ValueError: shapes (2,2,3) and (3,2,4) not aligned: 3 (dim 2) != 2 (dim 1)

エラーとなりました。

matmul

# matmulで計算
np.matmul(xt, yt)

---------------------------------------------------------------------------
ValueError                                Traceback (most recent call last)
<ipython-input-25-281cba2a720e> in <module>
      1 # matmulで計算
----> 2 np.matmul(xt, yt)

ValueError: matmul: Input operand 1 has a mismatch in its core dimension 0, with gufunc signature (n?,k),(k,m?)->(n?,m?) (size 2 is different from 3)

こちらもエラーとなりました。

ループ

# 要素ごとに計算
zt = np.zeros((2,2,4))
for i in range(2):
    for j in range(3):
        for k in range(2):
            for l in range(4):
                zt[k,i,l] += xt[k,i,j] * yt[j,i,l]
zt

array([[[ 38.,  44.,  50.,  56.],
        [ 38.,  44.,  50.,  56.]],

       [[ 83.,  98., 113., 128.],
        [ 83.,  98., 113., 128.]]])

結果を転置し、バッチを１次元目とすると結果が同じとなります。

zt.transpose(1,0,2)

array([[[ 38.,  44.,  50.,  56.],
        [ 83.,  98., 113., 128.]],

       [[ 38.,  44.,  50.,  56.],
        [ 83.,  98., 113., 128.]]])

einsum

for文中の計算式の添え字をそのままですね。
zt[k,i,l] += xt[k,i,j] * yt[j,i,l]

# einsumで計算
np.einsum("kij,jil->kil", xt, yt)

array([[[ 38.,  44.,  50.,  56.],
        [ 38.,  44.,  50.,  56.]],

       [[ 83.,  98., 113., 128.],
        [ 83.,  98., 113., 128.]]])

このようにeinsumを利用すれば、簡単に多次元配列の演算を行うことができます。
転置やtransposeにより変換を行い計算することもできますが、einsumを利用すればそのまま計算することができます。
einsumでは演算に時間がかかる場合もあるようなので、もし、性能的に問題なければ、einsumを利用すれば非常にシンプルに記述することができます。

JupyterでインタラクティブなGUI作ろうとしたけど、スライダーの幅が小さくて使いづらかったり、複数のコントロールを配置する方法が分からなかったりしたので調べてみた。

ウィジェットのスタイルの調整

CSSライクな記述でスタイルを調整できます

基本形

幅が狭く操作しづらいので、いろいろ調整したい

from ipywidgets import IntSlider

IntSlider(min=2000, max=2020,step=1,value=2010, description="year1: ")

幅と高さを変えてみる

幅を60%指定、高さを100px指定にします。

from ipywidgets import IntSlider, Layout

IntSlider(min=2000, max=2020,step=1,value=2010, description="year1: ",
                         orientation='horizontal',
                         layout=Layout(width='60%', height='100px'))

Borderもつけてみる

ついでに枠線もつけてみます。

from ipywidgets import IntSlider, Layout

IntSlider(min=2000, max=2020,step=1,value=2010, description="year1: ",
                         orientation='horizontal',
                         layout=Layout(width='60%', height='100px', border='solid'))

その他スタイルの調整

ウィジェットごとに固有なスタイルもあります。
例えば、スライダーのハンドルに色をつけることができる。

from ipywidgets import IntSlider

year1 = IntSlider(min=2000, max=2020,step=1,value=2010, description="year1: ")
year1.style.handle_color = 'lightblue'
year1

ハンドルが薄い青色になりました。

利用可能なスタイルは以下で確認できる。

year1.style.keys

他にもいろいろなスタイルを調整できます。
公式ドキュメントに利用可能なスタイルが記載されています。
Layout and Styling of Jupyter widgets — Jupyter Widgets 7.5.1 documentation

レイアウトを変える

複数のウィジェットがある場合に、レイアウトにそって並べることもできます。

横と棚に並べる

横方向のコンテナ(HBox)、縦方向のコンテナ(VBox)にウィジェットを配置できる。

from ipywidgets import IntSlider, Layout, HBox, VBox

year1 = IntSlider(min=2000, max=2020,step=1,value=2010, description="year1: ")
year2 = IntSlider(min=2000, max=2020,step=1,value=2010, description="year2: ")
year3 = IntSlider(min=2000, max=2020,step=1,value=2010, description="year3: ")
year4 = IntSlider(min=2000, max=2020,step=1,value=2010, description="year4: ")
year5 = IntSlider(min=2000, max=2020,step=1,value=2010, description="year5: ")
year6 = IntSlider(min=2000, max=2020,step=1,value=2010, description="year6: ")

VBox([HBox([year1,year2,year3]),HBox([year4,year5,year6])])

2×2で並べる

2×2のレイアウトでウィジェットを配置できます。

from ipywidgets import IntSlider, TwoByTwoLayout
year1 = IntSlider(min=2000, max=2020,step=1,value=2010, description="year1: ")
year2 = IntSlider(min=2000, max=2020,step=1,value=2010, description="year2: ")
year3 = IntSlider(min=2000, max=2020,step=1,value=2010, description="year3: ")
year4 = IntSlider(min=2000, max=2020,step=1,value=2010, description="year4: ")

TwoByTwoLayout(top_left=year1,
               top_right=year2,
               bottom_right=year3,
               bottom_left=year4)

グリッドに並べる

n × mの任意のサイズのグリッドにウィジェットを配置できます。

from ipywidgets import IntSlider, GridspecLayout

grid = GridspecLayout(2, 3)

grid[0,0] = IntSlider(min=2000, max=2020,step=1,value=2010, description="year1-1: ")
grid[0,1] = IntSlider(min=2000, max=2020,step=1,value=2010, description="year1-2: ")
grid[0,2] = IntSlider(min=2000, max=2020,step=1,value=2010, description="year1-3 ")
grid[1,0] = IntSlider(min=2000, max=2020,step=1,value=2010, description="year2-1: ")
grid[1,1] = IntSlider(min=2000, max=2020,step=1,value=2010, description="year2-2: ")
grid[1,2] = IntSlider(min=2000, max=2020,step=1,value=2010, description="year2-3: ")

grid

他のレイアウト

公式サイトの以下のドキュメントに記載があります。
アプリ風のApp Layoutなども使えます。
Using Layout Templates — Jupyter Widgets 7.5.1 documentation

まとめ

Jupyter使う場合は、凝ったUI作りこむ必要はないけど、コントロールを使いやすくしたり、ちょっとだけ見た目よくすると、より使いやすいと思います。
そういうことやりたい方の参考になればうれしいです。

はじめに

以前、「機械学習の分類」で取り上げたアルゴリズムについて、その理論とpythonでの実装、scikit-learnを使った分析についてステップバイステップで学習していく。個人の学習用として書いてるので間違いなんかは大目に見て欲しいと思います。

今回は、損失関数のパラメータ最適化の手法である勾配降下法について。数学的に解が見つかる場合はよいが、世の中そういう場合の方が少ない。最適値を探索する手法としていろんな勾配方があるが、いくつかをコードを交えてまとめてみたい。例によって参考にしたサイトは以下です。ありがとうございます。

• 確率的勾配降下法のメリットについて考えてみた
• OPTIMIZER 入門 ~線形回帰からAdamからEveまで
• 確率的勾配降下法とは何か、をPythonで動かして解説する
• 初歩からの機械学習：最急降下法による重回帰モデル～PythonとRでスクラッチから～

勾配法について

Wikipediaによると、勾配法とは

勾配法（こうばいほう、Gradient method）は、最適化問題において、関数の勾配に関する情報を解の探索に用いるアルゴリズムの総称。

と書いてある。単回帰では、$$ y = Ax+B $$という関数で近似させるために、損失関数$E$を残差の二乗和 $$ E = \sum_{i=0}^{N}(y_i - (Ax_i -B))^2 $$として、$E$を最小化するような$A$と$B$を求めた。この例では数学的に$A$と$B$を求めることができるのだが、損失関数がもっと複雑な場合であっても、最小となる（であろう）点を探すアプローチが勾配降下法である。

お題

例によってscikit-learnのdiabetes(糖尿病データ)を使います。

import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
from sklearn import datasets

diabetes = datasets.load_diabetes()

df = pd.DataFrame(diabetes.data, columns=diabetes.feature_names)

x = df['bmi'].values
y = diabetes.target
plt.scatter(x, y)

数学的に解いたときの傾きと切片はそれぞれ

傾きA:  949.43526038395
切片B:  152.1334841628967

でした。

最急勾配降下法

最急降下法(Steepest Gradient Descent)は、損失関数上のある点での勾配ベクトルと逆方向に少しずつ進んでいくというアイデアです。傾きを$\nabla E$とし、$$ E_{t+1}=E_{t} - \eta\nabla E_{t} $$を計算していきます。なお、$\eta$は学習率と呼ばれ、どの程度ゆっくり進んでいくかというものです。これが大きいと解が収束せずに発散してしまいます。

損失関数$$ E = \sum_{i=0}^{N}(y_i - (Ax_i -B))^2 $$について、$A$と$B$を求めるためには、$E$を$A$と$B$それぞれについて偏微分します。($\sum$の添え字は省略)

\frac{\partial{E}}{\partial{A}}=\sum(-2x_iy_i+2Ax_i^2+2Bx_i)
 \\
=-2\sum(Ax_i+B-y_i)x_i \\
\frac{\partial{E}}{\partial{B}}=\sum(-2x_iy_i+2Ax_i+2B)
 \\
=-2\sum(Ax_i+B-y_i) \\

となる。初期値を適当に決め、上の式を使って$A$と$B$を勾配方向に変化させていき、ほどよい段階で計算を打ち切ればよいことになる。

初期値の決め方、学習率の決め方、計算を打ち切る条件についてはさらなる考察が必要だが概略はこういうことだ。

pythonでの実装

最急降下法を実現するSteepestGradientDescentクラスを作成した。メソッドはなんとなくscikit-learnにあわせてみた。学習率は適当で、誤差が十分小さくなったらではなく回数で打ち切りとした。

class SteepestGradientDescent:
  def __init__(self, eta=0.001, n_iter=2000):
    self.eta = eta
    self.n_iter = n_iter
    self.grad = np.zeros((2,))
    self.loss = np.array([])

  def fit(self, X, Y, w0):
    self.w = w0
    for _ in range(self.n_iter):
      self.loss = np.append(self.loss, np.sum((Y-self.w[0]-self.w[1]*X)**2))
      self.grad[0] = np.sum(-2*(Y-self.w[0]-self.w[1]*X))
      self.grad[1] = np.sum(-2*X*(Y-self.w[0]-self.w[1]*X))

      self.w -= self.eta * self.grad

  def predict(self, x):
    return (self.w[0] + self.w[1]*x)

  @property
  def coef_(self):
    return self.w[1]

  @property
  def intercept_(self):
    return self.w[0]

  @property
  def loss_(self):
    return self.loss

fitメソッドの中で全データに対して勾配を算出し、係数を更新している。損失関数の値も同時に保存して、学習がどういう風に進んでいるかわかるようにした。

このクラスを使って糖尿病データに近似させ、損失の変化をグラフにプロットしてみる。

w0 = np.array([1.,1.])

model = SteepestGradientDescent()
model.fit(x, y, w0)

print("A: ", model.coef_)
print("B: ", model.intercept_)

loss = model.loss
plt.plot(np.arange(len(loss)), np.log10(loss))

plt.show()

A: 932.1335010668406
B: 152.13348416289668

数学的に解いた値と一緒になりました。損失が減っていく様子もみることができますね。

確率的勾配降下法

最急降下法は、サンプル全てを使って勾配を計算しました。確率的勾配降下法では、シャッフルしたデータを1つ〜複数使ってパラメータを更新します。複数使う方法をミニバッチ学習とも言います。

こうすると何がいいかという話なんですが、損失関数は形によって極小値を複数持つことがあります。地形を思い出して欲しいんですが、見えてる谷が一番低いとは限らず、山を越えたらもっと深い谷が見つかることもあります。この一見正しそうだけどそうではない谷底を局所解、もっと低い谷底のことを大域解と呼びます。

最急降下法は初期値の選び方で局所解におちいり易いのですが、確率的勾配降下法はパラメータ更新のたびに勾配計算に使うデータを選び直すので、低い山を越えて大域解にたどり着く可能性が高くなるそうです。

さらに、使うデータが決まっている必要がないのでリアルタイムの処理に向いてるとのことです。

pythonでの実装

先ほどと同じようにStochasticGradientDescentクラスを作りました。中身はパラメータの更新部分以外はほとんど最急降下法と同じです。ミニバッチのサイズは全データに対してどれくらい使うか指定できるようにしました。損失が最低だったパラメータを記憶しておき、それを返すようにしています。

class StochasticGradientDescent:
  def __init__(self, eta=0.01, n_iter=2000, sample_rate=0.1):
    self.eta = eta
    self.n_iter = n_iter
    self.sample_rate = sample_rate
    self.grad = np.zeros((2,))
    self.loss = np.array([])

  def fit(self, X, Y, w0):
    self.w = w0
    self.min_w = w0
    n_samples = int(np.ceil(len(X)*self.sample_rate))
    min_loss = 10**18
    for _ in range(self.n_iter):
      loss = np.sum((Y-self.w[0]-self.w[1]*X)**2)
      if min_loss>loss:
        min_loss = loss
        self.min_w = self.w
      self.loss = np.append(self.loss, loss)
      for i in range(n_samples):
        index = np.random.randint(0, len(X))
        batch_x = X[index]
        batch_y = Y[index]
        self.grad[0] = np.sum(-2*(batch_y-self.w[0]-self.w[1]*batch_x))
        self.grad[1] = np.sum(-2*batch_x*(batch_y-self.w[0]-self.w[1]*batch_x))
        self.w -= self.eta * self.grad

  def predict(self, x):
    return (self.w[0] + self.w[1]*x)

  @property
  def coef_(self):
    return self.min_w[1]

  @property
  def intercept_(self):
    return self.min_w[0]

  @property
  def loss_(self):
    return self.loss

これを実行して、損失の変化も合わせて見てみる。

w0 = np.array([1.,1.])

model = StochasticGradientDescent()
model.fit(x, y, w0)

print("A: ", model.coef_)
print("B: ", model.intercept_)

loss = model.loss
plt.plot(np.arange(len(loss)), np.log10(loss))

plt.show()

A:  925.5203159922469
B:  146.8724188770836

少し厳密解とは異なるのと、少し損失が暴れる感じですね。学習率を低くするとスパイクの高さが低くなるのと、ランダムサンプリングしている関係か、もとのサンプルの相関が低いと厳密解に収束しないのかなと思いました。この辺のパラメータのチューニングは課題ですね。

まとめ

勾配降下法として、最急降下法と買う率的勾配降下法について調べてみました。scikit-learnやディープラーニングの世界ではこれ以外でも様々な勾配方があるんですが、基本は一緒です。こうして順を追って見ていくことで、パラメータのチューニングについても想像がつきやすくなるかもしれません。

次回は回帰の総まとめと過学習や正則化あたりをまとめてみたいと思います。

以下を参考にしてサンプル画像大量ゲットできました

google image downloadが動かなかったのでその対応

よーしおじさん教師データつくっちゃうぞ、100件だと心もとないから1000件くらい画像ほしいな

ということでデバッグしました

Pythonさわったの今日がはじめてなんですけど解決できてよかった！

修正版は以下にあります

https://github.com/kituneponyo/google-images-download

俺としたことがgithubでforkしてプルリクまで投げてしまった、github童貞を捨ててしまった

機械学習と数理最適化

「コンビニのお店×商品ごとに需要予測がしたいんです」
　→　機械学習/統計モデリングっぽい話

「コンビニのお店×商品ごとに需要予測に基づいて利益の最大化をしたいんです」
　→　数理最適化っぽい話

意思決定のステップとアナリティクスの到達段階

データサイエンス/AI/アナリティクスといってもレベルは様々である

意思決定のステップ
　①集計（何が起きるのか）
　②説明（何故起きたのか）
　③予測（何が起きるのか）
　④意思決定（何をすればいいのか）
　⑤行動（実際の行動）

アナリティクスの到達段階
　①記述的アナリティクス
　②診断的アナリティクス
　③予測的アナリティクス
　④処方的アナリティクス（AIによる意思決定支援）
　⑤処方的アナリティクス（AIによる意思決定）

一般的に　
　①集計（何が起きるのか）　DWH/BIツールで行う
　　①記述的アナリティクス

　②説明（何故起きたのか）や③予測（何が起きるのか）までを機会学習/統計解析が行う
　　①記述的アナリティクス
　　②診断的アナリティクス
　　③予測的アナリティクス

　④意思決定（何をすればいいのか）や⑤行動（実際の行動）を数理最適化によって行う
　　①記述的アナリティクス
　　②診断的アナリティクス
　　③予測的アナリティクス
　　④処方的アナリティクス（AIによる意思決定支援）
　　⑤処方的アナリティクス（AIによる意思決定）

機械学習vs数理最適化

タスク	システムへの入力	システムのメカニズム	システムの出力	典型的な技術
予測・推定	⭕️	不明→推定	⭕️	機械学習・統計学
最適化	最良のものは不明→探索	⭕️	最小化・最大化したい	数理最適化

どこまでできるか？

	説明	予測	意思決定支援	意思決定
株トレード（DayTrading）	○	○	○	○（HFT）
スーパーの購買計画	○	○（需要予測）	○（最適購買）	○
建築機材のレンタル	○	○（需要予測）	○（支店間の機材配備の最適化）	○/?(支店間のインセンティブの対立)
コモディティトレード	○	○（価格予測）	○（売買タイミング示唆）	？（政治リスクの評価）
農業	○	○（収穫予想）	○(肥料最適化)/?(販売計画)	？
M&A	○	？（高い個別性）	?	?

課題・オペレーション依存になりやすい

参考

　DWHとBIツール
　　https://it-trend.jp/bi/article/bi_dwh

はじめに

前回、基本的な機械学習の手順：①分類モデルで、本当に基本的な分類モデルの作成手順を整理しました。
ただ、データを準備するところは、端折ってしまっていました。そこで、今回はBigQueryを使って下記のようなテーブルを作成し、Python環境に取り込むまでの手順を整理していきたいと思います。

id	result	product1	product2	product3	product4	product5
001	1	2500	1200	1890	530	null
002	0	750	3300	null	1250	2000

分析環境

Google BigQuery
Google Colaboratory

対象とするデータ

下記の様なテーブル構造で購買データが格納されています。

id	店舗名	購入日	購入商品	購入金額	購入個数
001	〇〇店	2020-01-10	product1	2500	1
002	☓☓店	2020-01-10	product1	750	3
002	☓☓店	2020-01-10	product2	1000	2

テーブル名は、BQ上で次のようになっています。（日別でテーブル分け）
myproject.mydataset.transaction_年月日

項目名は、残念ながら日本語ではつけられませんが、わかりやすくするためにここでは日本語にしています。
それぞれの環境に合わせて名称を変えてください。

1.BigQuery上でやること

BigQuery上でやることと書いていますが、実際の操作は全てPython（Google Colaboratory）から行っています。ちゃんと、何の処理をしたかを記録に残す意味でも、Notebookは重宝しますし。

ColaboratoryからBigQueryを操作するには、次の２行で簡単に認証できます。
URLが表示されたら自身のアカウントでアクセスして、ワンタイムパスワードを発行し、Notebookに表示されたセルに貼り付けましょう。

from google.colab import auth
auth.authenticate_user()

1-1.購買データの加工

それでは、実際に元の購買データから人別・商品別に集計する処理を進めていきましょう。

# ここでは2020年1月を対象期間とします
from_day = "20200101"
to_day = "20200131"

query=f"""
SELECT
 id, 購入商品, SUM(購入金額) AS 合計金額
 FROM `myproject.mydataset.transaction_*`
 WHERE _table_suffix between `{from_day}` and `{to_day}`
"""

# 出力するプロジェクト・データセット・テーブル名
project = "myproject"
client = bigquery.Client(project=project)
dataset = "mydataset"
ds = client.dataset(dataset )
table = "tmp"

job_config = bigquery.QueryJobConfig()
job_config.destination = ds.table(table)
job_config.write_disposition="WRITE_TRUNCATE" #テーブルに上書きする場合。追加する場合は"WRITE_APPEND"
job_config.allow_large_results=True #大規模な結果を許可するか。基本はTrue
job_config.use_legacy_sql=False #レガシーSQLを使うか。基本はFalse
job = client.query(query, job_config=job_config)

result = job.result()

これで、何も出ずに実行が完了したら、myproject.mydataset.tmpに実行結果が出力されています。
job_configでの設定が少々面倒ですが、最初はそれぞれが何の設定をしているのか意識しながら書いておきましょう。（慣れたらコピペでOK）

ここまで実行すると、次のようなテーブルになっています。

id	購入商品	合計金額
001	product1	2500
001	product2	1200
002	product1	750
002	product2	3300

1-2.人別テーブルへ加工

次に、機械学習のモデルに読み込むために、id × 購入金額のPivotTableを作成したいと思います。

PythonでPandasを使えば、pivot_tableが使えて楽なのですが、対象の期間等によっては、データサイズが数GBや数十GBあり、データ取り込みや加工に余計な時間がかかるので、PivotTable化までをBigQueryでやっています。

new_columns=[]

# 横持ちする「購入商品」の一覧を作成して、Dataframe化
new_column="購入商品"
query = f'SELECT DISTINCT {new_column} FROM `{project}.{dataset}.{tmp}` ORDER BY {new_column}'
df_product = client.query(query).to_dataframe()

## 購入商品のDataframeをループして、横持ちするSQLを自動生成
for index, row in df_product.iterrows():
    item_frag = []
    condition_frag = []

    for i, v in zip(row.index, row.values):
        formula=f"{i}='{v}'"
        condition_frag.append(formula)
        item_frag.append(v)

    item=''.join(item_frag)
    condition=' AND '.join(condition_frag)
    query_frag = f'SUM(IF(({condition}),"購入金額",0)) AS {new_column}_{item}'
    new_columns.append(query_frag)

# new_columnsというListをquery_partsという文字列に変更
query_parts = ',\n'.join(new_columns)
query = f'SELECT id + query_parts + FROM `{project}.{dataset}.{tmp}` GROUP BY id'

# 出力先のテーブル名
table2="pivoted_table"

# BigQueryの実行は同上
job_config = bigquery.QueryJobConfig()
job_config.destination = ds.table(table2)
job_config.write_disposition="WRITE_TRUNCATE" #テーブルに上書きする場合。追加する場合は"WRITE_APPEND"
job_config.allow_large_results=True #大規模な結果を許可するか。基本はTrue
job_config.use_legacy_sql=False #レガシーSQLを使うか。基本はFalse
job = client.query(query, job_config=job_config)

result = job.result()

Pivot TableのSQLを作るところの処理が、２重でループしていて少しややこしいですが、作りたいのは次のSQLです。

SELECT
 id,
SUM(IF((購入商品='product1'),購入金額,0)) AS 購入商品_product1,
SUM(IF((購入商品='product2'),購入金額,0)) AS 購入商品_product2,
SUM(IF((購入商品='product3'),購入金額,0)) AS 購入商品_product3,
...
FROM `myproject.mydataset.tmp`
GROUP BY id

テーブルをPivot化する定番のSQLですが、これのSUM(IF(～))部分を、商品数に応じて自動で作成するようにしています。
ここまで実行すると、次のテーブルがmyproject.mydataset.pivoted_tableとして保存されています。

id	product1	product2	product3	product4	product5
001	2500	1200	1890	530	null
002	750	3300	null	1250	2000

2.Pythonの環境への取り込み

最後に、BigQueryで作成したデータをPython環境にPandas Dataframeで取り込みます。
とはいえ、Pivot化の途中で既にDataframeで取り込んでいるので、プログラムを書くのも今更ですね。。。

query = f'SELECT * FROM `{project}.{dataset}.{table2}`'
df = client.query(query).to_dataframe()

おわりに

「基本的な機械学習の手順」と言いながら、今回はBigQueryでの操作に終始しました。
Colaboratoryからは、認証が簡単にできるため、BigQueryの操作へのハードルも低く感じますね。

BigQueryは今でも十分パワフルですが、日々新たな機能がリリースされています。上手く新たな機能を使いこなしていきたいですね。

• 自分用メモ
• このページに従ってやったら出来た
• AWSのEC2でjupyter notebookをたてた https://ljvmiranda921.github.io/notebook/2018/01/31/running-a-jupyter-notebook/

• 自分用メモ
• このページに従ってやったら出来た
• AWSのEC2でjupyter notebookをたてた https://ljvmiranda921.github.io/notebook/2018/01/31/running-a-jupyter-notebook/

この記事はTwitter　Dataの分析のtrend分析編である。このseriesは後で深めていく。初めての記事なので、簡単にTwitter　Dataの分析をしたら、詳しくこのcaseを説明します。日本語がきついから、英語にします。Git Repository

---

Why Analyze Twitter Data?

Twitter data analysis is used in a wide range of areas including but not restricted to analyzing the mentions of each political party in an election, detecting the reactions to the introduction of a new product, understanding the geographical scope of discussion of a news story.

How to analyze twitter data?

Twitter data can be retrieved from twitter API. Data can be wrangled and manipulated according to the purpose of analysis. For example, if we need to get insights on a certain product from twitter data, there're a few strategies we can apply:

• Collect mentions of the product and identify if people are talking about it positively. press
• Examine the size of the retweet network mentioning the product.
• Analyzing the geographical penetration of users mentioning the product. press

Collecting data via twitter API

This is a slightly complicated procedure if it's your first time to use twitter API. Here I'm going to explain how to collect data through twitter API.

1. Log in your twitter account
2. Apply for a developer account. Apply Here!
3. Create a new application.
4. Go to the application and find keys and tokens. (We'll need them later for authentication purposes.)

Authentication

from tweepy import OAuthHandler
from tweepy import API
import json

consumer_key = ...
consumer_secret = ...
access_token = ...
access_token_secret = ...
# Consumer key authentication
auth = OAuthHandler(consumer_key, consumer_secret)

# Access key authentication
auth.set_access_token(access_token, access_token_secret)

# Set up the API with the authentication handler
api = API(auth)

Getting the trend

Here I'm gonna use UK as an example. Information regarding WOE ID can be found here.

UK_WOE_ID = 23424975

UK_trends = api.trends_place(UK_WOE_ID)

trends = json.loads(json.dumps(UK_trends, indent=1))

Local and global thought patterns

# Loading json module
import json

# Load WW_trends and US_trends data into the the given variables respectively
WW_trends = json.loads(open('datasets/WWTrends.json').read())
US_trends = json.loads(open('datasets/USTrends.json').read())

# Inspecting data by printing out WW_trends and US_trends variables
print(WW_trends)
print(US_trends)

Prettifying the output

# Pretty-printing the results. First WW and then US trends.

print("WW trends:")
print(json.dumps(WW_trends, indent=1))

print("\n", "US trends:")
print(json.dumps(US_trends, indent=1))

Finding common trends

# Extracting all the WW trend names from WW_trends
world_trends = set([trend['name'] for trend in WW_trends[0]['trends']])

# Extracting all the US trend names from US_trends
us_trends = set([trend['name'] for trend in US_trends[0]['trends']]) 

# Getting the intersection of the two sets of trends
common_trends = world_trends.intersection(us_trends)

# Inspecting the data
print(world_trends, "\n")
print(us_trends, "\n")
print (len(common_trends), "common trends:", common_trends)

Exploring the hot trend

# Loading the data
tweets = json.loads(open('datasets/WeLoveTheEarth.json').read())

# Inspecting some tweets
tweets[0:2]

Digging deeper

# Extracting the text of all the tweets from the tweet object
texts = [tweet['text'] for tweet in tweets]

# Extracting screen names of users tweeting about #WeLoveTheEarth
names = [user_mention['screen_name'] for tweet in tweets for user_mention in tweet['entities']['user_mentions']]

# Extracting all the hashtags being used when talking about this topic
hashtags = [hashtag['text'] for tweet in tweets for hashtag in tweet['entities']['hashtags']]

# Inspecting the first 10 results
print (json.dumps(texts[0:10], indent=1),"\n")
print (json.dumps(names[0:10], indent=1),"\n")
print (json.dumps(hashtags[0:10], indent=1),"\n")

Frequency analysis

# Importing modules
from collections import Counter

# Counting occcurrences/ getting frequency dist of all names and hashtags
for item in [names, hashtags]:
    c = Counter(item)
    # Inspecting the 10 most common items in c
    print (c.most_common(10), "\n")

Activity around the trend

# Extracting useful information from retweets
retweets = [(tweet['retweet_count'], 
             tweet['retweeted_status']['favorite_count'],
             tweet['retweeted_status']['user']['followers_count'],
             tweet['retweeted_status']['user']['screen_name'], 
             tweet['text']) for tweet in tweets if 'retweeted_status' in tweet]

A table that speaks a 1000 words

# Importing modules
import matplotlib.pyplot as plt
import pandas as pd

# Create a DataFrame and visualize the data in a pretty and insightful format
df = pd.DataFrame(retweets, columns=['Retweets','Favorites', 'Followers', 'ScreenName', 'Text'])

df.style.background_gradient()
df = df.groupby(['ScreenName','Text','Followers']).sum().sort_values(by=['Followers'], ascending= False)

Analyzing used languages

# Extracting language for each tweet and appending it to the list of languages
tweets_languages = []
for tweet in tweets: 
    tweets_languages.append(tweet['lang'])

# Plotting the distribution of languages
%matplotlib inline
plt.hist(tweets_languages)

---

That's the end of our project. In the next project of this series we'll look into more complicated use of twitter API.

概要

　この記事は初心者の自分がRESTful なAPIとswiftでiPhone向けのクーポン配信サービスを開発した手順を順番に記事にしています。技術要素を1つずつ調べながら実装したため、とても遠回りな実装となっています。

前回の Django Rest Framework で JWTによるAPIの認証機能を実装 で認証機能を実装し、ある程度実用に耐えうるAPIとなりました。ただ現在のクーポンは文字情報だけでビジュアル面で弱いのが現状。

そこで今回はDjangoで画像ファイルを扱うための「Pillow」というパッケージを使い、クーポンの画像もAPIで配信出来るようにします。

Djangoで新しく画像を扱うAPIを作る場合も「Pillow」に関する部分は同じなので、参考になればと思います。

画像をwebAPIを介してクライアント側で表示する仕組み

1. webAPI側のサーバで画像を保持する
2. リクエストに対して画像のURLをJsonでレスポンスする
3. クライアントが再度サーバにアクセスして画像を取得し画面上に表示する

参考

• はじめてのDjango (7) 画像データの管理やページへの表示，アップロードの方法などについて知ろう
• 【Django】画像をアップロードして表示する
• Django本番環境でメディアファイル公開に必要な設定

環境

• Mac OS 10.15
• pipenv 2018.11.26
• Python 3.7.4
• Django 2.2.6
• pillow 7.0.0

手順

• pillowをインストールする
• モデルを定義する
• settings.pyにMEDIA_ROOTとMEDIA_URLを定義
• urls.pyに設定を追加する
• 必要に応じてpsettings.pyに設定を追加する
• マイグレートする
• 動作確認

pillowをインストールする

　自分の環境は pipenv でpythonの仮想環境を作っているので pipenvのシェルに入った後、下記の通りpillowをインストールします。インストール後に Pipfileを見てpillowがインストールされた事を確認します。

$ pipenv install pillow

$ cat Pipfile
[[source]]
name = "pypi"
url = "https://pypi.org/simple"
verify_ssl = true

[dev-packages]

[packages]
django = "*"
djangorestframework = "*"
django-filter = "*"
djangorestframework-jwt = "*"
pillow = "*" ←追加されている

[requires]
python_version = "3.7"

pipやpip3で環境構築している場合は、

$ sudo pip install pillow
又は
$ sudo pip3 install pillow

でインストールします。

モデルを定義する

　models.pyのCouponクラスに画像を格納するモデルフィールドを追加します。pillowをインストールすると画像ファイルを扱うフィールド型ImageFieldを使えるようになります。

　定義はこのようにします。upload_to=の部分は最低限必要です。

[フィールド名] = models.ImageField(upload_to=‘[画像ファイルを格納するフォルダの相対パス]’)

　「画像ファイルを格納するフォルダの相対パス」の起点は、settings.pyで定義するMEDIA_ROOTで指定したパスになります。

　下記の通り、imageというフィールド名でImageFieldを追加しました。
なお、私のケースではnullを許容（null=True)しないとマイグレーションファイルを作成する際にエラーになりました。

models.py

class Coupon(models.Model):
    code = models.CharField(max_length=20)
    benefit = models.CharField(max_length=1000)
    explanation = models.CharField(max_length=2000)
    image = models.ImageField(upload_to='images/', null=True) #追加
    store = models.CharField(max_length=1000)
    start = models.DateField()
    deadline = models.DateField()
    status = models.BooleanField()

settings.pyにMEDIA_ROOTとMEDIA_URLを定義

　前述の通り、プロジェクト名のフォルダ配下のsettings.pyにMEDIA_ROOTには画像ファイルを相対パスで参照する際の起点のURLを指定します。
　私の場合、画像ファイルを格納するフォルダをアプリのフォルダ配下に作りたかったので、アプリのフォルダのURLを指定しました。プロジェクトのフォルダまではBASE_DIRで定義されているので、下記のように記載しました。

　画像ファイルを格納するフォルダは初回の画像アップロード時に自動生成されるそうですが、知らなかったのでmkdir imagesで自分で作りました。

ami_coupon_api/settings.py

MEDIA_ROOT = os.path.join(BASE_DIR, 'coupon')

　MEDIA_URLはURLで参照する際に起点となるアドレスを指定します。
　私の場合、CouponモデルのデータをGETする際のアドレスが[ip(ドメイン）]:[ポート]/api/coupons/なので、イメージに参照する際は[ip(ドメイン）]:[ポート]/api/coupons/imagesとなるように/api/coupons/を起点として設定しました。

プロジェクト名のフォルダ/settings.py

MEDIA_URL = '/api/coupons/'

urls.py に設定を追加する

　URLで画像を参照するには urls.py に設定が必要です。
　プロジェクト名のディレクトリ配下のurls.pyを開き、下記import文とurlpatternsの宣言を追加します。画像のような静的ファイルを扱うための設定で、ほとんど定型文のように扱えるようです。

ami_coupon_api/urls.py

from django.conf import settings #画像参照のため追加
from django.contrib.staticfiles.urls import static #画像参照のため追加
from django.contrib.staticfiles.urls import staticfiles_urlpatterns #画像参照のため追加


urlpatterns += staticfiles_urlpatterns()
urlpatterns += static(settings.MEDIA_URL, document_root=settings.MEDIA_ROOT)

必要に応じてpsettings.pyに設定を追加する

　Django Rest Framework を使っていて serializer.py にレスポンスするモデルフィールドを設定している場合は、画像ファイルのURLをレスポンスするように設定を変更をします。

　下記のfields =の部分です。’__all__’の場合は全てのモデルフィールドの値をレスポンスする設定なので変更は不要です。

serializer.py

from rest_framework import serializers
from .models import Coupon

class CouponSerializer(serializers.ModelSerializer):
    class Meta:
        model = Coupon
        fields = '__all__'

マイグレートする

　モデルの定義を変更したので、DBに変更を反映するためにマイグレートをします。

　マイグレーションファイルを作成→マイグレートの流れになります。（pipenvを使っている場合は仮想環境のシェルに入っている状態で実行）

$ python manage.py makemigrations [アプリ名]
$ python manage.py migrate

動作確認

　以上で画像ファイルを扱うための設定は完了となりますので、djangoのadminページを使って直接画像を投入し、実際に表示されるか確認しました。

　djangoサーバを立ち上げdjangoのadminページにログインしてcouponモデルのページを開くと、画像をアップデートできるようになっています。クーポン用に作成した画像を直接アップデートします。

　次にcurlコマンドでGETリクエストに対するレスポンスの json を取得します。

$ curl -X GET http://127.0.0.1:8000/api/coupons/

　下記の通り、画像ファイルのURLが含まれているのでコピーしてブラウザでアクセスします。

[{"id":1,"code":"0001","benefit":"お会計から1,000円割引","explanation":"5,000円以上ご利用のお客様限定。他クーポンとの併用不可。","image":"http://127.0.0.1:8000/api/coupons/images/coupon-image-001_GKrT1ju.png" ,"store":"全店","start":"2019-10-01","deadline":"2019-12-31","status":true}

　ブラウザに設定したクーポンの画像が表示されたら成功です。

Anacondaにモジュールをインストール

※この記事はMacを対象にしています。
※例として、ortoolpyモジュールのインストールをとっています。他でも同じ。

Anacondaにパスを通す

自分もそうでしたが、Anacondaのパスが通っていない場合、
コマンドプロンプトで"conda"と打っても反応してくれません。

別記事にAnacondaへのパスの通し方があったので、こちらを参照にして下さい。

リンクはこちら：Anacondaのパスの設定

よし、condaを実行してモジュールのインストールだ！

しかしびっくり仰天

$ conda install ortoolpy

としても、

zsh: command not found: conda

となってしまう。

解決方法

$ source ~/.bash_profile

を実行することで、解決しました。

$ pip install ortoolpy

これにてインストール完了です。
※condaコマンドも実行できましたが、pipでないとortoolpyはインストールできないモジュールであることが発覚しました。

疑問点

最後pipでもインストールできるあたり謎です。
あと毎回

$ source ~/.bash_profile

とするのも面倒です。誰か良い方法を知っている方いたら教えて欲しいです、、、

若干詰まったのでメモ。

環境

• Python 3.7.4
• Django 3.0.3
• django-rest-framework 3.11.0

ModelSerializerについて

django-rest-framework(以降 DRF)では、ModelSerializerというオブジェクトを介して、Json - Model間の変換を行う。
APIで受け取った値を編集して登録したり、テーブルに存在しないレコードを返したり、データの加工を行ってJson化する場合、Serializerにメソッドを追加する場合が多い。

またDRFではトークン認証を行った場合、requestオブジェクトにuserインスタンスが格納される。
Serializerにuserインスタンスを渡すと、リレーション先の値を自動で入れられるので、利用シーンは多い。

model

前提となるVisitorモデルは以下。

import uuid
from django.db import models
from api.models import AppUser
from web.models import AdminUser


class Visitor(models.Model):
    class Meta:
        db_table = "visitor"
        verbose_name_plural = '訪問者'

    uuid = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False)
    parent_user = models.ForeignKey(AdminUser,
                                    on_delete=models.CASCADE,
                                    related_name='visitor')
    name = models.CharField(max_length=50)
    email = models.EmailField(max_length=255, null=True, blank=True)
    company = models.CharField(max_length=50, blank=True, null=True)
    memo = models.TextField(blank=True, null=True, max_length=300)
    visit_count = models.PositiveIntegerField(default=0)
    created_by = models.ForeignKey(AppUser,
                                   on_delete=models.SET_NULL,
                                   null=True, blank=True,
                                   related_name='visitor')
    created_at = models.DateTimeField(auto_now_add=True)
    updated_at = models.DateTimeField(auto_now=True)

    def __str__(self):
        return self.name

serializer

serializerはモデルと結び付いたfieldsを生成する。値を任意に変更するには、SerializerMethodField()でオーバーライドする。
SerializerMethodField()で定義したfieldはget_<field名>というメソッドを呼び出し、その返り値を格納する。

get_<field名>のメソッドは引数にinstanceを持つ。
そのため、API認証を行った際ここに渡されると思っていたのだが、ここはCreate時に作成されたインスタンスそのものが渡されるようだ。
(これを利用すれば、リレーション先のデータを自動作成も出来る。)

from rest_framework.serializers import ModelSerializer, SerializerMethodField

class VisitorCreateSerializer(ModelSerializer):
    created_by = SerializerMethodField()
    parent_user = SerializerMethodField()

    class Meta:
        model = Visitor
        read_only_fields = (
            'pk',
            'created_at',
            'created_by',
        )
        fields = (
            'pk',
            'parent_user',
            'created_at',
            'name',
            'email',
            'company',
            'memo',
            'updated_at',
        )

    def get_created_by(self, instance):
        """送信したユーザーがcreated_byに格納される"""
        return str(instance.pk)

    def get_parent_user(self, instance):
        """parent_userを自動的に格納する"""
        return str(instance.parent_user.pk)

ViewSet

Viewにおいて、Serializerをインスタンス化する部分があるので、ここでinstanceにuserを渡せばOK.
今回はcreateだけserializerを分けたかったのでcreateメソッドをオーバーライドして、Serializerを指定している。

# DRF
from rest_framework import status, viewsets
from rest_framework.permissions import IsAuthenticated
from .serializers import VisitorSerializer, VisitorCreateSerializer
from .models import Visitor
from utils.auth.authentication import APIAuthentication
from rest_framework.response import Response


class VisitorViewSet(viewsets.ModelViewSet):
    serializer_class = VisitorSerializer # create以外に使用するserializer
    queryset = Visitor.objects.all()

    authentication_classes = (APIAuthentication,)
    permission_classes = (IsAuthenticated,)

    def get_queryset(self):
        # API認証した場合,userインスタンスはrequest.userに格納される。
        admin = self.request.user.parent_user
        qs = Visitor.objects.filter(parent_user=admin)
        return qs

    # createメソッドをオーバーライド
    def create(self, request, *args, **kwargs):
        # Serializerにrequest.userを渡すと、get_<field名>メソッドにinstanceが渡る
        serializer = VisitorCreateSerializer(instance=request.user, data=request.data)
        # ここから下はそのまま
        serializer.is_valid(raise_exception=True)
        self.perform_create(serializer)
        headers = self.get_success_headers(serializer.data)
        return Response(serializer.data, status=status.HTTP_201_CREATED, headers=headers)

1 はじめに

　画像認識ができるプログラムをKerasにより作りました。
　実行環境は下記になります。

・python 3.5.6
・Keras 2.2.0
・tensorflow 1.5.
・matplotlib 2.2.2
・numpy 1.15.2
・opencv 3.4.2

参考にさせて頂いたURLは下記一覧となります。
https://qiita.com/ayumiya/items/e1e87df54c41519be6b4

2 画像を収集する

　今回flickr APIを利用して大量の画像を収集しました。こちらに関しては、下記URLを参考にflickrの登録・API, secret KEYの発行などを行うことで画像を収集させました。
http://ykubot.com/2017/11/05/flickr-api/

　収集に関してもpython上で下記プログラムを実行することで自動で画像を集めることができました。300枚収集するのに必要な時間は凡そ今回のケースだと5分程度でした。

下記がプログラム全文になります。

collection.py

import os
import time
import traceback

import flickrapi
from urllib.request import urlretrieve

import sys
from retry import retry
flickr_api_key = "xxxx" #ここに発行されたAPI KEYを入力
secret_key = "xxxx"　#ここに発行されたSECRET KEYを入力

keyword = sys.argv[1]


@retry()
def get_photos(url, filepath):
    urlretrieve(url, filepath)
    time.sleep(1)


if __name__ == '__main__':

    flicker = flickrapi.FlickrAPI(flickr_api_key, secret_key, format='parsed-json')
    response = flicker.photos.search(
        text='child',#検索したいワードを入力
        per_page=300,#何枚収集したいか指定
        media='photos',
        sort='relevance',
        safe_search=1,
        extras='url_q,license'
    )
    photos = response['photos']

    try:
        if not os.path.exists('./image-data/' + 'child'):
            os.mkdir('./image-data/' +'child')

        for photo in photos['photo']:
            url_q = photo['url_q']
            filepath = './image-data/' + 'child' + '/' + photo['id'] + '.jpg'
            get_photos(url_q, filepath)

    except Exception as e:
        traceback.print_exc()

　今回、「子供:child」と「老人：elder」を認識すべくそれら画像を300枚ずつ収集しました。
　先にも述べますが検索ワードelderは、老人だけでなく植物、若い女性等の画像も収集されたため、それにより認識率が高くなかったものと推定しています。

3 画像認識プログラムの作成

　以下記事にて、画像認識プログラムについて説明します。全体像から私がつまづいてしまったところや、注意すべきポイントに絞って抜粋し、そののちにプログラム全文を記載します。

インポートしたライブラリ、モジュール等

import os
import cv2
import numpy as np
import matplotlib.pyplot as plt
from keras.utils.np_utils import to_categorical
from keras.layers import Dense, Dropout, Flatten, Input
from keras.applications.vgg16 import VGG16
from keras.models import Model, Sequential
from keras import optimizers

モデル関数を作るうえでの注意

model = Model(input=vgg16.input, output=top_model(vgg16.output))

元々、仕様環境はtensorflowをインストールしており、その中のライブラリであるkerasとして、引用していました。
例：from tensorflow.keras.utils import to_categorical

このまま進めるていると、このModel関数を定義するうえで、
('Keyword argument not understood:', 'input')
となるエラーが発生しました。

これはどうもkerasのversionが古いと起こるとのことで、
keras自体のアップデートなども行ったのですが、エラーは出続けました。
従い、keras自体を読み込む書き方をしてみたら、解決しました。

本エラー概要及び、keras自体のアップデート
https://sugiyamayoshiaki.jp/%E3%80%90%E3%82%A8%E3%83%A9%E3%83%BC%E3%80%91typeerror-keyword-argument-not-understood-data_format/

モデルの選定

# モデルにvggを使います
input_tensor = Input(shape=(50, 50, 3))
vgg16 = VGG16(include_top=False, weights='imagenet', input_tensor=input_tensor)

# vggのoutputを受け取り、2クラス分類する層を定義します
top_model = Sequential()
top_model.add(Flatten(input_shape=vgg16.output_shape[1:]))
top_model.add(Dense(256, activation='relu'))
top_model.add(Dropout(0.5))
top_model.add(Dense(2, activation='softmax'))

今回選定したモデルはVGG16と呼ばれるモデルです。
http://cedro3.com/ai/mini-vgg-net/

VGGチームが作った畳み込み13層＋全結合層３層＝16層のニューラルネットワークとのことです。
詳細は下記に記載されていますが、なるほど理解できていません。
https://qiita.com/MuAuan/items/86a56637a1ebf455e180

子供か老人かを判別する関数を作る

def child(img):
    img = cv2.resize(img, (50, 50))
    pred = np.argmax(model.predict(np.array([img])))
    if pred == 0:
        return 'これは子供です'
    else:
        return 'これは老人です'

全文プログラム

train.py

import os
import cv2
import numpy as np
import matplotlib.pyplot as plt
from keras.utils.np_utils import to_categorical
from keras.layers import Dense, Dropout, Flatten, Input
from keras.applications.vgg16 import VGG16
from keras.models import Model, Sequential
from keras import optimizers


number = 100
path_child = os.listdir('image-data/child')
path_elder = os.listdir('image-data/elder')
path_test = os.listdir('image-data/test')
path_child_test = os.listdir('image-data/child_test')


img_child = []
img_elder = []
img_test = []
img_child_test= []


input_tensor = Input(shape=(50, 50, 3))
for i in range(len(path_child)):
    img = cv2.imread('image-data/child/' + path_child[i])
    if(type(img) != type(None)):
        img = cv2.resize(img ,(50,50))
        img_child.append(img)
for i in range(len(path_elder)):
    img = cv2.imread('image-data/elder/' + path_elder[i])
    if(type(img) != type(None)):
        img = cv2.resize(img, (50,50))
        img_elder.append(img)

for i in range(len(path_test)):
    img = cv2.imread('image-data/test/' + path_test[i])
    if(type(img) != type(None)):
        img = cv2.resize(img, (50,50))
        img_test.append(img)

for i in range(len(path_child_test)):
    img = cv2.imread('image-data/child_test/' + path_child_test[i])
    if(type(img) != type(None)):
        img = cv2.resize(img, (50,50))
        img_child_test.append(img)


X = np.array(img_child + img_elder)
y =  np.array([0]*len(img_child) + [1]*len(img_elder))

X_t = np.array(img_child_test + img_test)
y_t = np.array([0]*len(img_test) +[1]*len(img_child_test))

rand_index = np.random.permutation(np.arange(len(X)))
rand_index_2 = np.random.permutation(np.arange(len(X_t)))

X = X[rand_index]
y = y[rand_index]

X_t = X_t[rand_index_2]
y_t = y_t[rand_index_2]

# データの分割
X_train = X[:int(len(X)*0.8)]
y_train = y[:int(len(y)*0.8)]
X_test = X_t[int(len(X_t)*0.8):]  
y_test = y_t[int(len(y_t)*0.8):] 

# 正解ラベルをone-hotの形にします
y_train = to_categorical(y_train)
y_test = to_categorical(y_test)

# モデルにvggを使います
input_tensor = Input(shape=(50, 50, 3))
vgg16 = VGG16(include_top=False, weights='imagenet', input_tensor=input_tensor)

# vggのoutputを受け取り、2クラス分類する層を定義します
top_model = Sequential()
top_model.add(Flatten(input_shape=vgg16.output_shape[1:]))
top_model.add(Dense(256, activation='relu'))
top_model.add(Dropout(0.5))
top_model.add(Dense(2, activation='softmax'))

model = Model(input=vgg16.input, output=top_model(vgg16.output))

# vggの層の重みを変更不能にします
for layer in model.layers[:15]:
    layer.trainable = False

# コンパイルします
model.compile(loss='categorical_crossentropy',
              optimizer=optimizers.SGD(lr=1e-4, momentum=0.9),
              metrics=['accuracy'])

# 学習を行います
model.fit(X_train, y_train, batch_size=100, epochs=20, validation_data=(X_test, y_test))

#関数定義
def child(img):
    img = cv2.resize(img, (50, 50))
    pred = np.argmax(model.predict(np.array([img])))
    if pred == 0:
        return 'これは子供です'
    else:
        return 'これは老人です'


scores = model.evaluate(X_test, y_test, verbose=1)
print('Test loss 間違い具合:', scores[0])
print('Test accuracy 正解率:', scores[1])

# cream関数に写真を渡して性別を予測します
for i in range(10):
    img = cv2.imread('child_test/' + path_child_test[i])
    if(type(img) != type(None)):
        b,g,r = cv2.split(img)
        img = cv2.merge([r,g,b])
        plt.imshow(img)
        plt.show()
        print(child(img))    

以上になります。
Test loss 間違い具合: 2.303262777328491
Test accuracy 正解率: 0.5199999809265137

となり、2択なのに正解率52%とまったく当たっていませんので、改善が必要です。

LINE以外のアプリの通知を切っていますか？ 僕は切っています。
AtCoderコンテストの予定が追加されたらLINEで通知してほしいなと思い、Mr JJ と共に LINE BOTを作成しました。

作成したもの

毎時00分に予定されたコンテストの検索を行い、新たなコンテストが追加されている場合BOT君が通知してくれます。

一応メッセージを送ると返してくれる機能もあります。(おまけ)

作成したLINE BOTは以下のボタンから友達追加できます。

バックエンドの構成

バックエンドの処理はすべてAWS(Amazon Web Services)上で行っています。
AWS上の動作は以下の2つに分割できます。

• コンテスト通知セクション
  • AtCoderコンテストの検索と新たなコンテストの通知を行う
• LINE応答セクション
  • BOTに対しメッセージが送信された際に応答を行う

各セクションについて実装した機能を説明します。

コンテスト通知セクション

コンテスト通知セクションの流れ

1. Lambda関数: AtCoder_contest_Search がAtCoderホームページのスクレイピングを行い、予定されたコンテストリストを取得する
2. AtCoder_contest_Search が S3 から通知済みコンテスト一覧 JSON ファイルを取得する。
3. 差集合を計算し、未通知コンテストを抽出する。未通知コンテストが空なら終了。
4. 未通知コンテストデータをLambda関数: LINE_contest_notify に送る。
5. LINE_contest_notify が送られたデータを line-bot-sdk を用いてブロードキャスト形式で通知する。
6. [1] で得られたコンテストデータを S3 にアップロード

スクレイピング機能

Python3

def get_contests():
    # URLからデータを取ってくる
    url = "https://atcoder.jp/home?lang=ja"
    html_data = requests.get(url)

    # htmlパース
    soup = BeautifulSoup(html_data.text, "html.parser")

    tags = ["upcoming", "recent"]
    ret_dic = {}
    for tag in tags:
        div = soup.find("div", id="contest-table-"+tag)
        table = div.find("table")
        times  = table.find_all("time")
        titles = table.find_all("a")

        ret_list = []
        for i in range(len(times)):
            dic = {}
            dic["start"] = times[i].text
            dic["title"] = titles[i*2+1].text
            dic["url"] = "https://atcoder.jp" + titles[i*2+1].get("href")
            ret_list.append(dic)
        ret_dic[tag] = ret_list
    return ret_dic

	データ内容
引数	null
返り値	各コンテストに対してデータ(開始時刻・タイトル・URL)を要素に持つ辞書のリスト

本BOTでは ret_dic["upcoming"] のみ使用。

S3からのファイル取得

Python3

def get_data_s3():
    # S3上にある通知済みコンテストjsonファイル"contests.json"の読み込み

    client = boto3.client("s3")
    try:
        response = client.get_object(Bucket = "bucket_name", Key = "json_file_name")
    except Exception as e:
        print(e)
    return json.loads(response["Body"].read().decode())

	データ内容
引数	null
返り値	S3上のJSONファイルを読み込んだ辞書データ

get_object メソッドの返り値は辞書であり、キー"Body"にファイルの内容が含まれる。

S3へのファイルアップロード

Python3

def put_data_s3(save_dic):
    # 新たな通知済みコンテストjsonファイルのS3アップローダー

    text = json.dumps(save_dic, ensure_ascii = False)
    s3 = boto3.resource("s3")
    s3.Object("bucket_name", "json_file_name").put(Body = text)

	データ内容
引数	保存対象の辞書データ
返り値	null

辞書をJSON形式のテキストにする際、ensure_ascii の値をFalseにしておくと、
JSONファイル内の日本語が読みやすくなり、ファイルサイズも小さくできる。

ブロードキャスト形式のメッセージ送信

Python3

def DatetimeToString(date):
    if not isinstance(date, datetime.datetime):
        date = datetime.datetime.strptime(date, "%Y-%m-%d %H:%M:%S%z")
    week_list = "月火水木金土日"
    return f"{date.strftime('%m/%d')}({week_list[date.weekday()]}){date.strftime('%H:%M')}"

def MakeMessageText(event):
    text = DatetimeToString(event["start"]) + "\n" + event["title"] + "が開催されます！\n" + event["url"]
    return text

def main(event, context):
    # ***ここでLINE BOTのチャンネルアクセストークン取得しておく***

    line_bot_api = LineBotApi("channel_access_token")
    message = MakeMessageText(event)
    line_bot_api.broadcast(TextSendMessage(text=message))

    # *** main関数を終える処理 ***

DatetimeToString 関数でコンテスト開始時間のメッセージフォーマットを作成し、
MakeMessageText 関数で通知するメッセージのフォーマットを作成した。

line-bot-sdk の LineBotApi クラス broadcast メソッドを呼び出すだけでBOTの友達全員に通知を行うことができる。

LINE応答セクション

本システムの構成図

LINE応答セクションの流れ

1. LINE BOT がLINEメッセージを受け取るとWebhookを利用してAPI Gatewayへメッセージ内容をPOSTする。
2. API GatewayへのPOSTがトリガーとなりLambda関数: LINE_reply_message が実行される。
3. LINE_reply_message が line-bot-sdk を用いてLINEのリプライメッセージを送信する。

リプライメッセージ送信

Python3

def main(event, context):
    # ***ここでLINE BOTのチャンネルアクセストークン・秘密鍵を取得しておく***

    handler = WebhookHandler("channel_secret")
    line_bot_api = LineBotApi("channel_access_token")
    body = event["body"]

    # handlerにリプライ関数を追加
    @handler.add(MessageEvent)
    def SendReply(line_event):
        message = MakeReplyMessage(line_event.message.text)
        line_bot_api.reply_message(line_event.reply_token, TextSendMessage(text=message))

    try:
        handler.handle(body, event["headers"]["X-Line-Signature"])
    except # エラー処理

    # *** main関数を終える処理 ***

MakeReplyMessage 関数はBOTが受け取ったメッセージ内容に応じたリプライメッセージを文字列で返す。
ここの部分に関しては特に line-bot-sdkのドキュメント を読むことをおすすめします。

終わりに

作成した主な機能について紹介を行いました。
簡単にBOTの操作を行えるline-bot-sdk がとても便利です。
AWSに関する知識が少なく、かなり荒い設計になりましたが、気が向いたら今後機能追加＆改善します。(たぶん)

LINE以外のアプリの通知を切っていますか？ 僕は切っています。
AtCoderコンテストの予定が追加されたらLINEで通知してほしいなと思い、Mr JJ と共に LINE BOTを作成しました。

作成したもの

毎時00分に予定されたコンテストの検索を行い、新たなコンテストが追加されている場合BOT君が通知してくれます。

一応メッセージを送ると返してくれる機能もあります。(おまけ)

作成したLINE BOTは以下のボタンから友達追加できます。

バックエンドの構成

バックエンドの処理はすべてAWS(Amazon Web Services)上で行っています。
AWS上の動作は以下の2つに分割できます。

• コンテスト通知セクション
  • AtCoderコンテストの検索と新たなコンテストの通知を行う
• LINE応答セクション
  • BOTに対しメッセージが送信された際に応答を行う

各セクションについて実装した機能を説明します。

コンテスト通知セクション

コンテスト通知セクションの流れ

1. Lambda関数: AtCoder_contest_Search がAtCoderホームページのスクレイピングを行い、予定されたコンテスト一覧を取得する
2. AtCoder_contest_Search が S3 から通知済みコンテスト一覧 JSON ファイルを取得する。
3. 差集合を計算し、未通知コンテストを抽出する。未通知コンテストが空なら終了。
4. 未通知コンテストデータをLambda関数: LINE_contest_notify に送る。
5. LINE_contest_notify が送られたデータを line-bot-sdk を用いてブロードキャスト形式で通知する。
6. [1] で得られたコンテストデータを S3 にアップロード

スクレイピング機能

Python3

def get_contests():
    # URLからデータを取ってくる
    url = "https://atcoder.jp/home?lang=ja"
    html_data = requests.get(url)

    # htmlパース
    soup = BeautifulSoup(html_data.text, "html.parser")

    tags = ["upcoming", "recent"]
    ret_dic = {}
    for tag in tags:
        div = soup.find("div", id="contest-table-"+tag)
        table = div.find("table")
        times  = table.find_all("time")
        titles = table.find_all("a")

        ret_list = []
        for i in range(len(times)):
            dic = {}
            dic["start"] = times[i].text
            dic["title"] = titles[i*2+1].text
            dic["url"] = "https://atcoder.jp" + titles[i*2+1].get("href")
            ret_list.append(dic)
        ret_dic[tag] = ret_list
    return ret_dic

	データ内容
引数	null
返り値	各コンテストに対してデータ(開始時刻・タイトル・URL)を要素に持つ辞書のリスト

本BOTでは ret_dic["upcoming"] のみ使用。

S3からのファイル取得

Python3

def get_data_s3():
    # S3上にある通知済みコンテストjsonファイル"contests.json"の読み込み

    client = boto3.client("s3")
    try:
        response = client.get_object(Bucket = "bucket_name", Key = "json_file_name")
    except Exception as e:
        print(e)
    return json.loads(response["Body"].read().decode())

	データ内容
引数	null
返り値	S3上のJSONファイルを読み込んだ辞書データ

get_object メソッドの返り値は辞書であり、キー"Body"にファイルの内容が含まれる。

S3へのファイルアップロード

Python3

def put_data_s3(save_dic):
    # 新たな通知済みコンテストjsonファイルのS3アップローダー

    text = json.dumps(save_dic, ensure_ascii = False)
    s3 = boto3.resource("s3")
    s3.Object("bucket_name", "json_file_name").put(Body = text)

	データ内容
引数	保存対象の辞書データ
返り値	null

辞書をJSON形式のテキストにする際、ensure_ascii の値をFalseにしておくと、
JSONファイル内の日本語が読みやすくなり、ファイルサイズも小さくできる。

ブロードキャスト形式のメッセージ送信

Python3

def DatetimeToString(date):
    if not isinstance(date, datetime.datetime):
        date = datetime.datetime.strptime(date, "%Y-%m-%d %H:%M:%S%z")
    week_list = "月火水木金土日"
    return f"{date.strftime('%m/%d')}({week_list[date.weekday()]}){date.strftime('%H:%M')}"

def MakeMessageText(event):
    text = DatetimeToString(event["start"]) + "\n" + event["title"] + "が開催されます！\n" + event["url"]
    return text

def main(event, context):    
    line_bot_api = LineBotApi("channel_access_token")
    message = MakeMessageText(event)
    line_bot_api.broadcast(TextSendMessage(text=message))

    # *** main関数を終える処理 ***

DatetimeToString 関数でコンテスト開始時間のメッセージフォーマットを作成し、
MakeMessageText 関数で通知するメッセージのフォーマットを作成した。

line-bot-sdk の LineBotApi クラス broadcast メソッドを呼び出すだけでBOTの友達全員に通知を行うことができる。

LINE応答セクション

本システムの構成図

LINE応答セクションの流れ

1. LINE BOT がLINEメッセージを受け取るとWebhookを利用してAPI Gatewayへメッセージ内容をPOSTする。
2. API GatewayへのPOSTがトリガーとなりLambda関数: LINE_reply_message が実行される。
3. LINE_reply_message が line-bot-sdk を用いてLINEのリプライメッセージを送信する。

リプライメッセージ送信

Python3

def main(event, context):
    handler = WebhookHandler("channel_secret")
    line_bot_api = LineBotApi("channel_access_token")
    body = event["body"]

    # handlerにリプライ関数を追加
    @handler.add(MessageEvent)
    def SendReply(line_event):
        message = MakeReplyMessage(line_event.message.text)
        line_bot_api.reply_message(line_event.reply_token, TextSendMessage(text=message))

    try:
        handler.handle(body, event["headers"]["X-Line-Signature"])
    except # エラー処理

    # *** main関数を終える処理 ***

MakeReplyMessage 関数はBOTが受け取ったメッセージ内容に応じたリプライメッセージを文字列で返す。
ここの部分に関しては特に line-bot-sdkのドキュメント を読むことをおすすめします。

終わりに

作成した主な機能について紹介を行いました。
簡単にBOTの操作を行えるline-bot-sdk がとても便利です。
AWSに関する知識が少なく、かなり荒い設計になりましたが、気が向いたら今後機能追加＆改善します。(たぶん)