Coverage for src/renaissance/syntax_tree/batch_ast_processor.py: 98%
55 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"""AI: Process multiple ASTs in parallel by applying a shared action to each."""
9import concurrent.futures
10import re
11from collections.abc import Callable, Iterable, Sequence
12from functools import partial
13from typing import Any
15from .ast_factory import ASTFactory
16from .ast_node import ASTNode
17from .ast_processor import ASTProcessor
19AST_FACTORY_AND_ATU = tuple[ASTFactory, ASTNode]
20Action = Callable[[ASTProcessor], Callable[[], Any] | None]
21IterableProvider = Callable[[], Iterable[AST_FACTORY_AND_ATU]]
24class BatchASTProcessor:
25 """AI: Process multiple ASTs in parallel by applying a shared action to each."""
27 def __init__(self, in_memory: bool = False, max_processes: int = 4) -> None:
28 """Initialize the BatchASTProcessor.
30 Args:
31 in_memory (bool): Flag to indicate if processing should be done in memory. Defaults to False.
32 max_processes (int): The maximum number of processes to use. Defaults to 4.
34 """
35 self.in_memory: bool = in_memory
36 self.in_memory_files: dict[str, str] = {}
37 self.max_processes = max_processes
39 def once(
40 self,
41 iterable: Iterable[AST_FACTORY_AND_ATU] | IterableProvider,
42 actions: Action | Sequence[Action],
43 file_filter: str | re.Pattern[str] | None = None,
44 ) -> None:
45 """Process a given iterable of ATU objects or an IterableProvider with specified actions.
47 Args:
48 iterable (Iterable[ATU] | IterableProvider): The iterable or provider of ATU objects to process.
49 actions (Action | Sequence[Action]): The action or sequence of actions to apply to each item in the iterable.
50 file_filter (Optional[str | re.Pattern], optional): A filter to apply to file names. Defaults to None.
52 """
53 iterable = iterable() if callable(iterable) else iterable
54 self.__process(iterable, actions, self.in_memory, file_filter)
56 def repeat(
57 self,
58 iterable_provider: IterableProvider,
59 actions: Action | Sequence[Action],
60 file_filter: str | re.Pattern[str] | None = None,
61 max_repeat: int = 5,
62 ) -> None:
63 """Repeats the processing of items provided by the iterableProvider until no changes left.
65 Up to a maximum number of times.
67 Args:
68 iterable_provider (IterableProvider): A provider that yields items to be processed.
69 actions (Action | Sequence[Action]): A single action or a sequence of actions to be performed on each item.
70 file_filter (Optional[str | re.Pattern], optional): A filter to apply to the files being processed. Defaults to None.
71 max_repeat (int, optional): The maximum number of times to repeat the processing. Defaults to 5.
73 """
74 self.__process(iterable_provider(), actions, self.in_memory, file_filter, max_repeat)
76 def __process(
77 self,
78 iterable: Iterable[tuple[ASTFactory, ASTNode]],
79 actions: Action | Sequence[Action],
80 in_memory: bool = False,
81 file_filter: str | re.Pattern[str] | None = None,
82 max_repeat: int = 1,
83 ) -> None:
84 filter_pattern = (
85 file_filter if isinstance(file_filter, re.Pattern) else re.compile(file_filter) if file_filter is not None else None
86 )
88 def is_eligible(item: tuple[ASTFactory, ASTNode]) -> bool:
89 return BatchASTProcessor.__eligible_file(filter_pattern, item)
91 actions = actions if isinstance(actions, Sequence) else [actions]
92 # use parallel processing possible here
93 partial_process_item = partial(
94 process_atu,
95 self=self,
96 actions=actions,
97 in_memory=in_memory,
98 max_repeat=max_repeat,
99 )
100 with concurrent.futures.ThreadPoolExecutor(max_workers=self.max_processes) as executor:
101 for results in executor.map(partial_process_item, filter(is_eligible, iterable)):
102 for my_callable in results:
103 # the post-processing is done in the main thread
104 my_callable()
106 def _replace_if_in_memory(self, item: AST_FACTORY_AND_ATU) -> AST_FACTORY_AND_ATU:
107 if self.in_memory and self.in_memory_files.get(item[1].filename):
108 return item[0], item[0].create_from_text(
109 self.in_memory_files[item[1].filename],
110 item[1].filename,
111 )
112 return item
114 @staticmethod
115 def __eligible_file(file_filter: re.Pattern[str] | None, item: AST_FACTORY_AND_ATU) -> bool:
116 return file_filter is None or file_filter.match(item[1].filename) is not None
119def process_atu(
120 atu: AST_FACTORY_AND_ATU,
121 self: BatchASTProcessor,
122 actions: Sequence[Action],
123 in_memory: bool,
124 max_repeat: int,
125) -> Sequence[Callable[[], None]]:
126 """AI: Run the given actions against one ATU, repeating up to max_repeat times, and return deferred result callables."""
127 atu = self._replace_if_in_memory(atu)
128 ast_processor = ASTProcessor(atu[1], atu[0], in_memory)
129 results: list[Callable[[], None]] = []
131 for repeat in range(max_repeat):
132 for action in actions:
133 ast_processor.repeat_step = repeat
134 result = action(ast_processor)
135 if result:
136 results.append(result)
137 has_changed = ast_processor.has_changed()
138 if not has_changed:
139 return results
140 ast_processor = ast_processor.commit()
141 if self.in_memory:
142 self.in_memory_files[ast_processor.filename] = ast_processor.apply_to_string()
143 return results