「switchbotから部屋の不快度をGrafana,InfruxDBで可視化してみる」 可視化愛好家、前加賀美の今日も可視化日和!! #2

スポンサーリンク
tech系(python)
スポンサーリンク

主にこの記事に書いてある情報

  • pythonで不快指数をもとめる方法
  • 不快指数が閾値を超えるとLEDランプで知らせる機能の実装方法

はじめに

前加賀美
前加賀美

どうも、こんにちは、可視化愛好家の前加賀美 Necozay Gurlukovich(まえかがみ・ねこぜい・ごるるこびっち)です。

前加賀美
前加賀美

皆さんは不快指数というものを御存じでしょうか

不快指数とは、 「寒い」、「暑い」などの体感を数値化したものです。

以下の式で表されます。

$DI = 0.81T + 0.01H(0.99T – 14.3) + 46.3$

$DI$は不快指数

$T$は気温(℃)

$H$は湿度(%)

前回まで

前回はInfruxDBとGrafanaとpythonで部屋の温度を可視化させました。

今回は不快指数を可視化させます。

そして、不快指数が悪い値になったら警報(LED点灯)をするような実装を施します。

python

pythonコード、不快指数の部分は以下のようになりました。(前回のコードに追加しています)

★マークのついている「XX:XX:XX:XX:XX:XX」のところは各人のswitchbotのMACアドレスを入れてください。

from influxdb import InfluxDBClient
import struct
import sys
from bluepy.btle import Scanner, DefaultDelegate
import time
from datetime import datetime

# データベースに接続
dbclient = InfluxDBClient(host='localhost', port=8086, database='tempDB')

#Broadcastデータ取得用デリゲート
class SwitchbotScanDelegate(DefaultDelegate):
    #コンストラクタ
    def __init__(self, macaddr):
        #btle.DefaultDelegate.__init__(self)
        DefaultDelegate.__init__(self)
        #センサデータ保持用変数
        self.sensorValue = None
        self.macaddr = macaddr
 
    # スキャンハンドラー
    def handleDiscovery(self, dev, isNewDev, isNewData):
        # 対象Macアドレスのデバイスが見つかったら
        if dev.addr == self.macaddr:
            # アドバタイズデータを取り出し
            for (adtype, desc, value) in dev.getScanData():  
                #環境センサのとき、データ取り出しを実行
                if desc == '16b Service Data':
                    #センサデータ取り出し
                    self._decodeSensorData(value)
 
    # センサデータを取り出してdict形式に変換
    def _decodeSensorData(self, valueStr):
        #文字列からセンサデータ(4文字目以降)のみ取り出し、バイナリに変換
        valueBinary = bytes.fromhex(valueStr[4:])
        #バイナリ形式のセンサデータを数値に変換
        batt = valueBinary[2] & 0b01111111
        isTemperatureAboveFreezing = valueBinary[4] & 0b10000000
        temp = ( valueBinary[3] & 0b00001111 ) / 10 + ( valueBinary[4] & 0b01111111 )
        if not isTemperatureAboveFreezing:
            temp = -temp
        humid = valueBinary[5] & 0b01111111
        #dict型に格納
        self.sensorValue = {
            'SensorType': 'SwitchBot',
            'MacAddr': self.macaddr,
            'Temperature': temp,
            'Humidity': humid,
            'Battery': batt
        }


def send_temp_data(macaddr):
    # macaddr指定した温湿度センサの値を取得
    macaddr = str.lower(macaddr)
    scanner = Scanner().withDelegate( SwitchbotScanDelegate(macaddr) )
    #スキャン(timeout2s)
    scanner.scan( 9.0 )
    temp=0
    #出力
    try:
        temp = str(scanner.delegate.sensorValue['Temperature'])
    except TypeError as e:
        print("Temporarily lost communication with Bluetooth temperature device.",e) 
    except:
        import traceback
        traceback.print_exc()
    return  temp

def send_humi_data(macaddr):
    # macaddr指定した温湿度センサの値を取得
    macaddr = str.lower(macaddr)
    scanner = Scanner().withDelegate( SwitchbotScanDelegate(macaddr))
    #スキャン(timeout2s)
    scanner.scan( 9.0 )
    temp=0
    #出力
    try:
        humi = str(scanner.delegate.sensorValue['Humidity'])
    except TypeError as e:
        print("Temporarily lost communication with Bluetooth humidity device.",e) 
    except:
        import traceback
        traceback.print_exc()
    
    return  humi

 
if __name__ == '__main__':
    while True:
        temp = float(send_temp_data('XX:XX:XX:XX:XX:XX')) # ★
        humi = float(send_humi_data('XX:XX:XX:XX:XX:XX')) # ★
        discomfort = 0.81*temp + 0.01*humi*( 0.99*temp - 14.3 ) + 46.3

        json_body = [
            {
                "measurement": "temp_measurement",
                "time": datetime.utcnow(),
                "fields": {
                    "temp": temp,
                    "humi": humi,
                    "discomfort": discomfort,
                }
            }
        ]
        dbclient.write_points(json_body)
        time.sleep(60)

Grafana

試しに、上記のpythonコードをsudo実行すると不快指数をGrafanaに表示できました。(ここは前回の気温と一緒)

LEDライト

不快指数がある閾値を超えたら、もしくは下回ったらLEDライトを点灯させるようにします。

まずは試しに、Lチカを実施してみます。

前加賀美
前加賀美

ローソンで売っている、ジューシーな鳥のホットスナック?

それはLチキ。

LチカとはLEDをチカチカと点灯させることです。

今回使用する機器

今回使用する機器は以下のものです。

  • LED
  • 抵抗(1kΩ)
  • ジャンプワイヤー(オス-メス)
  • ブレッドボード

「LED」、「抵抗(1kΩ)」、「ジャンプワイヤー(オス-メス)」は前回の記事でも書いた、スターターキットに入っていました。

ブレッドボードはスターターキットとは別に購入しました。

LEDの接続に関して

LEDの接続に関してはこちらのサイトが参考になりました。

注意点としてはLEDにはプラスとマイナスがあること。

LEDの足が長い方がプラスです。

また、GPIOならびに各接続の番号は以下の画像を参考にしてみてください。引用元はこちら

https://www.raspberrypi.com/documentation/computers/os.html より引用
https://www.raspberrypi.com/documentation/computers/os.html より引用

以上を踏まえて今回の回路構成は以下のようになりました。

17番のGPIOにプラス側を接続、Groundにマイナス側を接続します。

ラズベリーパイのプラス端子であるGPIOとLEDのプラス側の間には抵抗を入れます。

構成図としては以下のようになります。

Lチカのpythonコード

Lチカのためのpythonコードは以下のようになっています。

import RPi.GPIO as GPIO
import time

# LED制御ピンに、GPIO 17 を使用する
LED_GPIO = 17

# GPIO.BCMを設定することで、GPIO番号で制御出来るようになる。
GPIO.setmode(GPIO.BCM)
# GPIO.OUTを設定することで、出力モードになる
# 出力モードになると電圧を ON/OFF することが出来る。
GPIO.setup(LED_GPIO, GPIO.OUT)

# 5回点灯するようなforループ
for i in range(5):
    # 出力1にしてLED点灯
    GPIO.output(LED_GPIO, 1)
    time.sleep(1)
    # 出力を0にしてLED消灯
    GPIO.output(LED_GPIO, 0)
    time.sleep(1)

# ピンの設定を初期化(この処理をしないと、次回 プログラムを実行した時に「ピンが使用中」のエラーが発生する.
GPIO.cleanup()

これでLチカができました。

【エラー】RuntimeError: Not running on a RPi!

上記のLチカpythonコードを実行すると、以下のエラーが発生することがあります。

RuntimeError: Not running on a RPi!

これはpythonコードをsudo実行していない場合に発生するエラーで、sudo実行すると解消されます。

不快指数でLチカ

そして、この記事の趣旨である、不快指数の値を基にLチカできるようにしました。

最終的なpythonコードは以下になりました。

from influxdb import InfluxDBClient
import struct
import sys
from bluepy.btle import Scanner, DefaultDelegate
import time
from datetime import datetime
import RPi.GPIO as GPIO

LED_GPIO = 17

# データベースに接続
dbclient = InfluxDBClient(host='localhost', port=8086, database='tempDB')

class SwitchbotScanDelegate(DefaultDelegate):
    #コンストラクタ
    def __init__(self, macaddr):
        DefaultDelegate.__init__(self)
        #センサデータ保持用変数
        self.sensorValue = None
        self.macaddr = macaddr
 
    # スキャンハンドラー
    def handleDiscovery(self, dev, isNewDev, isNewData):
        # 対象Macアドレスのデバイスが見つかったら
        if dev.addr == self.macaddr:
            # アドバタイズデータを取り出し
            for (adtype, desc, value) in dev.getScanData():  
                #環境センサのとき、データ取り出しを実行
                if desc == '16b Service Data':
                    #センサデータ取り出し
                    self._decodeSensorData(value)
 
    # センサデータを取り出してdict形式に変換
    def _decodeSensorData(self, valueStr):
        #文字列からセンサデータ(4文字目以降)のみ取り出し、バイナリに変換
        valueBinary = bytes.fromhex(valueStr[4:])
        #バイナリ形式のセンサデータを数値に変換
        batt = valueBinary[2] & 0b01111111
        isTemperatureAboveFreezing = valueBinary[4] & 0b10000000
        temp = ( valueBinary[3] & 0b00001111 ) / 10 + ( valueBinary[4] & 0b01111111 )
        if not isTemperatureAboveFreezing:
            temp = -temp
        humid = valueBinary[5] & 0b01111111
        #dict型に格納
        self.sensorValue = {
            'SensorType': 'SwitchBot',
            'MacAddr': self.macaddr,
            'Temperature': temp,
            'Humidity': humid,
            'Battery': batt
        }


def send_temp_data(macaddr):
    # macaddr指定した温湿度センサの値を取得
    macaddr = str.lower(macaddr)
    scanner = Scanner().withDelegate( SwitchbotScanDelegate(macaddr) )
    #スキャン(timeout9s)
    scanner.scan( 9.0 )
    temp=0
    #出力
    try:
        temp = str(scanner.delegate.sensorValue['Temperature'])
    except TypeError as e:
        print("Temporarily lost communication with Bluetooth temperature device.",e) 
    except:
        import traceback
        traceback.print_exc()
    return  temp

def send_humi_data(macaddr):
    # macaddr指定した温湿度センサの値を取得
    macaddr = str.lower(macaddr)
    scanner = Scanner().withDelegate( SwitchbotScanDelegate(macaddr))
    #スキャン(timeout2s)
    scanner.scan( 9.0 )
    humi=0
    #出力
    try:
        humi = str(scanner.delegate.sensorValue['Humidity'])
    except TypeError as e:
        print("Temporarily lost communication with Bluetooth humidity device.",e) 
    except:
        import traceback
        traceback.print_exc()
    return  humi
 
if __name__ == '__main__':
    GPIO.setmode(GPIO.BCM)
    GPIO.setup(LED_GPIO, GPIO.OUT)

    GPIO.output(LED_GPIO, 0)

    while True:
        temp = float(send_temp_data('XX:XX:XX:XX:XX:XX'))
        humi = float(send_humi_data('XX:XX:XX:XX:XX:XX'))
        discomfort = 0.81*temp + 0.01*humi*( 0.99*temp - 14.3 ) + 46.3
        print(discomfort)

        json_body = [
            {
                "measurement": "temp_measurement",
                "time": datetime.utcnow(),
                "fields": {
                    "temp": temp,
                    "humi": humi,
                    "discomfort": discomfort,
                }
            }
        ]
        dbclient.write_points(json_body)

        if discomfort <= 60 or 75 <= discomfort:
            # 出力1にしてLED点灯
            GPIO.output(LED_GPIO, 1)
        else:
            # 出力を0にしてLED消灯
            GPIO.output(LED_GPIO, 0)

        time.sleep(10)

試しに、不快指数の計算部分でわざと大きくマイナスさせて低い値になるようにしました。

そして、不快指数、

53.31649999999999

で点灯することが確認できました。

(上のLチカの画像とほぼ一緒ですが、ちゃんとpythonコードを実行した際に撮った写真です)

ちなみに、外出時などはエアコンを切ります。その際は点灯する可能性はあります。

ですので、その時に点灯し続けるのはアレなので、今は切ってます。

まとめ

今回は不快指数の値が悪い時に知らせるLEDライトの実装を行いました。

今後は、外出時や睡眠時などにはLEDライトが点灯しないような実装をし、実際の生活で起動し続けられるように修正していきたいと思います。

タイトルとURLをコピーしました