Raspberry Piを100%活用しよう（Vol.89掲載）

著者：米田聡

本連載では、人気のマイコンボード「Raspberry Pi Pico W/WH」を活用していきます。同ボードは、無線LANやBluetoothの通信機能を搭載し、入手しやすく価格も手頃なので、IoT機器を自作するのに最適なハードウエアです。第5回は、DHT-11で測った湿温度をスマートフォンに通知します。

シェルスクリプトマガジン Vol.89は以下のリンク先でご購入できます。

図4　LINE Notifyを使って通知を送る「notify()」関数を定義したPythonプログラム

import usocket as socket
from urequests import post
from micropython import const

TOKEN = const("your_access_token")

def notify(msg):
    url = 'https://notify-api.line.me/api/notify'
    rheaders = {
            'Content-type':'application/x-www-form-urlencoded',
            'Authorization':'Bearer ' + TOKEN
    }

    message = "message=" + msg
    req = post(url,headers=rheaders,data=message)
    return req.status_code

import usocket as socket

from urequests import post

from micropython import const

TOKEN = const("your_access_token")

def notify(msg):

url = 'https://notify-api.line.me/api/notify'

rheaders = {

'Content-type':'application/x-www-form-urlencoded',

'Authorization':'Bearer ' + TOKEN

}

message = "message=" + msg

req = post(url,headers=rheaders,data=message)

return req.status_code

図6　1時間おきに湿温度をスマートフォンに通知するサンプルプログラム（sample.py）

from machine import Pin
from machine import Timer
from micropython import const
from micropython import schedule
import _thread
import dht
import line
import time

DHT_PIN = const(15)
TICK = const(60)

class Climate():
    def __init__(self, dht_gpio=DHT_PIN, tick=TICK, callback=None, expire=TICK):
        self.dht = dht.DHT11(Pin(dht_gpio))
        self.tick = tick
        self.lock = _thread.allocate_lock()
        self.timer = Timer(period=tick*1000, mode=Timer.PERIODIC, callback=self._timer_handler)
        self._humidity = 0
        self._temperature = 0
        self._measurement_dt = 0
        self.measure()
        self._callback_func = callback
        self._expire = expire
        self._counter = 0
    
    def set_callback(self, callback, expire):
        self._callback_func = callback
        self._expire = expire
    
    def measure(self, arg=0):
        self.dht.measure()
        self.lock.acquire()
        self._humidity = self.dht.humidity()
        self._temperature = self.dht.temperature()
        self._measurement_dt = time.time()
        self.lock.release()
    
    def _timer_handler(self,t):
        schedule(self.measure, 0)
        self._counter -= 1
        if self._counter <= 0:
            self._counter = self._expire
            if self._callback_func is not None:
                schedule(self._callback_func, self)
    
    @property
    def temp(self):
        self.lock.acquire()
        rval = self._temperature
        self.lock.release()
        return rval
    
    @property
    def hum(self):
        self.lock.acquire()
        rval = self._humidity
        self.lock.release()
        return rval
    
    @property
    def measurement_date_time(self):
        self.lock.acquire()
        rval = self._measurement_dt
        self.lock.release()
        return rval
        

def mycalback(c):
    tm = time.localtime(c.measurement_date_time)
    line.notify("temp=%d,hum=%d@%4d/%02d/%02d_%02d:%02d:%02d" % (c.temp, c.hum, tm[0],tm[1],tm[2],tm[3],tm[4],tm[5]))
    
if __name__ == '__main__':
    cl = Climate(callback = mycalback)
    
    while(True):
        machine.idle()

from machine import Pin

from machine import Timer

from micropython import const

from micropython import schedule

import _thread

import dht

import line

import time

DHT_PIN = const(15)

TICK = const(60)

class Climate():

def __init__(self, dht_gpio=DHT_PIN, tick=TICK, callback=None, expire=TICK):

self.dht = dht.DHT11(Pin(dht_gpio))

self.tick = tick

self.lock = _thread.allocate_lock()

self.timer = Timer(period=tick*1000, mode=Timer.PERIODIC, callback=self._timer_handler)

self._humidity = 0

self._temperature = 0

self._measurement_dt = 0

self.measure()

self._callback_func = callback

self._expire = expire

self._counter = 0

def set_callback(self, callback, expire):

self._callback_func = callback

self._expire = expire

def measure(self, arg=0):

self.dht.measure()

self.lock.acquire()

self._humidity = self.dht.humidity()

self._temperature = self.dht.temperature()

self._measurement_dt = time.time()

self.lock.release()

def _timer_handler(self,t):

schedule(self.measure, 0)

self._counter -= 1

if self._counter <= 0:

self._counter = self._expire

if self._callback_func is not None:

schedule(self._callback_func, self)

@property

def temp(self):

self.lock.acquire()

rval = self._temperature

self.lock.release()

return rval

@property

def hum(self):

self.lock.acquire()

rval = self._humidity

self.lock.release()

return rval

@property

def measurement_date_time(self):

self.lock.acquire()

rval = self._measurement_dt

self.lock.release()

return rval

def mycalback(c):

tm = time.localtime(c.measurement_date_time)

line.notify("temp=%d,hum=%d@%4d/%02d/%02d_%02d:%02d:%02d" % (c.temp, c.hum, tm[0],tm[1],tm[2],tm[3],tm[4],tm[5]))

if __name__ == '__main__':

cl = Climate(callback = mycalback)

while(True):

machine.idle()