きっかけ

classの概念は初心者である私にとって非常に難しい。

今まで学習から逃げ続けてきたが、現在取り組んでいるkaggleコンペではよく、sklearnの準拠モデルから、分類器を作成しているのをよく見かけますので、復習も兼ねて再整理をしてみます。

クラスの定義とインスタンス(オブジェクト)の生成

class Car:
    def display(self):
        print('model:', self.model)
        print('model:', self.color)


car1 = Car()
car1.model = 'boxcar'
car1.color = 'white'
car1.display()

model: boxcar
model: white

car2 = Car()
car2.model = 'sedan'
car2.color = 'gray'
car2.display()

model: sedan
model: gray

メソッドとは、クラス内で定義した関数のこと。
構造は、

class クラス名:
     def メソッド名(self, 引数)
         処理

__init__メソッド

class Car:

    def __init__(self, model, color):
        self.model = model
        self.color = color

    def display(self):
        print('model:', self.model)
        print('color:', self.color)
car1 = Car('roadster', 'red')
car1.display()

model: roadster
color: red

インスタンスのデータ属性名に値を設定するには...

方法①

変数 = クラス名
変数.データ属性名 = 値

方法②　インスタンス生成と同時に、データ属性を設定

class クラス名:

    def __init__(self, 引数, ...):
        self.データ属性名 = 値

変数 = クラス名(引数, ...)

メリット：
①データ属性の設定が防げる!
②プログラムを簡潔にできる!

*コンストラクタとは、インスタンスが生成される時に実行されるメソッド（関数）の一種です。インスタンスの生成、扱う変数などの初期化を主に行います。

マングリング

クラス内部だけで使うメソッドを定義。
内部だけで使うメソッドを間違って外部から呼ぶのを防ぐ。

class Car:

    def __init__(self, model, color):  #__init__はマングリングではない
        self.model = model
        self.color = color
        self.speed = 0

    def display(self):
        print('model:', self.model)
        print('color:', self.color)
        print('speed:', self.speed)

    def accelerate(self, pressure):
        self.speed += pressure
        self.__check_speed()          #__check_speedメソッド呼び出し

    def __check_speed(self):          #__check_speedはマングリング
        if self.speed > 100:
            self.speed = 100

car1 = Car('roadster', 'red')
car1.display() #()が抜けるとメソッド呼び出しができない

model: roadster
color: red
speed: 0

car1.accelerate(50)
car1.accelerate(60)
car1.display()

model: roadster
color: red
speed: 100

クラス属性

class Car:

    SPEED_LIMIT = 90

    def __init__(self, model, color):  #__init__はマングリングではない
        self.model = model
        self.color = color
        self.speed = 0

    def display(self):
        print('model:', self.model)
        print('color:', self.color)
        print('speed:', self.speed)

    def accelerate(self, pressure):
        self.speed += pressure
        self.__check_speed()

    def __check_speed(self):          #__check_speedはマングリング
        if self.speed > Car.SPEED_LIMIT:  #or self.SPEED_LIMIT でもOK
            self.speed = Car.SPEED_LIMIT

car1 = Car('roadster', 'red')
car1.accelerate(40)
car1.accelerate(60)
car1.display()

継承とオーバーライド

'''
class Car:

    SPEED_LIMIT = 90

    def __init__(self, model, color):  #__init__はマングリングではない
        self.model = model
        self.color = color
        self.speed = 0

    def display(self):
        print('model:', self.model)
        print('color:', self.color)
        print('speed:', self.speed)

    def accelerate(self, pressure):
        self.speed += pressure
        self.__check_speed()

    def __check_speed(self):          #__check_speedはマングリング
        if self.speed > Car.SPEED_LIMIT:  #or self.SPEED_LIMIT でもOK
            self.speed = Car.SPEED_LIMIT
'''

class Truck(Car): #Carクラスを基底として、Truckクラスを派生される

    def __init__(self, model, color, cargo): #cargo追加
        Car.__init__(self, model, color)
        self.cargo = cargo

    def display(self):
        print('cargo:', self.cargo)

truck1 = Truck('trailer', 'yellow', 'potato')
truck1.accelerate(30)
truck1.display()

cargo: potato

継承：既存のクラスが持つ機能を新しいクラスですべて引き継ぐ

継承におけるメソッドの追加

'''
class Car:

    SPEED_LIMIT = 90

    def __init__(self, model, color):  #__init__はマングリングではない
        self.model = model
        self.color = color
        self.speed = 0

    def display(self):
        print('model:', self.model)
        print('color:', self.color)
        print('speed:', self.speed)

    def accelerate(self, pressure):
        self.speed += pressure
        self.__check_speed()

    def __check_speed(self):          #__check_speedはマングリング
        if self.speed > Car.SPEED_LIMIT:  #or self.SPEED_LIMIT でもOK
            self.speed = Car.SPEED_LIMIT
'''

class Truck(Car):

#オーバーライド    
    def __init__(self, model, color, cargo):
        Car.__init__(self, model, color)
        self.cargo = cargo

    def display(self):
        Car.display(self)
        print('cargo:', self.cargo)

#新規メソッド
    def load(self, cargo):
        self.cargo = cargo

truck1 = Truck('trailer', 'yellow', 'potato')
truck1.load('lettuce')
truck1.display()

model: trailer
color: yellow
speed: 0
cargo: lettuce

参照

・100問でわかるPython 単行本 松浦 健一郎 (著), 司 ゆき (著)
・sklearn準拠モデルの作り方
・https://techacademy.jp/magazine/18842

21世紀も20年経とうとしている今、Python スクリプトを走らせてスペルミスなどの単純ミスて止まるのは文明的じゃない; flake8 などの静的解析を使おう。それは多くの記事があるのでいいとして、自作の package に flake8 を使うと flake8

F401 'mypackage.module1' imported but unused

みたいなことをいう。その import には意味があるんだけど... という話。

設定

package とは複数のモジュールがひとつのディレクトリに入ったもの:

mypackage/
  + __init__.py
  + module1.py
  + module2.py

import mypackage

して使う。

module1.py では関数 function1() が、module2 には class Class2 が定義してあるとしよう。

問題

import mypackage

mypackage.module1.function1()
c = mypackage.Class2()

のように使うために __init__.py に

import mypackage.module1
from mypackage.module2 import Class2

と書きたい。すると、

$ flake8 mypackage/*.py
mypackage/__init__.py:1:1: F401 'mypackage.module1' imported but unused
mypackage/__init__.py:2:1: F401 'mypackage.module2.Class2' imported but unused

と言われる。確かに __init__.py では使ってないけど意味はあるんだ。これらの import 文が無いと

import mypackage.module1
from mypackage.module2 import Class2

と毎回書かなければならないので、それはナイ。PEP8的な正解は？

解決法１

編集中...

解決法２

from . import module1
from .module2 import Class2

__all__ = ['module1', 'Class2']

まず、. を使えば「このパッケージ」を意味するので、mypackage と繰り返し書かなくて良い。パッケージ名を変えて修正しなくていいという利点もある。そして __all__ というリストに import したものを書く。すると、unused でなくなったので flake8 さんは満足。

__all__ とは

flake8 を満足させるためだけのダミーの変数というわけではなく、一応意味はある。

__init__.py に import を書かず、

__all__ = ['module1']

と書くと、奨励されないアイツ from mypackage import * した時に

from mypackage import *

module1.funtion()

module1 が読み込まれる。もし、__all__ も from . import module1 も無ければ、import * でも　module1 は読み込まれない。

とはいえ、from . import module1 があれば __all__ = ['module1'] は不要なので __all__ は flake8 向けの重複とは言える。もっと良い方法があればコメントください。

参考

https://stackoverflow.com/questions/31079047/python-pep8-class-in-init-imported-but-not-used

21世紀も20年経とうとしている今、Python スクリプトを走らせてスペルミスなどの単純ミスて止まるのは文明的じゃない; flake8 などの静的解析を使おう。それは多くの記事があるのでいいとして、自作の package に flake8 を使うと

F401 'mypackage.module1' imported but unused

みたいなことをいう。その import には意味があるんだけど... という話。

設定

package とは複数のモジュールがひとつのディレクトリに入ったもの:

mypackage/
  + __init__.py
  + module1.py
  + module2.py

import mypackage

して使う。

module1.py では関数 function1() が、module2 には class Class2 が定義してあるとしよう。

問題

import mypackage

mypackage.module1.function1()
c = mypackage.Class2()

のように使うために __init__.py に

import mypackage.module1
from mypackage.module2 import Class2

と書きたい。すると、

$ flake8 mypackage/*.py
mypackage/__init__.py:1:1: F401 'mypackage.module1' imported but unused
mypackage/__init__.py:2:1: F401 'mypackage.module2.Class2' imported but unused

と言われる。確かに __init__.py では使ってないけど意味はあるんだ。これらの import 文が無いと

import mypackage.module1
from mypackage.module2 import Class2

と毎回書かなければならないので、それはナイ。PEP8的な正解は？

解決法１（手っ取り早い方法）

# noqa とコメントすると flake8 などの linter が無視する。

from . import module1  # noqa
from .module2 import Class2  #noqa

解決法２

やらなくても動くけど、package の公式の機能は __all__ にリストアップべき。

Any backwards compatibility guarantees apply only to public interfaces. Accordingly, it is important that users be able to clearly distinguish between public and internal interfaces.

To better support introspection, modules should explicitly declare the names in their public API using the __all__ attribute.

https://www.python.org/dev/peps/pep-0008/

すなわち、

from . import module1
from .module2 import Class2

__all__ = ['module1', 'Class2']

まず、. を使えば「このパッケージ」を意味するので、mypackage と繰り返し書かなくて良い。パッケージ名を変えても修正しなくていいという利点もある。そして __all__ というリストに import したものを書く。すると、unused でなくなったので flake8 さんは満足。

__all__ とは

flake8 を満足させるためだけのダミーの変数というわけではなく、一応意味はある。

__init__.py に import を書かず、

__all__ = ['module1']

と書くと、奨励されないアイツ from mypackage import * した時に

from mypackage import *

module1.funtion()

module1 が読み込まれる。もし、__all__ も from . import module1 も無ければ、import * でも　module1 は読み込まれない。

from . import module1 があれば __all__ = ['module1'] は不要なので __all__ は重複とは言える。逆に、__all__ だけでは import mypackage の時には自動 import してくれない。とはいえ、PEP8 に書かれていることなので、他の人も使う package として公開するときはちゃんと __all__ を書くのが良さそう。

参考

すべてはここに書いてあった。

https://stackoverflow.com/questions/31079047/python-pep8-class-in-init-imported-but-not-used

pyenvでインストールできるバージョンを確認

$ pyenv install --list

pyenvで Python 3.6.8をインストール

$ pyenv install 3.6.8

インストールしたPythonのバージョンを確認

$ pyenv versions

バージョンの切替　グローバル

$ pyenv global 3.6.8

バージョンの切替　ローカル

$ pyenv local 3.6.8

バージョン確認

$ python --version

仮想環境の作成

$ cd <作成したい場所>　*作成したいディレクトリに移動
$ python3 -m venv <作成する環境名>　*ディレクトリ内に仮想環境を作成

作成したディレクトリに入る

$ cd <作成する環境名>

仮想環境の有効化

$ source bin/activate

仮想環境でIDLEを起動

$ python -m idlelib.idle

IDLEを閉じる（ショートカット　⌘+W）

IDLE起動時に以下警告が表示される場合

WARNING: The version of Tcl/Tk (8.5.9) in use may be unstable. Visit
http://www.python.org/download/mac/tcltk/ for current information.
WARNING: The system preference "Prefer tabs when opening documents" is set to
"Always". This will cause various problems with IDLE. For the best experience,
change this setting when running IDLE (via System Preferences -> Dock).

メッセージのとおり
- Macのシステム環境設定でDockを開き、「書類を開くときはタブで開く：常に」 になっていたら他の項目に変更
- the ActiveState web site から ActiveTcl 8.6 をインストール（インストール後にPython の再インストールが必要）

Python の再インストール

$ pyenv uninstall 3.6.8
$ brew info tcl-tk #バージョンが8.6以上か確認
$ pyenv install 3.6.8

仮想環境を無効化

$ deactivate

参考：
How to launch python Idle from a virtual environment (virtualenv)

pyenvでインストールできるバージョンを確認

$ pyenv install --list

pyenvで Python 3.6.8をインストール

$ pyenv install 3.6.8

インストールしたPythonのバージョンを確認

$ pyenv versions

バージョンの切替　グローバル

$ pyenv global 3.6.8

バージョンの切替　ローカル

$ pyenv local 3.6.8

バージョン確認

$ python --version

仮想環境の作成

$ cd <作成したい場所>　*作成したいディレクトリに移動
$ python3 -m venv <作成する環境名>　*ディレクトリ内に仮想環境を作成

作成したディレクトリに入る

$ cd <作成する環境名>

仮想環境の有効化

$ source bin/activate

仮想環境でIDLEを起動

$ python -m idlelib.idle

IDLEを閉じる（ショートカット　⌘+W）

IDLE起動時に以下警告が表示される場合

WARNING: The version of Tcl/Tk (8.5.9) in use may be unstable. Visit
http://www.python.org/download/mac/tcltk/ for current information.
WARNING: The system preference "Prefer tabs when opening documents" is set to
"Always". This will cause various problems with IDLE. For the best experience,
change this setting when running IDLE (via System Preferences -> Dock).

メッセージのとおり
- Macのシステム環境設定でDockを開き、「書類を開くときはタブで開く：常に」 になっていたら他の項目に変更
- the ActiveState web site から ActiveTcl 8.6 をインストール（インストール後にPython の再インストールが必要）

Python の再インストール

$ pyenv uninstall 3.6.8
$ brew info tcl-tk #バージョンが8.6以上か確認
$ pyenv install 3.6.8

仮想環境を無効化

$ deactivate

参考：
How to launch python Idle from a virtual environment (virtualenv)

概要

Automation Anywhere でのボット作成にある程度慣れてくると、外部モジュールを実行してできることを拡張したくなります。たとえば、Automation Anywhere から PowerShell スクリプトの処理を呼び出したり、Python を呼び出して機械学習の処理をさせたり、といったことをしたくなります。これらをどう実行できるかについて整理したいと思います。

前提

• Automation Anywhere
  • Enterprise 11.3.2
  • Community Edition (2019年8月現在、UIは英語です)
• Windows 10 バージョン1903

Automation Anywhere で実行できる外部モジュール呼び出し方法

まず、Automation Anywhereでどのように外部モジュール呼び出し (=任意の操作の実行) ができるかについて一覧にしてみました。

方法	カテゴリ/コマンド名	概要と特徴
別タスクの呼び出し	Task / Run Task	Automation Anywhere で作成した別のボットを呼び出す。自分で作成したよく使うアクションリストをサブルーチン的に使うのに便利。引き継ぐローカル変数をダイアログボックスでビジュアルに任意数設定できるので、情報の受け渡しはとても楽。リピート有無や実行スピードも設定可能。
Meta Bot (およびBot Store)	Run Logic	Meta Bot Designerの権限を持つ人が汎用的につくったボットや、Bot Storeからインストールしたボットを部品として呼び出すのに使う。通常のボットでは使えない .DLL の呼び出しが可能なので、パッケージ内での複雑で汎用的なタスクが可能。Meta Botで定義されているInput / Output パラメータをダイアログボックスでビジュアルに設定可能。
Excelマクロの呼び出し	Excel / Run Excel Macro	開いているExcelファイルの中のExcelマクロを引数リストを渡して呼び出せる。
Windowsアプリの呼び出し	Open Program/File	Windows 実行形式ファイル(.exe)や、これに関連付けられたファイルを指定して実行できます。ファイルの開始ディレクトリや引数も指定可能。さまざまな種類の外部モジュールを最も汎用的に呼び出せる。ただし、グラフィカルユーザーインターフェイス (GUI) を持っている必要がある(持っていなくても実行されるがエラーが出る)。
OS サービスの呼び出し	Services / Start Service	インストール済みのサービスを開始できる。サービスモジュールを自分でプログラミングすることは少ないと思われるので、この方法はほぼ使わないかも。
VBScript/JScriptの呼び出し	Run Script	ローカルにある .vbs / .js を指定して実行します。パラメータや戻り値を得るための変数を設定することもできます。コマンドラインユーザーインターフェイス (CUI) のみでも実行可能。
データベース内のスクリプトの呼び出し	Database / Run Stored Procedure	ストアドプロシージャの関数名/引数を指定して実行できます。引数は任意数追加でき、結果をCSVにエクスポートするオプションもあります。
ウェブページ内のスクリプトの呼び出し	Web Recorder / Execute JavaScript Function	ウェブページ内のJavaScript関数と、実行結果を返す変数を指定して呼び出せます。
XML内のスクリプトの呼び出し	XML / Execute XPath Function	XPath Function を実行できます。実行するXPath 式と、戻り値を得る変数を指定します。
SOAP/REST ウェブサービスの呼び出し	SOAP Web Service	SOAP/REST ウェブサービスのURIを呼び出してサービスを実行して戻り値を変数に返します。必要に応じてヘッダーパラメーターを任意数、出力結果をXMLとして保存、認証がある場合のアカウント情報やクライアント証明書を指定できます。

今回呼び出したいPowerShellスクリプトの場合、Open Program/File アクションを利用するのが良さそうです。

Open Program/File による外部モジュールの呼び出し

Open Program/Fileは以下のダイアログボックスでパラメーターの指定が可能です。

ここに呼び出したいPowerShellスクリプトを指定するのですが、今回は2つの方法を考えてみました。
ちなみに、PowerShellスクリプトは以下のものを用意します。

Hello.ps1

Add-Type -Assembly System.Windows.Forms
[System.Windows.Forms.MessageBox]::Show("Hello World!!", "こんにちは")

方法1: PS1ファイルの関連付けを利用して実行する

PowerShell のスクリプトファイル .ps1 はアイコンを見ればわかる通り、関連付けがされています。
したがって、.ps1 を以下のダイアログボックスの「Program/File パス」に直接指定しても動作するはずです。

この状態で保存すると、以下のアクションリストができます。

これで実行してみると...スクリプトの中身が実行される代わりにメモ帳でスクリプトファイルが開かれてしまいました...

これがなぜだったかを見てみると、実は.ps1のアクションの関連付けは、少なくともWindows 10の環境ではメモ帳にされていることがわかります。

したがって、この関連付けをpowershell.exeに変更すると動作するはずですので、変更してみましょう。

PowerShell.exe は"%systemroot%\system32\windowspowershell\v1.0\powershell.exe"にあるので、このファイルを指定します。

また、表の中でも記載した通り、このコマンドはGUIがあるプログラムを前提としており、CUIのプログラムを呼び出した場合は「WaitForInputIdleに失敗しました。プロセスがグラフィック インターフェイスを含んでいない可能性があります。」というエラーが出ます。これは無視できるので、Error Handlingで囲みます。

これで実行すると、以下の通りスクリプトが実行されることがわかります。

(注) デフォルトのセキュリティレベルではPowerShellはスクリプトとして実行が禁止されているので、スクリプトを実行する前にPowerShellコマンドSet-ExecutionPolicy -ExecutionPolicy RemoteSigned -Scope CurrentUserを実行しておくようにしましょう。終わったらSet-ExecutionPolicy -ExecutionPolicy Restricted -Scope CurrentUserでセキュリティを元に戻します。

方法2: 「PowerShell実行ファイル」+「引数=PS1ファイル」を指定して実行する

さて、関連付けがされていればファイルを直接「Program/File パス」に指定すれば実行されることはわかりましたが、この関連付けは通常はされていないため、実行形式ファイルと引数としてのスクリプトファイルの2つを指定する方法を試してみたいと思います。

「Program/File パス」にはシステム変数をうまく使って汎用的に$System(SystemRoot)$\System32\WindowsPowerShell\v1.0\powershell.exeと記載します。「パラメータ」にはHello.ps1のパスを指定します。

前述のエラー処理も取り入れると、アクションリストは以下のようになります。

これで実行すると、やはり以下の通りスクリプトが実行されることがわかります。

外部モジュールでの処理を待つようにする

ちなみに、今回のOpen Program/File アクションは、外部スクリプト呼出し後、処理を待たずに次のアクションに移ってしまうので、外部スクリプトの結果を待つには、たとえば共通のテキストファイルをコミュニケーション用に用意して、値が書き込まれたら次の処理に移る、などの工夫が必要です。たとえば以下のようにアクションリストを設定してみると、PowerShellでのダイアログボックスで「OK」ボタンをクリックした直後に、アクションリストの次のダイアログボックスが表示されることがわかります。

Hello.ps1

Add-Type -Assembly System.Windows.Forms
[System.Windows.Forms.MessageBox]::Show("Hello World!!", "こんにちは")
"Done`n" > "C:\Automation-Anywhere-Temporary-Folder\result.txt"

(おまけ) 同様の方法でPython スクリプトも実行

以上の方法は、PowerShellに限らず、ほかのスクリプトでも応用可能です。Python については、インストーラでセットアップを行うと、.pyファイルがpython.exeに関連付けられているので、そのまま.pyファイルを指定して実行可能です。以下のようなアクションリストとPythonスクリプトを組むことで、同様の処理を行うことができます。(以下はPython 3.6以上での例)

Hello.py

import pathlib
key = input('Hello World!')

p_text = pathlib.Path('C:\Automation-Anywhere-Temporary-Folder\result.txt')
p_text.write_text('Done')

はじめに

2019年10月にリリースを予定しているPython3.8で新たに加わる変更をPython3.8の新機能 (まとめ)という記事でまとめ始めています。分量のある変更は別記事にしていて、これはその第3弾のPickleの改良についての解説です。

Pickleとは

PickleはPythonのデータシリアライゼーション(データ直列化)のライブラリです。Pythonで扱っているデータをファイルなどに書き出しておきたい場合に、データ構造や型の情報を含めて記録し、読み込み時にはそのファイルにある情報のみで再構成をできるようにする仕組みです。JSONなどとも用途が似通っていますが、Pickleは以下のような特徴を持ちます。

• バイナリフォーマットなのでJSONやXMLのようなテキストフォーマットよりコンパクト
• 一般的な型だけでなくユーザ定義のクラスのインスタンスの直列化も可能
• 標準ライブラリに含まれているので追加のパッケージインストールが不要
• Python専用のフォーマットなので他の言語へのデータ受け渡しには不向き

Python 3.8ではこのPickleにいくつかの変更が加えられています

3.8での変更点

大きく2つの変更点があります。

1. デフォルトのプロトコルバージョンが4になった
2. プロトコルバージョン5が追加され、out-of-bandデータのサポートが追加された

Pickleは1995年に最初に設計されて以来、これまでも何度か拡張されていてその度にプロトコルのバージョンが上がっています。これまでの最新プロトコルバージョンはPython3.4で導入された4でしたが、デフォルトのプロトコルバージョンはPython3.0で導入された3でした。これが、Python3.8からデフォルトのPickleプロトコルバージョンが4になり性能の改善が期待できます。バージョン4の詳細はPEP 3154にあります。

2つ目のOut-of-bandデータのサポートですが、これはPEP 574で提案された変更によるものです。ちょっと詳しく次の章で見てみたいと思います。

なぜOut-of-bandデータが必要か？

PEP-574によると、Pickleは最近ではファイルへの書き出しというよりも(それらにはより汎用なJSONなどのフォーマットが使われる)、Pythonのプロセス間でのデータのやり取りに使われているようです。そして、データ解析などの応用ではとても大きなデータをPythonで扱う事が増えていて、それらをpickleでシリアライズする時に別手段で(out-of-band)で引きた渡したいという要求があります。

Out-of-bandデータは元々はネットワーク用語で、通常経路(in-band)とは別の経路(わき道)でデータ転送することです。データ経路が詰まっている時でも緊急度の高いコマンドなどを送れるようにする為に使われてました。経路は物理的に別れている場合と単に論理的な分離である場合とがありますが、アプリからは2つの経路があるように見える点では一緒です。そしてPickleの場合、Out-of-bandはシリアライズデータの中に入れずに別の手段での受け渡しになるわけですが、なぜそれが必要なのか？ それはできるだけコピーを避けるためです。

データを含めてin-bandでシリアライズ(dump)するとまずその全てを保持するためのバッファを用意してそこにデータをコピーすることになります。そしてそれを受け取った側でもそのシリアライズデータをデシリアライズ(load)する時にまたコピーが発生します。データが小さければそれほど大きな差にはなりませんが、数MBを超えるデータとなると話が変わってきます。それをOut-of-bandでシリアライズデータから切り離して引き渡すことでコピーの回数を減らせる可能性が出てきます。データを値ではなくポインター or 参照で引き渡すイメージですね。

Out-of-bandデータの引き渡し方

まずは比較のためにProtocol Version 4の場合を見ます。'abc'を100回繰り返すバイト列をシリアライズしてみます。

pickle_test_4.py

import pickle
import pickletools

data = bytearray('abc' * 100, 'ascii')
pickled = pickletools.optimize(pickle.dumps(data, protocol=4))
print("Serialized:", pickled)
print("Deserialized:", pickle.loads(pickled))

結果はこうなります。

Serialized: b'\x80\x04\x95J\x01\x00\x00\x00\x00\x00\x00\x8c\x08builtins\x8c\tbytearray\x93B,\x01\x00\x00abcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabc\x85R.'
Deserialized: bytearray(b'abcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabc')

pickle.dumpsでバイト列にシリアライズされたデータがpickle.loadsで元に戻されている事がわかります。シリアライズされたデータの中身が良くわからないのでpickletools.dis(pickled)で逆アセンブル(？)してみます。

    0: \x80 PROTO      4
    2: \x95 FRAME      330
   11: \x8c SHORT_BINUNICODE 'builtins'
   21: \x8c SHORT_BINUNICODE 'bytearray'
   32: \x93 STACK_GLOBAL
   33: B    BINBYTES   b'abcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabc'
  338: \x85 TUPLE1
  339: R    REDUCE
  340: .    STOP
highest protocol among opcodes = 4

ほぼコード通りですが、組み込みのbytearrayを使って初期値b'abcabc...'からバイト列を生成するというという風になっている様です。Protocol Version 4までは、この初期値の部分がどんなに大きくてもここに格納することになり、シリアライズしたデータはその分大きくなってしまいます。そして生成の際にその分無駄なコピーが発生してしまいます。

これをProtocol Version 5の機能を使ってout-fo-band転送してみます。変更点は以下の3つ。

• out-of-band転送したいデータを含むクラスに__reduce_ex__メソッドを追加する。
• pickle.dumpsで buffer_callbackにコールバック関数を指定する。
• pickle.loadsで buffersに引き渡しのためのバッファを指定する。

これだけだと良くわからないので、サンプルのコードで説明します。上記の例は Out-of-band転送する場合にはこの様になります。

pickle_test_5.py

import pickle
import pickletools

class barray(bytearray):
    def __reduce_ex__(self, protocol):
        return type(self), (pickle.PickleBuffer(self),)

buffers = []

def buf_cb(pickle_buffer):
    data = pickle_buffer.raw().obj
    buffers.append(data)

data = barray('abc' * 100, 'ascii')
pickled = pickletools.optimize(pickle.dumps(
    data, protocol=5,  buffer_callback=buf_cb))
print("Serialized:", pickled)
print("Deserialized:", pickle.loads(pickled, buffers=buffers))

pickletools.dis(pickled)

まずひとつ目の__reduce_ex__メソッドですが、それぞれの型をどの様にシリアライズしたいかを定義できて、返り値として、2要素以上のタプルを返します。1要素目がそのデータを再構成する時に呼ばれるオブジェクト(通常はクラスオブジェクト)で、2要素目がその呼び出し引数(初期値)です。通常はデータそのものを2つ目の要素として指定しますが、Out-of-bandにしたい時はそれをpickle.PickleBuffer型にして返します。

なお、bytearrayは組み込みの型で直接メソッドを追加できないので、ここの例ではそれを基底クラスとする新しいクラス barrayを作り、そこに__reduce_ex__メソッドを追加しています。

そして2つ目の変更点がdumps()のbuffer_callback引数で、ここにPickleBufferを引数として取るコールバック関数を指定します。このコールバック関数はdumps()がデータをシリアライズする過程でPickleBufferに遭遇する度に呼ばれますが、ここの例ではOut-of-band転送したいデータを buffersという配列に追加していっています。

で、3つ目の変更点のloads()のbuffers引数。ここにdumps()のコールバック関数で設定されたbuffersという配列を指定します。それを理解するには上記のコードでどのようなシリアライズデータが作成されるのかを見るのが早いと思います。

実行結果はこんな風になります。

Serialized: b'\x80\x05\x95\x17\x00\x00\x00\x00\x00\x00\x00\x8c\x08__main__\x8c\x06barray\x93\x97\x85R.'
Deserialized: barray(b'abcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabcabc')
    0: \x80 PROTO      5
    2: \x95 FRAME      23
   11: \x8c SHORT_BINUNICODE '__main__'
   21: \x8c SHORT_BINUNICODE 'barray'
   29: \x93 STACK_GLOBAL
   30: \x97 NEXT_BUFFER
   31: \x85 TUPLE1
   32: R    REDUCE
   33: .    STOP
highest protocol among opcodes = 5

まずシリアライズされたデータのサイズが全然違います(Version 4が341バイトだったのに対してVersion 5では34バイト)。上記の逆アセンブルされたシリアライズデータを見るとわかると思いますが、初期化データはここには入っていません。代わりにNEXT_BUFFERというopcodeが見えます。loads()はこのopcodeに遭遇する度に引数で与えられた配列からデータを一つずつ取ってきて使います。上手くやればコピーなしでデータの引き渡しができるでしょう。

まとめ

これまであまりPickleって使ったことがなかったのですが、今回の変更をきっかけに少し理解が深まりました。ファイルに書き出しておくだけならJSONとかの方が汎用性もあって便利だろうなと思っていましたが、Python間でのデータの受け渡しに、ユーザ定義の型情報も含んだ形で行えるのはちょっと便利かも知れません。今後、なにか作るときの選択肢の一つとして考えたいと思います。

tl;dr

• 管理ソースが多くなるにつれていちいちコンソールで編集やアップロードが面倒になったのでCIからデプロイで自動化したくなった
• LambdaへのデプロイをLambdaにやらせるのが手っ取り早そう
• CI(zip化してS3デプロイ) -> S3 -> イベントトリガーでLambda起動(デプロイ) -> Lambda関数の更新

用意するもの

• CI（よしなに）
• zip放置用のS3
• デプロイ実行用のLambda関数
• デプロイされるLambda関数

CI（よしなに）

• 好きなものをどうぞ
• ビルド -> zip化 -> S3までデプロイする感じで

import boto3
import glob
import os
import shutil
import mimetypes
from datetime import datetime
from datetime import timedelta
import traceback
import json

BUILD_DIR = "YOUR_BUILD_DIR"

LAMBDA_PROJECT_LIST_JSON = "./YOUR_PROJECT.json"

AWS_ACCESS_KEY = "YOUR_AWS_ACCESS_KEY"
AWS_ACCESS_SECRET = "YOUR_AWS_ACCESS_SECRET"

AWS_S3_BUCKET = "YOUR_AWS_S3_BUCKET"
AWS_REGION = "YOUR_AWS_REGION"

def run():
  # YOUR_BUILD_DIRディレクトリ削除
  # YOUR_BUILD_DIRディレクトリ作成
  # zipファイル作ってYOUR_BUILD_DIRに突っ込む
  # s3デプロイ

  with open(LAMBDA_PROJECT_LIST_JSON,"r") as l:
    lambda_project_list = json.load(l)

    try:
      if (AWS_ACCESS_KEY in os.environ)==False or (AWS_ACCESS_SECRET in os.environ)==False:
        raise "undefined access_key"
    except:
      traceback.print_exception()
      return

    session = boto3.session.Session(
                aws_access_key_id=os.environ[AWS_ACCESS_KEY],
                aws_secret_access_key=os.environ[AWS_ACCESS_SECRET],
                region_name=AWS_REGION,
              )

    client = session.client('s3')

    shutil.rmtree(BUILD_DIR, ignore_errors=True)
    os.makedirs(BUILD_DIR, exist_ok=True)

    keys = lambda_project_list.keys()

    for keyname in keys:
      val = lambda_project_list[keyname]
      shutil.make_archive(BUILD_DIR+"/"+val["lambda"], 'zip', root_dir="./"+keyname)

    file_list = glob.glob(BUILD_DIR+"/*.zip")

    now = datetime.today()
    expires = now+timedelta(hours=1)

    print("DEPLOY:"+AWS_REGION)

    for val in file_list:
      mimetype = mimetypes.guess_type(val)[0]
      fileneme = os.path.basename(val)
      with open(val,"rb") as f:
        try:
          response = client.put_object(
            ACL='private',
            Body=f,
            Key=fileneme,
            Bucket=AWS_S3_BUCKET,
            ContentType=mimetype,
            Expires=expires,
          )
          print("UPLOAD:"+fileneme)
        except:
          traceback.print_exception()
          break

ビルドははしょり（懺悔）subprocess使えばいいんじゃなかろうか（適当）

デプロイ実行用のLambda関数

IAMはよしなに付与しましょう

import sys
import boto3
import os

AWS_REGION = "YOUR_AWS_REGION"

def lambda_handler(event, context):

    # s3に上がったzipを取得
    # lambdaへデプロイ
    if len(event["Records"])==0:
      return

    s3_data = event["Records"][0]["s3"]

    _,ext = os.path.splitext(s3_data["object"]["key"])

    # zip以外にアップロードされたものは除外
    if ext != ".zip":
      return

    function_name = s3_data["object"]["key"].split(".")[0]

    session = boto3.session.Session(
                region_name=AWS_REGION,
              )

    client = session.client("lambda")

    response = client.update_function_code(
      FunctionName=function_name,
      S3Bucket=s3_data["bucket"]["name"],
      S3Key=s3_data["object"]["key"],
      DryRun=False,
    )

    print(response,file=sys.stderr)

DryRun=False は必須。

https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/lambda.html#Lambda.Client.update_function_code

DryRun (boolean) -- Set to true to validate the request parameters and access permissions without modifying the function code.

zip放置用のS3

• よしなに作成。
• リージョンは デプロイ実行用のLambda関数 と合わせた方が楽。
• バケットの プロパティ > Events から PUTで デプロイ実行用のLambda関数 が起動するように設定

デプロイされるLambda関数

• よしなに指定。
• リージョンは合わせる
• LAMBDA_PROJECT_LIST_JSON に放り込んでいく

課題

• 雑に作ったのでビルドやってないので後日作る（といいなあ）

参考

https://dev.classmethod.jp/cloud/aws/2018-solo-boto3-advent-calendar-day14/
https://qiita.com/asahi0301/items/a88760dd38fe0f5fcee1

TL;DR

• Xilinx ISEがインストールされたdockerイメージを作成しました。
• Mac上から自動でビルドする環境を整備し、テンプレートとして公開しました。

はじめに

FPGAを販売するXilinx社では、ISE（最近のデバイス用にはVivado）と呼ばれる自社製の開発環境を用意しています。FPGAのコンフィグレーションに必要なBit stream file（*.bit）はこのISEを使って生成できますが、逆に言えばXilinx社謹製のISEを使わなければ生成できません。そして、ISEは対応OSがWindowsとLinuxのみなんですよね...見事にMacがハブれれてしまいました。メインで使っている端末がMacということもあり、MacからFPGAの開発ができると嬉しいな...と思いしばらくいじっていたので、備忘録を兼ねて記事にします。

以前の開発環境

一年ほど前にFPGAボードの開発をしていた際には、次のような環境で開発していました。

• ターゲットデバイス
  • Papilio pro (Gadget Factory製)
• 開発&シミュレーション
  • Mac端末
  • 言語: VHDL
  • エディタ: Visual Studio Code
  • テスト用コンパイラ: ghdl
• 実装
  • Windows端末
  • エディタ&コンパイラ: Xilinx ISE

状況としてはMacの方でVHDLコードを書き、ghdlでコンパイルして挙動を確認します。だいたいいい感じまでできたらソースコードをDropboxなりGithubなりにアップロードして、Windows端末にダウンロードし、Xilinx ISEで作成したプロジェクトにソースコードを加えて、実装用のBit stream fileを生成していました。
なぜこのような面倒な手順を踏んでいたかというと、Windows端末が研究室の端末で、自宅に持ち帰りが難しかったこと、加えてXilinx ISE上でのソースコードの編集がお世辞にもやりやすいとは言えなかった（補完やSyntax Highlightが貧弱だった）ため、ソースの編集はVS Code上でやりたかったという事情があります。Mac上でVMを使うことも考えたのですが、Mac上のストレージの容量が少なく、VMを立てたくもない状況でした。
そんなこんなで、このしち面倒臭い環境でしばらく開発をしており、そのときの目的が達成された後には、もう金輪際FPGAに関わることもないだろうと思っていたわけです。ただ、幸か不幸かまたFPGAに触る機会ができたので、この際Macで実装までできるようにしてやろうと思いまして、環境構築に取り組みました。

Mac上でのFPGA開発環境の構築

あらすじは次の通りです。

1. Docker for MacでXilinx ISEインストール済みのdockerイメージを作成
2. Pythonのinvokeパッケージを使ってコンパイル、テスト環境の構築

以下の手順では次のソフトウェアがインストールされていることを前提にしています。

• Docker for Mac
• Python 3.x
• XQuartz
• invoke (Python package)
• ghdl

docker、XQuartz、Python3、ghdlはHomebrewから、invokeはAnaconda経由で入手しました。invokeはpipでも入手できるようです。

1. Docker for MacでXilinx ISEインストール済みのdockerイメージを作成

~/~/~/~/~/~/~/~/~/~/~/~/~/~/~/~/~/~/~/~/~/~/~/~/~/~/~/~/~/~/~
以下の手順でISE Webpack editionをインストールしたものはDocker Hub上でseekworser/ise_webpackで公開しているため、Webpack版であれば

docker pull seekworser/ise_webpack

で事足ります。容量が30 GB以上あるので注意してください。
~/~/~/~/~/~/~/~/~/~/~/~/~/~/~/~/~/~/~/~/~/~/~/~/~/~/~/~/~/~/~

dockerを使ってXilinx Vivadoをインストールしている方がいらっしゃったので、その記事を参考にしました。
Xilinx の開発ツールを Docker コンテナに閉じ込める

この方はVivado HLx 2018.2: All OS installer Single-File Downloadをダウンロードしてきて、インストールまで含めてDockerfileで自動化しているのですが、ISEではSingle file Downloadが存在しないため、泣く泣くインストールは手動で行います。
そこで、まずfull installer for Linuxをダウンロードしてきて、次のようなDockerfileを作成しました。

Dockerfile

FROM ubuntu

RUN \
  sed -i -e "s%http://[^ ]\+%http://ftp.jaist.ac.jp/pub/Linux/ubuntu/%g" /etc/apt/sources.list && \
  apt update && \
  apt upgrade -y && \
  apt -y --no-install-recommends install \
  ca-certificates curl sudo xorg dbus dbus-x11 ubuntu-gnome-default-settings gtk2-engines \
  ttf-ubuntu-font-family fonts-ubuntu-font-family-console fonts-droid-fallback lxappearance && \
  apt-get autoclean && \
  apt-get autoremove && \
  rm -rf /var/lib/apt/lists/* && \
  echo "%sudo ALL=(ALL) NOPASSWD: ALL" >> /etc/sudoers

COPY Xilinx_ISE_DS_Lin_14.7_1015_1.tar /

RUN mkdir /ise-installer &&\
  tar --strip-components 1 -C /ise-installer \
  -xvf Xilinx_ISE_DS_Lin_14.7_1015_1.tar &&\
  rm /Xilinx_ISE_DS_Lin_14.7_1015_1.tar

前半は先のリンクの方から丸パクリで、X11環境のインストールその他です。後半でインストール用のフォルダを/ise_installer下に展開しておきます。（ファイル名はバージョンに合わせて変えてください）

ここからGUIインストーラを立ち上げ、ISEをインストールします。Docker for MacでX11 GUIを立ち上げる際には次を参考にしました。
Mac+dockerでx11アプリケーションを起動する

先ほどのDockerfile、ダウンロードしたインストーラを全て同じフォルダに入れ、さらにhost側でXQuartzを立ち上げた状態で以下のコマンドを実行します。

docker build --no-cache --rm -t ise-installer .
xhost + 127.0.0.1
docker run -e DISPLAY=docker.for.mac.localhost:0 --name ise-installed ise-installer /ise-installer/xsetup

その後XQuartzでインストーラが起動するため、指示にしたがってインストールを行います。インストール後に、

(host)$ docker commit ise-installed ise-installed
(host)$ docker run -i --name ise ise-installed /bin/bash
(container)$ rm -rf /ise-installer
(container)$ exit
(host)$ docker commit ise ise
(host)$ docker container rm ise-installer ise-installed ise
(host)$ docker image rm ise-installer ise-installed

を順に実行し、ISEのインストールされたイメージiseを作成します。

2. Pythonのinvokeパッケージを使ってコンパイル、テスト環境の構築

先ほど作成したimageと、Pythonのinvokeパッケージを使用して、コンフィギュレーション用のBit stream fileを作成する環境を作成します。seekworser/ise_project_templateでPapilio pro用に作成したものは公開しています。
ディレクトリ構成は次のようになっています。

ise_project_template

.
├── Readme.md
├── log/
├── out/
├── test/
├── src/
│   ├── project.yaml
│   ├── sample.vhd
│   ├── sample_ucf.ucf
│   └── test_sample.vhd
├── sample.bit
└── tasks.py

logファイルの出力、中間ファイルの出力、ソースコード保管、test用のファイルの出力用のディレクトリとinvoke用のPythonスクリプトが置いてあります。

sample.vhdがメインのVHDLファイルです。

sample.vhd

entity sample is
    -- write ports
end sample;
architecture behavior of sample is
    -- write behavior
end behavior;

また、sample_ucf.ucfはFPGAの制約ファイルです。
プロジェクトが変わるごとにMakefileもといtasks.pyをいじるのは嫌だったので、project.yamlにコンパイル用のパラメータを格納しておくようにします。ついでに使うFPGAの情報もpart:以下に書いておきます。

project.yaml

src_files:
  - file_name: sample.vhd
    language: vhdl
    library: work
    entity_names:
      - sample 

test_files:
  - file_name: test_sample.vhd
    entity_names:
      - test_sample
      - test_sample2

ucf_file: sample_ucf.ucf

top_module: sample

part:
  device: xc6slx9
  package: tqg144
  speed: -2

tasks.pyではこの情報を読み取って、ビルドやテストなどを行うようコードを書いておきます。例えば次のようなイメージです。

tasks.py（一部）

import yaml
import invoke

def get_project_parameter(*keys):
    param = yaml.load(open(PROJECT_YAML_FILE_NAME, "r+"))
    for item in keys:
        param = param[item]
    return param

@invoke.task
def bitgen(c):
    c.run(
        "docker run --rm -i -v $(PWD):/project seekworser/ise_webpack:latest sh <<_EOT_\n"
        "cd /project\n"
        "{ise_base:s}bitgen -w {outdir:s}{top_module:s}.ncd {top_module:s}.bit {outdir:s}{top_module:s}.pcf\n"
        "mv *.drc {logdir:s}{top_module:s}_drc.log\n"
        "mv *.bgn {logdir:s}{top_module:s}_bitgen.log\n"
        "rm -rf *.xwbt _xmsgs xilinx_device_details.xml *.xrpt\n"
        "_EOT_".format(
            ise_base=ISE_BASE,
            top_module=get_project_parameter(TOP_MODULE_KEY),
            outdir=OUTDIR,
            logdir=LOGDIR,
        )
    )
    return

# ~ ~ ~ ~ ~ ~ ~ ~ ~ ~ ~ ~ ~ ~ ~ ~ ~ ~ ~ ~ ~ ~ ~ ~ 
# some other tasks
# ~ ~ ~ ~ ~ ~ ~ ~ ~ ~ ~ ~ ~ ~ ~ ~ ~ ~ ~ ~ ~ ~ ~ ~ 

ISE_BASE = "/opt/Xilinx/14.7/ISE_DS/ISE/bin/lin64/"
OUTDIR = "./out/"
SRCDIR = "./src/"
LOGDIR = "./log/"
PROJECT_YAML_FILE_NAME = SRCDIR + "project.yaml"
TOP_MODULE_KEY = "top_module"

最終的にどうなったか

以上の結果として

invoke builde

のコマンド一発でMacからBit stream fileが生成できるようになりました。ISEの使いにくいエディタを使う必要もなくなったため、開発がよりスムーズに進むようになりました。

前回までのあらすじ

前回の記事では、スクレイピングを使って取得したいユーザーの投稿情報を取得するところまでをまとめました。

今回は、その投稿情報をもとにして投稿の詳細情報を描画するところまでまとめます。
前回の記事はこちら

https://qiita.com/Tokyo/items/a12ef262268c48be8ff1

描画したデータ

以下の25個の項目をグラフで表現しました。

• 各時間ごとの投稿数、いいねの平均/中央値、リツイートの平均/中央値
• 各曜日ごとの投稿数、いいねの平均/中央値、リツイートの平均/中央値
• 文字数ごとの投稿数、いいねの平均/中央値、リツイートの平均/中央値
• 改行数ごとの投稿数、 いいねの平均/中央値、リツイートの平均/中央値
• メディア数ごとの投稿数、 いいねの平均/中央値、リツイートの平均/中央値

環境は、Jupyter notebook上で行いました。

スクリプトについて

簡単な統計情報

まずは小手調べに簡単な統計情報を出してみました。

平均値と中央値のいずれもを出している理由は、私のツイート数が少なすぎて何度かバズった外れ値の影響を考慮したからです。
pandasで取得するデータは前回の記事で紹介した方法で取得済みの設定です。

easy.py

import pandas as pd
import matplotlib.pyplot as plot

df = pd.read_csv('otnk23.csv',encoding='utf-8')
#いいねの平均
print(df['likes'].mean())
#いいねの中央値
print(df['likes'].median())
#リツイートの平均
print(df['retweets'].mean())
#リツイートの中央値
print(df['retweets'].median())

時間帯ごとの分析

time.py

#時間帯ごとの投稿数

df['hour'] = df['date'].dt.hour
hour_size =df.groupby(pd.Grouper(key='hour' )).size().sort_index()
hour_size.plot.bar()
plot.title('hour_posts_size')
plot

#時間ごとのいいねの中央値
likes_hour_groupby_median = df.groupby('hour')[['likes']].median()
likes_hour_groupby_median.plot.bar()
plot.title('hour_likes_median')
plot

#時間ごとのいいねの平均
likes_hour_groupby_mean = df.groupby('hour')[['likes']].mean()
likes_hour_groupby_mean.plot.bar()
plot.title('hour_likes_mean')
plot


#時間ごとのリツイートの中央値
retweets_hour_groupby_median = df.groupby('hour')[['retweets']].median()
retweets_hour_groupby_median.plot.bar()
plot.title('hour_retweets_median')
plot

#時間ごとのリツイートの平均
retweets_hour_groupby_mean = df.groupby('hour')[['retweets']].mean()
retweets_hour_groupby_mean.plot.bar()
plot.title('hour_retweets_mean')
plot

曜日ごとの分析

day.py

#曜日ごとの投稿数

df['day_of_week'] = df['date'].dt.day_name()
cats = [ 'Monday', 'Tuesday', 'Wednesday', 'Thursday', 'Friday', 'Saturday', 'Sunday']


day_size =df.groupby(pd.Grouper(key='day_of_week' )).size().reindex(cats)
day_size.plot.bar()
plot.title('day_posts_size')
plot

#曜日ごとのいいねの中央値
likes_day_groupby_median = df.groupby('day_of_week')[['likes']].median().reindex(cats)
likes_day_groupby_median.plot.bar()
plot.title('day_likes_median')
plot

#曜日ごとのいいねの平均
likes_day_groupby_mean = df.groupby('day_of_week')[['likes']].mean().reindex(cats)
likes_day_groupby_mean.plot.bar()
plot.title('day_lkes_mean')
plot


#曜日ごとのリツイートの中央値
retweets_day_groupby_median = df.groupby('day_of_week')[['retweets']].median().reindex(cats)
retweets_day_groupby_median.plot.bar()
plot.title('day_retweets_median')
plot

#曜日ごとのリツイートの平均
retweets_day_groupby_mean = df.groupby('day_of_week')[['retweets']].mean().reindex(cats)
retweets_day_groupby_mean.plot.bar()
plot.title('day_retweets_mean')
plot

メディア数ごとの分析

media.py

#メディア数ごとのサイズ
size_media_groupby_median = df.groupby('media').size()
size_media_groupby_median.plot.bar()
plot.title('media_num_size')
plot


#メディア数ごとのいいねの中央値
likes_media_groupby_median = df.groupby('media')[['likes']].median()
likes_media_groupby_median.plot.bar()
plot.title('media_likes_median')
plot

#メディア数ごとのいいねの平均値
likes_media_groupby_mean = df.groupby('media')[['likes']].mean()
likes_media_groupby_mean.plot.bar()
plot.title('media_likes_mean')
plot

#メディア数ごとのリツイートの中央値
retweets_media_groupby_median = df.groupby('media')[['retweets']].median()
retweets_media_groupby_median.plot.bar()
plot.title('media_retweets_median')
plot

#メディア数ごとのリツイートの平均値
retweets_media_groupby_mean = df.groupby('media')[['retweets']].mean()
retweets_media_groupby_mean.plot.bar()
plot.title('media_retweets_mean')
plot

改行数ごとの分析

block.py

df_text  = df['text']
kaigyo = []
for text in df_text:
    count = text.count('\n\n')
    kaigyo += [count]
df['kaigyo'] = kaigyo

#改行(スペースの数)ごとのサイズ
likes_kaigyo_groupby_size = df.groupby('kaigyo').size()
likes_kaigyo_groupby_size.plot.bar()
plot.title('block_size')
plot




#改行ごとのいいね数の中央値
likes_kaigyo_groupby_median = df.groupby('kaigyo')[['likes']].median()
likes_kaigyo_groupby_median.plot.bar()
plot.title('block_likes_median')
plot

#改行ごとのいいね数の平均
likes_kaigyo_groupby_mean = df.groupby('kaigyo')[['likes']].mean()
likes_kaigyo_groupby_mean.plot.bar()
plot.title('block_likes_mean')
plot


#改行ごとのリツイート数の中央値
retweets_kaigyo_groupby_median = df.groupby('kaigyo')[['retweets']].median()
retweets_kaigyo_groupby_median.plot.bar()
plot.title('block_retweets_median')
plot

#改行ごとのリツイート数の平均
retweets_kaigyo_groupby_mean = df.groupby('kaigyo')[['retweets']].mean()
retweets_kaigyo_groupby_mean.plot.bar()
plot.title('block_retweets_mean')
plot

文字数ごとの分析

text_num.py

df_text_num_about = []
df_text_num = df['text_num']
for text in df_text_num:
    df_text_num_about.append(text//10*10)
df['about_num'] = df_text_num_about 

#文字数ごとのサイズ
about_num_groupby_size = df.groupby('about_num').size()
about_num_groupby_size.plot.bar()
plot.title('text_size')
plot

#文字数ごとのいいね数の中央値
likes_about_num_groupby_median = df.groupby('about_num')[['likes']].median()
likes_about_num_groupby_median.plot.bar()
plot.title('text_size_likes_median')
plot


#文字数ごとのいいね数の平均値
likes_about_num_groupby_mean = df.groupby('about_num')[['likes']].mean()
likes_about_num_groupby_mean.plot.bar()
plot.title('text_size_likes_mean')
plot


#文字数ごとのリツイート数の中央値
retweets_about_num_groupby_median = df.groupby('about_num')[['retweets']].mean()
retweets_about_num_groupby_median.plot.bar()
plot.title('text_size_retweets_median')
plot


#文字数ごとのリツイート数の平均値
retweets_about_num_groupby_mean = df.groupby('about_num')[['retweets']].mean()
retweets_about_num_groupby_mean.plot.bar()
plot.title('text_size_retweets_mean')
plot

感想と結果

せっかくなので、私の投稿の分析結果も載せておきます。

これは、文字数ごとの分析、改行数ごとの分析、メディア数ごとの分析の結果です。

テキスト数に見るからに文字数が多いほうがバズりやそうですね。とはいえ、母数が少ないので何とも言えませんが。。。

それから、改行も0の場合はあまり伸びていませんね。これは中央値、平均値共にその傾向があるので、一定の有意性があるのかもしれません。

これは、曜日ごとの分析と時間ごとの分析です。

曜日に関しては、日曜日の平均が高いのでバズった投稿が集中しているのかもしれません。また時間に関しても、夜になるにつれエンゲージメントが伸びています。

私の投稿数とフォロワーが少ないので伸びにくい傾向が出てしまいましたが、時々気になるデータが出てきているようです。

皆さんも自分のアカウントや気になるアカウントで試してみてください。

製作の動機

• ブラッドピット主演の映画「Fury」の戦車戦闘シーンを再現したい。
• 趣味のディスプレイ用プラモデルをraspberry piで動かしたい。
• ラジコン、電子回路等は未経験だけど、加工は得意なのでなんとかなりそう。
• GoogleのEdge TPUで物体検出すれば砲塔回転方向を割り出せるはず。
• 車体はとりあえずライントレースでまずは作ってみよう。

貫いたこだわり

• スペースに余裕のある1/16スケールは高価でかつ車種も少ないので、安価で車種も豊富だけどスペースに全く余裕のない1/35スケールでなんとかした。
• 今後のリアルなスケールモデル製作（塗装、ディティールアップ等）に繋げるために、Edge TPUのUSBケーブルの一部とカメラレンズ以外は全て車体内部および車体底面に配置し、「いかにも手作りラズパイマシン」な外観を徹底的に回避。
• 外観を損ねないために、航続距離を犠牲にしてでも絶対に単4電池の分散配置で動かす。
• 安全面を考慮してリチウムイオンバッテリーは使わない。

動画では、とりあえずデモとしてマウスを認識するかを見ています。
(動画中での砲塔上面の電池はただの重りで、現在はとれています。)
物体検出には、バナナロボのコードを一部改変させていただいたものを利用しています。
project-banana-robo
ライントレースは快調ですが、砲塔の指向先の精度が...。

コンセント電源で固定した戦車で視野をディスプレイ表示（バナナロボのコードのデモ）

紆余曲折の末、とりあえず1ヶ月でそれなりにはなったかな、と。今後はさらにEdge TPUのモデルを再学習し、敵のタイガー戦車を認識させ、クロマキー合成で特撮するのが究極的な目標です。

先は長い...。

外観

砲塔上部の黒い突起物がスパイカメラのレンズ部分です。

カメラはネジとナットでとりあえず留めています。

砲塔内部には、サーボモーターの動力を噛み合わせるためにコの字型に合わせたプラ板と、釣具店に売っている重りを適宜配置。

お手製の歯車を、

車体の砲塔リングと砲塔回転用サーボモーターに合わせます。

天板を外した中身の配置は以下の通り。
- 車体前面に収まるように加工したギアボックス
- ミニ四駆用の馬力のあるタイプの走行モーター（トルクチューン2）
- その直上に、大電流を要する走行モーターを制御できるDCモータードライバー（黒色基盤）
- 更にその直上に、raspberry pi電源用に5Vに昇圧できるコンバータ（青色基盤）
- 走行モーター駆動用の単4電池を車体左右、raspberry pi直下、車体外底部に直列で計4つ
- 車体後部に見える白いブレッドボードは、左右のライントレース用赤外線センサーの配線をまとめているもので、可変抵抗器（青色）で黒のテープと白の画用紙を区別するセンサー感度を調節しています。

raspberry piの下には、
- オレンジ色のトルクチューンモーターの後部
- 走行モーター用の単4電池のうちの一つ
- Edge TPU本体から伸びるケーブル
- それをraspberry piに繋ぐ短い自作ケーブル
- 手前に埋もれている砲塔回転用サーボモーター
- サーボモーターの後ろに、ライントレース用の電源配線をまとめているブレッドボード
が格納されています。

車体底面に赤外線センサー、電池、Edge TPUを配置。車体の左右に見える小さなボックスのようなものがライントレース用の赤外線センサーです。車体前面の単4電池2つがraspberry pi本体への電源供給用で、これでEdge TPUと砲塔回転用サーボモーターへの電流もまかないます。

電子回路全体図

模式的にFritzingで作成した回路図です。
電子回路もFritzingも初めてなので、配線には間違いや無駄があるかもしれません。
実際にはブレッドボードは車体に格納するために切り詰めています。
可変抵抗は500kΩ前後に調整することで、ラインの白黒を判別できました。
なお、モータードライバーの該当アイコンがFritzingに見つからなかったので、この模式図では代用品を利用しています。

使用したパーツ（主要部品のみ）

メイン

ブラピ搭乗のM4A3E8戦車
raspberry pi zero wh スターターキット
PowerBoost 1000 Basic - 5V USB Boost（昇圧コンバータ）

車体制御系

トルクチューン2モーター
ツインモーターギアボックス
Cytron 4V～16V DCモータードライバ（2チャンネル、連続3A）
赤外線センサー（フォトリフレクタ）
100Ω抵抗
可変抵抗器（1MΩ）
電圧・電流計測用のテスタ

砲塔制御系

サーボモーター(SG90)
Raspberry Pi Zero用スパイカメラ
Google Coral Edge TPU USB Accelerator
20cmのUSB type A - type C 通信ケーブル（型番は失念）
USB(microB)コネクタ
USB2.0コネクタ
シールドスリムロボットケーブル
ダイソーのグルーガンとグルースティック

その他

秋月電子通商、千石電商、Amazon、SWITCH SCIENCE、タミヤには大変お世話になりました。上記以外にも数多の電子パーツ、ホビー用品、工具等を用いています。

制御の流れ

1. ライントレースの要領で走行用DCモーターを制御
2. 走行中に砲塔上面のカメラとEdge TPUで対象物を物体検出
3. 検出した対象物の方向を計算
4. 2.の計算の間に車体が進んだ記録を元に、対象物の方向のズレを粗く予測
5. 3.と4.を加味した方向に砲塔旋回用サーボモーターを制御
6. 1.と並行して2.から5.の繰り返し

コード

物体検出には、バナナロボのコードを一部改変させていただいたものを利用しています。
project-banana-robo
冒頭のGIF動画の通り、砲塔の回転先の精度が甘く、まだまだ改良の余地があります。
物体検出には2-3s/frameほどかかるので、その間の車体の動きを"forward", "right", "left"の変数に記録しておき、その補正も加味して砲塔を動かそうとしていますがなかなか...。
何か良い案がありましたらお願いします。

concurrent_motor.py

from edgetpu.detection.engine import DetectionEngine
from PIL import Image
from PIL import ImageDraw
from PIL import ImageFont
import time, io, picamera, pigpio, threading
import numpy as np
import wiringpi as pi

forward = 0
right = 0
left = 0

# https://github.com/waveform80/picamera/issues/383
def _monkey_patch_picamera():
    original_send_buffer = picamera.mmalobj.MMALPortPool.send_buffer

    def silent_send_buffer(zelf, *args, **kwargs):
        try:
            original_send_buffer(zelf, *args, **kwargs)
        except picamera.exc.PiCameraMMALError as error:
            if error.status != 14:
                raise error

    picamera.mmalobj.MMALPortPool.send_buffer = silent_send_buffer


# Read labels.txt file provided by Coral website
def _read_label_file(file_path):
    with open(file_path, 'r', encoding="utf-8") as f:
        lines = f.readlines()
    ret = {}
    for line in lines:
        pair = line.strip().split(maxsplit=1)
        ret[int(pair[0])] = pair[1].strip()
    return ret


# Main loop
def detect_motor():
    global forward, right, left
    model_filename = "mobilenet_ssd_v2_coco_quant_postprocess_edgetpu.tflite"
    label_filename = "coco_labels.txt"
    engine = DetectionEngine(model_filename)
    labels = _read_label_file(label_filename)
    CAMERA_WIDTH = 640
    CAMERA_HEIGHT = 480

    fnt = ImageFont.load_default()

    # set up for servo motor
    servo_pin = 18
    pigp = pigpio.pi()
    pulsewidth = 1450
    pigp.set_servo_pulsewidth(servo_pin, pulsewidth)

    # To view preview on VNC,
    # https://raspberrypi.stackexchange.com/a/74390
    with picamera.PiCamera() as camera:
        _monkey_patch_picamera()
        camera.resolution = (CAMERA_WIDTH, CAMERA_HEIGHT)
        camera.framerate = 15
        camera.rotation = 180
        _, width, height, channels = engine.get_input_tensor_shape()
        print("{}, {}".format(width, height))
        overlay_renderer = None
        camera.vflip = True
        camera.hflip = True
        camera.start_preview()
        try:
            stream = io.BytesIO()
            for foo in camera.capture_continuous(stream,
                                                 format='rgb',
                                                 use_video_port=True):
                # Make Image object from camera stream
                stream.truncate()
                stream.seek(0)
                input = np.frombuffer(stream.getvalue(), dtype=np.uint8)
                input = input.reshape((CAMERA_HEIGHT, CAMERA_WIDTH, 3))
                image = Image.fromarray(input)
                # image.save("out.jpg")

                # Make overlay image plane
                img = Image.new('RGBA',
                                (CAMERA_WIDTH, CAMERA_HEIGHT),
                                (255, 0, 0, 0))
                draw = ImageDraw.Draw(img)
                draw.line((CAMERA_WIDTH//2, 0, CAMERA_WIDTH//2, CAMERA_HEIGHT), width=1)
                draw.line((3*CAMERA_WIDTH//8, 0, 3*CAMERA_WIDTH//8, CAMERA_HEIGHT), width=1)
                draw.line((5*CAMERA_WIDTH//8, 0, 5*CAMERA_WIDTH//8, CAMERA_HEIGHT), width=1)

                # Run detection
                forward = 0
                right = 0
                left = 0
                start_ms = time.time()
                results = engine.DetectWithImage(image,
                                                 threshold=0.2, top_k=5)
                elapsed_ms = (time.time() - start_ms)*1000.0
                obj = None
                if results:
                    obj = next((x for x in results if labels[x.label_id] == "mouse"), None)

                if obj:
                    box = obj.bounding_box.flatten().tolist()
                    box[0] *= CAMERA_WIDTH
                    box[1] *= CAMERA_HEIGHT
                    box[2] *= CAMERA_WIDTH
                    box[3] *= CAMERA_HEIGHT
                    draw.rectangle(box, outline='red')
                    draw.text((box[0], box[1]-10), labels[obj.label_id],
                              font=fnt, fill="red")
                    obj_width = box[2] - box[0]
                    obj_center = box[0] + obj_width // 2
                    draw.point((obj_center, box[1] + (box[3] - box[1])//2))
                    # if object on the right side of the sight
                    if (obj_center - CAMERA_WIDTH // 2) > 0:
                        print("TURN R")
                        mv_deg = 54 * ((obj_center - CAMERA_WIDTH // 2) / CAMERA_WIDTH)
                        # if gunport in the right direction of the vehicle head
                        if pulsewidth <= 1450:
                            mv_deg += int(54 * (forward / 105 + (right-left)/35) * np.sin(((1450-pulsewidth)/950) * (np.pi/2)))
                        # if gunport in the left direction of the vehicle head
                        elif pulsewidth > 1450:
                            mv_deg += int(54 * (-forward / 105 + (right-left)/35) * np.sin(((pulsewidth-1450)/900) * (np.pi/2)))
                        pulsewidth -= mv_deg * 10
                        print("mv_deg {}, pulsewidth {}".format(mv_deg, pulsewidth))
                        pigp.set_servo_pulsewidth(servo_pin, pulsewidth)
                    # if object on the left side of the sight
                    elif (obj_center - CAMERA_WIDTH // 2) < 0:
                        print("TURN L")
                        mv_deg = 54 * ((obj_center - CAMERA_WIDTH // 2) / CAMERA_WIDTH)
                        # if gunport in the right direction of the vehicle head
                        if pulsewidth <= 1450:
                            mv_deg += int(54 * (forward / 105 + (right-left)/35) * np.sin(((1450-pulsewidth)/950) * (np.pi/2)))
                        # if gunport in the left direction of the vehicle head
                        elif pulsewidth > 1450:
                            mv_deg += int(54 * (-forward / 105 + (right-left)/35) * np.sin(((pulsewidth-1450)/900) * (np.pi/2)))
                        pulsewidth -= mv_deg * 10
                        print("mv_deg {}, pulsewidth {}".format(mv_deg, pulsewidth))
                        pigp.set_servo_pulsewidth(servo_pin, pulsewidth)
                    camera.annotate_text = "{0:.2f}ms".format(elapsed_ms)

                if not overlay_renderer:
                    overlay_renderer = camera.add_overlay(
                        img.tobytes(),
                        size=(CAMERA_WIDTH, CAMERA_HEIGHT), layer=4, alpha=255)
                else:
                    overlay_renderer.update(img.tobytes())
        finally:
            if overlay_renderer:
                camera.remove_overlay(overlay_renderer)
            camera.stop_preview()

def sensor_motor():
    # set for IR sensors
    PIR_PIN1 = 23
    PIR_PIN2 = 24

    pi.wiringPiSetupGpio()
    pi.pinMode(PIR_PIN1, pi.INPUT)
    pi.pinMode(PIR_PIN2, pi.INPUT)

    # set for driving motors
    motor_right_pin1 = 16
    motor_right_pin2 = 20
    motor_left_pin1 = 5
    motor_left_pin2 = 6

    pi.pinMode(motor_right_pin1, pi.OUTPUT)
    pi.pinMode(motor_right_pin2, pi.OUTPUT)
    pi.pinMode(motor_left_pin1, pi.OUTPUT)
    pi.pinMode(motor_left_pin2, pi.OUTPUT)

    # set the speed ranges for motors
    pi.softPwmCreate(motor_right_pin1, 0, 100)
    pi.softPwmCreate(motor_right_pin2, 0, 100)
    pi.softPwmCreate(motor_left_pin1, 0, 100)
    pi.softPwmCreate(motor_left_pin2, 0, 100)

    # get the motors speed 0
    pi.softPwmWrite(motor_right_pin1, 0)
    pi.softPwmWrite(motor_right_pin2, 0)
    pi.softPwmWrite(motor_left_pin1, 0)
    pi.softPwmWrite(motor_left_pin2, 0)

    # raise up the speed
    speed = 0
    while ( speed <= 35 ):
        pi.softPwmWrite(motor_right_pin1, speed)
        pi.softPwmWrite(motor_right_pin2, 0)
        pi.softPwmWrite(motor_left_pin1, speed)
        pi.softPwmWrite(motor_left_pin2, 0)

        time.sleep(0.1)
        speed += 5

    # Black line trace
    global forward, right, left
    while True:
        if (pi.digitalRead(PIR_PIN1) == pi.HIGH) & (pi.digitalRead(PIR_PIN2) == pi.HIGH):
            pi.softPwmWrite(motor_right_pin1, speed)
            pi.softPwmWrite(motor_right_pin2, 0)
            pi.softPwmWrite(motor_left_pin1, speed)
            pi.softPwmWrite(motor_left_pin2, 0)
            forward += 1
        elif (pi.digitalRead(PIR_PIN1) == pi.HIGH) & (pi.digitalRead(PIR_PIN2) == pi.LOW):
            pi.softPwmWrite(motor_right_pin1, speed)
            pi.softPwmWrite(motor_right_pin2, 0)
            pi.softPwmWrite(motor_left_pin1, 100)
            pi.softPwmWrite(motor_left_pin2, 100)
            right += 1
        elif (pi.digitalRead(PIR_PIN1) == pi.LOW) & (pi.digitalRead(PIR_PIN2) == pi.HIGH):
            pi.softPwmWrite(motor_right_pin1, 100)
            pi.softPwmWrite(motor_right_pin2, 100)
            pi.softPwmWrite(motor_left_pin1, speed)
            pi.softPwmWrite(motor_left_pin2, 0)
            left += 1
        else:
            while ( speed >= 5 ):
                pi.softPwmWrite(motor_right_pin1, speed)
                pi.softPwmWrite(motor_right_pin2, 0)
                pi.softPwmWrite(motor_left_pin1, speed)
                pi.softPwmWrite(motor_left_pin2, 0)

                time.sleep(0.1)
                speed -= 5
            break

        time.sleep(0.1)

    # Get back
    time.sleep(3)

    while ( speed <= 35 ):
        pi.softPwmWrite(motor_right_pin1, 0)
        pi.softPwmWrite(motor_right_pin2, speed)
        pi.softPwmWrite(motor_left_pin1, 0)
        pi.softPwmWrite(motor_left_pin2, speed)

        time.sleep(0.1)
        speed += 5

    pi.softPwmWrite(motor_right_pin1, 0)
    pi.softPwmWrite(motor_right_pin2, speed)
    pi.softPwmWrite(motor_left_pin1, 0)
    pi.softPwmWrite(motor_left_pin2, speed)
    time.sleep(2)

    # Stop

    while ( speed >= 5 ):
        pi.softPwmWrite(motor_right_pin1, 0)
        pi.softPwmWrite(motor_right_pin2, speed)
        pi.softPwmWrite(motor_left_pin1, 0)
        pi.softPwmWrite(motor_left_pin2, speed)

        time.sleep(0.1)
        speed -= 5

    pi.softPwmWrite(motor_right_pin1, 100)
    pi.softPwmWrite(motor_right_pin2, 100)
    pi.softPwmWrite(motor_left_pin1, 100)
    pi.softPwmWrite(motor_left_pin2, 100)
    time.sleep(3)


if __name__ == "__main__":
    thread_1 = threading.Thread(target=detect_motor)
    thread_1.start()

    time.sleep(20)

    thread_2 = threading.Thread(target=sensor_motor)
    thread_2.start()

直近の今後の課題

• 砲塔回転の精度向上
• 目的の対象物の学習データ生成（webスクレイピングまたは観賞用回転ディスプレイを利用）
• 上記データを用いたモデルの再学習

作ったのは米軍の戦車なので、色々な独軍の戦車を認識するモデルをできれば作りたいなと思っています。ただ、スクレイピングで十分にデータが集まるかどうか。
砲塔天蓋のカメラからは自車の砲身が写るので、そのカメラで回転ディスプレイ上の対象物の連続コマ撮りの方がいいのか。でもそれで強いモデルができるのか。背景はどうしよ。などなど課題が山積していますが地道に続けていきます。

参考文献・リンク

電子部品ごとの制御を学べる! Raspberry Pi 電子工作 実践講座 改訂第2版
1/35 M1スーパーシャーマン RC化 ① : 朴念仁の艱苦
Raspberry Pi Zero用の小型電源とUSB WiFiモジュール用USBコネクタを自作する
project-banana-robo

製作の動機

• ブラッドピット主演の映画「Fury」の戦車戦闘シーンを再現したい。
• 趣味のディスプレイ用プラモデルをraspberry piで動かしたい。
• ラジコン、電子回路等は未経験だけど、加工は得意なのでなんとかなりそう。
• GoogleのEdge TPUで物体検出すれば砲塔回転方向を割り出せるはず。
• 車体はとりあえずライントレースでまずは作ってみよう。

貫いたこだわり

• スペースに余裕のある1/16スケールは高価でかつ車種も少ないので、安価で車種も豊富だけどスペースに全く余裕のない1/35スケールでなんとかした。
• 今後のリアルなスケールモデル製作（塗装、ディティールアップ等）に繋げるために、Edge TPUのUSBケーブルの一部とカメラレンズ以外は全て車体内部および車体底面に配置し、「いかにも手作りラズパイマシン」な外観を徹底的に回避。
• 外観を損ねないために、航続距離を犠牲にしてでも絶対に単4電池の分散配置で動かす。
• 安全面を考慮してリチウムイオンバッテリーは使わない。

動画では、とりあえずデモとしてマウスを認識するかを見ています。
(動画中での砲塔上面の電池はただの重りで、現在はとれています。)
物体検出には、バナナロボのコードを一部改変させていただいたものを利用しています。
project-banana-robo
ライントレースは快調ですが、砲塔の指向先の精度が...。

コンセント電源で固定した戦車で視野をディスプレイ表示（バナナロボのコードのデモ）

紆余曲折の末、とりあえず1ヶ月でそれなりにはなったかな、と。今後はさらにEdge TPUのモデルを再学習し、敵のタイガー戦車を認識させ、クロマキー合成で特撮するのが究極的な目標です。

先は長い...。

外観

砲塔上部の黒い突起物がスパイカメラのレンズ部分です。

カメラはネジとナットでとりあえず留めています。

砲塔内部には、サーボモーターの動力を噛み合わせるためにコの字型に合わせたプラ板と、釣具店に売っている重りを適宜配置。

お手製の歯車を、

車体の砲塔リングと砲塔回転用サーボモーターに合わせます。

天板を外した中身の配置は以下の通り。
- 車体前面に収まるように加工したギアボックス
- ミニ四駆用の馬力のあるタイプの走行モーター（トルクチューン2）
- その直上に、大電流を要する走行モーターを制御できるDCモータードライバー（黒色基盤）
- 更にその直上に、raspberry pi電源用に5Vに昇圧できるコンバータ（青色基盤）
- 走行モーター駆動用の単4電池を車体左右、raspberry pi直下、車体外底部に直列で計4つ
- 車体後部に見える白いブレッドボードは、左右のライントレース用赤外線センサーの配線をまとめているもので、可変抵抗器（青色）で黒のテープと白の画用紙を区別するセンサー感度を調節しています。

raspberry piの下には、
- オレンジ色のトルクチューンモーターの後部
- 走行モーター用の単4電池のうちの一つ
- Edge TPU本体から伸びるケーブル
- それをraspberry piに繋ぐ短い自作ケーブル
- 手前に埋もれている砲塔回転用サーボモーター
- サーボモーターの後ろに、ライントレース用の電源配線をまとめているブレッドボード
が格納されています。

車体底面に赤外線センサー、電池、Edge TPUを配置。車体の左右に見える小さなボックスのようなものがライントレース用の赤外線センサーです。車体前面の単4電池2つがraspberry pi本体への電源供給用で、これでEdge TPUと砲塔回転用サーボモーターへの電流もまかないます。

電子回路全体図

模式的にFritzingで作成した回路図です。
電子回路もFritzingも初めてなので、配線には間違いや無駄があるかもしれません。
実際にはブレッドボードは車体に格納するために切り詰めています。
可変抵抗は500kΩ前後に調整することで、ラインの白黒を判別できました。
なお、モータードライバーの該当アイコンがFritzingに見つからなかったので、この模式図では代用品を利用しています。

使用したパーツ（主要部品のみ）

メイン

ブラピ搭乗のM4A3E8戦車
raspberry pi zero wh スターターキット
PowerBoost 1000 Basic - 5V USB Boost（昇圧コンバータ）

車体制御系

トルクチューン2モーター
ツインモーターギアボックス
Cytron 4V～16V DCモータードライバ（2チャンネル、連続3A）
赤外線センサー（フォトリフレクタ）
100Ω抵抗
可変抵抗器（1MΩ）
電圧・電流計測用のテスタ

砲塔制御系

サーボモーター(SG90)
Raspberry Pi Zero用スパイカメラ
Google Coral Edge TPU USB Accelerator
20cmのUSB type A - type C 通信ケーブル（型番は失念）
USB(microB)コネクタ
USB2.0コネクタ
シールドスリムロボットケーブル
ダイソーのグルーガンとグルースティック

その他

秋月電子通商、千石電商、Amazon、SWITCH SCIENCE、タミヤには大変お世話になりました。上記以外にも数多の電子パーツ、ホビー用品、工具等を用いています。

制御の流れ

1. ライントレースの要領で走行用DCモーターを制御
2. 走行中に砲塔上面のカメラとEdge TPUで対象物を物体検出
3. 検出した対象物の方向を計算
4. 2.の計算の間に車体が進んだ記録を元に、対象物の方向のズレを粗く予測
5. 3.と4.を加味した方向に砲塔旋回用サーボモーターを制御
6. 1.と並行して2.から5.の繰り返し

コード

物体検出には、バナナロボのコードを一部改変させていただいたものを利用しています。
project-banana-robo
冒頭のGIF動画の通り、砲塔の回転先の精度が甘く、まだまだ改良の余地があります。
物体検出には2-3s/frameほどかかるので、その間の車体の動きを"forward", "right", "left"の変数に記録しておき、その補正も加味して砲塔を動かそうとしていますがなかなか...。
何か良い案がありましたらお願いします。

concurrent_motor.py

from edgetpu.detection.engine import DetectionEngine
from PIL import Image
from PIL import ImageDraw
from PIL import ImageFont
import time, io, picamera, pigpio, threading
import numpy as np
import wiringpi as pi

forward = 0
right = 0
left = 0

# https://github.com/waveform80/picamera/issues/383
def _monkey_patch_picamera():
    original_send_buffer = picamera.mmalobj.MMALPortPool.send_buffer

    def silent_send_buffer(zelf, *args, **kwargs):
        try:
            original_send_buffer(zelf, *args, **kwargs)
        except picamera.exc.PiCameraMMALError as error:
            if error.status != 14:
                raise error

    picamera.mmalobj.MMALPortPool.send_buffer = silent_send_buffer


# Read labels.txt file provided by Coral website
def _read_label_file(file_path):
    with open(file_path, 'r', encoding="utf-8") as f:
        lines = f.readlines()
    ret = {}
    for line in lines:
        pair = line.strip().split(maxsplit=1)
        ret[int(pair[0])] = pair[1].strip()
    return ret


# Main loop
def detect_motor():
    global forward, right, left
    model_filename = "mobilenet_ssd_v2_coco_quant_postprocess_edgetpu.tflite"
    label_filename = "coco_labels.txt"
    engine = DetectionEngine(model_filename)
    labels = _read_label_file(label_filename)
    CAMERA_WIDTH = 640
    CAMERA_HEIGHT = 480

    fnt = ImageFont.load_default()

    # set up for servo motor
    servo_pin = 18
    pigp = pigpio.pi()
    pulsewidth = 1450
    pigp.set_servo_pulsewidth(servo_pin, pulsewidth)

    # To view preview on VNC,
    # https://raspberrypi.stackexchange.com/a/74390
    with picamera.PiCamera() as camera:
        _monkey_patch_picamera()
        camera.resolution = (CAMERA_WIDTH, CAMERA_HEIGHT)
        camera.framerate = 15
        camera.rotation = 180
        _, width, height, channels = engine.get_input_tensor_shape()
        print("{}, {}".format(width, height))
        overlay_renderer = None
        camera.vflip = True
        camera.hflip = True
        camera.start_preview()
        try:
            stream = io.BytesIO()
            for foo in camera.capture_continuous(stream,
                                                 format='rgb',
                                                 use_video_port=True):
                # Make Image object from camera stream
                stream.truncate()
                stream.seek(0)
                input = np.frombuffer(stream.getvalue(), dtype=np.uint8)
                input = input.reshape((CAMERA_HEIGHT, CAMERA_WIDTH, 3))
                image = Image.fromarray(input)
                # image.save("out.jpg")

                # Make overlay image plane
                img = Image.new('RGBA',
                                (CAMERA_WIDTH, CAMERA_HEIGHT),
                                (255, 0, 0, 0))
                draw = ImageDraw.Draw(img)
                draw.line((CAMERA_WIDTH//2, 0, CAMERA_WIDTH//2, CAMERA_HEIGHT), width=1)
                draw.line((3*CAMERA_WIDTH//8, 0, 3*CAMERA_WIDTH//8, CAMERA_HEIGHT), width=1)
                draw.line((5*CAMERA_WIDTH//8, 0, 5*CAMERA_WIDTH//8, CAMERA_HEIGHT), width=1)

                # Run detection
                forward = 0
                right = 0
                left = 0
                start_ms = time.time()
                results = engine.DetectWithImage(image,
                                                 threshold=0.2, top_k=5)
                elapsed_ms = (time.time() - start_ms)*1000.0
                obj = None
                if results:
                    obj = next((x for x in results if labels[x.label_id] == "mouse"), None)

                if obj:
                    box = obj.bounding_box.flatten().tolist()
                    box[0] *= CAMERA_WIDTH
                    box[1] *= CAMERA_HEIGHT
                    box[2] *= CAMERA_WIDTH
                    box[3] *= CAMERA_HEIGHT
                    draw.rectangle(box, outline='red')
                    draw.text((box[0], box[1]-10), labels[obj.label_id],
                              font=fnt, fill="red")
                    obj_width = box[2] - box[0]
                    obj_center = box[0] + obj_width // 2
                    draw.point((obj_center, box[1] + (box[3] - box[1])//2))
                    # if object on the right side of the sight
                    if (obj_center - CAMERA_WIDTH // 2) > 0:
                        print("TURN R")
                        mv_deg = 54 * ((obj_center - CAMERA_WIDTH // 2) / CAMERA_WIDTH)
                        # if gunport in the right direction of the vehicle head
                        if pulsewidth <= 1450:
                            mv_deg += int(54 * (forward / 105 + (right-left)/35) * np.sin(((1450-pulsewidth)/950) * (np.pi/2)))
                        # if gunport in the left direction of the vehicle head
                        elif pulsewidth > 1450:
                            mv_deg += int(54 * (-forward / 105 + (right-left)/35) * np.sin(((pulsewidth-1450)/900) * (np.pi/2)))
                        pulsewidth -= mv_deg * 10
                        print("mv_deg {}, pulsewidth {}".format(mv_deg, pulsewidth))
                        pigp.set_servo_pulsewidth(servo_pin, pulsewidth)
                    # if object on the left side of the sight
                    elif (obj_center - CAMERA_WIDTH // 2) < 0:
                        print("TURN L")
                        mv_deg = 54 * ((obj_center - CAMERA_WIDTH // 2) / CAMERA_WIDTH)
                        # if gunport in the right direction of the vehicle head
                        if pulsewidth <= 1450:
                            mv_deg += int(54 * (forward / 105 + (right-left)/35) * np.sin(((1450-pulsewidth)/950) * (np.pi/2)))
                        # if gunport in the left direction of the vehicle head
                        elif pulsewidth > 1450:
                            mv_deg += int(54 * (-forward / 105 + (right-left)/35) * np.sin(((pulsewidth-1450)/900) * (np.pi/2)))
                        pulsewidth -= mv_deg * 10
                        print("mv_deg {}, pulsewidth {}".format(mv_deg, pulsewidth))
                        pigp.set_servo_pulsewidth(servo_pin, pulsewidth)
                    camera.annotate_text = "{0:.2f}ms".format(elapsed_ms)

                if not overlay_renderer:
                    overlay_renderer = camera.add_overlay(
                        img.tobytes(),
                        size=(CAMERA_WIDTH, CAMERA_HEIGHT), layer=4, alpha=255)
                else:
                    overlay_renderer.update(img.tobytes())
        finally:
            if overlay_renderer:
                camera.remove_overlay(overlay_renderer)
            camera.stop_preview()

def sensor_motor():
    # set for IR sensors
    PIR_PIN1 = 23
    PIR_PIN2 = 24

    pi.wiringPiSetupGpio()
    pi.pinMode(PIR_PIN1, pi.INPUT)
    pi.pinMode(PIR_PIN2, pi.INPUT)

    # set for driving motors
    motor_right_pin1 = 16
    motor_right_pin2 = 20
    motor_left_pin1 = 5
    motor_left_pin2 = 6

    pi.pinMode(motor_right_pin1, pi.OUTPUT)
    pi.pinMode(motor_right_pin2, pi.OUTPUT)
    pi.pinMode(motor_left_pin1, pi.OUTPUT)
    pi.pinMode(motor_left_pin2, pi.OUTPUT)

    # set the speed ranges for motors
    pi.softPwmCreate(motor_right_pin1, 0, 100)
    pi.softPwmCreate(motor_right_pin2, 0, 100)
    pi.softPwmCreate(motor_left_pin1, 0, 100)
    pi.softPwmCreate(motor_left_pin2, 0, 100)

    # get the motors speed 0
    pi.softPwmWrite(motor_right_pin1, 0)
    pi.softPwmWrite(motor_right_pin2, 0)
    pi.softPwmWrite(motor_left_pin1, 0)
    pi.softPwmWrite(motor_left_pin2, 0)

    # raise up the speed
    speed = 0
    while ( speed <= 35 ):
        pi.softPwmWrite(motor_right_pin1, speed)
        pi.softPwmWrite(motor_right_pin2, 0)
        pi.softPwmWrite(motor_left_pin1, speed)
        pi.softPwmWrite(motor_left_pin2, 0)

        time.sleep(0.1)
        speed += 5

    # Black line trace
    global forward, right, left
    while True:
        if (pi.digitalRead(PIR_PIN1) == pi.HIGH) & (pi.digitalRead(PIR_PIN2) == pi.HIGH):
            pi.softPwmWrite(motor_right_pin1, speed)
            pi.softPwmWrite(motor_right_pin2, 0)
            pi.softPwmWrite(motor_left_pin1, speed)
            pi.softPwmWrite(motor_left_pin2, 0)
            forward += 1
        elif (pi.digitalRead(PIR_PIN1) == pi.HIGH) & (pi.digitalRead(PIR_PIN2) == pi.LOW):
            pi.softPwmWrite(motor_right_pin1, speed)
            pi.softPwmWrite(motor_right_pin2, 0)
            pi.softPwmWrite(motor_left_pin1, 100)
            pi.softPwmWrite(motor_left_pin2, 100)
            right += 1
        elif (pi.digitalRead(PIR_PIN1) == pi.LOW) & (pi.digitalRead(PIR_PIN2) == pi.HIGH):
            pi.softPwmWrite(motor_right_pin1, 100)
            pi.softPwmWrite(motor_right_pin2, 100)
            pi.softPwmWrite(motor_left_pin1, speed)
            pi.softPwmWrite(motor_left_pin2, 0)
            left += 1
        else:
            while ( speed >= 5 ):
                pi.softPwmWrite(motor_right_pin1, speed)
                pi.softPwmWrite(motor_right_pin2, 0)
                pi.softPwmWrite(motor_left_pin1, speed)
                pi.softPwmWrite(motor_left_pin2, 0)

                time.sleep(0.1)
                speed -= 5
            break

        time.sleep(0.1)

    # Get back
    time.sleep(3)

    while ( speed <= 35 ):
        pi.softPwmWrite(motor_right_pin1, 0)
        pi.softPwmWrite(motor_right_pin2, speed)
        pi.softPwmWrite(motor_left_pin1, 0)
        pi.softPwmWrite(motor_left_pin2, speed)

        time.sleep(0.1)
        speed += 5

    pi.softPwmWrite(motor_right_pin1, 0)
    pi.softPwmWrite(motor_right_pin2, speed)
    pi.softPwmWrite(motor_left_pin1, 0)
    pi.softPwmWrite(motor_left_pin2, speed)
    time.sleep(2)

    # Stop

    while ( speed >= 5 ):
        pi.softPwmWrite(motor_right_pin1, 0)
        pi.softPwmWrite(motor_right_pin2, speed)
        pi.softPwmWrite(motor_left_pin1, 0)
        pi.softPwmWrite(motor_left_pin2, speed)

        time.sleep(0.1)
        speed -= 5

    pi.softPwmWrite(motor_right_pin1, 100)
    pi.softPwmWrite(motor_right_pin2, 100)
    pi.softPwmWrite(motor_left_pin1, 100)
    pi.softPwmWrite(motor_left_pin2, 100)
    time.sleep(3)


if __name__ == "__main__":
    thread_1 = threading.Thread(target=detect_motor)
    thread_1.start()

    time.sleep(20)

    thread_2 = threading.Thread(target=sensor_motor)
    thread_2.start()

直近の今後の課題

• 砲塔回転の精度向上
• 目的の対象物の学習データ生成（webスクレイピングまたは観賞用回転ディスプレイを利用）
• 上記データを用いたモデルの再学習

作ったのは米軍の戦車なので、色々な独軍の戦車を認識するモデルをできれば作りたいなと思っています。ただ、スクレイピングで十分にデータが集まるかどうか。
砲塔天蓋のカメラからは自車の砲身が写るので、そのカメラで回転ディスプレイ上の対象物の連続コマ撮りの方がいいのか。でもそれで強いモデルができるのか。背景はどうしよ。などなど課題が山積していますが地道に続けていきます。

参考文献・リンク

電子部品ごとの制御を学べる! Raspberry Pi 電子工作 実践講座 改訂第2版
1/35 M1スーパーシャーマン RC化 ① : 朴念仁の艱苦
Raspberry Pi Zero用の小型電源とUSB WiFiモジュール用USBコネクタを自作する
project-banana-robo

あらすじ

2019年8月10日。私はVue.jsとDjango REST Frameworkを使い、ある業務システムの画面開発を担当していた。
フレームワークなんて代物はまだ触り始めて２，３か月。とは言えどこを触れば何が変わるか多少は理解してきている自負もあった。
そんな良くも悪くも弾みのついていた私を戒めるかのように、単体テスト中、そいつは突然現れた。

出会い

『更新中の…ステータスが…… エラーコード：409』

「（動作確認では何度も正常に作動していた機能開発...そんなはずは...）」
NG判定とエラー内容をテスト記録として記述する。

"""
HTTP 409 Conflict はリクエストが現在のサーバーの状態と競合したことを示すステータスコード。

競合は PUT メソッドを使用したリクエストのレスポンスで最も発生しやすい。例えば、サーバーにすでに存在しているファイルよりも古いバージョンのファイルをアップロードした際に409の応答が返され、バージョン管理システムの競合が発生する可能性がある。
"""
409 Conflict

困った困った。なけなしの知識と経験であれやこれやと思考した。
そして画面の再描画後、再度そいつは現れた。

『更新中の…ステータスが……　エラーコード：409』

非同期処理の沼

「ここか。」
答えに辿り着くのに長くはかからなかった。

heavyHoge.ts

 // post(url, data, loading)
 // url : 呼び出し先のURL
 // data : REST側に送りたいデータ
 // loading : 処理終了までローディング画面を表示

 // 重ための処理
  private async heavyProcess(): Promise<void> {
    const payload = '画面から送る何かのデータ群';
    this.$http.post(url='hoge-url1', data={'payload': payload}, loading=false);
    this.$http.post(url='hoge-url2', data={'payload': payload}, loading=false);
    // 画面の更新処理
    this.onShown();
  }

ボタンを押下して、この処理が走っている間に他の作業も進められるようにしたい。数十秒～数分なんて流石に待っていられない。
そのような声は当然にあり、この処理は上記の通り非同期処理で実装していた。
ここでの2つのhttp.post(以下、処理①と処理②)は類似処理を行っているが、具体的には下記の処理を行っている。

１．画面内のデータを取得しREST側へPOST
２．取得したデータの条件が合えば、外部APIへ有料の関連データを取得
３．種々のCRUD処理を行う

更にエラーの発生パターンについて調べたところ、大きく下記の2つであった。

１．処理①が上記１～３を終えて、戻り値を返して処理②を行おうとした直後
２．処理①のみor処理②のみが走るように画面側で予め条件設定し、1度目の処理を終えた直後に同じ処理のトリガーを引いた直後

つまり、処理量の多さにCRUD処理が実際には完了しきいっていない状態の時に再び処理①②を行うことで
それを示すように awaitを付記するとloading画面は処理終了まで続くがエラーは消える。（当然っちゃ当然だが）

heavyHoge.ts

 // post(url, data, loading)
 // url :
 // data : 何かのデータ群
 // loading : 処理終了までローディング画面を表示

 // 重ための処理
  private async heavyProcess(): Promise<void> {
    const payload = '画面から送る何かのデータ群';
    await this.$http.post(url='hoge-url1', data={'payload': payload}, loading=false);
    await this.$http.post(url='hoge-url2', data={'payload': payload}, loading=false);
    // 画面の更新処理
    this.onShown();
  }

とにかく、今回のエラーの発生要因やその背景なんかを整理すると
・外部APIが有料ということもあり、連続or多数の本チャンデータ取得パターンの確認を怠った。
・1度のボタン押下で行う非同期処理を２つに分けてしまった。
　（※片方は他画面で使用しているViewに飛んでいた為、このような実装になった）
・REST側で各処理を行うViewを呼び出す為の、共通クラスを用意しなかった。
　（※やってることは2個目と同じ）

しかし、上記の課題を全てクリアして当初の設計通り実装しようとしても、まだエラーの発生パターン２が解決しない。
そう、ユーザーが短時間に連続で同じ処理を行おうとした時だ。完全に非同期処理の沼に嵌まってしまったのである。

2019年8月11日。27歳の誕生日を迎えた今日、未だ解決策は見つかっていない。

To Be Continued

Buttonクラス

インテリジェントブロックボタンに関するクラス

ev3devにおけるボタンの名前

wait_for_bump()やwait_for_pressed()で指定するボタンには以下の6つのいずれかが入る
1. enter
2. up
3. down
4. right
5. left
6. backspace

ボタンが 離れた/押された/押されて離された まで待つ

・wait_for_bump(buttons,timeout_ms=None)

指定されたボタンが指定した時間内に押されて離される（バンプ）まで待つ。
時間内にボタンがバンプされたらTrue，そうでなければFalseを返す。

引数	説明	デフォルト値
buttons(stringもしくはlist)	ボタンの名前(2つ以上の場合はリストで指定)
timeout_ms(float)	処理を中断する待ち時間(ミリ秒)	None

button_wait_for_bump.py

#!/usr/bin/env python3

# クラスのインポート
from ev3dev2.button import Button
from ev3dev2.sound import Sound

# インスタンス化
my_button = Button()
my_sound = Sound()

# 左ボタンが押されて離される（バンプ）まで待つ
my_button.wait_for_bump('left')
my_sound.beep()

・wait_for_pressed(buttons,timeout_ms=None)

指定されたボタンが指定した時間内に押し込まれるまで待つ。
時間内にボタンが押されたらTrue，押されていなかったらFalseを返す。

button_wait_for_pressed.py

#!/usr/bin/env python3

# インポート
from ev3dev2.button import Button
from ev3dev2.sound import Sound

# インスタンス化
my_button = Button()
my_sound = Sound()

# 上ボタンと下ボタンが押されるまでまつ
my_button.wait_for_pressed(['up','down'])
my_sound.beep()

・wait_for_pressed(buttons,timeout_ms=None)

指定されたボタンが指定した時間内に離されるまで待つ。
時間内にボタンが離されたらTrue，離されなかったらFalseを返す。

button_wait_for_released.py

#!/usr/bin/env python3

# インポート
from ev3dev2.button import Button
from ev3dev2.sound import Sound

# インスタンス化
my_button = Button()
my_sound = Sound()

# 右ボタンが離されるまでまつ
my_button.wait_for_released('right')
my_sound.beep()

いずれかのボタンが押されているかどうか

・any()

いずれかのボタンが押されていたらTrue何も押されていなけらばFalseを返す

button_any.py

#!/usr/bin/env python3

# インポート
from ev3dev2.button import Button
from ev3dev2.sound import Sound
import time

# インスタンス化
my_button = Button()
my_sound = Sound()

# ずっと繰り返す
while True:
    # もしボタンが押されていたら
    if my_button.any():
        my_sound.speak("PUSHED",espeak_opts="-a 200 -p 99")
        break # whileループから抜ける
    else:
        time.sleep(0.01)

○○ボタンが押されているかどうか

・up/down/left/right/enter/backspace

ボタンが押されているかどうかbool型で返される。指定したボタンが押されてたらTrueを返す。

button_push.py

#!/usr/bin/env python3

# インポート
from ev3dev2.button import Button
from ev3dev2.sound import Sound
import time

# インスタンス化
my_button = Button()
my_sound = Sound()

# ずっと繰り返す
while True:
    # もし上ボタンが押されてたら
    if my_button.up:
        my_sound.speak("UP",espeak_opts="-a 200 -p 99")
        break # whileループから抜ける
    else:
        time.sleep(0.01)

これらはプロパティとして実装されている。プロパティは読み込み書き込みを制限できる変数みたいなものという理解で良い

イベントハンドラー

イベントハンドラー・・・特定のイベントが発生した時にあらかじめ登録されている処理を実行する事（ボタンの場合、ボタンが押されたり離されたりした時に呼び出される関数）

・process(new_state=None)

最後にprocess関数を呼び出した時と現在の各ボタンの状態とを比較し、状態が変わったボタンのイベントハンドラーに登録された関数を実行する。

引数	説明	デフォルト値
new_state	リファレンスに情報なし(基本的な使い方の時は省略して良い)	None

ButtonEventHandler.py

#!/usr/bin/env python3
# ButtonEventHandler.py 押されたボタンの名前を喋らせる
from time import sleep

from ev3dev2.button import Button
from ev3dev2.sound import Sound

# インスタンス化
my_button = Button()
my_sound = Sound()

# 左ボタンが押された時の処理を関数で宣言
def left(state):
    if state:
        my_sound.speak("left")

# 右ボタンが押された時の処理を関数で宣言
def right(state):
    if state:
        my_sound.speak("right")

# 上ボタンが押された時の処理を関数で宣言
def up(state):
    if state:
        my_sound.speak("up")

# 下ボタンが押された時の処理を関数で宣言        
def down(state):
    if state:
        my_sound.speak("down")

# 決定ボタンが押された時の処理を関数で宣言
def enter(state):
    if state:
        my_sound.speak("enter")


# 各ボタンが押された時に呼びだす関数をそれぞれ紐づける
my_button.on_left = left
my_button.on_right = right
my_button.on_up = up
my_button.on_down = down
my_button.on_enter = enter


while True:
    # 前にprocess()を呼び出した状態と比較して、今の状態が違う場合
    # 上で紐づけられた関数を実行する
    my_button.process()
    sleep(0.01)