Refactor FSMBase to subclass transitions.Machine

This commit is contained in:
Kristóf Tóth 2018-07-04 15:48:16 +02:00
parent 91c257554f
commit 7a92d88b73
2 changed files with 36 additions and 40 deletions

View File

@ -1,8 +1,6 @@
# Copyright (C) 2018 Avatao.com Innovative Learning Kft.
# All Rights Reserved. See LICENSE file for details.
from collections import defaultdict
from tfw import BroadcastingEventHandler
from tfw.config.logs import logging
@ -13,7 +11,6 @@ class FSMManagingEventHandler(BroadcastingEventHandler):
def __init__(self, key, fsm_type):
super().__init__(key)
self.fsm = fsm_type()
self.fsm_manager = FSMManager(self.fsm)
self._fsm_updater = FSMUpdater(self.fsm)
self.command_handlers = {
@ -30,7 +27,7 @@ class FSMManagingEventHandler(BroadcastingEventHandler):
LOG.error('IGNORING MESSAGE: Invalid message received: %s', message)
def handle_trigger(self, data):
self.fsm_manager.trigger(data['value'])
self.fsm.step(data['value'])
return self.with_fsm_update(data)
def with_fsm_update(self, data):
@ -43,35 +40,6 @@ class FSMManagingEventHandler(BroadcastingEventHandler):
return self.with_fsm_update(data)
class FSMManager:
def __init__(self, fsm):
self.fsm = fsm
self.trigger_predicates = defaultdict(list)
def trigger(self, trigger):
predicate_results = [
predicate()
for predicate in self.trigger_predicates[trigger]
]
# TODO: think about what could we do when this prevents triggering
if all(predicate_results):
try:
self.fsm.trigger(trigger)
except AttributeError:
LOG.debug('FSM failed to execute nonexistent trigger: "%s"', trigger)
def subscribe_predicate(self, trigger, *predicates):
self.trigger_predicates[trigger].extend(predicates)
def unsubscribe_predicate(self, trigger, *predicates):
self.trigger_predicates[trigger] = [
predicate
for predicate in self.trigger_predicates[trigger]
not in predicates
]
class FSMUpdater:
def __init__(self, fsm):
self.fsm = fsm
@ -80,7 +48,7 @@ class FSMUpdater:
state = self.fsm.state
valid_transitions = [
{'trigger': trigger}
for trigger in self.fsm.machine.get_triggers(self.fsm.state)
for trigger in self.fsm.get_triggers(self.fsm.state)
]
return {
'current_state': state,

View File

@ -1,14 +1,17 @@
# Copyright (C) 2018 Avatao.com Innovative Learning Kft.
# All Rights Reserved. See LICENSE file for details.
from typing import List
from collections import defaultdict
from transitions import Machine
from tfw.mixins import CallbackMixin
from tfw.config.logs import logging
LOG = logging.getLogger(__name__)
class FSMBase(CallbackMixin):
class FSMBase(Machine, CallbackMixin):
"""
A general FSM base class you can inherit from to track user progress.
See linear_fsm.py for an example use-case.
@ -18,10 +21,12 @@ class FSMBase(CallbackMixin):
"""
states, transitions = [], []
def __init__(self, initial: str = None, accepted_states: List[str] = None):
def __init__(self, initial=None, accepted_states=None):
self.accepted_states = accepted_states or [self.states[-1]]
self.machine = Machine(
model=self,
self.trigger_predicates = defaultdict(list)
Machine.__init__(
self,
states=self.states,
transitions=self.transitions,
initial=initial or self.states[0],
@ -34,4 +39,27 @@ class FSMBase(CallbackMixin):
self._execute_callbacks(event_data.kwargs)
def is_solved(self):
return self.state in self.accepted_states # pylint: disable=no-member
return self.state in self.accepted_states # pylint: disable=no-member
def subscribe_predicate(self, trigger, *predicates):
self.trigger_predicates[trigger].extend(predicates)
def unsubscribe_predicate(self, trigger, *predicates):
self.trigger_predicates[trigger] = [
predicate
for predicate in self.trigger_predicates[trigger]
not in predicates
]
def step(self, trigger):
predicate_results = [
predicate()
for predicate in self.trigger_predicates[trigger]
]
# TODO: think about what could we do when this prevents triggering
if all(predicate_results):
try:
self.trigger(trigger)
except AttributeError:
LOG.debug('FSM failed to execute nonexistent trigger: "%s"', trigger)