猪咪闲来无事,创作"无中生有"阵。一边是一台可以播放声音的主机,一边是zero w绑着霍尔传感器。本来以为还要MQTT结果发现局域网里有更简单粗暴的实现方式,甚好。
zero端代码:
from gpiozero import DigitalInputDevice from signal import pause import time import json import urllib.request import logging import sys GPIO_PIN = 27 SERVER_URL = "http://192.168.50.2:5000/door/open" TOKEN = "your_shared_token_here" MIN_TRIGGER_INTERVAL = 1.0 last_trigger_ts = 0.0 LOG_FILE = "/home/pi/bug.txt" logging.basicConfig( filename=LOG_FILE, level=logging.ERROR, format="%(asctime)s [%(levelname)s] %(message)s", datefmt="%Y-%m-%d %H:%M:%S", ) def handle_uncaught_exception(exc_type, exc_value, exc_traceback): logging.error("Uncaught exception", exc_info=(exc_type, exc_value, exc_traceback)) sys.excepthook = handle_uncaught_exception sensor = DigitalInputDevice(GPIO_PIN, pull_up=True, bounce_time=0.05) def send_open_signal(): req = urllib.request.Request( SERVER_URL, data=json.dumps({"event": "door_open", "ts": time.time()}).encode("utf-8"), headers={ "Content-Type": "application/json", "X-Token": TOKEN }, method="POST", ) try: with urllib.request.urlopen(req, timeout=2): pass except Exception: logging.exception("Failed to send door_open signal") def on_inactive(): global last_trigger_ts now = time.time() if now - last_trigger_ts >= MIN_TRIGGER_INTERVAL: last_trigger_ts = now send_open_signal() sensor.when_deactivated = on_inactive pause()4B主机端:
import json import os import subprocess import threading from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer # ===== 配置区 ===== HOST = "0.0.0.0" PORT = 5000 TOKEN = "your_shared_token_here" AUDIO_FILE = "/home/pi/aaa.m4a" # ffplay 参数:无窗口、播完退出、日志静默 FFPLAY_CMD = [ "ffplay", "-nodisp", "-autoexit", "-loglevel", "error", AUDIO_FILE ] lock = threading.Lock() current_proc = None def play_audio_once(): global current_proc with lock: # 若正在播放,就忽略新的触发,避免叠音 if current_proc is not None and current_proc.poll() is None: return False if not os.path.exists(AUDIO_FILE): print("音频文件不存在:", AUDIO_FILE) return False current_proc = subprocess.Popen(FFPLAY_CMD) return True class Handler(BaseHTTPRequestHandler): def do_POST(self): if self.path != "/door/open": self.send_response(404) self.end_headers() return token = self.headers.get("X-Token", "") if token != TOKEN: self.send_response(401) self.end_headers() self.wfile.write(b"unauthorized") return # 读 body(可选,不强依赖内容) length = int(self.headers.get("Content-Length", "0")) if length > 0: _ = self.rfile.read(length) ok = play_audio_once() self.send_response(200) self.end_headers() if ok: self.wfile.write(b"playing") #print("收到开门信号,开始播放") else: self.wfile.write(b"ignored_or_error") def log_message(self, format, *args): return def main(): server = ThreadingHTTPServer((HOST, PORT), Handler) print(f"Door receiver listening on {HOST}:{PORT}") server.serve_forever() if __name__ == "__main__": main()接下来两台机器都做个开机自启,也就是各自装个service,daemon-reload一下。拿zero举例,
sudo vim /etc/systemd/system/door_open.service
[Unit] Description=Door Sensor Sender (Pi Zero) After=network-online.target Wants=network-online.target [Service] Type=simple User=pi WorkingDirectory=/home/pi ExecStart=/usr/bin/python3 /home/pi/door_open.py Restart=always RestartSec=2 [Install] WantedBy=multi-user.target