Coverage for src/rejuvenation/batch_process_examples.py: 94%

66 statements  

« prev     ^ index     » next       coverage.py v7.16.1, created at 2026-10-02 13:12 +0000

1# use clang to load and walk a compilation database 

2"""AI: Example script demonstrating batch AST processing across a Clang compilation database.""" 

3 

4import textwrap 

5from collections.abc import Callable, Iterable 

6from dataclasses import dataclass 

7 

8from renaissance.integrations.clang import ClangASTNode 

9from renaissance.integrations.clang.clang_json_ast_node import ClangJsonASTNode 

10from renaissance.recipes import CleanupRefactoring 

11from renaissance.syntax_tree import ( 

12 ASTFactory, 

13 ASTNode, 

14 ASTProcessor, 

15 BatchASTProcessor, 

16 TextUtils, 

17) 

18from renaissance.syntax_tree.recipe_ast_processor import ( 

19 RecipeASTProcessor, 

20 after_step, 

21 final_action, 

22 recipe_step, 

23) 

24from renaissance.syntax_tree.semantic_kind import SemanticKind 

25 

26example_1 = textwrap.dedent(""" 

27 void x(int a) {} 

28 void x1(int a) {} 

29 void x2(int a) {} 

30 

31 void f1(int a){ 

32 int unused = 0; 

33 int unused2 = 0; //must be removed 

34 if (a==1) { 

35 int unused = 0; 

36 int unused2 = 0; //should be kept 

37 int c = unused2; 

38 x1(c); 

39 } 

40 } 

41 """) 

42 

43example_2 = textwrap.dedent(""" 

44 void x(int a) {} 

45 void x1(int a) {} 

46 void x2(int a) {} 

47 void f2(int a){ 

48 int unused = 0; 

49 if (a==1) { 

50 int unused = 0; 

51 int another_unused = 0; 

52 int used2 = 0; //should be kept 

53 int c = used2; 

54 x2(c); 

55 } 

56 } 

57 """) 

58 

59 

60# generate a simple code base provider in real life use a compilation database 

61def simple_codebase_provider() -> Iterable[tuple[ASTFactory, ASTNode]]: 

62 """AI: Yield (factory, ATU) pairs for the two example C snippets, once per Clang integration.""" 

63 for impl_type in [ClangASTNode, ClangJsonASTNode]: 

64 factory = ASTFactory(impl_type) 

65 atu1 = factory.create_from_text(example_1, impl_type.__name__ + "1.c") 

66 yield factory, atu1 

67 atu2 = factory.create_from_text(example_2, impl_type.__name__ + "2.c") 

68 yield factory, atu2 

69 

70 

71def print_results(title, batch_processor): 

72 """AI: Print the given title followed by each in-memory file's path and rewritten content.""" 

73 print(title + ":") 

74 for file, code in batch_processor.in_memory_files.items(): 

75 print(TextUtils.shift_right(file, 4) + "\n") 

76 print(TextUtils.shift_right(code, 8) + "\n") 

77 

78 

79def batch_remove_unused_variable_once_example(): 

80 """Demonstrate a batch processing example using different AST node implementations. 

81 

82 It iterates over a list of AST node implementations (`ClangASTNode` and `ClangJsonASTNode`), 

83 and for each implementation, it generates a codebase provider that yields tuples of 

84 `ASTFactory` and `ASTNode` created from example source texts (`example_1` and `example_2`). 

85 The function then creates a `BatchASTProcessor` with in-memory storage enabled and processes 

86 the codebase using the `CleanupRefactoring.remove_unused_variables` refactoring operation. 

87 Finally, it prints the rewritten code stored in memory. 

88 """ 

89 # generate a batch processor for testing purposes we store into memory 

90 batch_processor = BatchASTProcessor(in_memory=True) 

91 batch_processor.once(simple_codebase_provider, CleanupRefactoring.remove_unused_variables) 

92 # print the rewritten code normally you would write to a file 

93 print_results("example batch remove unused variable once", batch_processor) 

94 

95 

96def batch_repeat_example(): 

97 """Demonstrates the use of a batch processor to perform multiple refactoring operations on a codebase. 

98 

99 This example creates an in-memory batch processor and applies two refactoring operations: 

100 1. CleanupRefactoring.remove_unused_variables: Removes unused variables from the codebase. 

101 2. remove_function: Removes all function calls from the codebase. 

102 The results of the refactoring operations are printed to the console. 

103 

104 Repeat is in action here: 

105 the first time the codebase is processed, the unused variables are removed. 

106 and the function calls are removed. 

107 the second time the codebase is processed, the new unused variables are removed again. 

108 

109 Note: 

110 In a real-world scenario, the rewritten code would typically be written to a file instead of being printed. 

111 

112 """ 

113 # generate a batch processor for testing purposes we store into memory 

114 batch_processor = BatchASTProcessor(in_memory=True) 

115 

116 # remove a function to create more unused variables 

117 def remove_function(ast_processor: ASTProcessor): 

118 [ast_processor.insert_before("// ", node, False, False) for node in ast_processor.find_semantic_kind(SemanticKind.CALL)] 

119 

120 # batch_processor.repeat(simple_codebase_provider, [remove_function]) 

121 batch_processor.repeat( 

122 simple_codebase_provider, 

123 [CleanupRefactoring.remove_unused_variables, remove_function], 

124 ) 

125 # print the rewritten code normally you would write to a file 

126 print_results("example batch repeat", batch_processor) 

127 

128 

129@dataclass 

130class CallInfo: 

131 """AI: Record the callee name and call-site text for one collected function call.""" 

132 

133 callee: str 

134 calls: str 

135 

136 

137class AnalysisRecipe: 

138 """AI: Recipe that collects function-call analysis results across the processed AST.""" 

139 

140 def __init__(self): 

141 """AI: Initialize an empty recipe for collecting function-call analysis results.""" 

142 self._calls = [] 

143 

144 @recipe_step(order=0) 

145 def store_function_call(self, ast_processor: ASTProcessor) -> Callable[[], None] | None: 

146 """AI: Collect all function-call nodes found by the processor and queue them for single-threaded storage.""" 

147 # find all function calls and store them, this routing is invoked in parallel! 

148 calls: list[CallInfo] = [] 

149 [AnalysisRecipe._add_function_call(node, calls) for node in ast_processor.find_semantic_kind(SemanticKind.CALL)] 

150 # the resulting lambda is invoked single threaded 

151 # this kind of mechanism is mainly used to store results from multiple processors 

152 # for refactoring operations this is not needed as a refactoring operation is single threaded 

153 if calls: 

154 return lambda: self._calls.extend(calls) 

155 return None 

156 

157 @after_step("store_function_call") 

158 def just_show_the_method(self): 

159 """AI: Print a marker showing this hook ran after store_function_call.""" 

160 print("called after store_function_call") 

161 

162 @final_action() 

163 def final_action(self): 

164 """AI: Print all collected function calls after the recipe finishes.""" 

165 print("Calls:") 

166 for call in self._calls: 

167 print(" " + call.callee + " -- calls --> " + call.calls) 

168 

169 @staticmethod 

170 def _add_function_call(call: ASTNode, calls: list[CallInfo]): 

171 callee = call.get_ancestor("(?i)Function_?Decl") 

172 if callee: 

173 calls.append(CallInfo(callee.name, call.children[0].name)) 

174 

175 

176def batch_recipe_example(): 

177 """AI: Run the analysis recipe over the example codebase and print discovered calls.""" 

178 print("example batch analysis using recipe:\n") 

179 recipe_ast_processor = RecipeASTProcessor(AnalysisRecipe(), simple_codebase_provider, r".*", in_memory=True) 

180 recipe_ast_processor.run() 

181 

182 

183if __name__ == "__main__": 

184 # a list of example to show batch processing of a code base 

185 batch_remove_unused_variable_once_example() 

186 batch_repeat_example() 

187 batch_recipe_example()