Because I spent so much time on this, doesn't mean we all have to. This python script configures a "doorbell" device in Home Assistant via MQTT.
This device has:
- A switch / toggle for the mute state. This also updates its own state;
- A button to ring your own doorbell "cuzz ya can";
- A trigger event as a "short button press" that you can use in automations;
#!/usr/bin/env python # (c) 2022-04-24 S.E.Jansen. # Copyright (c) 2014 Adafruit Industries # Author: Tony DiCola #versie = "1.0" import time import paho.mqtt.client as paho import subprocess import MQTT_Config import GPIO_Config import json #call back function def on_connect(client, userdata, flags, rc): if rc==0: client.connected_flag=True #set flag print("connected OK") else: print("Bad connection Returned code=",rc) client.bad_connection_flag=True #Make connection with MQTT broker and paho.Client.connected_flag=False #Create flag in class paho.Client.bad_connection_flag=False #another flag client= paho.Client() #create client object client.username_pw_set(MQTT_Config.user,MQTT_Config.password) client.on_connect=on_connect #bind call back function print("Connecting to broker ",MQTT_Config.broker) client.connect(MQTT_Config.broker,MQTT_Config.port) #establish connection client.loop_start() #start network loop while not client.connected_flag and not client.bad_connection_flag: #wait in loop print("In wait loop") time.sleep(1) if client.bad_connection_flag: client.loop_stop() #Stop loop sys.exit() #the main loop starts here print("in Main Loop") print("Adding Doorbell " + GPIO_Config.ID + " action trigger") #Add doorbell button; as an action trigger topic = MQTT_Config.HA_name + "/device_automation/Doorbell_Button_" + GPIO_Config.ID + "/action_pressed" config_payload = {"automation_type":"trigger", "payload": "visitor", "value_template": "{{value_json.value}}", "unique_id": "Doorbell_Button_" + GPIO_Config.ID, "topic":topic + "/action", "type":"action", "subtype":"button_short_press", "icon": "mdi:doorbell", "device": {"identifiers":[("Doorbell_" + GPIO_Config.ID)], "name":("Doorbell_" + GPIO_Config.ID), "manufacturer": "X", "model": "001", "suggested_area": GPIO_Config.location}} client.publish(topic + "/config",json.dumps(config_payload),0,True) time.sleep(1) print("Adding Doorbell_" + GPIO_Config.ID + " button") #Add doorbell; as output topic = MQTT_Config.HA_name + "/button/Doorbell_" + GPIO_Config.ID config_payload = {"name": GPIO_Config.location + " doorbell", "unique_id": "Doorbell_" + GPIO_Config.ID + "_ring", "name": "Doorbell_" + GPIO_Config.ID + "_ring", "command_topic": topic + "/set", "payload_press": "ring_short", "icon": "mdi:alarm-bell", "device": {"identifiers":[("Doorbell_" + GPIO_Config.ID)], "name":("Doorbell_" + GPIO_Config.ID), "manufacturer": "X", "model": "001", "suggested_area": GPIO_Config.location}} client.publish(topic + "/config",json.dumps(config_payload),0,True) time.sleep(1) print("Adding Doorbell_" + GPIO_Config.ID + " mute switch") #Add doorbell; As switch. With state and command topic topic = MQTT_Config.HA_name + "/switch/Doorbell_" + GPIO_Config.ID config_payload = { "unique_id": GPIO_Config.location + " mute switch " + GPIO_Config.ID, "name": GPIO_Config.location + " mute switch", "state_topic": topic + "/state", "command_topic": topic + "/set", #"payload_on": "{\"mute_set\":\"ON\"}", #"playload_off": "{\"mute_set\":\"OFF\"}", "state_on": "1", "state_off": "0", "value_template": "{{value_json.mute_state}}", "retain": True, "icon": "mdi:alarm-bell", "device": {"identifiers":[("Doorbell_" + GPIO_Config.ID)], "name":("Doorbell_" + GPIO_Config.ID), "manufacturer": "X", "model": "001", "suggested_area": GPIO_Config.location} } client.publish(topic + "/config",json.dumps(config_payload),0,True) time.sleep(1) client.loop_stop() #stop connection loop client.disconnect() #gracefully disconnect client.connected_flag=False #reset flag
I updated the python daemon accordingly. It has to listen to two topics and the action/trigger topic is also different.
Hope this is helpfull for someone. I spent almost a full day on the changes and it is still not perfect. I can't get the switch or the button to use the JSON payload. I really don't know where else I can put escapes or more single or double quotes. In the end I decided to skip the JSON for commands.
Also when the daemon encounters a non-JSON command on a subscribed topic, it crashes.. The code should use some proper exception handling, but I don't wanna.
#!/usr/bin/env python
# (c) 2022-04-24 S.E.Jansen.
#import lybraries
#import RPi.GPIO as GPIO
from gpiozero import Button as Button
from gpiozero import DigitalInputDevice as DigitalInputDevice
from gpiozero import DigitalOutputDevice as DigitalOutputDevice
import time
import GPIO_Config
import MQTT_Config
import json
import paho.mqtt.client as paho
#setup pins
button = Button(GPIO_Config.button, pull_up=False, bounce_time=0.05,hold_time = 3, hold_repeat=False)
bel = DigitalOutputDevice(GPIO_Config.bel)
mute_state = DigitalInputDevice(GPIO_Config.mute_state, pull_up=True, bounce_time=0.05)
mute_set = DigitalOutputDevice(GPIO_Config.mute_set)
mute_reset = DigitalOutputDevice(GPIO_Config.mute_reset)
#GPIO functions and callbacks
def visitor():
#on rise of input (debounce 50ms) the doorbell has rung
print("doorbel press detected")
data = {'value':"visitor"}
client.publish(MQTT_Config.HA_name + "/device_automation/Doorbell_Button_" + GPIO_Config.ID + "/action_pressed/action",json.dumps(data),0,False)
#MQTT connection + callback functions
def on_message(client, userdata, message):
#check and handle incomming messages to:
# - ring bell
# - check mute status
# - set bell to mute / unmute
data = message.payload
receive=data.decode("utf-8")
#print(m_decode)
#print (m_decode['Server_name'])
if message.topic == MQTT_Config.HA_name + "/button/Doorbell_" + GPIO_Config.ID + "/set":
#m_decode = json.loads(receive)
#buttonpress = m_decode["doorbell"]
buttonpress = receive
print ("Button message received: " + buttonpress)
if buttonpress == "ring_short":
bel.blink(on_time=0.1, n=1)
if message.topic == MQTT_Config.HA_name + "/switch/Doorbell_" + GPIO_Config.ID + "/set":
#muteCommand = m_decode["mute_set"]
muteCommand = receive
print ("Command received for mute: " + muteCommand)
#switch: for
if muteCommand == "ON":
print("setting mute")
mute_set.blink(on_time=0.5, n=1)
time.sleep(0.1)
data = {'mute_state':mute_state.value}
client.publish(MQTT_Config.HA_name + "/switch/Doorbell_" + GPIO_Config.ID + "/state",json.dumps(data),0,True)
elif muteCommand == "OFF":
print("resetting mute")
mute_reset.blink(on_time=0.5, n=1)
time.sleep(0.1)
data = {'mute_state':mute_state.value}
client.publish(MQTT_Config.HA_name + "/switch/Doorbell_" + GPIO_Config.ID + "/state",json.dumps(data),0,True)
elif muteCommand == "ENQ":
#check status of relays. B-side is inverse of "mute" and is connected to GND.
#Pi is set with internal pull-up. Normal operation is mute off (bell rings normaly)
#A-side is closed, B-side is open.Therefore internall pul-up gives a value of 1.
#value of mute is therefore inverse of pin-value.
#mute = True means silent operation. mute = False means ring!
data = {'mute_state':mute_state.value}
client.publish(MQTT_Config.HA_name + "/switch/Doorbell_" + GPIO_Config.ID + "/state",json.dumps(data),0,True)
#def MQTT_AutodiscoverConfig():
#send config to MQTT broker for autodiscover in Home Assistant
# print("send Autoconfig message with MQTT")
#init
#Set callbacks:
#Doorbell button
button.when_pressed = visitor
#button.when_long_press?
#Mute change detect
#mute_state.when_changed?
#MQTT client
#client_on_message = MQTT_message
#client_on_connect .. etc
#call back function for MQTT connection
def on_connect(client, userdata, flags, rc):
if rc==0:
client.connected_flag=True #set flag
print("connected OK")
# MQTT subscribe to command topic to receive commands (in json format)
client.subscribe(MQTT_Config.HA_name + "/button/Doorbell_" + GPIO_Config.ID + "/set",0)
print("Subscribed to: " + MQTT_Config.HA_name + "/button/Doorbell_" + GPIO_Config.ID + "/set")
client.subscribe(MQTT_Config.HA_name + "/switch/Doorbell_" + GPIO_Config.ID + "/set",0)
print("Subscribed to: " + MQTT_Config.HA_name + "/switch/Doorbell_" + GPIO_Config.ID + "/set")
data = {'mute_state':mute_state.value}
client.publish(MQTT_Config.HA_name + "/switch/Doorbell_" + GPIO_Config.ID + "/state",json.dumps(data),0,True)
print(json.dumps(data))
else:
print("Bad connection Returned code=",rc)
client.bad_connection_flag=True
#call back function for disconnect MQTT connection to reconnect
#def on_disconnect
# MQTT details
paho.Client.connected_flag=False #Create flag in class
paho.Client.bad_connection_flag=False #another flag
client= paho.Client() #create client object
client.username_pw_set(MQTT_Config.user,MQTT_Config.password)
client.on_connect=on_connect #bind call back function
client.on_message= on_message #attach function to callback
client.connect(MQTT_Config.broker,MQTT_Config.port) #establish connection
client.loop_start() #start network loop
while not client.connected_flag and not client.bad_connection_flag: #wait in loop
time.sleep(0.1)
if client.bad_connection_flag:
client.loop_stop() #Stop loop
sys.exit()
# MQTT Home Assistant autodiscover config
# MQTT subscribe to command topic to receive commands (in json format)
client.subscribe(MQTT_Config.HA_name + "/button/Doorbell_" + GPIO_Config.ID + "/set",0)
print("Subscribed to: " + MQTT_Config.HA_name + "/button/Doorbell_" + GPIO_Config.ID + "/set")
client.subscribe(MQTT_Config.HA_name + "/switch/Doorbell_" + GPIO_Config.ID + "/set",0)
print("Subscribed to: " + MQTT_Config.HA_name + "/switch/Doorbell_" + GPIO_Config.ID + "/set")
data = {'mute_state':mute_state.value}
client.publish(MQTT_Config.HA_name + "/switch/Doorbell_" + GPIO_Config.ID + "/state",json.dumps(data),0,True)
print(json.dumps(data))
#for silent testing..
#print("engage silent mode")
#mute_set.blink(on_time=0.5, n=1)
#time.sleep(0.1)
#print("mute status: ")
#print(mute_state.value)
while True:
try:
# print("still running")
time.sleep(10)
except KeyboardInterrupt:
client.loop_stop() #stop connection loop
client.disconnect() #gracefully disconnect
client.connected_flag=False #reset flag
mute_reset.blink(on_time=0.5, n=1)
time.sleep(0.1)
print("mute status: ")
print(mute_state.value)
print("returned doorbell to normal opperation")
print("Ending program")
exit()
And supporting config files:
#MQTT_config
broker="xxx.xxx.xxx.xxx"
port=1883
user="xx"
password="xx"
HA_name="homeassistant"
#Pins (GPIO numbering)
button=18
bel=23
mute_state=27
mute_set=24
mute_reset=25
#Unique sensor ID
ID="XX"
location="location"
Systemd services file:
[Unit]
Description=Managing doorbell with MQTT
After=network.target
[Service]
ExecStart=/home/user/python/GPIO_MQTT_daemon.py
WorkingDirectory=/home/user/python
StandardOutput=inherit
StandardError=inherit
Restart=on-failure
User=user
[Install]
WantedBy=multi-user.target
As always: use at your own risk :)
Discussions
Become a Hackaday.io Member
Create an account to leave a comment. Already have an account? Log In.