加速度センサ

以前にArduinoで使ってみた加速度センサのインタフェースがI2Cなので、これもRaspberry Piに接続して試してみる事にした。加速度センサはフリースケール社のMMA7455Lを14ピンのDIP基板に乗せたものだ。ストロベリーリナックスで購入できる。

動作電圧が2.4V~3.6Vで、5Vでは動作しないから5V動作のArduinoで使う時は注意が必要なのだが、Raspberry PiはGPIOが3.3Vなので都合が良い。まずはブレッドボード上でVCC/GND/SDA/SCLの各ピンをGPIOの対応するピンに接続。
MMA7455

MMA745Lモジュール

Raspberry Pi

1 : GND

6 : GND

2 : VDD  6 : AVDD  8 : CS

1 : 3.3V

13 : SDA

3 : SDA

14 : SCL

5 : SCL

Raspberry PiのI2Cポートにはプルアップ抵抗が付いているので、プルアップする必要は無いはずなのだが、抵抗を外付けしないと正しく認識されなかった。正しく認識されていれば以下のように検出される。

>pi@raspberrypi ~ $ sudo i2cdetect 1
WARNING! This program can confuse your I2C bus, cause data loss and worse!
I will probe file /dev/i2c-1.
I will probe address range 0x03-0x77.
Continue? [Y/n]
0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f
00:          -- -- -- -- -- -- -- -- -- -- -- -- --
10: -- -- -- -- -- -- -- -- -- -- -- -- -- 1d -- --
20: -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- --
30: -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- --
40: -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- --
50: -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- --
60: -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- --
70: -- -- -- -- -- -- -- --

あとはPythonで簡単なプログラムを書いて確認。加速度センサの値は簡単に取得できたので、ついでに値をPyGameでグラフ表示できるようにしてみた。プログラムは以下の通り。

#!/usr/bin/env python
# -*- coding: utf-8 -*-

import sys
import pygame
from pygame.locals import *
import smbus
import time

screen_rect = Rect(0, 0, 640, 320)
white = 255, 255, 255
red = 255, 0, 0
green = 0, 255, 0
blue = 0, 0, 255
yellow = 255, 255, 0

class MMA7455L:
    i2c = smbus.SMBus(1)   
    addr = 0x1d
    def __init__(self):
        self.i2c.write_byte_data(self.addr, 0x16, 0x05)     # 2G/Measurement Mode
        for i in range(0, 6):
            self.i2c.write_byte_data(self.addr, 0x10 + i, 0)
        time.sleep(0.1)
    def get(self):
        return self.i2c.read_i2c_block_data(self.addr, 0x06, 3)
    def drift(self):
        dd = [0, 0, 0]
        for i in range(0, 20):
            buf = self.i2c.read_i2c_block_data(self.addr, 0x06, 3)
            dd[0] = dd[0] + signed(buf[0])
            dd[1] = dd[1] + signed(buf[1])
            dd[2] = dd[2] + signed(buf[2]) - 64
        dd[0] = (dd[0] / 8) * -1
        dd[1] = (dd[1] / 8) * -1
        dd[2] = (dd[2] / 8) * -1
        for i in range(0, 3):
            self.i2c.write_byte_data(self.addr, 0x10 + i * 2, dd[i])
            self.i2c.write_byte_data(self.addr, 0x11 + i * 2, (dd[i] >> 8) & 0x07)
        time.sleep(0.1)

def signed(n):
    return n if n < 128 else n - 256 

if __name__ == "__main__":
    mma = MMA7455L()
    mma.drift()
    pygame.init()
    screen = pygame.display.set_mode(screen_rect.size)
    center = screen_rect.height / 2
    pygame.draw.line(screen, blue, (20, center),\
                    (screen_rect.width, center), 1)
    pygame.draw.line(screen, blue, (20, center - 64),\
                    (screen_rect.width, center - 64), 1)
    pygame.draw.line(screen, blue, (20, center + 64),\
                    (screen_rect.width, center + 64), 1)
    pygame.draw.line(screen, blue, (20, center - 128),\
                    (screen_rect.width, center - 128), 1)
    pygame.draw.line(screen, blue, (20, center + 128),\
                    (screen_rect.width, center + 128), 1)
    pygame.draw.line(screen, blue, (20, 0), (20, screen_rect.height), 1)
    font = pygame.font.Font(None, 16)
    text = font.render("+2G", True, white)
    screen.blit(text, (0, center - 133))
    text = font.render("+1G", True, white)
    screen.blit(text, (0, center - 69))
    text = font.render(" 0G", True, white)
    screen.blit(text, (0, center - 5))
    text = font.render("-1G", True, white)
    screen.blit(text, (0, center + 59))
    text = font.render("-2G", True, white)
    screen.blit(text, (0, center + 123))
    d1 = mma.get()
    d1[0] = signed(d1[0]) * -1
    d1[1] = signed(d1[1]) * -1
    d1[2] = signed(d1[2]) * -1
    x = 20
    try:
        while True:
            for event in pygame.event.get():
                if event.type == pygame.QUIT:
                    pygame.quit()
                    sys.exit()
            d2 = mma.get()
            d2[0] = signed(d2[0]) * -1
            d2[1] = signed(d2[1]) * -1
            d2[2] = signed(d2[2]) * -1
            pygame.draw.line(screen, red, (x, d1[0] + center),\
                            (x + 1, d2[0] + center), 1)
            pygame.draw.line(screen, green, (x, d1[1] + center),\
                            (x + 1, d2[1] + center), 1)
            pygame.draw.line(screen, yellow, (x, d1[2] + center),\
                            (x + 1, d2[2] + center), 1)
            pygame.display.update()
            x = x + 1
            d1 = d2
            time.sleep(0.1)
    except KeyboardInterrupt:
        print '\nbreak'
        pygame.quit()

実行すると以下のように表示される、グラフの赤線がX軸、緑線がY軸、黄色線がZ軸。Z軸が最初から1Gになっているのは、何もしなくても重力が加わっているからだ。グラフの最初は加速度センサを傾けた時、後半はそれぞれの軸に対して振ったときの値を示している。
MMATest プルアップ抵抗の謎はあったが、I2C液晶の時よりも簡単に使う事ができた。モジュールは千円以下で買えるので、手軽に色々な応用ができそうだ。

シリアルポート(2)

TCPソケットで受信したデータをRaspberry Piのシリアルポートに送信し、シリアルポートで受信したデータをTCPソケットに送信するプログラムをPythonで作成して、Raspberry Piを一種の変換器にする事ができた。

Raspberry Piに接続するPC側はtelnetクライアントを使用してテストを行ったが、PC側にもシリアルポートとTCPを変換するアプリケーションがあれば、シリアル通信ソフトを使ってRaspberry Piのシリアルポートに対する入出力を行う事ができる。

PCに仮想シリアルポートを追加し、同時にTCPへのトンネルを行うアプリはセンサー類をネットワークに接続して遠隔地からデータ収集を行うデバイスのメーカーが提供していて、ここのページからダウンロードできる。
マルチ接続バージョンは有料だが、シングル接続バージョンはフリー。その代わり広告が表示される。

hw-vsp3-single_3-1-2.exe をダウンロードしてインストールしたら、先にRaspberry Piで前回のプログラムを起動して接続待ち状態にしておく。次にHW Virtaul Serial Portを起動し、Loginをクリック。ログインパスワードはデフォルトではadmin。HWVSP1次にSettingタブをクリックして、NVT Enabledのチェックを外す。このアプリはデバイスと接続した時にコントロール情報を送出するので、それがゴミになってしまうからだ。
HWVSP2Virtual Serial Portのタブに戻り、Raspberry PiのIPアドレスとポートアドレス5555を 入力して仮想シリアルポートを作成する。正しくRaspberry Piと接続されれば、以下のような感じでRaspberry Piのコンソールに表示されるはずだ(表示されるIPは異なる)。
HWVSP4 これでPCに仮想シリアルポートが作成され、Pythonのプログラムを介してRaspberry Piのシリアルポートと接続された事になる。
HWVSP5 Raspberry Pi部分を無いものとして考えると、単純にPCのシリアルポートに外部デバイスが接続されているのと同じ状態になるので、あとはPCで追加された仮想シリアルポートを開き、通信を行えばよい。

かなり簡易的ではあるけれど、XPort等のシリアルデバイスサーバと同じような事が、これでRaspberry Piを使って実現できた。実用的にはまだ色々と問題が残されているが、実用に使うならそれこそXPort等を使用するべきだろう。

シリアルポート

Raspberry Piのシリアルポートはデフォルトではシリアルコンソール用として使用されていることを、以前にUSBシリアル変換モジュールを接続して確認した。今回はその続き。

システムが使用しているシリアルポートをユーザが使用する為には、二つの設定ファイルを修正する必要がある。一つはシステムの起動スクリプトである /etc/inittab、もう一つはコンソールデバイスの設定ファイルである /boot/cmdline.txt だ。

/etc/inittab は最後の行の

T0:23:respawn:/sbin/getty -L ttyAMA0 115200 vt100

をコメントアウト

/boot/cmdline.txt は

dwc_otg.lpm_enable=0 console=ttyAMA0,115200 kgdboc=ttyAMA0,115200 console=tty1 root=/dev/mmcblk0p2 rootfstype=ext$

dwc_otg.lpm_enable=0 console=tty1 root=/dev/mmcblk0p2 rootfstype=ext4 elevator=deadline rootwait

のように変更する。

両方のファイルを修正して保存したら、システムを再起動すればシリアルコンソールが無効になるはずなのだが、起動時の

Uncompressing Linux… done, booting the kernel.

というメッセージだけはシリアルポートに出力されてしまう。このメッセージはカーネルが出力しているらしいので、設定を変えただけでは禁止する事ができないようだ。Raspberry Piにシリアル機器を接続した場合、これがゴミデータとなってしまう可能性があるので、注意が必要。

シリアルポートをPythonで扱うためにはモジュールのインストールも必要なので

sudo apt-get install python-serial

でインストールしておく。これでRaspberry Piのシリアルポートを扱う準備は整った。

まずは以前作ったTCPソケット通信のプログラムを修正して、TCPとシリアルポート間で通信を行うプログラムを作ってみた。ソースは以下の通り。

#!/usr/bin/env python
# -*- coding:utf-8 -*-

import socket
import select
import Queue
import sys
import serial

def socketio(sock, comm):
    ilst = [sock, comm]
    olst = []
    ique = Queue.Queue() 
    oque = Queue.Queue() 
    connect = False
    while True:
        # socket i/o
        rlst, wlst, xlst = select.select(ilst, olst, [], 0)
        for si in rlst:
            # accept?
            if si == sock:
                if not connect:
                    # connect
                    conn, address = si.accept()
                    print 'connect', address
                    connect = True
                    ilst.append(conn)
                    olst.append(conn)
                else:
                    # discard
                    conn, address = si.accept()
                    conn.shutdown(socket.SHUT_RDWR)
                    conn.close
            elif si == comm:
                data = comm.read()
                oque.put(data)
            else:
            # recv?
                data = si.recv(1024)
                if data:
                    ique.put(data)
                else:
                    # disconnect
                    print 'disconnect'
                    connect = False
                    ilst.remove(si)
                    olst.remove(si)
                    si.close
                    with ique.mutex:
                        ique.queue.clear()
                    with oque.mutex:
                        oque.queue.clear()
        for so in wlst:
            if not oque.empty():
                # send
                data = oque.get_nowait()
                so.send(data)
        # Serial
        if not ique.empty():
            data = ique.get_nowait()
            comm.write(data)
                        
def start(host, port, bps):
    try:
        comm = serial.Serial('/dev/ttyAMA0', bps)
        sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
        sock.bind((host, port))
        sock.listen(1)
        socketio(sock, comm)
    except KeyboardInterrupt:
        print '\nbreak'
        sock.shutdown(socket.SHUT_RDWR)
        sock.close()

if __name__ == "__main__":
    start('', 5555, 115200)

このプログラムを起動して、Raspberry Piと同一のネットワークに接続されているPCから、Raspberry PiのIPアドレスのポート5555に対してtelnet接続を行う。接続が確立されたら、telnetクライアントからネットワーク経由でRaspberry Piに送られた文字がシリアルポートに出力され、逆にシリアルポートに入力された文字がtelnetクライアントに表示されるはずだ。

これで何ができるかと言うと、Raspberry Piのシリアルポートに接続されたデバイスを、遠隔地からネットワーク経由で使う事ができる。制御線を使用するようなデバイスには対応できないが、単純な無手順の伝送でなら使い道はあると思う。

この話はまだ続くのだが、長くなるので今回はここまで。後半は近日中に。