Python 3: Sortie propre via l'entrée-commande avec asyncio

Question

J'ai des problèmes pour obtenir une sortie propre via une commande d'entrée vers le terminal lors de l'exécution d'un script Python utilisant asyncio.

Le script fait un tas de choses mais fournit également un CLI pour l'entrée. Taper 'done' et presser enter devrait arrêter le script entier. J'ai essayé de le faire avec le code ci-dessous, mais cela ne semble pas fonctionner.

Le code pour le script:

try: 
    cm_cli = CmCli(iot_net, loop, logger) 
    cm_cli.start() 

    loop.run_forever() 
except KeyboardInterrupt: 
    print("[CerberOS_Manager] Shutting everything down.") 
    for task in asyncio.Task.all_tasks(): 
     task.close() 
    loop.stop() 
    sys.exit() 

Le code exécuté dans cm_cli en tapant 'fait':

def run(self): 
    print("[CM_CLI] Type 'help' for available commands, 'test' to run test.") 
    while True: 
     print('\033[0m', end=' ')  # stop printing yellow 
     params = (input('>> ')) 
     print('\033[1;33m', end=' ') # start printing yellow 

     params = params.split() 
     if len(params) < 1: 
      continue 

     if params[0] == "help": 
      print("[CM_CLI] Available commands are:") 
     elif params[0] == "": 
      continue 
     elif params[0] == "done": 
      print("[CM_CLI] Exiting.") 
      raise KeyboardInterrupt 
     else: 
      print("[CM_CLI] Unknown command.") 

Que se passe au moment où je commence simplement le script (appelé cerberos_manager. py) et le type fait:

pi@raspberrypi:~/git/yggdrasil_managers/cerberos_mgmt $ sudo python3 cerberos_manager.py 
[CM_CLI] Type 'help' for available commands, 'test' to run test. 
>> done 
[CM_CLI] Exiting. 
Exception in thread Thread-11: 
Traceback (most recent call last): 
    File "/usr/local/lib/python3.6/threading.py", line 916, in _bootstrap_inner 
    self.run() 
    File "/home/pi/git/yggdrasil_managers/cerberos_mgmt/cm_cli.py", line 57, in run 
    raise KeyboardInterrupt 
KeyboardInterrupt 

Donc il soulève l'interruption mais n'arrive pas à atteindre l'impression dans l'autre partie. J'ai expérimenté avec la fermeture de la boucle dans cm_cli mais je n'ai pas été plus proche. Des suggestions pour une sortie propre vers le terminal?

Answer 1

Jetez un oeil à cet exemple:

import asyncio 

def hello_world(loop): 
    print('Hello World') 
    loop.stop() 

loop = asyncio.get_event_loop() 

# Schedule a call to hello_world() 
loop.call_soon(hello_world, loop) 

# Blocking call interrupted by loop.stop() 
loop.run_forever() 
loop.close() 

Vous devez attraper exception dans la fonction de rappel (run()) et le feu loop.stop() en elle. Après cela, utilisez loop.close() dans le module principal.