センサーボードで学ぶ電子回路の制御（Vol.66掲載）

著者：米田聡

シェルスクリプトマガジンでは、小型コンピュータボード「Raspberry Pi」（ラズパイ）向けのセンサー搭載拡張ボード「ラズパイセンサーボード」を制作しました。第13 回では、同ボードに搭載された湿温度・気圧センサー「BME280」から取得したデータをグラフ化する準備をします。

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

図2　PythonからSQLiteを操作するサンプルプログラム（test_sqlite3.py）

import sqlite3
 
DBNAME = "test.sqlite3"         # データベースファイル名
conn = sqlite3.connect(DBNAME)  # SQLite 3に接続
cur = conn.cursor()             # カーソルを得る
cur.execute("SQL文")            # SQLを発行
conn.commit()                   # コミット
data = cur.fetchall()           # 結果をdataに格納
conn.close()                    # データベースを閉じる

import sqlite3

DBNAME = "test.sqlite3" # データベースファイル名

conn = sqlite3.connect(DBNAME) # SQLite 3に接続

cur = conn.cursor() # カーソルを得る

cur.execute("SQL文") # SQLを発行

conn.commit() # コミット

data = cur.fetchall() # 結果をdataに格納

conn.close() # データベースを閉じる

図3　温度、湿度、気圧のデータをSQLiteに保存するプログラム（storebme.py）

import time, os, sys, signal
import smbus2
import bme280
import sqlite3
 
BME280_ADDR = 0x76
BUS_NO = 1
DBNAME = 'weather.sqlite3'
 
def store_values(values):
  conn = sqlite3.connect(DBNAME)
  cur = conn.cursor()
  sql = "INSERT INTO bme(dt,temp,hum,press) VALUES(datetime('now', '+9 hours'),?,?,?)"
  cur.execute(sql,(values[0],values[1],values[2]))
  conn.commit()
  conn.close()
 
def signal_handler(sig, handler):
  sys.exit()
 
# テーブル作成
if not os.path.exists(DBNAME):
  conn = sqlite3.connect(DBNAME)
  cur = conn.cursor()
  cur.execute("CREATE TABLE bme(id INTEGER PRIMARY KEY AUTOINCREMENT, dt TEXT,temp REAL, hum REAL, press REAL)")
  conn.commit()
  conn.close()
 
# BME280
i2c = smbus2.SMBus(BUS_NO)
bme280.load_calibration_params(i2c, BME280_ADDR)
 
# signal
signal.signal(signal.SIGINT, signal_handler)
signal.signal(signal.SIGTERM, signal_handler)
 
# メインループ
while True:
  data = bme280.sample(i2c, BME280_ADDR)
  store_values([data.temperature,data.humidity,data.pressure ])
  time.sleep(60)

import time, os, sys, signal

import smbus2

import bme280

import sqlite3

BME280_ADDR = 0x76

BUS_NO = 1

DBNAME = 'weather.sqlite3'

def store_values(values):

conn = sqlite3.connect(DBNAME)

cur = conn.cursor()

sql = "INSERT INTO bme(dt,temp,hum,press) VALUES(datetime('now', '+9 hours'),?,?,?)"

cur.execute(sql,(values[0],values[1],values[2]))

conn.commit()

conn.close()

def signal_handler(sig, handler):

sys.exit()

# テーブル作成

if not os.path.exists(DBNAME):

conn = sqlite3.connect(DBNAME)

cur = conn.cursor()

cur.execute("CREATE TABLE bme(id INTEGER PRIMARY KEY AUTOINCREMENT, dt TEXT,temp REAL, hum REAL, press REAL)")

conn.commit()

conn.close()

# BME280

i2c = smbus2.SMBus(BUS_NO)

bme280.load_calibration_params(i2c, BME280_ADDR)

# signal

signal.signal(signal.SIGINT, signal_handler)

signal.signal(signal.SIGTERM, signal_handler)

# メインループ

while True:

data = bme280.sample(i2c, BME280_ADDR)

store_values([data.temperature,data.humidity,data.pressure ])

time.sleep(60)