A few weeks ago, we announced our new Typed Python programming library, which offers an asynchronous API in addition to the "traditional" one. This week, we'll take a look at how to use these asynchronous functions by updating the Home Assistant integration we described in this post.
The Yocto-ColorLedCluster integration lets you drive the LEDs of the Yocto-Color-V2, Yocto-MaxiBuzzer, and Yocto-MaxiKnob modules from Home Assistant. The aim of this integration was mainly to have a reference implementation for customers who want to use advanced functionalities of our modules in Home Assistant.
Home Assistant has been using asynchronous programming for some time now, and when we wrote this integration, we had to write wrappers that used async_add_executor_job to call our traditional Python library.
For example, here's the part of the code that turned off the LEDs.
from yoctopuce.yocto_colorledcluster import YColorLedCluster
def set_color(self, hwid: str, rgb_color) -> None:
yled = YColorLedCluster.FindColorLedCluster(hwid)
if yled.isOnline():
nbled = yled.get_activeLedCount()
_LOGGER.debug("Set %s to color 0x%x" % (hwid, rgb_color))
yled.rgb_move(0, nbled, rgb_color, 1000)
else:
_LOGGER.warning("Module %s not offline" % hwid)
def set_on_off(self) -> None:
if self._is_on:
color = (
(self._rgb_color[0] << 16)
+ (self._rgb_color[1] << 8)
+ self._rgb_color[2]
)
self.set_color(self._name, color)
self._is_on = True
else:
self.set_color(self._name, 0)
self._is_on = False
async def async_turn_off(self, **kwargs: Any) -> None:
self._is_on = False
await self.hass.async_add_executor_job(self.set_on_off)
Modifications
We'll see how easy it is to modify this extension to create a fully asynchronous version.
The first step is to modify the extension so that it references our new library. To do this, simply update the "requirements" field in the manifest to use the PyPi package of the Typed Python library. Specifically, we need to replace "yoctopuce==2.X.XXX" with "yoctolib==2.X.XXX".
Here's the updated manifest.json file.
"domain":"yocto_colorledcluster",
"name":"Yocto-ColorLedCluster",
"codeowners": [
"@yoctopuce"
],
"config_flow": true,
"dependencies": [],
"issue_tracker": "https://github.com/yoctopuce-examples/yocto_colorledcluster/issues",
"documentation": "https://github.com/yoctopuce-examples/yocto_colorledcluster/",
"homekit": {},
"iot_class": "local_polling",
"requirements": ["yoctolib==2.1.6832"],
"version":"1.1.0",
"ssdp": [],
"zeroconf": []
}
The second step is to modify the extension source files so that they import the correct version of the library. As explained in the post announcing the Typed Pyhton library, the Yoctopuce API is available in both synchronous and asynchronous versions. To use the asynchronous version, add the "_aio" suffix to imports.
So, to use the asynchronous version of the YAPI classes and the YColorLedCluster class, use the following code:
from yoctolib.yocto_colorledcluster_aio import YColorLedCluster
Next, modify the extension code to convert the synchronous code into asynchronous code.
We need to add awaits to all calls to asynchronous functions in our library, such as YAPI.RegisterHub(), isOnline(), rgb_move(), and so on.
Then, we need to modify the declaration of all functions that directly or indirectly use our library to declare them asynchronous. In concrete terms, this means adding the keyword async before the declaration and adding the keyword await when used.
The final step is to eliminate the use of the async_add_executor_job method and call our asynchronous functions directly.
In the end, the example above becomes:
from yoctolib.yocto_colorledcluster_aio import *
async def set_color(self, hwid: str, rgb_color) -> None:
yled = YColorLedCluster.FindColorLedClusterInContext(self.yctx, hwid)
if await yled.isOnline():
nbled = await yled.get_activeLedCount()
_LOGGER.debug("Set %s to color 0x%x" % (hwid, rgb_color))
await yled.rgb_move(0, nbled, rgb_color, 1000)
else:
_LOGGER.warning("Module %s not offline" % hwid)
async def set_on_off(self) -> None:
if self._is_on:
color = (
(self._rgb_color[0] << 16)
+ (self._rgb_color[1] << 8)
+ self._rgb_color[2]
)
await self.set_color(self._name, color)
self._is_on = True
else:
await self._hub.set_color(self._name, 0)
self._is_on = False
async def async_turn_off(self, **kwargs: Any) -> None:
self._is_on = False
await self.set_on_off()
These modifications aren't very complicated, especially as modern IDEs are able to detect if you've forgotten to use await when calling an asynchronous function, or vice versa.
If you want to see all the changes required for our extension, you can consult the change list on GitHub.
Once these changes have been made, you can use the extension and these modifications are invisible to the end user. Installing or updating the extension is done in the same way. Incidentally, the installation method documented in the previous post remains unchanged.
Bonus
Note that, as we already had our hands dirty, we took the opportunity to add other features to this extension, such as support for Yocto-Display. From now on, the integration adds a Text entity to display a value on Yocto-Display and Yocto-MaxiDisplay.
The Text entity for the Yocto-Display
This makes it possible for instance to create an automation that displays the solar production of the building.
Conclusion
As we have seen, the new Typed Python library makes it easy to convert classic Python code to interact with Yoctopuce modules in asynchronous mode. This new library thus allows for much better integration into applications that operate in asynchronous mode, such as Home Assistant.
