Discussion :

[mandrain]

Bonjour,
Je suis un débutant avec Python 3 et Raspberry 3 B+ et j’ai quelques difficultés avec ce petit programme.
Je voudrais que la boucle while s’arrête lorsque j’appuie sur CTRL+C.
Les deux fonctions suivantes ne servent que pour les essais.
La fonction Afficheur(Thread) comme son nom l’indique n’affiche que 10 fois le caractère ‘A’.
La fonction Essais_1(Thread) fait clignoter une led .
L’ensemble me permet d’essayer les threads.
Dans ce programme l’exception ne fonctionne pas et je ne vois pas pourquoi.

Code :

Sélectionner tout - Visualiser dans une fenêtre à part

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
 
# -*- coding: utf-8 -*-
"""
Created on Mar Mai 22 10:30:36 2018
@author: mandrin
"""
import os
import random
import sys
from threading import Thread
import datetime 
import time
import RPi.GPIO as GPIO #++
 
# broches GPIO
GPIO.setmode(GPIO.BCM) #++ GPIO21= broche 40
GPIO.setup(20, GPIO.OUT, initial=GPIO.LOW)# broche 38 LED #++
 
class Afficheur(Thread):
 
    """Thread chargé simplement d'afficher une lettre dans la console."""
 
    def __init__(self, lettre):
        Thread.__init__(self)
        self.lettre = lettre
 
    def run(self):
        """Code à exécuter pendant l'exécution du thread."""
        i = 0
        while i < 20:
            sys.stdout.write(self.lettre)
            sys.stdout.flush() # affichage immediat
            attente = 0.2
            attente += random.randint(1, 60) / 100 # attente aleatoire de 0.2 a 0.8 s
            time.sleep(attente)
            i += 1
# fin de Afficheur
 
# allumage d'une LED ==========================================================
class Essais_1(Thread):
 
    def __init__(self):
        Thread.__init__(self)
 
    def run(self):
        date=datetime.datetime.now() #*-*
        second_old=date.second+1 #*-*
        while 1:
            try:
                date=datetime.datetime.now() #*-*
                secs=int(date.second) #*-*<
                Led=date.microsecond/1e6
# commande de la LED toute les secondes (0.5s allumee, 0.5s eteinte)
                if  Led<=0.5:
                    GPIO.output(20,GPIO.LOW)# LED  allumee #++
                    pass # equiv a nop
                else:
                    GPIO.output(20,GPIO.HIGH)# LED eteinte #++
                    pass # equiv a nop                
                if secs==second_old:
                    print('secs->',secs) #*/*
                    second_old+=1  
                    if second_old==60: 
                        second_old=0                 
            except KeyboardInterrupt:
                GPIO.cleanup()
                thread_2.join()
                break
 
# fin de while 1 =============================================================
 
thread_2 = Essais_1()           
thread_3 = Afficheur("A")
 
# Lancement des threads
thread_2.start()
thread_3.start()
 
# Attend que les threads se terminent
thread_2.join()
thread_3.join()

Si quelqu'un voit pourquoi je suis preneur.
Merci d’avance
Cordialement

[wiztricks]

Salut,

Si quelqu'un voit pourquoi je suis preneur.

C'est le thread principal qui recevra le KeyboardInterrupt et qui devra le propager (comme écrit dans la documentation sur le module "signal". ) via, par exemple, un Event (testé à chaque itération par le thread).

- W

[mandrain]

Un petit complément d’infos :
Le thread Afficheur() ne dure que 10 secondes environ et ensuite il s'arrête.
Donc il ne reste plus que Essais_1().
Dans ce cas ^c devrait être efficace. Ce n'est pas le cas.

Un ^C provoque bien une interruption du déroulement du thread mais celui ci reprend instantanément son déroulement.

De plus l'exception n'est pas détectée et je n'en vois pas la raison.

[mandrain]

A l'attention de wiztricks

Bonjour,

C'est le thread principal qui recevra le KeyboardInterrupt et qui devra le propager (comme écrit dans la documentation sur le module "signal" via, par exemple, un Event (testé à chaque itération par le thread).

La je suis largué !

Je n'ai pas de thread principal. En réalité j'ai besoin d'un thread qui tourne en permanence afin d'exécuter des commandes à la seconde près et d'autres qui font d'autres fonctions à des temps plus long comme toute les minutes, heures et jours avec des durée de cycle qui peuvent dépasser les 10 s ou plus.

C'est cela qui m'a amené aux thread. Mais je rame, serait il possible d'avoir un exemple quelque part avec une sortie qui puisse fermer des fichiers ouverts avant de quitter le programme.

L'exemple de programme joint à mon post est un raccourci afin de ne pas transmettre un document trop volumineux.

Merci d'avance.
Cordialement

[wiztricks]

Je n'ai pas de thread principal.

C'est le thread du programme.

En réalité j'ai besoin d'un thread qui tourne en permanence afin d'exécuter des commandes à la seconde près et d'autres qui font d'autres fonctions à des temps plus long comme toute les minutes, heures et jours avec des durée de cycle qui peuvent dépasser les 10 s ou plus.

Si vous avez besoin de threads, il faut apprendre à les utiliser. A la base, les threads sont des objets système. Python ne donne qu'une interface pour les utiliser depuis vos programmes avec des restrictions (et un mode d'emploi).

- W

[mandrain]

Merci, j'apprécie

Bonne journée

[wiztricks]

Ah ben, je peux vous donner aussi du code:

Code :

Sélectionner tout - Visualiser dans une fenêtre à part

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
from threading import Thread, Event
import threading
import time
 
def do_work(event, delay):
    counter = 0
    while not event.is_set():
        print(threading.get_ident(), counter)
        counter += 1
        time.sleep(delay)
    print('done')
 
delay = 0.2
print('ident:', threading.get_ident())
event = Event()
tasks = []
for _ in range(3):
    p = Thread(target=do_work, args=(event, 2*delay))
    p.start()
    tasks.append(p)
 
while True:
    try:
        print ('waiting')
        time.sleep(delay)
    except KeyboardInterrupt:
        event.set()
        break
 
for p in tasks:
    p.join()
print('all done')

Mais pas sûr que vous le compreniez sans avant pris le temps d'ouvrir la documentation...

- W