Coverage for C:\src\imod-python\imod\mf6\mf6_wel_adapter.py: 98%

40 statements  

« prev     ^ index     » next       coverage.py v7.5.1, created at 2024-05-08 14:15 +0200

1""" 

2Module to store (prototype) Modflow 6 adapter classes. 

3 

4These are closer to the Modflow 6 data model, with a cellid, 

5instead of x, y coordinates. 

6 

7We plan to split up the present attributes of the classes in Package.py and BoundaryCondition.py into low-level and 

8high-level classes. 

9 

10The high-level class contains grids with x, y, z coordinates, closely linked to 

11GIS systems. The low-level classes contain a dataset based on cellid, 

12consisting of layer, row, and column, closely resembling input for Modflow6. 

13""" 

14 

15from typing import Optional 

16 

17import numpy as np 

18 

19from imod.mf6.boundary_condition import BoundaryCondition 

20from imod.schemata import DTypeSchema 

21 

22# FUTURE: There was an idea to autogenerate modflow 6 adapters. 

23# This was relevant: 

24# https://github.com/Deltares/xugrid/blob/main/xugrid/core/wrap.py#L90 

25 

26 

27class Mf6Wel(BoundaryCondition): 

28 """ 

29 Package resembling input for Modflow 6 List Input. This class has 

30 methods for the modflow 6 wel packages with time component. 

31 

32 This class only supports `list input 

33 <https://water.usgs.gov/water-resources/software/MODFLOW-6/mf6io_6.0.4.pdf#page=19>`_, 

34 not the array input which is used in :class:`Mf6Package`. 

35 """ 

36 

37 _pkg_id = "wel" 

38 

39 _period_data = ("cellid", "rate") 

40 _keyword_map = {} 

41 _template = BoundaryCondition._initialize_template(_pkg_id) 

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

43 

44 _init_schemata = { 

45 "cellid": [DTypeSchema(np.integer)], 

46 "rate": [DTypeSchema(np.floating)], 

47 "concentration": [DTypeSchema(np.floating)], 

48 } 

49 _write_schemata = {} 

50 

51 def __init__( 

52 self, 

53 cellid, 

54 rate, 

55 concentration=None, 

56 concentration_boundary_type="aux", 

57 save_flows: Optional[bool] = None, 

58 print_flows: Optional[bool] = None, 

59 print_input: Optional[bool] = None, 

60 validate: bool = True, 

61 ): 

62 dict_dataset = { 

63 "cellid": cellid, 

64 "rate": rate, 

65 "concentration": concentration, 

66 "concentration_boundary_type": concentration_boundary_type, 

67 "save_flows": save_flows, 

68 "print_flows": print_flows, 

69 "print_input": print_input, 

70 } 

71 super().__init__(dict_dataset) 

72 self._validate_init_schemata(validate) 

73 

74 def _ds_to_arrdict(self, ds): 

75 """ 

76 Prepares a dictionary with values needed for the _to_sparse method. 

77 """ 

78 arrdict = {} 

79 

80 arrdict["data_vars"] = [ 

81 var_name for var_name in ds.data_vars if var_name != "cellid" 

82 ] 

83 

84 dsvar = {} 

85 for var in arrdict["data_vars"]: 

86 dsvar[var] = ds[var] 

87 arrdict["var_values"] = dsvar 

88 

89 arrdict["cellid_names"] = ds.coords["nmax_cellid"].values 

90 arrdict["nrow"] = ds.coords["ncellid"].size 

91 arrdict["cellid"] = ds["cellid"] 

92 

93 return arrdict 

94 

95 def _to_struct_array(self, arrdict, _): 

96 index_spec = [(index, np.int32) for index in arrdict["cellid_names"]] 

97 field_spec = [(var, np.float64) for var in arrdict["data_vars"]] 

98 sparse_dtype = np.dtype(index_spec + field_spec) 

99 

100 # Initialize the structured array 

101 recarr = np.empty(arrdict["nrow"], dtype=sparse_dtype) 

102 for cellid_name in arrdict["cellid_names"]: 

103 recarr[cellid_name] = arrdict["cellid"].sel(nmax_cellid=cellid_name).values 

104 

105 for var in arrdict["data_vars"]: 

106 recarr[var] = arrdict["var_values"][var] 

107 

108 return recarr 

109 

110 @classmethod 

111 def from_file(cls, path, **kwargs) -> "Mf6Wel": 

112 """ 

113 Instantiate class from modflow 6 file 

114 """ 

115 # return cls.__new__(cls) 

116 raise NotImplementedError("from_file not implemented")