Python-LEGO Mindstormsでソケット通信

Python環境でLEGOMindsotrmsEV3(以降EV3)を利用してロボットプログラミンなどを開発していると当然機械学習などの分析を応用して制御をしたいと思うことがある。そんな時まず考えるのがEV3のセンサー値をPCに送って分析に利用したいと考えるだろう。
方法としてはいくつかあるかもしれないが、今回はソケット通信を用いてセンサーの値をPC側に送ったり、またPC側からメッセージをEV3に送ったりする方法を書いていく。

参考

本記事は以下の記事を参考にしている。
pythonでsocket通信を勉強しよう

また、以下の書籍を参考にしている。
Pythonを使ったEV3の基本的な制御等については以下に網羅されている。
ロボットではじめるAI入門

ソケット通信とは

ソケット(Socket)とは通信するアプリケーション(今回でいうと2つのpythonプログラム)同士の出入り口のことで、IPアドレスとポート番号の組み合わせで表現することができる。
ソケット通信にはTCP（Transmission Control Protocol）とUDP（User Diagram Protocol）の通信規格があり今回はTCPを用いる。2つの違いはTCPが信頼性を重視するのに対して、UDPは高速性を重視する通信規格になっている。

今回の通信関係

今回は以下の2通りの通信を行う。それぞれがサーバー・クライアントを行う方法を記載していく。
その1

その2

動作環境

• PC
  Windows10
  Python 3.7.3

• EV3
  ev3dev2-python
  APIリファレンス
  Python 3.5.3

PC:サーバー・EV3：クライアント

まずはPCがサーバー側となりEV3がクライアントとなる方法。といってもサーバー、クライアント共にデータの送受信ができるためできることに大きな違いはない。通信を確立させる際にサーバーが接続待ち状態からクライアントを受け付けるためクライアントが接続要求する際にサーバー側は接続待機状態である必要がある。

準備

EV3の環境や構築については以下の記事を参照してほしい。
EV3で機械学習その1　環境構築編

EV3をPCとBluetooth接続したらPCでコマンドプロンプトを開きPC-EV3間のBluetooth接続に利用しているIPアドレスを調べる。
PCとEV3をBluetooth接続すると169.254.XXX.YYYのリンクローカルアドレスが割り当てられる。以下の手順でIPアドレスを調べる。

1. コマンドプロンプトを開く
2. ipconfigコマンドを実行する
3. 表示されるイーサネットアダプター Bluetooth ネットワークアドレスに表示されるIPアドレスをメモする

###　プログラム
以下がPC側サーバープログラム

PC_server.py

import socket
import sys
import time

ev3_massage = None

with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
    s.bind(('169.254.85.105', 50010))  # your PC's Bluetooth IP & PORT
    s.listen(1)
    print('Start program...')
    while True:
        conn, addr = s.accept()
        with conn:
            while True:
                ev3_massage = conn.recv(1024)
                if ev3_massage is not None:
                    ev3_massage = ev3_massage.decode()
                    print('get' + ev3_massage)
                    time.sleep(1.0)
                    if ev3_massage == 'BACKSPACE':
                        break
                    ev3_massage = None
            print('End program')
            sys.exit()

前述したイーサネットアダプター Bluetooth ネットワークアドレスのIPアドレスをs.bindのカッコ内に記述する。

以下がEV3側クライアントプログラム

ev3_client.py

import socket
import sys
import time
from ev3dev2.button import Button
from ev3dev2.display import Display
import ev3dev2.fonts as fonts

# definition
button = Button()
screen = Display()
font = fonts.load('luBS12')
ev3_message = None


with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
    s.connect(('169.254.85.105', 50010))    # your PC's Bluetooth IP & PORT
    print('connected')
    while not(button.backspace):
        if button.up:
            screen.draw.text((10,10), 'UP Button Pressed', font=font)
            ev3_message = 'UP'
            s.send(ev3_message.encode())
            time.sleep(0.5)
        if button.down:
            screen.draw.text((10,10), 'DOWN Button Pressed', font=font)
            ev3_message = 'DOWN'
            s.send(ev3_message.encode())
            time.sleep(0.5)
        if button.left:
            screen.draw.text((10,10), 'LEFT Button Pressed', font=font)
            ev3_message = 'LEFT'
            s.send(ev3_message.encode())
            time.sleep(0.5)
        if button.right:
            screen.draw.text((10,10), 'RIGHT Button Pressed', font=font)
            ev3_message = 'RIGHT'
            s.send(ev3_message.encode())
            time.sleep(0.5)
        screen.update()
        screen.clear()
        
    screen.draw.text((10,10), 'ENTER Button Pressed', font=font)
    ev3_message = 'BACKSPACE'
    s.send(ev3_message.encode())
screen.clear()
print('End program')
sys.exit()

前述したイーサネットアダプター Bluetooth ネットワークアドレスのIPアドレスをs.connectのカッコ内に記述する。

実行

プログラムを実行してみる。
サーバープログラムであるPC側から実行する。

コマンドプロンプトからPC側プログラムが格納されているフォルダにcdコマンドで移動する。
以下ではデスクトップにtestというフォルダを作成し、その中にPC_server.pyを作成している。

サーバープログラムを実行するとStart program...と表示され接続待機状態になる。

次に、クライアントプログラムであるEV3側を実行する。VSCCodeでのEV3のプログラム実行については機械学習その1　環境構築編を参照してほしい。

実行後、ソケット通信が確立するとconnectedが表示される。
その後EV3の上下左右のボタンを押すと対応したメッセージをPC側プログラムに送信し、受信するとコマンドプロンプトに表示される。
　　　
左上のボタンを押すと、PC側、EV3側両方のプログラムが終了する。
PC側のプログラムにソケット通信を介してUPやRIGHTといった文字列が送信できているのがわかる。
正確にはソケット通信で送信できるデータはバイト型なので、文字列をencodeして変換してから送り、受け取ったらdecodeして文字列に変換し直してからprint出力している。

Appendix

ソケット通信で受け取った値をCSVファイルに記録

サーバー側のプログラムを書き換えることで、受け取った値をCSVファイルに蓄積することができる。

PC_server_csv.py

import socket
import sys
import time
import os.path
import csv

ev3_massage = None

if os.path.exists('button.csv') == False:
    writedata = ['button']
    f = open('button.csv', 'w', newline='')
    writer = csv.writer(f)
    writer.writerow(writedata)
    f.close()

def write(data):
    f = open('button.csv', 'a', newline='')
    writer = csv.writer(f)
    writer.writerow([data])
    f.close()



with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
    s.bind(('169.254.85.105', 50010))  # your PC's Bluetooth IP & PORT
    s.listen(1)
    print('Start program...')
    while True:
        conn, addr = s.accept()
        with conn:
            while True:
                ev3_massage = conn.recv(1024)
                if ev3_massage is not None:
                    ev3_massage = ev3_massage.decode()
                    print('get' + ev3_massage)
                    write(ev3_massage)
                    time.sleep(1.0)
                    if ev3_massage == 'BACKSPACE':
                        break
                    ev3_massage = None
                    
            print('End program')
            sys.exit()

モジュールにはcsv、os.pathを新たに利用している。
プログラムを実行するとCSVファイルが生成され各セルに受け取った文字列が記録されているのがわかる。