Update and Lint MacroPad RPC Home Assistant code to work with settings.toml

Author: makermelissa

MacroPad_RPC_Home_Assistant/boot.py

@@ -4,4 +4,4 @@
 
 import usb_cdc
 
-usb_cdc.enable(console=True, data=True)
+usb_cdc.enable(console=True, data=True)

MacroPad_RPC_Home_Assistant/code.py

@@ -4,14 +4,14 @@
 """
 Home Assistant Remote Procedure Call for MacroPad.
 """
+import os
 import time
 import displayio
 import terminalio
 from adafruit_display_shapes.rect import Rect
-from rpc import RpcClient, RpcError
 from adafruit_display_text import label
 from adafruit_macropad import MacroPad
-from secrets import secrets
+from rpc import RpcClient, RpcError
 
 macropad = MacroPad()
 rpc = RpcClient()
@@ -22,27 +22,40 @@
 KEY_LABELS = ("Demo", "Office")
 UPDATE_DELAY = 0.25
 NEOPIXEL_COLORS = {
-    "OFF": 0xff0000,
-    "ON": 0x00ff00,
+    "OFF": 0xFF0000,
+    "ON": 0x00FF00,
 }
 
 class MqttError(Exception):
     """For MQTT Specific Errors"""
-    pass
+
 # Set up displayio group with all the labels
 group = displayio.Group()
 for key_index in range(12):
     x = key_index % 3
     y = key_index // 3
-    group.append(label.Label(terminalio.FONT, text=(str(KEY_LABELS[key_index]) if key_index < len(KEY_LABELS) else ''), color=0xFFFFFF,
-                             anchored_position=((macropad.display.width - 1) * x / 2,
-                                                macropad.display.height - 1 -
-                                                (3 - y) * 12),
-                             anchor_point=(x / 2, 1.0)))
+    group.append(
+        label.Label(
+            terminalio.FONT,
+            text=(str(KEY_LABELS[key_index]) if key_index < len(KEY_LABELS) else ""),
+            color=0xFFFFFF,
+            anchored_position=(
+                (macropad.display.width - 1) * x / 2,
+                macropad.display.height - 1 - (3 - y) * 12,
+            ),
+            anchor_point=(x / 2, 1.0),
+        )
+    )
 group.append(Rect(0, 0, macropad.display.width, 12, fill=0xFFFFFF))
-group.append(label.Label(terminalio.FONT, text='Home Assistant', color=0x000000,
-                         anchored_position=(macropad.display.width//2, -2),
-                         anchor_point=(0.5, 0.0)))
+group.append(
+    label.Label(
+        terminalio.FONT,
+        text="Home Assistant",
+        color=0x000000,
+        anchored_position=(macropad.display.width // 2, -2),
+        anchor_point=(0.5, 0.0),
+    )
+)
 macropad.display.show(group)
 
 def rpc_call(function, *args, **kwargs):
@@ -54,16 +67,22 @@ def rpc_call(function, *args, **kwargs):
     return response["return_val"]
 
 def mqtt_init():
-    rpc_call("mqtt_init", secrets["mqtt_broker"], username=secrets["mqtt_username"], password=secrets["mqtt_password"], port=secrets["mqtt_port"])
+    rpc_call(
+        "mqtt_init",
+        os.getenv("MQTT_BROKER"),
+        username=os.getenv("MQTT_USERNAME"),
+        password=os.getenv("MQTT_PASSWORD"),
+        port=os.getenv("MQTT_PORT"),
+    )
     rpc_call("mqtt_connect")
 
-def update_key(key_number):
-    if key_number < len(SUBSCRIBE_TOPICS):
-        switch_state = rpc_call("mqtt_get_last_value", SUBSCRIBE_TOPICS[key_number])
+def update_key(key_id):
+    if key_id < len(SUBSCRIBE_TOPICS):
+        switch_state = rpc_call("mqtt_get_last_value", SUBSCRIBE_TOPICS[key_id])
         if switch_state is not None:
-            macropad.pixels[key_number] = NEOPIXEL_COLORS[switch_state]
+            macropad.pixels[key_id] = NEOPIXEL_COLORS[switch_state]
         else:
-            macropad.pixels[key_number] = 0
+            macropad.pixels[key_id] = 0
 
 server_is_running = False
 print("Waiting for server...")
@@ -93,7 +112,11 @@ def update_key(key_number):
         last_macropad_encoder_value = macropad.encoder
 
     macropad.encoder_switch_debounced.update()
-    if macropad.encoder_switch_debounced.pressed and "key_number" not in output and ENCODER_ITEM is not None:
+    if (
+        macropad.encoder_switch_debounced.pressed
+        and "key_number" not in output
+        and ENCODER_ITEM is not None
+    ):
         output["key_number"] = ENCODER_ITEM
 
     if output:

MacroPad_RPC_Home_Assistant/rpc.py

@@ -7,41 +7,42 @@
 
 import time
 import json
+
 try:
     import serial
     import adafruit_board_toolkit.circuitpython_serial
+
     json_decode_exception = json.decoder.JSONDecodeError
 except ImportError:
     import usb_cdc as serial
+
     json_decode_exception = ValueError
 
-RESPONSE_TIMEOUT=5
-DATA_TIMEOUT=0.5
+RESPONSE_TIMEOUT = 5
+DATA_TIMEOUT = 0.5
+
 
 class RpcError(Exception):
     """For RPC Specific Errors"""
-    pass
 
 class _Rpc:
     def __init__(self):
         self._serial = None
 
     @staticmethod
-    def create_response_packet(error=False, error_type="RPC", message=None, return_val=None):
+    def create_response_packet(
+        error=False, error_type="RPC", message=None, return_val=None
+    ):
         return {
             "error": error,
             "error_type": error_type if error else None,
             "message": message,
-            "return_val": return_val
+            "return_val": return_val,
         }
 
     @staticmethod
-    def create_request_packet(function, args=[], kwargs={}):
-        return {
-            "function": function,
-            "args": args,
-            "kwargs": kwargs
-        }
+    def create_request_packet(function, args=[], kwargs={}): # pylint: disable=dangerous-default-value
+        return {"function": function, "args": args, "kwargs": kwargs}
 
     def _wait_for_packet(self, timeout=None):
         incoming_packet = b""
@@ -51,11 +52,16 @@ def _wait_for_packet(self, timeout=None):
             if incoming_packet:
                 data_start_time = time.monotonic()
             while not self._serial.in_waiting:
-                if incoming_packet and (time.monotonic() - data_start_time) >= DATA_TIMEOUT:
+                if (
+                    incoming_packet
+                    and (time.monotonic() - data_start_time) >= DATA_TIMEOUT
+                ):
                     incoming_packet = b""
                 if not incoming_packet and timeout is not None:
                     if (time.monotonic() - response_start_time) >= timeout:
-                        return self.create_response_packet(error=True, message="Timed out waiting for response")
+                        return self.create_response_packet(
+                            error=True, message="Timed out waiting for response"
+                        )
                 time.sleep(0.001)
             data = self._serial.read(self._serial.in_waiting)
             if data:
@@ -66,13 +72,13 @@ def _wait_for_packet(self, timeout=None):
                     if sorted(tuple(packet.keys())) == sorted(self._packet_format()):
                         return packet
                 except json_decode_exception:
-                    pass # Incomplete packet
+                    pass  # Incomplete packet
 
 class RpcClient(_Rpc):
     def __init__(self):
         super().__init__()
         self._serial = serial.data
-    
+
     def _packet_format(self):
         return self.create_response_packet().keys()
 
@@ -97,13 +103,14 @@ def init_serial(self, baudrate):
         return serial.Serial(
             port,
             baudrate,
-            parity='N',
+            parity="N",
             rtscts=False,
             xonxoff=False,
             exclusive=True,
         )
 
-    def detect_port(self):
+    @staticmethod
+    def detect_port():
         """
         Detect the port automatically
         """
@@ -113,14 +120,16 @@ def detect_port(self):
             if len(ports) > 1:
                 print("Multiple devices detected, using the first detected port.")
             return ports[0]
-        raise RuntimeError("Unable to find any CircuitPython Devices with the CDC Data port enabled.")
+        raise RuntimeError(
+            "Unable to find any CircuitPython Devices with the CDC Data port enabled."
+        )
 
     def loop(self, timeout=None):
         packet = self._wait_for_packet(timeout)
         if "error" not in packet:
             response_packet = self._handler(packet)
             self._serial.write(bytes(json.dumps(response_packet), "utf-8"))
-    
+
     def close_serial(self):
         if self._serial is not None:
-            self._serial.close()
+            self._serial.close()

MacroPad_RPC_Home_Assistant/rpc_ha_server.py

@@ -9,32 +9,30 @@
 import adafruit_minimqtt.adafruit_minimqtt as MQTT
 from rpc import RpcServer
 
-mqtt_client = None
-mqtt_connected = False
+mqtt_status = {
+    "connected": False,
+    "client": None,
+}
 last_mqtt_messages = {}
 
 # For program flow purposes, we do not want these functions to be called remotely
 PROTECTED_FUNCTIONS = ["main", "handle_rpc"]
 
-def connect(mqtt_client, userdata, flags, rc):
-    global mqtt_connected
-    mqtt_connected = True
+def connect(_mqtt_client, _userdata, _flags, _rc):
+    mqtt_status["connected"] = True
 
-def disconnect(mqtt_client, userdata, rc):
-    global mqtt_connected
-    mqtt_connected = False
+def disconnect(_mqtt_client, _userdata, _rc):
+    mqtt_status["connected"] = False
 
-def message(client, topic, message):
-    last_mqtt_messages[topic] = message
+def message(_client, topic, payload):
+    last_mqtt_messages[topic] = payload
 
 class MqttError(Exception):
     """For MQTT Specific Errors"""
-    pass
 
-# Default to 1883 as SSL on CPython is not currently supported
+# Default to 1883 since SSL on CPython is not currently supported
 def mqtt_init(broker, port=1883, username=None, password=None):
-    global mqtt_client, mqtt_connect_info
-    mqtt_client = MQTT.MQTT(
+    mqtt_status["client"] = MQTT.MQTT(
         broker=broker,
         port=port,
         username=username,
@@ -43,28 +41,28 @@ def mqtt_init(broker, port=1883, username=None, password=None):
         ssl_context=ssl.create_default_context(),
     )
 
-    mqtt_client.on_connect = connect
-    mqtt_client.on_disconnect = disconnect
-    mqtt_client.on_message = message
+    mqtt_status["client"].on_connect = connect
+    mqtt_status["client"].on_disconnect = disconnect
+    mqtt_status["client"].on_message = message
 
 def mqtt_connect():
-    mqtt_client.connect()
+    mqtt_status["client"].connect()
 
 def mqtt_publish(topic, payload):
-    if mqtt_client is None:
+    if mqtt_status["client"] is None:
         raise MqttError("MQTT is not initialized")
     try:
-        return_val = mqtt_client.publish(topic, json.dumps(payload))
+        return_val = mqtt_status["client"].publish(topic, json.dumps(payload))
     except BrokenPipeError:
         time.sleep(0.5)
-        mqtt_client.connect()
-        return_val = mqtt_client.publish(topic, json.dumps(payload))
+        mqtt_status["client"].connect()
+        return_val = mqtt_status["client"].publish(topic, json.dumps(payload))
     return return_val
 
 def mqtt_subscribe(topic):
-    if mqtt_client is None:
+    if mqtt_status["client"] is None:
         raise MqttError("MQTT is not initialized")
-    return mqtt_client.subscribe(topic)
+    return mqtt_status["client"].subscribe(topic)
 
 def mqtt_get_last_value(topic):
     """Return the last value we have received regarding a topic"""
@@ -80,36 +78,42 @@ def handle_rpc(packet):
     call the method with parameters, and generate a response
     packet as the return value"""
     print("Received packet")
-    func_name = packet['function']
+    func_name = packet["function"]
     if func_name in PROTECTED_FUNCTIONS:
-        return rpc.create_response_packet(error=True, message=f"{func_name}'() is a protected function and can not be called.")
+        return rpc.create_response_packet(
+            error=True,
+            message=f"{func_name}'() is a protected function and can not be called.",
+        )
     if func_name not in globals():
-        return rpc.create_response_packet(error=True, message=f"Function {func_name}() not found")
+        return rpc.create_response_packet(
+            error=True, message=f"Function {func_name}() not found"
+        )
     try:
-        return_val = globals()[func_name](*packet['args'], **packet['kwargs'])
+        return_val = globals()[func_name](*packet["args"], **packet["kwargs"])
     except MqttError as err:
-        return rpc.create_response_packet(error=True, error_type="MQTT", message=str(err))
+        return rpc.create_response_packet(
+            error=True, error_type="MQTT", message=str(err)
+        )
 
     packet = rpc.create_response_packet(return_val=return_val)
     return packet
 
 def main():
     """Command line, entry point"""
-    global mqtt_connected
     while True:
         rpc.loop(0.25)
-        if mqtt_connected and mqtt_client is not None:
+        if mqtt_status["connected"] and mqtt_status["client"] is not None:
             try:
-                mqtt_client.loop(0.5)
+                mqtt_status["client"].loop(0.5)
             except AttributeError:
-                mqtt_connected = False
+                mqtt_status["connected"] = False
 
-if __name__ == '__main__':
+if __name__ == "__main__":
     rpc = RpcServer(handle_rpc)
     try:
-        print(f"Listening for RPC Calls, to stop press \"CTRL+C\"")
+        print('Listening for RPC Calls, to stop press "CTRL+C"')
         main()
     except KeyboardInterrupt:
         print("")
-        print(f"Caught interrupt, exiting...")
-    rpc.close_serial()
+        print("Caught interrupt, exiting...")
+    rpc.close_serial()