Coverage for brodata / cpt.py: 81%

1import logging 

2import tempfile 

3from functools import partial 

4 

5import pandas as pd 

6 

7from . import bro 

8 

9logger = logging.getLogger(__name__) 

10 

11 

12class ConePenetrationTest(bro.FileOrUrl): 

13 """Class to represent a Cone Penetration Test (CPT) from the BRO.""" 

14 

15 _rest_url = "https://publiek.broservices.nl/sr/cpt/v1" 

16 _xmlns = "http://www.broservices.nl/xsd/dscpt/1.1" 

17 _char = "CPT_C" 

18 

19 def _read_contents(self, tree): 

20 ns = { 

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

22 "gml": "http://www.opengis.net/gml/3.2", 

23 "cptcommon": "http://www.broservices.nl/xsd/cptcommon/1.1", 

24 "xmlns": self._xmlns, 

25 } 

26 cpt = self._get_main_object(tree, "CPT_O", ns) 

27 for key in cpt.attrib: 

28 setattr(self, key.split("}", 1)[1], cpt.attrib[key]) 

29 for child in cpt: 

30 key = self._get_tag(child) 

31 if len(child) == 0: 

32 setattr(self, key, child.text) 

33 elif key == "standardizedLocation": 

34 self._read_standardized_location(child) 

35 elif key == "deliveredLocation": 

36 self._read_delivered_location(child) 

37 elif key in ["researchReportDate"]: 

38 setattr(self, key, self._read_date(child)) 

39 elif key in ["deliveredVerticalPosition", "registrationHistory"]: 

40 self._read_children_of_children(child) 

41 elif key in ["conePenetrometerSurvey"]: 

42 for grandchild in child: 

43 key = self._get_tag(grandchild) 

44 if len(grandchild) == 0: 

45 setattr(self, key, grandchild.text) 

46 elif key in [ 

47 "finalProcessingDate", 

48 "trajectory", 

49 "conePenetrometer", 

50 "procedure", 

51 ]: 

52 self._read_children_of_children(grandchild) 

53 elif key == "parameters": 

54 self._read_parameters(grandchild) 

55 elif key == "conePenetrationTest": 

56 self._read_cone_penetration_test(grandchild, key) 

57 elif key == "dissipationTest": 

58 self._read_cone_penetration_test(grandchild, key) 

59 else: 

60 self._warn_unknown_tag(key) 

61 elif key == "additionalInvestigation": 

62 self._read_additional_investigation(child) 

63 else: 

64 self._warn_unknown_tag(key) 

65 if hasattr(self, "conePenetrationTest") and hasattr(self, "parameters"): 

66 self.conePenetrationTest.columns = self.parameters.index 

67 if "penetrationLength" in self.conePenetrationTest.columns: 

68 self.conePenetrationTest = self.conePenetrationTest.set_index( 

69 "penetrationLength" 

70 ) 

71 

72 def _read_parameters(self, node): 

73 self.parameters = pd.Series() 

74 for child in node: 

75 key = self._get_tag(child) 

76 self.parameters[key] = child.text 

77 

78 def _read_cone_penetration_test(self, node, name): 

79 for child in node: 

80 key = self._get_tag(child) 

81 if key in ["phenomenonTime", "resultTime"]: 

82 setattr(self, f"{name}_{key}", self._read_time_instant(child)) 

83 elif key in [ 

84 "procedure", 

85 "observedProperty", 

86 "featureOfInterest", 

87 "penetrationLength", 

88 ]: 

89 self._read_children_of_children(child) 

90 elif key in ["cptResult", "disResult"]: 

91 setattr(self, name, self._read_data_array(child)) 

92 else: 

93 self._warn_unknown_tag(key) 

94 

95 def _read_additional_investigation(self, node): 

96 for child in node: 

97 key = self._get_tag(child) 

98 if len(child) == 0: 

99 setattr(self, key, child.text) 

100 elif key == "removedLayer": 

101 if not hasattr(self, key): 

102 self.removedLayer = [] 

103 d = {} 

104 self._read_children_of_children( 

105 child, 

106 d=d, 

107 to_float=["upperBoundary", "lowerBoundary"], 

108 to_int="sequenceNumber", 

109 ) 

110 self.removedLayer.append(d) 

111 if hasattr(self, "removedLayer"): 

112 self.removedLayer = pd.DataFrame(self.removedLayer) 

113 if "sequenceNumber" in self.removedLayer.columns: 

114 self.removedLayer = self.removedLayer.set_index("sequenceNumber") 

115 

116 

117def get_graph_types(timeout=5): 

118 """ 

119 Get the graph types that can be generated for CPT by the REST API of the BRO. 

120 

121 Parameters 

122 ---------- 

123 timeout : int or float, optional 

124 A number indicating how many seconds to wait for the client to make a connection 

125 and/or send a response. The default is 5. 

126 

127 Returns 

128 ------- 

129 pd.DataFrame 

130 A Pandas DataFrame that contains the supported graph types, with the columns 

131 'name' and 'description'. The index of this DataFrame contains the strings that 

132 can be used for the graphType-argument in nlmod.cpt.graph(). 

133 

134 """ 

135 url = "https://publiek.broservices.nl/sr/cpt/v1/result/graph/types" 

136 r = bro.util.get_with_rate_limit(url) 

137 supported_graphs = r.json()["supportedGraphs"] 

138 assert len(supported_graphs) == 1 

139 return pd.DataFrame(supported_graphs[0]["graphs"]).set_index("graphType") 

140 

141 

142def graph( 

143 xml_file, graphType="cptCombinedLength", to_file=None, timeout=5, return_fname=False 

144): 

145 """ 

146 Generate a svg-graph of a cpt-file (ConePenetrationTest). 

147 

148 Parameters 

149 ---------- 

150 xml_file : str 

151 The filename of the xml-file to generate a graphical representation of. 

152 graphType : str, optional 

153 The type of graph. Run `brodata.cpt.get_graph_types()` to view available graph 

154 types. The default is "cptCombinedLength". 

155 to_file : str, optional 

156 The filename to save the svg-file to. The default is None. 

157 timeout : int or float, optional 

158 A number indicating how many seconds to wait for the client to make a connection 

159 and/or send a response. The default is 5. 

160 return_fname : bool, optional 

161 If True, Return the filename of the svg-file. The default is False. 

162 

163 Returns 

164 ------- 

165 IPython.display.SVG or str 

166 A graphical representation of the svg-file or the filename of the svg-file. 

167 

168 """ 

169 url = "https://publiek.broservices.nl/sr/cpt/v1/result/graph/dispatch" 

170 

171 params = {"graphType": graphType} 

172 with open(xml_file, "rb") as data: 

173 r = bro.util.post_with_rate_limit( 

174 url, data=data, timeout=timeout, params=params 

175 ) 

176 r.raise_for_status() 

177 if to_file is None: 

178 to_file = tempfile.NamedTemporaryFile(suffix=".svg").name 

179 with open(to_file, "w", encoding="utf-8") as f: 

180 f.write(r.text) 

181 if return_fname: 

182 return to_file 

183 else: 

184 from IPython.display import SVG 

185 

186 return SVG(to_file) 

187 

188 

189cl = ConePenetrationTest 

190 

191get_bro_ids_of_bronhouder = partial(bro._get_bro_ids_of_bronhouder, cl=cl) 

192get_bro_ids_of_bronhouder.__doc__ = bro._get_bro_ids_of_bronhouder.__doc__ 

193 

194get_data_for_bro_ids = partial(bro._get_data_for_bro_ids, cl) 

195get_data_for_bro_ids.__doc__ = bro._get_data_for_bro_ids.__doc__ 

196 

197get_characteristics = partial(bro._get_characteristics, cl) 

198get_characteristics.__doc__ = bro._get_characteristics.__doc__ 

199 

200get_data_in_extent = partial(bro._get_data_in_extent, cl) 

201get_data_in_extent.__doc__ = bro._get_data_in_extent.__doc__