Quelques concepts python

async.py

import asyncio

"""
asyncio: résoudre les problèmes d'attente
"""


class Progress:
    def __init__(self, start: float):
        self.value = start


async def telecharger_un_fichier(progression: Progress):
    progression.value = 1.0
    await asyncio.sleep(1.0)
    progression.value = 25.0
    await asyncio.sleep(1.0)
    progression.value = 55.0
    await asyncio.sleep(1.0)
    progression.value = 99.0
    await asyncio.sleep(1.0)
    progression.value = 100.0


async def afficher_progression(progression: Progress):
    print(f"{progression.value}%")
    previous_progress = progression.value
    while progression.value != 100.0:
        await asyncio.sleep(0.25)
        if previous_progress != progression.value:
            print(f"{progression.value}%")
            previous_progress = progression.value
    print("Done !")


async def main():
    progress = Progress(0.0)
    await asyncio.gather(
        telecharger_un_fichier(progress),
        afficher_progression(progress),
    )


asyncio.run(main())

context.py

"""
Gestionnaire de contexte : appliquer un comportement avant et après
"""


class MonFichier(object):
    def __init__(self, file_name: str):
        self.file = open(file_name, "w+")

    def __enter__(self):
        print("ouverture du fichier")
        return self.file

    def __exit__(self, exception_type, value, traceback):
        print(f"fermeture du fichier ({exception_type}, {value}, {traceback})")
        self.file.close()


import contextlib
@contextlib.contextmanager
def mon_fichier(file_name: str):
    print("ouverture du fichier")
    file = open(file_name, "w+")
    try:
        yield file
    finally:
        print(f"fermeture du fichier")
        file.close()


print("Utilisation sans accro")
with MonFichier("/tmp/toto.txt") as f:
    f.write("hello world")

print()

print("Utilisation avec accro")
with MonFichier("/tmp/toto.txt") as f:
    1/0

decorateurs.py

"""
Décorateurs : Remplacement dune fonction par une autre
"""


# Décorateur sans arguments
def mon_decorateur(fonction_decore):
    def ma_fonction_de_remplacement(*args, **kwargs):
        print("print depuis ma_fonction_de_remplacement")
        return fonction_decore(*args, **kwargs)
    return ma_fonction_de_remplacement


# Décorateur avec arguments
def mon_constructeur_de_decorateur(un_argument: str):
    def mon_decorateur(fonction_decore):
        def ma_fonction_de_remplacement(*args, **kwargs):
            print(f"print depuis ma_fonction_de_remplacement, j'aime jouer du: {un_argument}")
            return fonction_decore(*args, **kwargs)

        return ma_fonction_de_remplacement
    return mon_decorateur


@mon_decorateur
def ma_fonction_a_decorer(a: int) -> int:
    return a + 1


@mon_constructeur_de_decorateur("trombonne")
def ma_fonction_a_decorer2(a: int) -> int:
    return a + 1


print("Décorateur sans arguments")
print(ma_fonction_a_decorer)
print(ma_fonction_a_decorer(41))

print()

print("Décorateur avec arguments")
print(ma_fonction_a_decorer2)
print(ma_fonction_a_decorer2(41))

generateur.py

"""
Générateur : un stream
"""


def generator_function():
    yield 1
    yield 1
    yield 2


print("Comme un iterateur")
for i in generator_function():
    print(i)

print("Comme un objet gnerateur")
generateur = generator_function()
print(next(generateur))
print(next(generateur))
print(next(generateur))

print()

print("Si on l'appelle une fois de trop")
print(next(generateur))
print(next(generateur))
print(next(generateur))
print(next(generateur))

print()

print("Envoyer une valeur au generateur")
# Recevoir des valeurs dans le generateur
def generateur_pair_ou_impair():
    filtre = True
    input_ = None
    for i in range(20):
        if (i % 2 == 0) == filtre:
            input_ = yield i
        if input_ is not None:
            if input_ == "pair":
                filtre = True
            elif input_ == "impair":
                filtre = False


gen = generateur_pair_ou_impair()
for i in gen:
    print(i)
    if i == 10:
        print(gen.send("impair"))

process.py

"""
Processing : code concurrent
"""

import multiprocessing
import time


def recevoir_meteo(queue_: multiprocessing.Queue, stop_signal_: multiprocessing.Event):
    counter = 0
    soleil = True
    while not stop_signal_.is_set():
        counter += 1
        if counter % 5 == 0:
            soleil = not soleil
            counter = 0

        queue_.put(soleil)
        time.sleep(1.0)


stop_signal = multiprocessing.Event()
queue = multiprocessing.Queue()
meteo_process = multiprocessing.Process(target=recevoir_meteo, args=(queue, stop_signal))
meteo_process.start()

counter = 0
while True:
    soleil = queue.get(block=True)
    counter += 1

    if soleil:
        print("☀")
    else:
        print("☁")

    if counter == 10:
        stop_signal.set()
        break

thread.py

"""
Thread : sur deux fils mais pas concurrent
"""

import threading
import time


def mettre_a_jour_meteo(meteo_: dict, stop_signal_: threading.Event) -> None:
    counter = 0
    while not stop_signal_.is_set():
        counter += 1
        if counter % 5 == 0:
            meteo_["soleil"] = not meteo_["soleil"]
            meteo_["nuage"] = not meteo_["nuage"]
            counter = 0
        time.sleep(1.0)


meteo = {
    "soleil": True,
    "nuage": False,
}

stop_signal_ = threading.Event()
meteo_thread = threading.Thread(target=mettre_a_jour_meteo, args=(meteo, stop_signal_))
meteo_thread.start()

for _ in range(15):
    if meteo["soleil"]:
        print("☀")
    if meteo["nuage"]:
        print("☁")
    time.sleep(1)

stop_signal_.set()