猪咪闲来无事创作无中生有阵。一边是一台可以播放声音的主机一边是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( filenameLOG_FILE, levellogging.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_upTrue, bounce_time0.05) def send_open_signal(): req urllib.request.Request( SERVER_URL, datajson.dumps({event: door_open, ts: time.time()}).encode(utf-8), headers{ Content-Type: application/json, X-Token: TOKEN }, methodPOST, ) try: with urllib.request.urlopen(req, timeout2): 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(bunauthorized) 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(bplaying) #print(收到开门信号开始播放) else: self.wfile.write(bignored_or_error) def log_message(self, format, *args): return def main(): server ThreadingHTTPServer((HOST, PORT), Handler) print(fDoor 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] DescriptionDoor Sensor Sender (Pi Zero) Afternetwork-online.target Wantsnetwork-online.target [Service] Typesimple Userpi WorkingDirectory/home/pi ExecStart/usr/bin/python3 /home/pi/door_open.py Restartalways RestartSec2 [Install] WantedBymulti-user.target