From 2bda2c8c50797a6214126edc5afdc3fc5d8e9969 Mon Sep 17 00:00:00 2001 From: Vincent Vinet Date: Sun, 06 Dec 2009 19:43:16 +0000 Subject: improve coverage report and cleanup old stuff --- (limited to 'tutorius/core.py') diff --git a/tutorius/core.py b/tutorius/core.py deleted file mode 100644 index bfbe07b..0000000 --- a/tutorius/core.py +++ /dev/null @@ -1,618 +0,0 @@ -# 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