Coverage for pySDC / core / hooks.py: 96%
57 statements
« prev ^ index » next coverage.py v7.13.4, created at 2026-03-13 09:00 +0000
1import logging
2from typing import Any, Dict, Optional, TYPE_CHECKING
3from collections import namedtuple
5if TYPE_CHECKING:
6 from pySDC.core.step import Step
8# metadata with defaults
9meta_data = {
10 'process': None,
11 'process_sweeper': None,
12 'time': None,
13 'level': None,
14 'iter': None,
15 'sweep': None,
16 'type': None,
17 'num_restarts': None,
18}
19Entry = namedtuple('Entry', meta_data.keys())
22# noinspection PyUnusedLocal,PyShadowingBuiltins,PyShadowingNames
23class Hooks(object):
24 """
25 Hook class to contain the functions called during the controller runs (e.g. for calling user-routines)
27 When deriving a custom hook from this class make sure to always call the parent method using e.g.
28 `super().post_step(step, level_number)`. Otherwise bugs may arise when using `filer_recomputed` from the stats
29 helper for post processing.
31 Attributes:
32 logger: logger instance for output
33 __num_restarts (int): number of restarts of the current step
34 __stats (dict): dictionary for gathering the statistics of a run
35 entry (namedtuple): statistics entry containing all information to identify the value
36 """
38 entry = Entry
39 meta_data = meta_data
41 def __init__(self) -> None:
42 """
43 Initialization routine
44 """
45 self.__num_restarts: int = 0
47 self.logger: logging.Logger = logging.getLogger('hooks')
49 # create statistics and entry elements
50 self.__stats: Dict[Entry, Any] = {}
52 def add_to_stats(self, value: Any, **kwargs: Any) -> None:
53 """
54 Routine to add data to the statistics dict. Please supply the metadata as keyword arguments in accordance with
55 the entry class.
57 Args:
58 value: the actual data
59 """
60 # create named tuple for the key and add to dict
61 meta = {
62 **self.meta_data,
63 **kwargs,
64 'num_restarts': self.__num_restarts,
65 }
66 self.__stats[self.entry(**meta)] = value
68 def increment_stats(self, value: Any, initialize: Optional[Any] = None, **kwargs: Any) -> None:
69 """
70 Routine to increment data to the statistics dict. If the data is not yet created, it will be initialized to
71 initialize if applicable and to value otherwise. Please supply metadata as keyword arguments in accordance with
72 the entry class.
74 Args:
75 value: the actual data
76 initialize: if supplied and data does not exist already, this will be used over value
77 """
78 meta = {
79 **meta_data,
80 **kwargs,
81 'num_restarts': self.__num_restarts,
82 }
83 key = self.entry(**meta)
84 if key in self.__stats.keys():
85 self.__stats[key] += value
86 elif initialize is not None:
87 self.__stats[key] = initialize
88 else:
89 self.__stats[key] = value
91 def return_stats(self) -> Dict[Entry, Any]:
92 """
93 Getter for the stats
95 Returns:
96 dict: stats
97 """
98 return self.__stats
100 def reset_stats(self) -> None:
101 """
102 Function to reset the stats for multiple runs
103 """
104 self.__stats = {}
106 def pre_setup(self, step: Optional['Step'], level_number: int) -> None:
107 """
108 Default routine called before setup starts
110 Args:
111 step (pySDC.Step.step): the current step
112 level_number (int): the current level number
113 """
114 self.__num_restarts = step.status.get('restarts_in_a_row') if step is not None else 0
116 def pre_run(self, step: Optional['Step'], level_number: int) -> None:
117 """
118 Default routine called before time-loop starts
120 Args:
121 step (pySDC.Step.step): the current step
122 level_number (int): the current level number
123 """
124 self.__num_restarts = step.status.get('restarts_in_a_row') if step is not None else 0
126 def pre_predict(self, step: Optional['Step'], level_number: int) -> None:
127 """
128 Default routine called before predictor starts
130 Args:
131 step (pySDC.Step.step): the current step
132 level_number (int): the current level number
133 """
134 self.__num_restarts = step.status.get('restarts_in_a_row') if step is not None else 0
136 def pre_step(self, step: Optional['Step'], level_number: int) -> None:
137 """
138 Hook called before each step
140 Args:
141 step (pySDC.Step.step): the current step
142 level_number (int): the current level number
143 """
144 self.__num_restarts = step.status.get('restarts_in_a_row') if step is not None else 0
146 def pre_iteration(self, step: Optional['Step'], level_number: int) -> None:
147 """
148 Default routine called before iteration starts
150 Args:
151 step (pySDC.Step.step): the current step
152 level_number (int): the current level number
153 """
154 self.__num_restarts = step.status.get('restarts_in_a_row') if step is not None else 0
156 def pre_sweep(self, step: Optional['Step'], level_number: int) -> None:
157 """
158 Default routine called before sweep starts
160 Args:
161 step (pySDC.Step.step): the current step
162 level_number (int): the current level number
163 """
164 self.__num_restarts = step.status.get('restarts_in_a_row') if step is not None else 0
166 def pre_comm(self, step: Optional['Step'], level_number: int) -> None:
167 """
168 Default routine called before communication starts
170 Args:
171 step (pySDC.Step.step): the current step
172 level_number (int): the current level number
173 """
174 self.__num_restarts = step.status.get('restarts_in_a_row') if step is not None else 0
176 def post_comm(self, step: Optional['Step'], level_number: int, add_to_stats: bool = False) -> None:
177 """
178 Default routine called after each communication
180 Args:
181 step (pySDC.Step.step): the current step
182 level_number (int): the current level number
183 add_to_stats (bool): set if result should go to stats object
184 """
185 self.__num_restarts = step.status.get('restarts_in_a_row') if step is not None else 0
187 def post_sweep(self, step: Optional['Step'], level_number: int) -> None:
188 """
189 Default routine called after each sweep
191 Args:
192 step (pySDC.Step.step): the current step
193 level_number (int): the current level number
194 """
195 self.__num_restarts = step.status.get('restarts_in_a_row') if step is not None else 0
197 def post_iteration(self, step: Optional['Step'], level_number: int) -> None:
198 """
199 Default routine called after each iteration
201 Args:
202 step (pySDC.Step.step): the current step
203 level_number (int): the current level number
204 """
205 self.__num_restarts = step.status.get('restarts_in_a_row') if step is not None else 0
207 def post_step(self, step: Optional['Step'], level_number: int) -> None:
208 """
209 Default routine called after each step or block
211 Args:
212 step (pySDC.Step.step): the current step
213 level_number (int): the current level number
214 """
215 self.__num_restarts = step.status.get('restarts_in_a_row') if step is not None else 0
217 def post_predict(self, step: Optional['Step'], level_number: int) -> None:
218 """
219 Default routine called after each predictor
221 Args:
222 step (pySDC.Step.step): the current step
223 level_number (int): the current level number
224 """
225 self.__num_restarts = step.status.get('restarts_in_a_row') if step is not None else 0
227 def post_run(self, step: Optional['Step'], level_number: int) -> None:
228 """
229 Default routine called after each run
231 Args:
232 step (pySDC.Step.step): the current step
233 level_number (int): the current level number
234 """
235 self.__num_restarts = step.status.get('restarts_in_a_row') if step is not None else 0
237 def post_setup(self, step: Optional['Step'], level_number: int) -> None:
238 """
239 Default routine called after setup
241 Args:
242 step (pySDC.Step.step): the current step
243 level_number (int): the current level number
244 """
245 self.__num_restarts = step.status.get('restarts_in_a_row') if step is not None else 0