In preparation for next year, my team is working on some autonomous code for the AVR drone. However, we’re getting stuck at the basic parts.
I wrote a simple variation on the provided sandbox.py script, to flash the led strip. However, the mqtt message is not being sent and nothing changes. The debug statements before and after the message are being sent. I’m not sure quite what’s wrong.
While the __init__ method of the MQTTModule class (what you’re calling with super().__init__()) connects to the MQTT broker, it does not start what is called the “event loop”. The event loop is an infinite loop that processes MQTT messages. This includes receiving incoming messages, and running callback functions defined in the topic_map, but at the same time, it also deals with sending messages as well. Without this event loop, nothing happens besides connecting to the broker.
In order to start the event loop, there are two options.
The box.run() method. This starts the infinite loop, and prevents any more code from ever running. This is ideal for when you want your sandbox code to always be responding to incoming messages, like the example I made.
The box.run_non_blocking() method. This starts the event loop in the background, and allows more code to run. This is great for when you have code that needs to run independently of incoming messages, like your code. This is how the software that gets data from the ZED Mini tracking camera works. When you do this however, you need to have some kind of infinite loop running, so your code doesn’t reach the end and stop executing.
Putting all of that together, this is how I would rewrite your example to work:
import time
from bell.avr.mqtt.client import MQTTModule
from bell.avr.mqtt.payloads import AvrPcmSetBaseColorPayload
from loguru import logger
class Sandbox(MQTTModule):
def __init__(self) -> None:
super().__init__()
logger.debug("Class initialized")
def hello_world(self) -> None:
logger.debug("Hello world")
payload = AvrPcmSetBaseColorPayload(wrgb=(0, 100, 0, 0))
self.send_message("avr/pcm/set_base_color", payload)
logger.debug("Light changed")
if __name__ == "__main__":
box = Sandbox()
box.run_non_blocking()
box.hello_world()
while True:
time.sleep(0.1)
To answer your clarifying question, the open_servo function cannot be called from the unedited sandbox.py. It needs to be called in the callback to some incoming message (like, if I see this AprilTag, open this servo). I was trying not to give too much away and leave it as an exercise to the reader.