State Machine with Coroutine `send()`

From Generators to Coroutines: The yield Expression

So far, we've used yield as a statement to produce values from a generator. The key insight that unlocks coroutines is that yield can also be an expression that receives values.

Consider this line:
received_value = yield

When the generator is paused at this line, an external caller can use the .send(data) method. The data sent by the caller becomes the result of the yield expression and is assigned to received_value. This creates a two-way communication channel.

The execution flow has a critical first step: priming. A new generator hasn't reached a yield expression yet, so you can't send it data. You must first call next() on it to advance it to its first yield point.

Let's visualize this flow.

This mechanism is the foundation for using generators to model stateful processes.

Modeling a Finite State Machine (FSM)

Now, let's apply this to a classic finite state machine. Imagine a simple digital turnstile. It has two states: LOCKED and UNLOCKED.

• In the LOCKED state, if you insert a coin, it becomes UNLOCKED. If you push it, it stays LOCKED.
• In the UNLOCKED state, if you push it, it lets you through and becomes LOCKED. If you insert another coin, it stays UNLOCKED.

A traditional implementation might look like this:

# The "old" way with explicit state variable and conditionals
def turnstile_machine():
    state = 'LOCKED'
    while True:
        event = yield state # yield the current state, wait for an event
        if state == 'LOCKED':
            if event == 'coin':
                print("-> UNLOCKED")
                state = 'UNLOCKED'
            elif event == 'push':
                print("-> CLANG! (Still LOCKED)")
        elif state == 'UNLOCKED':
            if event == 'push':
                print("-> Welcome! (Now LOCKED)")
                state = 'LOCKED'
            elif event == 'coin':
                print("-> Thank you! (Still UNLOCKED)")




# --- Driver code ---
ts = turnstile_machine()
current_state = next(ts)  # Prime it and get initial state
print(f"Initial State: {current_state}")

current_state = ts.send('push') # Stays locked
print(f"State: {current_state}")

current_state = ts.send('coin') # Unlocks
print(f"State: {current_state}")

current_state = ts.send('push') # Locks again
print(f"State: {current_state}")

This works, but the logic is tangled in if/elif blocks. As the number of states and events grows, this becomes unmanageable.

The Coroutine Approach: States as Functions

A much cleaner approach is to model each state as its own coroutine (or a loop within a single coroutine). This localizes the logic for each state.

Let's see an elegant implementation structure.

This structure is powerful. Each state is a self-contained unit of logic. It defines its own transitions and what happens on valid or invalid input. The StopIteration that occurs when a state's generator ends is used as a signal for an invalid transition.

Deeper Dive: Schedulers and Advanced Coroutines

The taxi_sim.py example from the Fluent Python chapter is an advanced application of this concept. Each taxi_process is a state machine moving through states like 'leave garage', 'pick up passenger', and 'drop off passenger'. The Simulator class acts as a scheduler, driving multiple taxi coroutines concurrently by sending them time updates. This is the conceptual basis for modern asynchronous frameworks like asyncio.

Exercise: A Stateful Traffic Light

Your task is to model a traffic light as a state machine using a single generator coroutine.

Specification:

1. The coroutine, traffic_light(), should model a traffic light that cycles through GREEN, YELLOW, and RED.
2. It should accept two events sent to it: 'tick' (time passes) and 'emergency' (an emergency vehicle approaches).
3. The coroutine should yield its current color as a string.
4. The cycle logic:
   • Starts GREEN. Stays GREEN for 3 'tick's.
   • Transitions to YELLOW. Stays YELLOW for 1 'tick'.
   • Transitions to RED. Stays RED for 2 'tick's.
   • Transitions back to GREEN and repeats.
5. If it receives an 'emergency' event at any time, it should immediately transition to RED, yield 'RED', and then continue its normal cycle from the RED state.

Starter Code:

def traffic_light():



    # Your implementation here
    # Use an infinite loop and nested loops for each state.
    # `event = yield color` will be your core mechanic.
    pass # remove this line




# --- Driver Code (Do not change) ---
if __name__ == "__main__":
    light_coro = traffic_light()
    



    # Prime the coroutine and print its initial state
    current_color = next(light_coro)
    print(f"Light is {current_color}")




    # A list of events to send to the traffic light
    events = ['tick', 'tick', 'tick', 'tick', 'tick', 'tick', 'emergency', 'tick', 'tick']
    
    for event in events:
        print(f"\nSending: '{event}'")
        current_color = light_coro.send(event)
        print(f"Light is {current_color}")

def traffic_light():
    """A state machine for a traffic light, implemented as a coroutine."""
    while True:



        # --- GREEN STATE ---
        color = 'GREEN'
        for _ in range(3):
            event = yield color
            if event == 'emergency':
                break  # Exit to the emergency handler below
        else: # This `else` belongs to the `for` loop



            # --- YELLOW STATE ---
            color = 'YELLOW'
            event = yield color
            if event == 'emergency':
                pass # Will be handled by the outer emergency check




        # --- RED STATE / EMERGENCY HANDLER ---
        color = 'RED'
        for _ in range(2):
            event = yield color



            # An emergency during RED just resets its timer
            if event == 'emergency':
                continue

Conclusion

You have now learned how to use one of Python's most powerful and expressive features. By leveraging generator coroutines, you can model complex, stateful logic in a way that is clean, readable, and avoids the tangled mess of nested conditionals.

Key Takeaways:

• Coroutines are State Machines: Generators become coroutines when you use yield as an expression to receive data via .send(). This makes them ideal for implementing state machines.
• State is Encapsulated: The state of the process is maintained in the coroutine's local variables between yield points, eliminating the need for instance variables or global state managers.
• Logic is Sequential: Stateful logic, which is often spread across complex conditional blocks, can be written as a straightforward, sequential script inside the coroutine.
• Priming is Essential: A coroutine must be advanced to its first yield with next() before you can .send() it any data.

In our next module, we will shift gears from abstract control flow patterns to the practical realities of I/O. Our first lesson will cover how to use pathlib.Path for modern, object-oriented, and cross-platform path manipulation, replacing the older, string-based os.path module.