64 lines
1.8 KiB
Python
64 lines
1.8 KiB
Python
#!/usr/bin/env python3
|
|
# /// script
|
|
# requires-python = ">=3.10"
|
|
# dependencies = [
|
|
# "paho-mqtt>=2.0",
|
|
# ]
|
|
# ///
|
|
"""Fake Zigbee device for doorbell E2E validation.
|
|
|
|
Subscribes to a "set" topic and echoes a TOGGLE back to a "state" topic
|
|
(retained), emulating a Zigbee smart plug at the MQTT layer. This lets the
|
|
doorbell E2E probe confirm the full chain (ntfy.sh -> listener -> MQTT -> device)
|
|
without toggling a real lamp.
|
|
|
|
Env vars:
|
|
MQTT_HOST default "mosquitto"
|
|
MQTT_PORT default 1883
|
|
SET_TOPIC default "zigbee2mqtt/DoorbellProbe/set"
|
|
STATE_TOPIC default "zigbee2mqtt/DoorbellProbe"
|
|
"""
|
|
import json
|
|
import os
|
|
|
|
import paho.mqtt.client as mqtt
|
|
|
|
MQTT_HOST = os.environ.get("MQTT_HOST", "mosquitto")
|
|
MQTT_PORT = int(os.environ.get("MQTT_PORT", "1883"))
|
|
SET_TOPIC = os.environ.get("SET_TOPIC", "zigbee2mqtt/DoorbellProbe/set")
|
|
STATE_TOPIC = os.environ.get("STATE_TOPIC", "zigbee2mqtt/DoorbellProbe")
|
|
|
|
|
|
def log(msg: str) -> None:
|
|
print(msg, flush=True)
|
|
|
|
|
|
def on_connect(client, userdata, flags, reason_code, properties=None):
|
|
log(f"connected to {MQTT_HOST}:{MQTT_PORT} (rc={reason_code})")
|
|
client.subscribe(SET_TOPIC)
|
|
|
|
|
|
def on_message(client, userdata, msg):
|
|
log(f"echo -> {STATE_TOPIC} (from {msg.topic})")
|
|
client.publish(STATE_TOPIC, json.dumps({"state": "TOGGLE", "echo": True}), retain=True)
|
|
|
|
|
|
def main() -> None:
|
|
log(f"fake-device: {SET_TOPIC} -> {STATE_TOPIC}")
|
|
client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2)
|
|
client.on_connect = on_connect
|
|
client.on_message = on_message
|
|
while True:
|
|
try:
|
|
client.connect(MQTT_HOST, MQTT_PORT, 60)
|
|
break
|
|
except Exception as e:
|
|
log(f"connect failed: {e}; retry in 5s")
|
|
import time
|
|
time.sleep(5)
|
|
client.loop_forever()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|