# Copyright (C) 2009, Tutorius.org # Copyright (C) 2009, Erick Lavoie # # This program is free software; you can redistribute it and/or modify # it under the terms of the GNU General Public License as published by # the Free Software Foundation; either version 2 of the License, or # (at your option) any later version. # # This program is distributed in the hope that it will be useful, # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # GNU General Public License for more details. # # You should have received a copy of the GNU General Public License # along with this program; if not, write to the Free Software # Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA #TODO: For notification of modifications on the Tutorial check for GObject and PyDispatcher for inspiration from .constraints import ConstraintException from .properties import TPropContainer _NAME_SEPARATOR = "/" class Tutorial(object): """ This class replaces the previous Tutorial class and allows manipulation of the abstract representation of a tutorial as a state machine """ INIT = "INIT" END = "END" INITIAL_TRANSITION_NAME = INIT + "/transition0" def __init__(self, name, state_dict=None): """ The constructor for the Tutorial. By default, the tutorial contains only an initial state and an end state. The initial state doesn't contain any action but it contains a single automatic transition between the initial state and the end state . The end state doesn't contain any action nor transition. If state_dict is provided, a valid initial state and an end state must be provided. @param name The name of the tutorial @param state_dict optional, a valid dictionary of states @raise InvalidStateDictionary """ self.name = name # We will use an adjacency list representation through the # usage of state objects because our graph representation # is really sparse and mostly linear, for a brief # example of graph programming in python see: # http://www.python.org/doc/essays/graphs if not state_dict: self._state_dict = \ {Tutorial.INIT:State(name=Tutorial.INIT),\ Tutorial.END:State(name=Tutorial.END)} self.add_transition(Tutorial.INIT, \ (AutomaticTransitionEvent(), Tutorial.END)) else: self._state_dict = state_dict # Minimally check for the presence of an INIT and an END # state if not self._state_dict.has_key(Tutorial.INIT): raise Exception("No INIT state found in state_dict") if not self._state_dict.has_key(Tutorial.END): raise Exception("No END state found in state_dict") # TODO: Validate once validation is working #self.validate() # Initialize variables for generating unique names # TODO: We should take the max number from the # existing state names self._state_name_nb = 0 def add_state(self, action_list=(), transition_list=()): """ Add a new state to the state machine. The state is initialized with the action list and transition list and a new unique name is returned for this state. The actions are added using add_action. The transitions are added using add_transition. @param action_list The list of valid actions for this state @param transition_list The list of valid transitions @return unique name for this state """ name = self._generate_unique_state_name() for action in action_list: self._validate_action(action) for transition in transition_list: self._validate_transition(transition) state = State(name, action_list, transition_list) self._state_dict[name] = state return name def add_action(self, state_name, action): """ Add an action to a specific state. A name unique throughout the tutorial is generated to refer precisely to this action and is returned. The action is validated. @param state_name The name of the state to add an action to @param action The action to be added @return unique name for this action @raise LookupError if state_name doesn't exist """ if not self._state_dict.has_key(state_name): raise LookupError("Tutorial: state <" + state_name +\ "> is not defined") self._validate_action(action) return self._state_dict[state_name].add_action(action) def add_transition(self, state_name, transition): """ Add a transition to a specific state. A name unique throughout the tutorial is generated to refer precisely to this transition and is returned. Inserting a duplicate transition will raise an exception. The transition is validated. @param state_name The name of the state to add a transition to @param transition The transition to be added @return unique name for this action @raise LookupError if state_name doesn't exist @raise TransitionAlreadyExists """ if not self._state_dict.has_key(state_name): raise LookupError("Tutorial: state <" + state_name +\ "> is not defined") self._validate_transition(transition) # The unicity of the transition is validated by the state return self._state_dict[state_name].add_transition(transition) def update_action(self, action_name, new_properties): """ Update the action with action_name with a property dictionary new_properties. If one property update is invalid, the old values are restored and an exception is raised. @param action_name The name of the action to update @param new_properties The properties that will update the action @return old properties from the action @raise LookupError if action_name doesn't exist @raise ConstraintException if a property constraint is violated """ state_name = self._validate_state_name(action_name) #TODO: We should validate that only properties defined on the action # are passed in return self._state_dict[state_name].update_action(action_name, new_properties) def update_transition(self, transition_name, new_properties=None, new_state=None): """ Update the transition with transition_name with new properties and/or a new state to transition to. A None value means that the corresponding value won't be updated. If one property update is invalid, the old values are restored and an exception is raised. @param transition_name The name of the transition to replace @param new_properties The properties that will update the transition @param new_state The new state to transition to @return a tuple (old_properties, old_state) with previous values @raise LookupError if transition_name doesn't exist @raise ConstraintException if a property constraint is violated """ state_name = self._validate_state_name(transition_name) if not self._state_dict.has_key(state_name): raise LookupError("Tutorial: transition <" + transition_name +\ "> is not defined") if new_state and not self._state_dict.has_key(new_state): raise LookupError("Tutorial: destination state <" + new_state +\ "> is not defined") #TODO: We should validate that only properties defined on the action # are passed in return self._state_dict[state_name].update_transition(transition_name, new_properties, new_state) def delete_action(self, action_name): """ Delete the action identified by action_name. @param action_name The name of the action to be deleted @return the action that has been deleted @raise LookupError if transition_name doesn't exist """ state_name = self._validate_state_name(action_name) return self._state_dict[state_name].delete_action(action_name) def delete_transition(self, transition_name): """ Delete the transition identified by transition_name. @param transition_name The name of the transition to be deleted @return the transition that has been deleted @raise LookupError if transition_name doesn't exist """ state_name = self._validate_state_name(transition_name) return self._state_dict[state_name].delete_transition(transition_name) def delete_state(self, state_name): """ Delete the state, delete all the actions and transitions in this state, update the transitions from the state that pointed to this one to point to the next state and remove all the unreachable states recursively. All but the INIT and END states can be deleted. @param state_name The name of the state to remove @return The deleted state @raise StateDeletionError when trying to delete the INIT or the END state @raise LookupError if state_name doesn't exist """ self._validate_state_name(state_name) if state_name == Tutorial.INIT or state_name == Tutorial.END: raise StateDeletionError("<" + state_name + "> cannot be deleted") next_states = set(self.get_following_states_dict(state_name).values()) previous_states = set(self.get_previous_states_dict(state_name).values()) # For now tutorials should be completely linear, # let's make sure they are assert len(next_states) <= 1 and len(previous_states) <= 1 # Update transitions only if they existed if len(next_states) == 1 and len(previous_states) == 1: next_state = next_states.pop() previous_state = previous_states.pop() transitions = previous_state.get_transition_dict() for transition_name, (event, state_to_delete) in \ transitions.iteritems(): self.update_transition(transition_name, None, next_state.name) # Since we assume tutorials are linear for now, we do not need # to search for unreachable states return self._state_dict.pop(state_name) def get_action_dict(self, state_name=None): """ Returns a reference to the dictionary of all actions for a specific state. If no state_name is provided, returns an action dictionary containing actions for all states. @param state_name The name of the state to list actions from @return A dictionary of actions with action_name as key and action as value for state_name @raise LookupError if state_name doesn't exist """ if state_name and not self._state_dict.has_key(state_name): raise LookupError("Tutorial: state <" + state_name +\ "> is not defined") elif state_name: return self._state_dict[state_name].get_action_dict() else: action_dict = {} for state in self._state_dict.itervalues(): action_dict.update(state.get_action_dict()) return action_dict def get_transition_dict(self, state_name=None): """ Returns a dictionary of all actions for a specific state. If no state_name is provided, returns an action dictionary containing actions for all states. @param state_name The name of the state to list actions from @return A dictionary of transitions with transition_name as key and transition as value for state_name @raise LookupError if state_name doesn't exist """ if state_name and not self._state_dict.has_key(state_name): raise LookupError("Tutorial: state <" + state_name +\ "> is not defined") elif state_name: return self._state_dict[state_name].get_transition_dict() else: transition_dict = {} for state in self._state_dict.itervalues(): transition_dict.update(state.get_transition_dict()) return transition_dict def get_state_dict(self): """ Returns a reference to the internal state dictionary used by the Tutorial. @return A reference to the dictionary of all the states in the tutorial with state_name as key and state as value """ # Maybe we will need to change it for an immutable dictionary # to make sure the internal representation is not modified return self._state_dict def get_following_states_dict(self, state_name): """ Returns a dictionary of the states that are immediately reachable from a specific state. @param state_name The name of the state @raise LookupError if state_name doesn't exist """ if not self._state_dict.has_key(state_name): raise LookupError("Tutorial: state <" + state_name +\ "> is not defined") following_states_dict = {} for (event, next_state) in \ self._state_dict[state_name].get_transition_dict().itervalues(): following_states_dict[next_state] = self._state_dict[next_state] return following_states_dict def get_previous_states_dict(self, state_name): """ Returns a dictionary of the states that can transition to a specific state. @param state_name The name of the state @raise LookupError if state_name doesn't exist """ if not self._state_dict.has_key(state_name): raise LookupError("Tutorial: state <" + state_name +\ "> is not defined") previous_states_dict = {} for iter_state_name, state in \ self._state_dict.iteritems(): for (event, next_state) in \ self._state_dict[iter_state_name].get_transition_dict().itervalues(): if next_state != state_name: continue previous_states_dict[iter_state_name] = state # if we have found one, do not look for other transitions # from this state break return previous_states_dict # Convenience methods for common tutorial manipulations def add_state_before(self, state_name, action_list=[], event_list=[]): """ Add a new state just before another state state_name. All transitions going to state_name are updated to end on the new state and all events will be converted to transitions ending on state_name. When event_list is empty, an automatic transition to state_name will be added to maintain consistency. @param state_name The name of the state that will be preceded by the new state @param action_list The list of valid actions for this state @param event_list The list of events that will be converted to transitions to state_name @return unique name for this state @raise LookupError if state_name doesn't exist """ raise NotImplementedError # Callback mecanism to allow automatic change notification when # the tutorial is modified def register_action_added_cb(self, cb): """ Register a function cb that will be called when any action from the tutorial is added. cb should be of the form: cb(action_name, new_action) where: action_name is the unique name of the action that was added new_action is the new action @param cb The callback function to be called @raise InvalidCallbackFunction if the callback has less or more than 2 arguments """ raise NotImplementedError def register_action_updated_cb(self, cb): """ Register a function cb that will be called when any action from the tutorial is updated. cb should be of the form: cb(action_name, new_action) where: action_name is the unique name of the action that has changed new_action is the new action that replaces the old one @param cb The callback function to be called @raise InvalidCallbackFunction if the callback has less or more than 2 arguments """ raise NotImplementedError def register_action_deleted_cb(self, cb): """ Register a function cb that will be called when any action from the tutorial is deleted. cb should be of the form: cb(action_name, old_action) where: action_name is the unique name of the action that was deleted old_action is the new action that replaces the old one @param cb The callback function to be called @raise InvalidCallbackFunction if the callback has less or more than 2 arguments """ raise NotImplementedError def register_transition_updated_cb(self, cb): """ Register a function cb that will be called when any transition from the tutorial is updated. cb should be of the form: cb(transition_name, new_transition) where: transition_name is the unique name of the transition that has changed new_transition is the new transition that replaces the old one @param cb The callback function to be called @raise InvalidCallbackFunction if the callback has less or more than 2 arguments """ raise NotImplementedError # Validation to assert precondition def _validate_action(self, action): """ Validate that an action conforms to what we expect, throws an exception otherwise. @param action The action to validate @except InvalidAction if the action fails to conform to what we expect """ pass def _validate_transition(self, transition): """ Validate that a transition conforms to what we expect, throws an exception otherwise. @param transition The transition to validate @except InvalidTransition if the transition fails to conform to what we expect """ pass # Validation decorators to assert preconditions def _validate_state_name(self,name): """ Assert that the state name found in the first part of the string actually exists @param name The name that starts with a state name @return the state_name from name @raise LookupError if state_name doesn't exist """ state_name = name if name.find(_NAME_SEPARATOR) != -1: state_name = name[:name.find(_NAME_SEPARATOR)] if not self._state_dict.has_key(state_name): raise LookupError("Tutorial: state <" + str(state_name) +\ "> is not defined") return state_name def validate(self): """ Validate the state machine for a serie of properties: 1. No unreachable states 2. No dead end state (except END) 3. No branching in the main path 4. No loop in the main path 5. ... Throw an exception for the first condition that is not met. """ raise NotImplementedError def _generate_unique_state_name(self): name = "State" + str(self._state_name_nb) while name in self._state_dict: self._state_name_nb += 1 name = "State" + str(self._state_name_nb) return name # Python Magic Methods def __str__(self): """ Return a string representation of the tutorial """ return str(self._state_dict) def __eq__(self, other): return isinstance(other, type(self)) and self.get_state_dict() == other.get_state_dict() class State(object): """ This is a step in a tutorial. The state represents a collection of actions to undertake when entering the state, and a series of transitions to lead to next states. This class is not meant to be used explicitly as no validation is done on inputs, the validation should be done by the containing class. """ def __init__(self, name, actions={}, transitions={}): """ Initializes the content of the state, such as loading the actions that are required and building the correct transitions. @param actions list or dict of actions to perform when entering the state @param transitions list or dict of tuples of the form (event, next_state_name), that explains the outgoing links for this state For actions and transitions, dictionaries allow specifying the name. If lists are given, their contents will be added with add_action or add_transition """ object.__init__(self) self.name = name # Initialize internal variables for name generation self.action_name_nb = 0 self.transition_name_nb = 0 if type(actions) is dict: self._actions = dict(actions) else: self._actions = {} for action in actions: self.add_action(action) if type(transitions) is dict: self._transitions = dict(transitions) else: self._transitions = {} for transition in transitions: self.add_transition(transition) # Action manipulations def add_action(self, new_action): """ Adds an action to the state @param new_action The action to add @return a unique name for this action """ action_name = self._generate_unique_action_name(new_action) self._actions[action_name] = new_action return action_name def delete_action(self, action_name): """ Delete the action with the name action_name @param action_name The name of the action to delete @return The action deleted @raise LookupError if action_name doesn't exist """ if self._actions.has_key(action_name): return self._actions.pop(action_name) else: raise LookupError("Tutorial.State: action <" + action_name + "> is not defined") def update_action(self, action_name, new_properties): """ Update the action with action_name with a property dictionary new_properties. If one property update is invalid, the old values are restored and an exception is raised. @param action_name The name of the action to update @param new_properties The properties that will update the action @return The old properties from the action @raise LookupError if action_name doesn't exist @raise ConstraintException if a property constraint is violated """ if not self._actions.has_key(action_name): raise LookupError("Tutorial.State: action <" + action_name + "> is not defined") action = self._actions[action_name] old_properties = action.get_properties_dict_copy() try: for property_name, property_value in new_properties.iteritems(): action.__setattr__(property_name, property_value) return old_properties except ConstraintException, e: action._props = old_properties raise e def get_action_dict(self): """ Return the reference to the internal action dictionary. @return A dictionary of actions that the state will execute """ return self._actions def delete_actions(self): """ Removes all the action associated with this state. A cleared state will not do anything when entered or exited. """ self._actions = {} # Transition manipulations def add_transition(self, new_transition): """ Adds a transition from this state to another state. The same transition may not be added twice. @param transition The new transition. @return A unique name for the transition @raise TransitionAlreadyExists if an equivalent transition exists """ for transition in self._transitions.itervalues(): if transition == new_transition: raise TransitionAlreadyExists(str(transition)) transition_name = self._generate_unique_transition_name(new_transition) self._transitions[transition_name] = new_transition return transition_name def update_transition(self, transition_name, new_properties=None, new_state=None): """ Update the transition with transition_name with new properties and/or a new state to transition to. A None value means that the corresponding value won't be updated. If one property update is invalid, the old values are restored and an exception is raised. @param transition_name The name of the transition to replace @param new_properties The properties that will update the event on the transition @param new_state The new state to transition to @return a tuple (old_properties, old_state) with previous values @raise LookupError if transition_name doesn't exist @raise ConstraintException if a property constraint is violated """ if not self._transitions.has_key(transition_name): raise LookupError("Tutorial.State: transition <" + transition_name + "> is not defined") transition = self._transitions[transition_name] tmp_event = transition[0] tmp_state = transition[1] prop = new_properties or {} old_properties = transition[0].get_properties_dict_copy() old_state = transition[1] try: for property_name, property_value in prop.iteritems(): tmp_event.__setattr__(property_name, property_value) except ConstraintException, e: tmp_event._props = old_properties raise e if new_state: tmp_state = new_state self._transitions[transition_name] = (tmp_event, tmp_state) return (old_properties, old_state) def delete_transition(self, transition_name): """ Delete the transition with the name transition_name @param transition_name The name of the transition to delete @return The transition deleted @raise LookupError if transition_name doesn't exist """ if self._transitions.has_key(transition_name): return self._transitions.pop(transition_name) else: raise LookupError("Tutorial.State: transition <" + transition_name + "> is not defined") def get_transition_dict(self): """ Return the reference to the internal transition dictionary. @return The dictionary of transitions associated with this state. """ return self._transitions def delete_transitions(self): """ Delete all the transitions associated with this state. """ self._transitions = {} def _generate_unique_action_name(self, action): """ Returns a unique name for the action in this state, the actual content of the name should not be relied upon for correct behavior @param action The action to generate a name for @return A name garanteed to be unique within this state """ #TODO use the action class name to generate a name # to make it easier to debug and know what we are # manipulating name = self.name + _NAME_SEPARATOR + "action" + str(self.action_name_nb) while name in self._actions: self.action_name_nb += 1 name = self.name + _NAME_SEPARATOR + "action" + str(self.action_name_nb) return name def _generate_unique_transition_name(self, transition): """ Returns a unique name for the transition in this state, the actual content of the name should not be relied upon for correct behavior @param transition The transition to generate a name for @return A name garanteed to be unique within this state """ #TODO use the event class name from the transition to # generate a name to make it easier to debug and know # what we are manipulating name = self.name + _NAME_SEPARATOR + "transition" + str(self.transition_name_nb) while name in self._transitions: self.transition_name_nb += 1 name = self.name + _NAME_SEPARATOR + "transition" + str(self.transition_name_nb) return name def __eq__(self, otherState): """ Compare current state to otherState. Two states are considered equal if and only if: -every action in this state has a matching action in the other state with the same properties and values -every event filters in this state has a matching filter in the other state having the same properties and values -both states have the same name. @param otherState The state that will be compared to this one @return True if the states are the same, False otherwise ` """ return isinstance(otherState, type(self)) and \ self.get_action_dict() == otherState.get_action_dict() and \ self.get_transition_dict() == otherState.get_transition_dict() #TODO: Define the automatic transition in the same way as # other events class AutomaticTransitionEvent(TPropContainer): def __repr__(self): return str(self.__class__.__name__) ################## Error Handling and Exceptions ############################## class TransitionAlreadyExists(Exception): """ Raised when a duplicate transition is added to a state """ pass class InvalidStateDictionary(Exception): """ Raised when an initialization dictionary could not be used to initialize a tutorial """ pass class StateDeletionError(Exception): """ Raised when trying to delete an INIT or an END state from a tutorial """ pass