Mindstorms EV3 Server – DECODE X NET

レゴブロックでロボットを作るときなどに使われるコンピュータMindstorms EV3。
通常は、製品に付属するツールを使ってプログラミングをするのですが、EV3用のLinux OS(http://www.ev3dev.org/)を使うと、Python等の言語でインターフェイスを制御できるようになります。
そこで、以下サイトで作成したキットを使ってWebブラウザからEV3を制御できるサーバを作ってみました。
http://decode.red/ed/archives/369

from __future__ import print_function
import socket
import select
import re
import ev3dev.ev3 as ev3

def main():
	host = '0.0.0.0'
	port = 8000
	backlog = 10
	bufsize = 4096

	server_sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
	readfds = set([server_sock])
	try:
		server_sock.bind((host, port))
		server_sock.listen(backlog)

		t1 = ev3.TouchSensor()
		in3 = ev3.Device('lego-port', name_pattern='*', address='in3')
		in3.connected
		in3.set_attr_string('mode', 'nxt-analog')
		in3.set_attr_string('set_device', 'lego-nxt-touch')
		t2 = ev3.Sensor(address='in3')
		t2.connected
		ss = ev3.Sensor(ev3.INPUT_2)
		us = ev3.UltrasonicSensor()
	
		while True:
			rready, wready, xready = select.select(readfds, [], [])
			for sock in rready:
				if sock is server_sock:
					conn, address = server_sock.accept()
					readfds.add(conn)
				else:
					msg = sock.recv(bufsize)
					if len(msg) == 0:
						sock.close()
						readfds.remove(sock)
					else:
						r = re.compile("GET /(.+) HTTP")
						m = r.search(msg)
						if m is not None:
							print(m.group(1))
							cmd = m.group(1)
	
							m2 = re.compile("([a-z0-9]+)=([a-z0-9\-_]+)")
							para = m2.findall(cmd)
	
							print(para)
	
							for p in para:
								if p[0] == 'sp':
									rr = p[1].replace('_', ' ')
									print('speak: ' + rr)
									ev3.Sound.speak(rr)
								if p[0] == 'm1':
									m1 = ev3.Motor('outA')
									m1.run_timed(time_sp=3000, speed_sp=int(p[1]))
								if p[0] == 'm2':
									m2 = ev3.MediumMotor('outB')
									m2.run_timed(time_sp=3000, speed_sp=int(p[1]))
								if p[0] == 'm3':
									m3 = ev3.LargeMotor('outC')
									m3.run_timed(time_sp=3000, speed_sp=int(p[1]))
								if p[0] == 'm4':
									m4 = ev3.LargeMotor('outD')
									m4.run_timed(time_sp=3000, speed_sp=int(p[1]))
						
						res = str(t1.value()) + '\n'
						res = res + str(t2.get_attr_int('value0')) + '\n'
						res = res + str(ss.value())  + '\n'
						res = res + str(us.value()) + '\n'

						sock.send("HTTP/1.1 200 OK\r\nContent-Length: " + str(len(res)) + "\r\n\r\n" + res)
						sock.close()  
						readfds.remove(sock)

	finally:
		for sock in readfds:
			sock.close()
	return

if __name__ == '__main__':
  main()

from __future__ import print_function

import socket

import select

import re

import ev3dev.ev3 as ev3

def main():

host = '0.0.0.0'

port = 8000

backlog = 10

bufsize = 4096

server_sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)

readfds = set([server_sock])

try:

server_sock.bind((host, port))

server_sock.listen(backlog)

t1 = ev3.TouchSensor()

in3 = ev3.Device('lego-port', name_pattern='*', address='in3')

in3.connected

in3.set_attr_string('mode', 'nxt-analog')

in3.set_attr_string('set_device', 'lego-nxt-touch')

t2 = ev3.Sensor(address='in3')

t2.connected

ss = ev3.Sensor(ev3.INPUT_2)

us = ev3.UltrasonicSensor()

while True:

rready, wready, xready = select.select(readfds, [], [])

for sock in rready:

if sock is server_sock:

conn, address = server_sock.accept()

readfds.add(conn)

else:

msg = sock.recv(bufsize)

if len(msg) == 0:

sock.close()

readfds.remove(sock)

else:

r = re.compile("GET /(.+) HTTP")

m = r.search(msg)

if m is not None:

print(m.group(1))

cmd = m.group(1)

m2 = re.compile("([a-z0-9]+)=([a-z0-9\-_]+)")

para = m2.findall(cmd)

print(para)

for p in para:

if p[0] == 'sp':

rr = p[1].replace('_', ' ')

print('speak: ' + rr)

ev3.Sound.speak(rr)

if p[0] == 'm1':

m1 = ev3.Motor('outA')

m1.run_timed(time_sp=3000, speed_sp=int(p[1]))

if p[0] == 'm2':

m2 = ev3.MediumMotor('outB')

m2.run_timed(time_sp=3000, speed_sp=int(p[1]))

if p[0] == 'm3':

m3 = ev3.LargeMotor('outC')

m3.run_timed(time_sp=3000, speed_sp=int(p[1]))

if p[0] == 'm4':

m4 = ev3.LargeMotor('outD')

m4.run_timed(time_sp=3000, speed_sp=int(p[1]))

res = str(t1.value()) + '\n'

res = res + str(t2.get_attr_int('value0')) + '\n'

res = res + str(ss.value()) + '\n'

res = res + str(us.value()) + '\n'

sock.send("HTTP/1.1 200 OK\r\nContent-Length: " + str(len(res)) + "\r\n\r\n" + res)

sock.close()

readfds.remove(sock)

finally:

for sock in readfds:

sock.close()

return

if __name__ == '__main__':

main()

URLパラメータのHTTPリクエストでEV3へのコマンドを送り、レスポンスで全てのセンサ値を受け取ります。
<リクエスト>
sp= : 音声再生の文字列
m1= : モータ1
m2= : モータ2
m3= : モータ3
m4= : モータ4
<レスポンス>
タッチセンサ1
タッチセンサ2
サウンドセンサ
超音波センサ

古いNXT用のセンサの扱いがちょっと苦労しました。
ネットワーク接続はUSB WiFiです。まったく普通のLinuxとして扱えます。