香川大学SLPからお届け！（Vol.90掲載）

著者：小河原直道

Raspberry Pi 5（以下、ラズパイ5）が日本でも発売され、私も購入しました。センサーとの接続やサーバーの構築をしてみたいと思ったため、湿温度/気圧/CO₂濃度を測定できるセンサーを接続し、その値をWebブラウザから確認できるシステムを作成しました。このシステムの作成ポイントについて解説します。

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

図8　BME280のチップIDを読み出すPythonコード

import spidev
# 読み取りは8ビット目を「1」に, 書き込みは「0」にする
def get_reg_addr(rw: str, addr: int) -> int:
  rw_flag = 0b1000_0000
  if rw == "w":
    rw_flag = 0b0000_0000
  return rw_flag | (addr & 0b0111_1111)
spi = spidev.SpiDev()
spi.open(0, 0) # bus0, cs0
spi.mode = 0   # mode0, CPOL=CPHA=0
spi.max_speed_hz = 10 * 10**6 # 10MHz
bme_id = spi.xfer2([get_reg_addr("r", 0xD0), 0])
print(f"id: {hex(bme_id[1])}")
spi.close()

import spidev

# 読み取りは8ビット目を「1」に, 書き込みは「0」にする

def get_reg_addr(rw: str, addr: int) -> int:

rw_flag = 0b1000_0000

if rw == "w":

rw_flag = 0b0000_0000

return rw_flag | (addr & 0b0111_1111)

spi = spidev.SpiDev()

spi.open(0, 0) # bus0, cs0

spi.mode = 0 # mode0, CPOL=CPHA=0

spi.max_speed_hz = 10 * 10**6 # 10MHz

bme_id = spi.xfer2([get_reg_addr("r", 0xD0), 0])

print(f"id: {hex(bme_id[1])}")

spi.close()

図9　BME280の気温データを読み出すPythonコード

import spidev
# 読み取りは8ビット目を「1」に, 書き込みは「0」にする
def get_reg_addr(rw: str, addr: int) -> int:
  rw_flag = 0b1000_0000
  if rw == "w":
    rw_flag = 0b0000_0000
  return rw_flag | (addr & 0b0111_1111)
spi = spidev.SpiDev()
spi.open(0, 0) # bus0, cs0
spi.mode = 0   # mode0, CPOL=CPHA=0
spi.max_speed_hz = 10 * 10**6 # 10MHz
# 湿度の測定を有効化。オーバーサンプリングx1
r = spi.xfer2([get_reg_addr("r", 0xF2), 0])[1]
ctrl_hum = r & 0b1111_1000 | 0b001
spi.xfer2([get_reg_addr("w", 0xF2), ctrl_hum])
# 気圧、気温の測定を有効化。オーバーサンプリングx1、ノーマルモードに
ctrl_meas = (0b001 << 5) + (0b001 << 3) + 0b11
spi.xfer2([get_reg_addr("w", 0xF4), ctrl_meas])
# 気温データの読み出し
temp_raw_bins = spi.xfer2([get_reg_addr("r", 0xFA), 0, 0, 0])
temp_raw = (temp_raw_bins[1] << 8+4) | (temp_raw_bins[2] << 4) |\
  ((temp_raw_bins[3]>>4) & 0b00001111)
print(f"気温: {temp_raw}")
spi.close()

import spidev

# 読み取りは8ビット目を「1」に, 書き込みは「0」にする

def get_reg_addr(rw: str, addr: int) -> int:

rw_flag = 0b1000_0000

if rw == "w":

rw_flag = 0b0000_0000

return rw_flag | (addr & 0b0111_1111)

spi = spidev.SpiDev()

spi.open(0, 0) # bus0, cs0

spi.mode = 0 # mode0, CPOL=CPHA=0

spi.max_speed_hz = 10 * 10**6 # 10MHz

# 湿度の測定を有効化。オーバーサンプリングx1

r = spi.xfer2([get_reg_addr("r", 0xF2), 0])[1]

ctrl_hum = r & 0b1111_1000 | 0b001

spi.xfer2([get_reg_addr("w", 0xF2), ctrl_hum])

# 気圧、気温の測定を有効化。オーバーサンプリングx1、ノーマルモードに

ctrl_meas = (0b001 << 5) + (0b001 << 3) + 0b11

spi.xfer2([get_reg_addr("w", 0xF4), ctrl_meas])

# 気温データの読み出し

temp_raw_bins = spi.xfer2([get_reg_addr("r", 0xFA), 0, 0, 0])

temp_raw = (temp_raw_bins[1] << 8+4) | (temp_raw_bins[2] << 4) |\

((temp_raw_bins[3]>>4) & 0b00001111)

print(f"気温: {temp_raw}")

spi.close()

図10　MH-Z19CのCO2濃度データを読み出すPythonコード

import serial
import sys
import time

serial = serial.Serial(port="/dev/ttyAMA0", timeout=5)
# 念のためバッファをリセットする
serial.reset_input_buffer()
serial.reset_output_buffer()
time.sleep(2) # 待ち時間は適宜増減して調整
s = bytearray.fromhex("FF 01 86 00 00 00 00 00 79")
serial.write(s)
r = serial.read(size=9)
serial.close()
if r[0] != 0xFF:
  print("invalid data\n")
  sys.exit(-1)
for_check=0
for i, b in enumerate(r):
  if i==0 or i==8:
    continue
  # チェックサムは1バイトのみ、あふれた分は捨てる
  for_check = for_check + b & 0b1111_1111
for_check = 0xFF - for_check + 1
if r[8] != for_check:
  print("checksum error\n")
  sys.exit(-1)
print(f"{r[2]*256 + r[3]} ppm")

import serial

import sys

import time

serial = serial.Serial(port="/dev/ttyAMA0", timeout=5)

# 念のためバッファをリセットする

serial.reset_input_buffer()

serial.reset_output_buffer()

time.sleep(2) # 待ち時間は適宜増減して調整

s = bytearray.fromhex("FF 01 86 00 00 00 00 00 79")

serial.write(s)

r = serial.read(size=9)

serial.close()

if r[0] != 0xFF:

print("invalid data\n")

sys.exit(-1)

for_check=0

for i, b in enumerate(r):

if i==0 or i==8:

continue

# チェックサムは1バイトのみ、あふれた分は捨てる

for_check = for_check + b & 0b1111_1111

for_check = 0xFF - for_check + 1

if r[8] != for_check:

print("checksum error\n")

sys.exit(-1)

print(f"{r[2]*256 + r[3]} ppm")

図12　測定データをデータベースに保存するPythonコード

import mariadb
import sys

try:
  conn = mariadb.connect(
    user     = ユーザー名,  password = パスワード,
    host     = "127.0.0.1", port     = 3306,
    database = "sensor1",   autocommit = True
  )
except mariadb.Error as e:
  print(f"db connect error: {e}")
  sys.exit(-1)
cur = conn.cursor()
sql = 'INSERT INTO sensor (datetime, temp, humid, press, co2) VALUES(NOW(), ?, ?, ?, ?)'
データ（temp_data、humid_data、press_data、co2_data）取得用コードをここに記述
try:
  cur.execute(sql, (temp_data, humid_data, press_data, co2_data))
except Exception as e:
  print(f"insert error: {e}")
finally:
  cur.close()
conn.close()

import mariadb

import sys

try:

conn = mariadb.connect(

user = ユーザー名, password = パスワード,

host = "127.0.0.1", port = 3306,

database = "sensor1", autocommit = True

)

except mariadb.Error as e:

print(f"db connect error: {e}")

sys.exit(-1)

cur = conn.cursor()

sql = 'INSERT INTO sensor (datetime, temp, humid, press, co2) VALUES(NOW(), ?, ?, ?, ?)'

データ（temp_data、humid_data、press_data、co2_data）取得用コードをここに記述

try:

cur.execute(sql, (temp_data, humid_data, press_data, co2_data))

except Exception as e:

print(f"insert error: {e}")

finally:

cur.close()

conn.close()

図13　「db/db_pool.js」ファイルに記述するコード

const mariadb = require("mariadb");
const pool = mariadb.createPool({
  host:     "localhost",
  user:     ユーザー名,
  password: パスワード,
  database: "sensor1",
  connectionLimit: 5,
});
module.exports = pool;

const mariadb = require("mariadb");

const pool = mariadb.createPool({

host: "localhost",

user: ユーザー名,

password: パスワード,

database: "sensor1",

connectionLimit: 5,

});

module.exports = pool;

図14　「route/api_v1.js」ファイルに記述するコード（抜粋）

const router = require("express").Router();
const pool = require("../db/db_pool");
function getColumnName(sensor){
  switch(sensor){
    case "all":return "*";
    case "temperature": return "datetime, temp"
    case "humidity": return "datetime, humid"
    case "pressure": return "datetime, press"
    case "co2": return "datetime, co2"
    default: throw new Error("sensor name error");
  }
}
router.get("/:sensor/latest", async(req, res) => {
  let sqlCol;
  try{
    sqlCol = getColumnName(req.params.sensor);
  } catch (e) { return res.status(404).send(`bad sensor name`); }
  try{
    let dbRes = await pool.query(
      `select {sqlCol} from sensor order by datetime desc limit 1`
    );
    // クエリー結果の最初のレコードをJSON形式で送る
    return res.status(200).json(dbRes[0]);
  } catch(err) {
    return res.status(500).json({mes:err.toString()});
  }
});
module.exports = router;

const router = require("express").Router();

const pool = require("../db/db_pool");

function getColumnName(sensor){

switch(sensor){

case "all":return "*";

case "temperature": return "datetime, temp"

case "humidity": return "datetime, humid"

case "pressure": return "datetime, press"

case "co2": return "datetime, co2"

default: throw new Error("sensor name error");

}

router.get("/:sensor/latest", async(req, res) => {

let sqlCol;

try{

sqlCol = getColumnName(req.params.sensor);

} catch (e) { return res.status(404).send(`bad sensor name`); }

try{

let dbRes = await pool.query(

`select {sqlCol} from sensor order by datetime desc limit 1`

);

// クエリー結果の最初のレコードをJSON形式で送る

return res.status(200).json(dbRes[0]);

} catch(err) {

return res.status(500).json({mes:err.toString()});

}

});

module.exports = router;

図15　「route/index.js」ファイルに記述するコード

const express = require('express');
const router = express.Router();
const path = require('path');
router.get("/", (req, res) => {
  res.sendFile(path.join(__dirname, "../view/index.html"));
});
router.use("/api/v1", require("./api_v1"));
module.exports = router;

const express = require('express');

const router = express.Router();

const path = require('path');

router.get("/", (req, res) => {

res.sendFile(path.join(__dirname, "../view/index.html"));

});

router.use("/api/v1", require("./api_v1"));

module.exports = router;

図16　「app.js」ファイルに記述するコード

const express = require("express");
const app = express();
const path = require('path');
app.use("/", require("./route/index"));
app.use(express.static(path.join(__dirname, "public")));
app.all("*", (req, res) => {
  return res.sendStatus(404);
});
app.listen(3000, () => {
  console.log("Listening at port 3000");
});

const express = require("express");

const app = express();

const path = require('path');

app.use("/", require("./route/index"));

app.use(express.static(path.join(__dirname, "public")));

app.all("*", (req, res) => {

return res.sendStatus(404);

});

app.listen(3000, () => {

console.log("Listening at port 3000");

});