Coverage for C:\src\imod-python\imod\mf6\out\__init__.py: 81%
58 statements
« prev ^ index » next coverage.py v7.4.4, created at 2024-04-08 13:27 +0200
« prev ^ index » next coverage.py v7.4.4, created at 2024-04-08 13:27 +0200
1"""
2Read MODFLOW6 output
4The dis, disv, disu modules implement the following functions:
6```python
7Darray = Union[xr.DataArray, xu.UgridDataArray]
9def read_grb(f: BinaryIO, ntxt: int, lentxt: int) -> Dict[str, Any]:
10 return
12def read_times(*args) -> FloatArray:
13 return
15def read_hds_timestep(*args) -> FloatArray:
16 return
18def open_hds(path: FilePath, d: Dict[str, Any], dry_nan: bool) -> Darray:
19 return
21def open_imeth1_budgets(
22 cbc_path: FilePath, grb_content: dict, header_list: List["Imeth1Header"]
23) -> Darray:
24 return
26def open_imeth6_budgets(
27 cbc_path: FilePath, grb_content: dict, header_list: List["Imeth6Header"]
28) -> Darray:
29 return
31def open_cbc(
32 cbc_path: FilePath, grb_content: Dict[str, Any]
33) -> Dict[str, Darray]:
34 return
35```
37(These could be implemented via Reader classes, but why bother with mutable
38state or a class with exclusively staticmethods?)
39"""
41from typing import Any, Callable, Dict, Optional, Union
43import numpy as np
44import xarray as xr
45import xugrid as xu
47from imod.typing import GridDataArray, GridDataset
48from imod.typing.grid import merge_with_dictionary
50from . import dis, disu, disv
51from .cbc import read_cbc_headers
52from .common import FilePath, _grb_text
54_READ_GRB = {
55 "grid dis": dis.read_grb,
56 "grid disv": disv.read_grb,
57 "grid disu": disu.read_grb,
58}
60_OPEN_HDS = {
61 "dis": dis.open_hds,
62 "disv": disv.open_hds,
63 "disu": disu.open_hds,
64}
66_OPEN_CBC = {
67 "dis": dis.open_cbc,
68 "disv": disv.open_cbc,
69 "disu": disu.open_cbc,
70}
73def _get_function(d: Dict[str, Callable], key: str) -> Callable:
74 try:
75 func = d[key]
76 except KeyError:
77 valid_options = ", ".join(d.keys()).lower()
78 raise ValueError(f"Expected one of {valid_options}, got: {key}")
79 return func
82def read_grb(path: FilePath) -> Dict[str, Any]:
83 """
84 Read the data in a MODFLOW6 binary grid (.grb) file.
86 Parameters
87 ----------
88 path: Union[str, pathlib.Path]
90 Returns
91 -------
92 grb_content: Dict[str, Any]
93 """
94 with open(path, "rb") as f:
95 h1 = _grb_text(f)
96 _read = _get_function(_READ_GRB, h1)
97 h2 = _grb_text(f)
98 if h2 != "version 1":
99 raise ValueError(f"Only version 1 supported, got {h2}")
100 ntxt = int(_grb_text(f).split()[1])
101 lentxt = int(_grb_text(f).split()[1])
102 d = _read(f, ntxt, lentxt)
103 return d
106def open_hds(
107 hds_path: FilePath,
108 grb_path: FilePath,
109 dry_nan: bool = False,
110 simulation_start_time: Optional[np.datetime64] = None,
111 time_unit: Optional[str] = "d",
112) -> GridDataArray:
113 """
114 Open modflow6 heads (.hds) file.
116 The data is lazily read per timestep and automatically converted into
117 (dense) xr.DataArrays or xu.UgridDataArrays, for DIS and DISV respectively.
118 The conversion is done via the information stored in the Binary Grid file
119 (GRB).
122 Parameters
123 ----------
124 hds_path: Union[str, pathlib.Path]
125 grb_path: Union[str, pathlib.Path]
126 dry_nan: bool, default value: False.
127 Whether to convert dry values to NaN.
128 simulation_start_time : Optional datetime
129 The time and date correpsonding to the beginning of the simulation.
130 Use this to convert the time coordinates of the output array to
131 calendar time/dates. time_unit must also be present if this argument is present.
132 time_unit: Optional str
133 The time unit MF6 is working in, in string representation.
134 Only used if simulation_start_time was provided.
135 Admissible values are:
136 ns -> nanosecond
137 ms -> microsecond
138 s -> second
139 m -> minute
140 h -> hour
141 d -> day
142 w -> week
143 Units "month" or "year" are not supported, as they do not represent unambiguous timedelta values durations.
145 Returns
146 -------
147 head: Union[xr.DataArray, xu.UgridDataArray]
148 """
149 grb_content = read_grb(grb_path)
150 grb_content["name"] = "head"
151 distype = grb_content["distype"]
152 _open = _get_function(_OPEN_HDS, distype)
153 return _open(hds_path, grb_content, dry_nan, simulation_start_time, time_unit)
156def open_conc(
157 ucn_path: FilePath,
158 grb_path: FilePath,
159 dry_nan: bool = False,
160 simulation_start_time: Optional[np.datetime64] = None,
161 time_unit: Optional[str] = "d",
162) -> GridDataArray:
163 """
164 Open Modflow6 "Unformatted Concentration" (.ucn) file.
166 The data is lazily read per timestep and automatically converted into
167 (dense) xr.DataArrays or xu.UgridDataArrays, for DIS and DISV respectively.
168 The conversion is done via the information stored in the Binary Grid file
169 (GRB).
171 Parameters
172 ----------
173 ucn_path: Union[str, pathlib.Path]
174 grb_path: Union[str, pathlib.Path]
175 dry_nan: bool, default value: False.
176 Whether to convert dry values to NaN.
177 simulation_start_time : Optional datetime
178 The time and date correpsonding to the beginning of the simulation.
179 Use this to convert the time coordinates of the output array to
180 calendar time/dates. time_unit must also be present if this argument is present.
181 time_unit: Optional str
182 The time unit MF6 is working in, in string representation.
183 Only used if simulation_start_time was provided.
184 Admissible values are:
185 ns -> nanosecond
186 ms -> microsecond
187 s -> second
188 m -> minute
189 h -> hour
190 d -> day
191 w -> week
192 Units "month" or "year" are not supported, as they do not represent unambiguous timedelta values durations.
194 Returns
195 -------
196 concentration: Union[xr.DataArray, xu.UgridDataArray]
197 """
198 grb_content = read_grb(grb_path)
199 grb_content["name"] = "concentration"
200 distype = grb_content["distype"]
201 _open = _get_function(_OPEN_HDS, distype)
202 return _open(ucn_path, grb_content, dry_nan, simulation_start_time, time_unit)
205def open_hds_like(
206 path: FilePath,
207 like: Union[xr.DataArray, xu.UgridDataArray],
208 dry_nan: bool = False,
209) -> GridDataArray:
210 """
211 Open modflow6 heads (.hds) file.
213 The data is lazily read per timestep and automatically converted into
214 DataArrays. Shape and coordinates are inferred from ``like``.
216 Parameters
217 ----------
218 hds_path: Union[str, pathlib.Path]
219 like: Union[xr.DataArray, xu.UgridDataArray]
220 dry_nan: bool, default value: False.
221 Whether to convert dry values to NaN.
223 Returns
224 -------
225 head: Union[xr.DataArray, xu.UgridDataArray]
226 """
227 # TODO: check shape with hds metadata.
228 if isinstance(like, xr.DataArray):
229 d = dis.grid_info(like)
230 return dis.open_hds(path, d, dry_nan)
232 elif isinstance(like, xu.UgridDataArray):
233 d = disv.grid_info(like)
234 return disv.open_hds(path, d, dry_nan)
236 else:
237 raise TypeError(
238 "like should be a DataArray or UgridDataArray, "
239 f"received instead {type(like)}"
240 )
243def open_cbc(
244 cbc_path: FilePath,
245 grb_path: FilePath,
246 flowja: bool = False,
247 simulation_start_time: Optional[np.datetime64] = None,
248 time_unit: Optional[str] = "d",
249 merge_to_dataset: bool = False,
250) -> GridDataset | Dict[str, GridDataArray]:
251 """
252 Open modflow6 cell-by-cell (.cbc) file.
254 The data is lazily read per timestep and automatically converted into
255 (dense) xr.DataArrays or xu.UgridDataArrays, for DIS and DISV respectively.
256 The conversion is done via the information stored in the Binary Grid file
257 (GRB).
259 The ``flowja`` argument controls whether the flow-ja-face array (if present)
260 is returned in grid form as "as is". By default ``flowja=False`` and the
261 array is returned in "grid form", meaning:
263 * DIS: in right, front, and lower face flow. All flows are placed in
264 the cell.
265 * DISV: in horizontal and lower face flow.the horizontal flows are
266 placed on the edges and the lower face flow is placed on the faces.
268 When ``flowja=True``, the flow-ja-face array is returned as it is found in
269 the CBC file, with a flow for every cell to cell connection. Additionally,
270 a ``connectivity`` DataArray is returned describing for every cell (n) its
271 connected cells (m).
273 Parameters
274 ----------
275 cbc_path: str, pathlib.Path
276 Path to the cell-by-cell flows file
277 grb_path: str, pathlib.Path
278 Path to the binary grid file
279 flowja: bool, default value: False
280 Whether to return the flow-ja-face values "as is" (``True``) or in a
281 grid form (``False``).
282 simulation_start_time : Optional datetime
283 The time and date correpsonding to the beginning of the simulation.
284 Use this to convert the time coordinates of the output array to
285 calendar time/dates. time_unit must also be present if this argument is present.
286 time_unit: Optional str
287 The time unit MF6 is working in, in string representation.
288 Only used if simulation_start_time was provided.
289 Admissible values are:
290 ns -> nanosecond
291 ms -> microsecond
292 s -> second
293 m -> minute
294 h -> hour
295 d -> day
296 w -> week
297 Units "month" or "year" are not supported, as they do not represent unambiguous timedelta values durations.
298 merge_to_dataset: bool, default value: False
299 Merge output to dataset.
301 Returns
302 -------
303 cbc_content: xr.Dataset | Dict[str, xr.DataArray]
304 DataArray contains float64 data of the budgets, with dimensions ("time",
305 "layer", "y", "x").
307 Examples
308 --------
310 Open a cbc file:
312 >>> import imod
313 >>> cbc_content = imod.mf6.open_cbc("budgets.cbc", "my-model.grb")
315 Check the contents:
317 >>> print(cbc_content.keys())
319 Get the drainage budget, compute a time mean for the first layer:
321 >>> drn_budget = cbc_content["drn]
322 >>> mean = drn_budget.sel(layer=1).mean("time")
324 """
325 grb_content = read_grb(grb_path)
326 distype = grb_content["distype"]
327 _open = _get_function(_OPEN_CBC, distype)
328 cbc = _open(cbc_path, grb_content, flowja, simulation_start_time, time_unit)
329 if merge_to_dataset:
330 return merge_with_dictionary(cbc)
331 return cbc