SCD41でCO2センサーをAPIまでセットアップした
こちらのセンサーを使ってCO2測定が簡単にできるようになったので、Claudeに手伝ってもらってAPIを使えるところまでセットアップしたのでメモ。
ハードウェアインストールについては割愛。
ログファイルへの出力を設定
このあたりの記事を参考にしました。
- IndoorCorgi/cgsensor: Raspberry Pi拡張基板に搭載されているセンサーを一括して制御できるPythonパッケージとコマンドラインツールです。Python package and command line tool to control sensors on Raspberry Pi
- SCD41 CO2センサーとラズパイで二酸化炭素濃度を測定する – Indoor Corgi
まずは実行してログ出力する用の.shを作成
$ cat cgsensor_logger.sh
#!/bin/bash LOG_FILE="/home/choco/logs/sensor.log"
cgsensor all -c 10 -v -f "$LOG_FILE
-c 10 で10秒ごとなのだが、実用だともっと長くていいと思う。
実行権限をつける
サービスにして自動起動させる
$ cat /etc/systemd/system/cgsensor-logger.service
[Unit]
Description=CGSensor Logger
After=network.target
[Service]
Type=simple
ExecStart=/home/choco/cgsensor_logger.sh
Restart=always
RestartSec=3
User=choco
WorkingDirectory=/home/choco
[Install]
WantedBy=multi-user.target
有効化して再起動
sudo systemctl daemon-reload
sudo systemctl enable --now cgsensor-logger.service
sudo systemctl status cgsensor-logger.service
APIでLAN内から使えるようにする
/home/choco/sensor_api.py を作成
#!/usr/bin/env python3
import http.server
import socketserver
import json
from urllib.parse import urlparse, parse_qs
LOG_FILE = "/home/choco/logs/sensor.log"
PORT = 8080
DEFAULT_HISTORY_COUNT = 50
MAX_HISTORY_COUNT = 1000
def parse_line(line):
parts = [p.strip() for p in line.strip().split(",")]
if len(parts) < 4:
raise ValueError("フォーマットが想定と違います: " + line)
return {
"datetime": parts[0],
"co2_ppm": parts[1],
"temperature_c": parts[2],
"humidity_percent": parts[3],
}
class SensorHandler(http.server.BaseHTTPRequestHandler):
def _send_json(self, status, data):
body = json.dumps(data, ensure_ascii=False).encode("utf-8")
self.send_response(status)
self.send_header("Content-Type", "application/json; charset=utf-8")
self.send_header("Content-Length", str(len(body)))
self.end_headers()
self.wfile.write(body)
def do_GET(self):
parsed = urlparse(self.path)
try:
if parsed.path == "/sensor":
with open(LOG_FILE, "r") as f:
lines = [l for l in f if l.strip()]
if not lines:
raise ValueError("ログが空です")
self._send_json(200, parse_line(lines[-1]))
elif parsed.path == "/sensor/history":
qs = parse_qs(parsed.query)
count = DEFAULT_HISTORY_COUNT
if "count" in qs:
try:
count = int(qs["count"][0])
except ValueError:
raise ValueError("countは整数で指定してください")
count = max(1, min(count, MAX_HISTORY_COUNT))
with open(LOG_FILE, "r") as f:
lines = [l for l in f if l.strip()]
if not lines:
raise ValueError("ログが空です")
target_lines = lines[-count:]
data = [parse_line(l) for l in target_lines]
self._send_json(200, {"count": len(data), "data": data})
else:
self._send_json(404, {"error": "Not Found"})
except Exception as e:
self._send_json(500, {"error": str(e)})
def log_message(self, format, *args):
pass
if __name__ == "__main__":
with socketserver.TCPServer(("0.0.0.0", PORT), SensorHandler) as httpd:
print(f"Serving on port {PORT}")
httpd.serve_forever()
実行権限をつける
これもサービスにする
/etc/systemd/system/sensor-api.service を作成
[Unit]
Description=CGSensor API Server
After=network.target
[Service]
Type=simple
ExecStart=/usr/bin/python3 /home/choco/sensor_api.py
Restart=always
RestartSec=3
User=choco
WorkingDirectory=/home/choco
[Install]
WantedBy=multi-user.target
有効化して再起動
sudo systemctl daemon-reload
sudo systemctl enable --now sensor-api.service
sudo systemctl status sensor-api.service
使い方
LAN内の別のサーバーからcurlでJSON形式のレスポンスを取得できる。
直近一件
直近50件
直近10件 (最大1000件)
次はこれをWebサイトとかにリアルタイムチャートで表示して、まめに部屋の換気とかしたいですね。