Coverage for brodata / cpt.py: 81%

1import logging 

2import tempfile 

3from functools import partial 

4 

5import pandas as pd 

6import requests 

7 

8from . import bro 

9 

10logger = logging.getLogger(__name__) 

11 

12 

13class ConePenetrationTest(bro.FileOrUrl): 

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

15 

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

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

18 _char = "CPT_C" 

19 

20 def _read_contents(self, tree): 

21 ns = { 

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

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

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

25 "xmlns": self._xmlns, 

26 } 

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

28 for key in cpt.attrib: 

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

30 for child in cpt: 

31 key = self._get_tag(child) 

32 if len(child) == 0: 

33 setattr(self, key, child.text) 

34 elif key == "standardizedLocation": 

35 self._read_standardized_location(child) 

36 elif key == "deliveredLocation": 

37 self._read_delivered_location(child) 

38 elif key in ["researchReportDate"]: 

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

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

41 self._read_children_of_children(child) 

42 elif key in ["conePenetrometerSurvey"]: 

43 for grandchild in child: 

44 key = self._get_tag(grandchild) 

45 if len(grandchild) == 0: 

46 setattr(self, key, grandchild.text) 

47 elif key in [ 

48 "finalProcessingDate", 

49 "trajectory", 

50 "conePenetrometer", 

51 "procedure", 

52 ]: 

53 self._read_children_of_children(grandchild) 

54 elif key == "parameters": 

55 self._read_parameters(grandchild) 

56 elif key == "conePenetrationTest": 

57 self._read_cone_penetration_test(grandchild, key) 

58 elif key == "dissipationTest": 

59 self._read_cone_penetration_test(grandchild, key) 

60 else: 

61 self._warn_unknown_tag(key) 

62 elif key == "additionalInvestigation": 

63 self._read_additional_investigation(child) 

64 else: 

65 self._warn_unknown_tag(key) 

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

67 self.conePenetrationTest.columns = self.parameters.index 

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

69 self.conePenetrationTest = self.conePenetrationTest.set_index( 

70 "penetrationLength" 

71 ) 

72 

73 def _read_parameters(self, node): 

74 self.parameters = pd.Series() 

75 for child in node: 

76 key = self._get_tag(child) 

77 self.parameters[key] = child.text 

78 

79 def _read_cone_penetration_test(self, node, name): 

80 for child in node: 

81 key = self._get_tag(child) 

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

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

84 elif key in [ 

85 "procedure", 

86 "observedProperty", 

87 "featureOfInterest", 

88 "penetrationLength", 

89 ]: 

90 self._read_children_of_children(child) 

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

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

93 else: 

94 self._warn_unknown_tag(key) 

95 

96 def _read_additional_investigation(self, node): 

97 for child in node: 

98 key = self._get_tag(child) 

99 if len(child) == 0: 

100 setattr(self, key, child.text) 

101 elif key == "removedLayer": 

102 if not hasattr(self, key): 

103 self.removedLayer = [] 

104 d = {} 

105 self._read_children_of_children( 

106 child, 

107 d=d, 

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

109 to_int="sequenceNumber", 

110 ) 

111 self.removedLayer.append(d) 

112 if hasattr(self, "removedLayer"): 

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

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

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

116 

117 

118def get_graph_types(timeout=5): 

119 """ 

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

121 

122 Parameters 

123 ---------- 

124 timeout : int or float, optional 

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

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

127 

128 Returns 

129 ------- 

130 pd.DataFrame 

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

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

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

134 

135 """ 

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

137 r = requests.get(url) 

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

139 assert len(supported_graphs) == 1 

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

141 

142 

143def graph( 

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

145): 

146 """ 

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

148 

149 Parameters 

150 ---------- 

151 xml_file : str 

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

153 graphType : str, optional 

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

155 types. The default is "cptCombinedLength". 

156 to_file : str, optional 

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

158 timeout : int or float, optional 

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

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

161 return_fname : bool, optional 

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

163 

164 Returns 

165 ------- 

166 IPython.display.SVG or str 

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

168 

169 """ 

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

171 

172 params = {"graphType": graphType} 

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

174 r = requests.post(url, data=data, timeout=timeout, params=params) 

175 r.raise_for_status() 

176 if to_file is None: 

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

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

179 f.write(r.text) 

180 if return_fname: 

181 return to_file 

182 else: 

183 from IPython.display import SVG 

184 

185 return SVG(to_file) 

186 

187 

188cl = ConePenetrationTest 

189 

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

191get_bro_ids_of_bronhouder.__doc__ = bro._get_bro_ids_of_bronhouder.__doc__ 

192 

193get_data_for_bro_ids = partial(bro._get_data_for_bro_ids, cl) 

194get_data_for_bro_ids.__doc__ = bro._get_data_for_bro_ids.__doc__ 

195 

196get_characteristics = partial(bro._get_characteristics, cl) 

197get_characteristics.__doc__ = bro._get_characteristics.__doc__ 

198 

199get_data_in_extent = partial(bro._get_data_in_extent, cl) 

200get_data_in_extent.__doc__ = bro._get_data_in_extent.__doc__