Coverage for src / renaissance / syntax_tree / recipe_ast_processor.py: 100%

1import functools 

2from collections.abc import Callable, Sequence 

3from typing import Any, TypeVar 

4 

5from .ast_processor import ASTProcessor 

6from .batch_ast_processor import BatchASTProcessor, IterableProvider 

7 

8TRecipe = TypeVar("TRecipe") 

9TFunc = Callable[..., Any] 

10 

11 

12def annotate_decorator(foreign_decorator: TFunc, name: str): 

13 def new_decorator(func: TFunc) -> TFunc: 

14 r = foreign_decorator(func) # apply foreignDecorator, like call to foreignDecorator(method) would have done 

15 r.decorator = new_decorator # keep track of decorator 

16 r.recipe_action = name 

17 return r 

18 

19 new_decorator.__name__ = foreign_decorator.__name__ 

20 new_decorator.__doc__ = foreign_decorator.__doc__ 

21 return new_decorator 

22 

23 

24def get_methods_with_decorator(cls: Any, decorator: TFunc): 

25 for maybe_decorated in cls.__dict__.values(): 

26 if hasattr(maybe_decorated, "recipe_action") and maybe_decorated.recipe_action == decorator.__name__: 

27 yield maybe_decorated 

28 

29 

30# Decorators 

31 

32 

33def final_action() -> TFunc: 

34 def final_action_decorator(func: TFunc) -> TFunc: 

35 @functools.wraps(func) 

36 def final_action_wrapper(recipe: TFunc): 

37 func(recipe) 

38 

39 return final_action_wrapper 

40 

41 return annotate_decorator(final_action_decorator, final_action.__name__) 

42 

43 

44def recipe_step(order: int = 0, repeat: bool = False) -> TFunc: 

45 def recipe_step_decorator(func: TFunc) -> TFunc: 

46 @functools.wraps(func) 

47 def recipe_step_wrapper(step: int, recipe: TFunc, ast_processor: ASTProcessor): 

48 if step == order and (repeat or ast_processor.repeat_step == 0): 

49 result = func(recipe, ast_processor) 

50 

51 def callable_result(): 

52 if result: 

53 result() 

54 return func.__name__ 

55 

56 return callable_result() 

57 return None 

58 

59 return recipe_step_wrapper 

60 

61 return annotate_decorator(recipe_step_decorator, recipe_step.__name__) 

62 

63 

64def after_step(step: str) -> TFunc: 

65 def after_step_decorator(func: TFunc) -> TFunc: 

66 @functools.wraps(func) 

67 def after_step_wrapper(preceding_methods: Sequence[str], recipe: TFunc): 

68 if step in preceding_methods: 

69 func(recipe) 

70 

71 return after_step_wrapper 

72 

73 return annotate_decorator(after_step_decorator, after_step.__name__) 

74 

75 

76class RecipeASTProcessor[TRecipe]: 

77 def __init__( 

78 self, 

79 recipe: TRecipe, 

80 iterable_provider: IterableProvider, 

81 file_filter: str, 

82 in_memory: bool = False, 

83 max_processes: int = 4, 

84 ): 

85 self.__recipe: TRecipe = recipe 

86 self.__batch_processor = BatchASTProcessor(in_memory=in_memory, max_processes=max_processes) 

87 self.__iterableProvider = iterable_provider 

88 self.__file_filter = file_filter 

89 

90 def run(self): 

91 actions: list[TFunc] = [] 

92 results: list[Any] = [] 

93 for idx, recipe_step_method in enumerate(get_methods_with_decorator(type(self.__recipe), recipe_step)): 

94 results.append(None) 

95 

96 def recipe_action(ast_processor: ASTProcessor): 

97 result = recipe_step_method(step, self.__recipe, ast_processor) 

98 if result: 

99 results[idx] = result 

100 

101 actions.append(recipe_action) 

102 

103 after_step_actions: list[TFunc] = [] 

104 for after_step_method in get_methods_with_decorator(self.__recipe.__class__, after_step): 

105 

106 def after_step_action(): 

107 after_step_method(results, self.__recipe) 

108 

109 after_step_actions.append(after_step_action) 

110 

111 step = 0 

112 while len(actions) > 0: 

113 for idx in range(len(results)): 

114 results[idx] = None 

115 self.__batch_processor.repeat(self.__iterableProvider, actions, self.__file_filter) 

116 if all(result is None for result in results): 

117 break 

118 for after_step_action in after_step_actions: 

119 after_step_action() 

120 step += 1 

121 

122 for method in get_methods_with_decorator(self.__recipe.__class__, final_action): 

123 method(self.__recipe)