Coverage for brodata / gpd.py: 83%

1import logging 

2from functools import partial 

3import pandas as pd 

4 

5from . import bro, util 

6 

7logger = logging.getLogger(__name__) 

8 

9 

10class GroundwaterProductionDossier(bro.FileOrUrl): 

11 """Class to represent a Groundwater Production Dossier (GPD) from the BRO.""" 

12 

13 _rest_url = "https://publiek.broservices.nl/gu/gpd/v1" 

14 _xmlns = "http://www.broservices.nl/xsd/dsgpd/1.0" 

15 

16 def _read_contents(self, tree): 

17 ns = { 

18 "brocom": "http://www.broservices.nl/xsd/brocommon/3.0", 

19 "xmlns": self._xmlns, 

20 } 

21 

22 gpd = self._get_main_object(tree, "GPD_O", ns) 

23 

24 for key in gpd.attrib: 

25 setattr(self, key.split("}", 1)[1], gpd.attrib[key]) 

26 for child in gpd: 

27 key = self._get_tag(child) 

28 if len(child) == 0: 

29 setattr(self, key, child.text) 

30 elif key in ["registrationHistory", "lifespan"]: 

31 self._read_children_of_children(child) 

32 elif key == "report": 

33 if not hasattr(self, "report"): 

34 self.report = [] 

35 for grandchild in child: 

36 key = self._get_tag(grandchild) 

37 if key == "Report": 

38 self.report.append(self._read_report(grandchild)) 

39 else: 

40 self._warn_unknown_tag(key) 

41 else: 

42 self._warn_unknown_tag(key) 

43 

44 if hasattr(self, "report"): 

45 self.report = pd.DataFrame(self.report) 

46 if "reportId" in self.report.columns: 

47 self.report = self.report.set_index("reportId") 

48 if hasattr(self, "volumeSeries"): 

49 self.volumeSeries = pd.DataFrame(self.volumeSeries) 

50 for column in ["beginDate", "endDate"]: 

51 if column in self.volumeSeries.columns: 

52 self.volumeSeries[column] = pd.to_datetime( 

53 self.volumeSeries[column] 

54 ) 

55 if "volume" in self.volumeSeries.columns: 

56 self.volumeSeries["volume"] = pd.to_numeric(self.volumeSeries["volume"]) 

57 

58 def _read_report(self, node): 

59 d = {} 

60 for child in node: 

61 key = self._get_tag(child) 

62 if len(child) == 0: 

63 d[key] = child.text 

64 elif key == "reportPeriod": 

65 for grandchild in child: 

66 key = self._get_tag(grandchild) 

67 if key in ["beginDate", "endDate"]: 

68 d[key] = grandchild.text 

69 else: 

70 self._warn_unknown_tag(key) 

71 elif key == "volumeSeries": 

72 if not hasattr(self, "volumeSeries"): 

73 self.volumeSeries = [] 

74 vs = self._read_volume_series(child) 

75 vs["reportId"] = d["reportId"] 

76 self.volumeSeries.append(vs) 

77 

78 elif key == "installationOrFacility": 

79 for grandchild in child: 

80 key = self._get_tag(grandchild) 

81 if key == "InstallationOrFacility": 

82 self._read_installation_facility(grandchild) 

83 else: 

84 self._warn_unknown_tag(key) 

85 else: 

86 self._warn_unknown_tag(key) 

87 

88 if "volumeSeries" in d: 

89 d["volumeSeries"] = pd.DataFrame(d["volumeSeries"]) 

90 return d 

91 

92 def _read_volume_series(self, node): 

93 d = {} 

94 for child in node: 

95 key = self._get_tag(child) 

96 if len(child) == 0: 

97 d[key] = child.text 

98 elif key == "period": 

99 for grandchild in child: 

100 key = self._get_tag(grandchild) 

101 if key in ["beginDate", "endDate"]: 

102 d[key] = grandchild.text 

103 else: 

104 self._warn_unknown_tag(key) 

105 return d 

106 

107 def _read_installation_facility(self, node): 

108 for child in node: 

109 key = child.tag.split("}", 1)[1] 

110 if key == "relatedGroundwaterUsageFacility": 

111 for grandchild in child: 

112 key = self._get_tag(grandchild) 

113 if key == "GroundwaterUsageFacility": 

114 for greatgrandchild in grandchild: 

115 key2 = greatgrandchild.tag.split("}", 1)[1] 

116 if key2 == "broId": 

117 setattr(self, key, greatgrandchild.text) 

118 else: 

119 util._warn_unknown_key(key2, self) 

120 else: 

121 self._warn_unknown_tag(key) 

122 elif key == "relatedRealisedInstallation": 

123 for grandchild in child: 

124 key = self._get_tag(grandchild) 

125 if key == "RealisedInstallation": 

126 for greatgrandchild in grandchild: 

127 key2 = greatgrandchild.tag.split("}", 1)[1] 

128 if key2 == "broId": 

129 setattr(self, key, greatgrandchild.text) 

130 elif key2 == "realisedInstallationId": 

131 setattr(self, key2, greatgrandchild.text) 

132 else: 

133 util._warn_unknown_key(key2, self) 

134 else: 

135 self._warn_unknown_tag(key) 

136 else: 

137 self._warn_unknown_tag(key) 

138 

139 

140cl = GroundwaterProductionDossier 

141 

142get_bro_ids_of_bronhouder = partial(bro._get_bro_ids_of_bronhouder, cl) 

143get_bro_ids_of_bronhouder.__doc__ = bro._get_bro_ids_of_bronhouder.__doc__ 

144 

145get_data_for_bro_ids = partial(bro._get_data_for_bro_ids, cl) 

146get_data_for_bro_ids.__doc__ = bro._get_data_for_bro_ids.__doc__