PyPI 官网下载 | fysom-2.0.1.tar.gz

# coding=utf-8 # # fysom - pYthOn Finite State Machine - this is a port of Jake # Gordon's javascript-state-machine to python # https://github.com/jakesgordon/javascript-state-machine # # Copyright (C) 2011 Mansour Behabadi <mansour@oxplot.com>, Jake Gordon # and other contributors # # Permission is hereby granted, free of charge, to any person obtaining # a copy of this software and associated documentation files (the # "Software"), to deal in the Software without restriction, including # without limitation the rights to use, copy, modify, merge, publish, # distribute, sublicense, and/or sell copies of the Software, and to # permit persons to whom the Software is furnished to do so, subject to # the following conditions: # # The above copyright notice and this permission notice shall be # included in all copies or substantial portions of the Software. # # THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, # EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF # MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. # IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY # CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, # TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE # SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. # __author__ = 'Mansour Behabadi' __copyright__ = 'Copyright 2011, Mansour Behabadi and Jake Gordon' __credits__ = ['Mansour Behabadi', 'Jake Gordon'] __license__ = 'MIT' __version__ = '2.0.1' __maintainer__ = 'Mansour Behabadi' __email__ = 'mansour@oxplot.com' import collections WILDCARD = '*' class FysomError(Exception): ''' Raised whenever an unexpected event gets triggered. ''' pass class Canceled(FysomError): ''' Raised when an event is canceled due to the onbeforeevent handler returning False ''' class Fysom(object): ''' Wraps the complete finite state machine operations. ''' def __init__(self, cfg={}, initial=None, events=None, callbacks=None, final=None): ''' Construct a Finite State Machine. Arguments: cfg finite state machine specification, a dictionary with keys 'initial', 'events', 'callbacks', 'final' initial initial state events a list of dictionaries (keys: 'name', 'src', 'dst') or a list tuples (event name, source state or states, destination state or states) callbacks a dictionary mapping callback names to functions final a state of the FSM where its is_finished() method returns True Named arguments override configuration dictionary. Example: >>> fsm = Fysom(events=[('tic', 'a', 'b'), ('toc', 'b', 'a')], initial='a') >>> fsm.current 'a' >>> fsm.tic() >>> fsm.current 'b' >>> fsm.toc() >>> fsm.current 'a' ''' cfg = dict(cfg) # override cfg with named arguments if "events" not in cfg: cfg["events"] = [] if "callbacks" not in cfg: cfg["callbacks"] = {} if initial: cfg["initial"] = initial if final: cfg["final"] = final if events: cfg["events"].extend(list(events)) if callbacks: cfg["callbacks"].update(dict(callbacks)) # convert 3-tuples in the event specification to dicts events_dicts = [] for e in cfg["events"]: if isinstance(e, collections.Mapping): events_dicts.append(e) elif hasattr(e, "__iter__"): name, src, dst = list(e)[:3] events_dicts.append({"name": name, "src": src, "dst": dst}) cfg["events"] = events_dicts self._apply(cfg) def isstate(self, state): ''' Returns if the given state is the current state. ''' return self.current == state def can(self, event): ''' Returns if the given event be fired in the current machine state. ''' return (event in self._map and ((self.current in self._map[event]) or WILDCARD in self._map[event]) and not hasattr(self, 'transition')) def cannot(self, event): ''' Returns if the given event cannot be fired in the current state. ''' return not self.can(event) def is_finished(self): ''' Returns if the state machine is in its final state. ''' return self._final and (self.current == self._final) def _apply(self, cfg): ''' Does the heavy lifting of machine construction. More notably: >> Sets up the initial and finals states. >> Sets the event methods and callbacks into the same object namespace. >> Prepares the event to state transitions map. ''' init = cfg['initial'] if 'initial' in cfg else None if self._is_base_string(init): init = {'state': init} self._final = cfg['final'] if 'final' in cfg else None events = cfg['events'] if 'events' in cfg else [] callbacks = cfg['callbacks'] if 'callbacks' in cfg else {} tmap = {} self._map = tmap def add(e): ''' Adds the event into the machine map. ''' if 'src' in e: src = [e['src']] if self._is_base_string( e['src']) else e['src'] else: src = [WILDCARD] if e['name'] not in tmap: tmap[e['name']] = {} for s in src: tmap[e['name']][s] = e['dst'] # Consider initial state as any other state that can have transition from none to # initial value on occurance of startup / init event ( if specified). if init: if 'event' not in init: init['event'] = 'startup' add({'name': init['event'], 'src': 'none', 'dst': init['state']}) for e in events: add(e) # For all the events as present in machine map, construct the event handler. for name in tmap: setattr(self, name, self._build_event(name)) # For all the callbacks, register them into the current object namespace. for name in callbacks: setattr(self, name, callbacks[name]) self.current = 'none' # If initialization need not be deferred, trigger the event for transition to initial state. if init and 'defer' not in init: getattr(self, init['event'])() def _build_event(self, event): ''' For every event in the state machine, prepares the event handler and registers the same into current object namespace. ''' def fn(*args, **kwargs): if hasattr(self, 'transition'): raise FysomError( "event %s inappropriate because previous transition did not complete" % event) # Check if this event can be triggered in the current state. if not self.can(event): raise FysomError( "event %s inappropriate in current state %s" % (event, self.current)) # On event occurence, source will always be the current state. src = self.current # Finds the destination state, after this event is completed. dst = ((src in self._map[event] and self._map[event][src]) or WILDCARD in self._map[event] and self._map[event][WILDCARD]) # Prepares the object with all the meta data to be passed to callbacks. class _e_obj(object): pass e = _e_obj()