以下に、そのままコピペで使える 完成版2ファイル をまとめます。
機能は次のとおりです。
親機 parent_monitor.py
-
GPIOでトグルスイッチ監視
-
ON(HIGH)でイベント発生
-
メール送信
-
子機へHTTP通知
-
ログ保存
-
失敗時は再送キューへ保存
-
定期的に自動再送
子機 child_display.py
-
親機からHTTP通知を受信
-
GUI全画面表示
-
ログ保存
-
受信内容を画面に反映
-
GUI上で簡単なクリア操作あり
1. parent_monitor.py
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
import os
import time
import json
import socket
import logging
import threading
import smtplib
from logging.handlers import RotatingFileHandler
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
import requests
import RPi.GPIO as GPIO
# =========================================================
# 設定
# =========================================================
# GPIO
GPIO_PIN = 17
# 子機 Raspberry Pi の通知先
CHILD_PI_URLS = [
"http://192.168.1.50:5000/alert",
# "http://192.168.1.51:5000/alert", # 必要なら追加
]
# メール設定
MAIL_FROM = "your_sender@gmail.com"
MAIL_TO = "your_receiver@example.com"
SMTP_SERVER = "smtp.gmail.com"
SMTP_PORT = 587
SMTP_USER = "your_sender@gmail.com"
SMTP_PASSWORD = "your_app_password"
# 動作設定
DEBOUNCE_SEC = 0.2
RETRIGGER_GUARD_SEC = 5.0
RETRY_INTERVAL_SEC = 30
MAX_RETRY_COUNT = 20
REQUEST_TIMEOUT_SEC = 5
# 保存ファイル
BASE_DIR = "/home/pi"
LOG_FILE = os.path.join(BASE_DIR, "parent_monitor.log")
QUEUE_FILE = os.path.join(BASE_DIR, "alert_queue.json")
HOSTNAME = socket.gethostname()
# =========================================================
# グローバル
# =========================================================
last_trigger_time = 0
queue_lock = threading.Lock()
# =========================================================
# ログ設定
# =========================================================
logger = logging.getLogger("parent_monitor")
logger.setLevel(logging.INFO)
logger.propagate = False
if not logger.handlers:
formatter = logging.Formatter("%(asctime)s [%(levelname)s] %(message)s")
file_handler = RotatingFileHandler(
LOG_FILE,
maxBytes=1024 * 1024,
backupCount=5,
encoding="utf-8"
)
file_handler.setFormatter(formatter)
console_handler = logging.StreamHandler()
console_handler.setFormatter(formatter)
logger.addHandler(file_handler)
logger.addHandler(console_handler)
# =========================================================
# キュー管理
# =========================================================
def load_queue():
if not os.path.exists(QUEUE_FILE):
return []
try:
with open(QUEUE_FILE, "r", encoding="utf-8") as f:
data = json.load(f)
if isinstance(data, list):
return data
logger.warning("キューファイル形式が不正のため初期化します")
return []
except Exception as e:
logger.error(f"キュー読込失敗: {e}")
return []
def save_queue(queue):
tmp_file = QUEUE_FILE + ".tmp"
try:
with open(tmp_file, "w", encoding="utf-8") as f:
json.dump(queue, f, ensure_ascii=False, indent=2)
os.replace(tmp_file, QUEUE_FILE)
except Exception as e:
logger.error(f"キュー保存失敗: {e}")
def enqueue_retry(task_type, payload, destination=""):
task = {
"task_type": task_type, # "email" or "child_notify"
"payload": payload,
"destination": destination,
"retry_count": 0,
"last_error": "",
"created_at": time.strftime("%Y-%m-%d %H:%M:%S"),
"last_try_at": ""
}
with queue_lock:
queue = load_queue()
queue.append(task)
save_queue(queue)
logger.warning(f"再送キュー追加: task_type={task_type}, destination={destination}")
def replace_queue(new_queue):
with queue_lock:
save_queue(new_queue)
# =========================================================
# 送信処理
# =========================================================
def send_email(subject, body):
msg = MIMEMultipart()
msg["From"] = MAIL_FROM
msg["To"] = MAIL_TO
msg["Subject"] = subject
msg.attach(MIMEText(body, "plain", "utf-8"))
with smtplib.SMTP(SMTP_SERVER, SMTP_PORT, timeout=20) as server:
server.starttls()
server.login(SMTP_USER, SMTP_PASSWORD)
server.sendmail(MAIL_FROM, MAIL_TO, msg.as_string())
def notify_child(url, payload):
response = requests.post(url, json=payload, timeout=REQUEST_TIMEOUT_SEC)
response.raise_for_status()
# =========================================================
# アラート内容生成
# =========================================================
def build_alert_data():
alert_time = time.strftime("%Y-%m-%d %H:%M:%S")
email_subject = f"[Alert] Raspberry Pi Switch ON ({HOSTNAME})"
email_body = (
"トグルスイッチがONになりました。\n\n"
f"Host: {HOSTNAME}\n"
f"Time: {alert_time}\n"
f"GPIO: {GPIO_PIN}\n"
)
child_payload = {
"title": "警報",
"message": "親機のトグルスイッチがONになりました",
"time": alert_time,
"host": HOSTNAME,
"gpio": GPIO_PIN
}
return {
"email_subject": email_subject,
"email_body": email_body,
"child_payload": child_payload
}
# =========================================================
# アラート処理
# =========================================================
def process_alert():
alert = build_alert_data()
logger.info("アラート発生: スイッチON")
# メール送信
try:
send_email(alert["email_subject"], alert["email_body"])
logger.info("メール送信成功")
except Exception as e:
logger.error(f"メール送信失敗: {e}")
enqueue_retry(
task_type="email",
payload={
"subject": alert["email_subject"],
"body": alert["email_body"]
},
destination=MAIL_TO
)
# 子機通知
for url in CHILD_PI_URLS:
try:
notify_child(url, alert["child_payload"])
logger.info(f"子機通知成功: {url}")
except Exception as e:
logger.error(f"子機通知失敗: {url}, error={e}")
enqueue_retry(
task_type="child_notify",
payload=alert["child_payload"],
destination=url
)
# =========================================================
# GPIOコールバック
# =========================================================
def on_switch_on(channel):
global last_trigger_time
now = time.time()
if now - last_trigger_time < RETRIGGER_GUARD_SEC:
logger.info("短時間の連続トリガーのため無視")
return
time.sleep(DEBOUNCE_SEC)
if GPIO.input(GPIO_PIN) == GPIO.HIGH:
logger.info("GPIO HIGH検出")
process_alert()
last_trigger_time = now
# =========================================================
# 再送ワーカー
# =========================================================
def retry_worker():
while True:
time.sleep(RETRY_INTERVAL_SEC)
with queue_lock:
queue = load_queue()
if not queue:
continue
logger.info(f"再送処理開始: {len(queue)}件")
remaining = []
for item in queue:
task_type = item.get("task_type")
payload = item.get("payload", {})
destination = item.get("destination", "")
retry_count = item.get("retry_count", 0)
if retry_count >= MAX_RETRY_COUNT:
logger.error(
f"最大再送回数超過のため破棄: task_type={task_type}, destination={destination}"
)
continue
try:
if task_type == "email":
send_email(payload["subject"], payload["body"])
logger.info(
f"再送成功: task_type=email, retry_count={retry_count + 1}, destination={destination}"
)
elif task_type == "child_notify":
notify_child(destination, payload)
logger.info(
f"再送成功: task_type=child_notify, retry_count={retry_count + 1}, destination={destination}"
)
else:
logger.error(f"不明なtask_type: {task_type}")
continue
except Exception as e:
item["retry_count"] = retry_count + 1
item["last_error"] = str(e)
item["last_try_at"] = time.strftime("%Y-%m-%d %H:%M:%S")
remaining.append(item)
logger.warning(
f"再送失敗: task_type={task_type}, retry_count={item['retry_count']}, "
f"destination={destination}, error={e}"
)
replace_queue(remaining)
# =========================================================
# 初期化
# =========================================================
def initialize_gpio():
GPIO.setwarnings(False)
GPIO.setmode(GPIO.BCM)
GPIO.setup(GPIO_PIN, GPIO.IN, pull_up_down=GPIO.PUD_DOWN)
GPIO.add_event_detect(
GPIO_PIN,
GPIO.RISING,
callback=on_switch_on,
bouncetime=300
)
# =========================================================
# メイン
# =========================================================
def main():
logger.info("========================================")
logger.info("親機監視プログラム起動")
logger.info(f"HOSTNAME={HOSTNAME}")
logger.info(f"GPIO_PIN={GPIO_PIN}")
logger.info(f"CHILD_PI_URLS={CHILD_PI_URLS}")
initialize_gpio()
thread = threading.Thread(target=retry_worker, daemon=True)
thread.start()
logger.info("GPIO監視開始")
try:
while True:
time.sleep(1)
except KeyboardInterrupt:
logger.info("ユーザー操作で停止")
finally:
GPIO.cleanup()
logger.info("GPIOクリーンアップ完了")
if __name__ == "__main__":
main()
2. child_display.py
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
import os
import json
import queue
import threading
import logging
from logging.handlers import RotatingFileHandler
import tkinter as tk
from flask import Flask, request, jsonify
# =========================================================
# 設定
# =========================================================
BASE_DIR = "/home/pi"
LOG_FILE = os.path.join(BASE_DIR, "child_display.log")
RECEIVED_LOG_JSON = os.path.join(BASE_DIR, "child_received_events.json")
LISTEN_HOST = "0.0.0.0"
LISTEN_PORT = 5000
WINDOW_TITLE = "Alert Display"
# =========================================================
# ログ設定
# =========================================================
logger = logging.getLogger("child_display")
logger.setLevel(logging.INFO)
logger.propagate = False
if not logger.handlers:
formatter = logging.Formatter("%(asctime)s [%(levelname)s] %(message)s")
file_handler = RotatingFileHandler(
LOG_FILE,
maxBytes=1024 * 1024,
backupCount=5,
encoding="utf-8"
)
file_handler.setFormatter(formatter)
console_handler = logging.StreamHandler()
console_handler.setFormatter(formatter)
logger.addHandler(file_handler)
logger.addHandler(console_handler)
# =========================================================
# 受信データ保存
# =========================================================
def append_received_event(data):
events = []
if os.path.exists(RECEIVED_LOG_JSON):
try:
with open(RECEIVED_LOG_JSON, "r", encoding="utf-8") as f:
loaded = json.load(f)
if isinstance(loaded, list):
events = loaded
except Exception as e:
logger.error(f"受信JSON読込失敗: {e}")
events.append(data)
# 古いデータが増えすぎないように末尾200件だけ保存
events = events[-200:]
tmp_file = RECEIVED_LOG_JSON + ".tmp"
try:
with open(tmp_file, "w", encoding="utf-8") as f:
json.dump(events, f, ensure_ascii=False, indent=2)
os.replace(tmp_file, RECEIVED_LOG_JSON)
except Exception as e:
logger.error(f"受信JSON保存失敗: {e}")
# =========================================================
# Flask
# =========================================================
app = Flask(__name__)
gui_queue = queue.Queue()
@app.route("/alert", methods=["POST"])
def alert():
try:
data = request.get_json(force=True)
if not isinstance(data, dict):
logger.warning("不正なJSON受信")
return jsonify({"status": "error", "message": "invalid json"}), 400
logger.info(f"通知受信: {data}")
append_received_event(data)
gui_queue.put(data)
return jsonify({"status": "ok"}), 200
except Exception as e:
logger.exception(f"通知処理失敗: {e}")
return jsonify({"status": "error", "message": str(e)}), 500
@app.route("/health", methods=["GET"])
def health():
return jsonify({"status": "ok"}), 200
def run_flask():
logger.info(f"HTTP受信開始: http://{LISTEN_HOST}:{LISTEN_PORT}")
app.run(host=LISTEN_HOST, port=LISTEN_PORT, debug=False, use_reloader=False)
# =========================================================
# Tkinter GUI
# =========================================================
class AlertDisplayApp:
def __init__(self, root):
self.root = root
self.root.title(WINDOW_TITLE)
self.root.attributes("-fullscreen", True)
self.root.configure(bg="black")
self.default_bg = "black"
self.alert_bg = "darkred"
self.frame = tk.Frame(self.root, bg=self.default_bg)
self.frame.pack(fill="both", expand=True)
self.title_label = tk.Label(
self.frame,
text="待機中",
font=("Arial", 40, "bold"),
fg="white",
bg=self.default_bg
)
self.title_label.pack(pady=40)
self.message_label = tk.Label(
self.frame,
text="通知を待っています",
font=("Arial", 28),
fg="yellow",
bg=self.default_bg,
wraplength=1200,
justify="center"
)
self.message_label.pack(pady=30)
self.info_label = tk.Label(
self.frame,
text="",
font=("Arial", 20),
fg="lightgreen",
bg=self.default_bg,
justify="center"
)
self.info_label.pack(pady=20)
self.help_label = tk.Label(
self.frame,
text="C: 画面クリア Esc: 終了",
font=("Arial", 18),
fg="lightgray",
bg=self.default_bg
)
self.help_label.pack(side="bottom", pady=20)
self.root.bind("<Escape>", self.on_escape)
self.root.bind("<c>", self.on_clear)
self.root.bind("<C>", self.on_clear)
self.poll_gui_queue()
def set_bg_all(self, color):
self.root.configure(bg=color)
self.frame.configure(bg=color)
self.title_label.configure(bg=color)
self.message_label.configure(bg=color)
self.info_label.configure(bg=color)
self.help_label.configure(bg=color)
def update_alert(self, data):
title = data.get("title", "通知")
message = data.get("message", "メッセージなし")
alert_time = data.get("time", "")
host = data.get("host", "")
gpio = data.get("gpio", "")
info_lines = []
if alert_time:
info_lines.append(f"時刻: {alert_time}")
if host:
info_lines.append(f"送信元: {host}")
if gpio != "":
info_lines.append(f"GPIO: {gpio}")
self.title_label.config(text=title)
self.message_label.config(text=message)
self.info_label.config(text="\n".join(info_lines))
self.set_bg_all(self.alert_bg)
logger.info("GUI更新完了")
def clear_screen(self, event=None):
self.title_label.config(text="待機中")
self.message_label.config(text="通知を待っています")
self.info_label.config(text="")
self.set_bg_all(self.default_bg)
logger.info("画面クリア")
def on_clear(self, event=None):
self.clear_screen()
def on_escape(self, event=None):
logger.info("GUI終了")
self.root.destroy()
def poll_gui_queue(self):
try:
while True:
data = gui_queue.get_nowait()
self.update_alert(data)
except queue.Empty:
pass
self.root.after(200, self.poll_gui_queue)
# =========================================================
# メイン
# =========================================================
def main():
logger.info("========================================")
logger.info("子機表示プログラム起動")
flask_thread = threading.Thread(target=run_flask, daemon=True)
flask_thread.start()
root = tk.Tk()
app_ui = AlertDisplayApp(root)
try:
root.mainloop()
except KeyboardInterrupt:
logger.info("ユーザー操作で停止")
finally:
logger.info("子機表示プログラム終了")
if __name__ == "__main__":
main()
3. インストール
親機
子機
4. 変更が必要な箇所
親機 parent_monitor.py
最低でもここを変更してください。
5. 起動方法
子機を先に起動
親機を起動
6. 保存されるログ
親機
-
通常ログ
/home/pi/parent_monitor.log
-
再送キュー
/home/pi/alert_queue.json
子機
-
通常ログ
/home/pi/child_display.log
-
受信履歴JSON
/home/pi/child_received_events.json
7. 動作確認
子機の疎通確認
親機または別PCから:
{"status":"ok"} が返れば受信待機できています。
手動通知テスト
これで子機GUIが赤背景で更新されます。
8. 運用メモ
-
スイッチONで即送信
-
失敗時は再送キューへ保存
-
30秒ごとに自動再送
-
最大20回まで再試行
-
子機は受信内容をログにもJSONにも保存
