Pour interagir avec un programme en ligne de commande exécuté depuis un script Python, ce n’est pas aussi simple que ça n’y paraît…

Supposons que nous ayons un programme externe qui s’appelle question.py :

import random

if __name__ == '__main__':
    print("Faisons quelques multiplications...")
    score = 0
    for i in range(3):
        a = random.randrange(2, 20)
        b = random.randrange(2, 20)
        res = int(input(f"{a} x {b} = "))
        if res == a * b:
            print("Bravo !")
            score += 1
        else:
            print(f"Dommage, la réponse était {a * b}.")
    print(f"Vous avez {score} bonne(s) réponse(s) sur 3.")

Ce programme va poser plusieurs multiplications à la suite et comptera combien de bonnes réponses ont été données. Simple.

L’objectif est de créer un programme Python qui va :

• Exécuter question.py
• Lire les questions.
• Calculer pour chaque question la bonne réponse.
• Envoyer la réponse au programme.

jusqu’à ce que le programme se termine.

On pourrait se dire qu’il suffit d’exécuter le programme avec un Popen et de lire les lignes une à une.
Problème : comme le programme externe est en attente d’une réponse, la fonction readline()reste bloquée.

Pour résoudre le problème, on peut utiliser la librairie asyncio qui permet de manipuler des Stream de manière asynchrone. Ainsi on pourra lire tout en appliquant un timeout.

Exemple de solution :

# -*- coding: utf-8 -*-
import asyncio
import sys
from asyncio.subprocess import PIPE, STDOUT
import re 

async def read_line(process, timeout=None):
    """
    Read line from process until timeout.
    """
    line = b''
    while True:
        try:
            char = await asyncio.wait_for(process.stdout.read(1), timeout)
            if not char:
                return line.decode('utf-8')
            line += char
            
        except asyncio.TimeoutError:
            return line.decode('utf-8')


async def run_command(*args, timeout=None):
    process = await asyncio.create_subprocess_exec(*args, stdout=PIPE, stdin=PIPE, stderr=STDOUT)

    while True:
        txt = await read_line(process, timeout)
        if not txt:
            break
        print(txt)
        res = re.search("(\d+) x (\d+)", txt.split("\n")[-1])
        if res:
            a, b = res.groups()
            c = str(int(a) * int(b))
            print(f'>> On va répondre {c}')
            
            process.stdin.write(c.encode())
            process.stdin.write(b'\n')
    return await process.wait()


if __name__ == '__main__':
    if sys.platform == "win32":
        loop = asyncio.ProactorEventLoop()
        asyncio.set_event_loop(loop)
    else:
        loop = asyncio.get_event_loop()

    returncode = loop.run_until_complete(run_command('python', 'question.py', timeout=1))
    loop.close()