Coverage for pySDC/helpers/stats_helper.py: 100%
34 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-10-17 13:41 +0000
1import numpy as np
4def filter_stats(stats, comm=None, recomputed=None, **kwargs):
5 """
6 Helper function to extract data from the dictionary of statistics. Please supply metadata as keyword arguments.
8 Args:
9 stats (dict): raw statistics from a controller run
10 recomputed (bool): filter recomputed values from stats if set to anything other than None
11 comm (mpi4py.MPI.Intracomm): Communicator (or None if not applicable)
13 Returns:
14 dict: dictionary containing only the entries corresponding to the filter
15 """
16 # read the fields straight off the key: building `k._asdict()` for every entry made this the slow part
17 filters = [(k2, v2) for k2, v2 in kwargs.items() if v2 is not None]
18 result = {k: v for k, v in stats.items() if all(getattr(k, k2, None) == v2 for k2, v2 in filters)}
20 if comm is not None:
21 # gather the results across all ranks
22 result = {key: value for sub_result in comm.allgather(result) for key, value in sub_result.items()}
24 if recomputed is not None:
25 # the entries by time, gathered once rather than filtered out of `result` again for every time below
26 by_time = {}
27 for me in result:
28 by_time.setdefault(me.time, []).append(me)
30 # delete values that have been recorded and superseded by similar, but not identical keys
31 times_restarted = np.unique([me.time for me in result.keys() if me.num_restarts > 0])
32 for t in times_restarted:
33 restarts = {}
34 stats_now = {me: result[me] for me in by_time.get(t, [])} # a NaN time matches nothing, as `==` did
35 for me in stats_now.keys():
36 restarts[me.type] = max([restarts.get(me.type, 0), me.num_restarts])
38 [
39 [
40 [result.pop(you, None) for you in filter_stats(stats_now, type=type_, num_restarts=i).keys()]
41 for i in range(num_restarts_)
42 ]
43 for type_, num_restarts_ in restarts.items()
44 ]
46 # delete values that were recorded at times that shouldn't be recorded because we performed a different step after the restart
47 if kwargs.get('type', None) != '_recomputed':
48 other_restarted_steps = [
49 key for key, val in filter_stats(stats, type='_recomputed', recomputed=False, comm=comm).items() if val
50 ]
51 for step in other_restarted_steps:
52 for me in by_time.get(step.time, []):
53 result.pop(me, None)
55 return result
58def sort_stats(stats, sortby):
59 """
60 Helper function to transform stats dictionary to sorted list of tuples
62 Args:
63 stats (dict): dictionary of statistics
64 sortby (str): string to specify which key to use for sorting
66 Returns:
67 list: list of tuples containing the sortby item and the value
68 """
70 result = []
71 for k, v in stats.items():
72 # convert string to attribute and append key + value to result as tuple
73 item = getattr(k, sortby)
74 result.append((item, v))
76 # sort by first element of the tuple (which is the sortby key) and return
77 sorted_data = sorted(result, key=lambda tup: tup[0])
79 return sorted_data
82def get_list_of_types(stats):
83 """
84 Helper function to get list of types registered in stats
86 Args:
87 stats (dict): dictionary with statistics
89 Returns:
90 list: list of types registered
92 """
94 # in order of first appearance, as a dict keeps it, without searching a list for every entry
95 return list(dict.fromkeys(k.type for k in stats))
98def get_sorted(stats, sortby='time', **kwargs):
99 """
100 Utility for filtering and sorting stats in a single call. Pass a communicator if using MPI.
101 Keyword arguments are passed to `filter_stats` for filtering.
103 stats (dict): raw statistics from a controller run
104 sortby (str): string to specify which key to use for sorting
106 Returns:
107 list: list of tuples containing the sortby item and the value
108 """
110 return sort_stats(
111 filter_stats(stats, **kwargs),
112 sortby=sortby,
113 )