Il y a quelques semaines, nous annoncions notre nouvelle librairie de programmation Typed Python qui offre, en plus de l'api "traditionnelle", une API asynchrone. Cette semaine, on va voir comment utiliser ces fonctions asynchrones en mettant à jour l'intégration Home Assistant que nous avions décrite dans cet article.
L'intégration Yocto-ColorLedCluster permet de contrôler les LEDs des modules Yocto-Color-V2, Yocto-MaxiBuzzer et Yocto-MaxiKnob depuis Home Assistant. Le but de cette intégration était principalement d'avoir une implémentation de référence pour les clients qui aimeraient utiliser des fonctionnalités avancées de nos modules dans Home Assitant.
Home Assistant est passé depuis un bon moment en programmation asynchrone et, quand nous avions écrit cette intégration, nous avions dû écrire des wrappers qui utilisaient async_add_executor_job pour appeler notre librairie Python traditionnel.
Par exemple, voilà la partie du code qui éteignait les 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)
Les modifications
Nous allons voir qu'il n'est pas très compliqué de modifier cette extension pour avoir une version complètement async.
La première étape est de modifier l'extension pour qu'elle référence notre nouvelle librairie. Pour ce faire, il faut simplement mettre à jour le champ "requirements" du manifeste pour qu'il utilise le package PyPi de la libraire Typed Python. Concrètement, il faut remplacer "yoctopuce==2.X.XXX" par "yoctolib==2.X.XXX".
Voilà le fichier manifest.json mis à jour.
"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": []
}
La deuxième étape est de modifier les fichiers sources de l'extension pour qu'ils importent la bonne version de la librairie. Comme nous l'avons expliqué dans l'article qui annonce la librairie Typed Pyhton, l'API Yoctopuce est disponible en version synchrone et en version asynchrone. Pour utiliser la version asynchrone, il faut ajouter le suffixe "_aio" aux imports.
Donc, pour utiliser la version asynchrone des classes YAPI et la classe YColorLedCluster, il faut utiliser le code suivant:
from yoctolib.yocto_colorledcluster_aio import YColorLedCluster
Puis, il faut modifier le code de l'extension pour convertir le code synchrone en code asynchrones.
Il faut ajouter les await à tous les appels aux fonctions asynchrones de notre librairie, comme YAPI.RegisterHub(), isOnline(), rgb_move(), etc.
Ensuite, il faut modifier la déclaration de toutes les fonctions qui utilisent directement ou indirectement notre librairie pour les déclarer asynchrones. Concrètement, il faut ajouter le mot-clef async devant la déclaration et ajouter le mot-clef await lors de l'utilisation.
La dernière étape est de supprimer l'utilisation de la méthode async_add_executor_job et d'appeler directement nos fonctions asynchrones.
Au final, l'exemple plus haut devient:
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()
Ces modifications ne sont pas très compliquées, d'autant plus que les IDEs modernes sont capables de détecter si on a oublié d'utiliser await lors de l'appel à une fonction asynchrone, ou vice versa.
Si vous voulez voir tous les changements nécessaires pour notre extension, vous pouvez consulter la change list sur GitHub.
Une fois ces modifications faites, l'extension est utilisable et ces modifications sont invisibles pour l'utilisateur final. L'installation ou la mise à jour de l'extension se fait de la même manière. Du reste, la méthode d'installation documentée dans le précédent article est inchangée.
Bonus
Notez que, comme on avait les mains dans le cambouis, on en a profité pour ajouter d'autres fonctionnalités à cette extension, comme le support pour les Yocto-Display. Désormais, l'intégration ajoute une entité Text qui permet d'afficher une valeur sur les Yocto-Display et Yocto-MaxiDisplay.
L'entité Text pour le Yocto-Display
Cela permet de créer une automatisation qui affiche la production solaire de l'habitation.
Conclusion
Comme on a pu le voir, la nouvelle librairie Typed Python permet de convertir facilement du code Python classique pour intéragir avec le modules Yoctopuce en mode asynchrone. Cette nouvelle librairie permet ainsi de s'intégrer beaucoup mieux dans des applications qui fonctionnent en mode asynchrone, comme Home Assistant.
