Bonjour,

j'ai cherché sur le forum et je n'est pas trouvé ce que je cherchais.

J'ai un programme qui récupère un calendrier ICS sur un serveur, c'est vraiment simple, mais en fonction de la période que l'on veux, cela peut prendre un peut de temps. pour l'instant j'ai cela (c'est juste un ECM)

import urllib.request                               
import urllib.error

def telechargement():
    try:
        url = "http://moncalendar.com/cal.ics"
        print("Téléchargement du calendrier")   
        page = urllib.request.urlopen(url)
        print("Téléchargement fini")                                            
        calendar = page.read()
        calendar = calendar.decode("utf8")
        return str(calendar)
    except urllib.error.URLError as e:
        print(e.reason)
    except urllib.error.HTTPError as e:
        print(e.code)

calendar = telechargement()
#[...]traitement du calendrier, enregistrement dans un fichier et toussa toussa

 
comme vous pouvez le voir c'est pas vraiment compliqué, j'aimerais simplement que pendant l’exécution de urlopen() j'ai quelque chose comme ceci qui s'affiche :

Téléchargement                                            |

avec un tourniquet comme ceci =>|\-/|

une fois que le fichier est téléchargé, cela afficherait :

Téléchargement                                        [OK]

Je ne vois pas bien comment exécuter mon tourniquet pendant que urlopen() mouline... en C je ferais surement un fork, mais ici...

Merci d'avance.

Salut,

Tu peux tout à fait faire un fork en Python aussi avec os.fork() (sauf sous windows), ou mieux, le module multiprocessing, mais en réalité, que ce soit en Python ou en C, ce n'est pas la meilleure solution.

Pour ce genre de choses, tu as besoin que ton code principal ait au moins une variable partagée avec celui du tourniquet. Le plus simple est donc d'exécuter ces parties dans des threads séparés à l'intérieur du même processus, plutôt que dans des processus séparés. En C, ça se ferait avec des pthreads (POSIX threads). En Python, tu as le module threading, ou à plus bas niveau avec la fonction _thread.start_new_thread() qui suffit largement dans un cas aussi simple.

C'est un petit peu tricky si tu n'as pas l'habitude de travailler avec les threads, mais ça se fait plutôt bien. Voilà un exemple en Python3. Tu peux directement importer ce module pour l'utiliser dans ton code :

#!/usr/bin/env python3
# -*- coding: utf-8 -*-

from time import sleep
from sys import stdout
from itertools import cycle

try:
    from _thread import start_new_thread
except ImportError:
    from _dummy_thread import start_new_thread

def show_working(task, task_name=""):
    working = True       # Variable partagée pour arrêter le sablier
    def roll():
        for bar in cycle("|/-\\"):
            if not working:
                return
            stdout.write("\r%s\t%s" % (task_name, bar))
            stdout.flush()
            sleep(0.2)

    # Le sablier s'exécute dans un thread séparé
    start_new_thread(roll, ())
    try:
        res = task()
    finally:
        # Arrêt du sablier, même en cas d'erreur
        # dans le thread principal
        working = False  

    stdout.write("\r%s\t[OK]\n" % (task_name))
    return res

# Test
if __name__ == '__main__':
    def example_task():
        for i in range(10):
            sleep(0.5)
        return 10

    res = show_working(example_task, "Counting to ten in my head")
    print("task returned: ", res)

Dans ton code, il suffit de remplacer les lignes 7 à 9 par ceci :

page = show_working(lambda: urllib.request.urlopen(url), "Téléchargement")

Note à part

C'est typiquement le genre de problématique simple et intéressante pour s'initier aux threads. Un exercice sympa serait de définir une classe facile à utiliser pour afficher une progression un peu comme ça au lieu d'un sablier :

Nom de la tâche [########            ] 40%

Par exemple, on créerait l'afficheur avec le constructeur task = ProgressTask("Nom de la tâche"). Puis on mettrait l'affichache à jour avec task.progress = 40. Puis on terminerait avec task.finish() (qui afficherait le nom avec [OK]).

Et ensuite, étendre ça au multithread en loggant plusieurs tâches simultanées :

Tâche truc    [OK]
[Tâche machin 30%][Tâche bidule 70%]

Dès qu'une tâche est terminée, on affiche son nom suivi de [OK] puis on continue à afficher les progressions des autres sur une nouvelle ligne. Ça, ça nécessite clairement l'emploi du module threading, et c'est loin d'être trivial à réaliser, mais c'est carrément formateur.

-
Edité par nohar 3 février 2013 à 14:46:17

Wouah, j'ai essayer de chercher, mais je n'arrivais pas a une solution générique comme la tienne ! Merci de me donner le code.

j'ai essayé avec sa :

#!/usr/bin/env python3.3

import threading
import time
import urllib.request as wget                                                   
from os import system

class Attente(threading.Thread):
    def __init__(self):
        threading.Thread.__init__(self)
        self.Terminated = False
    def run(self):
        curseur = ("|","/","-","\\")
        while not self.Terminated:
            for x in curseur:
                print("Téléchargement                   ",x)
                time.sleep(.5)
    def stop(self):
        self.Terminated = True
        print("Téléchargement                                       [OK]")


progess = Attente()
progress.run()
myPage = wget.urlopen("http://www.google.com")
progress.stop()

Mais cela ne fonctionnais pas.

Edit : ton code chez moi, n'affiche quelque chose que lorsque le fichier a fini de télécharger.

Je suis sur mac, mais je ne vois pas en quoi tous cela serait différent de GNU/Linux ? même gestion (enfin en programmation) des thread et compagnie.

-
Edité par MrMi 3 février 2013 à 15:02:46

C'est bizarre. Je me demande si MacOS supporte les pthreads...

Je vais me renseigner et je te tiens au courant. Au pire, tu peux essayer de reprendre mon code en utilisant le module multiprocessing et en envoyant au processus fils un SIGTERM grâce à la méthode terminate(), mais ça semble légèrement overkill quand même.

Voilà une version qui utilise le module threading au lieu de l'interface bas niveau :

#!/usr/bin/env python3
# -*- coding: utf-8 -*-

from time import sleep
from sys import stdout
from itertools import cycle
from threading import Thread


def show_working(task, task_name=""):
    working = True       # Variable partagée pour arrêter le sablier
    def roll():
        for bar in cycle("|/-\\"):
            if not working:
                return
            stdout.write("\r%s\t%s" % (task_name, bar))
            stdout.flush()
            sleep(0.2)

    # Le sablier s'exécute dans un thread séparé
    t = Thread(target=roll)
    t.start()
    try:
        res = task()
    finally:
        # Arrêt du sablier, même en cas d'erreur
        # dans le thread principal
        working = False  

    t.join()
    stdout.write("\r%s\t[OK]\n" % (task_name))
    return res

# Test
if __name__ == '__main__':
    def example_task():
        sleep(5)
        return "It works!"

    res = show_working(example_task, "Counting to ten in my head")
    print("task returned: ", res)

Si celle-là ne fonctionne pas, alors il faudra utiliser le module multiprocessing. Le code ne change quasiment pas, mais la solution est beaucoup plus lourde (lors d'un fork, l'état interne du processus père est copié dans le nouveau processus, et la communication entre processus par signaux système n'est pas non plus ultra-efficace) :

#!/usr/bin/env python3
# -*- coding: utf-8 -*-

from time import sleep
from sys import stdout
from itertools import cycle
from multiprocessing import Process


def show_working(task, task_name=""):
    def roll():
        for bar in cycle("|/-\\"):
            stdout.write("\r%s\t%s" % (task_name, bar))
            stdout.flush()
            sleep(0.2)

    # Le sablier s'exécute dans un processus séparé
    p = Process(target=roll)
    p.start()
    try:
        res = task()
    finally:
        p.terminate()

    p.join()
    stdout.write("\r%s\t[OK]\n" % (task_name))
    return res

# Test
if __name__ == '__main__':
    def example_task():
        sleep(5)
        return "It works!"

    res = show_working(example_task, "Counting to ten in my head")
    print("task returned: ", res)

-
Edité par nohar 4 février 2013 à 1:13:26

À vrai dire. Il n'y a quasiment aucune chance pour que le dernier code ne fonctionne pas. C'est un peu le mortier pour écraser un moustique.

Le truc à retenir, c'est surtout que c'est un dernier recours. Utiliser un processus séparé à la place de threads, c'est justifié si :

• Tu as besoin que l'OS gère ses ressources (mémoire, CPU) de façon indépendante, par exemple s'il fait beaucoup d'IO ou de calculs,

• Le morceau de code à exécuter est suffisamment isolé du reste du programme pour que le besoin de communication entre processus soit minimal (si on oublie l'écriture dans un RAMFS que tout le monde n'a pas, il reste les signaux, les pipes et les sockets... autant dire que ça limite les possibilités !).

Ici, la communication est minimale (juste un SIGTERM), donc ça s'y prête, mais j'aurais quand même plutôt tendance à préférer les threads, puisque le code du sablier passe plus de 90% de son temps à dormir.

-
Edité par nohar 6 février 2013 à 11:05:45

Merci pour les précisions, du coup j'ai fait essayé avec multiprossessing, cela fonctionne en test, mais une fois intégrer dans mon programme sa plante un peu. J'ai une erreur de type "TypeError: 'str' object is not callable" pourtant, je n'ai utiliser str comme variable ou quelque chose comme sa...

bref je join un ECM

#je ne join pas show_working, même code que precedement
import urllib.request as wget
import datetime
import urllib.error 

def telecharge():
    #génération de l'heure
    dateDebut = datetime.date.today()
    dateDebut = str(dateDebut)
    #+3mois (environ)
    dateFin = datetime.date.today() + datetime.timedelta(weeks=3*4)
    dateFin = str(dateFin)

    try:
        url = "http://example.com/test.ics"
        page = wget.urlopen(url)
        calendar = page.read()
        calendar = calendar.decode("utf8")
        return str(calendar)
    except urllib.error.URLError as e:
        print(e.reason)
    except urllib.error.HTTPError as e:
        print(e.code)

cal = show_working(telecharge(), "Téléchargement")

alors que...

#!/usr/bin/env python3                                                          
# -*- coding: utf-8 -*-
 
from time import sleep
from sys import stdout
from itertools import cycle
from multiprocessing import Process


def show_working(task, task_name=""):
    def roll():
        for bar in cycle("|/-\\"):
            stdout.write("\r%s\t%s" % (task_name, bar))
            stdout.flush()
            sleep(0.2)

    # Le sablier s'exécute dans un processus séparé
    p = Process(target=roll)
    p.start()
    try:
        res = task()
    finally:
        p.terminate()

    p.join()
    stdout.write("\r%s\t[OK]\n" % (task_name))
    return res

# Test
if __name__ == '__main__':
    import urllib.request
    def test():
        req = urllib.request.urlopen("http://www.google.com")
        sleep(10)
        req = req.read()
        return req

    res = show_working(test, "test")

Dernière ligne :

cal = show_working(telecharge(), "Téléchargement")

C'est la fonction qu'il faut passer en premier argument à show_working, et non le résultat de son appel :

cal = show_working(telecharge, "Téléchargement")