import paho.mqtt.client as mqtt from rpi_rf import RFDevice import logging import subprocess import time MQTT_BROKER_HOST = "homeassistant" MQTT_BROKER_PORT = 1883 MQTT_USERNAME = "MQTTBrokerUser" MQTT_PASSWORD = "mqtt1510##" MQTT_TOPIC = "serverhost_torrent_sensor/topic" # Set the desired topic logging.basicConfig(level=logging.DEBUG, datefmt='%Y-%m-%d %H:%M:%S', format='%(asctime)-15s - [%(levelname)s] %(module)s: %(message)s',) # The callback for when the client receives a CONNACK response from the server. def on_connect(client, userdata, flags, reason_code, properties): logging.info(f"Connected with result code {reason_code}") # Subscribing in on_connect() means that if we lose the connection and # reconnect then subscriptions will be renewed. client.subscribe("home/livingroom/raspi/screen") client.subscribe("home/livingroom/raspi/beamer") client.subscribe("home/livingroom/raspi/teufelaudio") msg_info = client.publish("home/livingroom/raspi/available", "online", qos=1, retain=True) # Our application produce some messages # The callback for when a PUBLISH message is received from the server. def on_message(client, userdata, msg): timeout=5 if "screen down" in str(msg.payload): logging.info("Lowering Screen") logging.debug(msg.payload) # Lower Screen via RFdevice code=83716 protocol=3 pulselength=136 length=None rfdevice.tx_code(code, protocol, pulselength, length) time.sleep(0.2) rfdevice.tx_code(code, protocol, pulselength, length) if "screen up" in str(msg.payload): logging.info("Raising Screen") logging.debug(msg.payload) # Lower Screen via RFdevice code=83713 protocol=3 pulselength=136 length=None rfdevice.tx_code(code, protocol, pulselength, length) time.sleep(0.2) rfdevice.tx_code(code, protocol, pulselength, length) if "teufel power" in str(msg.payload): logging.info("Send ON/OFF to Teufel Audio") logging.debug(msg.payload) cmd="irsend send_once teufel_nf power" try: returned_output = subprocess.check_output(cmd, shell=True,timeout=timeout) except Exception as e: print("Command returned %s, %s"%(returned_output,e)) returned_output = subprocess.check_output("systemctl mqttService restart", shell=True,timeout=timeout) logging.info("Ret: " + str(returned_output)) if "teufel opt1" in str(msg.payload) or "teufel aux" in str(msg.payload): time.sleep(10) if "teufel opt1" in str(msg.payload): logging.info("Send Opt1 to Teufel Audio") logging.debug(msg.payload) cmd="irsend send_once teufel_nf opt1" try: returned_output = subprocess.check_output(cmd, shell=True,timeout=timeout) except Exception as e: print("Command returned %s, %s"%(returned_output,e)) returned_output = subprocess.check_output("systemctl mqttService restart", shell=True,timeout=timeout) logging.info("Ret: " + str(returned_output)) if "teufel aux" in str(msg.payload): logging.info("Send aux to Teufel Audio") logging.debug(msg.payload) cmd="irsend send_once teufel_nf aux" try: returned_output = subprocess.check_output(cmd, shell=True,timeout=timeout) except Exception as e: print("Command returned %s, %s"%(returned_output,e)) returned_output = subprocess.check_output("systemctl mqttService restart", shell=True,timeout=timeout) logging.info("Ret: " + str(returned_output)) if "teufel Mute" in str(msg.payload): logging.info("Send Mute to Teufel Audio") logging.debug(msg.payload) cmd="irsend send_once teufel_nf Mute" try: returned_output = subprocess.check_output(cmd, shell=True,timeout=timeout) except Exception as e: print("Command returned %s, %s"%(returned_output,e)) returned_output = subprocess.check_output("systemctl mqttService restart", shell=True,timeout=timeout) logging.info("Ret: " + str(returned_output)) if "teufel VolUp" in str(msg.payload): logging.info("Send VolUp to Teufel Audio") logging.debug(msg.payload) cmd="irsend send_once teufel_nf VolUp" for i in range(5): try: returned_output = subprocess.check_output(cmd, shell=True,timeout=timeout) except Exception as e: print("Command returned %s, %s"%(returned_output,e)) returned_output = subprocess.check_output("systemctl mqttService restart", shell=True,timeout=timeout) logging.info("Ret: " + str(returned_output)) time.sleep(1) if "teufel VolDown" in str(msg.payload): logging.info("Send VolDown to Teufel Audio") logging.debug(msg.payload) cmd="irsend send_once teufel_nf VolDown" for i in range(5): try: returned_output = subprocess.check_output(cmd, shell=True,timeout=timeout) except Exception as e: print("Command returned %s, %s"%(returned_output,e)) returned_output = subprocess.check_output("systemctl mqttService restart", shell=True,timeout=timeout) logging.info("Ret: " + str(returned_output)) time.sleep(1) if "beamer off" in str(msg.payload): logging.info("Send Off to Beamer") logging.debug(msg.payload) cmd="irsend send_once beamer_optoma poweroff" returned_output = subprocess.check_output(cmd, shell=True,timeout=timeout) logging.info("Ret: " + str(returned_output)) time.sleep(1) returned_output = subprocess.check_output(cmd, shell=True,timeout=timeout) logging.info("Ret: " + str(returned_output)) if "beamer on" in str(msg.payload): logging.info("Send On to Beamer") logging.debug(msg.payload) cmd="irsend send_once beamer_optoma poweron" returned_output = subprocess.check_output(cmd, shell=True,timeout=timeout) logging.info("Ret: " + str(returned_output)) def on_subscribe(client, userdata, mid, reason_code_list, properties): # Since we subscribed only for a single channel, reason_code_list contains # a single entry if reason_code_list[0].is_failure: logging.debug(f"Broker rejected you subscription: {reason_code_list[0]}") else: logging.info(f"Broker granted the following QoS: {reason_code_list[0].value}") def on_unsubscribe(client, userdata, mid, reason_code_list, properties): # Be careful, the reason_code_list is only present in MQTTv5. # In MQTTv3 it will always be empty if len(reason_code_list) == 0 or not reason_code_list[0].is_failure: logging.debug("unsubscribe succeeded (if SUBACK is received in MQTTv3 it success)") else: logging.info(f"Broker replied with failure: {reason_code_list[0]}") client.disconnect() ## RX Init print("Starting") gpio_port_board=27 rfdevice = RFDevice(gpio_port_board) rfdevice.enable_tx() rfdevice.tx_repeat = 10 mqttc = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2) mqttc.on_connect = on_connect mqttc.on_message = on_message mqttc.on_subscribe = on_subscribe mqttc.on_unsubscribe = on_unsubscribe unacked_publish = set() # Set username and password for authentication mqttc.username_pw_set(username=MQTT_USERNAME, password=MQTT_PASSWORD) # Connect to the MQTT broker mqttc.connect(MQTT_BROKER_HOST, MQTT_BROKER_PORT, 60) mqttc.loop_start() msg_info2 = mqttc.publish("sensor/test", "6", qos=1) unacked_publish.add(msg_info2.mid) # Wait for all message to be published while len(unacked_publish): time.sleep(0.1) # Due to race-condition described above, the following way to wait for all publish is safer msg_info2.wait_for_publish() mqttc.disconnect() mqttc.loop_stop() rfdevice.cleanup()