fetapi/statemachinetest.py

215 lines
4.5 KiB
Python
Raw Normal View History

2015-05-15 23:38:44 +02:00
import threading
2015-05-18 14:33:26 +02:00
import Queue as queue
2015-05-15 23:38:44 +02:00
class IllegalEventError(Exception):
pass
class AbstractState(object):
def __init__(self, controller):
pass
2015-05-18 14:42:54 +02:00
def on_registration_in_progress(self):
raise IllegalEventError()
def on_registration_successful(self):
raise IllegalEventError()
2015-05-15 23:38:44 +02:00
def on_gabelschalter_up(self):
raise IllegalEventError()
def on_gabelschalter_down(self):
raise IllegalEventError()
def on_incoming_call(self):
raise IllegalEventError()
def on_call_ended(self):
raise IllegalEventError()
def on_call_accepted(self):
raise IllegalEventError()
def on_call_ringing(self):
raise IllegalEventError()
def on_nummernschalter_input(self, num):
raise IllegalEventError()
def on_timeout(self):
raise IllegalEventError()
2015-05-18 14:42:54 +02:00
class InitState(AbstractState):
def on_registration_in_progress(self):
print('registration in progress')
2015-05-18 15:31:47 +02:00
return RegisteringState
2015-05-15 23:38:44 +02:00
2015-05-18 15:31:47 +02:00
class RegisteringState(AbstractState):
2015-05-18 14:42:54 +02:00
def on_registration_successful(self):
print('registration successfull')
return IdleState
2015-05-15 23:38:44 +02:00
class IdleState(AbstractState):
def on_incoming_call(self):
print('incomfing call')
def on_gabelschalter_up(self):
print('gabel up')
2015-05-18 15:42:55 +02:00
return DialingState
2015-05-15 23:38:44 +02:00
def on_gabelschalter_down(self):
pass
def on_nummernschalter_input(self, x):
pass
class SchelltState(AbstractState):
def on_gabelschalter_up(self):
return AcceptingState
def on_call_ended(self):
return IdleState
class AcceptingState(AbstractState):
def on_call_accepted(self):
return CallRunningState
class CallTerminatingState(AbstractState):
def on_call_ended(self):
return IdleState
def on_call_accepted(self):
return None
class ForgottenState(AbstractState):
def on_gabelschalter_down(self):
return IdleState
class BusyBeepingState(AbstractState):
def on_timeout(self):
return ForgottenState
def on_gabelschalter_down(self):
return IdleState
class CallRunningState(AbstractState):
def on_gabelschalter_down(self):
return CallTerminatingState
def on_call_ended(self):
return BusyBeepingState
class WecktState(AbstractState):
def on_gabelschalter_down(self):
return CallTerminatingState
def on_call_ended(self):
return BusyBeepingState
def on_call_accepted(self):
return CallRunningState
class ConnectingState(AbstractState):
def on_gabelschalter_down(self):
return CallTerminatingState
def on_call_ringing(self):
return WecktState
2015-05-18 15:42:55 +02:00
class DialingState(AbstractState):
2015-05-15 23:38:44 +02:00
def on_gabelschalter_down(self):
return IdleState
def on_nummernschalter_input(self, num):
print('nummernschalter: %d' % (num))
def on_timeout(self):
return ConnectingState
class TelefonapparatUserInterface(object):
def __init__(self):
pass
def add_gabelschalter_callback(self, cb):
pass
def add_nummernschalter_callback(self, cb):
pass
def set_wecker(self, enabled):
pass
def set_schauzeichen(self, enabled):
pass
class StateMachineController(object):
def __init__(self):
2015-05-18 15:31:47 +02:00
self.__state = InitState(self)
2015-05-15 23:38:44 +02:00
self.__running = True
self.__evqueue = queue.Queue()
self.__evthread = threading.Thread(target=self.__event_dispatcher)
self.__evthread.start()
def __event_dispatcher(self):
while self.__running:
(evname, evargs, evkwargs) = self.__evqueue.get()
if not evname:
return
print('!!! event: %s' % (evname))
handler = getattr(self.__state, 'on_%s' % (evname))
try:
newstate = handler(*evargs, **evkwargs)
except IllegalEventError:
print('illegal event occured!!!!!!!!!!!!!!!!!!!!')
if not newstate:
continue
oldstate = self.__state.__class__
print('%s -> %s' % (oldstate.__name__, newstate.__name__))
self.__state = newstate(self)
def queue_event(self, evname, *evargs, **evkwargs):
if not hasattr(AbstractState, 'on_%s' % (evname)):
raise ValueError('Illegal event name: %s' % (evname))
self.__evqueue.put((evname, evargs, evkwargs))
def stop(self, hard=False):
if hard:
self.__running = False
self.__evqueue.put((None, None, None))
if __name__ == '__main__':
c = StateMachineController()
2015-05-18 15:31:47 +02:00
c.queue_event('registration_in_progress')
c.queue_event('registration_successful')
2015-05-15 23:38:44 +02:00
c.queue_event('gabelschalter_up')
c.queue_event('nummernschalter_input', 4)
c.queue_event('nummernschalter_input', 2)
#c.queue_event('gabelschalter_down')
#c.queue_event('call_accepted')
c.queue_event('timeout')
c.queue_event('call_ringing')
#c.queue_event('gabelschalter_down')
c.queue_event('call_accepted')
c.queue_event('call_ended')
c.queue_event('timeout')
c.queue_event('gabelschalter_down')
c.stop()