ephase/xieme-art
1
0
Fork 0
Code Issues Pull requests Releases Wiki Activity

Compare commits

base: ephase:8cf698330428e5a180d0cdc64526b915f03ac5f4
Branches Tags
ephase:main
ephase:feat/flake_integration
..
compare: ephase:6db1545e36d6444418d738a35d195ac75d5d49f9
Branches Tags
ephase:main
ephase:feat/flake_integration

No commits in common. "8cf698330428e5a180d0cdc64526b915f03ac5f4" and "6db1545e36d6444418d738a35d195ac75d5d49f9" have entirely different histories.
8cf6983304 ... 6db1545e36

6 changed files with 10 additions and 651 deletions
Download patch file Download diff file Expand all files Collapse all files

42
content/articles/2023/adafruit_macropad_asyncio/files/async_array/code.py
View file

@ -1,42 +0,0 @@
from adafruit_macropad import MacroPad
import usb_cdc, asyncio, random
macropad = MacroPad()
macropad.pixels.brightness = 0.3
color = 0xad20b4
async def get_key(taskslist):
while True:
key_event = macropad.keys.events.get()
if key_event:
if key_event.pressed:
print("key k:{} t:{}".format(key_event.key_number, len(taskslist)))
taskslist.append(asyncio.create_task(blink(key_event.key_number)))
await asyncio.sleep(0)
async def blink(led):
timer = random.random() * 3
for _ in range(5):
macropad.pixels[led] = color
await asyncio.sleep(timer)
macropad.pixels[led] = 0x0
await asyncio.sleep(timer)
async def manage_tasks(taskslist):
tasks = 0
print("Run task manager t:{}".format(len(taskslist)))
while True:
for task_number in range(0, len(taskslist)):
if taskslist[task_number].done():
print("Remove task t:{}/{}".format(task_number + 1,len(taskslist)))
taskslist.pop(task_number)
break
await asyncio.sleep(0)
async def main():
tasks = []
tasks.append(asyncio.create_task(get_key(tasks)))
tasks.append(asyncio.create_task(manage_tasks(tasks)))
await asyncio.gather(*tasks)
asyncio.run(main())

48
content/articles/2023/adafruit_macropad_asyncio/files/async_blink/code.py
View file

@ -1,48 +0,0 @@
from adafruit_macropad import MacroPad
import usb_cdc, asyncio
macropad = MacroPad()
timer = 1
color = 0xad20b4
led = 0
def exec_command (data):
global timer
command,option = data.split()
print("cmd:{} opt:{}".format(command,option))
if command == 'time':
timer = float(option)
async def blink(led, color):
macropad.pixels.brightness = 0.3
while True:
try:
if macropad.pixels[led] == (0,0,0):
macropad.pixels[led] = color
else:
macropad.pixels[led] = 0x0
print("l:{} c:{} t:{}".format(led,hex(color),timer))
await asyncio.sleep(timer)
except asyncio.CancelledError:
pass
async def check_serial(task_led):
data = bytearray()
serial = usb_cdc.data
# Avec le timeout, même si la ligne ne contient pas \n, elle
# sera pris en compte
serial.timeout = 1
while True:
if serial.in_waiting:
data = serial.readline()
print("serial data received")
exec_command(data.decode("utf-8"))
task_led.cancel()
await asyncio.sleep(0)
async def main():
t_led = asyncio.create_task(blink(led, color))
t_serial = asyncio.create_task(check_serial(t_led))
await asyncio.gather(t_led, t_serial)
asyncio.run(main())

45
content/articles/2023/adafruit_macropad_asyncio/files/async_serie/code.py
View file

@ -1,45 +0,0 @@
from adafruit_macropad import MacroPad
import usb_cdc, asyncio
macropad = MacroPad()
timer = 1
color = 0xad20b4
led = 0
def exec_command (data):
global timer
command,option = data.split()
print("cmd:{} opt:{}".format(command,option))
if command == 'time':
timer = float(option)
async def blink(led, color):
#global timer
macropad.pixels.brightness = 0.3
while True:
if macropad.pixels[led] == (0,0,0):
macropad.pixels[led] = color
else:
macropad.pixels[led] = 0x0
print("l:{} c:{} t:{}".format(led,hex(color),timer))
await asyncio.sleep(timer)
async def check_serial():
data = bytearray()
serial = usb_cdc.data
# Avec le timeout, même si la ligne ne contient pas \n, elle
# sera pris en compte
serial.timeout = 1
while True:
if serial.in_waiting:
data = serial.readline()
print("serial data received")
exec_command(data.decode("utf-8"))
await asyncio.sleep(0)
async def main():
t_led = asyncio.create_task(blink(led, color))
t_serial = asyncio.create_task(check_serial())
await asyncio.gather(t_led, t_serial)
asyncio.run(main())

52
content/articles/2023/adafruit_macropad_asyncio/files/lock/code.py
View file

@ -1,52 +0,0 @@
from adafruit_macropad import MacroPad
import usb_cdc, asyncio, random
macropad = MacroPad()
macropad.pixels.brightness = 0.3
run_color = 0xad20b4
wait_color = 0x21f312
# déclaration de notre Mutex
blink_mutex = asyncio.Lock()
async def get_key(taskslist):
while True:
key_event = macropad.keys.events.get()
if key_event:
if key_event.pressed:
taskslist.append(asyncio.create_task(blink(key_event.key_number)))
await asyncio.sleep(0)
async def blink(led):
timer = random.random() * 3
macropad.pixels[led] = wait_color
if blink_mutex.locked():
print("Wait for mutex l:{}".format(led))
await blink_mutex.acquire()
print("Aquire mutex l:{}".format(led))
for _ in range(5):
macropad.pixels[led] = run_color
await asyncio.sleep(timer)
macropad.pixels[led] = 0x0
await asyncio.sleep(timer)
blink_mutex.release()
async def manage_tasks(taskslist):
tasks = 0
while True:
for task_number in range(0, len(taskslist)):
if taskslist[task_number].done():
taskslist.pop(task_number)
break
await asyncio.sleep(0)
async def main():
tasks = []
tasks.append(asyncio.create_task(get_key(tasks)))
tasks.append(asyncio.create_task(manage_tasks(tasks)))
await asyncio.gather(*tasks)
asyncio.run(main())

452
content/articles/2023/adafruit_macropad_asyncio/index.md
View file

@ -1,452 +0,0 @@
Title: Adafruit Macropad : multitâche coopératif avec asyncio
Category: linux
Tags: Adafruit, CircuitPython, Python
Date: 2023-07-11 15:10
cover: assets/backgrounds/Adafruit_Macropad.jpg
status: draft
Nous avons découvert dans mon
[précédent article]({filename}../adafruit_macropad_conn_serie/index.md)
le Macropad et comment utiliser la connexion série pour envoyer et recevoir des
données depuis un ordinateur.
Dans la première partie de cet article, nous avions été confronté à un problème
alors que nous voulions changer la fréquence du clignotement de la DEL : il
fallait attendre que l'instruction `time.sleep` soit terminée.
Nous allons vois ici comment utiliser `asyncio` et l'implémentation de `async` /
`await` pour corriger ce problème. Nous essaierons ensuite d'aller plus loin
afin d'appréhender cette bibliothèque.
## Qu'est-ce que `asyncio`
C'est une bibliothèque de *Python* qui permet de réaliser de la **programmation
asynchrone**. Le principe ici est d'éviter que le programme attende sans rien
faire. Ça tombe bien puisque dans l'exemple de la DEL qui clignote, notre
programme ne fait qu'attendre, il est d'ailleurs impossible de traiter les
entrées / sorties sur le port série à ce moment.
Attention, il n'est **pas question ici de parallélisme**, `asyncio` ne gère pas
de fils d'exécution. Dans l'interpréteur Python c'est la classe `Thread` qui
s'en charge. Mais d'abord *CircuitPython* ne [les supporte pas](l_threads_cp) et
ils sont plus difficiles à gérer (concurrences, sections critiques, etc.).
Dans le cas d'`asyncio`, nous parlerons [**de coroutines**][l_coroutines] pour
des fonctions déclarée avec le mot clef `async`.
[l_threads_cp]:https://github.com/adafruit/circuitpython/issues/1124
[l_coroutines]:https://fr.wikipedia.org/wiki/Coroutine
## Installer `asyncio` sur le Macropad
Il est possible que la bibliothèque `asyncio` et ses dépendances ne soient pas
installées, il est alors possible de les installer avec `circup`. Commençons pas
créer un *environnement virtuel Python*:
```
python -m venv ~/venv_circup
source ~/venv_circup/bin/activate
pip install circup
```
Branchez ensuite le Macropad, montez le périphérique de stockage CIRCUITPY puis
lancez la commande suivante:
```
circup install asyncio
```
Et voilà!
## La DEL clignotante
Utilisons maintenant `asyncio` pour réécrire le code de la DEL clignotante.
Voici le code à placer dans le fichier `code.py` à la racine du disque
*CIRCUITPY* vous pouvez télécharger le fichier `code.py`
[ici]({attach}./files/async_serie/code.py):
```python
from adafruit_macropad import MacroPad
import usb_cdc, asyncio
macropad = MacroPad()
timer = 1
color = 0xad20b4
led = 0
def exec_command (data):
global timer
command,option = data.split()
print("cmd:{} opt:{}".format(command,option))
if command == 'time':
timer = float(option)
async def blink(led, color):
#global timer
macropad.pixels.brightness = 0.3
while True:
if macropad.pixels[led] == (0,0,0):
macropad.pixels[led] = color
else:
macropad.pixels[led] = 0x0
print("l:{} c:{} t:{}".format(led,hex(color),timer))
await asyncio.sleep(timer)
async def check_serial():
data = bytearray()
serial = usb_cdc.data
# Avec le timeout, même si la ligne ne contient pas \n, elle
# sera pris en compte
serial.timeout = 1
while True:
if serial.in_waiting:
data = serial.readline()
print("serial data received")
exec_command(data.decode("utf-8"))
await asyncio.sleep(0)
async def main():
t_led = asyncio.create_task(blink(led, color))
t_serial = asyncio.create_task(check_serial())
await asyncio.gather(t_led, t_serial)
asyncio.run(main())
```
Notre fonction de clignotement est définie par `async def blink()` : c'est
maintenant une fonction asynchrone. Une fois l'état de la DEL changé, notre
fonction d'attente `asyncio.sleep()` remplace `timer.sleep`. C'est en effet
ce module qui se charge maintenant de l'attente, ainsi il pourra gérer
**l'enchainement des tâches**.
La partie du code utilisé pour la gestion des entrées par le port série est elle
aussi placée dans la fonction asynchrone `check_serial()`. Cette fois
`asyncio.sleep` est définis à 0 afin de laisser l'ordonnanceur du module `asyncio`
interrompre notre fonction et ainsi **empêcher une coroutine à l'exécution
longue bloquer les autres**.
Une nouvelle fonction asynchrone fait son apparition : `main()`. C'est ici que
nous définissions nos tâches (contenants nos *coroutines*) `t_led` et`t_serial`.
Ensuite `asyncio.gather()` permet de lancer l'exécution concurrente de nos deux
tâches.
Enfin, notre fonction `main()` est lancée via `asyncio.run` permettant alors de
de l'exécuter jusqu'à sa fin.
Une fois le fichier sauvegardé, la DEL n°0 devrait se mettre à clignoter avec
une fréquence d'une seconde.
### Oui mais!
Lançons *minicom* sur le port série affichant la console . Nous observons
ainsi les actions effectuées par notre programme grâce au différents `print`
disséminées dans le code.
```shell
minicom -D /dev/ttyACM0 -c on
```
Maintenant, depuis un autre terminal, changeons la fréquence de clignotement de
la DEL pour la placer à 20 secondes puis le remettre à 1 seconde:
```shell
printf "time 20\n" > /dev/ttyACM1
printf "time 1\n" > /dev/ttyACM1
```
En observant les messages affichés dans la *minicom*, nous pouvons voir que
contrairement à la version sans coroutine, **les messages sont reçus et traités
immédiatement**. Voici les messages affichés via nos différents `print()` dans
*minicom*:
```shell
[...]
l:0 c:0xad20b4 t:1.0
l:0 c:0xad20b4 t:1.0
l:0 c:0xad20b4 t:1.0
serial data received
cmd:time opt:20
l:0 c:0xad20b4 t:20.0
serial data received
cmd:time opt:1
l:0 c:0xad20b4 t:1.0
l:0 c:0xad20b4 t:1.0
[...]
```
Les lignes `serial data received` et `cmd:time opt:<>` sont apparues
immédiatement, mais lors du changement de fréquence de 20 à 1, il a fallu
**attendre que les 20 secondes définies précédement soient ecoulées** pour que
la modification soit prise en compte par la coroutine `blink()`.
## Prendre en compte immédiatement la modification de fréquence
Comment pouvons nous faire en sorte que la modification de fréquence soit
immédiatement répercutée? Tout simplement en **utilisant la méthode `cancel()`**
sur notre tache `t_led`. Celle-ci lève l'exception `CancelError` dans la
coroutine `blink()`. C'est bien entendu à nous d'implémenter la gestion de cette
exception.
Vous pouvez télécharger le fichier `code.py`
[ici]({attach}./files/async_blink/code.py), voici le code:
```python
from adafruit_macropad import MacroPad
import usb_cdc, asyncio
macropad = MacroPad()
timer = 1
color = 0xad20b4
led = 0
def exec_command (data):
global timer
command,option = data.split()
print("cmd:{} opt:{}".format(command,option))
if command == 'time':
timer = float(option)
async def blink(led, color):
macropad.pixels.brightness = 0.3
while True:
try:
if macropad.pixels[led] == (0,0,0):
macropad.pixels[led] = color
else:
macropad.pixels[led] = 0x0
print("l:{} c:{} t:{}".format(led,hex(color),timer))
await asyncio.sleep(timer)
except asyncio.CancelledError:
pass
async def check_serial(task_led):
data = bytearray()
serial = usb_cdc.data
# Avec le timeout, même si la ligne ne contient pas \n, elle
# sera pris en compte
serial.timeout = 1
while True:
if serial.in_waiting:
data = serial.readline()
print("serial data received")
exec_command(data.decode("utf-8"))
task_led.cancel()
await asyncio.sleep(0)
async def main():
t_led = asyncio.create_task(blink(led, color))
t_serial = asyncio.create_task(check_serial(t_led))
await asyncio.gather(t_led, t_serial)
asyncio.run(main())
```
C'est notre fonction `check_serial()` qui lance le `cancel()` à partir de la
tache `t_led` que nous lui passons en paramètre. L'effet est alors immédiat :
l'exception `CancelError` est lancée dans `blink()` forçant cette dernière à
recommencer depuis le début du `while`. Notre modification **est donc appliquée
immédiatement**.
## Gestions dynamique des tâches
Maintenant que nous avons vu le fonctionnement de base d'`asyncio`, prenons un
exemple un peu plus complet.
Vous pouvez télécharger le fichier `code.py`
[ici]({attach}./files/async_array/code.py), voici le code:
```python
from adafruit_macropad import MacroPad
import usb_cdc, asyncio, random
macropad = MacroPad()
macropad.pixels.brightness = 0.3
color = 0xad20b4
async def get_key(taskslist):
while True:
key_event = macropad.keys.events.get()
if key_event:
if key_event.pressed:
print("key k:{} t:{}".format(key_event.key_number, len(taskslist)))
taskslist.append(asyncio.create_task(blink(key_event.key_number)))
await asyncio.sleep(0)
async def blink(led):
timer = random.random() * 3
for _ in range(5):
macropad.pixels[led] = color
await asyncio.sleep(timer)
macropad.pixels[led] = 0x0
await asyncio.sleep(timer)
async def manage_tasks(taskslist):
tasks = 0
print("Run task manager t:{}".format(len(taskslist)))
while True:
for task_number in range(0, len(taskslist)):
if taskslist[task_number].done():
print("Remove task t:{}/{}".format(task_number + 1,len(taskslist)))
taskslist.pop(task_number)
break
await asyncio.sleep(0)
async def main():
tasks = []
tasks.append(asyncio.create_task(get_key(tasks)))
tasks.append(asyncio.create_task(manage_tasks(tasks)))
await asyncio.gather(*tasks)
asyncio.run(main())
```
Lorsque nous appuyons sur une des douze touches du clavier de notre *MacroPad*,
la DEL en dessous clignote cinq fois. Vous l'aurez compris chaque clignotement
est en fait une coroutine gérée par `asyncio`.
Les différentes tâche sont répertoriée dans un tableau qui est passé à la
fonction `asyncio.gather()`. Plus intéressant, afin de supprimer de notre
tableau les coroutines terminée, nous passons par la fonction `manage_tasks()`
en utilisant `asyncio.done()` afin de vérifier que la tâche testée soit bien
terminée.
Nous avons donc deux tâches lancées dès le départ:
* `get_key()` changée d'écouter les évènements clavier et de lancer les
différents clignotements en fonction de la touche appuyée;
* `manage_tasks` chargée de nettoyer le tableau des tâches;
Notre fonction `manage_tasks()` est simple, nous parcourons l'ensemble du
tableau et lorsque nous trouvons une coroutine terminée nous supprimons
l'élément du tableau. Remarquez le `break` lorsque notre fonction supprime un
élément de notre `taskslist`, c'est nécessaire afin d'éviter une erreur d'index
dut au fait que **notre tableau contiendra un élément de moins**;
Voici l'affichage des informations sur l'écran du *Macropad* (ou sur la console
*miniciom*, qui est plus agréable) :
```shell
Run task manager t:2
key k:0 t:2
key k:3 t:3
key k:6 t:4
Remove task t:3/5
Remove task t:4/4
Remove task t:3/3
```
Nous pouvons ainsi voir la création des tâches par la fonction `get_key()` puis
leur destruction dans `manage_tasks()` -- merci les `print()`.
## Gestion de la concurrence
Contrairement à sa grande [sœur *CPython*][l_cpython], notre implémentation de
Python ici présente ne contient pas de primitive de synchronisation comme les
sémaphores, variables conditions, ... : seul le `Lock` sont présent.
C'est l'implémentation Python des [Mutex][l_mutex]: l'accès à une
partie du code protégée par un *Mutex* sera sérialisé. Une seule coroutine y
aura accès à la fois, les autres attendent leur tour une à une.
Voici un exemple de code pour notre Macropad utilisant `Lock`, vous pouvez
télécharger le fichier `code.py` [ici]({attach}./files/lock/code.py):
```python
from adafruit_macropad import MacroPad
import usb_cdc, asyncio, random
macropad = MacroPad()
macropad.pixels.brightness = 0.3
run_color = 0xad20b4
wait_color = 0x21f312
# déclaration de notre Mutex
blink_mutex = asyncio.Lock()
async def get_key(taskslist):
while True:
key_event = macropad.keys.events.get()
if key_event:
if key_event.pressed:
taskslist.append(asyncio.create_task(blink(key_event.key_number)))
await asyncio.sleep(0)
async def blink(led):
timer = random.random() * 3
macropad.pixels[led] = wait_color
if blink_mutex.locked():
print("Wait for mutex l:{}".format(led))
await blink_mutex.acquire()
print("Aquire mutex l:{}".format(led))
for _ in range(5):
macropad.pixels[led] = run_color
await asyncio.sleep(timer)
macropad.pixels[led] = 0x0
await asyncio.sleep(timer)
blink_mutex.release()
async def manage_tasks(taskslist):
tasks = 0
while True:
for task_number in range(0, len(taskslist)):
if taskslist[task_number].done():
taskslist.pop(task_number)
break
await asyncio.sleep(0)
async def main():
tasks = []
tasks.append(asyncio.create_task(get_key(tasks)))
tasks.append(asyncio.create_task(manage_tasks(tasks)))
await asyncio.gather(*tasks)
asyncio.run(main())
```
Lors de l'appui sur une touche, la fonction `blink()` DEL passe au vert, si le
*Mutex* est libre, on passe de suite à la phase de clignotement. Ainsi une tâche
en attente sera matérialisée.
Nous pouvons observer sur l'écran du *Macropad* différentes informations
relatives à l'attente / acquisitions de notre `blink_mutex`. Ces informations
permettent de comprendre ce qui se passe lors de l'exécution de nos coroutines :
```shell
run task manager t:2
Aquire mutex l:3
Wait for mutex l:7
Wait for mutex l:11
Aquire mutex l:7
Aquire mutex l:11
Aquire mutex l:0
Wait for mutex l:5
Wait for mutex l:6
Aquire mutex l:5
Aquire mutex l:6
```
[l_cpython]:https://docs.python.org/3/library/asyncio-sync.html
[l_mutex]:https://fr.wikipedia.org/wiki/Exclusion_mutuelle
## En conclusion
Nous avons vu dans cet article, au fils des différents exemples, comment
utiliser `asyncio` pour gérer des coroutines, permettant d'implémenter dans nos
programme pour notre Macropad du multitĉhe coopératif.
Un petit tous sur [la documentation][L-circuip_async] spécifique à `asyncio`
dans CircuitPython vous permettra d'aller plus loin. Elle est cependant très
succincte et manque cruellement d'exemples. Si vous voulez en apprendre plus
sur la programmation asynchrone en Python, vous avez aussi l'excellent tutoriel
de nohar sur [Zeste de savoir][l_zeste], il m'a été d'une grande aide pour la
rédaction de cet article.
[L-circuip_async]:https://docs.circuitpython.org/projects/asyncio/en/latest/index.html
[l_zeste]:https://zestedesavoir.com/articles/1568/decouvrons-la-programmation-asynchrone-en-python/
*[DEL]: Diode ÉlectroLuminescente

22
requirements.txt
View file

@ -1,21 +1,19 @@
blinker==1.6.2 blinker==1.5
commonmark==0.9.1 commonmark==0.9.1
docutils==0.20.1 docutils==0.19
feedgenerator==2.1.0 feedgenerator==2.0.0
invoke==2.1.3 invoke==1.7.3
Jinja2==3.1.2 Jinja2==3.1.2
livereload==2.6.3 livereload==2.6.3
Markdown==3.4.3 Markdown==3.4.1
markdown-it-py==3.0.0 MarkupSafe==2.1.1
MarkupSafe==2.1.3
mdurl==0.1.2
pelican==4.8.0 pelican==4.8.0
Pygments==2.15.1 Pygments==2.13.0
python-dateutil==2.8.2 python-dateutil==2.8.2
pytz==2023.3 pytz==2022.6
rich==13.4.2 rich==12.6.0
six==1.16.0 six==1.16.0
smartypants==2.0.1 smartypants==2.0.1
tornado==6.3.2 tornado==6.2
typogrify==2.0.7 typogrify==2.0.7
Unidecode==1.3.6 Unidecode==1.3.6