1# -*- coding: utf-8 -*- 

2 

3# -- stdlib -- 

4from typing import Callable, Generic, List, Tuple, TypeVar, Union 

5import logging 

6import sys 

7import zlib 

8 

9# -- third party -- 

10# -- own -- 

11 

12# -- code -- 

13log = logging.getLogger('utils.events') 

14T = TypeVar('T') 

15 

16 

17class EventHub(Generic[T]): 

18 __slots__ = ('_subscribers', 'name') 

19 

20 class StopPropagation: 

21 __slots__ = () 

22 

23 STOP_PROPAGATION = StopPropagation() 

24 

25 _subscribers: List[Tuple[float, Callable[[T], Union[T, StopPropagation]]]] 

26 

27 def __init__(self): 

28 self._subscribers = [] 

29 self.name: str = '[Anonymous]' 

30 

31 def subscribe(self, cb: Callable[[T], Union[T, StopPropagation]], prio: float): 

32 self._subscribers.append((prio, cb)) 

33 self._subscribers.sort(key=lambda v: v[0]) 

34 return self 

35 

36 def __iadd__(self, cb: Callable[[T], Union[T, StopPropagation]]): 

37 # deterministic priority 

38 f = sys._getframe(1) 

39 s = '{}:{}'.format(f.f_code.co_filename, f.f_lineno).encode('utf-8') 

40 prio = zlib.crc32(s) * 1.0 / 0x100000000 

41 self.subscribe(cb, prio) 

42 return self 

43 

44 def emit(self, ev: T): 

45 log.debug('Handling event %s %s', self.name, ev) 

46 if not self._subscribers: 

47 log.debug('Emitting event %s when no subscribers present!', self.name) 

48 return 

49 

50 for prio, cb in self._subscribers: 

51 r = cb(ev) 

52 

53 if isinstance(r, EventHub.StopPropagation): 

54 return None 

55 else: 

56 ev = r 

57 

58 if ev is None: 58 ↛ 59line 58 didn't jump to line 59, because the condition on line 58 was never true

59 raise Exception("Returning None in EventHub, last callback: %s", cb) 

60 

61 return ev 

62 

63 

64class FSM(object): 

65 __slots__ = ('_context', '_valid', '_state', '_callback') 

66 

67 def __init__(self, context, valid, initial, callback): 

68 if initial not in valid: 68 ↛ 69line 68 didn't jump to line 69, because the condition on line 68 was never true

69 raise Exception('State not in valid choices!') 

70 

71 self._context = context 

72 self._valid = valid 

73 self._state = initial 

74 self._callback = callback 

75 

76 def transit(self, state): 

77 if state not in self._valid: 77 ↛ 78line 77 didn't jump to line 78, because the condition on line 77 was never true

78 raise Exception('Invalid state transition!') 

79 

80 if state == self._state: 

81 return 

82 

83 old, self._state = self._state, state 

84 self._callback(self._context, old, state) 

85 

86 def __eq__(self, other): 

87 return self._state == other 

88 

89 def __repr__(self): 

90 return f'FSM:<{self._state}>' 

91 

92 @property 

93 def state(self) -> str: 

94 return self._state 

95 

96 @staticmethod 

97 def to_evhub(evhub): 

98 return lambda ctx, f, t: evhub.emit((ctx, f, t))