Once you have CircuitPython installed, the first thing you will need to do is to enable the CDC Data device. CDC stands for "Communications Device Class" and is a USB term. When a CircuitPython device is booted up, by default it normally comes with a CDC Console Device enabled, which is awesome for debugging, but it can introduce special characters into the stream. The Data device is an additional serial device that can be enabled that overcomes this issue. You can read more about it in our Customizing USB Devices in CircuitPython guide.
To enable the CDC Data device, you just need to add the following into a boot.py file on the root level of the CIRCUITPY drive:
import usb_cdc usb_cdc.enable(console=True, data=True)
The settings.toml File
Like other devices, which are network enabled such as the Adafruit PyPortal or MagTag, this project uses a settings.toml file. This will contain the MQTT connection information that will be used to connect to your Home Assistant MQTT server. If you have done any of the other Adafruit HomeAssistant projects, you should just be able to copy an existing settings.toml file. If you haven't you can just create a settings.toml file at the root level of your CIRCUITPY drive with the following content:
CIRCUITPY_WIFI_SSID = "your-wifi-ssid" CIRCUITPY_WIFI_PASSWORD = "your-wifi-password" MQTT_BROKER = "192.168.1.1" MQTT_PORT = 1883 MQTT_USERNAME = "myusername" MQTT_PASSWORD = "mypassword"
The only items you really need to change the values on are the MQTT parameters.
Download the Project Bundle
Your project will use a specific set of CircuitPython libraries as well as the code.py and rpc.py files. In order to get the libraries you need, click on the Download Project Bundle link below, and decompress the .zip file.
Next, drag the contents of the CircuitPython 7.x folder in the uncompressed bundle directory onto you microcontroller board's CIRCUITPY drive, replacing any existing files or directories with the same names, and adding any new ones that are necessary.
The files on your MacroPad should look like this:
# SPDX-FileCopyrightText: Copyright (c) 2021 Melissa LeBlanc-Williams for Adafruit Industries
#
# SPDX-License-Identifier: Unlicense
"""
Home Assistant Remote Procedure Call for MacroPad.
"""
import os
import time
import displayio
import terminalio
from adafruit_display_shapes.rect import Rect
from adafruit_display_text import label
from adafruit_macropad import MacroPad
from rpc import RpcClient, RpcError, MqttError
macropad = MacroPad()
rpc = RpcClient()
COMMAND_TOPIC = "macropad/peripheral"
SUBSCRIBE_TOPICS = ("stat/demoswitch/POWER", "stat/office-light/POWER")
ENCODER_ITEM = 0
KEY_LABELS = ("Demo", "Office")
UPDATE_DELAY = 0.25
NEOPIXEL_COLORS = {
"OFF": 0xFF0000,
"ON": 0x00FF00,
}
# 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(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),
)
)
macropad.display.root_group = group
def rpc_call(function, *args, **kwargs):
response = rpc.call(function, *args, **kwargs)
if response["error"]:
if response["error_type"] == "mqtt":
raise MqttError(response["message"])
raise RpcError(response["message"])
return response["return_val"]
def mqtt_init():
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_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_id] = NEOPIXEL_COLORS[switch_state]
else:
macropad.pixels[key_id] = 0
server_is_running = False
print("Waiting for server...")
while not server_is_running:
try:
server_is_running = rpc_call("is_running")
print("Connected")
except RpcError:
pass
mqtt_init()
last_macropad_encoder_value = macropad.encoder
for key_number, topic in enumerate(SUBSCRIBE_TOPICS):
rpc_call("mqtt_subscribe", topic)
update_key(key_number)
while True:
output = {}
key_event = macropad.keys.events.get()
if key_event and key_event.pressed:
output["key_number"] = key_event.key_number
if macropad.encoder != last_macropad_encoder_value:
output["encoder"] = macropad.encoder - last_macropad_encoder_value
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
):
output["key_number"] = ENCODER_ITEM
if output:
try:
rpc_call("mqtt_publish", COMMAND_TOPIC, output)
if "key_number" in output:
time.sleep(UPDATE_DELAY)
update_key(output["key_number"])
elif ENCODER_ITEM is not None:
update_key(ENCODER_ITEM)
except MqttError:
mqtt_init()
except RpcError as err_msg:
print(err_msg)
Page last edited January 21, 2025
Text editor powered by tinymce.
