Coverage for omforme/omforme.py: 85.42%

1"""Provide the class Omforme for transformations on a generator. 

2 

3Use case example: 

4 

5from omforme.omforme import Omforme 

6 

7Given a stream of lines e.g. in gen 

8 

9gen = (x for x in ('a', 'b', 'c', 'c', 'c', 'd', '1', '2', '3', 'e')) 

10 

11with defining a playbook of: 

12 

13playbook = (('b', 'ignore', lambda: None), ('d', 'collect', list()), ('e', 'return', lambda: None),) 

14 

15when calling 

16 

17transform = Omforme(playbook)(gen) 

18 

19then transform should become: 

20 

21( 

22 ('b', 'ignore', <function <lambda> at 0x...>), 

23 ('d', 'collect', ['1', '2', '3']), 

24 ('e', 'return', <function <lambda> at 0x...>) 

25) 

26 

27""" 

28 

29import argparse 

30from typing import no_type_check 

31 

32from omforme import DEBUG, ENCODING, ENCODING_ERRORS_POLICY, log 

33 

34 

35class Omforme: 

36 """Provide a generator borrowing consumer that returns the transform applying playbook to the stream. 

37 

38 The playbook is a tuple of (trigger, action, function) triplets. 

39 

40 Actions are epected to be within ('ignore', 'collect', 'return') 

41 

42 Example: 

43 

44 >>> # from omforme.omforme import Omforme 

45 >>> gen = (x for x in ('a', 'b', 'c', 'c', 'c', 'd', '1', '2', '3', 'e')) 

46 >>> playbook = (('b', 'ignore', lambda: None), ('d', 'collect', list()), ('e', 'return', lambda: None),) 

47 >>> transform = Omforme(playbook)(gen) # doctest: +ELLIPSIS 

48 0 3 a b ignore <function <lambda> at 0x...> d 

49 0 3 b b ignore <function <lambda> at 0x...> d 

50 0 3 c b ignore <function <lambda> at 0x...> d 

51 0 3 c b ignore <function <lambda> at 0x...> d 

52 0 3 c b ignore <function <lambda> at 0x...> d 

53 1 3 1 d collect ['1'] e 

54 1 3 2 d collect ['1', '2'] e 

55 1 3 3 d collect ['1', '2', '3'] e 

56 1 3 e d collect ['1', '2', '3'] e 

57 >>> transform[1] 

58 ('d', 'collect', ['1', '2', '3']) 

59 >>> 

60 

61 """ 

62 

63 @no_type_check 

64 def __init__(self, playbook): 

65 """Later alligator.""" 

66 self.playbook = playbook 

67 

68 @no_type_check 

69 def __call__(self, gen): 

70 """Apply transform to generator stream of events.""" 

71 phase = 0 

72 trigger, what, where_to = self.playbook[phase] 

73 peek = self.playbook[phase + 1][0] 

74 stop = len(self.playbook) 

75 for data in gen: 

76 if data == peek: 

77 if phase + 2 < stop: 

78 phase += 1 

79 trigger, what, where_to = self.playbook[phase] 

80 peek = self.playbook[phase + 1][0] 

81 continue 

82 if what == 'return': 82 ↛ 83line 82 didn't jump to line 83, because the condition on line 82 was never true

83 break 

84 if what == 'collect' and data != peek: 

85 where_to.append(data) 

86 print(phase, stop, data, trigger, what, where_to, peek) 

87 

88 return self.playbook 

89 

90 

91def main(options: argparse.Namespace) -> int: 

92 log.info(f'{DEBUG=}, {ENCODING=}, {ENCODING_ERRORS_POLICY=}') 

93 return 0 

94 

95 

96if __name__ == '__main__': 96 ↛ 97line 96 didn't jump to line 97, because the condition on line 96 was never true

97 import doctest 

98 

99 doctest.testmod(verbose=True, optionflags=doctest.ELLIPSIS)