# Coverage for utils/events.py : 92%

## 61 statements 58 run 3 missing 2 excluded 3 partial

Hot-keys on this page

r m x p toggle line displays

j k next/prev highlighted chunk

0 (zero) top of page

1 (one) first highlighted chunk

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

2

3# -- stdlib --

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

5 import logging

6 import sys

7 import zlib

8

9# -- third party --

10# -- own --

11

12# -- code --

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

14 T = TypeVar('T')

15

16

17 class 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 ↛ 59 line 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

64 class FSM(object):

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

66

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

68 if initial not in valid: 68 ↛ 69 line 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 ↛ 78 line 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))