Skip to content

Instantly share code, notes, and snippets.

@rgov
Created July 10, 2020 21:22
Show Gist options
  • Star 0 You must be signed in to star a gist
  • Fork 0 You must be signed in to fork a gist
  • Save rgov/e9e369b8167e8bac9a4c05e140136c0e to your computer and use it in GitHub Desktop.
Save rgov/e9e369b8167e8bac9a4c05e140136c0e to your computer and use it in GitHub Desktop.
#!/usr/bin/env python3
'''
Attempt to implement a statically type-checked state machine in Python. Doesn't work super well.
'''
import inspect
import typeguard
from typing import Callable, ForwardRef, Optional, Type, Union
class Event:
pass
class Machine:
def __init__(self):
self.state = None
def process_event(self, event: Event) -> None:
for _, method in inspect.getmembers(self.state, inspect.ismethod):
for trigger, predicate in getattr(method, 'triggers', []):
try:
typeguard.check_type('event', event, trigger)
except TypeError:
continue
if predicate(event):
self.state = method(event)
return
raise Exception(f'No state transition found for event {event}')
class State:
pass
def transition(predicate: Optional[Callable[[Event], bool]] = None):
predicate = predicate or (lambda _: True)
def decorator(func):
trigger = inspect.signature(func).parameters['event'].annotation
if isinstance(trigger, str):
trigger = ForwardRef(trigger)
func.__dict__.setdefault('triggers', []).append((trigger, predicate))
return func
return decorator
class Door(Machine):
def __init__(self):
self.state = Door.Open()
class EvOpen(Event):
pass
class EvClose(Event):
pass
class Open(State):
@transition()
def close(self, event: 'Door.EvClose') -> 'Door.Closed':
return Door.Closed()
class Closed(State):
@transition()
def open(self, event: Union['Door.EvOpen']) -> 'Door.Open':
return Door.Open()
door = Door()
door.process_event(Door.EvClose())
print(door.state)
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment