diff options
Diffstat (limited to 'tutorius/tutorial.py')
| -rw-r--r-- | tutorius/tutorial.py | 806 |
1 files changed, 806 insertions, 0 deletions
diff --git a/tutorius/tutorial.py b/tutorius/tutorial.py new file mode 100644 index 0000000..9831a7b --- /dev/null +++ b/tutorius/tutorial.py @@ -0,0 +1,806 @@ +# Copyright (C) 2009, Tutorius.org +# Copyright (C) 2009, Erick Lavoie <erick.lavoie@gmail.com> +# +# 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 <Tutorial.INITIAL_TRANSITION_NAME> + between the initial state <Tutorial.INIT> and the end state + <Tutorial.END>. + + 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: + raise NotImplementedError("Tutorial: Initilization from a dictionary is not supported yet") + + + # 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) + self._state_name_nb += 1 + return name + + def __str__(self): + """ + Return a string representation of the tutorial + """ + return str(self._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, action_list=(), transition_list=()): + """ + Initializes the content of the state, such as loading the actions + that are required and building the correct transitions. + + @param action_list The list of actions to execute when entering this + state + @param transition_list A list of tuples of the form + (event, next_state_name), that explains the outgoing links for + this state + """ + object.__init__(self) + + self.name = name + + # Initialize internal variables for name generation + self.action_name_nb = 0 + self.transition_name_nb = 0 + + self._actions = {} + for action in action_list: + self.add_action(action) + + self._transitions = {} + for transition in transition_list: + 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) + self.action_name_nb += 1 + 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) + self.transition_name_nb += 1 + 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 +` """ + raise NotImplementedError + +#TODO: Define the automatic transition in the same way as +# other events +class AutomaticTransitionEvent(TPropContainer): + pass + + +################## 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 |