こんにちは！園部です。

今回は先日記事にした、研修の「IoTのためのクラウドシステム構築」の内容を書いていきます。

※本記事で扱うプログラムは全てpythonです

◇ 全体の流れ

  • マイコンへのセンサー接続

  • クラウド側へのデータ保存

  • センサデータの可視化

  • センサデータに基づく電子メールの送信

  • アクチュエータの制御

※本日はマイコンへのセンサー接続について説明します

◇ 用意する物

  • RaspberryPi

  • ブレッドボード

  • AE-BME280（温度センサ）

  • AE-DRV8830（モータードライバモジュール）

  • 積層セラミックコンデンサ 0.1μF、10μF、1000pF

  • モーター

※今回はマブチモーターのRE-280RAというモーターを使用しましたが、AE-DRV8830の過電流防止の仕様により、動作が安定しなかったため、もう少し消費電力の小さいモーターが良いと思います。

1.マイコンへのセンサー接続

まずはAE-BME280という温湿度気圧センサモジュールを使ってデータをコンソールに表示します。

■ RaspberryPiの用意

RaspberryPiを用意します。

今回はRaspberryPi4を使用しますが、RaspberryPiシリーズなら何でも良いです。

RaspberryPiの初期セットアップについては調べるとたくさん出てくるので他のブログを参考にしてください。

今回はTeraTeamを使ってSSHで操作しますので、RaspberryPi設定でSSHを有効化します。

またセンサはI2Cという通信方法を使用しますので、I2Cも有効にします。

■ センサモジュールの用意

今回のAE-BME280とAE-DRV8830は秋葉原の秋月電子で購入しました。

AE-BME280にピンヘッダをはんだ付けし、ブレッドボードに刺せるようにします。

今回はI2C接続にするので、モジュール右上のジャンパ設定のJ3をはんだでジャンパします。

bme画像

記事内画像

■ モータードライバモジュールの用意

すぐには使いませんが、とりあえず用意しておきます。

センサと同じように、AE-DRV8830にピンヘッダをはんだ付けし、ブレッドボードに刺せるようにします。

AE-DRV8830

記事内画像

■ ブレッドボードで回路作成

まずは図の通りにRaspberryPiのピンと接続します。

配線図

記事内画像

RaspberryPiはI2Cのチャンネルが１つしかないので、２つのモジュールを並列で接続しています。

コンデンサは電源電圧の安定化と、ノイズの吸収の役割をしています。

図はモーターとセンサの電源を電池ボックスとRaspberryPiの電源で分けていますが、どちらもRaspberryPiの電源で構いません。

配線後の画像

記事内画像

■ RaspberryPiにSSHで接続する

RaspberryPiがインターネットに繋がる状態にします。

TeraTermをインストールし、RaspberryPiのIPアドレスを入力し、接続します。

RaspberryPiのIPアドレスはRaspberryPiのターミナルで”ip a”とかコマンドを叩くと出ます。

■ プログラムの実行

I2Cをpythonから制御するためのライブラリをインストールします。

$ pip install smbus2

RaspberryPiはI2Cのチャンネルが１つしかないので、２つのモジュールを並列で接続しています。

i2ctoolsでセンサのI2Cアドレスが取れるか確認する。（0x64と0x76がとれるはず）

$ i2cdetect -y 1

例で載せている画像はDRV8830を接続していませんが、こんな感じで表示されます。

i2cアドレス画像

記事内画像

アドレスが確認出来たらハード的な準備は完了です。

■ pythonプログラムの作成

やり方はRaspberryPiのエディターを使っても良いですが、せっかくsshで繋げているので日頃使っているPCで作成して、winSCPなどのソフトを使ってRaspberryPiに渡すやり方が楽だと思います。

bme280_get_data.py

  • #coding: utf-8

  • import time

  • import datetime

  • import json

  • from smbus2 import SMBus

  • class BME280:

  • def init(self):

  • self.bus_number = 1

  • self.i2c_address = 0x76

  • self.bus = SMBus(self.bus_number)

  • self.digT = []

  • self.digP = []

  • self.digH = []

  • self.t_fine = 0.0

  • #時間を文字列に変換する

  • def support_datetime_default(sel,o):

  • if isinstance(o, datetime.datetime):

  • return o.isoformat()

  • def writeReg(self,reg_address, data):

  • self.bus.write_byte_data(self.i2c_address,reg_address,data)

  • def get_calib_param(self):

  • calib = []

  • for i in range (0x88,0x88+24):

  • calib.append(self.bus.read_byte_data(self.i2c_address,i))

  • calib.append(self.bus.read_byte_data(self.i2c_address,0xA1))

  • for i in range (0xE1,0xE1+7):

  • calib.append(self.bus.read_byte_data(self.i2c_address,i))

  • self.digT.append((calib[1] << 8) | calib[0])

  • self.digT.append((calib[3] << 8) | calib[2])

  • self.digT.append((calib[5] << 8) | calib[4])

  • self.digP.append((calib[7] << 8) | calib[6])

  • self.digP.append((calib[9] << 8) | calib[8])

  • self.digP.append((calib[11]<< 8) | calib[10])

  • self.digP.append((calib[13]<< 8) | calib[12])

  • self.digP.append((calib[15]<< 8) | calib[14])

  • self.digP.append((calib[17]<< 8) | calib[16])

  • self.digP.append((calib[19]<< 8) | calib[18])

  • self.digP.append((calib[21]<< 8) | calib[20])

  • self.digP.append((calib[23]<< 8) | calib[22])

  • self.digH.append( calib[24] )

  • self.digH.append((calib[26]<< 8) | calib[25])

  • self.digH.append( calib[27] )

  • self.digH.append((calib[28]<< 4) | (0x0F & calib[29]))

  • self.digH.append((calib[30]<< 4) | ((calib[29] >> 4) & 0x0F))

  • self.digH.append( calib[31] )

  • for i in range(1,2):

  • if self.digT[i] & 0x8000:

  • self.digT[i] = (-self.digT[i] ^ 0xFFFF) + 1

  • for i in range(1,8):

  • if self.digP[i] & 0x8000:

  • self.digP[i] = (-self.digP[i] ^ 0xFFFF) + 1

  • for i in range(0,6):

  • if self.digH[i] & 0x8000:

  • self.digH[i] = (-self.digH[i] ^ 0xFFFF) + 1

  • #センサデータをmqttメッセージにして返す

  • def getMessage(self):

  • data = self.readData()

  • sensor = {}

  • sensor[‘temp’] = float(data[0])

  • sensor[‘hum’] = float(data[2])

  • sensor[‘pres’] = float(data[1])

  • message = {}

  • message[‘timestamp’] = datetime.datetime.now() – datetime.timedelta(hours=9)#KIBANAに入力される時間はUTC扱いになるため、9時間引いて日本時間にする

  • message[‘sensor’] = sensor

  • messageJson = json.dumps(message, default=self.support_datetime_default)

  • return messageJson

  • #センサデータを読み取って、温度、湿度、気圧をリスト化して返す。

  • def readData(self):

  • self.setup()

  • self.get_calib_param()

  • data = []

  • for i in range (0xF7, 0xF7+8):

  • data.append(self.bus.read_byte_data(self.i2c_address,i))

  • pres_raw = (data[0] << 12) | (data[1] << 4) | (data[2] >> 4)

  • temp_raw = (data[3] << 12) | (data[4] << 4) | (data[5] >> 4)

  • hum_raw = (data[6] << 8) | data[7]

  • t = self.compensate_T(temp_raw)

  • p = self.compensate_P(pres_raw)

  • h = self.compensate_H(hum_raw)

  • list = [t,p,h]

  • return list

  • def compensate_P(self,adc_P):

  • pressure = 0.0

  • v1 = (self.t_fine / 2.0) – 64000.0

  • v2 = (((v1 / 4.0) * (v1 / 4.0)) / 2048) * self.digP[5]

  • v2 = v2 + ((v1 * self.digP[4]) * 2.0)

  • v2 = (v2 / 4.0) + (self.digP[3] * 65536.0)

  • v1 = (((self.digP[2] * (((v1 / 4.0) * (v1 / 4.0)) / 8192)) / 8) + ((self.digP[1] * v1) / 2.0)) / 262144

  • v1 = ((32768 + v1) * self.digP[0]) / 32768

  • if v1 == 0:

  • return 0

  • pressure = ((1048576 – adc_P) – (v2 / 4096)) * 3125

  • if pressure < 0x80000000:

  • pressure = (pressure * 2.0) / v1

  • else:

  • pressure = (pressure / v1) * 2

  • v1 = (self.digP[8] * (((pressure / 8.0) * (pressure / 8.0)) / 8192.0)) / 4096

  • v2 = ((pressure / 4.0) * self.digP[7]) / 8192.0

  • pressure = pressure + ((v1 + v2 + self.digP[6]) / 16.0)

  • pres = format(pressure,“7.2f”)

  • print (“pressure : %7.2f hPa” % (pressure/100))

  • return pres

  • def compensate_T(self,adc_T):

  • v1 = (adc_T / 16384.0 – self.digT[0] / 1024.0) * self.digT[1]

  • v2 = (adc_T / 131072.0 – self.digT[0] / 8192.0) * (adc_T / 131072.0 – self.digT[0] / 8192.0) * self.digT[2]

  • self.t_fine = v1 + v2

  • temperature = self.t_fine / 5120.0

  • temp = format(temperature,“-6.2f”)

  • print (“temp : %-6.2f ℃” % (temperature))

  • return temp

  • def compensate_H(self,adc_H):

  • var_h = self.t_fine – 76800.0

  • if var_h != 0:

  • var_h = (adc_H – (self.digH[3] * 64.0 + self.digH[4]/16384.0 * var_h)) * (self.digH[1] / 65536.0 * (1.0 + self.digH[5] / 67108864.0 * var_h * (1.0 + self.digH[2] / 67108864.0 * var_h)))

  • else:

  • return 0

  • var_h = var_h * (1.0 – self.digH[0] * var_h / 524288.0)

  • if var_h > 100.0:

  • var_h = 100.0

  • elif var_h < 0.0:

  • var_h = 0.0

  • print (“hum : %6.2f ％” % (var_h))

  • h = format(var_h,“6.2f”)

  • return h

  • def setup(self):

  • osrs_t = 1 #Temperature oversampling x 1

  • osrs_p = 1 #Pressure oversampling x 1

  • osrs_h = 1 #Humidity oversampling x 1

  • mode = 3 #Normal mode

  • t_sb = 5 #Tstandby 1000ms

  • filter = 0 #Filter off

  • spi3w_en = 0 #3-wire SPI Disable

  • ctrl_meas_reg = (osrs_t << 5) | (osrs_p << 2) | mode

  • config_reg = (t_sb << 5) | (filter << 2) | spi3w_en

  • ctrl_hum_reg = osrs_h

  • self.writeReg(0xF2,ctrl_hum_reg)

  • self.writeReg(0xF4,ctrl_meas_reg)

  • self.writeReg(0xF5,config_reg)

  • def main():

  • bme = BME280()

  • data_list = bme.readData()

  • if name == ‘main’:

  • main()

プログラムを実行します（ 以下の例はホームディレクトリに作成した場合）

$ python bme280_get_data.py

温度、湿度、気圧が表示されれば成功です！

・ センサデータ出力

記事内画像

bme280_get_data.pyのメソッド

  • compensate_P（気圧）

  • compensate_T（気温）

  • compensate_H（湿度）

の中でprintしています。

今後センサを使うプログラムはこのプログラムを呼び出すようにします。

今回はマイコンへのセンサー接続を紹介していきました。

次からはいよいよクラウドが登場しますので、お楽しみに！