Coverage for C:\src\imod-python\imod\mf6\riv.py: 97%

29 statements  

« prev     ^ index     » next       coverage.py v7.4.4, created at 2024-04-08 10:26 +0200

1from typing import Optional, Tuple 

2 

3import numpy as np 

4 

5from imod.logging import init_log_decorator 

6from imod.mf6.boundary_condition import BoundaryCondition 

7from imod.mf6.interfaces.iregridpackage import IRegridPackage 

8from imod.mf6.utilities.regrid import RegridderType 

9from imod.mf6.validation import BOUNDARY_DIMS_SCHEMA, CONC_DIMS_SCHEMA 

10from imod.schemata import ( 

11 AllInsideNoDataSchema, 

12 AllNoDataSchema, 

13 AllValueSchema, 

14 CoordsSchema, 

15 DimsSchema, 

16 DTypeSchema, 

17 IdentityNoDataSchema, 

18 IndexesSchema, 

19 OtherCoordsSchema, 

20) 

21 

22 

23class River(BoundaryCondition, IRegridPackage): 

24 """ 

25 River package. 

26 Any number of RIV Packages can be specified for a single groundwater flow 

27 model. 

28 https://water.usgs.gov/water-resources/software/MODFLOW-6/mf6io_6.0.4.pdf#page=71 

29 

30 Parameters 

31 ---------- 

32 stage: array of floats (xr.DataArray) 

33 is the head in the river. 

34 conductance: array of floats (xr.DataArray) 

35 is the riverbed hydraulic conductance. 

36 bottom_elevation: array of floats (xr.DataArray) 

37 is the elevation of the bottom of the riverbed. 

38 concentration: array of floats (xr.DataArray, optional) 

39 if this flow package is used in simulations also involving transport, then this array is used 

40 as the concentration for inflow over this boundary. 

41 concentration_boundary_type: ({"AUX", "AUXMIXED"}, optional) 

42 if this flow package is used in simulations also involving transport, then this keyword specifies 

43 how outflow over this boundary is computed. 

44 print_input: ({True, False}, optional) 

45 keyword to indicate that the list of river information will be written 

46 to the listing file immediately after it is read. Default is False. 

47 print_flows: ({True, False}, optional) 

48 Indicates that the list of river flow rates will be printed to the 

49 listing file for every stress period time step in which "BUDGET PRINT" 

50 is specified in Output Control. If there is no Output Control option and 

51 PRINT FLOWS is specified, then flow rates are printed for the last time 

52 step of each stress period. Default is False. 

53 save_flows: ({True, False}, optional) 

54 Indicates that river flow terms will be written to the file specified 

55 with "BUDGET FILEOUT" in Output Control. Default is False. 

56 observations: [Not yet supported.] 

57 Default is None. 

58 validate: {True, False} 

59 Flag to indicate whether the package should be validated upon 

60 initialization. This raises a ValidationError if package input is 

61 provided in the wrong manner. Defaults to True. 

62 repeat_stress: Optional[xr.DataArray] of datetimes 

63 Used to repeat data for e.g. repeating stress periods such as 

64 seasonality without duplicating the values. The DataArray should have 

65 dimensions ``("repeat", "repeat_items")``. The ``repeat_items`` 

66 dimension should have size 2: the first value is the "key", the second 

67 value is the "value". For the "key" datetime, the data of the "value" 

68 datetime will be used. Can also be set with a dictionary using the 

69 ``set_repeat_stress`` method. 

70 """ 

71 

72 _pkg_id = "riv" 

73 _period_data = ("stage", "conductance", "bottom_elevation") 

74 _keyword_map = {} 

75 

76 _init_schemata = { 

77 "stage": [ 

78 DTypeSchema(np.floating), 

79 IndexesSchema(), 

80 CoordsSchema(("layer",)), 

81 BOUNDARY_DIMS_SCHEMA, 

82 ], 

83 "conductance": [ 

84 DTypeSchema(np.floating), 

85 IndexesSchema(), 

86 CoordsSchema(("layer",)), 

87 BOUNDARY_DIMS_SCHEMA, 

88 ], 

89 "bottom_elevation": [ 

90 DTypeSchema(np.floating), 

91 IndexesSchema(), 

92 CoordsSchema(("layer",)), 

93 BOUNDARY_DIMS_SCHEMA, 

94 ], 

95 "concentration": [ 

96 DTypeSchema(np.floating), 

97 IndexesSchema(), 

98 CoordsSchema( 

99 ( 

100 "species", 

101 "layer", 

102 ) 

103 ), 

104 CONC_DIMS_SCHEMA, 

105 ], 

106 "print_input": [DTypeSchema(np.bool_), DimsSchema()], 

107 "print_flows": [DTypeSchema(np.bool_), DimsSchema()], 

108 "save_flows": [DTypeSchema(np.bool_), DimsSchema()], 

109 } 

110 _write_schemata = { 

111 "stage": [ 

112 AllValueSchema(">=", "bottom_elevation"), 

113 OtherCoordsSchema("idomain"), 

114 AllNoDataSchema(), # Check for all nan, can occur while clipping 

115 AllInsideNoDataSchema(other="idomain", is_other_notnull=(">", 0)), 

116 ], 

117 "conductance": [IdentityNoDataSchema("stage"), AllValueSchema(">", 0.0)], 

118 "bottom_elevation": [ 

119 IdentityNoDataSchema("stage"), 

120 # Check river bottom above layer bottom, else Modflow throws error. 

121 AllValueSchema(">=", "bottom"), 

122 ], 

123 "concentration": [IdentityNoDataSchema("stage"), AllValueSchema(">=", 0.0)], 

124 } 

125 

126 _template = BoundaryCondition._initialize_template(_pkg_id) 

127 _auxiliary_data = {"concentration": "species"} 

128 

129 _regrid_method = { 

130 "stage": (RegridderType.OVERLAP, "mean"), 

131 "conductance": (RegridderType.RELATIVEOVERLAP, "conductance"), 

132 "bottom_elevation": (RegridderType.OVERLAP, "mean"), 

133 "concentration": (RegridderType.OVERLAP, "mean"), 

134 } 

135 

136 @init_log_decorator() 

137 def __init__( 

138 self, 

139 stage, 

140 conductance, 

141 bottom_elevation, 

142 concentration=None, 

143 concentration_boundary_type="aux", 

144 print_input=False, 

145 print_flows=False, 

146 save_flows=False, 

147 observations=None, 

148 validate: bool = True, 

149 repeat_stress=None, 

150 ): 

151 dict_dataset = { 

152 "stage": stage, 

153 "conductance": conductance, 

154 "bottom_elevation": bottom_elevation, 

155 "concentration": concentration, 

156 "concentration_boundary_type": concentration_boundary_type, 

157 "print_input": print_input, 

158 "print_flows": print_flows, 

159 "save_flows": save_flows, 

160 "observations": observations, 

161 "repeat_stress": repeat_stress, 

162 } 

163 super().__init__(dict_dataset) 

164 

165 self._validate_init_schemata(validate) 

166 

167 def _validate(self, schemata, **kwargs): 

168 # Insert additional kwargs 

169 kwargs["stage"] = self["stage"] 

170 kwargs["bottom_elevation"] = self["bottom_elevation"] 

171 errors = super()._validate(schemata, **kwargs) 

172 

173 return errors 

174 

175 def get_regrid_methods(self) -> Optional[dict[str, Tuple[RegridderType, str]]]: 

176 return self._regrid_method