Beer monitor script

beer_monitor.py

# coding: utf-8
# python3

import smbus
import time
import requests
import json

SLACK_URL = "https://hooks.slack.com/services/********"
DURATION = 300  # 定期通知の間隔 (seconds)
GOAL_TEMP = 6  # 目標温度(℃)

i2c = smbus.SMBus(1)
address = 0x48

# 温度を読み取る
# 参考: I2C温度センサー(ADT7410)を使う | Make. http://make.bcde.jp/raspberry-pi/i2c温度センサーadt7410を使う/
def read_temp():
    global i2c
    global address

    block = i2c.read_i2c_block_data(address, 0x00, 12)
    temp = (block[0] << 8 | block[1]) >> 3
    if (temp > 4096):
        temp -= 8192

    return temp / 16.0

# Slackに通知する
def send2slack(payload):
    return requests.post(SLACK_URL, data={"payload": json.dumps(payload)})

def notify_temp(temp):
    return send2slack({"text": "ビール温度: %.2f ℃" % (temp)})

def notify_completed(temp):
    return send2slack({"text": "ビールが冷えました :sparkles: :beers: :sparkles: (%.2f ℃)" % (temp)})

def notify_start(temp):
    send2slack({"text": "ビール温度の監視を開始しました (現在温度: %.2f ℃,  目標温度: %.2f ℃)" % (temp, GOAL_TEMP)})

def notify_stop():
    send2slack({"text": "監視を終了しました。再開するにはI2Cデバイスを再接続してください"})

# 監視ループ
def read_loop():
    cnt = 0

    while True:
        temp = read_temp()
        print(temp)

        if (temp <= GOAL_TEMP):
            notify_completed(temp)
            while True:
                read_temp()
                time.sleep(1)

        cnt += 1

        if cnt >= DURATION:
            cnt = 0
            res = notify_temp(temp)
            print(res)

        time.sleep(1)

# 温度が読み取り可能か
def is_readable():
    try:
        read_temp()
        return True
    except IOError:
        return False

# 温度が読み取り可能になるまで待つ
def wait_until_readable():
    while True:
        if is_readable():
            time.sleep(10)
            if is_readable():
                return

        time.sleep(1)

while True:
    wait_until_readable()
    
    notify_start(read_temp())

    try:
        read_loop()
    except IOError:
        notify_stop()
    except Exception as e:
        print(type(e))
        print(e)