3月 « 2022 « shell-mag

香川大学SLPからお届け！（Vol.77掲載）

投稿日：2022.03.25　｜　カテゴリー：　コード

著者：石上椋一

毎年年末に開催されている競馬の「有馬記念」というレースをご存知でしょうか。今回は、その有馬記念の順位を予測するAIを開発した話を紹介します。2021年12月には、開発したAIを使って第66回有馬記念の順位を予測してみました。結果がどうだったのかについては、記事の最後に書いています。

シェルスクリプトマガジン Vol.77は以下のリンク先でご購入できます。

図2　RankNetを実装するPythonコード

from tensorflow.keras import layers, Model, Input
from tensorflow.nn import leaky_relu

class RankNet(Model):
  def __init__(self):
    super().__init__()
    self.dense = [layers.Dense(16, activation=leaky_relu),
                  layers.Dense(8, activation=leaky_relu)]
    self.o = layers.Dense(1, activation='linear')
    self.oi_minus_oj = layers.Subtract()
  def call(self, inputs):
    xi, xj = inputs
    densei = self.dense[0](xi)
    densej = self.dense[0](xj)
    for dense in self.dense[1:]:
      densei = dense(densei)
      densej = dense(densej)
    oi = self.o(densei)
    oj = self.o(densej)
    oij = self.oi_minus_oj([oi, oj])
    output = layers.Activation('sigmoid')(oij)
    return output

from tensorflow.keras import layers, Model, Input

from tensorflow.nn import leaky_relu

class RankNet(Model):

def __init__(self):

super().__init__()

self.dense = [layers.Dense(16, activation=leaky_relu),

layers.Dense(8, activation=leaky_relu)]

self.o = layers.Dense(1, activation='linear')

self.oi_minus_oj = layers.Subtract()

def call(self, inputs):

xi, xj = inputs

densei = self.dense[0](xi)

densej = self.dense[0](xj)

for dense in self.dense[1:]:

densei = dense(densei)

densej = dense(densej)

oi = self.o(densei)

oj = self.o(densej)

oij = self.oi_minus_oj([oi, oj])

output = layers.Activation('sigmoid')(oij)

return output

図3　データの整理とラベル付けをするPythonコード

import pandas as pd
import numpy as np
from itertools import combinations

years = [2020,2019,2018,2017,2016,2015,2014,2013,2012,2011]
# ここで任意のCSVファイルを指定する
# 今回はCSVにデータを残しているため、CSVから読み込んでいる
df = pd.read_csv(CSVFile)
df["タイム指数2-3"] = df["タイム指数2"] - df["タイム指数3"]
index_num = 0
xi = xj = pij = pair_ids = pair_query_id = []
for year in years:
  one_year_Data = df[df['年数'] == year]
  index_list = [i for i in range(len(one_year_Data))]
  random.shuffle(index_list)
  for pair_id in combinations(index_list, 2):
    pair_query_id.append(year)
    pair_ids.append(pair_id)
    i = pair_id[0]
    j = pair_id[1]
    xi.append([one_year_Data.at[i+index_num,"タイム指数2"],
               one_year_Data.at[i+index_num,"タイム指数2-3"],
               one_year_Data.at[i+index_num,"上り"]])
    xj.append([one_year_Data.at[j+index_num,"タイム指数2"],
               one_year_Data.at[j+index_num,"タイム指数2-3"],
               one_year_Data.at[j+index_num,"上り"]])
    if one_year_Data.at[i+index_num,"順位"] == one_year_Data.at[j+index_num,"順位"] :
      pij_com = 0.5
    elif one_year_Data.at[i+index_num,"順位"] > one_year_Data.at[j+index_num,"順位"] :
      pij_com = 0
    else:
      pij_com = 1
    pij.append(pij_com)
  index_num += len(one_year_Data)
  index_list.clear()
xi = np.array(xi)
xj = np.array(xj)
pij = np.array(pij)
pair_query_id = np.array(pair_query_id)

import pandas as pd

import numpy as np

from itertools import combinations

years = [2020,2019,2018,2017,2016,2015,2014,2013,2012,2011]

# ここで任意のCSVファイルを指定する

# 今回はCSVにデータを残しているため、CSVから読み込んでいる

df = pd.read_csv(CSVFile)

df["タイム指数2-3"] = df["タイム指数2"] - df["タイム指数3"]

index_num = 0

xi = xj = pij = pair_ids = pair_query_id = []

for year in years:

one_year_Data = df[df['年数'] == year]

index_list = [i for i in range(len(one_year_Data))]

random.shuffle(index_list)

for pair_id in combinations(index_list, 2):

pair_query_id.append(year)

pair_ids.append(pair_id)

i = pair_id[0]

j = pair_id[1]

xi.append([one_year_Data.at[i+index_num,"タイム指数2"],

one_year_Data.at[i+index_num,"タイム指数2-3"],

one_year_Data.at[i+index_num,"上り"]])

xj.append([one_year_Data.at[j+index_num,"タイム指数2"],

one_year_Data.at[j+index_num,"タイム指数2-3"],

one_year_Data.at[j+index_num,"上り"]])

if one_year_Data.at[i+index_num,"順位"] == one_year_Data.at[j+index_num,"順位"] :

pij_com = 0.5

elif one_year_Data.at[i+index_num,"順位"] > one_year_Data.at[j+index_num,"順位"] :

pij_com = 0

else:

pij_com = 1

pij.append(pij_com)

index_num += len(one_year_Data)

index_list.clear()

xi = np.array(xi)

xj = np.array(xj)

pij = np.array(pij)

pair_query_id = np.array(pair_query_id)

図4　学習用データと評価用データを仕分けるPythonコード

from sklearn.model_selection import train_test_split

xi_train, xi_test, xj_train, xj_test, pij_train, pij_test, \
pair_id_train, pair_id_test = train_test_split(
  xi, xj, pij, pair_ids, test_size=0.2, stratify=pair_query_id)

from sklearn.model_selection import train_test_split

xi_train, xi_test, xj_train, xj_test, pij_train, pij_test, \

pair_id_train, pair_id_test = train_test_split(

xi, xj, pij, pair_ids, test_size=0.2, stratify=pair_query_id)

図5　コンパイルと学習を実施するPythonコード

ranknet = RankNet()
# コンパイル
ranknet.compile(optimizer='sgd', loss='binary_crossentropy',metrics=['accuracy'])
# 学習
ranknet.fit([xi_train, xj_train], pij_train,
            epochs=85, batch_size=4,
            validation_data=([xi_test, xj_test], pij_test))

ranknet = RankNet()

# コンパイル

ranknet.compile(optimizer='sgd', loss='binary_crossentropy',metrics=['accuracy'])

# 学習

ranknet.fit([xi_train, xj_train], pij_train,

epochs=85, batch_size=4,

validation_data=([xi_test, xj_test], pij_test))

図8　ResNet50を使った学習モデルを実装するPythonコード

from tensorflow.keras.applications.resnet50 import ResNet50
from keras.layers import Dense, Dropout, Input, Flatten

# ResNetの準備
input_tensor = Input(shape=(64, 64, 3))
resnet50 = ResNet50(include_top=False, weights='imagenet',
                    input_tensor=input_tensor)
# FC層の準備
fc_model = Sequential()
fc_model.add(Flatten(input_shape=resnet50.output_shape[1:]))
fc_model.add(Dense(512, activation='relu'))
fc_model.add(Dropout(0.3))
fc_model.add(Dense(2, activation='sigmoid'))
# モデルの準備
resnet50_model = Model(resnet50.input, fc_model(resnet50.output))
# ResNet50の一部の重みを固定
for layer in resnet50_model.layers[:100]:
  layer.trainable = False

from tensorflow.keras.applications.resnet50 import ResNet50

from keras.layers import Dense, Dropout, Input, Flatten

# ResNetの準備

input_tensor = Input(shape=(64, 64, 3))

resnet50 = ResNet50(include_top=False, weights='imagenet',

input_tensor=input_tensor)

# FC層の準備

fc_model = Sequential()

fc_model.add(Flatten(input_shape=resnet50.output_shape[1:]))

fc_model.add(Dense(512, activation='relu'))

fc_model.add(Dropout(0.3))

fc_model.add(Dense(2, activation='sigmoid'))

# モデルの準備

resnet50_model = Model(resnet50.input, fc_model(resnet50.output))

# ResNet50の一部の重みを固定

for layer in resnet50_model.layers[:100]:

layer.trainable = False

図9　コンパイルと学習を実施するPythonコード

# コンパイル
resnet50_model.compile(optimizer='sgd',
                       loss='categorical_crossentropy',
                       metrics=['accuracy'])
# 学習
history = resnet50_model.fit(train_generator, 
                             batch_size=4, 
                             epochs=50, verbose=1,
                             validation_data=val_generator)

# コンパイル

resnet50_model.compile(optimizer='sgd',

loss='categorical_crossentropy',

metrics=['accuracy'])

# 学習

history = resnet50_model.fit(train_generator,

batch_size=4,

epochs=50, verbose=1,

validation_data=val_generator)

本連載では「Pythonを昔から使っているものの、それほど使いこなしてはいない」という筆者が、いろいろな日常業務をPythonで処理することで、立派な「蛇使い」に育つことを目指します。その過程を温
かく見守ってください。皆さんと共に勉強していきましょう。第7回は、Pythonで実装されたWebアプリケーションフレームワーク「Flask」の使い方と、Pythonのユニークな文法である「デコレータ」について紹介します。

シェルスクリプトマガジン Vol.77は以下のリンク先でご購入できます。

図5　ルーティングのサンプルWebアプリケーションのコード

from flask import Flask
app = Flask(__name__)
@app.route('/hello')
def hello_flask():
  return '<h1>Hello, Flask!</h1>'
@app.route('/goodbye')
def goodbye_flask():
  return '<h1>Good bye, Flask!</h1>'

from flask import Flask

app = Flask(__name__)

@app.route('/hello')

def hello_flask():

return '<h1>Hello, Flask!</h1>'

@app.route('/goodbye')

def goodbye_flask():

return '<h1>Good bye, Flask!</h1>'

図7　簡単なメッセージを表示するプログラム

#!/usr/bin/env python

def hello():
  print('Hello, World')
def goodbye():
  print('Goodbye, World')
def main():
  hello()
  goodbye()
if __name__ == '__main__':
  main()

#!/usr/bin/env python

def hello():

print('Hello, World')

def goodbye():

print('Goodbye, World')

def main():

hello()

goodbye()

if __name__ == '__main__':

main()

図9　デコレータを追加したプログラム

#!/usr/bin/env python

def deco(func):
  def decorator():
    print('-- start --')
    func()
    print('-- end --')
  return decorator
@deco
def hello():
  print('Hello, World')
@deco
def goodbye():
  print('Goodbye, World')
def main():
  hello()
  goodbye()
if __name__ == '__main__':
  main()

#!/usr/bin/env python

def deco(func):

def decorator():

print('-- start --')

func()

print('-- end --')

return decorator

@deco

def hello():

print('Hello, World')

@deco

def goodbye():

print('Goodbye, World')

def main():

hello()

goodbye()

if __name__ == '__main__':

main()

図11　Flaskのルーティング設定処理を模したプログラム

#!/usr/bin/env python
　
class Flask():
  def __init__(self):
    self.url_map = {}
    print(f'url_map on init: {self.url_map}')
  def add_url_rule(self, rule, func):
    self.url_map[rule] = func
  def route(self, rule):
    def decorator(f):
      self.add_url_rule(rule, f)
      return f
    return decorator
  def dispatch_request(self, rule):
    return self.url_map[rule]()

app = Flask()
print(f'url_map after create: {app.url_map}')
@app.route('/index')
def index():
  print("hello world")
print(f'url_map after function def: {app.url_map}')
app.dispatch_request('/index')

#!/usr/bin/env python

class Flask():

def __init__(self):

self.url_map = {}

print(f'url_map on init: {self.url_map}')

def add_url_rule(self, rule, func):

self.url_map[rule] = func

def route(self, rule):

def decorator(f):

self.add_url_rule(rule, f)

return f

return decorator

def dispatch_request(self, rule):

return self.url_map[rule]()

app = Flask()

print(f'url_map after create: {app.url_map}')

@app.route('/index')

def index():

print("hello world")

print(f'url_map after function def: {app.url_map}')

app.dispatch_request('/index')

図13　図9のプログラムを修正した「greeting_deco2.py」

#!/usr/bin/env python
　
def deco(rule):
  def decorator(func):
    print(f'rule = \'{rule}\', func = {func}')
    return func
  return decorator
@deco('/hello')
def hello():
  print('Hello, World')
@deco('/goodbye')
def goodbye():
  print('Goodbye, World')

#!/usr/bin/env python

def deco(rule):

def decorator(func):

print(f'rule = \'{rule}\', func = {func}')

return func

return decorator

@deco('/hello')

def hello():

print('Hello, World')

@deco('/goodbye')

def goodbye():

print('Goodbye, World')

図15　デコレータを使わないように修正した「greeting_deco3.py」

#!/usr/bin/env python

def deco(rule):
  def decorator(func):
    print(f'rule = \'{rule}\', func = {func}')
    return func
  return decorator
def hello():
  print('Hello, World')
hello = deco('/hello')(hello)
def goodbye():
  print('Goodbye, World')
goodbye = deco('/goodbye')(goodbye)

#!/usr/bin/env python

def deco(rule):

def decorator(func):

print(f'rule = \'{rule}\', func = {func}')

return func

return decorator

def hello():

print('Hello, World')

hello = deco('/hello')(hello)

def goodbye():

print('Goodbye, World')

goodbye = deco('/goodbye')(goodbye)

図16　関数の挙動を変えるデコレータの使用例

#!/usr/bin/env python

def plus_n(n):
  def decorator(func):
    def func_modified(*args, **kwargs):
      return (func(*args, **kwargs) + n)
    return func_modified
  return decorator
@plus_n(3)
def square(x):
  return x*x
def main():
  print(square(2))
if __name__ == '__main__':
  main()

#!/usr/bin/env python

def plus_n(n):

def decorator(func):

def func_modified(*args, **kwargs):

return (func(*args, **kwargs) + n)

return func_modified

return decorator

@plus_n(3)

def square(x):

return x*x

def main():

print(square(2))

if __name__ == '__main__':

main()

図17　関数に複数のデコレータを適用したプログラム「greeting_deco4.py」

#!/usr/bin/env python

def DECO(func):
  def decorator():
    print('-- START --')
    func()
    print('-- END --')
  return decorator
def deco(func):
  def decorator():
    print('-- start --')
    func()
    print('-- end --')
  return decorator
@DECO
@deco
def hello():
  print('Hello, World')
def main():
  hello()
if __name__ == '__main__':
  main()

#!/usr/bin/env python

def DECO(func):

def decorator():

print('-- START --')

func()

print('-- END --')

return decorator

def deco(func):

def decorator():

print('-- start --')

func()

print('-- end --')

return decorator

@DECO

@deco

def hello():

print('Hello, World')

def main():

hello()

if __name__ == '__main__':

main()

機械学習ことはじめ（Vol.77掲載）

投稿日：2022.03.25　｜　カテゴリー：　コード

筆者：川嶋宏彰

本連載では、機械学習の基礎となるさまざまな手法の仕組みや、それらの手法のPythonでの利用方法を解説していきます。最終回となる今回は、ニューラルネットの仕組みと、基本的なモデルについて解説します。

シェルスクリプトマガジン Vol.77は以下のリンク先でご購入できます。

図3　単純パーセプトロンを学習するための関数を定義するPythonコード

import numpy as np

def h_step(x):
  """ ステップ関数: 0より大きければ1、それ以外は0 """
  return int(x > 0)
def train_perceptron(X, y, lr=0.5, max_it=20, random_state=None):
  """ パーセプトロンの学習アルゴリズム
      lr: 学習率 (learning rate)
      max_it: 最大反復回数 (maximum number of iterations)
  """
  N, d = X.shape  # データ数x次元
  X1 = np.c_[np.ones(N), X]  # 1だけの列を1列目に追加
  d1 = d + 1  # バイアス項込みの次元数
  np.random.seed(random_state)
  w = np.random.rand(d1)  # (w0, w1, w2)
  print('initial w', w)
  w_log = [w.copy()]  # 初期の重み係数
  info_log = [[-1] * 5]  # [it, i, y, o, y-o]
  for it in range(max_it):  # ① 反復 (iteration)
    print('--- iteration:', it)
    err = 0
    for i in range(N):  # ② 各データを入力
      x = X1[i, :]
      y_pred = h_step(np.dot(w, x))
      err += (y[i] - y_pred) ** 2
      print('yhat:', y_pred, 'y:', y[i])
      if y_pred != y[i]:
        w += lr * (y[i] - y_pred) * x  # ③ wを更新
      w_log.append(w.copy())
      info_log.append([it, i, y[i], y_pred, y[i] - y_pred])
    print('err:', err)
    if err == 0:
      print('## converged @ it=', it)
      break
  return w_log, info_log
def get_fourpoints(xor_flag=False):
  """4点のデータを準備
     xor_flag: 各データのラベルをどう設定するか
         True: 入力が異なる符号なら1, False: 入力が共に正なら1
  """
  X = np.array([[-1, -1], [-1, 1], [1, -1], [1, 1]])
  y = np.array([0, 1, 1, 0]) if xor_flag else np.array([0, 0, 0, 1])
  return X, y

import numpy as np

def h_step(x):

""" ステップ関数: 0より大きければ1、それ以外は0 """

return int(x > 0)

def train_perceptron(X, y, lr=0.5, max_it=20, random_state=None):

""" パーセプトロンの学習アルゴリズム

lr: 学習率 (learning rate)

max_it: 最大反復回数 (maximum number of iterations)

"""

N, d = X.shape # データ数x次元

X1 = np.c_[np.ones(N), X] # 1だけの列を1列目に追加

d1 = d + 1 # バイアス項込みの次元数

np.random.seed(random_state)

w = np.random.rand(d1) # (w0, w1, w2)

print('initial w', w)

w_log = [w.copy()] # 初期の重み係数

info_log = [[-1] * 5] # [it, i, y, o, y-o]

for it in range(max_it): # ① 反復 (iteration)

print('--- iteration:', it)

err = 0

for i in range(N): # ② 各データを入力

x = X1[i, :]

y_pred = h_step(np.dot(w, x))

err += (y[i] - y_pred) ** 2

print('yhat:', y_pred, 'y:', y[i])

if y_pred != y[i]:

w += lr * (y[i] - y_pred) * x # ③ wを更新

w_log.append(w.copy())

info_log.append([it, i, y[i], y_pred, y[i] - y_pred])

print('err:', err)

if err == 0:

print('## converged @ it=', it)

break

return w_log, info_log

def get_fourpoints(xor_flag=False):

"""4点のデータを準備

xor_flag: 各データのラベルをどう設定するか

True: 入力が異なる符号なら1, False: 入力が共に正なら1

"""

X = np.array([[-1, -1], [-1, 1], [1, -1], [1, 1]])

y = np.array([0, 1, 1, 0]) if xor_flag else np.array([0, 0, 0, 1])

return X, y

図4　単純パーセプトロンの学習過程を可視化する関数を定義するPythonコード

import matplotlib.pyplot as plt
import seaborn as sns

plt.rcParams['font.size'] = 14
def plot_2dline_with_data(X, y, w, title='', i=-1):
  """ データと w0 + w1 x1 + w2 x2 = 0 をプロット """
  fig = plt.figure(figsize=(5, 5))
  sns.scatterplot(x=X[:, 0], y=X[:, 1], hue=y, s=150)
  xlim = [-1.5, 1.5]
  ylim = [-1.5, 1.5]
  plt.xlim(xlim)
  plt.ylim(ylim)
  plt.xticks([-1, 1])
  plt.yticks([-1, 1])
  # w0 + w1*x1 + w2*x2 = 0 のプロット
  if w[2] == 0:  # w2 == 0 のとき: x1 = -w0/w1
    x2 = np.linspace(*ylim, 2)
    plt.plot([-w[0]/w[1]] * x2.size, x2, '-', linewidth=3, color='r')
  else:  # w2 != 0 のとき: x2 = -(w0 + w1*x1)/w2
    x1 = np.linspace(*xlim, 2)
    plt.plot(x1, -(w[0] + w[1]*x1)/w[2], '-', linewidth=3, color='r')
  if i >= 0: plt.scatter(X[i, 0], X[i, 1], s=300, facecolor='none', 
                         edgecolor='r', linewidth=2)
  plt.title(title)
  return fig

import matplotlib.pyplot as plt

import seaborn as sns

plt.rcParams['font.size'] = 14

def plot_2dline_with_data(X, y, w, title='', i=-1):

""" データと w0 + w1 x1 + w2 x2 = 0 をプロット """

fig = plt.figure(figsize=(5, 5))

sns.scatterplot(x=X[:, 0], y=X[:, 1], hue=y, s=150)

xlim = [-1.5, 1.5]

ylim = [-1.5, 1.5]

plt.xlim(xlim)

plt.ylim(ylim)

plt.xticks([-1, 1])

plt.yticks([-1, 1])

# w0 + w1*x1 + w2*x2 = 0 のプロット

if w[2] == 0: # w2 == 0 のとき: x1 = -w0/w1

x2 = np.linspace(*ylim, 2)

plt.plot([-w[0]/w[1]] * x2.size, x2, '-', linewidth=3, color='r')

else: # w2 != 0 のとき: x2 = -(w0 + w1*x1)/w2

x1 = np.linspace(*xlim, 2)

plt.plot(x1, -(w[0] + w[1]*x1)/w[2], '-', linewidth=3, color='r')

if i >= 0: plt.scatter(X[i, 0], X[i, 1], s=300, facecolor='none',

edgecolor='r', linewidth=2)

plt.title(title)

return fig

図5　単純パーセプトロンの学習と結果表示をするPythonコード

# データの読み込みと学習
xor_flag = False  # (★) True: XOR,  False: AND
X, y = get_fourpoints(xor_flag)  # 学習データを取得
print(f' X:\n{X},\n y: {y}')
w_log, info_log = train_perceptron(X, y, random_state=0)
# 学習過程の表示
for step in range(len(w_log)):
  title = 'it: {}, i: {}, y: {}, y_pred: {}, y - y_pred:{}'\
          .format(*info_log[step])
  print(title)
  w = w_log[step]
  it = info_log[step][0]
  i = info_log[step][1]
  print('w:', w)
  plot_flag = False if (it >= 3) and (it % 5 != 0) else True
  if plot_flag:
    plot_2dline_with_data(X, y, w, title, i)
    plt.show()

# データの読み込みと学習

xor_flag = False # (★) True: XOR, False: AND

X, y = get_fourpoints(xor_flag) # 学習データを取得

print(f' X:\n{X},\n y: {y}')

w_log, info_log = train_perceptron(X, y, random_state=0)

# 学習過程の表示

for step in range(len(w_log)):

title = 'it: {}, i: {}, y: {}, y_pred: {}, y - y_pred:{}'\

.format(*info_log[step])

print(title)

w = w_log[step]

it = info_log[step][0]

i = info_log[step][1]

print('w:', w)

plot_flag = False if (it >= 3) and (it % 5 != 0) else True

if plot_flag:

plot_2dline_with_data(X, y, w, title, i)

plt.show()

図12　決定境界を可視化するための関数を定義するPythonコード

import seaborn as sns
import matplotlib as mpl

def plot_decision_boundary(X, y, clf, xylabels=None, palette=None, fig=None, ngrid=50):
  """ 分類器 clf の決定境界を描画 """
  if fig is None: fig = plt.figure(figsize=(5, 5))
  else: plt.figure(fig.number)
  # 2次元空間にグリッド点を準備
  xmin = X.min(axis=0)  # 各列の最小値
  xmax = X.max(axis=0)  # 各列の最大値
  xstep = [(xmax[j]-xmin[j]) / ngrid for j in range(2)]  # グリッドのステップ幅
  xmin = [xmin[j] - 8*xstep[j] for j in range(2)]  # 少し広めに
  xmax = [xmax[j] + 8*xstep[j] for j in range(2)]
  aranges = [np.arange(xmin[j], xmax[j] + xstep[j], xstep[j]) for j in range(2)]
  x0grid, x1grid = np.meshgrid(*aranges)
  # 各グリッド点でクラスを判定
  y_pred = clf.predict(np.c_[x0grid.ravel(), x1grid.ravel()])
  y_pred = y_pred.reshape(x0grid.shape)  # 2次元へ
  y_pred = np.searchsorted(np.unique(y_pred), y_pred)  # 値をindexへ
  clist = palette.values() if type(palette) is dict else palette
  cmap = mpl.colors.ListedColormap(clist) if palette is not None else None
  plt.contourf(x0grid, x1grid, y_pred, alpha=0.3, cmap=cmap)
  sns.scatterplot(x=X[:, 0], y=X[:, 1], hue=y, palette=palette)
  plt.legend()
  plt.xlim([xmin[0], xmax[0]])
  plt.ylim([xmin[1], xmax[1]])
  if xylabels is not None:
    plt.xlabel(xylabels[0])
    plt.ylabel(xylabels[1])
  return fig

import seaborn as sns

import matplotlib as mpl

def plot_decision_boundary(X, y, clf, xylabels=None, palette=None, fig=None, ngrid=50):

""" 分類器 clf の決定境界を描画 """

if fig is None: fig = plt.figure(figsize=(5, 5))

else: plt.figure(fig.number)

# 2次元空間にグリッド点を準備

xmin = X.min(axis=0) # 各列の最小値

xmax = X.max(axis=0) # 各列の最大値

xstep = [(xmax[j]-xmin[j]) / ngrid for j in range(2)] # グリッドのステップ幅

xmin = [xmin[j] - 8*xstep[j] for j in range(2)] # 少し広めに

xmax = [xmax[j] + 8*xstep[j] for j in range(2)]

aranges = [np.arange(xmin[j], xmax[j] + xstep[j], xstep[j]) for j in range(2)]

x0grid, x1grid = np.meshgrid(*aranges)

# 各グリッド点でクラスを判定

y_pred = clf.predict(np.c_[x0grid.ravel(), x1grid.ravel()])

y_pred = y_pred.reshape(x0grid.shape) # 2次元へ

y_pred = np.searchsorted(np.unique(y_pred), y_pred) # 値をindexへ

clist = palette.values() if type(palette) is dict else palette

cmap = mpl.colors.ListedColormap(clist) if palette is not None else None

plt.contourf(x0grid, x1grid, y_pred, alpha=0.3, cmap=cmap)

sns.scatterplot(x=X[:, 0], y=X[:, 1], hue=y, palette=palette)

plt.legend()

plt.xlim([xmin[0], xmax[0]])

plt.ylim([xmin[1], xmax[1]])

if xylabels is not None:

plt.xlabel(xylabels[0])

plt.ylabel(xylabels[1])

return fig

図13　多層パーセプトロンによる学習をするPythonコード

from sklearn.neural_network import MLPClassifier

clf = MLPClassifier(hidden_layer_sizes=3, learning_rate_init=0.05, random_state=0)
# XORデータの準備
xor_flag = True  # True: XOR,  False: AND
X, y = get_fourpoints(xor_flag)  # 学習データを取得
print(f' X:\n{X},\n y: {y}')
clf.fit(X, y)  # 学習（誤差逆伝播法）の実行
plot_decision_boundary(X, y, clf)  # 決定境界
plt.show()

from sklearn.neural_network import MLPClassifier

clf = MLPClassifier(hidden_layer_sizes=3, learning_rate_init=0.05, random_state=0)

# XORデータの準備

xor_flag = True # True: XOR, False: AND

X, y = get_fourpoints(xor_flag) # 学習データを取得

print(f' X:\n{X},\n y: {y}')

clf.fit(X, y) # 学習（誤差逆伝播法）の実行

plot_decision_boundary(X, y, clf) # 決定境界

plt.show()

図15　学習曲線や結合荷重などを表示するPythonコード

# 学習曲線の表示
plt.plot(range(1, clf.n_iter_ + 1), clf.loss_curve_)
plt.xlabel('Epoch')
plt.ylabel('Loss')
plt.ylim(0,)
plt.show()
# 構造を確認
print('n_outputs:', clf.n_outputs_)  # 出力層の素子数
print('n_layers (with input layer):', clf.n_layers_)  # 入力層を含めた層の数
# 結合荷重
print('coefs:\n', clf.coefs_)  # バイアス項以外の結合荷重（行列のリスト）
print('intercepts:\n', clf.intercepts_)  # バイアス項（ベクトルのリスト）

# 学習曲線の表示

plt.plot(range(1, clf.n_iter_ + 1), clf.loss_curve_)

plt.xlabel('Epoch')

plt.ylabel('Loss')

plt.ylim(0,)

plt.show()

# 構造を確認

print('n_outputs:', clf.n_outputs_) # 出力層の素子数

print('n_layers (with input layer):', clf.n_layers_) # 入力層を含めた層の数

# 結合荷重

print('coefs:\n', clf.coefs_) # バイアス項以外の結合荷重（行列のリスト）

print('intercepts:\n', clf.intercepts_) # バイアス項（ベクトルのリスト）

図17　Breast Cancerデータセットの準備をするPythonコード

from sklearn.datasets import load_breast_cancer
from sklearn.model_selection import train_test_split
from sklearn.metrics import roc_curve, roc_auc_score
from sklearn.preprocessing import StandardScaler

data = load_breast_cancer()  # データセット読み込み
X_orig = data['data']
y = 1 - data['target']  # 悪性を1、良性を0とする正解ラベル
X = X_orig[:, :10]  # 一部の特徴量を利用する
# 訓練データとテストデータに2分割する（検証データは今回は切り分けない）
X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2,
                                                    random_state=1)
# ROCカーブの表示用に関数を準備
def plot_roc(fpr, tpr, marker=None):
    plt.figure(figsize=(5, 5))
    plt.plot(fpr, tpr, marker=marker)
    plt.xlabel('False Positive Rate')
    plt.ylabel('True Positive Rate')
    plt.grid()

from sklearn.datasets import load_breast_cancer

from sklearn.model_selection import train_test_split

from sklearn.metrics import roc_curve, roc_auc_score

from sklearn.preprocessing import StandardScaler

data = load_breast_cancer() # データセット読み込み

X_orig = data['data']

y = 1 - data['target'] # 悪性を1、良性を0とする正解ラベル

X = X_orig[:, :10] # 一部の特徴量を利用する

# 訓練データとテストデータに2分割する（検証データは今回は切り分けない）

X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2,

random_state=1)

# ROCカーブの表示用に関数を準備

def plot_roc(fpr, tpr, marker=None):

plt.figure(figsize=(5, 5))

plt.plot(fpr, tpr, marker=marker)

plt.xlabel('False Positive Rate')

plt.ylabel('True Positive Rate')

plt.grid()

図18　Breast Cancerデータセットによる学習と評価をするPythonコード

clf = MLPClassifier(hidden_layer_sizes=(5, 3), learning_rate_init=0.005, random_state=1)
with_scaling = False  # （★）標準化するか否か
if with_scaling:  # 標準化によるスケーリングあり
  scaler = StandardScaler().fit(X_train)  # 訓練データで平均と標準偏差を求める
  Xscaled_train = scaler.transform(X_train)  # X_train -> Xscaled_train
  Xscaled_test = scaler.transform(X_test)  # X_test -> Xscaled_test
  clf.fit(Xscaled_train, y_train)  # 学習
  y_proba_train = clf.predict_proba(Xscaled_train)  # 訓練データで事後確率予測
  y_proba_test = clf.predict_proba(Xscaled_test)  # テストデータ事後確率予測
else:  # 標準化なし（そのまま）
  clf.fit(X_train, y_train)  # 学習
  y_proba_train = clf.predict_proba(X_train)  # 訓練データで事後確率予測
  y_proba_test = clf.predict_proba(X_test)  # テストデータ事後確率予測
# テストデータに対するROCカーブ
fpr, tpr, threshold = roc_curve(y_test, y_proba_test[:, 1])
plot_roc(fpr, tpr)
plt.show()
# 訓練・テストデータのAUC（Area Under the Curve）スコア
print('AUC (train):', roc_auc_score(y_train, y_proba_train[:, 1]))
print('AUC (test):', roc_auc_score(y_test, y_proba_test[:, 1]))

clf = MLPClassifier(hidden_layer_sizes=(5, 3), learning_rate_init=0.005, random_state=1)

with_scaling = False # （★）標準化するか否か

if with_scaling: # 標準化によるスケーリングあり

scaler = StandardScaler().fit(X_train) # 訓練データで平均と標準偏差を求める

Xscaled_train = scaler.transform(X_train) # X_train -> Xscaled_train

Xscaled_test = scaler.transform(X_test) # X_test -> Xscaled_test

clf.fit(Xscaled_train, y_train) # 学習

y_proba_train = clf.predict_proba(Xscaled_train) # 訓練データで事後確率予測

y_proba_test = clf.predict_proba(Xscaled_test) # テストデータ事後確率予測

else: # 標準化なし（そのまま）

clf.fit(X_train, y_train) # 学習

y_proba_train = clf.predict_proba(X_train) # 訓練データで事後確率予測

y_proba_test = clf.predict_proba(X_test) # テストデータ事後確率予測

# テストデータに対するROCカーブ

fpr, tpr, threshold = roc_curve(y_test, y_proba_test[:, 1])

plot_roc(fpr, tpr)

plt.show()

# 訓練・テストデータのAUC（Area Under the Curve）スコア

print('AUC (train):', roc_auc_score(y_train, y_proba_train[:, 1]))

print('AUC (test):', roc_auc_score(y_test, y_proba_test[:, 1]))

図21　PyTorchによる画像認識の準備をするPythonコード

import torch
import torch.nn as nn
import torch.nn.functional as F
import torchvision
from torchvision import datasets
from torchvision import transforms as transforms
import matplotlib.pyplot as plt
import numpy as np

# GPUが使えるか否かでデータの転送先を指定
device = 'cuda:0' if torch.cuda.is_available() else 'cpu'
img_mean, img_std = (0.1307, 0.3081)  # 画素値の標準化用
# 標準化などの前処理を設定
trans = transforms.Compose([
  transforms.ToTensor(),
  transforms.Normalize((img_mean,), (img_std,))
])
# データセットのダウンロード
rootdir = './data'  # ダウンロード先
train_dataset = datasets.MNIST(root=rootdir, train=True,
  transform=trans, download=True)
test_dataset = datasets.MNIST(root=rootdir, train=False,
  transform=trans, download=True)
# データを読み込む専用のクラスDataLoaderを準備
batch_size = 32
train_loader = torch.utils.data.DataLoader(
  dataset=train_dataset, batch_size=batch_size, shuffle=True)
test_loader = torch.utils.data.DataLoader(
  dataset=test_dataset, batch_size=batch_size, shuffle=False)
# クラス情報
num_classes = len(train_dataset.classes)
cls_names = [f'{i}' for i in range (num_classes)]
# 画像表示用の関数を定義しておく
def tensor_imshow(img, title=None):
  """ Tensorを画像として表示 """
  img = img.numpy().transpose((1, 2, 0))  # (C,H,W)->(H,W,C)
  img = img_std * img + img_mean
  plt.imshow(np.clip(img, 0, 1))
  if title is not None:
    plt.title(title)
  plt.show()
def test_with_examples(model, loader):
  """ loaderのミニバッチの画像を予測結果と共にグリッド表示
    model: 予測に用いるニューラルネット
    loader: DataLoader
  """
  model.eval()  # 推論モード
  with torch.no_grad():  # 推論のみ（勾配計算なし）
    imgs, labels = next(iter(loader))  # ミニバッチ取得
    imgs_in = imgs.view(-1, imgs.shape[2]*imgs.shape[3])
    outputs = model(imgs_in.to(device))  # 順伝播による推論
  _, pred = torch.max(outputs, 1)  # 予測ラベル
  grid = torchvision.utils.make_grid(imgs)  # グリッド画像生成
  title_str1 = '\n'.join(
    [', '.join([cls_names[y] for y in x]) 
      for x in pred.view(-1, 8).tolist()])
  title_str2 = '\n'.join(
    [', '.join([cls_names[y] for y in x]) 
      for x in labels.view(-1, 8).tolist()])
  tensor_imshow(grid, title='Predicted classes:\n'
    + f'{title_str1}\n\nTrue classes:\n{title_str2}\n')
# 訓練データ例の表示
print('\n--- Training data (example) ---\n')
imgs, labels = next(iter(train_loader))
grid = torchvision.utils.make_grid(imgs)
title_str = '\n'.join([', '.join([cls_names[y] for y in x])
  for x in labels.view(-1, 8).tolist()])
tensor_imshow(grid, title=f'True classes:\n{title_str}\n')
# ニューラルネットの構造を定義
class Net(nn.Module):
  def __init__(self):
    super(Net, self).__init__()
    self.fc1 = nn.Linear(28*28, 512)
    self.fc2 = nn.Linear(512, 256)
    self.fc3 = nn.Linear(256, num_classes)
  def forward(self, x):
    x = F.relu(self.fc1(x))
    x = F.relu(self.fc2(x))
    x = self.fc3(x)  # 活性化関数なし（損失関数側でsoftmax）
    return x
net = Net()  # インスタンス生成
net = net.to(device)  # モデルをGPUへ転送（もしGPUがあれば）
# 学習前にテストデータを入力してみる
print('\n--- Result BEFORE training ---\n')
test_with_examples(net, test_loader)

import torch

import torch.nn as nn

import torch.nn.functional as F

import torchvision

from torchvision import datasets

from torchvision import transforms as transforms

import matplotlib.pyplot as plt

import numpy as np

# GPUが使えるか否かでデータの転送先を指定

device = 'cuda:0' if torch.cuda.is_available() else 'cpu'

img_mean, img_std = (0.1307, 0.3081) # 画素値の標準化用

# 標準化などの前処理を設定

trans = transforms.Compose([

transforms.ToTensor(),

transforms.Normalize((img_mean,), (img_std,))

])

# データセットのダウンロード

rootdir = './data' # ダウンロード先

train_dataset = datasets.MNIST(root=rootdir, train=True,

transform=trans, download=True)

test_dataset = datasets.MNIST(root=rootdir, train=False,

transform=trans, download=True)

# データを読み込む専用のクラスDataLoaderを準備

batch_size = 32

train_loader = torch.utils.data.DataLoader(

dataset=train_dataset, batch_size=batch_size, shuffle=True)

test_loader = torch.utils.data.DataLoader(

dataset=test_dataset, batch_size=batch_size, shuffle=False)

# クラス情報

num_classes = len(train_dataset.classes)

cls_names = [f'{i}' for i in range (num_classes)]

# 画像表示用の関数を定義しておく

def tensor_imshow(img, title=None):

""" Tensorを画像として表示 """

img = img.numpy().transpose((1, 2, 0)) # (C,H,W)->(H,W,C)

img = img_std * img + img_mean

plt.imshow(np.clip(img, 0, 1))

if title is not None:

plt.title(title)

plt.show()

def test_with_examples(model, loader):

""" loaderのミニバッチの画像を予測結果と共にグリッド表示

model: 予測に用いるニューラルネット

loader: DataLoader

"""

model.eval() # 推論モード

with torch.no_grad(): # 推論のみ（勾配計算なし）

imgs, labels = next(iter(loader)) # ミニバッチ取得

imgs_in = imgs.view(-1, imgs.shape[2]*imgs.shape[3])

outputs = model(imgs_in.to(device)) # 順伝播による推論

_, pred = torch.max(outputs, 1) # 予測ラベル

grid = torchvision.utils.make_grid(imgs) # グリッド画像生成

title_str1 = '\n'.join(

[', '.join([cls_names[y] for y in x])

for x in pred.view(-1, 8).tolist()])

title_str2 = '\n'.join(

[', '.join([cls_names[y] for y in x])

for x in labels.view(-1, 8).tolist()])

tensor_imshow(grid, title='Predicted classes:\n'

+ f'{title_str1}\n\nTrue classes:\n{title_str2}\n')

# 訓練データ例の表示

print('\n--- Training data (example) ---\n')

imgs, labels = next(iter(train_loader))

grid = torchvision.utils.make_grid(imgs)

title_str = '\n'.join([', '.join([cls_names[y] for y in x])

for x in labels.view(-1, 8).tolist()])

tensor_imshow(grid, title=f'True classes:\n{title_str}\n')

# ニューラルネットの構造を定義

class Net(nn.Module):

def __init__(self):

super(Net, self).__init__()

self.fc1 = nn.Linear(28*28, 512)

self.fc2 = nn.Linear(512, 256)

self.fc3 = nn.Linear(256, num_classes)

def forward(self, x):

x = F.relu(self.fc1(x))

x = F.relu(self.fc2(x))

x = self.fc3(x) # 活性化関数なし（損失関数側でsoftmax）

return x

net = Net() # インスタンス生成

net = net.to(device) # モデルをGPUへ転送（もしGPUがあれば）

# 学習前にテストデータを入力してみる

print('\n--- Result BEFORE training ---\n')

test_with_examples(net, test_loader)

図22　PyTorchによるニューラルネットの学習と結果表示をするPythonコード

# 損失関数（softmax + cross entropy）
criterion = nn.CrossEntropyLoss()
# 最適化手法の選択
optimizer = torch.optim.SGD(net.parameters(), lr=0.01)
# 学習のループ
num_epochs = 5
train_loss_log = []
train_acc_log = []
for epoch in range(num_epochs):
  print(f'Epoch {epoch + 1}/{num_epochs}\n', '-' * 10)
  # 訓練フェーズ
  sum_loss, sum_ok = [0.0, 0]
  for inputs, labels in train_loader:  # ミニバッチ入力
    inputs = inputs.view(-1, 28*28).to(device)
    labels = labels.to(device)  # deviceに転送
    outputs = net(inputs)  # 順伝播
    loss = criterion(outputs, labels)  # 損失計算
    optimizer.zero_grad()  # 勾配をクリア
    loss.backward()  # 誤差逆伝播
    optimizer.step()  # パラメータ更新
    # ミニバッチの評価値（1バッチ分）を計算
    _, preds = torch.max(outputs, 1)  # 予測ラベル
    sum_loss += loss.item() * inputs.size(0)  # 損失
    sum_ok += torch.sum(preds == labels.data)  # 正解数
  # 1エポック分の評価値を計算
  e_loss = sum_loss / len(train_dataset)  # 平均損失
  e_acc = sum_ok.cpu().double() / len(train_dataset)  # 正解率
  print(f'Loss: {e_loss:.4f} Acc: {e_acc:.4f}')
  train_loss_log.append(e_loss)
  train_acc_log.append(e_acc)
# 学習曲線をプロット
fig = plt.figure(figsize=(12, 6))
fig.add_subplot(121)  # 損失
plt.plot(range(1, num_epochs + 1), train_loss_log, '.-')
plt.title('Training data')
plt.xlabel('Epoch')
plt.ylabel('Loss')
plt.ylim(0)
fig.add_subplot(122)  # 正解率
plt.plot(range(1, num_epochs + 1), train_acc_log, '.-')
plt.title('Training data')
plt.xlabel('Epoch')
plt.ylabel('Accuracy')
plt.show()
# テストデータを入力してみる
print('\n--- Result AFTER training ---\n')
test_with_examples(net, test_loader)

# 損失関数（softmax + cross entropy）

criterion = nn.CrossEntropyLoss()

# 最適化手法の選択

optimizer = torch.optim.SGD(net.parameters(), lr=0.01)

# 学習のループ

num_epochs = 5

train_loss_log = []

train_acc_log = []

for epoch in range(num_epochs):

print(f'Epoch {epoch + 1}/{num_epochs}\n', '-' * 10)

# 訓練フェーズ

sum_loss, sum_ok = [0.0, 0]

for inputs, labels in train_loader: # ミニバッチ入力

inputs = inputs.view(-1, 28*28).to(device)

labels = labels.to(device) # deviceに転送

outputs = net(inputs) # 順伝播

loss = criterion(outputs, labels) # 損失計算

optimizer.zero_grad() # 勾配をクリア

loss.backward() # 誤差逆伝播

optimizer.step() # パラメータ更新

# ミニバッチの評価値（1バッチ分）を計算

_, preds = torch.max(outputs, 1) # 予測ラベル

sum_loss += loss.item() * inputs.size(0) # 損失

sum_ok += torch.sum(preds == labels.data) # 正解数

# 1エポック分の評価値を計算

e_loss = sum_loss / len(train_dataset) # 平均損失

e_acc = sum_ok.cpu().double() / len(train_dataset) # 正解率

print(f'Loss: {e_loss:.4f} Acc: {e_acc:.4f}')

train_loss_log.append(e_loss)

train_acc_log.append(e_acc)

# 学習曲線をプロット

fig = plt.figure(figsize=(12, 6))

fig.add_subplot(121) # 損失

plt.plot(range(1, num_epochs + 1), train_loss_log, '.-')

plt.title('Training data')

plt.xlabel('Epoch')

plt.ylabel('Loss')

plt.ylim(0)

fig.add_subplot(122) # 正解率

plt.plot(range(1, num_epochs + 1), train_acc_log, '.-')

plt.title('Training data')

plt.xlabel('Epoch')

plt.ylabel('Accuracy')

plt.show()

# テストデータを入力してみる

print('\n--- Result AFTER training ---\n')

test_with_examples(net, test_loader)

Raspberry Piを100%活用しよう（Vol.77掲載）

投稿日：2022.03.25　｜　カテゴリー：　コード

筆者：米田聡

小型コンピュータボード「Raspberry Pi」（ラズパイ）向けにさまざまな拡張ボードが発売されています。その拡張ボードとラズパイを組み合わせれば、ラズパイでいろいろなことが簡単に試せます。第10回は、サーボモーターを制御できる拡張基板を扱います。

シェルスクリプトマガジン Vol.77は以下のリンク先でご購入できます。

図6　チャンネル1に接続したSG90を動かすサンプルプログラム（sv.py）

import smbus, SVO
import time

servo = SVO.SVO()

#init

servo.Servo_Switch('b', '0')         # 両チャンネルオフ
servo.Set_servo_cycle('20000')       # PWMピリオド20ミリ秒
servo.Set_servo_duty_min('b','500')  # 最小デューティサイクル0.5ミリ秒
servo.Set_servo_duty_max('b','2400') # 最大デューティサイクル2.4ミリ秒
servo.Set_servo_duty('b', '500')     # -90度に初期化
servo.Set_servo_Write()              # 初期値を書き込み

# 作動
servo.Servo_Switch('b', '1')         # 両チャンネルオン
time.sleep(3)

for r in range(10, 190, 10):         # 10度から180度まで
    print("r=" + str(r))
    sval = str(int( r*1900/180 + 500 ))
    servo.Set_servo_duty('1', sval)
    servo.Set_servo_Write()
    time.sleep(3)

import smbus, SVO

import time

servo = SVO.SVO()

#init

servo.Servo_Switch('b', '0') # 両チャンネルオフ

servo.Set_servo_cycle('20000') # PWMピリオド20ミリ秒

servo.Set_servo_duty_min('b','500') # 最小デューティサイクル0.5ミリ秒

servo.Set_servo_duty_max('b','2400') # 最大デューティサイクル2.4ミリ秒

servo.Set_servo_duty('b', '500') # -90度に初期化

servo.Set_servo_Write() # 初期値を書き込み

# 作動

servo.Servo_Switch('b', '1') # 両チャンネルオン

time.sleep(3)

for r in range(10, 190, 10): # 10度から180度まで

print("r=" + str(r))

sval = str(int( r*1900/180 + 500 ))

servo.Set_servo_duty('1', sval)

servo.Set_servo_Write()

time.sleep(3)

特別企画 ESP32とラズパイで作る本格IoT（Vol.77記載）

投稿日：2022.03.25　｜　カテゴリー：　コード

著者：魔法少女

マイコン「ESP32」と小型コンピュータ「Raspberry Pi」を使って本格的なIoT（モノのインターネット）環境を構築してみましょう。本企画では、五つの会議室の温度、湿度、利用状況を監視・分析し、会議室内の電源を制御するシステムを構築します。

シェルスクリプトマガジン Vol.77は以下のリンク先でご購入できます。

図4　ユーザー名とパスワードの生成スクリプト（user_create.sh）

#!/bin/sh

tee ~/confer_pw.txt << EOF > /dev/null
pub1:passwd1
pub2:passwd2
pub3:passwd3
pub4:passwd4
pub5:passwd5
sub1:passwd6
EOF
for i in $(seq 6)
do
sed -i -e "s|passwd${i}|$(pwgen 8 1)|" ~/confer_pw.txt
done

#!/bin/sh

tee ~/confer_pw.txt << EOF > /dev/null

pub1:passwd1

pub2:passwd2

pub3:passwd3

pub4:passwd4

pub5:passwd5

sub1:passwd6

EOF

for i in $(seq 6)

sed -i -e "s|passwd${i}|$(pwgen 8 1)|" ~/confer_pw.txt

done

図5　Eclipse Mosquittoの認証設定スクリプト（mosquitto_setting.sh）

#!/bin/sh

sudo cp ~/confer_pw.txt /etc/mosquitto/confer_pwfile
sudo mosquitto_passwd -U /etc/mosquitto/confer_pwfile
sudo tee /etc/mosquitto/conf.d/confer.conf << EOF > /dev/null
listener 1883
allow_anonymous false
password_file /etc/mosquitto/confer_pwfile
EOF
sudo systemctl reload mosquitto

#!/bin/sh

sudo cp ~/confer_pw.txt /etc/mosquitto/confer_pwfile

sudo mosquitto_passwd -U /etc/mosquitto/confer_pwfile

sudo tee /etc/mosquitto/conf.d/confer.conf << EOF > /dev/null

listener 1883

allow_anonymous false

password_file /etc/mosquitto/confer_pwfile

EOF

sudo systemctl reload mosquitto

図7　サブスクライバのプログラム（subscriber.py)

#!/bin/env python3

import paho.mqtt.client as mqtt
import time

##
mqtt_topic = "confer/#"
csv_file = "./room1.csv"
subscriber_username = "sub1"
subscriber_password = "IBehie1h"
broker_hostname = "localhost"

##
def on_connect(client, userdata, flags, rc):
    print("Connected with result code "+str(rc))
    client.subscribe(mqtt_topic)

def on_message(client, userdata, msg):
    now_unixtime = time.time()
    iot_device = msg.topic + "," + str(msg.payload, encoding='utf-8', errors='replace') + "," + str(now_unixtime)
    print(iot_device)
    with open(csv_file, mode='a') as f:
        f.write(iot_device + "\n")

##
client = mqtt.Client()
client.on_connect = on_connect
client.on_message = on_message
client.username_pw_set(username=subscriber_username, password=subscriber_password)
client.connect(broker_hostname)
client.loop_forever()

#!/bin/env python3

import paho.mqtt.client as mqtt

import time

mqtt_topic = "confer/#"

csv_file = "./room1.csv"

subscriber_username = "sub1"

subscriber_password = "IBehie1h"

broker_hostname = "localhost"

def on_connect(client, userdata, flags, rc):

print("Connected with result code "+str(rc))

client.subscribe(mqtt_topic)

def on_message(client, userdata, msg):

now_unixtime = time.time()

iot_device = msg.topic + "," + str(msg.payload, encoding='utf-8', errors='replace') + "," + str(now_unixtime)

print(iot_device)

with open(csv_file, mode='a') as f:

f.write(iot_device + "\n")

client = mqtt.Client()

client.on_connect = on_connect

client.on_message = on_message

client.username_pw_set(username=subscriber_username, password=subscriber_password)

client.connect(broker_hostname)

client.loop_forever()

図24　ESP32搭載ボートの起動時に無線LANに接続するプログラム（boot.py）

import network

SSID = 'USP_OFFICE_B'
PASSWORD = 'uspuspusp1234'

wlan_if = network.WLAN(network.STA_IF)
wlan_if.active(True)
wlan_if.connect(SSID, PASSWORD)

import network

SSID = 'USP_OFFICE_B'

PASSWORD = 'uspuspusp1234'

wlan_if = network.WLAN(network.STA_IF)

wlan_if.active(True)

wlan_if.connect(SSID, PASSWORD)

図25　パブリッシャのプログラム（main.py）

import time
import dht
from machine import Pin
from umqtt.simple import MQTTClient

mqtt_topic = "confer/room1"
publisher_username = "pub1"
publisher_password = "パスワード"
publisher_id = "room1_esp32"
broker_address = "192.168.1.100"
interval_time = "1"

hf_sensor = Pin(5, Pin.IN, Pin.PULL_UP)
th_sensor = dht.DHT11(Pin(22))

time.sleep(5)

while True:
  value1 = hf_sensor.value()
  th_sensor.measure()
  value2 = th_sensor.temperature()
  value3 = th_sensor.humidity()
  iot_value = str(value1)+","+str(value2)+","+str(value3)
  print(iot_value)
  publisher = MQTTClient(publisher_id,broker_address,user=publisher_username,password=publisher_password)
  publisher.connect()
  publisher.publish(mqtt_topic, msg=iot_value)
  publisher.disconnect()
  time.sleep(int(interval_time))

import time

import dht

from machine import Pin

from umqtt.simple import MQTTClient

mqtt_topic = "confer/room1"

publisher_username = "pub1"

publisher_password = "パスワード"

publisher_id = "room1_esp32"

broker_address = "192.168.1.100"

interval_time = "1"

hf_sensor = Pin(5, Pin.IN, Pin.PULL_UP)

th_sensor = dht.DHT11(Pin(22))

time.sleep(5)

while True:

value1 = hf_sensor.value()

th_sensor.measure()

value2 = th_sensor.temperature()

value3 = th_sensor.humidity()

iot_value = str(value1)+","+str(value2)+","+str(value3)

print(iot_value)

publisher = MQTTClient(publisher_id,broker_address,user=publisher_username,password=publisher_password)

publisher.connect()

publisher.publish(mqtt_topic, msg=iot_value)

publisher.disconnect()

time.sleep(int(interval_time))

特集1 AWKプログラミング入門（Vol.77記載）

投稿日：2022.03.25　｜　カテゴリー：　コード

著者：斉藤博文

シェルスクリプトで複雑なテキストデータ処理を実行する場合、「AWK」（オーク）というプログラミング言語が利用されます。AWKの処理系（実行環境）はとても小さく、シェルスクリプトと組み合わせて使うのに適しています。本特集では、このAWKのプログラミングを分かりやすく解説します。

シェルスクリプトマガジン Vol.77は以下のリンク先でご購入できます。

図8　ユーザー定義関数「factorial」の実装例

{
    print factorial($0);
}

function factorial(n,    i, ret) {
    ret = 1;

    for (i = 1; i <= n; i++) {
        ret *= i
    }

    return ret;
}

{

print factorial($0);

}

function factorial(n, i, ret) {

ret = 1;

for (i = 1; i <= n; i++) {

ret *= i

}

return ret;

}

図9　改変した階乗処理のAWKプログラム（factorial.awk）

BEGIN {
    n = n ? n : 5;

    print factorial(n);
}

function factorial(n) {
    ret = 1;

    for (i = 1; i <= n; i++) {
        ret *= i
    }

    return ret;
}

BEGIN {

n = n ? n : 5;

print factorial(n);

}

function factorial(n) {

ret = 1;

for (i = 1; i <= n; i++) {

ret *= i

}

return ret;

}

シェルスクリプトマガジンvol.77 Web掲載記事まとめ

投稿日：2022.03.25　｜　カテゴリー：　コード

004　レポート　古いPC向けOS「Chrome OS Flex」
005　レポート　WebブラウザでDebianが動作
006　NEWS FLASH
008　特集1　AWKプログラミング入門/斉藤博文　コード掲載
017　Hello Nogyo!
018　特集2　MIRACLE LINUXという選択肢/弦本春樹
022　特集3　デスクトップ版Power Automateを活用しよう/三沢友治
032　特別企画　ESP32とラズパイで作る本格IoT/魔法少女　コード掲載
050　先端技術　量子コンピュータ
052　Raspberry Piを100%活用しよう/米田聡　コード掲載
056　機械学習ことはじめ/川嶋宏彰　コード掲載
068　Pythonあれこれ/飯尾淳　コード掲載
074　JSON/桑原滝弥、イケヤシロウ
076　中小企業手作りIT化奮戦記/菅雄一
080　法林浩之のFIGHTING TALKS/法林浩之
082　タイ語から分かる現地生活/つじみき
088　香川大学SLPからお届け！/石上椋一　コード掲載
094　Bash入門/大津真
104　Techパズル/gori.sh
105　コラム「ユニケージ流の業務システム開発」/シェル魔人

Vol.77

投稿日：2022.03.25　｜　カテゴリー：　バックナンバー

　シェルスクリプトで複雑なテキストデータ処理を行う場合によく使われるのが「AWK」（オーク）というプログラミング言語です。AWKは、1977年という古い時代からあり、シェルスクリプトとの親和性が良い点から現在も利用されています。特集1では、このAWKのプログラミングを初めての人でも分かりやすいように解説しています。
　2021年末にRed Hat Enterprise Linux（RHEL） 8互換の無料OSである「CentOS Linux（CentOS） 8」の開発およびサポートが終了しました。今後、業務システムなどでCentOSを使い続けることが難しい状況になっています。特集2では、CentOSの代替となる「MIRACLE LINUX」を紹介しました。国内開発のMIRACLE LINUXで、CentOSからの安心でスムーズな移行を実現しましょう。
　人間が行っていた業務処理などを自動化する技術である「RPA」（Robotic Process Automation）は、まだまだ注目されています。このRPAですが、昨年3月にマイクロソフトが無料提供を開始した「Power Automate for Desktop」により身近なものになりました。特集3では、Power Automate for Desktopの機能や使い方を分かりやすく解説しています。RPAはどのようなもので、どのように構築すればよいのかを、Power Automate for Desktopから学びましょう。
　このほか、特別企画では、マイコン「ESP32」と小型コンピュータボード「Raspberry Pi」を使って、五つの会議室の状態を監視するIoT環境の構築方法を紹介しています。IoTによく利用される「MQTT」（Message Queueing Telemetry Transport）のプロトコルを用いています。さまざまなIoT環境に応用してください。
　今回も読み応え十分のシェルスクリプトマガジン Vol.77。お見逃しなく!

※記事掲載のコードはこちら。記事の補足情報はこちら

※読者アンケートはこちら

Vol.77 補足情報

投稿日：2022.03.25　｜　カテゴリー：　コード

特別企画　ESP32とラズパイで作る本格IoT

　p.41の左側中央にある「>>> upip,install(‘umqtt.simple’)」の「,」（カンマ）は「.」（ピリオド）の誤りです。お詫びして訂正します。

>>> upip.install('umqtt.simple')

1	>>> upip.install('umqtt.simple')

　p.47の右側中央やや上にある「最後の「COM5」がCOM番号です。「5」の数字は、使っているパソコンの環境によって変わりますので、」の「COM5」は「COM9」、「5」は「9」の誤りです。お詫びして訂正します。

情報は随時更新致します。

shell-mag ブログの 2022年3月のアーカイブを表示しています。

-->

シェルスクリプトマガジン

香川大学SLPからお届け！（Vol.77掲載）

Pythonあれこれ（Vol.77掲載）

機械学習ことはじめ（Vol.77掲載）

Raspberry Piを100%活用しよう（Vol.77掲載）

特別企画 ESP32とラズパイで作る本格IoT（Vol.77記載）

特集1 AWKプログラミング入門（Vol.77記載）

シェルスクリプトマガジンvol.77 Web掲載記事まとめ

Vol.77

Vol.77 補足情報

特別企画　ESP32とラズパイで作る本格IoT

固定ページ

アーカイブ

カテゴリー