From 52c83b1aab72721a42134a19e5db3acc6c235a39 Mon Sep 17 00:00:00 2001 From: Simon Poirier Date: Thu, 05 Nov 2009 13:13:43 +0000 Subject: half overlayer/half creator --- (limited to 'src/tutorius/core.py') diff --git a/src/tutorius/core.py b/src/tutorius/core.py new file mode 100644 index 0000000..bfbe07b --- /dev/null +++ b/src/tutorius/core.py @@ -0,0 +1,618 @@ +# Copyright (C) 2009, Tutorius.org +# Copyright (C) 2009, Vincent Vinet +# +# 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 +""" +Core + +This module contains the core classes for tutorius + +""" + +import logging +import os + +from .TProbe import ProbeManager +from .dbustools import save_args +from . import addon + +logger = logging.getLogger("tutorius") + +class Tutorial (object): + """ + Tutorial Class, used to run through the FSM. + """ + #Properties + probeManager = property(lambda self: self._probeMgr) + activityId = property(lambda self: self._activity_id) + + def __init__(self, name, fsm, filename=None): + """ + Creates an unattached tutorial. + """ + object.__init__(self) + self.name = name + self.activity_init_state_filename = filename + + self.state_machine = fsm + self.state_machine.set_tutorial(self) + + self.state = None + + self.handlers = [] + self._probeMgr = ProbeManager() + self._activity_id = None + #Rest of initialisation happens when attached + + def attach(self, activity_id): + """ + Attach to a running activity + + @param activity_id the id of the activity to attach to + """ + #For now, absolutely detach if a previous one! + if self._activity_id: + self.detach() + self._activity_id = activity_id + self._probeMgr.attach(activity_id) + self._probeMgr.currentActivity = activity_id + self._prepare_activity() + self.state_machine.set_state("INIT") + + def detach(self): + """ + Detach from the current activity + """ + + # Uninstall the whole FSM + self.state_machine.teardown() + + if not self._activity_id is None: + self._probeMgr.detach(self._activity_id) + self._activity_id = None + + def set_state(self, name): + """ + Switch to a new state + """ + logger.debug("==== NEW STATE: %s ====" % name) + + self.state_machine.set_state(name) + + def _prepare_activity(self): + """ + Prepare the activity for the tutorial by loading the saved state and + emitting gtk signals + """ + #Load the saved activity if any + if self.activity_init_state_filename is not None: + #For now the file will be saved in the data folder + #of the activity root directory + filename = os.getenv("SUGAR_ACTIVITY_ROOT") + "/data/" +\ + self.activity_init_state_filename + readfile = addon.create("ReadFile", filename=filename) + if readfile: + self._probeMgr.install(readfile) + #Uninstall now while we have the reference handy + self._probeMgr.uninstall(readfile) + +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 event filters + with associated actions that point to a possible next state. + """ + + def __init__(self, name="", action_list=None, event_filter_list=None, tutorial=None): + """ + Initializes the content of the state, like loading the actions + that are required and building the correct tests. + + @param action_list The list of actions to execute when entering this + state + @param event_filter_list A list of tuples of the form + (event_filter, next_state_name), that explains the outgoing links for + this state + @param tutorial The higher level container of the state + """ + object.__init__(self) + + self.name = name + + self._actions = action_list or [] + + self._transitions= dict(event_filter_list or []) + + self._installedEvents = set() + + self.tutorial = tutorial + + def set_tutorial(self, tutorial): + """ + Associates this state with a tutorial. A tutorial must be set prior + to executing anything in the state. The reason for this is that the + states need to have access to the activity (via the tutorial) in order + to properly register their callbacks on the activities' widgets. + + @param tutorial The tutorial that this state runs under. + """ + if self.tutorial == None : + self.tutorial = tutorial + else: + raise RuntimeWarning(\ + "The state %s was already associated with a tutorial." % self.name) + + def setup(self): + """ + Install the state itself, by first registering the event filters + and then triggering the actions. + """ + for (event, next_state) in self._transitions.items(): + self._installedEvents.add(self.tutorial.probeManager.subscribe(event, save_args(self._event_filter_state_done_cb, next_state ))) + + for action in self._actions: + self.tutorial.probeManager.install(action) + + def teardown(self): + """ + Uninstall all the event filters that were active in this state. + Also undo every action that was installed for this state. This means + removing dialogs that were displayed, removing highlights, etc... + """ + # Remove the handlers for the all of the state's event filters + while len(self._installedEvents) > 0: + self.tutorial.probeManager.unsubscribe(self._installedEvents.pop()) + + # Undo all the actions related to this state + for action in self._actions: + self.tutorial.probeManager.uninstall(action) + + def _event_filter_state_done_cb(self, next_state, event): + """ + Callback for event filters. This function needs to inform the + tutorial that the state is over and tell it what is the next state. + + @param next_state The next state for the transition + @param event The event that occured + """ + # Run the tests here, if need be + + # Warn the higher level that we wish to change state + self.tutorial.set_state(next_state) + + # Model manipulation + # These functions are used to simplify the creation of states + def add_action(self, new_action): + """ + Adds an action to the state + + @param new_action The new action to execute when in this state + @return True if added, False otherwise + """ + self._actions.append(new_action) + return True + + # remove_action - We did not define names for the action, hence they're + # pretty hard to remove on a precise basis + + def get_action_list(self): + """ + @return A list of actions that the state will execute + """ + return self._actions + + def clear_actions(self): + """ + Removes all the action associated with this state. A cleared state will + not do anything when entered or exited. + """ + #FIXME What if the action is currently installed? + self._actions = [] + + def add_event_filter(self, event, next_state): + """ + Adds an event filter that will cause a transition from this state. + + The same event filter may not be added twice. + + @param event The event that will trigger a transition + @param next_state The state to which the transition will lead + @return True if added, False otherwise + """ + if event not in self._transitions.keys(): + self._transitions[event]=next_state + return True + return False + + def get_event_filter_list(self): + """ + @return The list of event filters associated with this state. + """ + return self._transitions.items() + + def clear_event_filters(self): + """ + Removes all the event filters associated with this state. A state that + was just cleared will become a sink and will be the end of the + tutorial. + """ + self._transitions = {} + + def __eq__(self, otherState): + """ + Compares two states and tells whether they contain the same states with the + same actions and event filters. + + @param otherState The other State that we wish to match + @returns True if every action in this state has a matching action in the + other state with the same properties and values AND if every + event filters in this state has a matching filter in the + other state having the same properties and values AND if both + states have the same name. +` """ + if not isinstance(otherState, State): + return False + if self.name != otherState.name: + return False + + # Do they have the same actions? + if len(self._actions) != len(otherState._actions): + return False + + if len(self._transitions) != len(otherState._transitions): + return False + + for act in self._actions: + found = False + # For each action in the other state, try to match it with this one. + for otherAct in otherState._actions: + if act == otherAct: + found = True + break + if found == False: + # If we arrive here, then we could not find an action with the + # same values in the other state. We know they're not identical + return False + + # Do they have the same event filters? + if self._transitions != otherState._transitions: + return False + + # If nothing failed up to now, then every actions and every filters can + # be found in the other state + return True + +class FiniteStateMachine(State): + """ + This is a collection of states, with a start state and an end callback. + It is used to simplify the development of the various tutorials by + encapsulating a collection of states that represent a given learning + process. + + For now, we will consider that there can only be states + inserted in the FSM, and that there are no nested FSM inside. + """ + + def __init__(self, name, tutorial=None, state_dict=None, start_state_name="INIT", action_list=None): + """ + The constructor for a FSM. Pass in the start state and the setup + actions that need to be taken when the FSM itself start (which may be + different from what is done in the first state of the machine). + + @param name A short descriptive name for this FSM + @param tutorial The tutorial that will execute this FSM. If None is + attached on creation, then one must absolutely be attached before + executing the FSM with set_tutorial(). + @param state_dict A dictionary containing the state names as keys and + the state themselves as entries. + @param start_state_name The name of the starting state, if different + from "INIT" + @param action_list The actions to undertake when initializing the FSM + """ + State.__init__(self, name) + + self.name = name + self.tutorial = tutorial + + # Dictionnary of states contained in the FSM + self._states = state_dict or {} + + self.start_state_name = start_state_name + # Set the current state to None - we are not executing anything yet + self.current_state = None + + # Register the actions for the FSM - They will be processed at the + # FSM level, meaning that when the FSM will start, it will first + # execute those actions. When the FSM closes, it will tear down the + # inner actions of the state, then close its own actions + self.actions = action_list or [] + + # Flag to mention that the FSM was initialized + self._fsm_setup_done = False + # Flag that must be raised when the FSM is to be teared down + self._fsm_teardown_done = False + # Flag used to declare that the FSM has reached an end state + self._fsm_has_finished = False + + def set_tutorial(self, tutorial): + """ + This associates the FSM to the given tutorial. It MUST be associated + either in the constructor or with this function prior to executing the + FSM. + + @param tutorial The tutorial that will execute this FSM. + """ + # If there was no tutorial associated + if self.tutorial == None: + # Associate it with this FSM and all the underlying states + self.tutorial = tutorial + for state in self._states.itervalues(): + state.set_tutorial(tutorial) + else: + raise RuntimeWarning(\ + "The FSM %s is already associated with a tutorial."%self.name) + + def setup(self): + """ + This function initializes the FSM the first time it is called. + Then, every time it is called, it initializes the current state. + """ + # Are we associated with a tutorial? + if self.tutorial == None: + raise UnboundLocalError("No tutorial was associated with FSM %s" % self.name) + + # If we never initialized the FSM itself, then we need to run all the + # actions associated with the FSM. + if self._fsm_setup_done == False: + # Remember the initial state - we might want to reset + # or rewind the FSM at a later moment + self.start_state = self._states[self.start_state_name] + self.current_state = self.start_state + # Flag the FSM level setup as done + self._fsm_setup_done = True + # Execute all the FSM level actions + for action in self.actions: + self.tutorial.probeManager.install(action) + + # Then, we need to run the setup of the current state + self.current_state.setup() + + def set_state(self, new_state_name): + """ + This functions changes the current state of the finite state machine. + + @param new_state The identifier of the state we need to go to + """ + # TODO : Since we assume no nested FSMs, we don't set state on the + # inner States / FSMs +## # Pass in the name to the internal state - it might be a FSM and +## # this name will apply to it +## self.current_state.set_state(new_state_name) + + # Make sure the given state is owned by the FSM + if not self._states.has_key(new_state_name): + # If we did not recognize the name, then we do not possess any + # state by that name - we must ignore this state change request as + # it will be done elsewhere in the hierarchy (or it's just bogus). + return + + if self.current_state != None: + if new_state_name == self.current_state.name: + # If we already are in this state, we do not need to change + # anything in the current state - By design, a state may not point + # to itself + return + + new_state = self._states[new_state_name] + + # Undo the actions of the old state + self.teardown() + + # Insert the new state + self.current_state = new_state + + # Call the initial actions in the new state + self.setup() + + def get_current_state_name(self): + """ + Returns the name of the current state. + + @return A string representing the name of the current state + """ + return self.current_state.name + + def teardown(self): + """ + Revert any changes done by setup() + """ + # Teardown the current state + if self.current_state is not None: + self.current_state.teardown() + + # If we just finished the whole FSM, we need to also call the teardown + # on the FSM level actions + if self._fsm_has_finished == True: + # Flag the FSM teardown as not needed anymore + self._fsm_teardown_done = True + # Undo all the FSM level actions here + for action in self.actions: + self.tutorial.probeManager.uninstall(action) + + # TODO : It might be nice to have a start() and stop() method for the + # FSM. + + # Data manipulation section + # These functions are dedicated to the building and editing of a graph. + def add_state(self, new_state): + """ + Inserts a new state in the FSM. + + @param new_state The State object that will now be part of the FSM + @raise KeyError In the case where a state with this name already exists + """ + if self._states.has_key(new_state.name): + raise KeyError("There is already a state by this name in the FSM") + + self._states[new_state.name] = new_state + + # Not such a great name for the state accessor... We already have a + # set_state name, so get_state would conflict with the notion of current + # state - I would recommend having a set_current_state instead. + def get_state_by_name(self, state_name): + """ + Fetches a state from the FSM, based on its name. If there is no + such state, the method will throw a KeyError. + + @param state_name The name of the desired state + @return The State object having the given name + """ + return self._states[state_name] + + def remove_state(self, state_name): + """ + Removes a state from the FSM. Raises a KeyError when the state is + not existent. + + Warning : removing a state will also remove all the event filters that + point to this given name, to preserve the FSM's integrity. If you only + want to edit a state, you would be better off fetching this state with + get_state_by_name(). + + @param state_name A string being the name of the state to remove + @raise KeyError When the state_name does not a represent a real state + stored in the dictionary + """ + + state_to_remove = self._states[state_name] + + # Remove the state from the states' dictionnary + for st in self._states.itervalues(): + # Iterate through the list of event filters and remove those + # that point to the state that will be removed + + #TODO : Move this code inside the State itself - we're breaking + # encap :P + for event in st._transitions: + if st._transitions[event] == state_name: + del st._transitions[event] + + # Remove the state from the dictionary + del self._states[state_name] + + # Exploration methods - used to know more about a given state + def get_following_states(self, state_name): + """ + Returns a tuple of the names of the states that point to the given + state. If there is no such state, the function raises a KeyError. + + @param state_name The name of the state to analyse + @raise KeyError When there is no state by this name in the FSM + """ + state = self._states[state_name] + + next_states = set() + + for event, state in state._transitions.items(): + next_states.add(state) + + return tuple(next_states) + + def get_previous_states(self, state_name): + """ + Returns a tuple of the names of the state that can transition to + the given state. If there is no such state, the function raises a + KeyError. + + @param state_name The name of the state that the returned states might + transition to. + """ + # This might seem a bit funny, but we don't verify if the given + # state is present or not in the dictionary. + # This is due to the fact that when building a graph, we might have a + # prototypal state that has not been inserted yet. We could not know + # which states are pointing to it until we insert it in the graph. + + states = [] + # Walk through the list of states + for st in self._states.itervalues(): + for event, state in st._transitions.items(): + if state == state_name: + states.append(state) + continue + + return tuple(states) + + # Convenience methods to see the content of a FSM + def __str__(self): + out_string = "" + for st in self._states.itervalues(): + out_string += st.name + ", " + return out_string + + def __eq__(self, otherFSM): + """ + Compares the elements of two FSM to ensure and returns true if they have the + same set of states, containing the same actions and the same event filters. + + @returns True if the two FSMs have the same content, False otherwise + """ + if not isinstance(otherFSM, FiniteStateMachine): + return False + + # Make sure they share the same name + if not (self.name == otherFSM.name) or \ + not (self.start_state_name == otherFSM.start_state_name): + return False + + # Ensure they have the same number of FSM-level actions + if len(self._actions) != len(otherFSM._actions): + return False + + # Test that we have all the same FSM level actions + for act in self._actions: + found = False + # For every action in the other FSM, try to match it with the + # current one. + for otherAct in otherFSM._actions: + if act == otherAct: + found = True + break + if found == False: + return False + + # Make sure we have the same number of states in both FSMs + if len(self._states) != len(otherFSM._states): + return False + + # For each state, try to find a corresponding state in the other FSM + for state_name in self._states.keys(): + state = self._states[state_name] + other_state = None + try: + # Attempt to use this key in the other FSM. If it's not present + # the dictionary will throw an exception and we'll know we have + # at least one different state in the other FSM + other_state = otherFSM._states[state_name] + except: + return False + # If two states with the same name exist, then we want to make sure + # they are also identical + if not state == other_state: + return False + + # If we made it here, then all the states in this FSM could be matched to an + # identical state in the other FSM. + return True -- cgit v0.9.1