Python asyncio: unterbrechbare Aufgabe

Question

Ich versuche, ein Barebones Skelett Kampfspiel mit Python asyncio zu machen.

class Skeleton(Creature): 
    pass 

class SkeletonAI(): 
    def __init__(self, skeleton,loop = None): 
     self.loop = loop or asyncio.new_event_loop() 
     self.skeleton = skeleton 
     self.action_task = None 


    async def run(self): 
     while True: 
      #print(self.action_task,) 
      if self.skeleton.alive and self.skeleton.target.alive: 
       if self.skeleton.state == 'idle': 
        #ATTACK 
        self.skeleton.begin_attack() 
        self.action_task = self.loop.call_later(3, self.skeleton.action_complete) 

      else: 
       break 


class Player(Creature): 
    def attack_target(self, target): 
     target.take_damage(self.damage) 
     if target.state == 'attacking': 
      target.state = 'idle' 
      #interrupt attack 

class Game(): 

    #Super simple game 

    #The skeleton launches an attack, press anything to interrupt it 
    async def handle_sending(self): 
     loop = asyncio.get_event_loop() 
     executor = concurrent.futures.ThreadPoolExecutor(
      max_workers=1, 
     ) 
     while True: 
      msg = await loop.run_in_executor(executor, input) 
      print('got a message') 
      if self.skeleton_ai.action_task: 
       print('cancelling attack') 
       self.skeleton_ai.action_task.cancel() 
       self.skeleton_ai.skeleton.machine.set_state('idle') 

       print('cancelled attack') 
       self.skeleton_ai.action_task = None 

    async def game_loop(self): 
     player_task = asyncio.ensure_future(self.handle_sending()) 
     skeleton_task = asyncio.ensure_future(self.skeleton_ai.run()) 
    def __init__(self): 
     self.task = None 
     self.loop = asyncio.get_event_loop() 
     self.player = Player(name='ply') 
     self.skeleton_ai = SkeletonAI(skeleton=Skeleton(name='bobby')) 
     self.skeleton_ai.skeleton.target = self.player 
     self.loop.run_until_complete(self.game_loop()) 
     try: 
      self.loop.run_forever() 
     finally: 
      pass 
     loop.close() 

Game() 

Hier ist, was ich zu tun versucht:

1. Spielereingang und Spiel ausgegeben werden async, so input() nicht blockiert. Das funktioniert.
2. Das Skelett bereitet einen Angriff vor, wenn der Angriff nicht innerhalb von 3 Sekunden unterbrochen wird, fügt der Angriff dem Spieler Schaden zu.
3. Der Spieler kann beliebigen Text eingeben, um den Skelettangriff zu unterbrechen.

Wie kann ich den Angriff des Skeletts machen? Ich möchte eine Aufgabe, die ich beliebig unterbrechen kann und später einen Callback aufrufen kann. Momentan bleibt einfach alles hängen. Die call_later ruft nie an.

Answer 1

Dies ist das Muster für eine asynchrone Funktion mit Timeout- und Callback-Funktion. Der Schlüssel ist, den asyncio.TimeoutError abzufangen und Ihre Zeitablauflogik auszuführen. Die abgebrochene Funktion wird nach der aktuellen Warteposition nicht fortgesetzt.

import asyncio 

async def slow_function(seconds): 
    print('starting slow computations') 
    await asyncio.sleep(seconds) 
    print('slow computations done') 

async def timeout_callback(): 
    print('timeout called') 

async def timeout_with_cb(fut, timeout, timeout_fut): 
    try: 
     await asyncio.wait_for(fut, timeout) 
    except asyncio.TimeoutError: 
     await timeout_fut 

loop = asyncio.get_event_loop() 
loop.run_until_complete(asyncio.ensure_future(
         timeout_with_cb(slow_function(2), 1, 
         timeout_callback()))) 

Dies drucken:

starting slow computations 
timeout called 

Ich denke, das Sie Ihr Beispiel anzupassen helfen kann (das bereitgestellte Beispiel nicht kompiliert).