"""Classes to help in running and / or chaining long running statefull operations
This is useful if you want to implement long running processes (ie across many frames).
You can write these as simple generators and you don't have to worry about keeping
track of where you are in the sequence as this is done for you. The sequence can
include many steps and can branch and repeat etc.
The normal way to use these classes is to make your State a subclass of StateMachine.
This requires you to implement two methods to set up and render your on screen items.
There is nothing special going on here, it just keeps your logic away from the
statefull machinery.
1. When initialising the state will call a method initUI that you must implement.
You can do whatever you want in there but normally you would initialise any UI
components.
2. Calls drawUI(surface, scale) whenever you need to redraw the current frame.
The rest is where the magic happens,
You can start a long running process with a call to add_generator('<name>', some_generator)
Your generator will be called repeatedly until it exits. You can also stop it manually by
calling stop_generator() on the state, by calling stop() on the StateExecutor that
is returned by add_generator, or by raising a StopIteration in the generator.
Your generator can yield in two ways,
A - "yield": stops execution and returns to the same point next frame
B - "yield <N>: stops execution and returns to the same point <N> ms later
For an example of usage, let's say the player does something and you want to
move a card across the screen, flash it three times, then move it down the screen.
If the player clicks somewhere else in the meantime then the motion should stop.
The generator approach allows you to write this very logically and not worry about
keeping state (moving across, flashing, moving down etc).
In your State you would detect two conditions a) the motion should state, b) the
motion should be interrupted, and a generator method move_card()
class MyState(StateMachine):
def drawUI(self, surface, scale):
...
if condition_to_initiate_motion:
self.card_mover = self.add_generator('move-card', self.move_card(50, 100))
if condition_to_interrupt_card:
self.card_mover.stop()
...
def move_card(self, dx, dy):
# Move card over to the right
for x in range(dx):
self.card.x += 1
yield 100 # Pause for 100ms
# Flash the card three times
for repeat in range(dy):
self.card.visible = False
yield 100 # Hide for short time
self.card.visible = True
yield 900 # Show for a longer time
# Move the card down
for y in range(100):
self.card.y -= 1
yield 100 # Pause for 100ms
The above example shows how a single stateful operation can be written very cleanly.
Another use case is to chain states together.
Suppose you want to show cards shuffling, then being dealt and then being turned
over, one-by-one. You can chain these by calling add_generator at the end of each
operation.
class State(StateMachine):
def drawUI(...):
if condition_to_start_dealing:
self.add_generator('shuffle-cards', self.shuffle_cards())
def shuffle_cards(self):
... code to show shuffling of cards (runs over many frames, yielding as needed) ...
self.add_generator('deal-cards', self.deal_cards())
def deal_cards(self):
... code to move cards from deck to table (like move_card above) ...
self.add_generator('turn-cards', self.turn_cards)
def turn_cards(self):
for card in self.cards:
card.turn_over()
yield 1000
... add_generator for the next step in the process ...
You states can easily wait for conditions, eg if you detect a click on a card by
setting a property "card_selected" on the state.
def wait_for_card_selection(self):
while self.card_selected is None:
yield # We will wait in this state until a card is selected
if self.card_selected in deck:
self.add_generator('move-from-deck', self.move_card_from_deck(self.card_selected)
else:
self.add_generator('move-from-hand', self.move_card_from_hand(self.card_selected)
"""
import serge.common
import serge.actor
import pygame
[docs]class NotFound(Exception):
"""Generator was not found"""
[docs]class NextStep(StopIteration):
"""Move to the next step"""
def __init__(self, step):
"""Intialise the step"""
self.step = step
[docs]class StateExecutor(serge.common.Loggable):
"""Executes a generator through a sequence with delays"""
def __init__(self, name, generator, delay=0):
"""Initialise the executor"""
self.addLogger()
self.name = name
self.generator = generator
self.last_delay = self.delay = delay
self.done = False
self.verbose = False
self.paused = False
[docs] def update(self, dt):
"""Update the state"""
if not self.paused:
self.delay -= dt
if not self.done and self.delay < 0:
if self.verbose:
self.log.debug('{0} {1} doing action'.format(self.name, id(self)))
try:
try:
self.last_delay = self.delay = self.generator.send(dt)
except TypeError:
self.last_delay = self.delay = next(self.generator)
except NextStep, step:
self.generator = step.step
self.delay = 0
except StopIteration:
self.done = True
[docs] def next_step(self):
"""Immediately make the generator go to the next step"""
self.delay = 0
[docs] def update_interval(self, interval):
"""Update the interval we are waiting for
This behaves as though the last delay requested is
immediately changed to the specified value.
If we have already waited for the specified time
then immediately go to the next step
"""
time_elapsed = self.last_delay - self.delay
time_to_go = interval - time_elapsed
self.delay = time_to_go
[docs] def stop(self):
"""Stop this executor"""
self.done = True
[docs] def get_fraction_to_go(self):
"""Return the fraction of our time to go"""
if self.last_delay == 0:
return 1
else:
return max(0, min(1, self.delay / self.last_delay))
[docs]class StateMachine(serge.actor.Actor):
"""A state machine to use for handling the screen"""
verbose = True
def __init__(self, tag, name):
"""Initialise the machine"""
super(StateMachine, self).__init__(tag, name)
#
self.addLogger()
self.generators = []
self.state_clock = pygame.time.Clock()
self.delay = 0.0
self.verbose = True
self.dt = 0
[docs] def updateActor(self, interval, world):
"""Update the game state"""
self.dt = interval
self.state_clock.tick(interval)
self.delay -= self.state_clock.get_time()
#
# Process all states
for executor in list(self.generators):
executor.update(interval)
if executor.done:
self.generators.remove(executor)
[docs] def add_generator(self, name, generator):
"""Add a new generator to run"""
new_generator = StateExecutor(name, generator)
self.log.debug('Adding new executor {0}, {1}'.format(name, id(new_generator)))
self.generators.append(new_generator)
return new_generator
[docs] def stop_generator(self, name):
"""Stop a generator with a specific name"""
for generator in self.generators[:]:
if generator.name == name:
self.generators.remove(generator)
self.log.debug('Removing executor {0}, {1}'.format(name, id(generator)))
break
else:
raise NotFound('A generator named {0} was not found'.format(name))