1  # SPDX-FileCopyrightText: 2021 Melissa LeBlanc-Williams for Adafruit Industries
  2  #
  3  # SPDX-License-Identifier: MIT
  4  
  5  import time
  6  import json
  7  import ssl
  8  import socket
  9  import adafruit_minimqtt.adafruit_minimqtt as MQTT
 10  from rpc import RpcServer
 11  
 12  mqtt_client = None
 13  mqtt_connected = False
 14  last_mqtt_messages = {}
 15  
 16  # For program flow purposes, we do not want these functions to be called remotely
 17  PROTECTED_FUNCTIONS = ["main", "handle_rpc"]
 18  
 19  def connect(mqtt_client, userdata, flags, rc):
 20      global mqtt_connected
 21      mqtt_connected = True
 22  
 23  def disconnect(mqtt_client, userdata, rc):
 24      global mqtt_connected
 25      mqtt_connected = False
 26  
 27  def message(client, topic, message):
 28      last_mqtt_messages[topic] = message
 29  
 30  class MqttError(Exception):
 31      """For MQTT Specific Errors"""
 32      pass
 33  
 34  # Default to 1883 as SSL on CPython is not currently supported
 35  def mqtt_init(broker, port=1883, username=None, password=None):
 36      global mqtt_client, mqtt_connect_info
 37      mqtt_client = MQTT.MQTT(
 38          broker=broker,
 39          port=port,
 40          username=username,
 41          password=password,
 42          socket_pool=socket,
 43          ssl_context=ssl.create_default_context(),
 44      )
 45  
 46      mqtt_client.on_connect = connect
 47      mqtt_client.on_disconnect = disconnect
 48      mqtt_client.on_message = message
 49  
 50  def mqtt_connect():
 51      mqtt_client.connect()
 52  
 53  def mqtt_publish(topic, payload):
 54      if mqtt_client is None:
 55          raise MqttError("MQTT is not initialized")
 56      try:
 57          return_val = mqtt_client.publish(topic, json.dumps(payload))
 58      except BrokenPipeError:
 59          time.sleep(0.5)
 60          mqtt_client.connect()
 61          return_val = mqtt_client.publish(topic, json.dumps(payload))
 62      return return_val
 63  
 64  def mqtt_subscribe(topic):
 65      if mqtt_client is None:
 66          raise MqttError("MQTT is not initialized")
 67      return mqtt_client.subscribe(topic)
 68  
 69  def mqtt_get_last_value(topic):
 70      """Return the last value we have received regarding a topic"""
 71      if topic in last_mqtt_messages.keys():
 72          return last_mqtt_messages[topic]
 73      return None
 74  
 75  def is_running():
 76      return True
 77  
 78  def handle_rpc(packet):
 79      """This function will verify good data in packet,
 80      call the method with parameters, and generate a response
 81      packet as the return value"""
 82      print("Received packet")
 83      func_name = packet['function']
 84      if func_name in PROTECTED_FUNCTIONS:
 85          return rpc.create_response_packet(error=True, message=f"{func_name}'() is a protected function and can not be called.")
 86      if func_name not in globals():
 87          return rpc.create_response_packet(error=True, message=f"Function {func_name}() not found")
 88      try:
 89          return_val = globals()[func_name](*packet['args'], **packet['kwargs'])
 90      except MqttError as err:
 91          return rpc.create_response_packet(error=True, error_type="MQTT", message=str(err))
 92  
 93      packet = rpc.create_response_packet(return_val=return_val)
 94      return packet
 95  
 96  def main():
 97      """Command line, entry point"""
 98      global mqtt_connected
 99      while True:
100          rpc.loop(0.25)
101          if mqtt_connected and mqtt_client is not None:
102              try:
103                  mqtt_client.loop(0.5)
104              except AttributeError:
105                  mqtt_connected = False
106  
107  if __name__ == '__main__':
108      rpc = RpcServer(handle_rpc)
109      try:
110          print(f"Listening for RPC Calls, to stop press \"CTRL+C\"")
111          main()
112      except KeyboardInterrupt:
113          print("")
114          print(f"Caught interrupt, exiting...")
115      rpc.close_serial()