Coverage for src/rejuvenation/batch_process_examples.py: 94%
66 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-10-15 10:34 +0000
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
7# use clang to load and walk a compilation database
8"""AI: Example script demonstrating batch AST processing across a Clang compilation database."""
10import textwrap
11from collections.abc import Callable, Iterable
12from dataclasses import dataclass
14from renaissance.integrations.clang import ClangASTNode
15from renaissance.integrations.clang.clang_json_ast_node import ClangJsonASTNode
16from renaissance.recipes import CleanupRefactoring
17from renaissance.syntax_tree import (
18 ASTFactory,
19 ASTNode,
20 ASTProcessor,
21 BatchASTProcessor,
22 TextUtils,
23)
24from renaissance.syntax_tree.recipe_ast_processor import (
25 RecipeASTProcessor,
26 after_step,
27 final_action,
28 recipe_step,
29)
30from renaissance.syntax_tree.semantic_kind import SemanticKind
32example_1 = textwrap.dedent("""
33 void x(int a) {}
34 void x1(int a) {}
35 void x2(int a) {}
37 void f1(int a){
38 int unused = 0;
39 int unused2 = 0; //must be removed
40 if (a==1) {
41 int unused = 0;
42 int unused2 = 0; //should be kept
43 int c = unused2;
44 x1(c);
45 }
46 }
47 """)
49example_2 = textwrap.dedent("""
50 void x(int a) {}
51 void x1(int a) {}
52 void x2(int a) {}
53 void f2(int a){
54 int unused = 0;
55 if (a==1) {
56 int unused = 0;
57 int another_unused = 0;
58 int used2 = 0; //should be kept
59 int c = used2;
60 x2(c);
61 }
62 }
63 """)
66# generate a simple code base provider in real life use a compilation database
67def simple_codebase_provider() -> Iterable[tuple[ASTFactory, ASTNode]]:
68 """AI: Yield (factory, ATU) pairs for the two example C snippets, once per Clang integration."""
69 for impl_type in [ClangASTNode, ClangJsonASTNode]:
70 factory = ASTFactory(impl_type)
71 atu1 = factory.create_from_text(example_1, impl_type.__name__ + "1.c")
72 yield factory, atu1
73 atu2 = factory.create_from_text(example_2, impl_type.__name__ + "2.c")
74 yield factory, atu2
77def print_results(title, batch_processor) -> None:
78 """AI: Print the given title followed by each in-memory file's path and rewritten content."""
79 print(title + ":")
80 for file, code in batch_processor.in_memory_files.items():
81 print(TextUtils.shift_right(file, 4) + "\n")
82 print(TextUtils.shift_right(code, 8) + "\n")
85def batch_remove_unused_variable_once_example() -> None:
86 """Demonstrate a batch processing example using different AST node implementations.
88 It iterates over a list of AST node implementations (`ClangASTNode` and `ClangJsonASTNode`),
89 and for each implementation, it generates a codebase provider that yields tuples of
90 `ASTFactory` and `ASTNode` created from example source texts (`example_1` and `example_2`).
91 The function then creates a `BatchASTProcessor` with in-memory storage enabled and processes
92 the codebase using the `CleanupRefactoring.remove_unused_variables` refactoring operation.
93 Finally, it prints the rewritten code stored in memory.
94 """
95 # generate a batch processor for testing purposes we store into memory
96 batch_processor = BatchASTProcessor(in_memory=True)
97 batch_processor.once(simple_codebase_provider, CleanupRefactoring.remove_unused_variables)
98 # print the rewritten code normally you would write to a file
99 print_results("example batch remove unused variable once", batch_processor)
102def batch_repeat_example() -> None:
103 """Demonstrates the use of a batch processor to perform multiple refactoring operations on a codebase.
105 This example creates an in-memory batch processor and applies two refactoring operations:
106 1. CleanupRefactoring.remove_unused_variables: Removes unused variables from the codebase.
107 2. remove_function: Removes all function calls from the codebase.
108 The results of the refactoring operations are printed to the console.
110 Repeat is in action here:
111 the first time the codebase is processed, the unused variables are removed.
112 and the function calls are removed.
113 the second time the codebase is processed, the new unused variables are removed again.
115 Note:
116 In a real-world scenario, the rewritten code would typically be written to a file instead of being printed.
118 """
119 # generate a batch processor for testing purposes we store into memory
120 batch_processor = BatchASTProcessor(in_memory=True)
122 # remove a function to create more unused variables
123 def remove_function(ast_processor: ASTProcessor) -> None:
124 [ast_processor.insert_before("// ", node, False, False) for node in ast_processor.find_semantic_kind(SemanticKind.CALL)]
126 # batch_processor.repeat(simple_codebase_provider, [remove_function])
127 batch_processor.repeat(
128 simple_codebase_provider,
129 [CleanupRefactoring.remove_unused_variables, remove_function],
130 )
131 # print the rewritten code normally you would write to a file
132 print_results("example batch repeat", batch_processor)
135@dataclass
136class CallInfo:
137 """AI: Record the callee name and call-site text for one collected function call."""
139 callee: str
140 calls: str
143class AnalysisRecipe:
144 """AI: Recipe that collects function-call analysis results across the processed AST."""
146 def __init__(self) -> None:
147 """AI: Initialize an empty recipe for collecting function-call analysis results."""
148 self._calls = []
150 @recipe_step(order=0)
151 def store_function_call(self, ast_processor: ASTProcessor) -> Callable[[], None] | None:
152 """AI: Collect all function-call nodes found by the processor and queue them for single-threaded storage."""
153 # find all function calls and store them, this routing is invoked in parallel!
154 calls: list[CallInfo] = []
155 [AnalysisRecipe._add_function_call(node, calls) for node in ast_processor.find_semantic_kind(SemanticKind.CALL)]
156 # the resulting lambda is invoked single threaded
157 # this kind of mechanism is mainly used to store results from multiple processors
158 # for refactoring operations this is not needed as a refactoring operation is single threaded
159 if calls:
160 return lambda: self._calls.extend(calls)
161 return None
163 @after_step("store_function_call")
164 def just_show_the_method(self) -> None:
165 """AI: Print a marker showing this hook ran after store_function_call."""
166 print("called after store_function_call")
168 @final_action()
169 def final_action(self) -> None:
170 """AI: Print all collected function calls after the recipe finishes."""
171 print("Calls:")
172 for call in self._calls:
173 print(" " + call.callee + " -- calls --> " + call.calls)
175 @staticmethod
176 def _add_function_call(call: ASTNode, calls: list[CallInfo]) -> None:
177 callee = call.get_ancestor("(?i)Function_?Decl")
178 if callee:
179 calls.append(CallInfo(callee.name, call.children[0].name))
182def batch_recipe_example() -> None:
183 """AI: Run the analysis recipe over the example codebase and print discovered calls."""
184 print("example batch analysis using recipe:\n")
185 recipe_ast_processor = RecipeASTProcessor(AnalysisRecipe(), simple_codebase_provider, r".*", in_memory=True)
186 recipe_ast_processor.run()
189if __name__ == "__main__":
190 # a list of example to show batch processing of a code base
191 batch_remove_unused_variable_once_example()
192 batch_repeat_example()
193 batch_recipe_example()