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

1# Copyright (C) 2024-2026 TNO 

2# 

3# Licensed under the terms specified in the LICENSE file in the root directory 

4# of this repository, also available at: 

5# https://github.com/TNO/Renaissance.Py/blob/main/LICENSE 

6 

7"""AI: Decorator support for annotating recipe actions on AST processors.""" 

8 

9import functools 

10from collections.abc import Callable, Sequence 

11from typing import Any, TypeVar 

12 

13from .ast_processor import ASTProcessor 

14from .batch_ast_processor import BatchASTProcessor, IterableProvider 

15 

16TRecipe = TypeVar("TRecipe") 

17TFunc = Callable[..., Any] 

18 

19 

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

21 """AI: Wrap foreign_decorator so the decorated function also records its own decorator and recipe-action name.""" 

22 

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

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

25 r.decorator = new_decorator # keep track of decorator 

26 r.recipe_action = name 

27 return r 

28 

29 new_decorator.__name__ = foreign_decorator.__name__ 

30 new_decorator.__doc__ = foreign_decorator.__doc__ 

31 return new_decorator 

32 

33 

34def get_methods_with_decorator(cls: type, decorator: TFunc): 

35 """AI: Yield all methods of cls that were annotated with the given recipe-action decorator.""" 

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

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

38 yield maybe_decorated 

39 

40 

41# Decorators 

42 

43 

44def final_action() -> TFunc: 

45 """AI: Decorator marking a recipe method as a final action, run once after all steps complete.""" 

46 

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

48 @functools.wraps(func) 

49 def final_action_wrapper(recipe: TFunc) -> None: 

50 func(recipe) 

51 

52 return final_action_wrapper 

53 

54 return annotate_decorator(final_action_decorator, final_action.__name__) 

55 

56 

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

58 """AI: Decorator marking a recipe method as a numbered step, optionally repeatable, in the recipe's workflow.""" 

59 

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

61 @functools.wraps(func) 

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

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

64 result = func(recipe, ast_processor) 

65 

66 def callable_result(): 

67 if result: 

68 result() 

69 return func.__name__ 

70 

71 return callable_result() 

72 return None 

73 

74 return recipe_step_wrapper 

75 

76 return annotate_decorator(recipe_step_decorator, recipe_step.__name__) 

77 

78 

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

80 """AI: Decorator marking a recipe method to run only after the named step has already executed.""" 

81 

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

83 @functools.wraps(func) 

84 def after_step_wrapper(preceding_methods: Sequence[str], recipe: TFunc) -> None: 

85 if step in preceding_methods: 

86 func(recipe) 

87 

88 return after_step_wrapper 

89 

90 return annotate_decorator(after_step_decorator, after_step.__name__) 

91 

92 

93class RecipeASTProcessor[TRecipe]: 

94 """AI: Processor that runs a recipe's steps across a batch of files, optionally in parallel.""" 

95 

96 def __init__( 

97 self, 

98 recipe: TRecipe, 

99 iterable_provider: IterableProvider, 

100 file_filter: str, 

101 in_memory: bool = False, 

102 max_processes: int = 4, 

103 ) -> None: 

104 """AI: Prepare a processor that runs a recipe's steps across a batch of files, optionally in parallel.""" 

105 self.__recipe: TRecipe = recipe 

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

107 self.__iterableProvider = iterable_provider 

108 self.__file_filter = file_filter 

109 

110 def run(self) -> None: 

111 """AI: Run each of the recipe's decorated steps across the batch of files and return the collected results.""" 

112 actions: list[TFunc] = [] 

113 results: list[Any] = [] 

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

115 results.append(None) 

116 

117 def recipe_action(ast_processor: ASTProcessor) -> None: 

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

119 if result: 

120 results[idx] = result 

121 

122 actions.append(recipe_action) 

123 

124 after_step_actions: list[TFunc] = [] 

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

126 

127 def after_step_action() -> None: 

128 after_step_method(results, self.__recipe) 

129 

130 after_step_actions.append(after_step_action) 

131 

132 step = 0 

133 while len(actions) > 0: 

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

135 results[idx] = None 

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

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

138 break 

139 for after_step_action in after_step_actions: 

140 after_step_action() 

141 step += 1 

142 

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

144 method(self.__recipe)