Coverage for pySDC / core / level.py: 100%

1from typing import Any, Dict, List, Optional, Type 

2from pySDC.core.sweeper import Sweeper 

3from pySDC.core.problem import Problem 

4 

5from pySDC.helpers.pysdc_helper import FrozenClass 

6 

7 

8# short helper class to add params as attributes 

9class _Pars(FrozenClass): 

10 def __init__(self, params: Dict[str, Any]) -> None: 

11 self.dt: Optional[float] = None 

12 self.dt_initial: Optional[float] = None 

13 self.restol: float = -1.0 

14 self.nsweeps: int = 1 

15 self.residual_type: str = 'full_abs' 

16 for k, v in params.items(): 

17 setattr(self, k, v) 

18 # freeze class, no further attributes allowed from this point 

19 self._freeze() 

20 

21 self.dt_initial = self.dt * 1.0 if self.dt is not None else None 

22 

23 

24# short helper class to bundle all status variables 

25class _Status(FrozenClass): 

26 """ 

27 This class carries the status of the level. All variables that the core SDC / PFASST functionality depend on are 

28 initialized here. 

29 """ 

30 

31 def __init__(self) -> None: 

32 self.residual: Optional[float] = None 

33 self.unlocked: bool = False 

34 self.updated: bool = False 

35 self.time: Optional[float] = None 

36 self.dt_new: Optional[float] = None 

37 self.sweep: Optional[int] = None 

38 # freeze class, no further attributes allowed from this point 

39 self._freeze() 

40 

41 

42class Level(FrozenClass): 

43 """ 

44 Level class containing all management functionality for a single level 

45 

46 A level contains all data structures, types and objects to perform sweeps on this particular level. It does not 

47 know about other levels. 

48 

49 Attributes: 

50 params (__Pars): parameter object containing the custom parameters passed by the user 

51 status (__Status): status object 

52 level_index (int): custom string naming this level 

53 uend: dof values at the right end point of the interval 

54 u (list of dtype_u): dof values at the nodes 

55 uold (list of dtype_u): copy of dof values for saving data during restriction) 

56 f (list of dtype_f): RHS values at the nodes 

57 fold (list of dtype_f): copy of RHS values for saving data during restriction 

58 tau (list of dtype_u): FAS correction, allocated via step class if necessary 

59 """ 

60 

61 def __init__( 

62 self, 

63 problem_class: Type[Problem], 

64 problem_params: Dict[str, Any], 

65 sweeper_class: Type[Sweeper], 

66 sweeper_params: Dict[str, Any], 

67 level_params: Dict[str, Any], 

68 level_index: int, 

69 ) -> None: 

70 """ 

71 Initialization routine 

72 

73 Args: 

74 problem_class: problem class 

75 problem_params (dict): parameters for the problem to be initialized 

76 sweeper_class: sweeper class 

77 sweeper_params (dict): parameters for the sweeper (contains collocation) 

78 level_params (dict): parameters given by the user, will be added as attributes 

79 level_index (int): custom name for this level 

80 """ 

81 

82 # set level parameters and status 

83 self.params: _Pars = _Pars(level_params) 

84 self.status: _Status = _Status() 

85 

86 # instantiate sweeper, problem and hooks 

87 self.__sweep: Sweeper = sweeper_class(sweeper_params, self) 

88 self.__prob: Problem = problem_class(**problem_params) 

89 

90 # set name 

91 self.level_index: int = level_index 

92 

93 # empty data at the nodes, the right end point and tau 

94 self.uend: Optional[Any] = None 

95 self.u: List[Optional[Any]] = [None] * (self.sweep.coll.num_nodes + 1) 

96 self.uold: List[Optional[Any]] = [None] * (self.sweep.coll.num_nodes + 1) 

97 self.u_avg: List[Optional[Any]] = [None] * self.sweep.coll.num_nodes 

98 self.residual: List[Optional[Any]] = [None] * self.sweep.coll.num_nodes 

99 self.increment: List[Optional[Any]] = [None] * self.sweep.coll.num_nodes 

100 self.f: List[Optional[Any]] = [None] * (self.sweep.coll.num_nodes + 1) 

101 self.fold: List[Optional[Any]] = [None] * (self.sweep.coll.num_nodes + 1) 

102 

103 self.tau: List[Optional[Any]] = [None] * self.sweep.coll.num_nodes 

104 

105 self.__tag: Optional[Any] = None 

106 

107 # freeze class, no further attributes allowed from this point 

108 self._freeze() 

109 

110 def reset_level(self, reset_status: bool = True) -> None: 

111 """ 

112 Routine to clean-up the level for the next time step 

113 

114 Args: 

115 reset_status (bool): Reset the status or only the solution 

116 

117 Returns: 

118 None 

119 """ 

120 

121 # reset status 

122 if reset_status: 

123 self.status = _Status() 

124 

125 # all data back to None 

126 self.uend = None 

127 self.u = [None] * (self.sweep.coll.num_nodes + 1) 

128 self.uold = [None] * (self.sweep.coll.num_nodes + 1) 

129 self.f = [None] * (self.sweep.coll.num_nodes + 1) 

130 self.fold = [None] * (self.sweep.coll.num_nodes + 1) 

131 self.tau = [None] * self.sweep.coll.num_nodes 

132 

133 @property 

134 def sweep(self) -> Sweeper: 

135 """ 

136 Getter for the sweeper 

137 

138 Returns: 

139 pySDC.Sweeper.sweeper: the sweeper associated to this level 

140 """ 

141 return self.__sweep 

142 

143 @property 

144 def prob(self) -> Problem: 

145 """ 

146 Getter for the problem 

147 

148 Returns: 

149 pySDC.Problem.ptype: the problem associated to this level 

150 """ 

151 return self.__prob 

152 

153 @property 

154 def time(self) -> Optional[float]: 

155 """ 

156 Meta-getter for the current time 

157 

158 Returns: 

159 float: referencing status time for convenience 

160 """ 

161 return self.status.time 

162 

163 @property 

164 def dt(self) -> Optional[float]: 

165 """ 

166 Meta-getter for the time-step size 

167 

168 Returns: 

169 float: referencing dt from parameters for convenience 

170 """ 

171 return self.params.dt 

172 

173 @property 

174 def tag(self) -> Optional[Any]: 

175 """ 

176 Getter for tag 

177 

178 Returns: 

179 tag for sending/receiving 

180 """ 

181 return self.__tag 

182 

183 @tag.setter 

184 def tag(self, t: Any) -> None: 

185 """ 

186 Setter for tag 

187 

188 Args: 

189 t: new tag for sending/receiving 

190 """ 

191 self.__tag = t